PyPI · praisonaiagents
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)
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.
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 "***".
praisonaiagents/ui/a2a/a2a.py:
# 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.
| 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. |
The /publish and /events routes are the load-bearing ones. Together they let an unauthenticated network-adjacent attacker:
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./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./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.
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.
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
Make AgentServer reuse the A2A pattern. Smallest fix:
# 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 publi
Is your project exposed to this? Stateward checks every dependency on every pull request and flags it only if your code actually reaches it.
Check my repoSources: CISA KEV (public domain), OSV.dev & GitHub Advisory Database (CC-BY-4.0), FIRST EPSS, NVD/CWE (public domain). Served live from the Stateward advisory database.