diff --git a/.github/actions/conformance/expected-failures.yml b/.github/actions/conformance/expected-failures.yml index b542e788b4..dc18a647d5 100644 --- a/.github/actions/conformance/expected-failures.yml +++ b/.github/actions/conformance/expected-failures.yml @@ -1,4 +1,11 @@ # Known conformance test failures for v1.x # These are tracked and should be removed as they're fixed. server: [] -client: [] +client: + # The pinned harness (0.1.13) serves authorization server metadata whose `issuer` + # omits the tenant path its resource metadata advertises (`/tenant1`), so a client + # that checks RFC 8414 section 3.3 refuses it. The mock includes the path from + # conformance 0.1.15 (modelcontextprotocol/conformance#152); drop these two entries + # when the pin moves past it. + - auth/metadata-var2 + - auth/metadata-var3 diff --git a/src/mcp/client/auth/extensions/client_credentials.py b/src/mcp/client/auth/extensions/client_credentials.py index e2f3f08a4d..804e117282 100644 --- a/src/mcp/client/auth/extensions/client_credentials.py +++ b/src/mcp/client/auth/extensions/client_credentials.py @@ -11,6 +11,7 @@ import time from collections.abc import Awaitable, Callable from typing import Any, Literal +from urllib.parse import urlparse from uuid import uuid4 import httpx @@ -18,14 +19,50 @@ from pydantic import BaseModel, Field from mcp.client.auth import OAuthClientProvider, OAuthFlowError, OAuthTokenError, TokenStorage +from mcp.client.auth.oauth2 import OAuthContext +from mcp.client.auth.utils import issuers_match from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata +def _checked_issuer(issuer: str | None) -> str | None: + if issuer is not None and urlparse(issuer).scheme not in ("http", "https"): + raise ValueError(f"issuer must be the authorization server's http(s) issuer URL, got {issuer!r}") + return issuer + + +def _preferred_authorization_server(advertised: list[str], issuer: str | None) -> str: + """The advertised server matching the configured issuer if there is one, else the first.""" + return next( + (server for server in advertised if issuer is not None and issuers_match(server, issuer)), advertised[0] + ) + + +def _require_metadata_for_configured_issuer(context: OAuthContext, issuer: str | None) -> None: + """With an issuer configured, a token request is only built from metadata discovered for that issuer. + + Anything else held is dropped along with the tokens, so the next request starts discovery afresh + rather than refreshing against it. + """ + if issuer is None: + return + metadata = context.oauth_metadata + if metadata is not None and issuers_match(str(metadata.issuer), issuer): + return + context.oauth_metadata = None + context.clear_tokens() + if metadata is None: + raise OAuthFlowError(f"No authorization server metadata discovered for configured issuer {issuer}") + raise OAuthFlowError(f"Authorization server metadata issuer mismatch: {metadata.issuer} != {issuer}") + + class ClientCredentialsOAuthProvider(OAuthClientProvider): """OAuth provider for client_credentials grant with client_id + client_secret. This provider sets client_info directly, bypassing dynamic client registration. Use this when you already have client credentials (client_id and client_secret). + Pass `issuer` to name the authorization server those credentials belong to: token + requests are then only built from authorization server metadata for that issuer, and + the flow stops if the MCP server leads anywhere else. Example: ```python @@ -34,6 +71,7 @@ class ClientCredentialsOAuthProvider(OAuthClientProvider): storage=my_token_storage, client_id="my-client-id", client_secret="my-client-secret", + issuer="https://auth.example.com", ) ``` """ @@ -46,6 +84,7 @@ def __init__( client_secret: str, token_endpoint_auth_method: Literal["client_secret_basic", "client_secret_post"] = "client_secret_basic", scopes: str | None = None, + issuer: str | None = None, ) -> None: """Initialize client_credentials OAuth provider. @@ -57,6 +96,11 @@ def __init__( token_endpoint_auth_method: Authentication method for token endpoint. Either "client_secret_basic" (default) or "client_secret_post". scopes: Optional space-separated list of scopes to request. + issuer: The issuer identifier of the authorization server that issued + `client_id` and `client_secret`. When set, token requests are only built from + discovered authorization server metadata whose `issuer` is exactly this string; + otherwise the flow stops with `OAuthFlowError`. When omitted, whichever + authorization server discovery yields is used. """ # Build minimal client_metadata for the base class client_metadata = OAuthClientMetadata( @@ -66,6 +110,7 @@ def __init__( scope=scopes, ) super().__init__(server_url, client_metadata, storage, None, None, 300.0) + self._issuer = _checked_issuer(issuer) # Store client_info to be set during _initialize - no dynamic registration needed self._fixed_client_info = OAuthClientInformationFull( redirect_uris=None, @@ -82,12 +127,17 @@ async def _initialize(self) -> None: self.context.client_info = self._fixed_client_info self._initialized = True + def _select_authorization_server(self, advertised: list[str]) -> str: + return _preferred_authorization_server(advertised, self._issuer) + async def _perform_authorization(self) -> httpx.Request: """Perform client_credentials authorization.""" return await self._exchange_token_client_credentials() async def _exchange_token_client_credentials(self) -> httpx.Request: """Build token exchange request for client_credentials grant.""" + _require_metadata_for_configured_issuer(self.context, self._issuer) + token_data: dict[str, Any] = { "grant_type": "client_credentials", } @@ -198,7 +248,10 @@ class PrivateKeyJWTOAuthProvider(OAuthClientProvider): The JWT assertion's audience MUST be the authorization server's issuer identifier (per RFC 7523bis security updates). The `assertion_provider` callback receives - this audience value and must return a JWT with that audience. + this audience value and must return a JWT with that audience. Pass `issuer` to name + the authorization server this client is registered with: an assertion is then only + minted once metadata for that issuer has been discovered, and token requests are only + built from that metadata. **Option 1: Pre-built JWT via Workload Identity Federation** @@ -258,6 +311,7 @@ def __init__( client_id: str, assertion_provider: Callable[[str], Awaitable[str]], scopes: str | None = None, + issuer: str | None = None, ) -> None: """Initialize private_key_jwt OAuth provider. @@ -271,6 +325,11 @@ def __init__( `static_assertion_provider()` for pre-built JWTs, or provide your own callback for workload identity federation. scopes: Optional space-separated list of scopes to request. + issuer: The issuer identifier of the authorization server `client_id` is + registered with. When set, an assertion is only minted, and token requests + are only built, once authorization server metadata whose `issuer` is exactly this + string has been discovered; otherwise the flow stops with `OAuthFlowError`. + When omitted, whichever authorization server discovery yields is used. """ # Build minimal client_metadata for the base class client_metadata = OAuthClientMetadata( @@ -281,6 +340,7 @@ def __init__( ) super().__init__(server_url, client_metadata, storage, None, None, 300.0) self._assertion_provider = assertion_provider + self._issuer = _checked_issuer(issuer) # Store client_info to be set during _initialize - no dynamic registration needed self._fixed_client_info = OAuthClientInformationFull( redirect_uris=None, @@ -296,6 +356,9 @@ async def _initialize(self) -> None: self.context.client_info = self._fixed_client_info self._initialized = True + def _select_authorization_server(self, advertised: list[str]) -> str: + return _preferred_authorization_server(advertised, self._issuer) + async def _perform_authorization(self) -> httpx.Request: """Perform client_credentials authorization with private_key_jwt.""" return await self._exchange_token_client_credentials() @@ -316,6 +379,8 @@ async def _add_client_authentication_jwt(self, *, token_data: dict[str, Any]) -> async def _exchange_token_client_credentials(self) -> httpx.Request: """Build token exchange request for client_credentials grant with private_key_jwt.""" + _require_metadata_for_configured_issuer(self.context, self._issuer) + token_data: dict[str, Any] = { "grant_type": "client_credentials", } diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py index 0ec0879688..aea037f1f0 100644 --- a/src/mcp/client/auth/oauth2.py +++ b/src/mcp/client/auth/oauth2.py @@ -17,7 +17,7 @@ import anyio import httpx -from pydantic import BaseModel, Field, ValidationError +from pydantic import AnyHttpUrl, BaseModel, Field, ValidationError from mcp.client.auth.exceptions import OAuthFlowError, OAuthTokenError from mcp.client.auth.utils import ( @@ -26,6 +26,7 @@ create_client_info_from_metadata_url, create_client_registration_request, create_oauth_metadata_request, + credentials_match_issuer, extract_field_from_www_auth, extract_resource_metadata_from_www_auth, extract_scope_from_www_auth, @@ -36,6 +37,7 @@ handle_token_response_scopes, is_valid_client_metadata_url, should_use_client_metadata_url, + validate_metadata_issuer, ) from mcp.client.streamable_http import MCP_PROTOCOL_VERSION from mcp.shared.auth import ( @@ -214,6 +216,14 @@ def prepare_token_auth( return data, headers +def _origin_issuer(server_url: str) -> str: + """The resource server's origin as an issuer identifier: `scheme://authority`, rendered the way + `OAuthMetadata.issuer` renders URLs (host case, default ports, trailing slash) so the two compare + as strings.""" + parsed = urlparse(server_url) + return str(AnyHttpUrl(f"{parsed.scheme}://{parsed.netloc}")) + + class OAuthClientProvider(httpx.Auth): """ OAuth2 authentication for httpx. @@ -488,6 +498,16 @@ async def _handle_oauth_metadata_response(self, response: httpx.Response) -> Non metadata = OAuthMetadata.model_validate_json(content) self.context.oauth_metadata = metadata + def _select_authorization_server(self, advertised: list[str]) -> str: + """Which of the servers listed in protected resource metadata to use: the first (the list is never empty).""" + return advertised[0] + + def _expected_issuer(self) -> str: + """The issuer that authorization server metadata and client credentials must belong to: the + PRM-advertised server, or on the legacy no-PRM path the resource server's origin, which is what + the 2025-03-26 well-known URL is built from (RFC 8414 ยง3.3).""" + return self.context.auth_server_url or _origin_issuer(self.context.server_url) + async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.Request, httpx.Response]: """HTTPX auth flow integration.""" async with self.context.lock: @@ -511,55 +531,89 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx. response = yield request - if response.status_code == 401: + step_up = ( + response.status_code == 403 and extract_field_from_www_auth(response, "error") == "insufficient_scope" + ) + + if response.status_code == 401 or step_up: # Perform full OAuth flow try: - # OAuth flow must be inline due to generator constraints - www_auth_resource_metadata_url = extract_resource_metadata_from_www_auth(response) - - # Step 1: Discover protected resource metadata (SEP-985 with fallback support) - prm_discovery_urls = build_protected_resource_metadata_discovery_urls( - www_auth_resource_metadata_url, self.context.server_url - ) - - for url in prm_discovery_urls: # pragma: no branch - discovery_request = create_oauth_metadata_request(url) - - discovery_response = yield discovery_request # sending request - - prm = await handle_protected_resource_response(discovery_response) - if prm: - # Validate PRM resource matches server URL (RFC 8707) - await self._validate_resource_match(prm) - self.context.protected_resource_metadata = prm - - # todo: try all authorization_servers to find the OASM - assert ( - len(prm.authorization_servers) > 0 - ) # this is always true as authorization_servers has a min length of 1 + # OAuth flow must be inline due to generator constraints. + # Steps 1-2 run on every 401. A scope step-up reuses the metadata discovered earlier + # in this process, and discovers it first when none is held yet (for example when + # tokens were loaded from storage), so re-authorization targets the right server. + if response.status_code == 401 or self.context.oauth_metadata is None: + www_auth_resource_metadata_url = extract_resource_metadata_from_www_auth(response) + + # Step 1: Discover protected resource metadata (SEP-985 with fallback support) + prm_discovery_urls = build_protected_resource_metadata_discovery_urls( + www_auth_resource_metadata_url, self.context.server_url + ) - self.context.auth_server_url = str(prm.authorization_servers[0]) - break + prm_request_failed: int | None = None + for url in prm_discovery_urls: + discovery_request = create_oauth_metadata_request(url) + + discovery_response = yield discovery_request # sending request + + if discovery_response.status_code >= 500 or discovery_response.status_code == 429: + prm_request_failed = discovery_response.status_code + prm = await handle_protected_resource_response(discovery_response) + if prm: + # Validate PRM resource matches server URL (RFC 8707) + await self._validate_resource_match(prm) + self.context.protected_resource_metadata = prm + self.context.auth_server_url = self._select_authorization_server( + [str(url) for url in prm.authorization_servers] + ) + break + else: + logger.debug(f"Protected resource metadata discovery failed: {url}") else: - logger.debug(f"Protected resource metadata discovery failed: {url}") - - asm_discovery_urls = build_oauth_authorization_server_metadata_discovery_urls( - self.context.auth_server_url, self.context.server_url - ) + if prm_request_failed is not None: + # A server error says nothing about whether the resource publishes + # metadata, so it must not send the flow down the legacy path. + raise OAuthFlowError( + f"Protected resource metadata request failed: HTTP {prm_request_failed}" + ) + + expected_issuer = self._expected_issuer() + + # SEP-2352: stored credentials are bound to the issuer that registered them. + # Decided before any metadata is fetched: if the expected issuer is a different + # server, drop them (and the old tokens) so the flow re-registers instead of + # presenting another server's credentials. + if self.context.client_info is not None and not credentials_match_issuer( + self.context.client_info, expected_issuer, self.context.client_metadata_url + ): + logger.debug( + "Authorization server changed; discarding bound credentials and re-registering" + ) + self.context.client_info = None + self.context.clear_tokens() + # Any cached AS metadata is for the old server; drop it so a failed + # rediscovery cannot leak the old registration/token endpoints into Step 4. + self.context.oauth_metadata = None + + asm_discovery_urls = build_oauth_authorization_server_metadata_discovery_urls( + self.context.auth_server_url, self.context.server_url + ) - # Step 2: Discover OAuth Authorization Server Metadata (OASM) (with fallback for legacy servers) - for url in asm_discovery_urls: # pragma: no cover - oauth_metadata_request = create_oauth_metadata_request(url) - oauth_metadata_response = yield oauth_metadata_request - - ok, asm = await handle_auth_metadata_response(oauth_metadata_response) - if not ok: - break - if ok and asm: - self.context.oauth_metadata = asm - break - else: - logger.debug(f"OAuth metadata discovery failed: {url}") + # Step 2: Discover OAuth Authorization Server Metadata (OASM) (with fallback for legacy servers) + for url in asm_discovery_urls: # pragma: no branch + oauth_metadata_request = create_oauth_metadata_request(url) + oauth_metadata_response = yield oauth_metadata_request + + ok, asm = await handle_auth_metadata_response(oauth_metadata_response) + if not ok: + break + if ok and asm: + # SEP-2468 / RFC 8414 section 3.3: the metadata must name the expected issuer + validate_metadata_issuer(asm, expected_issuer) + self.context.oauth_metadata = asm + break + else: + logger.debug(f"OAuth metadata discovery failed: {url}") # Step 3: Apply scope selection strategy self.context.client_metadata.scope = get_client_metadata_scopes( @@ -570,58 +624,56 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx. # Step 4: Register client or use URL-based client ID (CIMD) if not self.context.client_info: + # SEP-2352: the issuer to bind these credentials to, once metadata for it + # was actually found. + discovered_issuer = self._expected_issuer() if self.context.oauth_metadata is not None else None + if should_use_client_metadata_url( self.context.oauth_metadata, self.context.client_metadata_url ): - # Use URL-based client ID (CIMD) + # Use URL-based client ID (CIMD). CIMD records are portable across + # authorization servers, so the issuer stamp is informational. logger.debug(f"Using URL-based client ID (CIMD): {self.context.client_metadata_url}") client_information = create_client_info_from_metadata_url( self.context.client_metadata_url, # type: ignore[arg-type] redirect_uris=self.context.client_metadata.redirect_uris, ) + client_information.issuer = discovered_issuer self.context.client_info = client_information await self.context.storage.set_client_info(client_information) else: # Fallback to Dynamic Client Registration + fallback_base = self.context.get_authorization_base_url(self.context.server_url) registration_request = create_client_registration_request( - self.context.oauth_metadata, - self.context.client_metadata, - self.context.get_authorization_base_url(self.context.server_url), + self.context.oauth_metadata, self.context.client_metadata, fallback_base ) registration_response = yield registration_request client_information = await handle_registration_response(registration_response) + # Only record the issuer when the registration above actually targeted + # the discovered AS - either via its published registration_endpoint, + # or because the resource-origin /register fallback is on the issuer's + # own host (legacy same-origin embedded AS). Otherwise the fallback hit + # a different server and recording a binding to the PRM-advertised AS + # would persist a binding that was never established. + if ( + self.context.oauth_metadata is not None + and discovered_issuer is not None + and ( + self.context.oauth_metadata.registration_endpoint is not None + or self.context.get_authorization_base_url(discovered_issuer) == fallback_base + ) + ): + client_information.issuer = discovered_issuer self.context.client_info = client_information await self.context.storage.set_client_info(client_information) # Step 5: Perform authorization and complete token exchange token_response = yield await self._perform_authorization() await self._handle_token_response(token_response) - except Exception: # pragma: no cover + except Exception: logger.exception("OAuth flow error") raise # Retry with new tokens self._add_auth_header(request) yield request - elif response.status_code == 403: - # Step 1: Extract error field from WWW-Authenticate header - error = extract_field_from_www_auth(response, "error") - - # Step 2: Check if we need to step-up authorization - if error == "insufficient_scope": # pragma: no branch - try: - # Step 2a: Update the required scopes - self.context.client_metadata.scope = get_client_metadata_scopes( - extract_scope_from_www_auth(response), self.context.protected_resource_metadata - ) - - # Step 2b: Perform (re-)authorization and token exchange - token_response = yield await self._perform_authorization() - await self._handle_token_response(token_response) - except Exception: # pragma: no cover - logger.exception("OAuth flow error") - raise - - # Retry with new tokens - self._add_auth_header(request) - yield request diff --git a/src/mcp/client/auth/utils.py b/src/mcp/client/auth/utils.py index b4426be7f8..807d1da984 100644 --- a/src/mcp/client/auth/utils.py +++ b/src/mcp/client/auth/utils.py @@ -1,11 +1,13 @@ import logging import re +from typing import Any, cast from urllib.parse import urljoin, urlparse from httpx import Request, Response from pydantic import AnyUrl, ValidationError +from pydantic_core import from_json -from mcp.client.auth import OAuthRegistrationError, OAuthTokenError +from mcp.client.auth import OAuthFlowError, OAuthRegistrationError, OAuthTokenError from mcp.client.streamable_http import MCP_PROTOCOL_VERSION from mcp.shared.auth import ( OAuthClientInformationFull, @@ -58,7 +60,7 @@ def extract_resource_metadata_from_www_auth(response: Response) -> str | None: Returns: Resource metadata URL if found in WWW-Authenticate header, None otherwise """ - if not response or response.status_code != 401: + if not response or response.status_code not in (401, 403): return None # pragma: no cover return extract_field_from_www_auth(response, "resource_metadata") @@ -208,6 +210,35 @@ async def handle_auth_metadata_response(response: Response) -> tuple[bool, OAuth return True, None +def validate_metadata_issuer(oauth_metadata: OAuthMetadata, expected_issuer: str) -> None: + """Validate that authorization server metadata `issuer` matches the discovery issuer. + + Per RFC 8414 section 3.3 / SEP-2468, the `issuer` in the metadata must match the issuer + used to construct the well-known URL, compared as a simple string (RFC 3986 section 6.2.1). + The one tolerance is an origin with an empty path versus the same origin with a lone `/` + (RFC 3986 section 6.2.3): the SDK's URL type always renders a root issuer with the `/`, and + servers commonly render it either way. + + Raises: + OAuthFlowError: If the metadata issuer does not match `expected_issuer`. + """ + if not issuers_match(str(oauth_metadata.issuer), expected_issuer): + raise OAuthFlowError( + f"Authorization server metadata issuer mismatch: {oauth_metadata.issuer} != {expected_issuer}" + ) + + +def issuers_match(a: str, b: str) -> bool: + """Simple string comparison of two issuer identifiers (RFC 8414 section 3.3), except that a root + issuer with and without its trailing slash (`scheme://authority` and `scheme://authority/`) name + the same server.""" + if a == b: + return True + shorter, longer = sorted((a, b), key=len) + parsed = urlparse(shorter) + return longer == f"{shorter}/" and shorter == f"{parsed.scheme}://{parsed.netloc}" + + def create_oauth_metadata_request(url: str) -> Request: return Request("GET", url, headers={MCP_PROTOCOL_VERSION: LATEST_PROTOCOL_VERSION}) @@ -235,12 +266,17 @@ async def handle_registration_response(response: Response) -> OAuthClientInforma try: content = await response.aread() - client_info = OAuthClientInformationFull.model_validate_json(content) - return client_info - # self.context.client_info = client_info - # await self.context.storage.set_client_info(client_info) - except ValidationError as e: # pragma: no cover - raise OAuthRegistrationError(f"Invalid registration response: {e}") + body = from_json(content) + # `issuer` is the SDK's own binding of these credentials to the server they were + # registered with (SEP-2352), stamped by the auth flow - never sourced from the + # wire, so it is dropped before the body is parsed rather than trusted or cleared. + if isinstance(body, dict): + cast(dict[str, Any], body).pop("issuer", None) + return OAuthClientInformationFull.model_validate(body) + except ValueError as e: + # `from_json` reports malformed bytes/JSON as ValueError, and pydantic's + # ValidationError is itself a ValueError, so both parse layers surface here. + raise OAuthRegistrationError(f"Invalid registration response: {e}") from e def is_valid_client_metadata_url(url: str | None) -> bool: @@ -263,6 +299,26 @@ def is_valid_client_metadata_url(url: str | None) -> bool: return False +def credentials_match_issuer( + client_info: OAuthClientInformationFull, issuer: str, client_metadata_url: str | None +) -> bool: + """Whether stored client credentials may be reused against `issuer` (SEP-2352). + + A URL-based client ID (CIMD) is portable across authorization servers - the same self-hosted + document is resolved by whichever server is in use - so it always matches; CIMD is identified + by the client ID being the configured `client_metadata_url`, not by URL shape (a registration + server may also issue URL-shaped IDs that are bound to it). Credentials with a recorded issuer + match only when it names the same server as `issuer` (`issuers_match`). Credentials with no + recorded issuer (pre-registered, or stored before issuer binding existed) carry no binding to + enforce and are left as-is. + """ + if client_metadata_url is not None and client_info.client_id == client_metadata_url: + return True + if client_info.issuer is None: + return True + return issuers_match(client_info.issuer, issuer) + + def should_use_client_metadata_url( oauth_metadata: OAuthMetadata | None, client_metadata_url: str | None, diff --git a/src/mcp/shared/auth.py b/src/mcp/shared/auth.py index a985bef3f1..59cf2f5723 100644 --- a/src/mcp/shared/auth.py +++ b/src/mcp/shared/auth.py @@ -121,6 +121,9 @@ class OAuthClientInformationFull(OAuthClientMetadata): client_secret: str | None = None client_id_issued_at: int | None = None client_secret_expires_at: int | None = None + # SEP-2352: the issuer these credentials were registered with, recorded by the SDK (not an + # RFC 7591 field) to detect authorization-server migration and avoid cross-AS credential reuse. + issuer: str | None = None class OAuthMetadata(BaseModel): diff --git a/tests/client/auth/extensions/test_client_credentials.py b/tests/client/auth/extensions/test_client_credentials.py index 6d134af742..ec058aa595 100644 --- a/tests/client/auth/extensions/test_client_credentials.py +++ b/tests/client/auth/extensions/test_client_credentials.py @@ -1,9 +1,13 @@ import urllib.parse +from collections.abc import AsyncGenerator +import httpx import jwt import pytest +from inline_snapshot import snapshot from pydantic import AnyHttpUrl, AnyUrl +from mcp.client.auth import OAuthClientProvider, OAuthFlowError from mcp.client.auth.extensions.client_credentials import ( ClientCredentialsOAuthProvider, JWTParameters, @@ -429,3 +433,207 @@ async def test_returns_static_token(self): assert result1 == token assert result2 == token + + +_SERVER_URL = "https://api.example.com/v1/mcp" +_CONFIGURED_ISSUER = "https://auth.example.com" + + +def _metadata_for(issuer: str) -> dict[str, str]: + return {"issuer": issuer, "authorization_endpoint": f"{issuer}/authorize", "token_endpoint": f"{issuer}/token"} + + +def _provider_with_issuer(kind: str, storage: MockTokenStorage, audiences: list[str]) -> OAuthClientProvider: + """A ClientCredentials ("secret") or PrivateKeyJWT ("jwt") provider configured for _CONFIGURED_ISSUER; + `audiences` records every audience an assertion is minted for.""" + if kind == "secret": + return ClientCredentialsOAuthProvider( + server_url=_SERVER_URL, storage=storage, client_id="cid", client_secret="csecret", issuer=_CONFIGURED_ISSUER + ) + + async def assertion_provider(audience: str) -> str: + audiences.append(audience) + return "signed-assertion" + + return PrivateKeyJWTOAuthProvider( + server_url=_SERVER_URL, + storage=storage, + client_id="cid", + assertion_provider=assertion_provider, + issuer=_CONFIGURED_ISSUER, + ) + + +async def _answer_discovery( + flow: AsyncGenerator[httpx.Request, httpx.Response], + *, + authorization_server: str | list[str] | None, + metadata: dict[str, str] | None, +) -> httpx.Request: + """Answer the provider's first request with a 401 and its discovery requests as described; + return the request it builds once discovery is over. + + `authorization_server` is what protected-resource metadata advertises (None: no PRM is + served); `metadata` is the authorization server metadata document (None: every well-known + 404s). + """ + request = await flow.__anext__() + request = await flow.asend(httpx.Response(401, request=request)) + while "/.well-known/oauth-protected-resource" in str(request.url): + if authorization_server is None: + response = httpx.Response(404, request=request) + else: + advertised = authorization_server if isinstance(authorization_server, list) else [authorization_server] + prm = {"resource": _SERVER_URL, "authorization_servers": advertised} + response = httpx.Response(200, json=prm, request=request) + request = await flow.asend(response) + while "/.well-known/" in str(request.url): + if metadata is None: + response = httpx.Response(404, request=request) + else: + response = httpx.Response(200, json=metadata, request=request) + request = await flow.asend(response) + return request + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "served_issuer", [_CONFIGURED_ISSUER, f"{_CONFIGURED_ISSUER}/"], ids=["as-configured", "root-slash"] +) +@pytest.mark.parametrize("kind", ["secret", "jwt"]) +async def test_provider_with_configured_issuer_exchanges_at_that_issuer( + mock_storage: MockTokenStorage, kind: str, served_issuer: str +): + """SDK-defined: with `issuer=` set and metadata discovered for that issuer (a root issuer served with + its trailing slash is the same server), the token request goes to its token endpoint (positive + control for the refusals below).""" + audiences: list[str] = [] + provider = _provider_with_issuer(kind, mock_storage, audiences) + flow = provider.async_auth_flow(httpx.Request("POST", _SERVER_URL)) + metadata = {**_metadata_for(_CONFIGURED_ISSUER), "issuer": served_issuer} + + token_request = await _answer_discovery(flow, authorization_server=served_issuer, metadata=metadata) + + assert (token_request.method, str(token_request.url)) == ("POST", "https://auth.example.com/token") + # The SDK's URL type renders a root issuer with its trailing slash, which is the audience used. + assert audiences == ([] if kind == "secret" else ["https://auth.example.com/"]) + await flow.aclose() + + +@pytest.mark.anyio +@pytest.mark.parametrize("kind", ["secret", "jwt"]) +async def test_provider_picks_its_configured_issuer_among_several_advertised_servers( + mock_storage: MockTokenStorage, kind: str +): + """SDK-defined: when the resource lists several authorization servers, the one matching `issuer=` is + discovered and used even if it is not listed first.""" + provider = _provider_with_issuer(kind, mock_storage, []) + flow = provider.async_auth_flow(httpx.Request("POST", _SERVER_URL)) + + token_request = await _answer_discovery( + flow, + authorization_server=["https://other-as.example.com", _CONFIGURED_ISSUER], + metadata=_metadata_for(_CONFIGURED_ISSUER), + ) + + assert provider.context.auth_server_url == f"{_CONFIGURED_ISSUER}/" + assert str(token_request.url) == "https://auth.example.com/token" + await flow.aclose() + + +def test_an_issuer_that_is_not_an_http_url_is_rejected_at_construction(mock_storage: MockTokenStorage) -> None: + """SDK-defined: `issuer=` is the authorization server's issuer URL; anything else is a configuration + error on both machine-to-machine providers.""" + with pytest.raises(ValueError) as cc_error: + ClientCredentialsOAuthProvider( + server_url=_SERVER_URL, storage=mock_storage, client_id="cid", client_secret="s", issuer="auth.example.com" + ) + with pytest.raises(ValueError) as jwt_error: + PrivateKeyJWTOAuthProvider( + server_url=_SERVER_URL, + storage=mock_storage, + client_id="cid", + assertion_provider=static_assertion_provider("jwt"), + issuer="auth.example.com", + ) + assert ( + str(cc_error.value) + == str(jwt_error.value) + == snapshot("issuer must be the authorization server's http(s) issuer URL, got 'auth.example.com'") + ) + + +@pytest.mark.anyio +@pytest.mark.parametrize("kind", ["secret", "jwt"]) +async def test_provider_refuses_metadata_for_a_different_issuer(mock_storage: MockTokenStorage, kind: str): + """SDK-defined: when discovery ends at an authorization server other than the configured `issuer`, + no token request is built and no assertion is minted.""" + audiences: list[str] = [] + provider = _provider_with_issuer(kind, mock_storage, audiences) + flow = provider.async_auth_flow(httpx.Request("POST", _SERVER_URL)) + + with pytest.raises(OAuthFlowError) as exc_info: + await _answer_discovery( + flow, + authorization_server="https://other-as.example.com", + metadata=_metadata_for("https://other-as.example.com"), + ) + + assert str(exc_info.value) == snapshot( + "Authorization server metadata issuer mismatch: https://other-as.example.com/ != https://auth.example.com" + ) + assert audiences == [] + + +@pytest.mark.anyio +@pytest.mark.parametrize("kind", ["secret", "jwt"]) +async def test_provider_refuses_to_exchange_without_metadata_when_issuer_configured( + mock_storage: MockTokenStorage, kind: str +): + """SDK-defined: with `issuer=` set, the 2025-03-26 default `/token` on the resource origin is not + used when no authorization server metadata could be discovered.""" + audiences: list[str] = [] + provider = _provider_with_issuer(kind, mock_storage, audiences) + flow = provider.async_auth_flow(httpx.Request("POST", _SERVER_URL)) + + with pytest.raises(OAuthFlowError) as exc_info: + await _answer_discovery(flow, authorization_server=None, metadata=None) + + assert str(exc_info.value) == snapshot( + "No authorization server metadata discovered for configured issuer https://auth.example.com" + ) + assert audiences == [] + + +@pytest.mark.anyio +@pytest.mark.parametrize("kind", ["secret", "jwt"]) +async def test_a_refused_authorization_server_is_forgotten_so_the_next_request_rediscovers( + mock_storage: MockTokenStorage, kind: str +): + """SDK-defined: when the exchange is refused because discovery ended somewhere other than the + configured issuer, the refused metadata and any token held are dropped; the next request goes out + unauthenticated and discovery starts again, rather than a refresh being built from what was refused.""" + provider = _provider_with_issuer(kind, mock_storage, []) + flow = provider.async_auth_flow(httpx.Request("POST", _SERVER_URL)) + token_request = await _answer_discovery( + flow, authorization_server=_CONFIGURED_ISSUER, metadata=_metadata_for(_CONFIGURED_ISSUER) + ) + token = {"access_token": "first", "token_type": "Bearer", "expires_in": 3600, "refresh_token": "rt"} + retried = await flow.asend(httpx.Response(200, json=token, request=token_request)) + with pytest.raises(StopAsyncIteration): + await flow.asend(httpx.Response(200, request=retried)) + + flow = provider.async_auth_flow(httpx.Request("POST", _SERVER_URL)) + with pytest.raises(OAuthFlowError): + await _answer_discovery( + flow, + authorization_server="https://other-as.example.com", + metadata=_metadata_for("https://other-as.example.com"), + ) + assert provider.context.oauth_metadata is None + assert provider.context.current_tokens is None + + flow = provider.async_auth_flow(httpx.Request("POST", _SERVER_URL)) + request = await flow.__anext__() + assert (str(request.url), request.headers.get("Authorization")) == (_SERVER_URL, None) + await flow.aclose() diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py index 5f8bc14107..3ebd2e061a 100644 --- a/tests/client/test_auth.py +++ b/tests/client/test_auth.py @@ -3,9 +3,10 @@ """ import base64 +import json import time from unittest import mock -from urllib.parse import unquote +from urllib.parse import parse_qs, unquote, urlparse import httpx import pytest @@ -13,13 +14,14 @@ from pydantic import AnyHttpUrl, AnyUrl from mcp.client.auth import OAuthClientProvider, PKCEParameters -from mcp.client.auth.exceptions import OAuthFlowError +from mcp.client.auth.exceptions import OAuthFlowError, OAuthRegistrationError from mcp.client.auth.utils import ( build_oauth_authorization_server_metadata_discovery_urls, build_protected_resource_metadata_discovery_urls, create_client_info_from_metadata_url, create_client_registration_request, create_oauth_metadata_request, + credentials_match_issuer, extract_field_from_www_auth, extract_resource_metadata_from_www_auth, extract_scope_from_www_auth, @@ -27,6 +29,7 @@ handle_registration_response, is_valid_client_metadata_url, should_use_client_metadata_url, + validate_metadata_issuer, ) from mcp.shared.auth import ( OAuthClientInformationFull, @@ -1144,8 +1147,11 @@ async def mock_callback() -> tuple[str, str | None]: request=request, ) - # Trigger step-up - should get token exchange request - token_exchange_request = await auth_flow.asend(response_403) + # Trigger step-up - discovery runs first (nothing published here), then the token exchange + prm_request = await auth_flow.asend(response_403) + prm_request = await auth_flow.asend(httpx.Response(404, request=prm_request)) + asm_request = await auth_flow.asend(httpx.Response(404, request=prm_request)) + token_exchange_request = await auth_flow.asend(httpx.Response(404, request=asm_request)) # Verify scope was updated assert oauth_provider.context.client_metadata.scope == "admin:write admin:delete" @@ -1403,8 +1409,8 @@ async def callback_handler() -> tuple[str, str | None]: prm_request_1 = await auth_flow.asend(response) assert str(prm_request_1.url) == "https://custom.prm.com/.well-known/oauth-protected-resource" - # Returns 500 - prm_response_1 = httpx.Response(500, request=prm_request_1) + # Not served there + prm_response_1 = httpx.Response(404, request=prm_request_1) # Try path-based fallback prm_request_2 = await auth_flow.asend(prm_response_1) @@ -2113,3 +2119,757 @@ async def test_get_resource_url_falls_back_when_prm_mismatches( # get_resource_url should return the canonical server URL, not the PRM resource assert provider.context.get_resource_url() == "https://api.example.com/v1/mcp" + + +def _prepare_full_flow(provider: OAuthClientProvider, client_info: OAuthClientInformationFull | None) -> list[str]: + """Reset `provider` for a full flow with `client_info` as the stored registration, and wire a + redirect/callback pair that echoes the `state` of the last authorization URL it was sent to. + Returns the list the redirect handler appends authorization URLs to.""" + provider.context.current_tokens = None + provider.context.token_expiry_time = None + provider._initialized = True + provider.context.client_info = client_info + redirects: list[str] = [] + + async def record_redirect(url: str) -> None: + redirects.append(url) + + async def echo_callback() -> tuple[str, str | None]: + return "auth_code", parse_qs(urlparse(redirects[-1]).query)["state"][0] + + provider.context.redirect_handler = record_redirect + provider.context.callback_handler = echo_callback + return redirects + + +def _asm(issuer: str, *, token_origin: str | None = None, registration: bool = False) -> bytes: + metadata = { + "issuer": issuer, + "authorization_endpoint": f"{issuer}/authorize", + "token_endpoint": f"{token_origin or issuer}/token", + } + if registration: + metadata["registration_endpoint"] = f"{issuer}/register" + return json.dumps(metadata).encode() + + +@pytest.mark.anyio +async def test_metadata_issuer_must_match_the_advertised_authorization_server(oauth_provider: OAuthClientProvider): + """RFC 8414 section 3.3: metadata fetched for the PRM-advertised authorization server must + name that server as its issuer; metadata naming another issuer is refused before + registration, authorization or token requests are built from it.""" + _prepare_full_flow(oauth_provider, None) + auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + + prm_req = await auth_flow.asend(httpx.Response(401, request=request)) + prm_response = httpx.Response( + 200, + content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}', + request=prm_req, + ) + asm_req = await auth_flow.asend(prm_response) + assert str(asm_req.url) == "https://auth.example.com/.well-known/oauth-authorization-server" + + asm_response = httpx.Response(200, content=_asm("https://other-as.example.com", registration=True), request=asm_req) + with pytest.raises(OAuthFlowError) as exc_info: + await auth_flow.asend(asm_response) + + assert str(exc_info.value) == snapshot( + "Authorization server metadata issuer mismatch: https://other-as.example.com/ != https://auth.example.com/" + ) + + +@pytest.mark.anyio +async def test_legacy_fallback_metadata_naming_a_different_issuer_is_refused(oauth_provider: OAuthClientProvider): + """RFC 8414 section 3.3 on the legacy no-PRM path: metadata served from the resource server's + own well-known must name that origin as its issuer. + + Metadata naming a different authorization server is refused before any authorization or + token request is built, so a stored confidential client is never presented to the endpoints + that metadata lists. + """ + _prepare_full_flow( + oauth_provider, + OAuthClientInformationFull( + client_id="confidential-client", + client_secret="stored-secret", + token_endpoint_auth_method="client_secret_post", + redirect_uris=[AnyUrl("http://localhost:3030/callback")], + ), + ) + auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + + # 401 without WWW-Authenticate; both PRM well-knowns 404; legacy root ASM discovery. + prm_req = await auth_flow.asend(httpx.Response(401, request=request)) + prm_req = await auth_flow.asend(httpx.Response(404, request=prm_req)) + asm_req = await auth_flow.asend(httpx.Response(404, request=prm_req)) + assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server" + + # The resource origin's well-known names another server as issuer while listing its own + # token endpoint. + asm_response = httpx.Response( + 200, content=_asm("https://other-as.example.com", token_origin="https://api.example.com"), request=asm_req + ) + with pytest.raises(OAuthFlowError) as exc_info: + await auth_flow.asend(asm_response) + + assert str(exc_info.value) == snapshot( + "Authorization server metadata issuer mismatch: https://other-as.example.com/ != https://api.example.com/" + ) + + +_ISSUER = "https://as.example.com/tenant" + + +def _issuer_metadata(issuer: str = _ISSUER) -> OAuthMetadata: + return OAuthMetadata.model_validate( + {"issuer": issuer, "authorization_endpoint": f"{issuer}/authorize", "token_endpoint": f"{issuer}/token"} + ) + + +def test_validate_metadata_issuer_accepts_match(): + validate_metadata_issuer(_issuer_metadata(_ISSUER), _ISSUER) + + +def test_validate_metadata_issuer_rejects_mismatch(): + with pytest.raises(OAuthFlowError, match="issuer mismatch"): + validate_metadata_issuer(_issuer_metadata("https://other-as.example.com/tenant"), _ISSUER) + + +@pytest.mark.parametrize( + ("issuer", "expected"), + [ + pytest.param("https://as.example.com/", "https://as.example.com", id="metadata-has-root-slash"), + pytest.param("https://as.example.com/", "https://as.example.com/", id="both-have-root-slash"), + ], +) +def test_validate_metadata_issuer_treats_empty_path_and_root_slash_as_the_same_issuer(issuer: str, expected: str): + """SDK-defined tolerance: an origin with an empty path and the same origin with a lone `/` + identify the same server (RFC 3986 section 6.2.3). A root issuer always parses to the `/` + form here, while the legacy discovery URL is built from the bare origin, so the two must + compare equal.""" + validate_metadata_issuer(_issuer_metadata(issuer), expected) + + +@pytest.mark.parametrize( + ("issuer", "expected"), + [ + pytest.param("https://as.example.com/tenant/", "https://as.example.com/tenant", id="non-root-trailing-slash"), + pytest.param("https://as.example.com/tenant", "https://as.example.com", id="different-path"), + pytest.param("http://as.example.com/", "https://as.example.com", id="different-scheme"), + pytest.param("https://as.example.com:8443/", "https://as.example.com", id="different-port"), + pytest.param("https://as.example.com//", "https://as.example.com", id="double-slash"), + ], +) +def test_validate_metadata_issuer_root_slash_tolerance_does_not_extend_further(issuer: str, expected: str): + """The empty-path tolerance is exactly that: any other difference is still a mismatch.""" + with pytest.raises(OAuthFlowError, match="metadata issuer mismatch"): + validate_metadata_issuer(_issuer_metadata(issuer), expected) + + +def test_credentials_match_issuer_same_issuer(): + info = OAuthClientInformationFull(client_id="c", redirect_uris=[AnyUrl("http://localhost/cb")], issuer="https://as") + assert credentials_match_issuer(info, "https://as", None) is True + + +def test_credentials_match_issuer_root_slash_is_the_same_issuer(): + """A binding written as the bare origin matches the `/` form a root URL parses to, and back.""" + info = OAuthClientInformationFull(client_id="c", redirect_uris=[AnyUrl("http://localhost/cb")], issuer="https://as") + assert credentials_match_issuer(info, "https://as/", None) is True + info.issuer = "https://as/" + assert credentials_match_issuer(info, "https://as", None) is True + + +def test_credentials_match_issuer_root_slash_tolerance_does_not_extend_to_other_paths(): + """SDK-defined: a trailing slash on a non-root path is a different issuer.""" + info = OAuthClientInformationFull( + client_id="c", redirect_uris=[AnyUrl("http://localhost/cb")], issuer="https://as.example.com/tenant" + ) + assert credentials_match_issuer(info, "https://as.example.com/tenant/", None) is False + + +def test_credentials_match_issuer_different_issuer(): + info = OAuthClientInformationFull(client_id="c", redirect_uris=[AnyUrl("http://localhost/cb")], issuer="https://as") + assert credentials_match_issuer(info, "https://other", None) is False + + +def test_credentials_match_issuer_no_recorded_issuer_is_left_alone(): + """Credentials with no bound issuer (pre-registered / legacy) carry no binding to enforce.""" + info = OAuthClientInformationFull(client_id="c", redirect_uris=[AnyUrl("http://localhost/cb")]) + assert credentials_match_issuer(info, "https://as", None) is True + + +def test_credentials_match_issuer_cimd_is_portable(): + """A client_id equal to the configured client_metadata_url (CIMD) is portable across servers.""" + cimd_url = "https://client.example/metadata.json" + info = OAuthClientInformationFull( + client_id=cimd_url, + redirect_uris=[AnyUrl("http://localhost/cb")], + token_endpoint_auth_method="none", + issuer="https://as", + ) + assert credentials_match_issuer(info, "https://other", cimd_url) is True + + +def test_credentials_match_issuer_url_shaped_dcr_id_is_not_portable(): + """A URL-shaped client_id from DCR (not the configured CIMD URL) stays bound to its issuer.""" + info = OAuthClientInformationFull( + client_id="https://as.example.com/clients/123", + redirect_uris=[AnyUrl("http://localhost/cb")], + issuer="https://as.example.com", + ) + assert credentials_match_issuer(info, "https://other", "https://client.example/metadata.json") is False + + +@pytest.mark.anyio +@pytest.mark.parametrize("echoed_issuer", ["https://not-the-flow.example", 12345], ids=["string", "not-a-string"]) +async def test_registration_response_does_not_seed_the_issuer_binding_from_the_body(echoed_issuer: object): + """The issuer binding (SEP-2352) is the SDK's record of which server it registered with, + stamped by the auth flow; an "issuer" member in the untrusted response body is dropped + before parsing - never populating the binding, and never failing the parse either, so a + mismatched or malformed value cannot discard the credentials on every 401.""" + body = json.dumps( + {"client_id": "issued-id", "redirect_uris": ["http://localhost:3030/callback"], "issuer": echoed_issuer} + ).encode() + + client_info = await handle_registration_response(httpx.Response(201, content=body)) + + assert client_info.client_id == "issued-id" + assert client_info.issuer is None + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "content", + [b"not json", b'["json", "but", "not", "an", "object"]', '{"client_id": "caf\xe9"}'.encode("latin-1")], + ids=["not-json", "not-an-object", "not-utf8"], +) +async def test_a_2xx_body_that_is_not_client_information_is_an_oauth_registration_error(content: bytes): + """A success status whose body is not client information - unparseable, not an object, or + not valid UTF-8 - surfaces as OAuthRegistrationError rather than a raw parse failure, so a + single OAuthFlowError handler still covers registration.""" + with pytest.raises(OAuthRegistrationError): + await handle_registration_response(httpx.Response(201, content=content)) + + +@pytest.mark.anyio +async def test_stored_credentials_are_not_presented_to_a_different_authorization_server( + oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage +): + """SEP-2352: stored credentials are bound to the authorization server that registered them. + + Steps: + 1. Storage holds a confidential client bound to `https://auth.example.com/`. + 2. PRM now advertises `https://other-as.example.com` -> the stored client and its tokens are + discarded before that server's metadata is fetched. + 3. Metadata for the new server is discovered -> the flow registers there and the token + request carries the new client, not the discarded `client_id`/`client_secret`. + 4. The new registration is recorded as bound to the new server. + """ + redirects = _prepare_full_flow( + oauth_provider, + OAuthClientInformationFull( + client_id="confidential-client", + client_secret="stored-secret", + token_endpoint_auth_method="client_secret_post", + redirect_uris=[AnyUrl("http://localhost:3030/callback")], + issuer="https://auth.example.com/", + ), + ) + auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + + prm_req = await auth_flow.asend(httpx.Response(401, request=request)) + prm_response = httpx.Response( + 200, + content=( + b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://other-as.example.com"]}' + ), + request=prm_req, + ) + asm_req = await auth_flow.asend(prm_response) + assert str(asm_req.url) == "https://other-as.example.com/.well-known/oauth-authorization-server" + assert oauth_provider.context.client_info is None + + register_req = await auth_flow.asend( + httpx.Response(200, content=_asm("https://other-as.example.com", registration=True), request=asm_req) + ) + assert register_req.method == "POST" + assert str(register_req.url) == "https://other-as.example.com/register" + register_response = httpx.Response( + 201, + json={"client_id": "new-client", "redirect_uris": ["http://localhost:3030/callback"]}, + request=register_req, + ) + + token_req = await auth_flow.asend(register_response) + assert str(token_req.url) == "https://other-as.example.com/token" + assert redirects[-1].startswith("https://other-as.example.com/authorize?") + token_form = parse_qs(token_req.content.decode()) + assert token_form["client_id"] == ["new-client"] + assert "client_secret" not in token_form + + stored = await mock_storage.get_client_info() + assert stored is not None + assert stored.client_id == "new-client" + assert stored.issuer == "https://other-as.example.com/" + await auth_flow.aclose() + + +@pytest.mark.anyio +async def test_stored_credentials_bound_to_the_advertised_authorization_server_are_kept( + oauth_provider: OAuthClientProvider, +): + """SEP-2352 positive control: a stored client bound to the server PRM advertises (written + with or without the root slash) is reused - no registration request is made.""" + _prepare_full_flow( + oauth_provider, + OAuthClientInformationFull( + client_id="bound-client", + redirect_uris=[AnyUrl("http://localhost:3030/callback")], + issuer="https://auth.example.com", + ), + ) + auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + + prm_req = await auth_flow.asend(httpx.Response(401, request=request)) + prm_response = httpx.Response( + 200, + content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}', + request=prm_req, + ) + asm_req = await auth_flow.asend(prm_response) + assert oauth_provider.context.client_info is not None + + token_req = await auth_flow.asend( + httpx.Response(200, content=_asm("https://auth.example.com", registration=True), request=asm_req) + ) + assert str(token_req.url) == "https://auth.example.com/token" + assert parse_qs(token_req.content.decode())["client_id"] == ["bound-client"] + await auth_flow.aclose() + + +@pytest.mark.anyio +async def test_issuer_binding_evaluated_against_the_server_origin_when_prm_discovery_failed( + oauth_provider: OAuthClientProvider, +): + """SEP-2352: on the legacy no-PRM path the binding check uses the resource server's origin. + + PRM discovery fails (404) so `auth_server_url` stays `None`; the legacy well-known URL is + built from the resource server's origin, which is therefore the issuer any metadata found + there must carry (RFC 8414 section 3.3). Stored credentials bound to a different issuer are + discarded before that metadata is fetched, and the flow re-registers. + """ + _prepare_full_flow( + oauth_provider, + OAuthClientInformationFull( + client_id="stale-client", + redirect_uris=[AnyUrl("http://localhost:3030/callback")], + issuer="https://old-as.example.com", + ), + ) + auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + + # PRM discovery: path-based then root, both 404. + prm_req = await auth_flow.asend(httpx.Response(401, request=request)) + assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp" + prm_req = await auth_flow.asend(httpx.Response(404, request=prm_req)) + assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource" + + # ASM discovery via root fallback (no auth_server_url): the stale credentials are already + # gone when the request is issued. + asm_req = await auth_flow.asend(httpx.Response(404, request=prm_req)) + assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server" + assert oauth_provider.context.client_info is None + + # The stale bound credentials are discarded, so the next yield is a DCR request rather than + # the authorize redirect. + next_req = await auth_flow.asend( + httpx.Response(200, content=_asm("https://api.example.com", registration=True), request=asm_req) + ) + assert oauth_provider.context.auth_server_url is None + assert next_req.method == "POST" + assert str(next_req.url) == "https://api.example.com/register" + await auth_flow.aclose() + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "first_response", + [(401, {}), (403, {"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="admin"'})], + ids=["401", "403-insufficient-scope"], +) +async def test_legacy_fallback_without_metadata_re_registers_instead_of_presenting_credentials_bound_elsewhere( + oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken, first_response: tuple[int, dict[str, str]] +): + """SEP-2352 on the legacy no-PRM path when no metadata is served at all, whether the flow starts + from a 401 or from a 403 scope challenge with no metadata held. + + Steps: + 1. Storage holds a token and a confidential client bound to a different authorization server. + 2. Both PRM well-knowns 404 -> the expected issuer is the resource server's origin, so the + stored client is discarded before ASM discovery. + 3. The origin's ASM well-known 404s too -> the flow registers a fresh client at the + origin's default `/register` and authorizes with it; the token request to the origin's + default `/token` carries the new client and none of the discarded credentials. + """ + redirects = _prepare_full_flow( + oauth_provider, + OAuthClientInformationFull( + client_id="confidential-client", + client_secret="stored-secret", + token_endpoint_auth_method="client_secret_post", + redirect_uris=[AnyUrl("http://localhost:3030/callback")], + issuer="https://other-as.example.com", + ), + ) + oauth_provider.context.current_tokens = valid_tokens + oauth_provider.context.token_expiry_time = time.time() + 1800 + auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + + status, headers = first_response + prm_req = await auth_flow.asend(httpx.Response(status, headers=headers, request=request)) + prm_req = await auth_flow.asend(httpx.Response(404, request=prm_req)) + asm_req = await auth_flow.asend(httpx.Response(404, request=prm_req)) + assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server" + assert oauth_provider.context.client_info is None + + # No metadata at the origin either: register at the origin's default endpoint. + register_req = await auth_flow.asend(httpx.Response(404, request=asm_req)) + assert register_req.method == "POST" + assert str(register_req.url) == "https://api.example.com/register" + register_response = httpx.Response( + 201, + json={ + "client_id": "origin-client", + "redirect_uris": ["http://localhost:3030/callback"], + "token_endpoint_auth_method": "none", + }, + request=register_req, + ) + + token_req = await auth_flow.asend(register_response) + assert str(token_req.url) == "https://api.example.com/token" + assert redirects[-1].startswith("https://api.example.com/authorize?") + token_form = parse_qs(token_req.content.decode()) + assert token_form["client_id"] == ["origin-client"] + assert "client_secret" not in token_form + await auth_flow.aclose() + + +@pytest.mark.anyio +async def test_scope_step_up_discovers_the_authorization_server_before_reauthorizing( + oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken +): + """SDK-defined: a 403 scope challenge runs discovery first when no metadata is held, so + re-authorization targets the advertised server. + + Steps: + 1. A restarted client holds a token and a registration but no authorization server metadata. + 2. The first response is 403 insufficient_scope -> the next requests are PRM (at the challenge's + `resource_metadata` URL) then ASM discovery. + 3. The authorization redirect and the token request use the discovered server's endpoints and + ask for the challenged scope. + """ + redirects = _prepare_full_flow( + oauth_provider, + OAuthClientInformationFull( + client_id="registered-client", redirect_uris=[AnyUrl("http://localhost:3030/callback")] + ), + ) + oauth_provider.context.current_tokens = valid_tokens + oauth_provider.context.token_expiry_time = time.time() + 1800 + auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + response_403 = httpx.Response( + 403, + headers={ + "WWW-Authenticate": 'Bearer error="insufficient_scope", scope="admin",' + ' resource_metadata="https://api.example.com/v1/mcp/resource-metadata"' + }, + request=request, + ) + + prm_request = await auth_flow.asend(response_403) + assert (prm_request.method, str(prm_request.url)) == ("GET", "https://api.example.com/v1/mcp/resource-metadata") + prm = b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}' + asm_request = await auth_flow.asend(httpx.Response(200, content=prm, request=prm_request)) + assert str(asm_request.url) == "https://auth.example.com/.well-known/oauth-authorization-server" + + token_request = await auth_flow.asend( + httpx.Response(200, content=_asm("https://auth.example.com"), request=asm_request) + ) + + assert redirects[-1].startswith("https://auth.example.com/authorize?") + assert parse_qs(urlparse(redirects[-1]).query)["scope"] == ["admin"] + assert (token_request.method, str(token_request.url)) == ("POST", "https://auth.example.com/token") + await auth_flow.aclose() + + +@pytest.mark.anyio +async def test_scope_step_up_reuses_metadata_discovered_earlier_in_the_process( + oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken +): + """SDK-defined: once metadata has been discovered in this process, a step-up re-authorizes with it + directly (no discovery requests) and asks for the challenged scope.""" + redirects = _prepare_full_flow( + oauth_provider, + OAuthClientInformationFull( + client_id="registered-client", redirect_uris=[AnyUrl("http://localhost:3030/callback")] + ), + ) + oauth_provider.context.current_tokens = valid_tokens + oauth_provider.context.token_expiry_time = time.time() + 1800 + oauth_provider.context.auth_server_url = "https://auth.example.com/" + oauth_provider.context.oauth_metadata = OAuthMetadata.model_validate_json(_asm("https://auth.example.com")) + auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + response_403 = httpx.Response( + 403, headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="admin"'}, request=request + ) + + token_request = await auth_flow.asend(response_403) + + assert (token_request.method, str(token_request.url)) == ("POST", "https://auth.example.com/token") + assert redirects[-1].startswith("https://auth.example.com/authorize?") + assert parse_qs(urlparse(redirects[-1]).query)["scope"] == ["admin"] + await auth_flow.aclose() + + +@pytest.mark.anyio +async def test_403_without_a_scope_challenge_is_returned_to_the_caller( + oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken +): + """SDK-defined: a 403 that is not an insufficient_scope challenge ends the flow; the request is + not retried.""" + _prepare_full_flow(oauth_provider, None) + oauth_provider.context.current_tokens = valid_tokens + oauth_provider.context.token_expiry_time = time.time() + 1800 + auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + + with pytest.raises(StopAsyncIteration): + await auth_flow.asend( + httpx.Response(403, headers={"WWW-Authenticate": 'Bearer error="access_denied"'}, request=request) + ) + + +@pytest.mark.anyio +@pytest.mark.parametrize("status", [500, 503, 429]) +async def test_a_failing_resource_metadata_request_stops_the_flow_and_keeps_stored_credentials( + oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken, status: int +): + """SDK-defined: a server error (or 429) on a protected resource metadata request says nothing about + whether the server publishes that metadata. The remaining well-known locations are still tried, but + when none answers the flow stops instead of taking the legacy path, and a registration bound to the + advertised authorization server and its tokens stay as they were.""" + bound = OAuthClientInformationFull( + client_id="registered-client", + redirect_uris=[AnyUrl("http://localhost:3030/callback")], + issuer="https://auth.example.com/", + ) + _prepare_full_flow(oauth_provider, bound) + oauth_provider.context.current_tokens = valid_tokens + oauth_provider.context.token_expiry_time = time.time() + 1800 + flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await flow.__anext__() + prm_request = await flow.asend(httpx.Response(401, request=request)) + root_prm_request = await flow.asend(httpx.Response(status, request=prm_request)) + assert str(root_prm_request.url) == "https://api.example.com/.well-known/oauth-protected-resource" + + with pytest.raises(OAuthFlowError) as exc_info: + await flow.asend(httpx.Response(404, request=root_prm_request)) + + assert str(exc_info.value) == f"Protected resource metadata request failed: HTTP {status}" + assert oauth_provider.context.client_info == bound + assert oauth_provider.context.current_tokens == valid_tokens + + +@pytest.mark.anyio +async def test_a_failing_resource_metadata_location_does_not_matter_when_another_one_answers( + oauth_provider: OAuthClientProvider, +): + """SDK-defined: the well-known locations are tried in order; an error at one of them is forgotten + once a later one returns the metadata.""" + _prepare_full_flow(oauth_provider, None) + flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await flow.__anext__() + prm_request = await flow.asend(httpx.Response(401, request=request)) + root_prm_request = await flow.asend(httpx.Response(503, request=prm_request)) + prm = b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}' + + asm_request = await flow.asend(httpx.Response(200, content=prm, request=root_prm_request)) + + assert str(asm_request.url) == "https://auth.example.com/.well-known/oauth-authorization-server" + await flow.aclose() + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "served_issuer", ["https://api.example.com", "https://api.example.com/"], ids=["bare", "root-slash"] +) +async def test_legacy_fallback_accepts_the_origin_issuer_for_a_server_url_in_any_spelling( + client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage, served_issuer: str +): + """SDK-defined: on the legacy no-PRM path the expected issuer is the resource server's origin; a + `server_url` written with an upper-case host and an explicit default port still matches metadata + naming that origin, with or without its trailing slash, and the flow proceeds to registration.""" + + async def redirect_handler(url: str) -> None: + raise NotImplementedError + + async def callback_handler() -> tuple[str, str | None]: + raise NotImplementedError + + provider = OAuthClientProvider( + server_url="https://API.Example.com:443/v1/mcp", + client_metadata=client_metadata, + storage=mock_storage, + redirect_handler=redirect_handler, + callback_handler=callback_handler, + ) + auth_flow = provider.async_auth_flow(httpx.Request("GET", "https://API.Example.com:443/v1/mcp")) + request = await auth_flow.__anext__() + prm_req = await auth_flow.asend(httpx.Response(401, request=request)) + prm_req = await auth_flow.asend(httpx.Response(404, request=prm_req)) + asm_req = await auth_flow.asend(httpx.Response(404, request=prm_req)) + asm = { + "issuer": served_issuer, + "authorization_endpoint": "https://api.example.com/authorize", + "token_endpoint": "https://api.example.com/token", + "registration_endpoint": "https://api.example.com/register", + } + + register_req = await auth_flow.asend(httpx.Response(200, json=asm, request=asm_req)) + + assert (register_req.method, str(register_req.url)) == ("POST", "https://api.example.com/register") + await auth_flow.aclose() + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "asm_responses", + [ + pytest.param([httpx.Response(404), httpx.Response(404)], id="asm-discovery-failed"), + pytest.param( + [httpx.Response(200, content=_asm("https://new-as.example.com"))], + id="asm-metadata-without-registration-endpoint", + ), + ], +) +async def test_issuer_is_not_stamped_when_registration_falls_back_to_the_resource_origin( + oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage, asm_responses: list[httpx.Response] +): + """SEP-2352: a fallback registration is not recorded as bound to the PRM-advertised AS. + + PRM advertises a new authorization server, so the stored credentials (bound to the old + issuer) are discarded. DCR then falls back to the resource-server origin's `/register` + because the new AS's metadata either could not be discovered or omits + `registration_endpoint`. That registration was not derived from the new AS's metadata, + so persisting it as bound to the new AS would wedge the binding check on later flows; + instead the issuer is left unset. + """ + _prepare_full_flow( + oauth_provider, + OAuthClientInformationFull( + client_id="stale-client", + redirect_uris=[AnyUrl("http://localhost:3030/callback")], + issuer="https://api.example.com/", + ), + ) + auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + response_401 = httpx.Response( + 401, + headers={ + "WWW-Authenticate": 'Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"' + }, + request=request, + ) + + # PRM succeeds and advertises a new AS - the discard block fires. + prm_req = await auth_flow.asend(response_401) + assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource" + prm_response = httpx.Response( + 200, + content=( + b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://new-as.example.com"]}' + ), + request=prm_req, + ) + + # ASM discovery for the new AS yields no usable registration_endpoint - either every + # well-known URL 404s, or metadata is returned without one. + next_req = await auth_flow.asend(prm_response) + assert oauth_provider.context.client_info is None + assert oauth_provider.context.oauth_metadata is None + assert str(next_req.url) == "https://new-as.example.com/.well-known/oauth-authorization-server" + for asm_response in asm_responses: + asm_response.request = next_req + next_req = await auth_flow.asend(asm_response) + + # Step 4 falls back to the resource-server origin's /register. + dcr_req = next_req + assert dcr_req.method == "POST" + assert str(dcr_req.url) == "https://api.example.com/register" + dcr_response = httpx.Response( + 201, + json={"client_id": "fallback-client", "redirect_uris": ["http://localhost:3030/callback"]}, + request=dcr_req, + ) + await auth_flow.asend(dcr_response) + + # The persisted record carries no issuer binding - not the PRM-advertised AS we never reached. + stored = await mock_storage.get_client_info() + assert stored is not None + assert stored.client_id == "fallback-client" + assert stored.issuer is None + await auth_flow.aclose() + + +@pytest.mark.anyio +async def test_issuer_is_stamped_when_same_origin_fallback_register_is_on_the_discovered_issuer( + oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage +): + """SEP-2352: a fallback registration on the discovered issuer's own host is still bound. + + Legacy same-origin embedded AS: PRM is absent, root ASM discovery succeeds with `issuer` + equal to the resource origin and no `registration_endpoint`. DCR falls back to + `/register` - the issuer's own host - so the binding was established and + is recorded, preserving auto-recovery on a later AS migration. + """ + _prepare_full_flow(oauth_provider, None) + auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + + # PRM discovery 404s on both well-known URLs. + prm_req = await auth_flow.asend(httpx.Response(401, request=request)) + assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp" + prm_req = await auth_flow.asend(httpx.Response(404, request=prm_req)) + assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource" + + # Root ASM discovery succeeds with the resource origin as issuer and no registration_endpoint. + asm_req = await auth_flow.asend(httpx.Response(404, request=prm_req)) + assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server" + + # DCR falls back to the resource origin's /register - the issuer's own host. + dcr_req = await auth_flow.asend(httpx.Response(200, content=_asm("https://api.example.com"), request=asm_req)) + assert dcr_req.method == "POST" + assert str(dcr_req.url) == "https://api.example.com/register" + dcr_response = httpx.Response( + 201, + json={"client_id": "embedded-client", "redirect_uris": ["http://localhost:3030/callback"]}, + request=dcr_req, + ) + await auth_flow.asend(dcr_response) + + stored = await mock_storage.get_client_info() + assert stored is not None + assert stored.client_id == "embedded-client" + assert stored.issuer == "https://api.example.com/" + await auth_flow.aclose()