Résumé

praisonaiagents: AgentServer declares auth_token but never enforces it on any route

Détails de l’avis

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:

# 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:

# 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

Références