Description

praisonaiagents: AgentServer declares auth_token but never enforces it on any route Researcher: Kai Aizen — SnailSploit (@SnailSploit), Adversarial & Offensive Security Research Target: https://github.com/MervinPraison/PraisonAI --- Package: praisonaiagents on PyPI Affected version (empirically tested): 1.6.48 Component: praisonaiagents.server.AgentServer (the bundled HTTP / SSE server) --- ## TL;DR 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. It 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. The 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. ## Root cause Expected behavior when setting ServerConfig(auth_token="…"): "Only requests with a matching Authorization header will be accepted on /publish, /events, /info." Actual behavior (server/server.py, dist 1.6.48): - line 31 auth_token: Optional[str] = None # declared - line 39 "auth_token": "***" if self.auth_token else None # displayed - lines 122-204: no auth middleware, no Depends, no request.headers["Authorization"] read, no comparison to self.config.auth_token. Impact: The configuration knob is dead code from the route handlers' perspective. All routes always run. The operator has no signal that their auth_token was discarded — /info even confirms it was received by displaying "***". ## Sibling proof — the same package gets it right elsewhere praisonaiagents/ui/a2a/a2a.py: python # line 163 async def _verify_auth(authorization: Optional[str] = Header(None)): """Verify bearer token if auth_token is configured.""" if self.auth_token is None: return ... if len(parts) != 2 or parts[0].lower() != "bearer" \ or parts[1] != self.auth_token: raise HTTPException(status_code=401, ...) # line 192 from fastapi import Depends _a2a_deps = [Depends(_verify_auth)] if self.auth_token else [] That 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. ## Affected routes (empirically tested) | Route | Method | Accepts unauth requests? | Impact | |-------------|--------|--------------------------|------------------------------------------------------------------------| | /info | GET | Yes (200) | Leaks server config; confirms auth_token is set ("***"); reveals client count and CORS config. | | /publish | POST | Yes (200) | Anyone broadcasts arbitrary {type, data} to every subscribed agent. Event payload is whatever the attacker sends. | | /events | GET | Yes (200) | Anyone subscribes to the SSE stream and observes every event published by the server (and by any other anonymous attacker). | | /health | GET | Yes (200) | Leaks live SSE client count. | ## Impact The /publish and /events routes are the load-bearing ones. Together they let an unauthenticated network-adjacent attacker: 1. 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. 2. 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. 3. 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. 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. ## Anchors praisonaiagents 1.6.48, file praisonaiagents/server/server.py: | Line | Symbol | What it shows | |-------|---------------------------------------------------------|---------------| | 31 | auth_token: Optional[str] = None | Declared. | | 39 | "auth_token": "***" if self.auth_token else None | Displayed (masked) in /info. | | 121 | def _create_app(self): | Route + middleware setup begins. | | 132 | async def health(request): | No auth check. | | 139 | async def events(request): | No auth check. | | 164 | async def publish(request): | No auth check. | | 182 | async def info(request): | No auth check. | | 190 | routes = [Route("/health", …), Route("/events", …), Route("/publish", …), Route("/info", …)] | Routes registered without Depends/middleware. | | 197 | app = Starlette(routes=routes) | App created. | | 200 | app = CORSMiddleware(app, …) | Only middleware added. | Source sha256 (1.6.48, praisonaiagents/server/server.py): aac9497d515b5cb928070267b860b11ef38b537605e64659feef895b524ca7e4 (9,962 bytes). Sibling (same package, same field name, enforced): praisonaiagents/ui/a2a/a2a.py:163-193. ## Reproduction (empirical PoC) poc/poc.py starts AgentServer with ServerConfig(auth_token="supersecret-not-actually-checked") and then sends unauthenticated requests to each route. Run log (poc/run-log.txt): [1] GET /info (no Authorization) -> HTTP 200 body: {"name":"PraisonAI Agent Server","version":"1.0.0","clients":0, "config":{"host":"127.0.0.1","port":18765,"cors_origins":[], "auth_token":"***","max_connections":100}} [2] POST /publish (no Authorization) -> HTTP 200 body: {"success":true,"clients":0} [3] GET /health (no Authorization) -> HTTP 200 [4] GET /events (no Authorization) -> HTTP 200 VULNERABLE: 4 unauthenticated routes VERDICT: VULNERABLE EXIT 0 ## Suggested fix Make AgentServer reuse the A2A pattern. Smallest fix: python # in _create_app, after `app = Starlette(routes=routes)`: if self.config.auth_token: from starlette.middleware.base import BaseHTTPMiddleware from starlette.responses import JSONResponse expected = "Bearer " + self.config.auth_token class _Auth(BaseHTTPMiddleware): async def dispatch(self, request, call_next): if request.url.path == "/health": # if /health should remain public return await call_next(request) got = request.headers.get("authorization", "") if not hmac.compare_digest(got, expected): return JSONResponse({"error": "unauthorized"}, status_code=401) return await call_next(request) app = _Auth(app) app = CORSMiddleware(app, ...) Constant-time comparison (hmac.compare_digest) is appropriate since this is a network-comparable secret. ## Steps to reproduce 1. Clone the target: `git clone --depth 1

Mitigation

Aliases

References