diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 820852e..eef1eb8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -80,7 +80,9 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.11", "3.12", "3.13", "3.14", "3.15"] + python-version: ["3.11", "3.12", "3.13", "3.14", "3.15", "3.14t", "3.15t"] + env: + UV_PYTHON: ${{ matrix.python-version }} steps: - uses: actions/checkout@v7 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index af9cdac..dfff7f3 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -26,7 +26,7 @@ uv run ty check uv run pytest ``` -CI also runs the tests on Python 3.11 to 3.15, and against the oldest FastAPI release the project supports (0.123.0). +CI also runs the tests on Python 3.11 to 3.15, on the free-threaded 3.14t and 3.15t builds, and against the oldest FastAPI release the project supports (0.123.0). ## Tests diff --git a/benchmarks/test_perf.py b/benchmarks/test_perf.py index cb7aa87..5de3e56 100644 --- a/benchmarks/test_perf.py +++ b/benchmarks/test_perf.py @@ -221,7 +221,7 @@ def test_jwt_es256_jwks_validate(benchmark): def test_introspection_cache_hit(benchmark): - """Opaque token already introspected: the hash and LRU lookup every request pays.""" + """Opaque token already introspected: the hash and cache lookup every request pays.""" def provider(request: httpx.Request) -> httpx.Response: return httpx.Response( diff --git a/docs/security-model.md b/docs/security-model.md index 30062a0..6b097cc 100644 --- a/docs/security-model.md +++ b/docs/security-model.md @@ -35,6 +35,7 @@ What the library takes care of, and what stays with you. To report a vulnerabili - **ID tokens are untrusted until validated** with `oauth_validate_id_token`. - **Introspection answers are cached.** A token revoked at the provider keeps working until its cached answer expires, `cache_ttl` seconds at most. - **The discovery URL is configuration.** Never build it from request input. +- **Sources and validators can be shared across threads.** One instance can serve several threads, each with its own event loop, as on a free-threaded Python build. Requests on one loop share a provider fetch in flight; another loop makes its own. Your validators and `replay_check` run concurrently too, so keep them thread-safe. ## Status codes diff --git a/pyproject.toml b/pyproject.toml index 5c45b69..eedc984 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", + "Programming Language :: Python :: Free Threading :: 3 - Stable", "Topic :: Internet :: WWW/HTTP :: Session", "Topic :: Security", "Topic :: Software Development :: Libraries :: Python Modules", @@ -30,7 +31,6 @@ classifiers = [ ] dependencies = [ "anyio>=3.7", - "async-lru>=2.0", "fastapi>=0.123.0,!=0.137.0,!=0.137.1", "httpx>=0.25.0", "itsdangerous>=2.1", diff --git a/src/fastapi_multiauth/abc.py b/src/fastapi_multiauth/abc.py index 572deb8..8d9a59d 100644 --- a/src/fastapi_multiauth/abc.py +++ b/src/fastapi_multiauth/abc.py @@ -208,10 +208,13 @@ def _carry_scheme(dependency: Any, scheme: SecurityBase) -> None: cls = type(dependency) promoted = _SCHEME_CLASSES.get(cls) if promoted is None: - promoted = _SCHEME_CLASSES[cls] = type( - cls.__name__, - (cls, SecurityBase), - {"__module__": cls.__module__, "__doc__": cls.__doc__}, + promoted = _SCHEME_CLASSES.setdefault( + cls, + type( + cls.__name__, + (cls, SecurityBase), + {"__module__": cls.__module__, "__doc__": cls.__doc__}, + ), ) dependency.__class__ = promoted dependency.model = scheme.model diff --git a/src/fastapi_multiauth/introspection.py b/src/fastapi_multiauth/introspection.py index 5186fd3..c8f4792 100644 --- a/src/fastapi_multiauth/introspection.py +++ b/src/fastapi_multiauth/introspection.py @@ -4,18 +4,18 @@ import copy import logging import time -from collections import OrderedDict from collections.abc import Callable, Sequence from typing import Any, ClassVar from urllib.parse import quote_plus -import anyio import httpx from ._http import DEFAULT_TIMEOUT, _request_json, _require_https from .exceptions import UnauthorizedError from .jwt import AnyAudience from .utils import ( + SingleFlight, + TTLCache, ensure_async, finite_number, hash_token, @@ -53,15 +53,6 @@ def copy(self) -> dict[str, Any]: return copied -class _Lookup: - """One introspection in flight, awaited by every request carrying its token.""" - - def __init__(self) -> None: - self.done = anyio.Event() - self.answer: _Answer | None = None - self.error: Exception | None = None - - def _basic_authorization(client_id: str, client_secret: str) -> str: """`client_secret_basic`: both halves form-encoded before base64 (RFC 6749 ยง2.3.1).""" pair = f"{quote_plus(client_id, safe='')}:{quote_plus(client_secret, safe='')}" @@ -153,9 +144,8 @@ def __init__( ensure_async(claims_to_identity) if claims_to_identity else None ) self._cache_ttl = cache_ttl - self._cache_size = cache_size - self._cache: OrderedDict[str, tuple[float, _Answer]] = OrderedDict() - self._lookups: dict[str, _Lookup] = {} + self._cache = TTLCache(cache_size) + self._lookups = SingleFlight() self._timeout = timeout self._client = client @@ -168,7 +158,9 @@ async def __call__(self, token: str, scopes: list[str] | None = None) -> Any: HTTPException: 503 if the introspection endpoint cannot be used. """ key = hash_token(token) - answer = self._cached(key) or await self._lookup(token, key) + answer = self._cache.get(key) or await self._lookups.run( + key, lambda: self._lookup(token, key) + ) claims = answer.copy() if scopes: require_scopes(claims, self._scopes_claim, scopes) @@ -178,26 +170,11 @@ async def __call__(self, token: str, scopes: list[str] | None = None) -> Any: return claims async def _lookup(self, token: str, key: str) -> _Answer: - """The answer for `token`, from the introspection in flight or a new one.""" - while (lookup := self._lookups.get(key)) is not None: - await lookup.done.wait() - if lookup.error is not None: - raise copy.copy(lookup.error) - if lookup.answer is not None: - return lookup.answer - lookup = self._lookups[key] = _Lookup() - try: - claims = await self._introspect(token) - self._check_claims(claims) - answer = _Answer(claims) - self._store(key, answer) - lookup.answer = answer - except Exception as exc: - lookup.error = exc - raise - finally: - del self._lookups[key] - lookup.done.set() + """Introspect `token`, check its claims and cache the answer.""" + claims = await self._introspect(token) + self._check_claims(claims) + answer = _Answer(claims) + self._store(key, answer) return answer async def _introspect(self, token: str) -> dict[str, Any]: @@ -258,27 +235,11 @@ def _check_claims(self, claims: dict[str, Any]) -> None: if self._issuer is not None and claims.get("iss") != self._issuer: raise refusal(UnauthorizedError(), "invalid_issuer") - def _cached(self, key: str) -> _Answer | None: - """A cached answer still within its lifetime, `None` otherwise.""" - entry = self._cache.get(key) - if entry is None: - return None - expires_at, answer = entry - if time.monotonic() >= expires_at: - del self._cache[key] - return None - self._cache.move_to_end(key) - return answer - def _store(self, key: str, answer: _Answer) -> None: """Cache an active answer for `cache_ttl`, never past the token's `exp`.""" ttl = self._cache_ttl exp = finite_number(answer.claims.get("exp")) if exp is not None: ttl = min(ttl, exp - time.time()) - if ttl <= 0: - return - self._cache[key] = (time.monotonic() + ttl, answer) - self._cache.move_to_end(key) - while len(self._cache) > self._cache_size: - self._cache.popitem(last=False) + if ttl > 0: + self._cache.put(key, answer, ttl) diff --git a/src/fastapi_multiauth/jwt.py b/src/fastapi_multiauth/jwt.py index c4bb299..765f33e 100644 --- a/src/fastapi_multiauth/jwt.py +++ b/src/fastapi_multiauth/jwt.py @@ -2,16 +2,18 @@ import enum import logging +import threading import time from collections.abc import Callable, Sequence from typing import Any, ClassVar -import anyio +from anyio.lowlevel import current_token from ._http import DEFAULT_TIMEOUT, _get_json, _require_https from ._imports import require_extra from .exceptions import UnauthorizedError from .utils import ( + SingleFlight, ensure_async, record_verified_jkt, refusal, @@ -118,9 +120,10 @@ def __init__( self._refresh_cooldown = refresh_cooldown self._timeout = timeout self._keys: dict[str, Any] = {} - self._fetched_at = 0.0 - self._attempted_at = float("-inf") - self._refresh_lock = anyio.Lock() + self._fetched_at = float("-inf") + self._attempts: dict[Any, float] = {} + self._lock = threading.Lock() + self._refreshes = SingleFlight() async def key_for_token(self, token: str) -> Any: """The key that signed `token`, found by its header `kid`. @@ -159,27 +162,42 @@ def _cache_stale(self) -> bool: async def _refresh(self, force: bool = False) -> None: """Refetch the JWKS, coalescing concurrent and repeated refreshes. - At most one attempt per `refresh_cooldown`; a failed fetch falls + Requests on one event loop share its refresh in flight. Each loop + attempts at most one fetch per `refresh_cooldown`, and none within + it of a fetch that succeeded on any loop; a failed fetch falls back on cached keys, or raises when the cache is empty. """ - async with self._refresh_lock: - # Re-check after acquiring: another request may just have tried. - if time.monotonic() - self._attempted_at < self._refresh_cooldown: - if not self._keys: - raise _JWKSUnavailable() - return - if not force and not self._cache_stale(): - return - self._attempted_at = time.monotonic() - try: - await self._fetch() - except _JWKSFetchError as exc: - logger.warning("JWKS refresh failed: %s", exc) - if not self._keys: - raise _JWKSUnavailable() from exc - # Stale fallback: keep validating with the cached keys. + loop = current_token() + await self._refreshes.run(None, lambda: self._attempt(loop, force)) + + async def _attempt(self, loop: Any, force: bool) -> None: + if not self._claim_attempt(loop, force): + if not self._keys: + raise _JWKSUnavailable() + return + try: + await self._fetch() + except _JWKSFetchError as exc: + logger.warning("JWKS refresh failed: %s", exc) + if not self._keys: + raise _JWKSUnavailable() from exc + + def _claim_attempt(self, loop: Any, force: bool) -> bool: + """Whether `loop` may fetch now, recording the attempt when it may.""" + now = time.monotonic() + cooldown = self._refresh_cooldown + with self._lock: + last = max(self._fetched_at, self._attempts.get(loop, float("-inf"))) + if now - last < cooldown or not (force or self._cache_stale()): + return False + self._attempts = { + seen: at for seen, at in self._attempts.items() if now - at < cooldown + } + self._attempts[loop] = now + return True async def _fetch(self) -> None: + started = time.monotonic() document = await _get_json( self._url, timeout=self._timeout, @@ -221,8 +239,10 @@ async def _fetch(self) -> None: f"JWKS document has no usable signing key " f"({len(document['keys'])} entries)" ) - self._keys = keys - self._fetched_at = time.monotonic() + with self._lock: + if started >= self._fetched_at: + self._keys = keys + self._fetched_at = started class JWTValidator: diff --git a/src/fastapi_multiauth/oauth/client.py b/src/fastapi_multiauth/oauth/client.py index 1ba64f7..bb2a24a 100644 --- a/src/fastapi_multiauth/oauth/client.py +++ b/src/fastapi_multiauth/oauth/client.py @@ -6,11 +6,11 @@ from urllib.parse import urlsplit import httpx -from async_lru import alru_cache from .._http import DEFAULT_TIMEOUT, _get_json, _require_https from .._imports import require_extra from ..exceptions import OAuthDiscoveryError, OAuthExchangeError, OAuthUserinfoError +from ..utils import SingleFlight, TTLCache __all__ = [ "OIDCEndpoints", @@ -21,6 +21,7 @@ _DISCOVERY_SUFFIX = "/.well-known/openid-configuration" _DISCOVERY_CACHE_TTL = 3600 +_DISCOVERY_CACHE_SIZE = 32 TokenEndpointAuthMethod = Literal["client_secret_post", "client_secret_basic"] @@ -114,7 +115,15 @@ def get_httpx_client(self) -> Any: return _TimeoutOAuth2 -@alru_cache(maxsize=32, ttl=_DISCOVERY_CACHE_TTL) +_discovery_cache = TTLCache(_DISCOVERY_CACHE_SIZE) +_discovery_flights = SingleFlight() + + +def _clear_discovery_cache() -> None: + """Forget every cached discovery document.""" + _discovery_cache.clear() + + async def oauth_resolve_provider_urls( discovery_url: str, timeout: float = DEFAULT_TIMEOUT, @@ -138,6 +147,24 @@ async def oauth_resolve_provider_urls( ones only pass from a loopback discovery URL), or names another issuer. """ + key = (discovery_url, timeout) + + async def fetch() -> OIDCEndpoints: + endpoints = await _fetch_provider_urls(discovery_url, timeout) + _discovery_cache.put(key, endpoints, _DISCOVERY_CACHE_TTL) + return endpoints + + cached = _discovery_cache.get(key) + if cached is not None: + return cached + return await _discovery_flights.run(key, fetch) + + +setattr(oauth_resolve_provider_urls, "cache_clear", _clear_discovery_cache) # noqa: B010 + + +async def _fetch_provider_urls(discovery_url: str, timeout: float) -> OIDCEndpoints: + """Fetch and check one discovery document.""" _require_https(discovery_url, "OIDC discovery URL") cfg = await _get_json( discovery_url, diff --git a/src/fastapi_multiauth/sources/dpop.py b/src/fastapi_multiauth/sources/dpop.py index c99d184..6ee5fc6 100644 --- a/src/fastapi_multiauth/sources/dpop.py +++ b/src/fastapi_multiauth/sources/dpop.py @@ -6,6 +6,7 @@ import json import re import string +import threading import time from collections import OrderedDict from collections.abc import Callable, Mapping, Sequence @@ -176,6 +177,7 @@ def __init__( ) self._jkt = ensure_async(jkt) if jkt else None self._keys: OrderedDict[tuple[str, str], Any] = OrderedDict() + self._keys_lock = threading.Lock() super().__init__( validator, HTTPBase( @@ -243,11 +245,11 @@ def _request_url(self, request: HTTPConnection) -> str | None: ) def _remember_key(self, cache_key: tuple[str, str], key: Any) -> None: - """Remember the parsed key of a verified proof, by thumbprint and algorithm.""" - self._keys[cache_key] = key - self._keys.move_to_end(cache_key) - if len(self._keys) > _KEY_CACHE_SIZE: - self._keys.popitem(last=False) + """Remember the parsed key of a verified proof, evicting the oldest past the size.""" + with self._keys_lock: + self._keys[cache_key] = key + if len(self._keys) > _KEY_CACHE_SIZE: + self._keys.popitem(last=False) def _verify_proof( self, token: str, request: HTTPConnection | None @@ -280,7 +282,8 @@ def _verify_proof( raise self._proof_error("invalid_dpop_proof") cache_key = (thumbprint, alg) try: - key = self._keys.get(cache_key) or self._jwt.PyJWK(jwk, algorithm=alg) + cached = self._keys.get(cache_key) + key = cached or self._jwt.PyJWK(jwk, algorithm=alg) claims = self._jwt.decode( proofs[0], key=key, @@ -294,7 +297,8 @@ def _verify_proof( ) except (self._jwt.exceptions.PyJWTError, TypeError, ValueError): raise self._proof_error("invalid_dpop_proof") from None - self._remember_key(cache_key, key) + if cached is None: + self._remember_key(cache_key, key) now = time.time() iat = finite_number(claims["iat"]) if iat is None or not ( diff --git a/src/fastapi_multiauth/utils.py b/src/fastapi_multiauth/utils.py index 860a45f..3653e2d 100644 --- a/src/fastapi_multiauth/utils.py +++ b/src/fastapi_multiauth/utils.py @@ -1,6 +1,7 @@ """Standalone helpers: public token utilities plus internal source helpers.""" import base64 +import copy import functools import hashlib import hmac @@ -9,15 +10,19 @@ import math import secrets import string +import threading import time import zlib +from collections import OrderedDict from collections.abc import Awaitable, Callable, Mapping, Sequence from contextvars import ContextVar from datetime import UTC, datetime from typing import Any, TypeVar from urllib.parse import urlsplit +import anyio import anyio.to_thread +from anyio.lowlevel import current_token from fastapi import HTTPException, status from fastapi.requests import HTTPConnection @@ -315,6 +320,93 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any: return wrapper +class _Flight: + """One call in flight and its outcome.""" + + __slots__ = ("done", "error", "landed", "result") + + def __init__(self) -> None: + self.done = anyio.Event() + self.error: Exception | None = None + self.landed = False + self.result: Any = None + + +class SingleFlight: + """Calls in flight, one per key and event loop, shared by the requests awaiting them. + + A request on another event loop makes its own call: nothing waits + across threads. + """ + + def __init__(self) -> None: + self._flights: dict[tuple[Any, Any], _Flight] = {} + + async def run(self, key: Any, call: Callable[[], Awaitable[Any]]) -> Any: + """The outcome of `call()`, or of the call already in flight for `key`. + + A failure is raised to the requests sharing it as a copy. + """ + slot = (current_token(), key) + while (flight := self._flights.get(slot)) is not None: + await flight.done.wait() + if flight.error is not None: + raise copy.copy(flight.error) + if flight.landed: + return flight.result + flight = self._flights[slot] = _Flight() + try: + flight.result = await call() + flight.landed = True + except Exception as exc: + flight.error = exc + raise + finally: + del self._flights[slot] + flight.done.set() + return flight.result + + +class TTLCache: + """A bounded cache of expiring entries, safe to share across threads. + + A read takes no lock. The entry stored first is evicted first. + """ + + def __init__(self, size: int) -> None: + self._size = size + self._entries: OrderedDict[Any, tuple[float, Any]] = OrderedDict() + self._lock = threading.Lock() + + def __len__(self) -> int: + return len(self._entries) + + def get(self, key: Any) -> Any | None: + """The value stored for `key`, `None` when absent or expired.""" + entry = self._entries.get(key) + if entry is None: + return None + expires_at, value = entry + if time.monotonic() >= expires_at: + with self._lock: + self._entries.pop(key, None) + return None + return value + + def put(self, key: Any, value: Any, ttl: float) -> None: + """Store `value` for `ttl` seconds, evicting the oldest entries past the size.""" + with self._lock: + self._entries[key] = (time.monotonic() + ttl, value) + self._entries.move_to_end(key) + while len(self._entries) > self._size: + self._entries.popitem(last=False) + + def clear(self) -> None: + """Forget every entry.""" + with self._lock: + self._entries.clear() + + def granted_scopes(claims: dict[str, Any], claim: str) -> set[str]: """Read the scopes `claim`: a space-separated string or a list.""" value = claims.get(claim) diff --git a/tests/test_multiauth.py b/tests/test_multiauth.py index 8f662c1..35a44ed 100644 --- a/tests/test_multiauth.py +++ b/tests/test_multiauth.py @@ -10,8 +10,12 @@ import math import secrets import string +import sys +import sysconfig +import threading import time from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor from datetime import UTC, datetime from enum import StrEnum from operator import itemgetter @@ -19,6 +23,7 @@ from unittest.mock import patch from urllib.parse import parse_qs, urlparse +import anyio import httpx import jwt as pyjwt import pyotp @@ -98,9 +103,11 @@ oauth_resolve_provider_urls, oauth_validate_id_token, ) +from fastapi_multiauth.oauth.client import _clear_discovery_cache from fastapi_multiauth.oauth.id_token import _key_set as _id_key_set from fastapi_multiauth.sources.dpop import _jwk_thumbprint from fastapi_multiauth.utils import ( + TTLCache, VerifiedBinding, ensure_async, first_use, @@ -1572,7 +1579,7 @@ async def test_returns_endpoints(self): route = respx.get(DISCOVERY_URL) route.respond(json=_discovery_doc()) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() endpoints = await oauth_resolve_provider_urls(DISCOVERY_URL) assert isinstance(endpoints, OIDCEndpoints) assert endpoints.authorization_endpoint == "https://auth.example.com/authorize" @@ -1589,7 +1596,7 @@ async def test_returns_endpoints(self): end_session_endpoint="https://auth.example.com/logout", ) ) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() endpoints = await oauth_resolve_provider_urls(DISCOVERY_URL) assert endpoints.jwks_uri == "https://auth.example.com/jwks" assert endpoints.end_session_endpoint == "https://auth.example.com/logout" @@ -1600,7 +1607,7 @@ async def test_returns_endpoints(self): async def test_caches_discovery_document(self): route = respx.get(DISCOVERY_URL).respond(json=_discovery_doc()) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() await oauth_resolve_provider_urls(DISCOVERY_URL) await oauth_resolve_provider_urls(DISCOVERY_URL) @@ -1629,7 +1636,7 @@ async def test_unusable_document_raises_discovery_error(self): ), ]: route.mock(side_effect=[response]) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() with pytest.raises(OAuthDiscoveryError, match=match): await oauth_resolve_provider_urls(DISCOVERY_URL) @@ -1662,7 +1669,7 @@ def test_http_authorization_url_rejected(self): @respx.mock async def test_http_discovery_url_rejected_before_fetch(self): # No respx routes are registered: any HTTP request would error out. - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() with pytest.raises(ValueError, match="discovery URL must use https"): await oauth_resolve_provider_urls( "http://auth.example.com/.well-known/openid-configuration" @@ -1686,7 +1693,7 @@ async def test_http_endpoint_in_discovery_document_rejected(self): **{endpoint: "http://attacker.example.com/endpoint"} ) ) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() with pytest.raises(OAuthDiscoveryError, match=f"{endpoint} must use https"): await oauth_resolve_provider_urls(DISCOVERY_URL) @@ -1697,7 +1704,7 @@ async def test_loopback_endpoint_in_remote_discovery_document_rejected(self): for endpoint in ["token_endpoint", "userinfo_endpoint", "jwks_uri"]: for url in ["http://127.0.0.1:8080/x", "http://localhost/x"]: route.respond(json=_discovery_doc(**{endpoint: url})) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() with pytest.raises( OAuthDiscoveryError, match=f"{endpoint} must use https" ): @@ -1722,7 +1729,7 @@ async def test_http_loopback_discovery_allowed(self): } ) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() endpoints = await oauth_resolve_provider_urls( f"http://{host}:8080/.well-known/openid-configuration" ) @@ -1751,7 +1758,7 @@ async def test_foreign_or_missing_issuer_rejected(self): ]: with respx.mock: respx.get(url).respond(json=doc) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() with pytest.raises(OAuthDiscoveryError, match="issuer"): await oauth_resolve_provider_urls(url) @@ -1766,7 +1773,7 @@ async def test_issuer_variants_accepted(self): ]: with respx.mock: respx.get(url).respond(json=doc) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() endpoints = await oauth_resolve_provider_urls(url) assert endpoints.token_endpoint == "https://auth.example.com/token" assert endpoints.issuer == doc["issuer"] @@ -1779,19 +1786,19 @@ async def test_nonstandard_discovery_url(self): route = respx.get(url) route.respond(json=_discovery_doc(issuer="https://unrelated.example.com")) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() with pytest.raises(OAuthDiscoveryError, match="origin"): await oauth_resolve_provider_urls(url) no_issuer = _discovery_doc() del no_issuer["issuer"] route.respond(json=no_issuer) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() with pytest.raises(OAuthDiscoveryError, match="issuer"): await oauth_resolve_provider_urls(url) route.respond(json=_discovery_doc(issuer="https://auth.example.com/t1")) - oauth_resolve_provider_urls.cache_clear() + _clear_discovery_cache() endpoints = await oauth_resolve_provider_urls(url) assert endpoints.issuer == "https://auth.example.com/t1" @@ -6588,10 +6595,14 @@ def _active(**overrides) -> dict: **overrides, } - def _validator(self, **kwargs) -> IntrospectionValidator: + @staticmethod + def _validator(**kwargs) -> IntrospectionValidator: kwargs.setdefault("audience", "my-api") return IntrospectionValidator( - url=self.URL, client_id="my-api", client_secret="api-secret", **kwargs + url=TestIntrospectionValidator.URL, + client_id="my-api", + client_secret="api-secret", + **kwargs, ) @respx.mock @@ -7695,3 +7706,277 @@ def test_jwks_refusals(self): ("jwks_unavailable", 503) ] self._assert_no_leak(token) + + +_THREADS = 4 + + +def _in_threads(work: Callable[[int], Any]) -> list[Any]: + """Run `work(index)` in `_THREADS` threads at once, each result in order.""" + barrier = threading.Barrier(_THREADS) + + def run(index: int) -> Any: + barrier.wait() + return work(index) + + with ThreadPoolExecutor(_THREADS) as pool: + return list(pool.map(run, range(_THREADS))) + + +def _on_loops(call: Callable[[], Any], requests: int = 3) -> list[Any]: + """The outcomes of `requests` concurrent `call()`s on each of `_THREADS` event loops.""" + + async def one(outcomes: list[Any]) -> None: + try: + outcomes.append(await call()) + except (HTTPException, OAuthDiscoveryError) as exc: + outcomes.append(exc) + + async def main() -> list[Any]: + outcomes: list[Any] = [] + with anyio.fail_after(10): + async with anyio.create_task_group() as group: + for _ in range(requests): + group.start_soon(one, outcomes) + return outcomes + + return [ + outcome for loop in _in_threads(lambda _: anyio.run(main)) for outcome in loop + ] + + +def _hammer(turn: Callable[[int, int], Any]) -> None: + """Await `turn(thread, count)` many times in a row on each of `_THREADS` event loops.""" + + async def main(thread: int) -> None: + for count in range(300): + await turn(thread, count) + + _in_threads(lambda thread: anyio.run(main, thread)) + + +def _kid_token(key, kid: str) -> str: + return pyjwt.encode(_claims(), key, algorithm="RS256", headers={"kid": kid}) + + +class TestSharedAcrossThreads: + """One validator or source serving several threads, each with its own event loop.""" + + @staticmethod + def _jwks(monkeypatch, keys: list[dict] | None, delay: float = 0.05) -> list[str]: + """Serve `keys` as the JWKS, or fail when `None`; returns the fetch log.""" + fetches: list[str] = [] + + async def get_json(url, *, error_cls, **kwargs): + fetches.append(url) + await anyio.sleep(delay) + if keys is None: + raise error_cls("provider down") + return {"keys": list(keys)} + + monkeypatch.setattr("fastapi_multiauth.jwt._get_json", get_json) + return fetches + + def test_jwks_cold_start(self, monkeypatch): + fetches = self._jwks(monkeypatch, [_jwk(_RSA_KEY_1, "key1")]) + validator = _jwks_validator("https://idp.example.com/jwks") + token = _kid_token(_RSA_KEY_1, "key1") + + answers = _on_loops(lambda: validator(token)) + + assert [claims["sub"] for claims in answers] == ["user-1"] * (_THREADS * 3) + assert 1 <= len(fetches) <= _THREADS + + def test_jwks_cold_start_with_the_provider_down(self, monkeypatch): + fetches = self._jwks(monkeypatch, None) + validator = _jwks_validator("https://idp.example.com/jwks") + token = _kid_token(_RSA_KEY_1, "key1") + + for _ in range(2): + refusals = _on_loops(lambda: validator(token)) + assert [exc.status_code for exc in refusals] == [503] * (_THREADS * 3) + assert 1 <= len(fetches) <= _THREADS * 2 + + def test_jwks_unknown_kid_respects_the_cooldown(self, monkeypatch): + keys = [_jwk(_RSA_KEY_1, "key1")] + fetches = self._jwks(monkeypatch, keys, delay=0.01) + validator = _jwks_validator("https://idp.example.com/jwks") + anyio.run(validator, _kid_token(_RSA_KEY_1, "key1")) + junk = _kid_token(_RSA_KEY_2, "unknown") + + for _ in range(3): + refusals = _on_loops(lambda: validator(junk)) + assert [exc.status_code for exc in refusals] == [401] * (_THREADS * 3) + assert len(fetches) == 1 + + key_set = validator._key_set + assert key_set is not None + key_set._fetched_at -= 31 + key_set._attempts.clear() + keys.append(_jwk(_RSA_KEY_2, "unknown")) + answers = _on_loops(lambda: validator(junk)) + assert [claims["sub"] for claims in answers] == ["user-1"] * (_THREADS * 3) + assert 2 <= len(fetches) <= 1 + _THREADS + + @pytest.mark.anyio + async def test_a_slower_older_jwks_never_replaces_a_newer_one(self, monkeypatch): + answers = [ + (0.05, {"keys": [_jwk(_RSA_KEY_1, "old")]}), + (0.0, {"keys": [_jwk(_RSA_KEY_2, "new")]}), + ] + + async def get_json(url, **kwargs): + delay, document = answers.pop(0) + await anyio.sleep(delay) + return document + + monkeypatch.setattr("fastapi_multiauth.jwt._get_json", get_json) + key_set = _jwks_validator("https://idp.example.com/jwks")._key_set + assert key_set is not None + + async with anyio.create_task_group() as group: + group.start_soon(key_set._fetch) + await anyio.sleep(0.01) + group.start_soon(key_set._fetch) + + assert list(key_set._keys) == ["new"] + + def test_introspection_in_flight(self, monkeypatch): + lookups = [] + + async def request_json(method, url, **kwargs): + lookups.append(url) + await anyio.sleep(0.05) + return TestIntrospectionValidator._active() + + monkeypatch.setattr( + "fastapi_multiauth.introspection._request_json", request_json + ) + validator = TestIntrospectionValidator._validator() + + answers = _on_loops(lambda: validator("opaque-token")) + + assert [claims["sub"] for claims in answers] == ["user-1"] * (_THREADS * 3) + assert 1 <= len(lookups) <= _THREADS + + def test_introspection_cache_eviction(self, monkeypatch): + async def request_json(method, url, **kwargs): + return TestIntrospectionValidator._active(sub=kwargs["data"]["token"]) + + monkeypatch.setattr( + "fastapi_multiauth.introspection._request_json", request_json + ) + validator = TestIntrospectionValidator._validator(cache_size=2) + + async def turn(thread: int, count: int) -> None: + token = f"token-{(count * (thread + 1)) % 6}" + assert (await validator(token))["sub"] == token + + _hammer(turn) + assert len(validator._cache) <= 2 + + def test_dpop_key_cache_eviction(self, monkeypatch): + monkeypatch.setattr("fastapi_multiauth.sources.dpop._KEY_CACHE_SIZE", 1) + source = TestDPoP._source() + scopes = [] + for key in (_EC_KEY, _EC_KEY_2): + token = TestDPoP._token(key) + scopes.append( + { + "type": "http", + "method": "GET", + "scheme": "http", + "path": "/me", + "raw_path": b"/me", + "query_string": b"", + "root_path": "", + "server": ("testserver", 80), + "headers": [ + (b"authorization", f"DPoP {token}".encode()), + (b"dpop", TestDPoP._proof(token, key).encode()), + ], + } + ) + + async def turn(thread: int, count: int) -> None: + request = Request(dict(scopes[(count + thread) % 2])) + assert (await source.dispatch(request, []))["sub"] == "user-1" + + _hammer(turn) + assert len(source._keys) == 1 + + def test_discovery(self, monkeypatch): + fetches = [] + + async def get_json(url, **kwargs): + fetches.append(url) + await anyio.sleep(0.05) + return _discovery_doc() + + monkeypatch.setattr("fastapi_multiauth.oauth.client._get_json", get_json) + _clear_discovery_cache() + + answers = _on_loops(lambda: oauth_resolve_provider_urls(DISCOVERY_URL)) + _clear_discovery_cache() + + assert [endpoints.issuer for endpoints in answers] == [ + "https://auth.example.com" + ] * (_THREADS * 3) + assert 1 <= len(fetches) <= _THREADS + + def test_discovery_with_the_provider_down(self, monkeypatch): + fetches = [] + + async def get_json(url, *, error_cls, **kwargs): + fetches.append(url) + await anyio.sleep(0.05) + raise error_cls("provider down") + + monkeypatch.setattr("fastapi_multiauth.oauth.client._get_json", get_json) + _clear_discovery_cache() + + failures = _on_loops(lambda: oauth_resolve_provider_urls(DISCOVERY_URL)) + + assert all(isinstance(exc, OAuthDiscoveryError) for exc in failures) + assert len(failures) == _THREADS * 3 + assert 1 <= len(fetches) <= _THREADS + + def test_the_discovery_cache_is_cleared_from_the_function(self): + assert ( + vars(oauth_resolve_provider_urls)["cache_clear"] == _clear_discovery_cache + ) + + +class TestTTLCache: + def test_entries_expire(self): + cache = TTLCache(2) + cache.put("a", 1, 60) + assert cache.get("a") == 1 + assert cache.get("b") is None + with patch("time.monotonic", return_value=time.monotonic() + 61): + assert cache.get("a") is None + assert len(cache) == 0 + + def test_the_entry_stored_first_is_evicted_first(self): + cache = TTLCache(2) + for key in ["a", "b", "a", "c"]: + cache.put(key, key, 60) + assert cache.get("b") is None + assert (cache.get("a"), cache.get("c")) == ("a", "c") + cache.clear() + assert len(cache) == 0 + + +@pytest.mark.skipif( + not sysconfig.get_config_var("Py_GIL_DISABLED"), + reason="needs a free-threaded Python build", +) +def test_free_threaded_build_keeps_the_gil_disabled(): + import jwt.algorithms + + import fastapi_multiauth.oauth + import fastapi_multiauth.totp # noqa: F401 + + assert sys.version_info >= (3, 13) + assert jwt.algorithms.has_crypto + assert not sys._is_gil_enabled() diff --git a/uv.lock b/uv.lock index 450f93b..64d4f8e 100644 --- a/uv.lock +++ b/uv.lock @@ -37,15 +37,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/12/b8/4bd346e22b28902df4d651910f5242c28d84e4a5c2435ca5c3f797ed7e2e/anyio-4.15.1-py3-none-any.whl", hash = "sha256:6152fdbbf9a77fdec97731721bebf7c4c44f7c29b424b0065826173efc7ed101", size = 132079, upload-time = "2026-09-05T10:42:37.923Z" }, ] -[[package]] -name = "async-lru" -version = "2.4.0" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/8f/43/9d22679e8dd0ed965182cca990d32c68b150730cce8046ad61b3179b5b7a/async_lru-2.4.0.tar.gz", hash = "sha256:b0b0077d9b7596615b3c12d0aad4f6aeccaaa0ac75ceae472608ede89e764c08", size = 22927, upload-time = "2026-10-12T18:12:54.276Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/ec/0b/6a817103fb78faca4970c686a61065047da7226aba0d7cae6d4299b20c59/async_lru-2.4.0-py3-none-any.whl", hash = "sha256:1b46a96419ef91fe11431d58db98d792c9b4250001427191a40faf1031934212", size = 10045, upload-time = "2026-10-12T18:12:53.091Z" }, -] - [[package]] name = "certifi" version = "2026.10.23" @@ -398,7 +389,6 @@ version = "1.1.0" source = { editable = "." } dependencies = [ { name = "anyio" }, - { name = "async-lru" }, { name = "fastapi" }, { name = "httpx" }, { name = "itsdangerous" }, @@ -447,7 +437,6 @@ tests = [ [package.metadata] requires-dist = [ { name = "anyio", specifier = ">=3.7" }, - { name = "async-lru", specifier = ">=2.0" }, { name = "fastapi", specifier = ">=0.123.0,!=0.137.0,!=0.137.1" }, { name = "httpx", specifier = ">=0.25.0" }, { name = "httpx-oauth", marker = "extra == 'oauth'", specifier = ">=0.16" },