Robert Helewka
e7c84885d9
The UI never sent the bearer token, so with API_TOKEN set every data call 401'd and /ws/events was rejected pre-accept (the 403s in the uvicorn log). The API client now keeps the token in localStorage, attaches Authorization to every request, prompts once on a 401 and retries, and appends ?token= to the WebSocket connect — the page's reconnect loop picks the token up after the first prompt. require_token also accepts a ?token= query parameter (same convention as the WebSocket) because <audio> elements fetching recordings can't set headers; recordingUrl() rides the token there. The dashboard header's status call moved to a new authenticated GET /api/v1/status — its old source was the JSON root endpoint that the dashboard itself replaced at /. Two new auth tests (query-param accepted / wrong query-param 401); dashboard rebuilt. Verified live: WS rejected without token and connected with ?token=, status 200, ?token=wrong 401. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
"""
|
|
API surface tests — bearer-token enforcement and route registration order.
|
|
|
|
The app is exercised without its lifespan: auth runs before any handler,
|
|
so a 503 ("Gateway not initialized") proves the token was accepted.
|
|
"""
|
|
|
|
import httpx
|
|
import pytest
|
|
from pydantic import SecretStr
|
|
from starlette.routing import Match
|
|
|
|
import main
|
|
from config import get_settings
|
|
|
|
TOKEN = "test-token-for-suite"
|
|
|
|
|
|
@pytest.fixture
|
|
def token_enabled(monkeypatch):
|
|
monkeypatch.setattr(get_settings(), "api_token", SecretStr(TOKEN))
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
class TestBearerToken:
|
|
async def test_missing_token_rejected(self, token_enabled, client):
|
|
resp = await client.get("/api/v1/calls/active")
|
|
assert resp.status_code == 401
|
|
assert resp.headers["www-authenticate"] == "Bearer"
|
|
|
|
async def test_wrong_token_rejected(self, token_enabled, client):
|
|
resp = await client.get(
|
|
"/api/v1/calls/active", headers={"Authorization": "Bearer wrong"}
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
async def test_valid_token_reaches_handler(self, token_enabled, client):
|
|
resp = await client.get(
|
|
"/api/v1/calls/active", headers={"Authorization": f"Bearer {TOKEN}"}
|
|
)
|
|
# No lifespan ran, so the handler itself 503s — auth was accepted
|
|
assert resp.status_code == 503
|
|
|
|
async def test_empty_token_disables_auth(self, monkeypatch, client):
|
|
monkeypatch.setattr(get_settings(), "api_token", SecretStr(""))
|
|
resp = await client.get("/api/v1/calls/active")
|
|
assert resp.status_code == 503
|
|
|
|
async def test_query_param_token_accepted(self, token_enabled, client):
|
|
"""<audio>/<a> elements can't set headers — ?token= must work."""
|
|
resp = await client.get(f"/api/v1/calls/active?token={TOKEN}")
|
|
assert resp.status_code == 503 # auth accepted, handler 503s (no lifespan)
|
|
|
|
async def test_wrong_query_param_token_rejected(self, token_enabled, client):
|
|
resp = await client.get("/api/v1/calls/active?token=wrong")
|
|
assert resp.status_code == 401
|
|
|
|
async def test_all_api_routers_protected(self, token_enabled, client):
|
|
for path in ("/api/v1/calls/active", "/api/v1/call-flows/", "/api/v1/devices/",
|
|
"/api/v1/routing/rules", "/api/v1/calls/history"):
|
|
resp = await client.get(path)
|
|
assert resp.status_code == 401, path
|
|
|
|
|
|
class TestRouteOrder:
|
|
def _resolve(self, path: str):
|
|
scope = {
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": path,
|
|
"root_path": "",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
}
|
|
for route in main.app.router.routes:
|
|
match, _ = route.matches(scope)
|
|
if match == Match.FULL:
|
|
return route
|
|
return None
|
|
|
|
def test_history_not_shadowed_by_call_id(self):
|
|
route = self._resolve("/api/v1/calls/history")
|
|
assert route is not None
|
|
assert route.endpoint.__name__ == "list_history"
|
|
|
|
def test_call_id_still_matches(self):
|
|
route = self._resolve("/api/v1/calls/call_abc123")
|
|
assert route is not None
|
|
assert route.endpoint.__name__ == "get_call"
|
|
|
|
def test_mcp_mounted(self):
|
|
mounted = [getattr(r, "path", "") for r in main.app.router.routes]
|
|
assert "/mcp" in mounted
|