ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 07:09:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

Extract TokenCache utility, add caching to GitHubTokenVerifier (#3547)

Browse source 
* feat: add TokenCache utility and caching to GitHubTokenVerifier

Extract the caching machinery from IntrospectionTokenVerifier into a
shared TokenCache class in fastmcp.utilities.token_cache, then wire
it into both IntrospectionTokenVerifier and GitHubTokenVerifier.

* Remove dead constant, validate negative cache params

* Fix overwrite eviction bug, skip cache on scope lookup failure
This commit is contained in:
Jeremiah Lowin 2026-03-18 15:26:09 -04:00 • committed by GitHub
commit 269c9c9f46
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 648 additions and 154 deletions
Download patch file Download diff file Expand all files Collapse all files

34
src/fastmcp/server/auth/providers/github.py
View file

@ -33,6 +33,7 @@ from fastmcp.server.auth.auth import AccessToken
from fastmcp.server.auth.oauth_proxy import OAuthProxy
from fastmcp.utilities.auth import parse_scopes
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.token_cache import TokenCache
logger = get_logger(__name__)
@ -42,6 +43,10 @@ class GitHubTokenVerifier(TokenVerifier):
GitHub OAuth tokens are opaque (not JWTs), so we verify them
by calling GitHub's API to check if they're valid and get user info.
Caching is disabled by default. Set ``cache_ttl_seconds`` to a positive
integer to cache successful verification results and avoid repeated
GitHub API calls for the same token.
"""
def __init__(
@ -49,6 +54,8 @@ class GitHubTokenVerifier(TokenVerifier):
*,
required_scopes: list[str] | None = None,
timeout_seconds: int = 10,
cache_ttl_seconds: int | None = None,
max_cache_size: int | None = None,
http_client: httpx.AsyncClient | None = None,
):
"""Initialize the GitHub token verifier.
@ -56,6 +63,10 @@ class GitHubTokenVerifier(TokenVerifier):
Args:
required_scopes: Required OAuth scopes (e.g., ['user:email'])
timeout_seconds: HTTP request timeout
cache_ttl_seconds: How long to cache verification results in seconds.
Caching is disabled by default (None). Set to a positive integer
to enable (e.g., 300 for 5 minutes).
max_cache_size: Maximum number of tokens to cache. Default: 10 000.
http_client: Optional httpx.AsyncClient for connection pooling. When provided,
the client is reused across calls and the caller is responsible for its
lifecycle. When None (default), a fresh client is created per call.
@ -63,9 +74,18 @@ class GitHubTokenVerifier(TokenVerifier):
super().__init__(required_scopes=required_scopes)
self.timeout_seconds = timeout_seconds
self._http_client = http_client
self._cache = TokenCache(
ttl_seconds=cache_ttl_seconds,
max_size=max_cache_size,
)
async def verify_token(self, token: str) -> AccessToken | None:
"""Verify GitHub OAuth token by calling GitHub API."""
is_cached, cached_result = self._cache.get(token)
if is_cached:
logger.debug("GitHub token cache hit")
return cached_result
try:
async with (
contextlib.nullcontext(self._http_client)
@ -104,6 +124,7 @@ class GitHubTokenVerifier(TokenVerifier):
)
# Extract scopes from X-OAuth-Scopes header if available
scopes_verified = scopes_response.status_code == 200
oauth_scopes_header = scopes_response.headers.get("x-oauth-scopes", "")
token_scopes = [
scope.strip()
@ -128,7 +149,7 @@ class GitHubTokenVerifier(TokenVerifier):
return None
# Create AccessToken with GitHub user info
return AccessToken(
result = AccessToken(
token=token,
client_id=str(user_data.get("id", "unknown")), # Use GitHub user ID
scopes=token_scopes,
@ -142,6 +163,9 @@ class GitHubTokenVerifier(TokenVerifier):
"github_user_data": user_data,
},
)
if scopes_verified:
self._cache.set(token, result)
return result
except httpx.RequestError as e:
logger.debug("Failed to verify GitHub token: %s", e)
@ -189,6 +213,8 @@ class GitHubProvider(OAuthProxy):
redirect_path: str | None = None,
required_scopes: list[str] | None = None,
timeout_seconds: int = 10,
cache_ttl_seconds: int | None = None,
max_cache_size: int | None = None,
allowed_client_redirect_uris: list[str] | None = None,
client_storage: AsyncKeyValue | None = None,
jwt_signing_key: str | bytes | None = None,
@ -207,6 +233,10 @@ class GitHubProvider(OAuthProxy):
redirect_path: Redirect path configured in GitHub OAuth app (defaults to "/auth/callback")
required_scopes: Required GitHub scopes (defaults to ["user"])
timeout_seconds: HTTP request timeout for GitHub API calls (defaults to 10)
cache_ttl_seconds: How long to cache token verification results in seconds.
Caching is disabled by default (None). Set to a positive integer to
enable (e.g., 300 for 5 minutes).
max_cache_size: Maximum number of tokens to cache. Default: 10 000.
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
client_storage: Storage backend for OAuth state (client registrations, encrypted tokens).
@ -234,6 +264,8 @@ class GitHubProvider(OAuthProxy):
token_verifier = GitHubTokenVerifier(
required_scopes=required_scopes_final,
timeout_seconds=timeout_seconds,
cache_ttl_seconds=cache_ttl_seconds,
max_cache_size=max_cache_size,
http_client=http_client,
)

111
src/fastmcp/server/auth/providers/introspection.py
View file

@ -25,9 +25,7 @@ from __future__ import annotations
import base64
import contextlib
import hashlib
import time
from dataclasses import dataclass
from typing import Any, Literal, get_args
import httpx
@ -36,18 +34,11 @@ from pydantic import AnyHttpUrl, SecretStr
from fastmcp.server.auth import AccessToken, TokenVerifier
from fastmcp.utilities.auth import parse_scopes
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.token_cache import TokenCache
logger = get_logger(__name__)
@dataclass
class _IntrospectionCacheEntry:
"""Cached introspection result with expiration."""
result: AccessToken
expires_at: float
ClientAuthMethod = Literal["client_secret_basic", "client_secret_post"]
@ -86,9 +77,6 @@ class IntrospectionTokenVerifier(TokenVerifier):
```
"""
# Default cache settings
DEFAULT_MAX_CACHE_SIZE = 10000
def __init__(
self,
*,
@ -154,96 +142,9 @@ class IntrospectionTokenVerifier(TokenVerifier):
self._http_client = http_client
self.logger = get_logger(__name__)
# Cache configuration (None or 0 = disabled)
self._cache_ttl = cache_ttl_seconds or 0
self._max_cache_size = (
max_cache_size
if max_cache_size is not None
else self.DEFAULT_MAX_CACHE_SIZE
)
self._cache: dict[str, _IntrospectionCacheEntry] = {}
self._last_cleanup = time.monotonic()
self._cleanup_interval = 60 # Cleanup every 60 seconds
def _hash_token(self, token: str) -> str:
"""Hash token for use as cache key.
Using SHA-256 for memory efficiency (fixed 64-char hex digest
regardless of token length).
"""
return hashlib.sha256(token.encode("utf-8")).hexdigest()
def _cleanup_expired_cache(self) -> None:
"""Remove expired entries from cache."""
now = time.time()
expired = [key for key, entry in self._cache.items() if entry.expires_at < now]
for key in expired:
del self._cache[key]
if expired:
self.logger.debug("Cleaned up %d expired cache entries", len(expired))
def _maybe_cleanup(self) -> None:
"""Periodically cleanup expired entries to prevent unbounded growth."""
now = time.monotonic()
if now - self._last_cleanup > self._cleanup_interval:
self._cleanup_expired_cache()
self._last_cleanup = now
def _get_cached(self, token: str) -> tuple[bool, AccessToken | None]:
"""Get cached introspection result.
Returns:
Tuple of (is_cached, result):
- (True, AccessToken) if cached valid token
- (False, None) if not in cache or expired
"""
if self._cache_ttl <= 0 or self._max_cache_size <= 0:
return (False, None) # Caching disabled
cache_key = self._hash_token(token)
entry = self._cache.get(cache_key)
if entry is None:
return (False, None) # Not in cache
if entry.expires_at < time.time():
del self._cache[cache_key]
return (False, None) # Expired
# Return a copy to prevent mutations from affecting cached value
return (True, entry.result.model_copy(deep=True))
def _set_cached(self, token: str, result: AccessToken) -> None:
"""Cache a valid introspection result with TTL.
Only successful validations are cached. Failures (inactive, expired,
missing scopes, errors) are never cached to avoid sticky false negatives.
"""
if self._cache_ttl <= 0 or self._max_cache_size <= 0:
return # Caching disabled
# Periodic cleanup
self._maybe_cleanup()
# Check cache size limit
if len(self._cache) >= self._max_cache_size:
self._cleanup_expired_cache()
# If still at limit after cleanup, evict oldest entry
if len(self._cache) >= self._max_cache_size:
oldest_key = next(iter(self._cache))
del self._cache[oldest_key]
cache_key = self._hash_token(token)
# Use token's expiration if available and sooner than TTL
expires_at = time.time() + self._cache_ttl
if result.expires_at:
expires_at = min(expires_at, float(result.expires_at))
# Store a deep copy to prevent mutations from affecting cached value
self._cache[cache_key] = _IntrospectionCacheEntry(
result=result.model_copy(deep=True),
expires_at=expires_at,
self._cache = TokenCache(
ttl_seconds=cache_ttl_seconds,
max_size=max_cache_size,
)
def _create_basic_auth_header(self) -> str:
@ -293,7 +194,7 @@ class IntrospectionTokenVerifier(TokenVerifier):
AccessToken object if valid and active, None if invalid, inactive, or expired
"""
# Check cache first
is_cached, cached_result = self._get_cached(token)
is_cached, cached_result = self._cache.get(token)
if is_cached:
self.logger.debug("Token introspection cache hit")
return cached_result
@ -388,7 +289,7 @@ class IntrospectionTokenVerifier(TokenVerifier):
expires_at=int(exp) if exp else None,
claims=introspection_data, # Store full response for extensibility
)
self._set_cached(token, result)
self._cache.set(token, result)
return result
except httpx.TimeoutException:

173
src/fastmcp/utilities/token_cache.py Normal file
View file

@ -0,0 +1,173 @@
"""In-memory cache for token verification results.
Provides a generic TTL-based cache for ``AccessToken`` objects, designed to
reduce repeated network calls during opaque-token verification. Only
*successful* verifications should be cached; errors and failures must be
retried on every request.
Example:
```python
from fastmcp.utilities.token_cache import TokenCache
cache = TokenCache(ttl_seconds=300, max_size=10000)
# On cache miss, call the upstream verifier and store the result.
hit, token = cache.get(raw_token)
if not hit:
token = await _call_upstream(raw_token)
if token is not None:
cache.set(raw_token, token)
```
"""
from __future__ import annotations
import hashlib
import time
from dataclasses import dataclass
from fastmcp.server.auth.auth import AccessToken
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
DEFAULT_MAX_CACHE_SIZE = 10_000
_CLEANUP_INTERVAL = 60 # seconds between periodic sweeps
@dataclass
class _CacheEntry:
"""A cached token result with its absolute expiration timestamp."""
result: AccessToken
expires_at: float
class TokenCache:
"""TTL-based in-memory cache for ``AccessToken`` objects.
Features:
- SHA-256 hashed cache keys (fixed size, regardless of token length).
- Per-entry TTL that respects both the configured ``ttl_seconds`` and the
token's own ``expires_at`` claim (whichever is sooner).
- Bounded size with FIFO eviction when the cache is full.
- Periodic cleanup of expired entries to prevent unbounded growth.
- Defensive deep copies on both store and retrieve to prevent
callers from mutating cached values.
Caching is disabled when ``ttl_seconds`` is ``None`` or ``0``, or
when ``max_size`` is ``0``. Negative values raise ``ValueError``.
"""
def __init__(
self,
*,
ttl_seconds: int | None = None,
max_size: int | None = None,
) -> None:
"""Initialise the cache.
Args:
ttl_seconds: How long cached entries remain valid, in seconds.
``None`` or ``0`` disables caching entirely.
max_size: Upper bound on the number of entries. When the limit is
reached, expired entries are swept first; if still full the
oldest entry is evicted. Defaults to 10 000.
"""
if ttl_seconds is not None and ttl_seconds < 0:
raise ValueError(
f"cache_ttl_seconds must be non-negative, got {ttl_seconds}"
)
if max_size is not None and max_size < 0:
raise ValueError(f"max_cache_size must be non-negative, got {max_size}")
self._ttl = ttl_seconds or 0
self._max_size = max_size if max_size is not None else DEFAULT_MAX_CACHE_SIZE
self._entries: dict[str, _CacheEntry] = {}
self._last_cleanup = time.monotonic()
@property
def enabled(self) -> bool:
"""Return whether caching is active."""
return self._ttl > 0 and self._max_size > 0
# -- public API ----------------------------------------------------------
def get(self, token: str) -> tuple[bool, AccessToken | None]:
"""Look up a cached verification result.
Returns:
``(True, AccessToken)`` on a cache hit, ``(False, None)`` on a miss
or when caching is disabled. The returned ``AccessToken`` is a deep
copy that is safe to mutate.
"""
if not self.enabled:
return (False, None)
cache_key = self._hash_token(token)
entry = self._entries.get(cache_key)
if entry is None:
return (False, None)
if entry.expires_at < time.time():
del self._entries[cache_key]
return (False, None)
return (True, entry.result.model_copy(deep=True))
def set(self, token: str, result: AccessToken) -> None:
"""Store a *successful* verification result.
Only successful verifications should be cached. Failures (inactive
tokens, missing scopes, HTTP errors, timeouts) must **not** be cached
so that transient problems do not produce sticky false negatives.
"""
if not self.enabled:
return
cache_key = self._hash_token(token)
self._maybe_cleanup()
if cache_key not in self._entries:
self._enforce_size_limit()
expires_at = time.time() + self._ttl
if result.expires_at:
expires_at = min(expires_at, float(result.expires_at))
self._entries[cache_key] = _CacheEntry(
result=result.model_copy(deep=True),
expires_at=expires_at,
)
# -- internals -----------------------------------------------------------
@staticmethod
def _hash_token(token: str) -> str:
"""Return the SHA-256 hex digest of *token*."""
return hashlib.sha256(token.encode("utf-8")).hexdigest()
def _cleanup_expired(self) -> None:
"""Remove all entries whose TTL has elapsed."""
now = time.time()
expired = [k for k, v in self._entries.items() if v.expires_at < now]
for key in expired:
del self._entries[key]
if expired:
logger.debug("Cleaned up %d expired cache entries", len(expired))
def _maybe_cleanup(self) -> None:
"""Run ``_cleanup_expired`` at most once per cleanup interval."""
now = time.monotonic()
if now - self._last_cleanup > _CLEANUP_INTERVAL:
self._cleanup_expired()
self._last_cleanup = now
def _enforce_size_limit(self) -> None:
"""Ensure there is room for at least one new entry."""
if len(self._entries) < self._max_size:
return
self._cleanup_expired()
if len(self._entries) >= self._max_size:
oldest_key = next(iter(self._entries))
del self._entries[oldest_key]

174
tests/server/auth/providers/test_github.py
View file

@ -1,6 +1,6 @@
"""Unit tests for GitHub OAuth provider."""
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from key_value.aio.stores.memory import MemoryStore
@ -100,8 +100,6 @@ class TestGitHubTokenVerifier:
async def test_verify_token_success(self):
"""Test successful token verification."""
from unittest.mock import AsyncMock
verifier = GitHubTokenVerifier(required_scopes=["user"])
# Mock the httpx.AsyncClient directly
@ -139,3 +137,173 @@ class TestGitHubTokenVerifier:
assert result.scopes == ["user", "repo"]
assert result.claims["login"] == "testuser"
assert result.claims["name"] == "Test User"
def _mock_github_success(mock_client: AsyncMock) -> None:
"""Configure *mock_client* to return a successful GitHub user + scopes response."""
user_response = MagicMock()
user_response.status_code = 200
user_response.json.return_value = {
"id": 12345,
"login": "testuser",
"name": "Test User",
"email": "test@example.com",
"avatar_url": "https://github.com/testuser.png",
}
scopes_response = MagicMock()
scopes_response.status_code = 200
scopes_response.headers = {"x-oauth-scopes": "user,repo"}
mock_client.get.side_effect = [user_response, scopes_response]
def _mock_github_failure(mock_client: AsyncMock) -> None:
"""Configure *mock_client* to return a 401 GitHub response."""
fail_response = MagicMock()
fail_response.status_code = 401
fail_response.text = "Bad credentials"
mock_client.get.return_value = fail_response
class TestGitHubTokenVerifierCaching:
"""Test caching behaviour on GitHubTokenVerifier."""
def test_cache_disabled_by_default(self):
verifier = GitHubTokenVerifier()
assert not verifier._cache.enabled
def test_cache_enabled_with_ttl(self):
verifier = GitHubTokenVerifier(cache_ttl_seconds=300)
assert verifier._cache.enabled
async def test_cache_hit_avoids_second_api_call(self):
verifier = GitHubTokenVerifier(
required_scopes=["user"],
cache_ttl_seconds=300,
)
mock_client = AsyncMock()
with patch(
"fastmcp.server.auth.providers.github.httpx.AsyncClient"
) as mock_cls:
mock_cls.return_value.__aenter__.return_value = mock_client
_mock_github_success(mock_client)
result1 = await verifier.verify_token("tok-1")
assert result1 is not None
assert mock_client.get.call_count == 2 # /user + /user/repos
result2 = await verifier.verify_token("tok-1")
assert result2 is not None
assert result2.client_id == result1.client_id
assert mock_client.get.call_count == 2 # no additional calls
async def test_cache_disabled_makes_every_call(self):
verifier = GitHubTokenVerifier(
required_scopes=["user"],
cache_ttl_seconds=0,
)
mock_client = AsyncMock()
with patch(
"fastmcp.server.auth.providers.github.httpx.AsyncClient"
) as mock_cls:
mock_cls.return_value.__aenter__.return_value = mock_client
_mock_github_success(mock_client)
await verifier.verify_token("tok-1")
assert mock_client.get.call_count == 2
_mock_github_success(mock_client)
await verifier.verify_token("tok-1")
assert mock_client.get.call_count == 4
async def test_failures_are_not_cached(self):
verifier = GitHubTokenVerifier(cache_ttl_seconds=300)
mock_client = AsyncMock()
with patch(
"fastmcp.server.auth.providers.github.httpx.AsyncClient"
) as mock_cls:
mock_cls.return_value.__aenter__.return_value = mock_client
_mock_github_failure(mock_client)
result1 = await verifier.verify_token("bad-tok")
assert result1 is None
_mock_github_success(mock_client)
result2 = await verifier.verify_token("bad-tok")
assert result2 is not None
async def test_cached_result_is_defensive_copy(self):
verifier = GitHubTokenVerifier(
required_scopes=["user"],
cache_ttl_seconds=300,
)
mock_client = AsyncMock()
with patch(
"fastmcp.server.auth.providers.github.httpx.AsyncClient"
) as mock_cls:
mock_cls.return_value.__aenter__.return_value = mock_client
_mock_github_success(mock_client)
result1 = await verifier.verify_token("tok-1")
assert result1 is not None
result1.claims["login"] = "MUTATED"
result2 = await verifier.verify_token("tok-1")
assert result2 is not None
assert result2.claims["login"] == "testuser"
async def test_scope_failure_skips_cache(self):
"""Token verified with fallback scopes (scope API failed) should not be cached."""
verifier = GitHubTokenVerifier(cache_ttl_seconds=300)
mock_client = AsyncMock()
user_response = MagicMock()
user_response.status_code = 200
user_response.json.return_value = {
"id": 12345,
"login": "testuser",
"name": "Test User",
"email": "test@example.com",
"avatar_url": "https://github.com/testuser.png",
}
scopes_response = MagicMock()
scopes_response.status_code = 500
scopes_response.headers = {}
with patch(
"fastmcp.server.auth.providers.github.httpx.AsyncClient"
) as mock_cls:
mock_cls.return_value.__aenter__.return_value = mock_client
mock_client.get.side_effect = [user_response, scopes_response]
result = await verifier.verify_token("tok-1")
assert result is not None
# Should NOT be cached because scope response was not 200
assert not verifier._cache.enabled or len(verifier._cache._entries) == 0
def test_provider_passes_cache_params(self, memory_storage: MemoryStore):
provider = GitHubProvider(
client_id="cid",
client_secret="csec",
base_url="https://example.com",
cache_ttl_seconds=120,
max_cache_size=500,
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
verifier = provider._token_validator
assert isinstance(verifier, GitHubTokenVerifier)
assert verifier._cache.enabled
assert verifier._cache._ttl == 120
assert verifier._cache._max_size == 500

80
tests/server/auth/providers/test_introspection.py
View file

@ -554,8 +554,7 @@ class TestIntrospectionCaching:
client_id="test-client",
client_secret="test-secret",
)
assert verifier._cache_ttl == 0 # Disabled by default
assert verifier._max_cache_size == 10000
assert not verifier._cache.enabled
def test_custom_cache_settings(self):
"""Test that cache settings can be customized."""
@ -566,8 +565,8 @@ class TestIntrospectionCaching:
cache_ttl_seconds=60,
max_cache_size=500,
)
assert verifier._cache_ttl == 60
assert verifier._max_cache_size == 500
assert verifier._cache._ttl == 60
assert verifier._cache._max_size == 500
def test_cache_disabled_with_zero_ttl(self):
"""Test that cache is disabled when TTL is 0 or None."""
@ -578,7 +577,7 @@ class TestIntrospectionCaching:
client_secret="test-secret",
cache_ttl_seconds=0,
)
assert verifier._cache_ttl == 0
assert not verifier._cache.enabled
# Explicit None (same as default)
verifier2 = IntrospectionTokenVerifier(
@ -587,25 +586,20 @@ class TestIntrospectionCaching:
client_secret="test-secret",
cache_ttl_seconds=None,
)
assert verifier2._cache_ttl == 0
assert not verifier2._cache.enabled
async def test_cache_disabled_with_zero_or_negative_max_size(
self, httpx_mock: HTTPXMock
):
"""Test that cache is disabled when max_cache_size is 0 or negative."""
# Add two responses for the two verifiers
for _ in range(2):
httpx_mock.add_response(
url="https://auth.example.com/oauth/introspect",
method="POST",
json={
"active": True,
"client_id": "user-123",
"scope": "read",
},
)
async def test_cache_disabled_with_zero_max_size(self, httpx_mock: HTTPXMock):
"""Test that cache is disabled when max_cache_size is 0."""
httpx_mock.add_response(
url="https://auth.example.com/oauth/introspect",
method="POST",
json={
"active": True,
"client_id": "user-123",
"scope": "read",
},
)
# Zero max_cache_size should disable caching (not raise StopIteration)
verifier = IntrospectionTokenVerifier(
introspection_url="https://auth.example.com/oauth/introspect",
client_id="test-client",
@ -617,16 +611,16 @@ class TestIntrospectionCaching:
assert result is not None
assert result.client_id == "user-123"
# Negative max_cache_size should also disable caching
verifier2 = IntrospectionTokenVerifier(
introspection_url="https://auth.example.com/oauth/introspect",
client_id="test-client",
client_secret="test-secret",
cache_ttl_seconds=300,
max_cache_size=-1,
)
result2 = await verifier2.verify_token("test-token")
assert result2 is not None
def test_negative_max_cache_size_raises(self):
"""Negative max_cache_size is a caller bug and should raise."""
with pytest.raises(ValueError, match="max_cache_size must be non-negative"):
IntrospectionTokenVerifier(
introspection_url="https://auth.example.com/oauth/introspect",
client_id="test-client",
client_secret="test-secret",
cache_ttl_seconds=300,
max_cache_size=-1,
)
async def test_cache_hit_returns_cached_result(
self, verifier_with_cache: IntrospectionTokenVerifier, httpx_mock: HTTPXMock
@ -854,9 +848,9 @@ class TestIntrospectionCaching:
def test_token_hashing(self, verifier_with_cache: IntrospectionTokenVerifier):
"""Test that tokens are hashed consistently."""
hash1 = verifier_with_cache._hash_token("test-token")
hash2 = verifier_with_cache._hash_token("test-token")
hash3 = verifier_with_cache._hash_token("different-token")
hash1 = verifier_with_cache._cache._hash_token("test-token")
hash2 = verifier_with_cache._cache._hash_token("test-token")
hash3 = verifier_with_cache._cache._hash_token("different-token")
# Same token produces same hash
assert hash1 == hash2
@ -885,8 +879,8 @@ class TestIntrospectionCaching:
await verifier_with_cache.verify_token("test-token")
# Check that cache entry uses the shorter expiration
cache_key = verifier_with_cache._hash_token("test-token")
entry = verifier_with_cache._cache[cache_key]
cache_key = verifier_with_cache._cache._hash_token("test-token")
entry = verifier_with_cache._cache._entries[cache_key]
# Cache expiration should be at or before token expiration
assert entry.expires_at <= short_exp
@ -917,8 +911,8 @@ class TestIntrospectionCaching:
assert len(httpx_mock.get_requests()) == 1
# Expire the cache entry manually
cache_key = verifier._hash_token("test-token")
verifier._cache[cache_key].expires_at = time.time() - 1
cache_key = verifier._cache._hash_token("test-token")
verifier._cache._entries[cache_key].expires_at = time.time() - 1
# Second call — cache miss, new introspection
await verifier.verify_token("test-token")
@ -944,15 +938,15 @@ class TestIntrospectionCaching:
# Fill cache to capacity
await verifier.verify_token("token-0")
await verifier.verify_token("token-1")
assert len(verifier._cache) == 2
assert len(verifier._cache._entries) == 2
# Third token should evict the oldest entry
await verifier.verify_token("token-2")
assert len(verifier._cache) == 2
assert len(verifier._cache._entries) == 2
# token-0 should have been evicted (FIFO)
hash_0 = verifier._hash_token("token-0")
assert hash_0 not in verifier._cache
hash_0 = verifier._cache._hash_token("token-0")
assert hash_0 not in verifier._cache._entries
class TestIntrospectionTokenVerifierIntegration:

4
tests/server/auth/providers/test_propelauth.py
View file

@ -133,8 +133,8 @@ class TestPropelAuthProvider:
)
assert isinstance(provider.token_verifier, IntrospectionTokenVerifier)
assert provider.token_verifier._cache_ttl == 300
assert provider.token_verifier._max_cache_size == 500
assert provider.token_verifier._cache._ttl == 300
assert provider.token_verifier._cache._max_size == 500
def test_token_introspection_overrides_http_client(self):
"""Test that http_client override is passed to the verifier."""

226
tests/utilities/test_token_cache.py Normal file
View file

@ -0,0 +1,226 @@
"""Tests for the shared TokenCache utility."""
import time
import pytest
from fastmcp.server.auth.auth import AccessToken
from fastmcp.utilities.token_cache import TokenCache
def _make_token(
*,
token: str = "tok",
client_id: str = "client-1",
scopes: list[str] | None = None,
expires_at: int | None = None,
) -> AccessToken:
return AccessToken(
token=token,
client_id=client_id,
scopes=scopes or [],
expires_at=expires_at,
)
class TestTokenCacheDisabled:
"""Verify behaviour when caching is turned off."""
@pytest.mark.parametrize(
"ttl, max_size",
[
(None, None),
(0, 100),
(300, 0),
],
)
def test_disabled_configurations(self, ttl: int | None, max_size: int | None):
cache = TokenCache(ttl_seconds=ttl, max_size=max_size)
assert not cache.enabled
def test_negative_ttl_raises(self):
with pytest.raises(ValueError, match="cache_ttl_seconds must be non-negative"):
TokenCache(ttl_seconds=-1)
def test_negative_max_size_raises(self):
with pytest.raises(ValueError, match="max_cache_size must be non-negative"):
TokenCache(max_size=-1)
def test_get_returns_miss_when_disabled(self):
cache = TokenCache(ttl_seconds=0)
cache.set("tok", _make_token())
hit, result = cache.get("tok")
assert not hit
assert result is None
def test_set_is_noop_when_disabled(self):
cache = TokenCache(ttl_seconds=0)
cache.set("tok", _make_token())
assert len(cache._entries) == 0
class TestTokenCacheEnabled:
"""Core get/set behaviour with caching on."""
@pytest.fixture
def cache(self) -> TokenCache:
return TokenCache(ttl_seconds=300, max_size=100)
def test_enabled(self, cache: TokenCache):
assert cache.enabled
def test_set_and_get(self, cache: TokenCache):
access = _make_token(client_id="user-1")
cache.set("tok-1", access)
hit, result = cache.get("tok-1")
assert hit
assert result is not None
assert result.client_id == "user-1"
def test_miss_for_unknown_token(self, cache: TokenCache):
hit, result = cache.get("unknown")
assert not hit
assert result is None
def test_different_tokens_cached_separately(self, cache: TokenCache):
cache.set("tok-a", _make_token(client_id="a"))
cache.set("tok-b", _make_token(client_id="b"))
_, a = cache.get("tok-a")
_, b = cache.get("tok-b")
assert a is not None and a.client_id == "a"
assert b is not None and b.client_id == "b"
class TestTokenCacheDefensiveCopy:
"""Mutating a returned token must not affect the cached value."""
def test_get_returns_deep_copy(self):
cache = TokenCache(ttl_seconds=300, max_size=100)
access = _make_token(client_id="orig")
access.claims = {"key": "original"}
cache.set("tok", access)
_, first = cache.get("tok")
assert first is not None
first.claims["key"] = "mutated"
first.scopes.append("admin")
_, second = cache.get("tok")
assert second is not None
assert second.claims["key"] == "original"
assert "admin" not in second.scopes
def test_mutating_source_does_not_affect_cache(self):
cache = TokenCache(ttl_seconds=300, max_size=100)
access = _make_token(client_id="orig")
access.claims = {"key": "original"}
cache.set("tok", access)
access.claims["key"] = "mutated"
_, cached = cache.get("tok")
assert cached is not None
assert cached.claims["key"] == "original"
class TestTokenCacheTTL:
"""Expiration and TTL behaviour."""
def test_expired_entry_is_evicted_on_get(self):
cache = TokenCache(ttl_seconds=300, max_size=100)
cache.set("tok", _make_token())
key = cache._hash_token("tok")
cache._entries[key].expires_at = time.time() - 1
hit, result = cache.get("tok")
assert not hit
assert result is None
assert key not in cache._entries
def test_token_expires_at_caps_ttl(self):
cache = TokenCache(ttl_seconds=300, max_size=100)
short_exp = int(time.time()) + 30
cache.set("tok", _make_token(expires_at=short_exp))
key = cache._hash_token("tok")
assert cache._entries[key].expires_at <= short_exp
def test_ttl_used_when_no_token_expiry(self):
cache = TokenCache(ttl_seconds=60, max_size=100)
before = time.time()
cache.set("tok", _make_token(expires_at=None))
after = time.time()
key = cache._hash_token("tok")
entry = cache._entries[key]
assert before + 60 <= entry.expires_at <= after + 60
class TestTokenCacheSizeLimit:
"""Eviction and size-limit behaviour."""
def test_evicts_oldest_when_full(self):
cache = TokenCache(ttl_seconds=300, max_size=2)
cache.set("tok-0", _make_token(client_id="0"))
cache.set("tok-1", _make_token(client_id="1"))
cache.set("tok-2", _make_token(client_id="2"))
assert len(cache._entries) == 2
hit_0, _ = cache.get("tok-0")
assert not hit_0
hit_1, _ = cache.get("tok-1")
hit_2, _ = cache.get("tok-2")
assert hit_1
assert hit_2
def test_cleanup_expired_before_eviction(self):
cache = TokenCache(ttl_seconds=300, max_size=2)
cache.set("tok-0", _make_token(client_id="0"))
cache.set("tok-1", _make_token(client_id="1"))
key_0 = cache._hash_token("tok-0")
cache._entries[key_0].expires_at = time.time() - 1
cache.set("tok-2", _make_token(client_id="2"))
assert len(cache._entries) == 2
hit_1, _ = cache.get("tok-1")
hit_2, _ = cache.get("tok-2")
assert hit_1
assert hit_2
def test_overwrite_does_not_evict(self):
"""Overwriting an existing key should not evict another entry."""
cache = TokenCache(ttl_seconds=300, max_size=2)
cache.set("tok-0", _make_token(client_id="0"))
cache.set("tok-1", _make_token(client_id="1"))
# Overwrite tok-0 — should NOT evict tok-1
cache.set("tok-0", _make_token(client_id="0-updated"))
assert len(cache._entries) == 2
hit_0, result_0 = cache.get("tok-0")
hit_1, _ = cache.get("tok-1")
assert hit_0
assert hit_1
assert result_0 is not None
assert result_0.client_id == "0-updated"
class TestTokenCacheHashing:
"""SHA-256 key hashing."""
def test_consistent_hashing(self):
assert TokenCache._hash_token("abc") == TokenCache._hash_token("abc")
def test_different_tokens_different_hashes(self):
assert TokenCache._hash_token("abc") != TokenCache._hash_token("xyz")
def test_hash_is_64_hex_chars(self):
h = TokenCache._hash_token("anything")
assert len(h) == 64
int(h, 16) # must be valid hex