diff --git a/CLAUDE.md b/CLAUDE.md index 808d92d..3eb7e54 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -213,7 +213,14 @@ work** — raise them. estate services, there's no exposition endpoint here yet. - **No health-probe access-log filter.** Every `/health` poll hits the access log. Other estate services suppress probe noise; this one doesn't. -- **No rate limiting** on API endpoints (README Phase 4, unchecked). +- ~~**No rate limiting** on API endpoints.~~ Done, but *narrowly*: only the + unauthenticated `/auth/*` routes are limited + ([core/rate_limit.py](core/rate_limit.py)), because every other surface is + already owner-gated and a limit there would throttle the sole operator. The + limiter keys on the **socket peer, not `X-Forwarded-For`** — behind the + estate's reverse proxy that means per-proxy, not per-caller. Per-caller limits + need an explicit trusted-proxy config; don't silently start trusting the + header. - **Docker: single-image `Dockerfile` + `docker-compose.yaml`** (app + `postgres:17`) ship in-repo; the Gitea CI (`cve-scan-docker-build.yml`) builds the image on push to `main`. The compose stack requires SSO enabled diff --git a/README.md b/README.md index d12fdc2..8ecaedf 100644 --- a/README.md +++ b/README.md @@ -106,6 +106,7 @@ hold-slayer/ │ ├── pjsua_engine.py # PJSUA2 SIP engine (call control + media) │ ├── media_pipeline.py # PJSUA2 audio routing │ ├── logging_config.py # Text/JSON log formatting +│ ├── rate_limit.py # Fixed-window limiter for the /auth/* edge │ ├── call_manager.py # Active call state management │ └── event_bus.py # Async pub/sub event bus ├── services/ @@ -468,15 +469,15 @@ Full documentation is in [`/docs`](docs/README.md): - [x] Notification service (WebSocket + SMS) - [x] Service wiring in main.py lifespan -### Phase 4: Production Hardening 🚧 +### Phase 4: Production Hardening ✅ - [x] Alembic database migrations (baseline + upgrade-on-boot) - [x] API authentication — Casdoor SSO (browser JWT) + owner-minted PATs, owner-only across REST/WS/MCP - [x] Emergency-number guard + concurrent-call cap on outbound calls -- [ ] Rate limiting on API endpoints +- [x] Rate limiting on the unauthenticated `/auth/*` edge (everything else is owner-gated; see [core/rate_limit.py](core/rate_limit.py)) - [x] Structured JSON logging (`LOG_FORMAT=json`, uvicorn access log included) - [x] Honest /health — engine mode, DB ping, trunk registration, STT/TTS availability -- [ ] Graceful degradation (classifier works without STT, etc.) +- [x] Graceful degradation — a down STT/LLM/TTS degrades the call and publishes an `ERROR` event naming the service, rather than aborting it - [x] Docker Compose (Hold Slayer + PostgreSQL) ### Phase 5: Additional Services 🚧 diff --git a/api/auth.py b/api/auth.py index 929ecb3..b8ac71c 100644 --- a/api/auth.py +++ b/api/auth.py @@ -15,15 +15,25 @@ non-owner sees an "access denied" screen instead of a bare 401. import secrets from urllib.parse import urlencode -from fastapi import APIRouter, HTTPException, Query, Request +from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from auth import get_sdk, is_owner, resolve_from_header_or_query from config import get_settings +from core.rate_limit import rate_limit from db.database import session_scope router = APIRouter(prefix="/auth", tags=["auth"]) +# These are the only routes that must answer before an identity exists, so they +# are the only ones worth limiting — everything else is already owner-gated. +# `/callback` and `/refresh-callback` each trigger an outbound token exchange +# with Casdoor, and `/me` opens a DB session per call; all three are +# unauthenticated work an attacker controls. See core/rate_limit.py. +_limit_callback = [Depends(rate_limit("auth:callback", limit=10))] +_limit_me = [Depends(rate_limit("auth:me"))] +_limit_redirect = [Depends(rate_limit("auth:redirect"))] + def _build_casdoor_auth_url( callback: str, @@ -50,7 +60,7 @@ def _build_casdoor_auth_url( return f"{c.endpoint.rstrip('/')}/login/oauth/authorize?{urlencode(params)}" -@router.get("/login") +@router.get("/login", dependencies=_limit_redirect) async def login(request: Request, redirect_uri: str = Query(None)): """Redirect the browser to the Casdoor authorization page. @@ -64,7 +74,7 @@ async def login(request: Request, redirect_uri: str = Query(None)): return RedirectResponse(url=_build_casdoor_auth_url(callback)) -@router.get("/callback") +@router.get("/callback", dependencies=_limit_callback) async def callback( code: str = Query(...), state: str = Query(None), @@ -89,7 +99,7 @@ async def callback( return RedirectResponse(url=f"/#token={access_token}") -@router.get("/silent-refresh") +@router.get("/silent-refresh", dependencies=_limit_redirect) async def silent_refresh(request: Request): """Start a silent token refresh via hidden iframe (``prompt=none``). @@ -104,7 +114,7 @@ async def silent_refresh(request: Request): return RedirectResponse(url=_build_casdoor_auth_url(callback, prompt="none")) -@router.get("/refresh-callback") +@router.get("/refresh-callback", dependencies=_limit_callback) async def refresh_callback( code: str = Query(None), error: str = Query(None), @@ -140,7 +150,7 @@ async def refresh_callback( ) -@router.get("/me") +@router.get("/me", dependencies=_limit_me) async def me(request: Request): """Return the current authenticated user's profile + ``is_owner``. diff --git a/core/rate_limit.py b/core/rate_limit.py new file mode 100644 index 0000000..1321936 --- /dev/null +++ b/core/rate_limit.py @@ -0,0 +1,132 @@ +""" +Rate limiting for the unauthenticated edge. + +**Scope, and why it is narrow.** Every REST/WS/MCP surface is owner-only: an +unauthenticated request is rejected by `resolve_bearer`/`is_owner` before any +handler runs, and real spend control for outbound calls is +`max_concurrent_calls` in `gateway.make_call`. Blanket per-endpoint limits would +therefore mostly rate-limit the single legitimate operator. What is genuinely +exposed is the handful of `/auth/*` routes that must answer before an identity +exists — those are limited here, and nothing else. + +**What this defends against.** Not credential guessing: PATs are +`secrets.token_urlsafe(32)` (256 bits) compared by SHA-256 digest, so brute +force is not a practical threat. The concern is *unauthenticated work an +attacker controls*: `/auth/callback` makes an outbound token-exchange round-trip +to Casdoor on every request, and `/auth/me` opens a DB session and runs a token +lookup. Both are free to trigger and neither is cheap to serve. + +**Fixed-window, in-process, no dependency.** One operator and a single process +mean a shared counter store would be infrastructure without a purpose. The +trade-off of a fixed window is a burst of up to 2× the limit across a boundary; +that is irrelevant at these thresholds, and it costs one dict lookup with no +background task. + +**Client identity is the socket peer, deliberately.** `X-Forwarded-For` is +attacker-controlled unless a trusted proxy overwrites it, and this app does not +currently establish that trust (`_public_base_url` reads forwarded headers, but +only to build URLs). Keying on a spoofable header would let one client present +as thousands and make the limiter worse than useless. Behind the estate's +reverse proxy this means the limit applies per-proxy rather than per-caller — +correct for exhaustion, and honest about what it can enforce. If per-caller +limits are ever needed, that needs an explicit trusted-proxy config, not a +silent `X-Forwarded-For` read. +""" + +import time + +from fastapi import HTTPException, Request + +# Requests allowed per window, per client, per route. Sized to be invisible to a +# human — a dashboard load touches /auth/me once — while capping automated +# hammering. +DEFAULT_LIMIT = 30 +DEFAULT_WINDOW_SECONDS = 60 + +# Stop the bucket store growing without bound under a spray of source addresses. +# Eviction is oldest-first and only runs when the cap is exceeded. +MAX_TRACKED_CLIENTS = 10_000 + + +class RateLimiter: + """Fixed-window request counter, keyed by (route, client).""" + + def __init__( + self, + limit: int = DEFAULT_LIMIT, + window_seconds: int = DEFAULT_WINDOW_SECONDS, + max_clients: int = MAX_TRACKED_CLIENTS, + ): + self.limit = limit + self.window_seconds = window_seconds + self.max_clients = max_clients + # key -> [window_start, count]. Insertion-ordered, which is what makes + # oldest-first eviction a cheap `next(iter(...))`. + self._buckets: dict[str, list[float]] = {} + + def check( + self, key: str, limit: int | None = None, now: float | None = None + ) -> tuple[bool, int]: + """Record a hit. Returns (allowed, retry_after_seconds). + + `retry_after` is 0 when allowed, and the seconds remaining in the + current window when not. `limit` overrides the instance default for + routes that want a tighter cap. + """ + now = time.monotonic() if now is None else now + effective = self.limit if limit is None else limit + bucket = self._buckets.get(key) + + if bucket is None or now - bucket[0] >= self.window_seconds: + self._buckets.pop(key, None) # re-insert so ordering tracks recency + self._buckets[key] = [now, 1] + self._evict_if_needed() + return True, 0 + + bucket[1] += 1 + if bucket[1] > effective: + remaining = self.window_seconds - (now - bucket[0]) + return False, max(1, int(remaining) + 1) + return True, 0 + + def _evict_if_needed(self) -> None: + while len(self._buckets) > self.max_clients: + self._buckets.pop(next(iter(self._buckets))) + + def reset(self) -> None: + self._buckets.clear() + + +def client_key(request: Request, scope: str) -> str: + """Identify the caller for limiting purposes. + + Uses the socket peer, never a forwarded header — see the module docstring. + """ + client = request.client.host if request.client else "unknown" + return f"{scope}:{client}" + + +_limiter = RateLimiter() + + +def get_limiter() -> RateLimiter: + return _limiter + + +def rate_limit(scope: str, limit: int | None = None): + """FastAPI dependency factory limiting one route. + + Applied per-route rather than as middleware so the authenticated surfaces — + which are already owner-gated — pay nothing. + """ + + async def _dependency(request: Request) -> None: + allowed, retry_after = get_limiter().check(client_key(request, scope), limit=limit) + if not allowed: + raise HTTPException( + status_code=429, + detail="Too many requests", + headers={"Retry-After": str(retry_after)}, + ) + + return _dependency diff --git a/docs/api-reference.md b/docs/api-reference.md index 672ce84..3e3fe68 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -274,10 +274,21 @@ All errors follow a consistent format: | Status Code | Meaning | |-------------|---------| | `400` | Bad request (invalid parameters) | +| `401` | Not authenticated (missing or invalid bearer token) | +| `403` | Authenticated but not the owner | | `404` | Resource not found (call, flow, device) | | `409` | Conflict (call already ended, device already registered) | +| `429` | Rate limited — `/auth/*` routes only. Carries `Retry-After` (seconds) | | `500` | Internal server error | +`429` applies solely to the unauthenticated `/auth/*` edge: those routes must +answer before an identity exists, and each does real work (`/auth/callback` +makes an outbound token exchange with Casdoor, `/auth/me` opens a DB session). +The owner-gated API is not rate limited — it is already restricted to a single +operator. Limits are per client, per route, in a fixed 60-second window; the +client is the **socket peer**, so behind a reverse proxy the limit applies +per-proxy. See [core/rate_limit.py](../core/rate_limit.py). + ## WebSocket ### Event Stream diff --git a/tests/test_rate_limit.py b/tests/test_rate_limit.py new file mode 100644 index 0000000..7c233f4 --- /dev/null +++ b/tests/test_rate_limit.py @@ -0,0 +1,185 @@ +""" +Rate-limiting tests. + +The limiter guards the unauthenticated `/auth/*` edge — the only routes that +must answer before an identity exists. Everything else is owner-gated, so a +limit there would mostly throttle the single legitimate operator. + +The properties worth pinning: the cap actually blocks, windows expire, clients +and routes don't share a bucket, the store can't grow without bound, and +identity comes from the socket peer rather than a spoofable header. +""" + +import httpx +import pytest +from fastapi import Depends, FastAPI + +import main +from core.rate_limit import RateLimiter, client_key, get_limiter, rate_limit + + +@pytest.fixture(autouse=True) +def _clean_limiter(): + get_limiter().reset() + yield + get_limiter().reset() + + +class TestRateLimiter: + def test_allows_up_to_the_limit(self): + rl = RateLimiter(limit=3, window_seconds=60) + assert [rl.check("k", now=100.0)[0] for _ in range(3)] == [True, True, True] + + def test_blocks_past_the_limit(self): + rl = RateLimiter(limit=3, window_seconds=60) + for _ in range(3): + rl.check("k", now=100.0) + allowed, retry_after = rl.check("k", now=100.0) + assert allowed is False + assert retry_after > 0 + + def test_window_expiry_resets_the_count(self): + rl = RateLimiter(limit=2, window_seconds=60) + rl.check("k", now=100.0) + rl.check("k", now=100.0) + assert rl.check("k", now=100.0)[0] is False + # A full window later, the caller is welcome again. + assert rl.check("k", now=161.0)[0] is True + + def test_retry_after_shrinks_as_the_window_drains(self): + rl = RateLimiter(limit=1, window_seconds=60) + rl.check("k", now=100.0) + early = rl.check("k", now=110.0)[1] + late = rl.check("k", now=150.0)[1] + assert early > late >= 1 + + def test_clients_do_not_share_a_bucket(self): + rl = RateLimiter(limit=1, window_seconds=60) + assert rl.check("auth:me:1.1.1.1", now=100.0)[0] is True + # A different client must be unaffected by the first one's usage. + assert rl.check("auth:me:2.2.2.2", now=100.0)[0] is True + + def test_routes_do_not_share_a_bucket(self): + rl = RateLimiter(limit=1, window_seconds=60) + assert rl.check("auth:me:1.1.1.1", now=100.0)[0] is True + assert rl.check("auth:callback:1.1.1.1", now=100.0)[0] is True + + def test_per_call_limit_overrides_the_default(self): + rl = RateLimiter(limit=100, window_seconds=60) + rl.check("k", limit=1, now=100.0) + assert rl.check("k", limit=1, now=100.0)[0] is False + + def test_bucket_store_is_bounded(self): + # Otherwise a spray of source addresses is itself a memory exhaustion + # vector — the thing the limiter exists to prevent. + rl = RateLimiter(limit=5, window_seconds=60, max_clients=10) + for i in range(50): + rl.check(f"client-{i}", now=100.0 + i) + assert len(rl._buckets) <= 10 + + def test_eviction_drops_oldest_first(self): + rl = RateLimiter(limit=5, window_seconds=600, max_clients=3) + for i in range(4): + rl.check(f"client-{i}", now=100.0 + i) + assert "client-0" not in rl._buckets + assert "client-3" in rl._buckets + + +class TestClientKey: + def _request(self, peer: str | None, headers: dict | None = None): + scope = { + "type": "http", + "method": "GET", + "path": "/auth/me", + "headers": [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()], + "client": (peer, 12345) if peer else None, + } + from starlette.requests import Request + + return Request(scope) + + def test_uses_the_socket_peer(self): + assert client_key(self._request("10.0.0.5"), "auth:me") == "auth:me:10.0.0.5" + + def test_ignores_x_forwarded_for(self): + # Trusting a spoofable header would let one client present as + # thousands, making the limiter worse than useless. + key = client_key( + self._request("10.0.0.5", {"X-Forwarded-For": "1.2.3.4"}), "auth:me" + ) + assert key == "auth:me:10.0.0.5" + assert "1.2.3.4" not in key + + def test_missing_peer_does_not_crash(self): + assert client_key(self._request(None), "auth:me") == "auth:me:unknown" + + +class TestDependency: + """The FastAPI integration: a 429 with a Retry-After header.""" + + def _app(self, limit=2): + app = FastAPI() + + @app.get("/limited", dependencies=[Depends(rate_limit("test", limit=limit))]) + async def limited(): + return {"ok": True} + + @app.get("/unlimited") + async def unlimited(): + return {"ok": True} + + return app + + async def _get(self, app, path, n=1): + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: + return [await c.get(path) for _ in range(n)] + + async def test_returns_429_past_the_limit(self): + responses = await self._get(self._app(limit=2), "/limited", n=3) + assert [r.status_code for r in responses] == [200, 200, 429] + + async def test_429_carries_retry_after(self): + responses = await self._get(self._app(limit=1), "/limited", n=2) + blocked = responses[-1] + assert blocked.status_code == 429 + assert int(blocked.headers["retry-after"]) >= 1 + + async def test_unlimited_routes_are_untouched(self): + responses = await self._get(self._app(limit=1), "/unlimited", n=10) + assert {r.status_code for r in responses} == {200} + + +class TestAuthRoutesAreLimited: + """The wiring: the unauthenticated edge is covered, the rest is not.""" + + def _is_limited(self, path: str) -> bool: + """True if the route carries a dependency built by `rate_limit`. + + Identified by the closure's qualname rather than a string search, so + this can't pass on an unrelated dependency that happens to stringify + similarly. + """ + for route in main.app.routes: + if getattr(route, "path", None) != path: + continue + return any( + getattr(d.dependency, "__qualname__", "").startswith("rate_limit") + for d in getattr(route, "dependencies", []) + ) + raise AssertionError(f"route {path} not found") + + @pytest.mark.parametrize( + "path", ["/auth/login", "/auth/callback", "/auth/me", "/auth/refresh-callback"] + ) + def test_unauthenticated_auth_routes_are_limited(self, path): + assert self._is_limited(path) + + def test_owner_gated_routes_are_not_limited(self): + # They're already behind is_owner; limiting them would throttle the + # single legitimate operator. + assert not self._is_limited("/api/v1/calls/active") + + def test_logout_is_not_limited(self): + # Pure redirect builder — no I/O, nothing to exhaust. + assert not self._is_limited("/auth/logout")