Admin auth: harden JWKS cache against outage amplification (P2) and corrupt/unknown-key handling (P3)

- P2: verify_admin_token parses and enforces the JWT header (alg ==
  RS256, kid present) BEFORE consulting the JWKS cache, so a
  structurally invalid token costs no fetch, no lock, no provider load.
- P2: JwksCache gains a short negative-cache backoff window (5s,
  time.monotonic-recorded) after a failed fetch: within it, get()
  re-raises the recorded fault without touching the fetcher — and
  without holding or queueing on the lock — so an outage costs one
  fetch per window instead of one per request. A successful fetch
  clears the backoff; nothing is cached indefinitely (TTL unchanged).
- P3: _matching_key widens the from_jwk handling to
  (PyJWTError, binascii.Error, ValueError, TypeError, KeyError) so a
  JWKS entry matching the kid but carrying corrupt n/e maps to the
  SAME uniform 401 instead of surfacing a 500.
- P3: an unmatched kid now forces ONE TTL-bypassing refetch and
  retries the match once, so rotated keys verify without waiting out
  the TTL; the forced refetch is itself bounded (inside the backoff
  window or the 5s kid-refetch interval the cached document is served
  as-is), so unknown-kid floods cannot force a fetch per request.
- Tests: no-fetch proofs for garbage/non-RS256 tokens, corrupt-entry
  401, outage fail-fast (sequential + concurrent burst), backoff
  lapse/recovery, outage-as-server-fault, and the unknown-kid
  refetch: one forced, throttled, resume-after-interval, rotation
  picked up, failed forced refetch propagates then fails fast.
  Existing tests unchanged (uniform 401 contract intact).
This commit is contained in:
2026-09-23 00:30:04 -04:00
parent 3cdedb37c8
commit 80ac786a54
2 changed files with 494 additions and 30 deletions
+175 -30
View File
@@ -20,12 +20,20 @@ Every rejection raises the SAME :class:`~fastapi.HTTPException` with
the same detail — nothing in the response distinguishes which check
failed. The check FAILS CLOSED: a missing app config, a malformed or
missing ``Authorization`` header, a non-RS256 algorithm, an unknown
``kid``, and any PyJWT verification error all yield the one 401. A
JWKS FETCH failure is deliberately NOT mapped to 401: the provider
being unreachable is a server-side fault (an HTTP 500 surfaces) and
never poisons the cache, so the plane stays closed while the real
cause is visible in the server log instead of disguised as a bad
credential.
``kid``, and any PyJWT verification error all yield the one 401.
The token's header (``alg``, ``kid``) is parsed and checked BEFORE the
JWKS cache is consulted, so structurally invalid tokens can never
amplify load on the provider. An unknown ``kid`` forces ONE
TTL-bypassing refetch (throttled, see :class:`JwksCache`) so a rotated
key is picked up without waiting out the TTL, and a JWKS entry that
matches the ``kid`` but carries corrupt key material maps to the same
401. A JWKS FETCH failure is deliberately NOT mapped to 401: the
provider being unreachable is a server-side fault (an HTTP 500
surfaces) and never poisons the cache, so the plane stays closed while
the real cause is visible in the server log instead of disguised as a
bad credential; a short backoff window after a failed fetch fails fast
(re-raising the recorded fault) rather than fetching once per request.
This module validates the TOKEN only — it is authentication, not
authorization. Membership in the ``nestquest-admins`` group is the
@@ -35,6 +43,7 @@ claims are already its return value, so no rewrite is needed).
from __future__ import annotations
import asyncio
import binascii
import json
import time
from collections.abc import Awaitable, Callable
@@ -66,6 +75,18 @@ _WWW_AUTHENTICATE = {"WWW-Authenticate": "Bearer"}
#: verifying while staying far above a per-request fetch.
_JWKS_TTL_SECONDS = 300.0
#: Seconds a FAILED JWKS fetch blocks further fetch attempts. Within
#: this window after a failure the cache fails fast — it re-raises the
#: recorded fault instead of touching the provider — so an outage is
#: never amplified into one fetch per request.
_JWKS_BACKOFF_SECONDS = 5.0
#: Seconds that must pass before ANOTHER forced (TTL-bypassing) refetch
#: may run. An unmatched ``kid`` forces one refetch to pick up rotated
#: keys; this interval bounds it, so a flood of unknown-``kid`` tokens
#: cannot turn each request into a provider fetch either.
_JWKS_KID_REFETCH_SECONDS = 5.0
#: Seconds before the real JWKS fetch gives up on the provider.
_JWKS_TIMEOUT_SECONDS = 5.0
@@ -101,8 +122,22 @@ class JwksCache:
— the dependency NEVER fetches per request. A single fetch is
shared between concurrent requests via an :class:`asyncio.Lock`
(double-checked around it), and a failed fetch is never cached, so
the next request retries rather than being locked out until the TTL
lapses.
the next request retries once the backoff window lapses rather than
being locked out until the TTL does.
Two bounds keep the provider out of an amplification loop:
- ``backoff_seconds``: after a FAILED fetch, :meth:`get` fails fast
for this long — re-raising the recorded fault WITHOUT touching
the fetcher or even the lock — instead of retrying per request.
- ``kid_refetch_seconds``: forced (TTL-bypassing) refetches — the
unmatched-``kid`` path — may run at most this often; a forced
refetch that is bounded out (or one inside the backoff window)
is served from the cached document as-is.
A successful fetch always clears the backoff; the document itself
is still trusted only for ``ttl_seconds`` — nothing is cached
indefinitely.
One instance lives on ``app.state.jwks_cache`` (installed by
:func:`api.app.create_app`); tests replace it with one built around
@@ -114,30 +149,67 @@ class JwksCache:
url: str,
*,
ttl_seconds: float = _JWKS_TTL_SECONDS,
backoff_seconds: float = _JWKS_BACKOFF_SECONDS,
kid_refetch_seconds: float = _JWKS_KID_REFETCH_SECONDS,
fetcher: JwksFetcher | None = None,
) -> None:
self._url = url
self._ttl_seconds = ttl_seconds
self._backoff_seconds = backoff_seconds
self._kid_refetch_seconds = kid_refetch_seconds
self._fetcher: JwksFetcher = (
fetcher if fetcher is not None else _fetch_jwks
)
self._document: dict[str, object] | None = None
self._fetched_at: float | None = None
self._failed_until: float | None = None
self._last_error: BaseException | None = None
self._forced_at: float | None = None
self._lock = asyncio.Lock()
async def get(self) -> dict[str, object]:
"""Return the JWKS document, refetching it if the TTL lapsed."""
fresh = self._fresh_document()
if fresh is not None:
return fresh
async with self._lock:
async def get(self, *, force: bool = False) -> dict[str, object]:
"""Return the JWKS document, refetching it if the TTL lapsed.
With ``force=True`` the TTL is bypassed so an unmatched ``kid``
can pick up a rotated key without waiting it out; the forced
refetch is itself bounded (see the class docstring): inside the
backoff window or the ``kid_refetch_seconds`` interval the
cached document is served as-is instead. A failed fetch is
re-raised to the caller (a provider outage is a server fault,
never a 401) and starts the backoff window, during which every
caller — including forced ones with nothing cached — fails fast
on the recorded fault.
"""
if not force:
fresh = self._fresh_document()
if fresh is not None:
return fresh
document = await self._fetcher(self._url)
self._document = document
self._fetched_at = time.monotonic()
return document
error = self._backoff_error()
if error is not None:
# Fail fast OUTSIDE the lock: a request inside the backoff
# window must not queue behind an in-flight fetch.
if force:
document = self._document
if document is not None:
return document
raise error
async with self._lock:
if not force:
fresh = self._fresh_document()
if fresh is not None:
return fresh
error = self._backoff_error()
if error is not None:
if force:
document = self._document
if document is not None:
return document
raise error
if force and self._refetch_throttled():
document = self._document
if document is not None:
return document
return await self._fetch_locked(force=force)
def _fresh_document(self) -> dict[str, object] | None:
"""The cached document if one exists and is inside its TTL."""
@@ -151,6 +223,48 @@ class JwksCache:
return document
return None
def _backoff_error(self) -> BaseException | None:
"""The recorded fetch fault while the backoff window is open."""
failed_until = self._failed_until
last_error = self._last_error
if (
failed_until is not None
and last_error is not None
and time.monotonic() < failed_until
):
return last_error
return None
def _refetch_throttled(self) -> bool:
"""Whether a forced refetch ran too recently to run another."""
forced_at = self._forced_at
return forced_at is not None and (
time.monotonic() - forced_at
) < self._kid_refetch_seconds
async def _fetch_locked(self, *, force: bool) -> dict[str, object]:
"""Run ONE fetch attempt (caller holds the lock) and record it.
A failed fetch starts the ``backoff_seconds`` window and is
re-raised; a successful one stores the document, restarts the
TTL and clears the backoff. A forced attempt stamps
``_forced_at`` so the ``kid_refetch_seconds`` interval bounds
how often forced refetches may run back-to-back.
"""
if force:
self._forced_at = time.monotonic()
try:
document = await self._fetcher(self._url)
except Exception as error:
self._failed_until = time.monotonic() + self._backoff_seconds
self._last_error = error
raise
self._document = document
self._fetched_at = time.monotonic()
self._failed_until = None
self._last_error = None
return document
def _reject() -> HTTPException:
"""Build the single 401 the dependency raises for every failure."""
@@ -161,14 +275,18 @@ def _reject() -> HTTPException:
)
def _matching_key(jwks: dict[str, object], kid: str) -> object:
def _matching_key(jwks: dict[str, object], kid: str) -> object | None:
"""Return the RSA signing key named ``kid`` from the JWKS.
Only ``kty: RSA`` entries are considered (the pinned RS256 scheme
is asymmetric RSA); anything else — a missing ``keys`` list, no
entry with that ``kid`` — raises the one 401 rather than falling
back to "some" key. The key is built through PyJWT's JWKS decoder;
malformed key data raises there and is caught by the caller.
entry with that ``kid`` — returns ``None`` rather than falling back
to "some" key. The key is built through PyJWT's JWKS decoder, and
an entry that MATCHES the ``kid`` but carries corrupt key material
(a malformed ``n``/``e`` raises ``ValueError`` — ``binascii.Error``
is its subclass — or ``TypeError`` instead of a ``PyJWTError``) is
mapped to the one 401 like every other rejection: the caller must
never see a half-built key.
"""
keys = jwks.get("keys")
if not isinstance(keys, list):
@@ -178,8 +296,17 @@ def _matching_key(jwks: dict[str, object], kid: str) -> object:
continue
if entry.get("kid") != kid or entry.get("kty") != "RSA":
continue
return jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(entry))
raise _reject()
try:
return jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(entry))
except (
jwt.PyJWTError,
binascii.Error,
ValueError,
TypeError,
KeyError,
):
raise _reject() from None
return None
async def verify_admin_token(
@@ -198,6 +325,13 @@ async def verify_admin_token(
``exp`` is REQUIRED (a token that never expires is never accepted)
and PyJWT enforces ``nbf`` when the token carries it.
The token's HEADER (``alg``, ``kid``) is parsed and enforced BEFORE
the JWKS cache is consulted, so a structurally invalid token costs
no fetch, no lock, no provider load. When the cached document has
no entry for the ``kid``, ONE forced (TTL-bypassing, throttled)
refetch is made and the match retried once — a rotated key is
picked up without waiting out the TTL — before the 401.
A JWKS fetch failure propagates (see the module docstring): the
provider being down is a server fault, not a bad credential.
"""
@@ -212,15 +346,26 @@ async def verify_admin_token(
cache = getattr(request.app.state, "jwks_cache", None)
if cache is None:
raise _reject()
jwks = await cache.get()
try:
header = jwt.get_unverified_header(token)
if header.get("alg") != _ALLOWED_ALGORITHM:
raise _reject()
kid = header.get("kid")
if not isinstance(kid, str) or not kid:
raise _reject()
except jwt.PyJWTError:
raise _reject() from None
if header.get("alg") != _ALLOWED_ALGORITHM:
raise _reject()
kid = header.get("kid")
if not isinstance(kid, str) or not kid:
raise _reject()
jwks = await cache.get()
try:
key = _matching_key(jwks, kid)
if key is None:
# The kid is not in the cached document: force ONE
# TTL-bypassing refetch (bounded by the cache) and retry
# the match once before rejecting.
jwks = await cache.get(force=True)
key = _matching_key(jwks, kid)
if key is None:
raise _reject()
claims = jwt.decode(
token,
key=key,
+319
View File
@@ -12,9 +12,17 @@ JWKS fetcher — NO network access:
token returns 200 with the verified subject.
- Every 401 carries the SAME detail — nothing in the response
distinguishes which check failed.
- The token's HEADER is validated BEFORE the JWKS cache is touched: a
garbage or non-RS256 token triggers NO fetch at all.
- A JWKS entry that matches the ``kid`` but carries corrupt key
material maps to the same uniform 401.
- The JWKS document is fetched ONCE through the app's
:class:`api.auth.JwksCache` and then reused across requests (the
stub fetcher counts calls); a zero TTL refetches.
- A JWKS OUTAGE fails fast: after one failed fetch the cache re-raises
the fault for the whole backoff window without further fetches, even
under a burst of concurrent requests; an unknown ``kid`` forces one
TTL-bypassing refetch, throttled to at most one per interval.
``ApiConfig.from_env`` requires the three OIDC variables
(``NESTQUEST_OIDC_ISSUER``, ``NESTQUEST_OIDC_AUDIENCE``,
@@ -29,7 +37,9 @@ case signs with a SECOND locally generated key pair.
"""
from __future__ import annotations
import asyncio
import base64
import json
import time
from pathlib import Path
@@ -105,6 +115,45 @@ class _StubbedFetch:
return self.document
class _FailingFetch:
"""A JWKS fetcher stub that always fails, counting the attempts."""
def __init__(self) -> None:
self.calls = 0
async def __call__(self, url: str) -> dict[str, object]:
self.calls += 1
raise httpx.ConnectError("provider unreachable")
class _FlakyFetch:
"""A fetcher that succeeds ONCE, then fails on every later call."""
def __init__(self, document: dict[str, object]) -> None:
self.document = document
self.calls = 0
async def __call__(self, url: str) -> dict[str, object]:
self.calls += 1
if self.calls == 1:
return self.document
raise httpx.ConnectError("provider unreachable")
class _FailThenServe:
"""A fetcher that fails the FIRST call, then serves the document."""
def __init__(self, document: dict[str, object]) -> None:
self.document = document
self.calls = 0
async def __call__(self, url: str) -> dict[str, object]:
self.calls += 1
if self.calls == 1:
raise httpx.ConnectError("provider unreachable")
return self.document
def _make_token(
signing_key: rsa.RSAPrivateKey = ADMIN_KEY,
*,
@@ -130,6 +179,33 @@ def _make_token(
)
def _token_with_header(
header: dict[str, object], payload: dict[str, object]
) -> str:
"""Build a JWT with an EXACT header (bypassing jwt.encode's rules).
Used to prove header checks (``alg``, ``kid``) happen before the
JWKS cache is consulted; the signature segment is never reached.
"""
def part(obj: dict[str, object]) -> str:
raw = json.dumps(obj).encode("utf-8")
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
return f"{part(header)}.{part(payload)}.c2lnbmF0dXJl"
def _verified_payload() -> dict[str, object]:
"""Claims that would verify — the header checks fire before them."""
now = int(time.time())
return {
"iss": ISSUER,
"aud": AUDIENCE,
"sub": SUBJECT,
"exp": now + 300,
"nbf": now - 1,
}
class _AdminRunner:
"""Run the app's lifespan around an httpx AsyncClient.
@@ -148,6 +224,8 @@ class _AdminRunner:
jwks_document: dict[str, object],
*,
ttl_seconds: float | None = None,
backoff_seconds: float | None = None,
kid_refetch_seconds: float | None = None,
) -> None:
self.app = create_app(
ApiConfig(
@@ -162,6 +240,10 @@ class _AdminRunner:
cache_kwargs: dict[str, float] = {}
if ttl_seconds is not None:
cache_kwargs["ttl_seconds"] = ttl_seconds
if backoff_seconds is not None:
cache_kwargs["backoff_seconds"] = backoff_seconds
if kid_refetch_seconds is not None:
cache_kwargs["kid_refetch_seconds"] = kid_refetch_seconds
self.app.state.jwks_cache = JwksCache(
JWKS_URL, fetcher=self.fetch, **cache_kwargs
)
@@ -359,6 +441,70 @@ async def test_bad_signature_is_unauthorized(
assert response.status_code == 401
async def test_corrupt_jwks_entry_matching_kid_is_unauthorized(
temp_db_path: str,
) -> None:
"""A JWKS entry matching the kid but with corrupt key data → 401.
``RSAAlgorithm.from_jwk`` raises ValueError/TypeError (NOT a
PyJWTError) on such an entry; it must map to the SAME uniform 401,
not surface as a server error.
"""
corrupt_jwks: dict[str, object] = {
"keys": [
{"kty": "RSA", "kid": KID, "n": "AQ", "e": "AQAB"},
]
}
runner = _AdminRunner(temp_db_path, corrupt_jwks)
async with runner as client:
response = await client.get(
"/api/v1/admin/ping",
headers={"Authorization": f"Bearer {_make_token()}"},
)
assert response.status_code == 401
assert response.json()["detail"] == REJECTION_DETAIL
# --- the header is checked BEFORE the JWKS cache -----------------------------
async def test_garbage_token_triggers_no_jwks_fetch(
temp_db_path: str,
) -> None:
"""A token that is not a JWT costs NO fetch, lock or provider load."""
runner = _AdminRunner(temp_db_path, _jwks_for(ADMIN_KEY, KID))
async with runner as client:
response = await client.get(
"/api/v1/admin/ping",
headers={"Authorization": "Bearer not-a-jwt"},
)
assert response.status_code == 401
assert response.json()["detail"] == REJECTION_DETAIL
assert runner.fetch.calls == 0
async def test_non_rs256_token_triggers_no_jwks_fetch(
temp_db_path: str,
) -> None:
"""A structurally valid JWT with a non-RS256 alg is 401 pre-cache.
The header carries a kid pointing at the published key, so ONLY the
early alg check can reject it — the fetcher must never be called.
"""
runner = _AdminRunner(temp_db_path, _jwks_for(ADMIN_KEY, KID))
token = _token_with_header(
{"alg": "HS256", "typ": "JWT", "kid": KID}, _verified_payload()
)
async with runner as client:
response = await client.get(
"/api/v1/admin/ping",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 401
assert response.json()["detail"] == REJECTION_DETAIL
assert runner.fetch.calls == 0
# --- the uniform 401 contract -----------------------------------------------
@@ -431,6 +577,179 @@ async def test_expired_ttl_refetches_the_jwks(temp_db_path: str) -> None:
assert runner.fetch.calls == 2
# --- the JWKS cache under an outage: fail fast, don't amplify ---------------
async def test_jwks_outage_fails_fast_during_backoff() -> None:
"""After one failed fetch, later requests fail fast for the window."""
fetch = _FailingFetch()
cache = JwksCache(JWKS_URL, fetcher=fetch)
for _ in range(4):
with pytest.raises(httpx.ConnectError):
await cache.get()
assert fetch.calls == 1
async def test_jwks_outage_burst_calls_the_fetcher_once() -> None:
"""A burst of concurrent requests costs ONE fetch, not one each.
The first caller fetches and fails; every other caller — queued on
the lock or arriving inside the backoff window — fails fast on the
recorded fault without touching the fetcher.
"""
fetch = _FailingFetch()
cache = JwksCache(JWKS_URL, fetcher=fetch)
results = await asyncio.gather(
*(cache.get() for _ in range(6)), return_exceptions=True
)
assert fetch.calls == 1
assert all(
isinstance(result, httpx.ConnectError) for result in results
)
async def test_jwks_backoff_lapses_and_recovers() -> None:
"""A backoff window is not a lockout: retries resume after it.
With a zero-length window the very next request retries (and the
successful document is then served from the cache), proving a
failed state is never cached for good.
"""
document = _jwks_for(ADMIN_KEY, KID)
fetch = _FailThenServe(document)
cache = JwksCache(JWKS_URL, fetcher=fetch, backoff_seconds=0.0)
with pytest.raises(httpx.ConnectError):
await cache.get()
assert await cache.get() == document
assert await cache.get() == document
assert fetch.calls == 2
async def test_jwks_outage_is_a_server_fault_not_a_401(
temp_db_path: str,
) -> None:
"""A provider outage propagates as a fault, never as a 401."""
runner = _AdminRunner(temp_db_path, _jwks_for(ADMIN_KEY, KID))
runner.app.state.jwks_cache = JwksCache(JWKS_URL, fetcher=_FailingFetch())
async with runner as client:
with pytest.raises(httpx.ConnectError):
await client.get(
"/api/v1/admin/ping",
headers={"Authorization": f"Bearer {_make_token()}"},
)
# --- the unknown-kid refetch: one forced, bounded attempt --------------------
async def test_unknown_kid_refetch_picks_up_a_rotated_key(
temp_db_path: str,
) -> None:
"""An unmatched kid forces ONE refetch, and the retry can verify.
The JWKS first publishes a DIFFERENT kid: the token is 401 after
exactly two fetches (normal + forced). When the rotated key lands
in the provider's document, the NEXT token forces another refetch
that picks it up — the same token verifies WITHOUT waiting out the
TTL.
"""
runner = _AdminRunner(
temp_db_path, _jwks_for(ADMIN_KEY, "other-key"), kid_refetch_seconds=0.0
)
token = _make_token()
async with runner as client:
response = await client.get(
"/api/v1/admin/ping",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 401
assert response.json()["detail"] == REJECTION_DETAIL
assert runner.fetch.calls == 2
# The rotation lands at the provider (same stub fetcher).
runner.fetch.document = _jwks_for(ADMIN_KEY, KID)
response = await client.get(
"/api/v1/admin/ping",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 200
assert runner.fetch.calls == 3
async def test_unknown_kid_refetch_is_throttled(temp_db_path: str) -> None:
"""Back-to-back unknown-kid tokens force at most one refetch.
The interval bounds forced refetches: the second unknown-kid token
inside it is served from the cached document (still unmatched →
401) with no further fetch.
"""
runner = _AdminRunner(temp_db_path, _jwks_for(ADMIN_KEY, "other-key"))
token = _make_token()
async with runner as client:
for _ in range(2):
response = await client.get(
"/api/v1/admin/ping",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 401
assert runner.fetch.calls == 2
async def test_unknown_kid_refetch_resumes_after_the_interval(
temp_db_path: str,
) -> None:
"""A lapsed interval allows the next forced refetch again.
With a zero-length interval every request may force its refetch:
the first costs one normal fetch plus one forced (two), and the
second — its cached document still fresh — costs just the forced
one (three in total), proving the throttle is a bounded interval
rather than a permanent lockout.
"""
runner = _AdminRunner(
temp_db_path, _jwks_for(ADMIN_KEY, "other-key"), kid_refetch_seconds=0.0
)
token = _make_token()
async with runner as client:
for _ in range(2):
response = await client.get(
"/api/v1/admin/ping",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 401
assert runner.fetch.calls == 3
async def test_failed_forced_refetch_propagates_then_fails_fast(
temp_db_path: str,
) -> None:
"""A failed forced refetch is a fault, and its backoff bounds the rest.
First request: cache fills (1 fetch), the kid is unmatched, the
forced refetch FAILS (2 fetches) and propagates as the server
fault. Second request inside the backoff window: the cached
document is served, the match fails, and the token is 401 — with
NO third fetch.
"""
runner = _AdminRunner(temp_db_path, _jwks_for(ADMIN_KEY, KID))
runner.app.state.jwks_cache = JwksCache(
JWKS_URL, fetcher=_FlakyFetch(_jwks_for(ADMIN_KEY, "other-key"))
)
token = _make_token()
async with runner as client:
with pytest.raises(httpx.ConnectError):
await client.get(
"/api/v1/admin/ping",
headers={"Authorization": f"Bearer {token}"},
)
response = await client.get(
"/api/v1/admin/ping",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 401
assert response.json()["detail"] == REJECTION_DETAIL
# --- ApiConfig.from_env: the OIDC settings -----------------------------------