Bound unauthenticated SSO work: JWKS refreshes and pending logins

src/cerberus/identity/session_store.py

@@ -33,6 +33,11 @@ class AdminSession:
 # nonce, verifier, expires — the PKCE/replay material stashed between login start and callback
 Pending = tuple[str, str, float]
 
+# /admin/login/start is unauthenticated and each call stores a pending row that
+# lives for the login TTL. Past this many, the soonest-expiring rows are evicted
+# so a request loop cannot grow memory or the state file without bound.
+MAX_PENDING = 10_000
+
 
 class SessionStore(Protocol):
     def put_pending(self, state: str, nonce: str, verifier: str, expires: float) -> None: ...
@@ -51,6 +56,10 @@ def __init__(self) -> None:
         self._sessions: dict[str, AdminSession] = {}
 
     def put_pending(self, state: str, nonce: str, verifier: str, expires: float) -> None:
+        current = time.time()
+        self._pending = {s: v for s, v in self._pending.items() if v[2] > current}  # gc
+        while len(self._pending) >= MAX_PENDING:
+            del self._pending[min(self._pending, key=lambda s: self._pending[s][2])]
         self._pending[state] = (nonce, verifier, expires)
 
     def pop_pending(self, state: str, *, now: float | None = None) -> Pending | None:
@@ -121,6 +130,12 @@ def put_pending(self, state: str, nonce: str, verifier: str, expires: float) ->
                 " verifier=excluded.verifier, expires=excluded.expires",
                 (state, nonce, verifier, expires),
             )
+            self._connection.execute("DELETE FROM oidc_pending WHERE expires <= ?", (time.time(),))  # gc
+            self._connection.execute(
+                "DELETE FROM oidc_pending WHERE state IN (SELECT state FROM oidc_pending"
+                " ORDER BY expires DESC LIMIT -1 OFFSET ?)",
+                (MAX_PENDING,),
+            )
             self._connection.commit()
 
     def pop_pending(self, state: str, *, now: float | None = None) -> Pending | None:

src/cerberus/identity/sso.py

@@ -53,6 +53,10 @@ def __init__(
         self._http = httpx.AsyncClient(transport=transport, timeout=8.0, verify=verify)
         self._keys: dict[str, jwt.PyJWK] = {}
         self._fetched = 0.0
+        # _validate runs for any bearer on any /admin/* request before any other
+        # check, so an unknown kid must not buy an outbound JWKS fetch per request
+        self._refresh_attempted_at = 0.0
+        self._min_refresh_interval = 10.0
         # pending logins + sessions live in a shared store so they survive across
         # workers and restarts; the in-memory default is single-worker only
         self._store: SessionStore = store if store is not None else InMemorySessionStore()
@@ -187,7 +191,10 @@ async def _refresh_keys(self) -> None:
         self._fetched = time.time()
 
     async def _key_for(self, kid: str) -> jwt.PyJWK | None:
-        if kid not in self._keys or time.time() - self._fetched > 300:
+        now = time.time()
+        wants_refresh = kid not in self._keys or now - self._fetched > 300
+        if wants_refresh and now - self._refresh_attempted_at >= self._min_refresh_interval:
+            self._refresh_attempted_at = now
             try:
                 await self._refresh_keys()
             except (httpx.HTTPError, ValueError):

tests/test_admin_sso.py

@@ -229,3 +229,35 @@ async def test_docs_and_openapi_follow_the_admin_boundary(monkeypatch, tmp_path)
             # and the schema is real, not a stub
             schema = await client.get("/openapi.json", headers={"authorization": f"Bearer {break_glass()}"})
             assert "/v1/chat/completions" in schema.json()["paths"]
+
+
+@pytest.mark.asyncio
+async def test_unknown_kids_do_not_hammer_the_jwks_endpoint(monkeypatch, tmp_path):
+    """A bearer with a made-up kid on any /admin/* request must not buy an
+    outbound JWKS fetch each time; refreshes are rate-limited."""
+
+    monkeypatch.setenv("ALPHA_KEY", "alpha-secret")
+    monkeypatch.setenv("CB_OIDC_ID", "cerberus")
+    monkeypatch.setenv("CB_OIDC_SECRET", "shh")
+    monkeypatch.setenv("CB_SESSION_SECRET", "session-signing-secret")
+    raw = raw_config("cerberus-2026-07-16.1")
+    raw["server"] = {"host": "127.0.0.1", "port": 4000}
+    raw["admin_sso"] = sso_block()
+    doc = load_config_document(write_config(tmp_path, "v.yaml", raw))
+    fetches = []
+
+    def sso_handler(request: httpx.Request) -> httpx.Response:
+        fetches.append(request.url.path)
+        return httpx.Response(200, json={"keys": [_JWK]})
+
+    app = create_app(
+        doc, http_transport=httpx.MockTransport(ok_upstream), sso_transport=httpx.MockTransport(sso_handler)
+    )
+    forged = jwt.encode({"sub": "x"}, _KEY, algorithm="RS256", headers={"kid": "unknown-kid"})
+    async with app.router.lifespan_context(app):
+        transport = httpx.ASGITransport(app=app, client=("203.0.113.9", 40001))
+        async with httpx.AsyncClient(transport=transport, base_url="http://cerberus.example") as client:
+            for _ in range(20):
+                response = await client.get("/admin/status", headers={"authorization": f"Bearer {forged}"})
+                assert response.status_code == 401
+    assert len(fetches) == 1

tests/test_session_store.py

@@ -79,3 +79,25 @@ def test_sqlite_session_survives_reopen(tmp_path):
         assert reopened.get_session("sid") is not None
     finally:
         reopened.close()
+
+
+def test_pending_logins_are_capped(store, monkeypatch):
+    """/admin/login/start is unauthenticated; a request loop must not grow the
+    pending table without bound. The newest entry always survives."""
+
+    import cerberus.identity.session_store as session_store
+
+    monkeypatch.setattr(session_store, "MAX_PENDING", 3)
+    now = time.time()
+    for i in range(5):
+        store.put_pending(f"s{i}", "n", "v", now + 100 + i)
+    assert store.pop_pending("s0") is None
+    assert store.pop_pending("s1") is None
+    assert store.pop_pending("s4") is not None
+
+
+def test_put_pending_sweeps_expired_rows(store):
+    store.put_pending("old", "n", "v", time.time() - 1)
+    store.put_pending("new", "n", "v", time.time() + 100)
+    assert store.pop_pending("new") is not None
+    assert store.pop_pending("old") is None