mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Add jwks tests
This commit is contained in:
Jeremiah Lowin
parent
da9c51e132
commit
172f67c617
5 changed files with 318 additions and 293 deletions
|
|
@ -52,6 +52,7 @@ dev = [
|
|||
"pytest-cov>=6.1.1",
|
||||
"pytest-env>=1.1.5",
|
||||
"pytest-flakefinder",
|
||||
"pytest-httpx>=0.35.0",
|
||||
"pytest-report>=0.2.1",
|
||||
"pytest-timeout>=2.4.0",
|
||||
"pytest-xdist>=3.6.1",
|
||||
|
|
|
|||
|
|
@ -1,256 +0,0 @@
|
|||
"""
|
||||
Simple JWT Bearer Token validation for hosted MCP servers.
|
||||
|
||||
Uses RS256 (asymmetric) where your control plane signs with a private key
|
||||
and hosted MCP servers validate with the corresponding public key.
|
||||
|
||||
Example usage:
|
||||
# Static public key
|
||||
provider = BearerTokenValidatorProvider(
|
||||
public_key='''-----BEGIN PUBLIC KEY-----
|
||||
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
|
||||
-----END PUBLIC KEY-----''',
|
||||
issuer="https://auth.yourservice.com"
|
||||
)
|
||||
|
||||
# Or JWKS URI (recommended for production - allows key rotation)
|
||||
provider = BearerTokenValidatorProvider(
|
||||
jwks_uri="https://auth.yourservice.com/.well-known/jwks.json",
|
||||
issuer="https://auth.yourservice.com"
|
||||
)
|
||||
"""
|
||||
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from authlib.jose import JsonWebKey, JsonWebToken
|
||||
from authlib.jose.errors import JoseError
|
||||
from mcp.server.auth.provider import (
|
||||
AccessToken,
|
||||
AuthorizationCode,
|
||||
AuthorizationParams,
|
||||
RefreshToken,
|
||||
)
|
||||
from mcp.shared.auth import (
|
||||
OAuthClientInformationFull,
|
||||
OAuthToken,
|
||||
)
|
||||
|
||||
from fastmcp.server.auth.auth import (
|
||||
ClientRegistrationOptions,
|
||||
OAuthProvider,
|
||||
RevocationOptions,
|
||||
)
|
||||
|
||||
|
||||
class BearerTokenValidatorProvider(OAuthProvider):
|
||||
"""
|
||||
Simple JWT Bearer Token validator for hosted MCP servers.
|
||||
Uses RS256 asymmetric encryption. Supports either static public key
|
||||
or JWKS URI for key rotation.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
issuer: str,
|
||||
public_key: str | None = None,
|
||||
jwks_uri: str | None = None,
|
||||
audience: str | None = None,
|
||||
required_scopes: list[str] | None = None,
|
||||
):
|
||||
"""
|
||||
Initialize the provider.
|
||||
|
||||
Args:
|
||||
issuer: Expected issuer claim (your control plane)
|
||||
public_key: RSA public key in PEM format (for static key)
|
||||
jwks_uri: URI to fetch keys from (for key rotation)
|
||||
audience: Expected audience claim (optional)
|
||||
required_scopes: List of required scopes for access
|
||||
"""
|
||||
if not (public_key or jwks_uri):
|
||||
raise ValueError("Either public_key or jwks_uri must be provided")
|
||||
if public_key and jwks_uri:
|
||||
raise ValueError("Provide either public_key or jwks_uri, not both")
|
||||
|
||||
super().__init__(
|
||||
issuer_url=issuer,
|
||||
client_registration_options=ClientRegistrationOptions(enabled=False),
|
||||
revocation_options=RevocationOptions(enabled=False),
|
||||
required_scopes=required_scopes,
|
||||
)
|
||||
|
||||
self.issuer = issuer
|
||||
self.audience = audience
|
||||
self.public_key = public_key
|
||||
self.jwks_uri = jwks_uri
|
||||
self.jwt = JsonWebToken(["RS256"])
|
||||
|
||||
# Simple JWKS cache
|
||||
self._jwks_cache: dict[str, str] = {}
|
||||
self._jwks_cache_time: float = 0
|
||||
self._cache_ttl = 3600 # 1 hour
|
||||
|
||||
async def _get_verification_key(self, token: str) -> str:
|
||||
"""Get the verification key for the token."""
|
||||
if self.public_key:
|
||||
return self.public_key
|
||||
|
||||
# Extract kid from token header for JWKS lookup
|
||||
try:
|
||||
import base64
|
||||
import json
|
||||
|
||||
header_b64 = token.split(".")[0]
|
||||
header_b64 += "=" * (4 - len(header_b64) % 4) # Add padding
|
||||
header = json.loads(base64.urlsafe_b64decode(header_b64))
|
||||
kid = header.get("kid")
|
||||
|
||||
if not kid:
|
||||
raise ValueError("Token missing key ID (kid)")
|
||||
|
||||
return await self._get_jwks_key(kid)
|
||||
|
||||
except Exception as e:
|
||||
raise ValueError(f"Failed to extract key ID from token: {e}")
|
||||
|
||||
async def _get_jwks_key(self, kid: str) -> str:
|
||||
"""Fetch key from JWKS with simple caching."""
|
||||
if not self.jwks_uri:
|
||||
raise ValueError("JWKS URI not configured")
|
||||
|
||||
current_time = time.time()
|
||||
|
||||
# Check cache
|
||||
if (
|
||||
current_time - self._jwks_cache_time < self._cache_ttl
|
||||
and kid in self._jwks_cache
|
||||
):
|
||||
return self._jwks_cache[kid]
|
||||
|
||||
# Fetch JWKS
|
||||
try:
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(self.jwks_uri)
|
||||
response.raise_for_status()
|
||||
jwks_data = response.json()
|
||||
|
||||
# Cache all keys
|
||||
self._jwks_cache = {}
|
||||
for key_data in jwks_data.get("keys", []):
|
||||
key_kid = key_data.get("kid")
|
||||
if key_kid:
|
||||
jwk = JsonWebKey.import_key(key_data)
|
||||
self._jwks_cache[key_kid] = jwk.get_public_key()
|
||||
|
||||
self._jwks_cache_time = current_time
|
||||
|
||||
if kid not in self._jwks_cache:
|
||||
raise ValueError(f"Key ID '{kid}' not found in JWKS")
|
||||
|
||||
return self._jwks_cache[kid]
|
||||
|
||||
except Exception as e:
|
||||
raise ValueError(f"Failed to fetch JWKS: {e}")
|
||||
|
||||
async def load_access_token(self, token: str) -> AccessToken | None:
|
||||
"""
|
||||
Validates the provided JWT bearer token.
|
||||
|
||||
Args:
|
||||
token: The JWT token string to validate
|
||||
|
||||
Returns:
|
||||
AccessToken object if valid, None if invalid or expired
|
||||
"""
|
||||
try:
|
||||
# Get verification key (static or from JWKS)
|
||||
verification_key = await self._get_verification_key(token)
|
||||
|
||||
# Decode and verify the JWT token
|
||||
claims = self.jwt.decode(token, verification_key)
|
||||
|
||||
# Validate expiration
|
||||
exp = claims.get("exp")
|
||||
if exp and exp < time.time():
|
||||
return None
|
||||
|
||||
# Validate issuer
|
||||
if claims.get("iss") != self.issuer:
|
||||
return None
|
||||
|
||||
# Validate audience if configured
|
||||
if self.audience:
|
||||
aud = claims.get("aud")
|
||||
if isinstance(aud, list):
|
||||
if self.audience not in aud:
|
||||
return None
|
||||
elif aud != self.audience:
|
||||
return None
|
||||
|
||||
# Extract claims
|
||||
client_id = claims.get("sub") or claims.get("client_id") or "unknown"
|
||||
scopes = self._extract_scopes(claims)
|
||||
|
||||
return AccessToken(
|
||||
token=token,
|
||||
client_id=str(client_id),
|
||||
scopes=scopes,
|
||||
expires_at=int(exp) if exp else None,
|
||||
)
|
||||
|
||||
except JoseError:
|
||||
return None
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def _extract_scopes(self, claims: dict[str, Any]) -> list[str]:
|
||||
"""Extract scopes from JWT claims."""
|
||||
scope_claim = claims.get("scope", "")
|
||||
if isinstance(scope_claim, str):
|
||||
return scope_claim.split()
|
||||
elif isinstance(scope_claim, list):
|
||||
return scope_claim
|
||||
return []
|
||||
|
||||
# --- Unused OAuth server methods ---
|
||||
async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
|
||||
raise NotImplementedError("Client management not supported")
|
||||
|
||||
async def register_client(self, client_info: OAuthClientInformationFull) -> None:
|
||||
raise NotImplementedError("Client registration not supported")
|
||||
|
||||
async def authorize(
|
||||
self, client: OAuthClientInformationFull, params: AuthorizationParams
|
||||
) -> str:
|
||||
raise NotImplementedError("Authorization flow not supported")
|
||||
|
||||
async def load_authorization_code(
|
||||
self, client: OAuthClientInformationFull, authorization_code: str
|
||||
) -> AuthorizationCode | None:
|
||||
raise NotImplementedError("Authorization code flow not supported")
|
||||
|
||||
async def exchange_authorization_code(
|
||||
self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
|
||||
) -> OAuthToken:
|
||||
raise NotImplementedError("Authorization code exchange not supported")
|
||||
|
||||
async def load_refresh_token(
|
||||
self, client: OAuthClientInformationFull, refresh_token: str
|
||||
) -> RefreshToken | None:
|
||||
raise NotImplementedError("Refresh token flow not supported")
|
||||
|
||||
async def exchange_refresh_token(
|
||||
self,
|
||||
client: OAuthClientInformationFull,
|
||||
refresh_token: RefreshToken,
|
||||
scopes: list[str],
|
||||
) -> OAuthToken:
|
||||
raise NotImplementedError("Refresh token exchange not supported")
|
||||
|
||||
async def revoke_token(
|
||||
self,
|
||||
token: AccessToken | RefreshToken,
|
||||
) -> None:
|
||||
raise NotImplementedError("Token revocation not supported")
|
||||
|
|
@ -6,7 +6,7 @@ and hosted MCP servers validate with the corresponding public key.
|
|||
|
||||
Example usage:
|
||||
# Static public key
|
||||
provider = BearerTokenValidatorProvider(
|
||||
provider = BearerAuthProvider(
|
||||
public_key='''-----BEGIN PUBLIC KEY-----
|
||||
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
|
||||
-----END PUBLIC KEY-----''',
|
||||
|
|
@ -14,7 +14,7 @@ provider = BearerTokenValidatorProvider(
|
|||
)
|
||||
|
||||
# Or JWKS URI (recommended for production - allows key rotation)
|
||||
provider = BearerTokenValidatorProvider(
|
||||
provider = Bear(
|
||||
jwks_uri="https://auth.yourservice.com/.well-known/jwks.json",
|
||||
issuer="https://auth.yourservice.com"
|
||||
)
|
||||
|
|
@ -22,7 +22,7 @@ provider = BearerTokenValidatorProvider(
|
|||
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
from typing import Any, TypedDict
|
||||
|
||||
import httpx
|
||||
from authlib.jose import JsonWebKey, JsonWebToken
|
||||
|
|
@ -48,6 +48,25 @@ from fastmcp.server.auth.auth import (
|
|||
)
|
||||
|
||||
|
||||
class JWKData(TypedDict, total=False):
|
||||
"""JSON Web Key data structure."""
|
||||
|
||||
kty: str # Key type (e.g., "RSA") - required
|
||||
kid: str # Key ID (optional but recommended)
|
||||
use: str # Usage (e.g., "sig")
|
||||
alg: str # Algorithm (e.g., "RS256")
|
||||
n: str # Modulus (for RSA keys)
|
||||
e: str # Exponent (for RSA keys)
|
||||
x5c: list[str] # X.509 certificate chain (for JWKs)
|
||||
x5t: str # X.509 certificate thumbprint (for JWKs)
|
||||
|
||||
|
||||
class JWKSData(TypedDict):
|
||||
"""JSON Web Key Set data structure."""
|
||||
|
||||
keys: list[JWKData]
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True, repr=False)
|
||||
class RSAKeyPair:
|
||||
private_key: SecretStr
|
||||
|
|
@ -96,6 +115,7 @@ class RSAKeyPair:
|
|||
scopes: list[str] | None = None,
|
||||
expires_in_seconds: int = 3600,
|
||||
additional_claims: dict[str, Any] | None = None,
|
||||
kid: str | None = None,
|
||||
) -> str:
|
||||
"""
|
||||
Generate a test JWT token for testing purposes.
|
||||
|
|
@ -108,6 +128,7 @@ class RSAKeyPair:
|
|||
scopes: List of scopes to include
|
||||
expires_in_seconds: Token expiration time in seconds
|
||||
additional_claims: Any additional claims to include
|
||||
kid: Key ID for JWKS lookup (optional)
|
||||
|
||||
Returns:
|
||||
Signed JWT token string
|
||||
|
|
@ -135,6 +156,8 @@ class RSAKeyPair:
|
|||
|
||||
# Create header
|
||||
header = {"alg": "RS256"}
|
||||
if kid:
|
||||
header["kid"] = kid
|
||||
|
||||
# Sign and return token
|
||||
token_bytes = jwt.encode(
|
||||
|
|
@ -209,27 +232,25 @@ class BearerAuthProvider(OAuthProvider):
|
|||
header = json.loads(base64.urlsafe_b64decode(header_b64))
|
||||
kid = header.get("kid")
|
||||
|
||||
if not kid:
|
||||
raise ValueError("Token missing key ID (kid)")
|
||||
|
||||
return await self._get_jwks_key(kid)
|
||||
|
||||
except Exception as e:
|
||||
raise ValueError(f"Failed to extract key ID from token: {e}")
|
||||
|
||||
async def _get_jwks_key(self, kid: str) -> str:
|
||||
async def _get_jwks_key(self, kid: str | None) -> str:
|
||||
"""Fetch key from JWKS with simple caching."""
|
||||
if not self.jwks_uri:
|
||||
raise ValueError("JWKS URI not configured")
|
||||
|
||||
current_time = time.time()
|
||||
|
||||
# Check cache
|
||||
if (
|
||||
current_time - self._jwks_cache_time < self._cache_ttl
|
||||
and kid in self._jwks_cache
|
||||
):
|
||||
return self._jwks_cache[kid]
|
||||
# Check cache first
|
||||
if current_time - self._jwks_cache_time < self._cache_ttl:
|
||||
if kid and kid in self._jwks_cache:
|
||||
return self._jwks_cache[kid]
|
||||
elif not kid and len(self._jwks_cache) == 1:
|
||||
# If no kid but only one key cached, use it
|
||||
return next(iter(self._jwks_cache.values()))
|
||||
|
||||
# Fetch JWKS
|
||||
try:
|
||||
|
|
@ -242,16 +263,32 @@ class BearerAuthProvider(OAuthProvider):
|
|||
self._jwks_cache = {}
|
||||
for key_data in jwks_data.get("keys", []):
|
||||
key_kid = key_data.get("kid")
|
||||
jwk = JsonWebKey.import_key(key_data)
|
||||
public_key = jwk.get_public_key()
|
||||
|
||||
if key_kid:
|
||||
jwk = JsonWebKey.import_key(key_data)
|
||||
self._jwks_cache[key_kid] = jwk.get_public_key()
|
||||
self._jwks_cache[key_kid] = public_key
|
||||
else:
|
||||
# Key without kid - use a default identifier
|
||||
self._jwks_cache["_default"] = public_key
|
||||
|
||||
self._jwks_cache_time = current_time
|
||||
|
||||
if kid not in self._jwks_cache:
|
||||
raise ValueError(f"Key ID '{kid}' not found in JWKS")
|
||||
|
||||
return self._jwks_cache[kid]
|
||||
# Select the appropriate key
|
||||
if kid:
|
||||
if kid not in self._jwks_cache:
|
||||
raise ValueError(f"Key ID '{kid}' not found in JWKS")
|
||||
return self._jwks_cache[kid]
|
||||
else:
|
||||
# No kid in token - only allow if there's exactly one key
|
||||
if len(self._jwks_cache) == 1:
|
||||
return next(iter(self._jwks_cache.values()))
|
||||
elif len(self._jwks_cache) > 1:
|
||||
raise ValueError(
|
||||
"Multiple keys in JWKS but no key ID (kid) in token"
|
||||
)
|
||||
else:
|
||||
raise ValueError("No keys found in JWKS")
|
||||
|
||||
except Exception as e:
|
||||
raise ValueError(f"Failed to fetch JWKS: {e}")
|
||||
|
|
|
|||
|
|
@ -3,10 +3,15 @@ from typing import Any
|
|||
|
||||
import httpx
|
||||
import pytest
|
||||
from pytest_httpx import HTTPXMock
|
||||
|
||||
from fastmcp import Client, FastMCP
|
||||
from fastmcp.client.auth import BearerAuth
|
||||
from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
|
||||
from fastmcp.server.auth.providers.bearer import (
|
||||
BearerAuthProvider,
|
||||
JWKSData,
|
||||
RSAKeyPair,
|
||||
)
|
||||
from fastmcp.utilities.tests import run_server_in_process
|
||||
|
||||
|
||||
|
|
@ -103,6 +108,194 @@ class TestRSAKeyPair:
|
|||
# We'll validate the scopes in the BearerToken tests
|
||||
|
||||
|
||||
class TestBearerTokenJWKS:
|
||||
"""Tests for JWKS URI functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def jwks_provider(self, rsa_key_pair: RSAKeyPair) -> BearerAuthProvider:
|
||||
"""Provider configured with JWKS URI."""
|
||||
return BearerAuthProvider(
|
||||
jwks_uri="https://test.example.com/.well-known/jwks.json",
|
||||
issuer="https://test.example.com",
|
||||
audience="https://api.example.com",
|
||||
)
|
||||
|
||||
@pytest.fixture
|
||||
def mock_jwks_data(self, rsa_key_pair: RSAKeyPair) -> JWKSData:
|
||||
"""Create mock JWKS data from RSA key pair."""
|
||||
from authlib.jose import JsonWebKey
|
||||
|
||||
# Create JWK from the RSA public key
|
||||
jwk = JsonWebKey.import_key(rsa_key_pair.public_key)
|
||||
jwk_data = jwk.as_dict()
|
||||
jwk_data["kid"] = "test-key-1"
|
||||
jwk_data["alg"] = "RS256"
|
||||
|
||||
return {"keys": [jwk_data]}
|
||||
|
||||
async def test_jwks_token_validation(
|
||||
self,
|
||||
rsa_key_pair: RSAKeyPair,
|
||||
jwks_provider: BearerAuthProvider,
|
||||
mock_jwks_data: JWKSData,
|
||||
httpx_mock: HTTPXMock,
|
||||
):
|
||||
"""Test token validation using JWKS URI."""
|
||||
httpx_mock.add_response(
|
||||
url="https://test.example.com/.well-known/jwks.json",
|
||||
json=mock_jwks_data,
|
||||
)
|
||||
token = rsa_key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer="https://test.example.com",
|
||||
audience="https://api.example.com",
|
||||
)
|
||||
|
||||
access_token = await jwks_provider.load_access_token(token)
|
||||
assert access_token is not None
|
||||
assert access_token.client_id == "test-user"
|
||||
|
||||
async def test_jwks_token_validation_with_invalid_key(
|
||||
self,
|
||||
rsa_key_pair: RSAKeyPair,
|
||||
jwks_provider: BearerAuthProvider,
|
||||
mock_jwks_data: JWKSData,
|
||||
httpx_mock: HTTPXMock,
|
||||
):
|
||||
httpx_mock.add_response(
|
||||
url="https://test.example.com/.well-known/jwks.json",
|
||||
json=mock_jwks_data,
|
||||
)
|
||||
token = RSAKeyPair.generate().create_token(
|
||||
subject="test-user",
|
||||
issuer="https://test.example.com",
|
||||
audience="https://api.example.com",
|
||||
)
|
||||
|
||||
access_token = await jwks_provider.load_access_token(token)
|
||||
assert access_token is None
|
||||
|
||||
async def test_jwks_token_validation_with_kid(
|
||||
self,
|
||||
rsa_key_pair: RSAKeyPair,
|
||||
jwks_provider: BearerAuthProvider,
|
||||
mock_jwks_data: JWKSData,
|
||||
httpx_mock: HTTPXMock,
|
||||
):
|
||||
mock_jwks_data["keys"][0]["kid"] = "test-key-1"
|
||||
httpx_mock.add_response(
|
||||
url="https://test.example.com/.well-known/jwks.json",
|
||||
json=mock_jwks_data,
|
||||
)
|
||||
token = rsa_key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer="https://test.example.com",
|
||||
audience="https://api.example.com",
|
||||
kid="test-key-1",
|
||||
)
|
||||
|
||||
access_token = await jwks_provider.load_access_token(token)
|
||||
assert access_token is not None
|
||||
assert access_token.client_id == "test-user"
|
||||
|
||||
async def test_jwks_token_validation_with_kid_and_no_kid_in_token(
|
||||
self,
|
||||
rsa_key_pair: RSAKeyPair,
|
||||
jwks_provider: BearerAuthProvider,
|
||||
mock_jwks_data: JWKSData,
|
||||
httpx_mock: HTTPXMock,
|
||||
):
|
||||
mock_jwks_data["keys"][0]["kid"] = "test-key-1"
|
||||
httpx_mock.add_response(
|
||||
url="https://test.example.com/.well-known/jwks.json",
|
||||
json=mock_jwks_data,
|
||||
)
|
||||
token = rsa_key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer="https://test.example.com",
|
||||
audience="https://api.example.com",
|
||||
)
|
||||
|
||||
access_token = await jwks_provider.load_access_token(token)
|
||||
assert access_token is not None
|
||||
assert access_token.client_id == "test-user"
|
||||
|
||||
async def test_jwks_token_validation_with_no_kid_and_kid_in_jwks(
|
||||
self,
|
||||
rsa_key_pair: RSAKeyPair,
|
||||
jwks_provider: BearerAuthProvider,
|
||||
mock_jwks_data: JWKSData,
|
||||
httpx_mock: HTTPXMock,
|
||||
):
|
||||
mock_jwks_data["keys"][0]["kid"] = "test-key-1"
|
||||
httpx_mock.add_response(
|
||||
url="https://test.example.com/.well-known/jwks.json",
|
||||
json=mock_jwks_data,
|
||||
)
|
||||
token = rsa_key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer="https://test.example.com",
|
||||
audience="https://api.example.com",
|
||||
)
|
||||
|
||||
access_token = await jwks_provider.load_access_token(token)
|
||||
assert access_token is not None
|
||||
assert access_token.client_id == "test-user"
|
||||
|
||||
async def test_jwks_token_validation_with_kid_mismatch(
|
||||
self,
|
||||
rsa_key_pair: RSAKeyPair,
|
||||
jwks_provider: BearerAuthProvider,
|
||||
mock_jwks_data: JWKSData,
|
||||
httpx_mock: HTTPXMock,
|
||||
):
|
||||
mock_jwks_data["keys"][0]["kid"] = "test-key-1"
|
||||
httpx_mock.add_response(
|
||||
url="https://test.example.com/.well-known/jwks.json",
|
||||
json=mock_jwks_data,
|
||||
)
|
||||
token = rsa_key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer="https://test.example.com",
|
||||
audience="https://api.example.com",
|
||||
kid="test-key-2",
|
||||
)
|
||||
|
||||
access_token = await jwks_provider.load_access_token(token)
|
||||
assert access_token is None
|
||||
|
||||
async def test_jwks_token_validation_with_multiple_keys_and_no_kid_in_token(
|
||||
self,
|
||||
rsa_key_pair: RSAKeyPair,
|
||||
jwks_provider: BearerAuthProvider,
|
||||
mock_jwks_data: JWKSData,
|
||||
httpx_mock: HTTPXMock,
|
||||
):
|
||||
mock_jwks_data["keys"] = [
|
||||
{
|
||||
"kid": "test-key-1",
|
||||
"alg": "RS256",
|
||||
},
|
||||
{
|
||||
"kid": "test-key-2",
|
||||
"alg": "RS256",
|
||||
},
|
||||
]
|
||||
|
||||
httpx_mock.add_response(
|
||||
url="https://test.example.com/.well-known/jwks.json",
|
||||
json=mock_jwks_data,
|
||||
)
|
||||
token = rsa_key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer="https://test.example.com",
|
||||
audience="https://api.example.com",
|
||||
)
|
||||
|
||||
access_token = await jwks_provider.load_access_token(token)
|
||||
assert access_token is None
|
||||
|
||||
|
||||
class TestBearerToken:
|
||||
def test_initialization_with_public_key(self, rsa_key_pair: RSAKeyPair):
|
||||
"""Test provider initialization with public key."""
|
||||
|
|
@ -143,7 +336,6 @@ class TestBearerToken:
|
|||
issuer="https://test.example.com",
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_valid_token_validation(
|
||||
self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
|
||||
):
|
||||
|
|
@ -163,7 +355,6 @@ class TestBearerToken:
|
|||
assert "write" in access_token.scopes
|
||||
assert access_token.expires_at is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_expired_token_rejection(
|
||||
self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
|
||||
):
|
||||
|
|
@ -178,7 +369,6 @@ class TestBearerToken:
|
|||
access_token = await bearer_provider.load_access_token(token)
|
||||
assert access_token is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_issuer_rejection(
|
||||
self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
|
||||
):
|
||||
|
|
@ -192,7 +382,6 @@ class TestBearerToken:
|
|||
access_token = await bearer_provider.load_access_token(token)
|
||||
assert access_token is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_audience_rejection(
|
||||
self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
|
||||
):
|
||||
|
|
@ -206,7 +395,6 @@ class TestBearerToken:
|
|||
access_token = await bearer_provider.load_access_token(token)
|
||||
assert access_token is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_issuer_validation_when_none(self, rsa_key_pair: RSAKeyPair):
|
||||
"""Test that issuer validation is skipped when provider has no issuer configured."""
|
||||
provider = BearerAuthProvider(
|
||||
|
|
@ -221,7 +409,6 @@ class TestBearerToken:
|
|||
access_token = await provider.load_access_token(token)
|
||||
assert access_token is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_audience_validation_when_none(self, rsa_key_pair: RSAKeyPair):
|
||||
"""Test that audience validation is skipped when provider has no audience configured."""
|
||||
provider = BearerAuthProvider(
|
||||
|
|
@ -239,7 +426,6 @@ class TestBearerToken:
|
|||
access_token = await provider.load_access_token(token)
|
||||
assert access_token is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiple_audiences_validation(self, rsa_key_pair: RSAKeyPair):
|
||||
"""Test validation with multiple audiences in token."""
|
||||
provider = BearerAuthProvider(
|
||||
|
|
@ -259,7 +445,6 @@ class TestBearerToken:
|
|||
access_token = await provider.load_access_token(token)
|
||||
assert access_token is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scope_extraction_string(
|
||||
self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
|
||||
):
|
||||
|
|
@ -276,7 +461,6 @@ class TestBearerToken:
|
|||
assert access_token is not None
|
||||
assert set(access_token.scopes) == {"read", "write", "admin"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scope_extraction_list(
|
||||
self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
|
||||
):
|
||||
|
|
@ -293,7 +477,6 @@ class TestBearerToken:
|
|||
assert access_token is not None
|
||||
assert set(access_token.scopes) == {"read", "write"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_scopes(
|
||||
self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
|
||||
):
|
||||
|
|
@ -310,7 +493,6 @@ class TestBearerToken:
|
|||
assert access_token is not None
|
||||
assert access_token.scopes == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_malformed_token_rejection(self, bearer_provider: BearerAuthProvider):
|
||||
"""Test rejection of malformed tokens."""
|
||||
malformed_tokens = [
|
||||
|
|
@ -325,7 +507,6 @@ class TestBearerToken:
|
|||
access_token = await bearer_provider.load_access_token(token)
|
||||
assert access_token is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_signature_rejection(
|
||||
self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
|
||||
):
|
||||
|
|
@ -341,7 +522,6 @@ class TestBearerToken:
|
|||
access_token = await bearer_provider.load_access_token(token)
|
||||
assert access_token is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_id_fallback(
|
||||
self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
|
||||
):
|
||||
|
|
@ -367,9 +547,10 @@ class TestFastMCPBearerAuth:
|
|||
assert isinstance(mcp.auth, BearerAuthProvider)
|
||||
|
||||
async def test_unauthorized_access(self, mcp_server_url: str):
|
||||
with pytest.raises(httpx.HTTPStatusError, match="401"):
|
||||
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
||||
async with Client(mcp_server_url) as client:
|
||||
tools = await client.list_tools() # noqa: F841
|
||||
assert exc_info.value.response.status_code == 401
|
||||
assert "tools" not in locals()
|
||||
|
||||
async def test_authorized_access(self, mcp_server_url: str, bearer_token):
|
||||
|
|
@ -378,9 +559,10 @@ class TestFastMCPBearerAuth:
|
|||
assert tools
|
||||
|
||||
async def test_invalid_token_raises_401(self, mcp_server_url: str):
|
||||
with pytest.raises(httpx.HTTPStatusError, match="401"):
|
||||
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
||||
async with Client(mcp_server_url, auth=BearerAuth("invalid")) as client:
|
||||
tools = await client.list_tools() # noqa: F841
|
||||
assert exc_info.value.response.status_code == 401
|
||||
assert "tools" not in locals()
|
||||
|
||||
async def test_expired_token(self, mcp_server_url: str, rsa_key_pair: RSAKeyPair):
|
||||
|
|
@ -391,16 +573,62 @@ class TestFastMCPBearerAuth:
|
|||
expires_in_seconds=-3600,
|
||||
)
|
||||
|
||||
with pytest.raises(httpx.HTTPStatusError, match="401"):
|
||||
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
||||
async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
|
||||
tools = await client.list_tools() # noqa: F841
|
||||
assert exc_info.value.response.status_code == 401
|
||||
assert "tools" not in locals()
|
||||
|
||||
async def test_token_with_bad_signature(self, mcp_server_url: str):
|
||||
rsa_key_pair = RSAKeyPair.generate()
|
||||
token = rsa_key_pair.create_token()
|
||||
|
||||
with pytest.raises(httpx.HTTPStatusError, match="401"):
|
||||
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
||||
async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
|
||||
tools = await client.list_tools() # noqa: F841
|
||||
assert exc_info.value.response.status_code == 401
|
||||
assert "tools" not in locals()
|
||||
|
||||
async def test_token_with_insufficient_scopes(
|
||||
self, mcp_server_url: str, rsa_key_pair: RSAKeyPair
|
||||
):
|
||||
token = rsa_key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer="https://test.example.com",
|
||||
audience="https://api.example.com",
|
||||
scopes=["read"],
|
||||
)
|
||||
|
||||
with run_server_in_process(
|
||||
run_mcp_server,
|
||||
public_key=rsa_key_pair.public_key,
|
||||
auth_kwargs=dict(required_scopes=["read", "write"]),
|
||||
run_kwargs=dict(transport="streamable-http"),
|
||||
) as url:
|
||||
mcp_server_url = f"{url}/mcp"
|
||||
with pytest.raises(httpx.HTTPStatusError) as exc_info:
|
||||
async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
|
||||
tools = await client.list_tools() # noqa: F841
|
||||
assert exc_info.value.response.status_code == 403
|
||||
assert "tools" not in locals()
|
||||
|
||||
async def test_token_with_sufficient_scopes(
|
||||
self, mcp_server_url: str, rsa_key_pair: RSAKeyPair
|
||||
):
|
||||
token = rsa_key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer="https://test.example.com",
|
||||
audience="https://api.example.com",
|
||||
scopes=["read", "write"],
|
||||
)
|
||||
|
||||
with run_server_in_process(
|
||||
run_mcp_server,
|
||||
public_key=rsa_key_pair.public_key,
|
||||
auth_kwargs=dict(required_scopes=["read", "write"]),
|
||||
run_kwargs=dict(transport="streamable-http"),
|
||||
) as url:
|
||||
mcp_server_url = f"{url}/mcp"
|
||||
async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
|
||||
tools = await client.list_tools()
|
||||
assert tools
|
||||
|
|
|
|||
15
uv.lock
generated
15
uv.lock
generated
|
|
@ -456,6 +456,7 @@ dev = [
|
|||
{ name = "pytest-cov" },
|
||||
{ name = "pytest-env" },
|
||||
{ name = "pytest-flakefinder" },
|
||||
{ name = "pytest-httpx" },
|
||||
{ name = "pytest-report" },
|
||||
{ name = "pytest-timeout" },
|
||||
{ name = "pytest-xdist" },
|
||||
|
|
@ -490,6 +491,7 @@ dev = [
|
|||
{ name = "pytest-cov", specifier = ">=6.1.1" },
|
||||
{ name = "pytest-env", specifier = ">=1.1.5" },
|
||||
{ name = "pytest-flakefinder" },
|
||||
{ name = "pytest-httpx", specifier = ">=0.35.0" },
|
||||
{ name = "pytest-report", specifier = ">=0.2.1" },
|
||||
{ name = "pytest-timeout", specifier = ">=2.4.0" },
|
||||
{ name = "pytest-xdist", specifier = ">=3.6.1" },
|
||||
|
|
@ -1160,6 +1162,19 @@ wheels = [
|
|||
{ url = "https://files.pythonhosted.org/packages/33/8b/06787150d0fd0cbd3a8054262b56f91631c7778c1bc91bf4637e47f909ad/pytest_flakefinder-1.1.0-py2.py3-none-any.whl", hash = "sha256:741e0e8eea427052f5b8c89c2b3c3019a50c39a59ce4df6a305a2c2d9ba2bd13", size = 4644, upload-time = "2022-10-26T18:27:52.128Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pytest-httpx"
|
||||
version = "0.35.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "httpx" },
|
||||
{ name = "pytest" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/1f/89/5b12b7b29e3d0af3a4b9c071ee92fa25a9017453731a38f08ba01c280f4c/pytest_httpx-0.35.0.tar.gz", hash = "sha256:d619ad5d2e67734abfbb224c3d9025d64795d4b8711116b1a13f72a251ae511f", size = 54146, upload-time = "2024-11-28T19:16:54.237Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/b0/ed/026d467c1853dd83102411a78126b4842618e86c895f93528b0528c7a620/pytest_httpx-0.35.0-py3-none-any.whl", hash = "sha256:ee11a00ffcea94a5cbff47af2114d34c5b231c326902458deed73f9c459fd744", size = 19442, upload-time = "2024-11-28T19:16:52.787Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pytest-report"
|
||||
version = "0.2.1"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue