Add client_secret_basic authentication support

Add support for HTTP Basic Authentication (client_secret_basic) as a
client authentication method for the token and revoke endpoints, alongside
the existing client_secret_post method. This improves compatibility with
OAuth servers like Keycloak that use Basic auth.

Key changes:
- Update OAuthClientMetadata to accept "client_secret_basic" as valid
  token_endpoint_auth_method
- Return 401 status for authentication failures (was 400)
- Update metadata endpoints to advertise both auth methods
- Add tests for both auth methods and edge cases

Author: jonshea

src/mcp/server/auth/handlers/revoke.py

@@ -40,28 +40,25 @@ async def handle(self, request: Request) -> Response:
         Handler for the OAuth 2.0 Token Revocation endpoint.
         """
         try:
-            form_data = await request.form()
-            revocation_request = RevocationRequest.model_validate(dict(form_data))
-        except ValidationError as e:
+            client = await self.client_authenticator.authenticate_request(request)
+        except AuthenticationError as e:
             return PydanticJSONResponse(
-                status_code=400,
+                status_code=401,
                 content=RevocationErrorResponse(
-                    error="invalid_request",
-                    error_description=stringify_pydantic_error(e),
+                    error="unauthorized_client",
+                    error_description=e.message,
                 ),
             )
 
-        # Authenticate client
         try:
-            client = await self.client_authenticator.authenticate(
-                revocation_request.client_id, revocation_request.client_secret
-            )
-        except AuthenticationError as e:
+            form_data = await request.form()
+            revocation_request = RevocationRequest.model_validate(dict(form_data))
+        except ValidationError as e:
             return PydanticJSONResponse(
-                status_code=401,
+                status_code=400,
                 content=RevocationErrorResponse(
-                    error="unauthorized_client",
-                    error_description=e.message,
+                    error="invalid_request",
+                    error_description=stringify_pydantic_error(e),
                 ),
             )
 

src/mcp/server/auth/handlers/token.py

@@ -91,6 +91,22 @@ def response(self, obj: TokenSuccessResponse | TokenErrorResponse):
         )
 
     async def handle(self, request: Request):
+        try:
+            client_info = await self.client_authenticator.authenticate_request(request)
+        except AuthenticationError as e:
+            # Authentication failures should return 401
+            return PydanticJSONResponse(
+                content=TokenErrorResponse(
+                    error="unauthorized_client",
+                    error_description=e.message,
+                ),
+                status_code=401,
+                headers={
+                    "Cache-Control": "no-store",
+                    "Pragma": "no-cache",
+                },
+            )
+
         try:
             form_data = await request.form()
             token_request = TokenRequest.model_validate(dict(form_data)).root
@@ -102,19 +118,6 @@ async def handle(self, request: Request):
                 )
             )
 
-        try:
-            client_info = await self.client_authenticator.authenticate(
-                client_id=token_request.client_id,
-                client_secret=token_request.client_secret,
-            )
-        except AuthenticationError as e:
-            return self.response(
-                TokenErrorResponse(
-                    error="unauthorized_client",
-                    error_description=e.message,
-                )
-            )
-
         if token_request.grant_type not in client_info.grant_types:
             return self.response(
                 TokenErrorResponse(

src/mcp/server/auth/middleware/client_auth.py

@@ -1,6 +1,9 @@
+import base64
 import time
 from typing import Any
 
+from starlette.requests import Request
+
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider
 from mcp.shared.auth import OAuthClientInformationFull
 
@@ -30,19 +33,67 @@ def __init__(self, provider: OAuthAuthorizationServerProvider[Any, Any, Any]):
         """
         self.provider = provider
 
-    async def authenticate(self, client_id: str, client_secret: str | None) -> OAuthClientInformationFull:
-        # Look up client information
-        client = await self.provider.get_client(client_id)
+    async def authenticate_request(self, request: Request) -> OAuthClientInformationFull:
+        """
+        Authenticate a client from an HTTP request.
+
+        Extracts client credentials from the appropriate location based on the
+        client's registered authentication method and validates them.
+
+        Args:
+            request: The HTTP request containing client credentials
+
+        Returns:
+            The authenticated client information
+
+        Raises:
+            AuthenticationError: If authentication fails
+        """
+        form_data = await request.form()
+        client_id = form_data.get("client_id")
+        if not client_id:
+            raise AuthenticationError("Missing client_id")
+
+        client = await self.provider.get_client(str(client_id))
         if not client:
             raise AuthenticationError("Invalid client_id")
 
-        # If client from the store expects a secret, validate that the request provides
-        # that secret
+        request_client_secret = None
+        auth_header = request.headers.get("Authorization", "")
+
+        if client.token_endpoint_auth_method == "client_secret_basic":
+            if not auth_header.startswith("Basic "):
+                raise AuthenticationError("Missing or invalid Basic authentication in Authorization header")
+
+            try:
+                encoded_credentials = auth_header[6:]  # Remove "Basic " prefix
+                decoded = base64.b64decode(encoded_credentials).decode("utf-8")
+                if ":" not in decoded:
+                    raise ValueError("Invalid Basic auth format")
+                basic_client_id, request_client_secret = decoded.split(":", 1)
+
+                if basic_client_id != client_id:
+                    raise AuthenticationError("Client ID mismatch in Basic auth")
+            except AuthenticationError:
+                raise
+            except Exception:
+                raise AuthenticationError("Invalid Basic authentication header")
+
+        elif client.token_endpoint_auth_method == "client_secret_post":
+            request_client_secret = form_data.get("client_secret")
+            if request_client_secret:
+                request_client_secret = str(request_client_secret)
+
+        elif client.token_endpoint_auth_method == "none":
+            request_client_secret = None
+        else:
+            raise AuthenticationError(f"Unsupported auth method: {client.token_endpoint_auth_method}")
+
         if client.client_secret:
-            if not client_secret:
+            if not request_client_secret:
                 raise AuthenticationError("Client secret is required")
 
-            if client.client_secret != client_secret:
+            if client.client_secret != request_client_secret:
                 raise AuthenticationError("Invalid client_secret")
 
             if client.client_secret_expires_at and client.client_secret_expires_at < int(time.time()):

src/mcp/server/auth/routes.py

@@ -164,7 +164,7 @@ def build_metadata(
         response_types_supported=["code"],
         response_modes_supported=None,
         grant_types_supported=["authorization_code", "refresh_token"],
-        token_endpoint_auth_methods_supported=["client_secret_post"],
+        token_endpoint_auth_methods_supported=["client_secret_post", "client_secret_basic"],
         token_endpoint_auth_signing_alg_values_supported=None,
         service_documentation=service_documentation_url,
         ui_locales_supported=None,
@@ -181,7 +181,7 @@ def build_metadata(
     # Add revocation endpoint if supported
     if revocation_options.enabled:
         metadata.revocation_endpoint = AnyHttpUrl(str(issuer_url).rstrip("/") + REVOCATION_PATH)
-        metadata.revocation_endpoint_auth_methods_supported = ["client_secret_post"]
+        metadata.revocation_endpoint_auth_methods_supported = ["client_secret_post", "client_secret_basic"]
 
     return metadata
 

src/mcp/shared/auth.py

@@ -42,10 +42,7 @@ class OAuthClientMetadata(BaseModel):
     """
 
     redirect_uris: list[AnyUrl] = Field(..., min_length=1)
-    # token_endpoint_auth_method: this implementation only supports none &
-    # client_secret_post;
-    # ie: we do not support client_secret_basic
-    token_endpoint_auth_method: Literal["none", "client_secret_post"] = "client_secret_post"
+    token_endpoint_auth_method: Literal["none", "client_secret_post", "client_secret_basic"] = "client_secret_post"
     # grant_types: this implementation only supports authorization_code & refresh_token
     grant_types: list[Literal["authorization_code", "refresh_token"]] = [
         "authorization_code",

tests/client/test_auth.py

@@ -832,10 +832,10 @@ def test_build_metadata(
             "registration_endpoint": Is(registration_endpoint),
             "scopes_supported": ["read", "write", "admin"],
             "grant_types_supported": ["authorization_code", "refresh_token"],
-            "token_endpoint_auth_methods_supported": ["client_secret_post"],
+            "token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
             "service_documentation": Is(service_documentation_url),
             "revocation_endpoint": Is(revocation_endpoint),
-            "revocation_endpoint_auth_methods_supported": ["client_secret_post"],
+            "revocation_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
             "code_challenge_methods_supported": ["S256"],
         }
     )

tests/server/fastmcp/auth/test_auth_integration.py

@@ -12,7 +12,7 @@
 
 import httpx
 import pytest
-from pydantic import AnyHttpUrl
+from pydantic import AnyHttpUrl, AnyUrl
 from starlette.applications import Starlette
 
 from mcp.server.auth.provider import (
@@ -357,7 +357,7 @@ async def test_metadata_endpoint(self, test_client: httpx.AsyncClient):
         assert metadata["revocation_endpoint"] == "https://auth.example.com/revoke"
         assert metadata["response_types_supported"] == ["code"]
         assert metadata["code_challenge_methods_supported"] == ["S256"]
-        assert metadata["token_endpoint_auth_methods_supported"] == ["client_secret_post"]
+        assert metadata["token_endpoint_auth_methods_supported"] == ["client_secret_post", "client_secret_basic"]
         assert metadata["grant_types_supported"] == [
             "authorization_code",
             "refresh_token",
@@ -376,8 +376,8 @@ async def test_token_validation_error(self, test_client: httpx.AsyncClient):
             },
         )
         error_response = response.json()
-        assert error_response["error"] == "invalid_request"
-        assert "error_description" in error_response  # Contains validation error messages
+        assert error_response["error"] == "unauthorized_client"
+        assert "error_description" in error_response  # Contains error message
 
     @pytest.mark.anyio
     async def test_token_invalid_auth_code(
@@ -942,6 +942,147 @@ async def test_client_registration_invalid_grant_type(self, test_client: httpx.A
         assert error_data["error"] == "invalid_client_metadata"
         assert error_data["error_description"] == "grant_types must be authorization_code and refresh_token"
 
+    @pytest.mark.anyio
+    async def test_client_secret_basic_authentication(
+        self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider, pkce_challenge: dict[str, str]
+    ):
+        """Test that client_secret_basic authentication works correctly."""
+        client_metadata = {
+            "redirect_uris": ["https://client.example.com/callback"],
+            "client_name": "Basic Auth Client",
+            "token_endpoint_auth_method": "client_secret_basic",
+            "grant_types": ["authorization_code", "refresh_token"],
+        }
+
+        response = await test_client.post("/register", json=client_metadata)
+        assert response.status_code == 201
+        client_info = response.json()
+        assert client_info["token_endpoint_auth_method"] == "client_secret_basic"
+
+        auth_code = f"code_{int(time.time())}"
+        mock_oauth_provider.auth_codes[auth_code] = AuthorizationCode(
+            code=auth_code,
+            client_id=client_info["client_id"],
+            code_challenge=pkce_challenge["code_challenge"],
+            redirect_uri=AnyUrl("https://client.example.com/callback"),
+            redirect_uri_provided_explicitly=True,
+            scopes=["read", "write"],
+            expires_at=time.time() + 600,
+        )
+
+        credentials = f"{client_info['client_id']}:{client_info['client_secret']}"
+        encoded_credentials = base64.b64encode(credentials.encode()).decode()
+
+        response = await test_client.post(
+            "/token",
+            headers={"Authorization": f"Basic {encoded_credentials}"},
+            data={
+                "grant_type": "authorization_code",
+                "client_id": client_info["client_id"],
+                "code": auth_code,
+                "code_verifier": pkce_challenge["code_verifier"],
+                "redirect_uri": "https://client.example.com/callback",
+            },
+        )
+        assert response.status_code == 200
+        token_response = response.json()
+        assert "access_token" in token_response
+
+    @pytest.mark.anyio
+    async def test_wrong_auth_method_without_valid_credentials_fails(
+        self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider, pkce_challenge: dict[str, str]
+    ):
+        """Test that using the wrong authentication method fails when credentials are missing."""
+        client_metadata = {
+            "redirect_uris": ["https://client.example.com/callback"],
+            "client_name": "Post Auth Client",
+            "token_endpoint_auth_method": "client_secret_post",
+            "grant_types": ["authorization_code", "refresh_token"],
+        }
+
+        response = await test_client.post("/register", json=client_metadata)
+        assert response.status_code == 201
+        client_info = response.json()
+        assert client_info["token_endpoint_auth_method"] == "client_secret_post"
+
+        auth_code = f"code_{int(time.time())}"
+        mock_oauth_provider.auth_codes[auth_code] = AuthorizationCode(
+            code=auth_code,
+            client_id=client_info["client_id"],
+            code_challenge=pkce_challenge["code_challenge"],
+            redirect_uri=AnyUrl("https://client.example.com/callback"),
+            redirect_uri_provided_explicitly=True,
+            scopes=["read", "write"],
+            expires_at=time.time() + 600,
+        )
+
+        # Try to use Basic auth when client_secret_post is registered (without secret in body)
+        # This should fail because the secret is missing from the expected location
+
+        credentials = f"{client_info['client_id']}:{client_info['client_secret']}"
+        encoded_credentials = base64.b64encode(credentials.encode()).decode()
+
+        response = await test_client.post(
+            "/token",
+            headers={"Authorization": f"Basic {encoded_credentials}"},
+            data={
+                "grant_type": "authorization_code",
+                "client_id": client_info["client_id"],
+                # client_secret NOT in body where it should be
+                "code": auth_code,
+                "code_verifier": pkce_challenge["code_verifier"],
+                "redirect_uri": "https://client.example.com/callback",
+            },
+        )
+        assert response.status_code == 401
+        error_response = response.json()
+        assert error_response["error"] == "unauthorized_client"
+        assert "Client secret is required" in error_response["error_description"]
+
+    @pytest.mark.anyio
+    async def test_basic_auth_without_header_fails(
+        self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider, pkce_challenge: dict[str, str]
+    ):
+        """Test that omitting Basic auth when client_secret_basic is registered fails."""
+        client_metadata = {
+            "redirect_uris": ["https://client.example.com/callback"],
+            "client_name": "Basic Auth Client",
+            "token_endpoint_auth_method": "client_secret_basic",
+            "grant_types": ["authorization_code", "refresh_token"],
+        }
+
+        response = await test_client.post("/register", json=client_metadata)
+        assert response.status_code == 201
+        client_info = response.json()
+        assert client_info["token_endpoint_auth_method"] == "client_secret_basic"
+
+        auth_code = f"code_{int(time.time())}"
+        mock_oauth_provider.auth_codes[auth_code] = AuthorizationCode(
+            code=auth_code,
+            client_id=client_info["client_id"],
+            code_challenge=pkce_challenge["code_challenge"],
+            redirect_uri=AnyUrl("https://client.example.com/callback"),
+            redirect_uri_provided_explicitly=True,
+            scopes=["read", "write"],
+            expires_at=time.time() + 600,
+        )
+
+        response = await test_client.post(
+            "/token",
+            data={
+                "grant_type": "authorization_code",
+                "client_id": client_info["client_id"],
+                "client_secret": client_info["client_secret"],  # Secret in body (ignored)
+                "code": auth_code,
+                "code_verifier": pkce_challenge["code_verifier"],
+                "redirect_uri": "https://client.example.com/callback",
+            },
+        )
+        assert response.status_code == 401
+        error_response = response.json()
+        assert error_response["error"] == "unauthorized_client"
+        assert "Missing or invalid Basic authentication" in error_response["error_description"]
+
 
 class TestAuthorizeEndpointErrors:
     """Test error handling in the OAuth authorization endpoint."""