coverage increase

Author: pcarleton

src/mcp/client/auth/extensions/client_credentials.py

@@ -83,14 +83,14 @@ def __init__(
 
     async def _exchange_token_authorization_code(
         self, auth_code: str, code_verifier: str, *, token_data: dict[str, Any] | None = None
-    ) -> httpx.Request:  # pragma: no cover
+    ) -> httpx.Request:
         """Build token exchange request for authorization_code flow."""
         token_data = token_data or {}
         if self.context.client_metadata.token_endpoint_auth_method == "private_key_jwt":
             self._add_client_authentication_jwt(token_data=token_data)
         return await super()._exchange_token_authorization_code(auth_code, code_verifier, token_data=token_data)
 
-    async def _perform_authorization(self) -> httpx.Request:  # pragma: no cover
+    async def _perform_authorization(self) -> httpx.Request:
         """Perform the authorization flow."""
         if "client_credentials" in self.context.client_metadata.grant_types:
             # SEP-1046: client_credentials grant with private_key_jwt authentication
@@ -103,12 +103,12 @@ async def _perform_authorization(self) -> httpx.Request:  # pragma: no cover
         else:
             return await super()._perform_authorization()
 
-    def _add_client_authentication_jwt(self, *, token_data: dict[str, Any]):  # pragma: no cover
+    def _add_client_authentication_jwt(self, *, token_data: dict[str, Any]):
         """Add JWT assertion for client authentication to token endpoint parameters."""
         if not self.jwt_parameters:
-            raise OAuthTokenError("Missing JWT parameters for private_key_jwt flow")
+            raise OAuthTokenError("Missing JWT parameters for private_key_jwt flow")  # pragma: no cover
         if not self.context.oauth_metadata:
-            raise OAuthTokenError("Missing OAuth metadata for private_key_jwt flow")
+            raise OAuthTokenError("Missing OAuth metadata for private_key_jwt flow")  # pragma: no cover
 
         # We need to set the audience to the issuer identifier of the authorization server
         # https://datatracker.ietf.org/doc/html/draft-ietf-oauth-rfc7523bis-01#name-updates-to-rfc-7523
@@ -122,7 +122,7 @@ def _add_client_authentication_jwt(self, *, token_data: dict[str, Any]):  # prag
         # it represents the resource server that will validate the token
         token_data["audience"] = self.context.get_resource_url()
 
-    async def _exchange_token_client_credentials(self) -> httpx.Request:  # pragma: no cover
+    async def _exchange_token_client_credentials(self) -> httpx.Request:
         """Build token exchange request for client_credentials grant.
 
         This implements SEP-1046: OAuth Client Credentials Extension for MCP.

tests/client/auth/extensions/test_client_credentials.py

@@ -1,3 +1,4 @@
+import base64
 import urllib.parse
 
 import jwt
@@ -161,3 +162,292 @@ async def test_token_exchange_request_jwt(self, rfc7523_oauth_provider: RFC7523O
         assert claims["name"] == "John Doe"
         assert claims["admin"]
         assert claims["iat"] == 1516239022
+
+    @pytest.mark.anyio
+    async def test_exchange_token_client_credentials_with_private_key_jwt(
+        self, rfc7523_oauth_provider: RFC7523OAuthClientProvider
+    ):
+        """Test client_credentials token exchange with private_key_jwt authentication."""
+        # Set up required context for client_credentials with private_key_jwt
+        rfc7523_oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id="test-client",
+            grant_types=["client_credentials"],
+            token_endpoint_auth_method="private_key_jwt",
+            redirect_uris=[AnyUrl("http://localhost:0/unused")],
+            scope="read write",
+        )
+        rfc7523_oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        )
+        rfc7523_oauth_provider.context.client_metadata = rfc7523_oauth_provider.context.client_info
+        rfc7523_oauth_provider.context.protocol_version = "2025-06-18"
+        rfc7523_oauth_provider.jwt_parameters = JWTParameters(
+            issuer="test-client",
+            subject="test-client",
+            jwt_signing_algorithm="HS256",
+            jwt_signing_key="a-string-secret-at-least-256-bits-long",
+            jwt_lifetime_seconds=300,
+        )
+
+        request = await rfc7523_oauth_provider._exchange_token_client_credentials()
+
+        assert request.method == "POST"
+        assert str(request.url) == "https://auth.example.com/token"
+        assert request.headers["Content-Type"] == "application/x-www-form-urlencoded"
+
+        # Check form data
+        content = urllib.parse.unquote_plus(request.content.decode())
+        assert "grant_type=client_credentials" in content
+        assert "scope=read write" in content
+        assert "resource=https://api.example.com/v1/mcp" in content
+        assert "client_assertion=" in content
+        assert "client_assertion_type=urn:ietf:params:oauth:client-assertion-type:jwt-bearer" in content
+
+    @pytest.mark.anyio
+    async def test_exchange_token_client_credentials_with_client_secret_basic(
+        self, rfc7523_oauth_provider: RFC7523OAuthClientProvider
+    ):
+        """Test client_credentials token exchange with client_secret_basic authentication."""
+        # Set up required context for client_credentials with client_secret_basic
+        rfc7523_oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id="test-client",
+            client_secret="test-secret",
+            grant_types=["client_credentials"],
+            token_endpoint_auth_method="client_secret_basic",
+            redirect_uris=[AnyUrl("http://localhost:0/unused")],
+            scope="read write",
+        )
+        rfc7523_oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        )
+        rfc7523_oauth_provider.context.client_metadata = rfc7523_oauth_provider.context.client_info
+        rfc7523_oauth_provider.context.protocol_version = "2025-06-18"
+        # No JWT parameters needed for client_secret_basic
+
+        request = await rfc7523_oauth_provider._exchange_token_client_credentials()
+
+        assert request.method == "POST"
+        assert str(request.url) == "https://auth.example.com/token"
+        assert request.headers["Content-Type"] == "application/x-www-form-urlencoded"
+
+        # Check Authorization header (Basic auth)
+        assert "Authorization" in request.headers
+        auth_header = request.headers["Authorization"]
+        assert auth_header.startswith("Basic ")
+
+        # Decode and verify credentials
+        encoded_creds = auth_header[6:]  # Remove "Basic " prefix
+        decoded = base64.b64decode(encoded_creds).decode()
+        assert decoded == "test-client:test-secret"
+
+        # Check form data
+        content = urllib.parse.unquote_plus(request.content.decode())
+        assert "grant_type=client_credentials" in content
+        assert "scope=read write" in content
+        assert "resource=https://api.example.com/v1/mcp" in content
+        # client_secret should NOT be in body for client_secret_basic
+        assert "client_secret=" not in content
+
+    @pytest.mark.anyio
+    async def test_perform_authorization_routes_to_client_credentials(
+        self, rfc7523_oauth_provider: RFC7523OAuthClientProvider
+    ):
+        """Test that _perform_authorization routes to client_credentials when configured."""
+        # Set up required context for client_credentials flow
+        rfc7523_oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id="test-client",
+            client_secret="test-secret",
+            grant_types=["client_credentials"],
+            token_endpoint_auth_method="client_secret_basic",
+            redirect_uris=[AnyUrl("http://localhost:0/unused")],
+        )
+        rfc7523_oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        )
+        rfc7523_oauth_provider.context.client_metadata = rfc7523_oauth_provider.context.client_info
+        rfc7523_oauth_provider.context.protocol_version = "2025-06-18"
+
+        request = await rfc7523_oauth_provider._perform_authorization()
+
+        # Should route to client_credentials flow
+        content = urllib.parse.unquote_plus(request.content.decode())
+        assert "grant_type=client_credentials" in content
+
+    @pytest.mark.anyio
+    async def test_perform_authorization_routes_to_jwt_bearer(
+        self, rfc7523_oauth_provider: RFC7523OAuthClientProvider
+    ):
+        """Test that _perform_authorization routes to jwt-bearer when configured."""
+        # Set up required context for jwt-bearer flow
+        rfc7523_oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id="test-client",
+            grant_types=["urn:ietf:params:oauth:grant-type:jwt-bearer"],
+            token_endpoint_auth_method="private_key_jwt",
+            redirect_uris=[AnyUrl("http://localhost:0/unused")],
+        )
+        rfc7523_oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        )
+        rfc7523_oauth_provider.context.client_metadata = rfc7523_oauth_provider.context.client_info
+        rfc7523_oauth_provider.context.protocol_version = "2025-06-18"
+        rfc7523_oauth_provider.jwt_parameters = JWTParameters(
+            issuer="test-client",
+            subject="test-client",
+            jwt_signing_algorithm="HS256",
+            jwt_signing_key="a-string-secret-at-least-256-bits-long",
+        )
+
+        request = await rfc7523_oauth_provider._perform_authorization()
+
+        # Should route to jwt-bearer flow
+        content = urllib.parse.unquote_plus(request.content.decode())
+        assert "grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer" in content
+
+    @pytest.mark.anyio
+    async def test_add_client_authentication_jwt(
+        self, rfc7523_oauth_provider: RFC7523OAuthClientProvider
+    ):
+        """Test _add_client_authentication_jwt adds correct JWT assertion parameters."""
+        # Set up required context
+        rfc7523_oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        )
+        rfc7523_oauth_provider.jwt_parameters = JWTParameters(
+            issuer="test-client",
+            subject="test-client",
+            jwt_signing_algorithm="HS256",
+            jwt_signing_key="a-string-secret-at-least-256-bits-long",
+        )
+
+        token_data: dict = {}
+        rfc7523_oauth_provider._add_client_authentication_jwt(token_data=token_data)
+
+        # Check that JWT assertion parameters were added
+        assert "client_assertion" in token_data
+        assert token_data["client_assertion_type"] == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
+        assert token_data["audience"] == "https://api.example.com/v1/mcp"
+
+        # Verify the JWT assertion is valid and has correct audience (issuer identifier)
+        claims = jwt.decode(
+            token_data["client_assertion"],
+            key="a-string-secret-at-least-256-bits-long",
+            algorithms=["HS256"],
+            audience="https://auth.example.com/",  # Should be issuer, not token endpoint
+            verify=True,
+        )
+        assert claims["iss"] == "test-client"
+        assert claims["sub"] == "test-client"
+
+    @pytest.mark.anyio
+    async def test_exchange_token_authorization_code_with_private_key_jwt(
+        self, rfc7523_oauth_provider: RFC7523OAuthClientProvider
+    ):
+        """Test authorization_code token exchange adds JWT when using private_key_jwt."""
+        # Set up required context
+        rfc7523_oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id="test-client",
+            grant_types=["authorization_code"],
+            token_endpoint_auth_method="private_key_jwt",
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        )
+        rfc7523_oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        )
+        rfc7523_oauth_provider.context.client_metadata = rfc7523_oauth_provider.context.client_info
+        rfc7523_oauth_provider.context.protocol_version = "2025-06-18"
+        rfc7523_oauth_provider.jwt_parameters = JWTParameters(
+            issuer="test-client",
+            subject="test-client",
+            jwt_signing_algorithm="HS256",
+            jwt_signing_key="a-string-secret-at-least-256-bits-long",
+        )
+
+        request = await rfc7523_oauth_provider._exchange_token_authorization_code(
+            "test-auth-code", "test-verifier"
+        )
+
+        # Check form data contains JWT assertion
+        content = urllib.parse.unquote_plus(request.content.decode())
+        assert "grant_type=authorization_code" in content
+        assert "code=test-auth-code" in content
+        assert "client_assertion=" in content
+        assert "client_assertion_type=urn:ietf:params:oauth:client-assertion-type:jwt-bearer" in content
+
+    @pytest.mark.anyio
+    async def test_exchange_token_authorization_code_without_private_key_jwt(
+        self, rfc7523_oauth_provider: RFC7523OAuthClientProvider
+    ):
+        """Test authorization_code token exchange without private_key_jwt uses standard auth."""
+        # Set up required context with client_secret_post (not private_key_jwt)
+        rfc7523_oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id="test-client",
+            client_secret="test-secret",
+            grant_types=["authorization_code"],
+            token_endpoint_auth_method="client_secret_post",
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        )
+        rfc7523_oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        )
+        rfc7523_oauth_provider.context.client_metadata = rfc7523_oauth_provider.context.client_info
+        rfc7523_oauth_provider.context.protocol_version = "2025-06-18"
+
+        request = await rfc7523_oauth_provider._exchange_token_authorization_code(
+            "test-auth-code", "test-verifier"
+        )
+
+        # Check form data does NOT contain JWT assertion
+        content = urllib.parse.unquote_plus(request.content.decode())
+        assert "grant_type=authorization_code" in content
+        assert "code=test-auth-code" in content
+        assert "client_assertion=" not in content
+        # Should have client_secret in body for client_secret_post
+        assert "client_secret=test-secret" in content
+
+    @pytest.mark.anyio
+    async def test_perform_authorization_falls_back_to_parent(
+        self, rfc7523_oauth_provider: RFC7523OAuthClientProvider
+    ):
+        """Test that _perform_authorization falls back to parent when not client_credentials or jwt-bearer."""
+        # Set up required context with authorization_code grant (not client_credentials or jwt-bearer)
+        rfc7523_oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id="test-client",
+            client_secret="test-secret",
+            grant_types=["authorization_code", "refresh_token"],
+            token_endpoint_auth_method="client_secret_post",
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        )
+        rfc7523_oauth_provider.context.oauth_metadata = OAuthMetadata(
+            issuer=AnyHttpUrl("https://auth.example.com"),
+            authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+            token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        )
+        rfc7523_oauth_provider.context.client_metadata = rfc7523_oauth_provider.context.client_info
+        rfc7523_oauth_provider.context.protocol_version = "2025-06-18"
+
+        # Mock the parent class's _perform_authorization since it would try to do real OAuth
+        from unittest.mock import AsyncMock, patch
+
+        mock_request = AsyncMock()
+        with patch.object(
+            rfc7523_oauth_provider.__class__.__bases__[0],
+            "_perform_authorization",
+            new=AsyncMock(return_value=mock_request),
+        ) as mock_parent:
+            result = await rfc7523_oauth_provider._perform_authorization()
+            mock_parent.assert_called_once()
+            assert result == mock_request