feat: fully support Basic Authorization header at token request

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

@@ -19,7 +19,7 @@ class AuthorizationCodeRequest(BaseModel):
     grant_type: Literal["authorization_code"]
     code: str = Field(..., description="The authorization code")
     redirect_uri: AnyUrl | None = Field(None, description="Must be the same as redirect URI provided in /authorize")
-    client_id: str
+    client_id: str | None = Field(None, description="If none, client_id must be provided via basic auth header")
     # we use the client_secret param, per https://datatracker.ietf.org/doc/html/rfc6749#section-2.3.1
     client_secret: str | None = None
     # See https://datatracker.ietf.org/doc/html/rfc7636#section-4.5
@@ -33,7 +33,7 @@ class RefreshTokenRequest(BaseModel):
     grant_type: Literal["refresh_token"]
     refresh_token: str = Field(..., description="The refresh token")
     scope: str | None = Field(None, description="Optional scope parameter")
-    client_id: str
+    client_id: str | None = Field(None, description="If none, client_id must be provided via basic auth header")
     # we use the client_secret param, per https://datatracker.ietf.org/doc/html/rfc6749#section-2.3.1
     client_secret: str | None = None
     # RFC 8707 resource indicator
@@ -131,7 +131,7 @@ async def handle(self, request: Request):
         match token_request:
             case AuthorizationCodeRequest():
                 auth_code = await self.provider.load_authorization_code(client_info, token_request.code)
-                if auth_code is None or auth_code.client_id != token_request.client_id:
+                if auth_code is None or auth_code.client_id != client_info.client_id:
                     # if code belongs to different client, pretend it doesn't exist
                     return self.response(
                         TokenErrorResponse(
@@ -197,7 +197,7 @@ async def handle(self, request: Request):
 
             case RefreshTokenRequest():  # pragma: no cover
                 refresh_token = await self.provider.load_refresh_token(client_info, token_request.refresh_token)
-                if refresh_token is None or refresh_token.client_id != token_request.client_id:
+                if refresh_token is None or refresh_token.client_id != client_info.client_id:
                     # if token belongs to different client, pretend it doesn't exist
                     return self.response(
                         TokenErrorResponse(

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

@@ -2,7 +2,8 @@
 import binascii
 import hmac
 import time
-from typing import Any
+from dataclasses import dataclass
+from typing import Any, Literal
 from urllib.parse import unquote
 
 from starlette.requests import Request
@@ -16,6 +17,13 @@ def __init__(self, message: str):
         self.message = message  # pragma: no cover
 
 
+@dataclass
+class ClientCredentials:
+    auth_method: Literal["client_secret_basic", "client_secret_post"]
+    client_id: str
+    client_secret: str | None = None
+
+
 class ClientAuthenticator:
     """
     ClientAuthenticator is a callable which validates requests from a client
@@ -52,64 +60,86 @@ async def authenticate_request(self, request: Request) -> OAuthClientInformation
         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_credentials = await self._get_credentials(request)
+        client = await self.provider.get_client(client_credentials.client_id)
 
-        client = await self.provider.get_client(str(client_id))
         if not client:
             raise AuthenticationError("Invalid client_id")  # pragma: no cover
 
-        request_client_secret: str | None = 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)
-
-                # URL-decode both parts per RFC 6749 Section 2.3.1
-                basic_client_id = unquote(basic_client_id)
-                request_client_secret = unquote(request_client_secret)
-
-                if basic_client_id != client_id:
-                    raise AuthenticationError("Client ID mismatch in Basic auth")
-            except (ValueError, UnicodeDecodeError, binascii.Error):
-                raise AuthenticationError("Invalid Basic authentication header")
-
+            if client_credentials.auth_method != "client_secret_basic":
+                raise AuthenticationError(
+                    f"Expected client_secret_basic authentication method, but got {client_credentials.auth_method}"
+                )
         elif client.token_endpoint_auth_method == "client_secret_post":
-            raw_form_data = form_data.get("client_secret")
-            # form_data.get() can return a UploadFile or None, so we need to check if it's a string
-            if isinstance(raw_form_data, str):
-                request_client_secret = str(raw_form_data)
-
+            if client_credentials.auth_method != "client_secret_post":
+                raise AuthenticationError(
+                    f"Expected client_secret_post authentication method, but got {client_credentials.auth_method}"
+                )
         elif client.token_endpoint_auth_method == "none":
-            request_client_secret = None
-        else:
-            raise AuthenticationError(  # pragma: no cover
-                f"Unsupported auth method: {client.token_endpoint_auth_method}"
-            )
+            pass
+        else:  # pragma: no cover
+            raise AuthenticationError(f"Unsupported auth method: {client.token_endpoint_auth_method}")
 
         # If client from the store expects a secret, validate that the request provides
         # that secret
         if client.client_secret:  # pragma: no branch
-            if not request_client_secret:
+            if not client_credentials.client_secret:
                 raise AuthenticationError("Client secret is required")  # pragma: no cover
 
             # hmac.compare_digest requires that both arguments are either bytes or a `str` containing
             # only ASCII characters. Since we do not control `request_client_secret`, we encode both
             # arguments to bytes.
-            if not hmac.compare_digest(client.client_secret.encode(), request_client_secret.encode()):
+            if not hmac.compare_digest(client.client_secret.encode(), client_credentials.client_secret.encode()):
                 raise AuthenticationError("Invalid client_secret")  # pragma: no cover
 
             if client.client_secret_expires_at and client.client_secret_expires_at < int(time.time()):
                 raise AuthenticationError("Client secret has expired")  # pragma: no cover
 
         return client
+
+    async def _get_credentials(self, request: Request) -> ClientCredentials:
+        """
+        Extract client credentials from request, either from form data or Basic auth header.
+
+        Basic auth header takes precedence over form data.
+
+        Args:
+            request: The HTTP request containing client credentials
+        Returns:
+            The extracted client credentials
+        """
+        # First, check for Basic auth header
+        auth_header = request.headers.get("Authorization", "")
+        if auth_header.startswith("Basic "):
+            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")
+                client_id, client_secret = decoded.split(":", 1)
+
+                # URL-decode the client_id per RFC 6749 Section 2.3.1
+                client_id = unquote(client_id)
+                client_secret = unquote(client_secret)
+                return ClientCredentials(
+                    auth_method="client_secret_basic",
+                    client_id=client_id,
+                    client_secret=client_secret,
+                )
+            except (ValueError, UnicodeDecodeError, binascii.Error):
+                raise AuthenticationError("Invalid Basic authentication header")
+
+        # If not, check for client_id and client_secret in form data
+        form_data = await request.form()
+        client_id = form_data.get("client_id")
+        if not client_id:
+            raise AuthenticationError("Missing client_id")
+
+        raw_client_secret = form_data.get("client_secret")
+        client_secret = str(raw_client_secret) if isinstance(raw_client_secret, str) else None
+        return ClientCredentials(
+            auth_method="client_secret_post",
+            client_id=str(client_id),
+            client_secret=client_secret,
+        )

tests/server/fastmcp/auth/test_auth_integration.py

@@ -1089,7 +1089,7 @@ async def test_client_secret_basic_authentication(
         assert "access_token" in token_response
 
     @pytest.mark.anyio
-    async def test_wrong_auth_method_without_valid_credentials_fails(
+    async def test_wrong_auth_method_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."""
@@ -1117,7 +1117,7 @@ async def test_wrong_auth_method_without_valid_credentials_fails(
         )
 
         # 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
+        # This should fail despite that credentials are provided via Basic auth, because the method is wrong
 
         credentials = f"{client_info['client_id']}:{client_info['client_secret']}"
         encoded_credentials = base64.b64encode(credentials.encode()).decode()
@@ -1138,7 +1138,7 @@ async def test_wrong_auth_method_without_valid_credentials_fails(
         error_response = response.json()
         # RFC 6749: authentication failures return "invalid_client"
         assert error_response["error"] == "invalid_client"
-        assert "Client secret is required" in error_response["error_description"]
+        assert "Expected client_secret_post authentication method" in error_response["error_description"]
 
     @pytest.mark.anyio
     async def test_basic_auth_without_header_fails(
@@ -1183,7 +1183,7 @@ async def test_basic_auth_without_header_fails(
         error_response = response.json()
         # RFC 6749: authentication failures return "invalid_client"
         assert error_response["error"] == "invalid_client"
-        assert "Missing or invalid Basic authentication" in error_response["error_description"]
+        assert "Expected client_secret_basic authentication method" in error_response["error_description"]
 
     @pytest.mark.anyio
     async def test_basic_auth_invalid_base64_fails(
@@ -1279,10 +1279,10 @@ async def test_basic_auth_no_colon_fails(
         assert "Invalid Basic authentication header" in error_response["error_description"]
 
     @pytest.mark.anyio
-    async def test_basic_auth_client_id_mismatch_fails(
+    async def test_basic_auth_takes_precedence(
         self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider, pkce_challenge: dict[str, str]
     ):
-        """Test that client_id mismatch between body and Basic auth fails."""
+        """Test that even client_id at body is invalid, Basic auth passes because of the priority."""
         client_metadata = {
             "redirect_uris": ["https://client.example.com/callback"],
             "client_name": "Basic Auth Client",
@@ -1308,23 +1308,83 @@ async def test_basic_auth_client_id_mismatch_fails(
         # Send different client_id in Basic auth header
         import base64
 
-        wrong_creds = base64.b64encode(f"wrong-client-id:{client_info['client_secret']}".encode()).decode()
+        creds = base64.b64encode(f"{client_info['client_id']}:{client_info['client_secret']}".encode()).decode()
         response = await test_client.post(
             "/token",
-            headers={"Authorization": f"Basic {wrong_creds}"},
+            headers={"Authorization": f"Basic {creds}"},
             data={
                 "grant_type": "authorization_code",
-                "client_id": client_info["client_id"],  # Correct client_id in body
+                "client_id": "wrong-client-id",  # Wrong client_id in body
                 "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()
-        # RFC 6749: authentication failures return "invalid_client"
-        assert error_response["error"] == "invalid_client"
-        assert "Client ID mismatch" in error_response["error_description"]
+
+        # Header takes precedence, so this should succeed
+        assert response.status_code == 200
+
+    @pytest.mark.anyio
+    async def test_basic_auth_without_client_id_at_body(
+        self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider, pkce_challenge: dict[str, str]
+    ):
+        """Test that Basic auth works even if client_id is missing from body."""
+        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()
+
+        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 omitted from body
+                "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
+        assert "refresh_token" in token_response
+
+        refresh_token = token_response["refresh_token"]
+
+        # Now, use the refresh token without client_id in body
+        response = await test_client.post(
+            "/token",
+            headers={"Authorization": f"Basic {encoded_credentials}"},
+            data={
+                "grant_type": "refresh_token",
+                # client_id omitted from body
+                "refresh_token": refresh_token,
+            },
+        )
+        assert response.status_code == 200
+        new_token_response = response.json()
+        assert "access_token" in new_token_response
 
     @pytest.mark.anyio
     async def test_none_auth_method_public_client(