feat(oauth): support client_secret_basic

Author: sam-tombury

src/mcp/client/auth.py

@@ -10,6 +10,7 @@
 import secrets
 import string
 import time
+from base64 import b64encode
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from typing import Protocol
 from urllib.parse import urlencode, urljoin
@@ -359,22 +360,35 @@ async def _exchange_code_for_token(self, auth_code: str, client_info: OAuthClien
             auth_base_url = self._get_authorization_base_url(self.server_url)
             token_url = urljoin(auth_base_url, "/token")
 
+        extra_headers = {}
+
         token_data = {
             "grant_type": "authorization_code",
             "code": auth_code,
             "redirect_uri": str(self.client_metadata.redirect_uris[0]),
-            "client_id": client_info.client_id,
             "code_verifier": self._code_verifier,
         }
 
-        if client_info.client_secret:
-            token_data["client_secret"] = client_info.client_secret
+        match client_info.token_endpoint_auth_method:
+            case "none":
+                token_data["client_id"] = client_info.client_id
+            case "client_secret_post" if client_info.client_secret:
+                token_data["client_id"] = client_info.client_id
+                token_data["client_secret"] = client_info.client_secret
+            case "client_secret_basic" if client_info.client_secret:
+                basic = b64encode(
+                    f"{client_info.client_id}:{client_info.client_secret}".encode()
+                ).decode()
+                extra_headers = {"Authorization": f"Basic {basic}"}
+            case _:
+                pass
 
         async with httpx.AsyncClient() as client:
             response = await client.post(
                 token_url,
                 data=token_data,
-                headers={"Content-Type": "application/x-www-form-urlencoded"},
+                headers={"Content-Type": "application/x-www-form-urlencoded"}
+                | extra_headers,
                 timeout=30.0,
             )
 
@@ -419,21 +433,35 @@ async def _refresh_access_token(self) -> bool:
             auth_base_url = self._get_authorization_base_url(self.server_url)
             token_url = urljoin(auth_base_url, "/token")
 
+        extra_headers = {}
+
         refresh_data = {
             "grant_type": "refresh_token",
             "refresh_token": self._current_tokens.refresh_token,
             "client_id": client_info.client_id,
         }
 
-        if client_info.client_secret:
-            refresh_data["client_secret"] = client_info.client_secret
+        match client_info.token_endpoint_auth_method:
+            case "none":
+                refresh_data["client_id"] = client_info.client_id
+            case "client_secret_post" if client_info.client_secret:
+                refresh_data["client_id"] = client_info.client_id
+                refresh_data["client_secret"] = client_info.client_secret
+            case "client_secret_basic" if client_info.client_secret:
+                basic = b64encode(
+                    f"{client_info.client_id}:{client_info.client_secret}".encode()
+                ).decode()
+                extra_headers = {"Authorization": f"Basic {basic}"}
+            case _:
+                pass
 
         try:
             async with httpx.AsyncClient() as client:
                 response = await client.post(
                     token_url,
                     data=refresh_data,
-                    headers={"Content-Type": "application/x-www-form-urlencoded"},
+                    headers={"Content-Type": "application/x-www-form-urlencoded"}
+                    | extra_headers,
                     timeout=30.0,
                 )
 

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

@@ -1,6 +1,7 @@
 import base64
 import hashlib
 import time
+from base64 import b64decode
 from dataclasses import dataclass
 from typing import Annotated, Any, Literal
 
@@ -19,9 +20,6 @@ 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
-    # 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
     code_verifier: str = Field(..., description="PKCE code verifier")
 
@@ -31,9 +29,50 @@ 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")
+
+
+class NoneCredentials(BaseModel):
+    client_id: str
+    client_secret: None = None
+
+
+class PostCredentials(BaseModel):
     client_id: str
     # we use the client_secret param, per https://datatracker.ietf.org/doc/html/rfc6749#section-2.3.1
-    client_secret: str | None = None
+    client_secret: str
+
+
+class FormCredentials(
+    RootModel[
+        Annotated[
+            NoneCredentials | PostCredentials,
+            Field(discriminator="client_secret"),
+        ]
+    ]
+):
+    root: Annotated[
+        NoneCredentials | PostCredentials,
+        Field(discriminator="client_secret"),
+    ]
+
+
+class BasicCredentials(BaseModel):
+    client_id: str
+    client_secret: str
+
+    @classmethod
+    def from_authorization(cls, authorization: str):
+        try:
+            if authorization.startswith("Basic "):
+                [client_id, client_secret] = b64decode(authorization.removeprefix("Basic ")).decode().split(":", 1)
+                return cls(client_id=client_id, client_secret=client_secret)
+        except Exception:
+            # TODO: better error here??
+            return None
+        return None
+
+
+Credentials = NoneCredentials | PostCredentials | BasicCredentials
 
 
 class TokenRequest(
@@ -90,19 +129,42 @@ async def handle(self, request: Request):
         try:
             form_data = await request.form()
             token_request = TokenRequest.model_validate(dict(form_data)).root
+            try:
+                credentials = FormCredentials.model_validate(dict(form_data)).root
+            except ValidationError:
+                credentials = (
+                    BasicCredentials.from_authorization(authorization)
+                    if (authorization := request.headers.get("Authorization"))
+                    else None
+                )
+                if not credentials:
+                    return self.response(
+                        TokenErrorResponse(
+                            error="invalid_request",
+                            error_description="missing credentials",
+                        )
+                    )
         except ValidationError as validation_error:
             return self.response(
                 TokenErrorResponse(
                     error="invalid_request",
                     error_description=stringify_pydantic_error(validation_error),
                 )
             )
-
         try:
             client_info = await self.client_authenticator.authenticate(
-                client_id=token_request.client_id,
-                client_secret=token_request.client_secret,
+                client_id=credentials.client_id,
+                client_secret=credentials.client_secret,
             )
+            match client_info.token_endpoint_auth_method:
+                case "none" if not isinstance(credentials, NoneCredentials):
+                    raise AuthenticationError("Invalid credentials for client token_endpoint_auth_method")
+                case "client_secret_post" if not isinstance(credentials, PostCredentials):
+                    raise AuthenticationError("Invalid credentials for client token_endpoint_auth_method")
+                case "client_secret_basic" if not isinstance(credentials, BasicCredentials):
+                    raise AuthenticationError("Invalid credentials for client token_endpoint_auth_method")
+                case _:
+                    pass
         except AuthenticationError as e:
             return self.response(
                 TokenErrorResponse(
@@ -126,7 +188,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 != credentials.client_id:
                     # if code belongs to different client, pretend it doesn't exist
                     return self.response(
                         TokenErrorResponse(
@@ -185,7 +247,7 @@ async def handle(self, request: Request):
 
             case RefreshTokenRequest():
                 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 != credentials.client_id:
                     # if token belongs to different client, pretend it doesn't exist
                     return self.response(
                         TokenErrorResponse(

src/mcp/server/auth/routes.py

@@ -159,7 +159,10 @@ 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,
@@ -176,6 +179,9 @@ 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",