diff --git a/src/fastmcp/server/auth/providers/aws.py b/src/fastmcp/server/auth/providers/aws.py index fc5ca0666..aa654e966 100644 --- a/src/fastmcp/server/auth/providers/aws.py +++ b/src/fastmcp/server/auth/providers/aws.py @@ -38,15 +38,39 @@ logger = get_logger(__name__) class AWSCognitoTokenVerifier(JWTVerifier): - """Token verifier that filters claims to Cognito-specific subset.""" + """Token verifier for Cognito access tokens. + + Cognito access tokens use a ``client_id`` claim instead of the + standard ``aud`` claim. This subclass passes ``audience=None`` + to the parent (skipping the ``aud`` check) and validates the + ``client_id`` claim directly. + """ + + def __init__(self, *, audience: str | list[str] | None = None, **kwargs): + self._expected_client_id = audience + super().__init__(audience=None, **kwargs) async def verify_token(self, token: str) -> AccessToken | None: """Verify token and filter claims to Cognito-specific subset.""" - # Use base JWT verification access_token = await super().verify_token(token) if not access_token: return None + # Validate client_id claim (Cognito's equivalent of aud) + if self._expected_client_id: + token_client_id = access_token.claims.get("client_id") + if isinstance(self._expected_client_id, list): + valid = token_client_id in self._expected_client_id + else: + valid = token_client_id == self._expected_client_id + if not valid: + self.logger.debug( + "Token validation failed: client_id mismatch (expected %s, got %s)", + self._expected_client_id, + token_client_id, + ) + return None + # Filter claims to Cognito-specific subset cognito_claims = { "sub": access_token.claims.get("sub"), @@ -54,7 +78,6 @@ class AWSCognitoTokenVerifier(JWTVerifier): "cognito:groups": access_token.claims.get("cognito:groups", []), } - # Return new AccessToken with filtered claims return AccessToken( token=access_token.token, client_id=access_token.client_id, diff --git a/tests/server/auth/providers/test_aws.py b/tests/server/auth/providers/test_aws.py index 8bbde78f7..4f5232b53 100644 --- a/tests/server/auth/providers/test_aws.py +++ b/tests/server/auth/providers/test_aws.py @@ -103,8 +103,8 @@ class TestAWSCognitoProvider: assert provider._upstream_token_endpoint is not None assert "amazoncognito.com" in provider._upstream_authorization_endpoint - def test_token_verifier_defaults_audience_to_client_id(self): - """Test Cognito token verifier enforces the configured client ID by default.""" + def test_token_verifier_checks_client_id_not_aud(self): + """Cognito verifier should check client_id claim, not aud.""" with mock_cognito_oidc_discovery(): provider = AWSCognitoProvider( user_pool_id="us-east-1_XXXXXXXXX", @@ -116,10 +116,11 @@ class TestAWSCognitoProvider: verifier = provider.get_token_verifier() - assert verifier.audience == "test_client" + assert verifier._expected_client_id == "test_client" + assert verifier.audience is None def test_token_verifier_supports_audience_override(self): - """Test Cognito token verifier still allows explicit audience overrides.""" + """Audience param maps to client_id validation in Cognito verifier.""" with mock_cognito_oidc_discovery(): provider = AWSCognitoProvider( user_pool_id="us-east-1_XXXXXXXXX", @@ -131,7 +132,8 @@ class TestAWSCognitoProvider: verifier = provider.get_token_verifier(audience="custom-audience") - assert verifier.audience == "custom-audience" + assert verifier._expected_client_id == "custom-audience" + assert verifier.audience is None # Token verification functionality is now tested as part of the OIDC provider integration