diff --git a/fastmcp_slim/fastmcp/server/auth/providers/jwt.py b/fastmcp_slim/fastmcp/server/auth/providers/jwt.py index 6ed151fdf..9167eb9d1 100644 --- a/fastmcp_slim/fastmcp/server/auth/providers/jwt.py +++ b/fastmcp_slim/fastmcp/server/auth/providers/jwt.py @@ -347,11 +347,26 @@ class JWTVerifier(TokenVerifier): try: jwks_data = await self._fetch_jwks() - # Cache all keys + # Cache all usable keys. A key that cannot be converted (e.g. an + # unsupported kty like OKP/Ed25519) is skipped rather than failing + # the whole set — per RFC 7517 §5, clients should ignore JWKs they + # don't understand. Otherwise one exotic key published by the + # authorization server would reject every token, including ones + # signed by supported keys in the same set (#4515). self._jwks_cache = {} + skipped_kids: set[str] = set() for key_data in jwks_data.get("keys", []): + if not isinstance(key_data, dict): + self.logger.debug("Skipping non-object JWKS entry: %r", key_data) + continue key_kid = key_data.get("kid") - public_key = _jwk_to_pem(key_data) + try: + public_key = _jwk_to_pem(key_data) + except (JoseError, TypeError, KeyError, ValueError) as e: + self.logger.debug("Skipping unusable JWKS key %r: %s", key_kid, e) + if key_kid: + skipped_kids.add(key_kid) + continue if key_kid: self._jwks_cache[key_kid] = public_key @@ -364,6 +379,16 @@ class JWTVerifier(TokenVerifier): # Select the appropriate key if kid: if kid not in self._jwks_cache: + if kid in skipped_kids: + self.logger.debug( + "JWKS key lookup failed: key ID '%s' is present " + "but its key type is unsupported", + kid, + ) + raise ValueError( + f"Key ID '{kid}' found in JWKS but its key type " + "is unsupported" + ) self.logger.debug( "JWKS key lookup failed: key ID '%s' not found", kid ) diff --git a/tests/server/auth/test_jwt_provider.py b/tests/server/auth/test_jwt_provider.py index 8ccb1f37c..1961a0c18 100644 --- a/tests/server/auth/test_jwt_provider.py +++ b/tests/server/auth/test_jwt_provider.py @@ -1,6 +1,6 @@ import time from collections.abc import AsyncGenerator -from typing import Any +from typing import Any, cast from unittest.mock import patch import pytest @@ -559,6 +559,85 @@ class TestBearerTokenJWKS: assert access_token.claims.get("iss") == issuer assert access_token.claims.get("aud") == audience + async def test_jwks_skips_unsupported_key_types( + self, + rsa_key_pair: RSAKeyPair, + jwks_provider: JWTVerifier, + mock_jwks_data: JWKSData, + httpx_mock: HTTPXMock, + mock_dns, + ): + """An unsupported key type in the JWKS (e.g. OKP/Ed25519) must be + skipped, not poison the whole key set - #4515. + + Some authorization servers (e.g. Rauthy, Ory Hydra) publish an + Ed25519 key alongside RSA keys; tokens signed by the RSA keys must + still verify. + """ + okp_key = cast( + "JWKData", + { + "kty": "OKP", + "crv": "Ed25519", + "kid": "ed25519-key", + "alg": "EdDSA", + "use": "sig", + "x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo", + }, + ) + mock_jwks_data["keys"][0]["kid"] = "test-key-1" + # Unsupported key FIRST, so an unguarded conversion loop would + # abort before reaching the RSA key the token needs + mock_jwks_data["keys"].insert(0, okp_key) + httpx_mock.add_response(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_with_only_unsupported_keys_rejects_cleanly( + self, + rsa_key_pair: RSAKeyPair, + jwks_provider: JWTVerifier, + httpx_mock: HTTPXMock, + mock_dns, + ): + """If every key in the JWKS is unsupported, verification fails + cleanly (returns None) rather than crashing - #4515.""" + okp_only = { + "keys": [ + cast( + "JWKData", + { + "kty": "OKP", + "crv": "Ed25519", + "kid": "ed25519-key", + "alg": "EdDSA", + "use": "sig", + "x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo", + }, + ) + ] + } + httpx_mock.add_response(json=okp_only) + + token = rsa_key_pair.create_token( + subject="test-user", + issuer="https://test.example.com", + audience="https://api.example.com", + kid="ed25519-key", + ) + + access_token = await jwks_provider.load_access_token(token) + assert access_token is None + async def test_jwks_token_validation_with_invalid_key( self, rsa_key_pair: RSAKeyPair,