Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion src/mcp/server/auth/middleware/client_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,16 @@ async def authenticate_request(self, request: Request) -> OAuthClientInformation
"""
form_data = await request.form()
client_id = form_data.get("client_id")
auth_header = request.headers.get("Authorization", "")
if not client_id and auth_header.startswith("Basic "):
# RFC 6749 §2.3.1 lets a client_secret_basic client carry its id
# only in the Basic header, so fall back to it before giving up.
try:
decoded = base64.b64decode(auth_header[6:]).decode("utf-8")
if ":" in decoded:
client_id = unquote(decoded.split(":", 1)[0]) or None
except (ValueError, UnicodeDecodeError, binascii.Error):
client_id = None
if not client_id:
raise AuthenticationError("Missing client_id")

Expand All @@ -61,7 +71,6 @@ async def authenticate_request(self, request: Request) -> OAuthClientInformation
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 "):
Expand Down
77 changes: 77 additions & 0 deletions tests/server/auth/middleware/test_client_auth.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
"""Direct tests for ClientAuthenticator's Basic-header client_id fallback."""

import base64

import pytest
from starlette.requests import Request

from mcp.server.auth.middleware.client_auth import (
AuthenticationError,
ClientAuthenticator,
)
from mcp.server.auth.provider import OAuthAuthorizationServerProvider
from mcp.shared.auth import OAuthClientInformationFull


class _StubProvider(OAuthAuthorizationServerProvider[None, None, None]): # type: ignore[type-arg]
def __init__(self, client: OAuthClientInformationFull | None):
self._client = client

async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
if self._client and self._client.client_id == client_id:
return self._client
return None # pragma: no cover

async def register_client(self, client_info: OAuthClientInformationFull): ... # pragma: no cover
async def authorize(self, client: OAuthClientInformationFull, params): ... # pragma: no cover


def _client() -> OAuthClientInformationFull:
return OAuthClientInformationFull(
client_id="header-only-client",
client_secret="s3cret",
token_endpoint_auth_method="client_secret_basic",
redirect_uris=["https://client.example.com/callback"],
grant_types=["authorization_code"],
)


def _request(body: bytes, auth_header: str | None) -> Request:
async def receive() -> dict:
return {"type": "http.request", "body": body, "more_body": False}

headers = [(b"content-type", b"application/x-www-form-urlencoded")]
if auth_header is not None:
headers.append((b"authorization", auth_header.encode()))
return Request(
{
"type": "http",
"method": "POST",
"path": "/token",
"headers": headers,
},
receive=receive,
)


@pytest.mark.anyio
async def test_basic_client_id_falls_back_to_authorization_header() -> None:
"""A client_secret_basic request with no body client_id authenticates via the header."""
stored = _client()
auth = ClientAuthenticator(_StubProvider(stored))
creds = base64.b64encode(f"{stored.client_id}:{stored.client_secret}".encode()).decode()
req = _request(b"grant_type=authorization_code&code=abc", f"Basic {creds}")

result = await auth.authenticate_request(req)

assert result.client_id == stored.client_id


@pytest.mark.anyio
async def test_no_credentials_anywhere_still_rejected() -> None:
"""With neither body nor header credentials, the original Missing client_id stands."""
auth = ClientAuthenticator(_StubProvider(_client()))
req = _request(b"grant_type=authorization_code&code=abc", None)

with pytest.raises(AuthenticationError, match="Missing client_id"):
await auth.authenticate_request(req)
Loading