diff --git a/pyproject.toml b/pyproject.toml index 35b538b58..a6fce4ad6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", diff --git a/src/fastmcp/server/auth/bearer.py b/src/fastmcp/server/auth/bearer.py deleted file mode 100644 index 729fcf1dc..000000000 --- a/src/fastmcp/server/auth/bearer.py +++ /dev/null @@ -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") diff --git a/src/fastmcp/server/auth/providers/bearer.py b/src/fastmcp/server/auth/providers/bearer.py index ccf25f0bd..419ed255c 100644 --- a/src/fastmcp/server/auth/providers/bearer.py +++ b/src/fastmcp/server/auth/providers/bearer.py @@ -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}") diff --git a/tests/auth/providers/test_bearer.py b/tests/auth/providers/test_bearer.py index ccc84fe73..a54f4d416 100644 --- a/tests/auth/providers/test_bearer.py +++ b/tests/auth/providers/test_bearer.py @@ -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 diff --git a/uv.lock b/uv.lock index e1cd22726..5097b2bcc 100644 --- a/uv.lock +++ b/uv.lock @@ -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"