feat(auth): rate-limit the unauthenticated /auth/* edge

Blanket per-endpoint limits would have been the wrong shape here. Every
REST/WS/MCP surface is owner-only — an unauthenticated request is rejected by
resolve_bearer/is_owner before any handler runs — so limiting them would
mostly throttle the single legitimate operator, and real spend control for
outbound calls is already max_concurrent_calls in gateway.make_call.

What is genuinely exposed is the handful of /auth/* routes that must answer
before an identity exists. /auth/callback and /auth/refresh-callback each make
an outbound token exchange with Casdoor on every request; /auth/me opens a DB
session and runs a token lookup. All are free to trigger and none are cheap to
serve. /auth/logout is left unlimited — it builds a redirect URL and does no
I/O.

Not a defence against credential guessing: PATs are secrets.token_urlsafe(32)
(256 bits) compared by SHA-256 digest, so brute force was never the threat.
This is about unauthenticated work an attacker controls.

Fixed-window, in-process, no new dependency — one operator and one process
make a shared counter store infrastructure without a purpose. The bucket store
is bounded and evicts oldest-first, since an unbounded map keyed by source
address would itself be the exhaustion vector.

The limiter keys on the socket peer and deliberately ignores X-Forwarded-For.
That header is attacker-controlled unless a trusted proxy overwrites it, and
this app establishes no such trust; keying on it would let one client present
as thousands and make the limiter worse than useless. Behind the estate's
reverse proxy the limit is therefore per-proxy, not per-caller — correct for
exhaustion and honest about what it can enforce. Per-caller limits need an
explicit trusted-proxy config, noted in CLAUDE.md so it isn't added silently.

Verified against a real server: exactly 30 requests pass, then 429 with
Retry-After: 60, while an owner-gated route serves 40/40. The 429s appear in
the JSON access log with queryable status_code and client_addr, so an attack
is visible in Loki. The wiring test identifies the dependency by qualname
rather than string search, and was mutation-checked by removing the limit from
/auth/me.

Also documents 401/403/429 in the API reference — 401 and 403 have existed
since auth landed but were never in the status-code table. Phase 4 is now
complete.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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

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 🚧

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``.
 

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

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

tests/test_graceful_degradation.py

@@ -0,0 +1,187 @@
+"""
+Graceful-degradation tests.
+
+Every external dependency — STT, LLM, TTS — is reachable over the network and
+can be down. The gateway's rule is that a dead dependency degrades the call
+rather than killing it, and says so: each failure publishes an `ERROR` event
+naming the service, so a down Speaches reads as "transcription failed" rather
+than "the AI is making bad decisions".
+
+The behaviour is already implemented across the services; these tests exist so a
+later refactor can't quietly remove it. The failure mode being guarded against is
+silent: an un-caught exception in one of these paths aborts a live phone call.
+"""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from config import Settings
+from core.event_bus import EventBus
+from models.events import EventType
+
+
+def _gateway():
+    """A gateway stand-in with a real event bus, so events can be asserted."""
+    gw = MagicMock()
+    gw.settings = Settings(database_url="sqlite+aiosqlite:///:memory:")
+    gw.event_bus = EventBus()
+    gw.call_manager = MagicMock()
+    gw.call_manager.add_transcript = AsyncMock()
+    return gw
+
+
+def _hold_slayer(gateway, transcription):
+    from services.audio_classifier import AudioClassifier
+    from services.hold_slayer import HoldSlayerService
+
+    return HoldSlayerService(
+        gateway=gateway,
+        call_manager=gateway.call_manager,
+        sip_engine=MagicMock(),
+        classifier=AudioClassifier(gateway.settings.classifier),
+        transcription=transcription,
+        settings=gateway.settings,
+    )
+
+
+async def _errors_for(bus: EventBus, coro):
+    """Run `coro` while subscribed, returning the ERROR events it published."""
+    import asyncio
+
+    sub = bus.subscribe(event_types={EventType.ERROR})
+    try:
+        result = await coro
+        seen = []
+        while True:
+            try:
+                seen.append(sub._queue.get_nowait())
+            except asyncio.QueueEmpty:
+                break
+        return result, seen
+    finally:
+        bus.unsubscribe(sub)
+
+
+class TestClassifierWithoutSTT:
+    """The classifier is spectral: it must not depend on STT at all."""
+
+    def _classifier(self):
+        from services.audio_classifier import AudioClassifier
+
+        return AudioClassifier(Settings(database_url="sqlite+aiosqlite:///:memory:").classifier)
+
+    def test_classify_takes_only_audio(self):
+        # A transcript parameter would make STT a hard dependency of hold
+        # detection — the thing this checkbox is about.
+        import inspect
+
+        params = set(inspect.signature(self._classifier().classify_chunk).parameters)
+        assert params == {"audio_data"}
+
+    async def test_classifies_with_no_stt_service_anywhere(self):
+        # Silence is the cheapest deterministic input; the point is that a
+        # classification is produced at all with no STT in the picture.
+        result = await self._classifier().classify(b"\x00\x00" * 16000)
+        assert result.audio_type is not None
+
+
+class TestTranscriptionDegradation:
+    async def test_hold_slayer_transcribe_returns_empty_on_failure(self):
+        gw = _gateway()
+        stt = MagicMock()
+        stt.transcribe = AsyncMock(side_effect=RuntimeError("Connection refused"))
+        svc = _hold_slayer(gw, stt)
+
+        text, errors = await _errors_for(gw.event_bus, svc._transcribe("call-1", b"\x00" * 320))
+
+        assert text == ""  # empty transcript, not an exception
+        assert len(errors) == 1
+        assert errors[0].data["service"] == "transcription"
+
+    async def test_error_event_names_the_service(self):
+        # "transcription failed" vs "the AI decided badly" — the whole reason
+        # transcribe() raises instead of swallowing.
+        gw = _gateway()
+        stt = MagicMock()
+        stt.transcribe = AsyncMock(side_effect=RuntimeError("Connection refused"))
+        svc = _hold_slayer(gw, stt)
+
+        _, errors = await _errors_for(gw.event_bus, svc._transcribe("call-1", b"\x00" * 320))
+        assert "Connection refused" in errors[0].data["error"]
+
+    async def test_service_error_survives_a_dead_event_bus(self):
+        # Degradation reporting must not itself become a failure path.
+        gw = _gateway()
+        gw.event_bus.publish = AsyncMock(side_effect=RuntimeError("bus down"))
+        stt = MagicMock()
+        stt.transcribe = AsyncMock(side_effect=RuntimeError("stt down"))
+        svc = _hold_slayer(gw, stt)
+
+        assert await svc._transcribe("call-1", b"\x00" * 320) == ""
+
+    async def test_transcription_marks_itself_unavailable(self):
+        # /health reads this flag; a failure that doesn't record itself makes
+        # the probe lie.
+        import httpx
+
+        from services.transcription import TranscriptionService
+
+        svc = TranscriptionService(Settings(database_url="sqlite+aiosqlite:///:memory:").speaches)
+        client = MagicMock()
+        client.post = AsyncMock(side_effect=httpx.ConnectError("refused"))
+        svc._client = client
+        svc._client.is_closed = False
+
+        with pytest.raises(Exception):
+            await svc.transcribe(b"\x00" * 320)
+        assert svc.available is False
+
+
+class TestReceptionistDegradation:
+    def _receptionist(self, gateway, **kw):
+        from services.receptionist import ReceptionistService
+
+        return ReceptionistService(gateway=gateway, **kw)
+
+    async def test_llm_failure_falls_back_to_a_usable_decision(self):
+        gw = _gateway()
+        svc = self._receptionist(gw)
+        call = MagicMock(id="call-1", remote_number="+15551234567")
+
+        llm = MagicMock()
+        llm.chat_json = AsyncMock(side_effect=RuntimeError("LLM down"))
+        import services.llm_client as llm_mod
+
+        original = llm_mod.get_llm
+        llm_mod.get_llm = lambda: llm
+        try:
+            result, errors = await _errors_for(
+                gw.event_bus, svc._classify(call, "I need to speak to someone", None)
+            )
+        finally:
+            llm_mod.get_llm = original
+
+        # A decision still comes back, so the call can proceed.
+        assert result["recommended_action"] in {"ring", "message", "reject"}
+        assert errors[0].data["service"] == "llm"
+
+    async def test_no_transcription_service_yields_empty_not_crash(self):
+        # transcription=None is a valid wiring (STT not configured).
+        gw = _gateway()
+        svc = self._receptionist(gw, transcription=None)
+        assert svc.transcription is None
+
+
+class TestHealthReportsDegradation:
+    """A degraded gateway must read as degraded — /health may not lie."""
+
+    def test_availability_helper_distinguishes_unknown_from_down(self):
+        # Four distinct states, because "not wired up" and "wired up but the
+        # remote is refusing connections" are different operator problems.
+        import main
+
+        assert main._availability(None) == "not attached"
+        assert main._availability(MagicMock(available=None)) == "unknown (no requests yet)"
+        assert main._availability(MagicMock(available=True)) == "ok"
+        assert main._availability(MagicMock(available=False)) == "unreachable"

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")