From 5717fa285eb4e0c38c210c32beb9f3a42ff4807f Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Wed, 2 Sep 2026 16:01:56 +0000 Subject: [PATCH 1/2] [v1.x] Derive the expected issuer before fetching authorization server metadata Backport of the discovery half of #3398, together with the SEP-2352 credential binding that main has had since 2.0 and that change relies on (the parts of #2921, #2933, #2936, #2946 and #3181 it needs). The discovery step now knows which issuer the authorization server metadata must carry before fetching it: the PRM-advertised server, or on the 2025-03-26 no-PRM fallback the resource server's origin, which is what that well-known URL is built from (RFC 8414 section 3.3). On this line there was no issuer check at all; it now runs on both paths (`validate_metadata_issuer`, with `issuers_match`: exact, except that a root issuer with and without its trailing slash is the same server, which matters here because every parsed issuer renders with the slash). Stored client credentials gain the binding main has: the SDK records the issuer a registration was made with (`OAuthClientInformationFull.issuer`, never read from the registration response), and before any metadata is fetched a registration bound to a different issuer is discarded together with its tokens so the flow re-registers instead of presenting another server's credentials (`credentials_match_issuer`); newly registered clients are bound to the expected issuer when metadata for it was found and the registration actually went to that server. When no protected resource metadata location answers and one of them failed with a server error (or 429), the flow stops rather than reading that as "no such metadata". A 403 insufficient_scope step-up takes the same path as a 401: it discovers first when no metadata is held (extract_resource_metadata_from_www_auth also reads the `resource_metadata` hint from a 403 challenge) and then re-authorizes with the challenged scope; metadata already discovered in this process is reused as before. A 403 that is not a scope challenge is handed back to the caller instead of being retried unchanged. The conformance harness pinned on this line (0.1.13) still serves, in its two path-based authorization server scenarios, metadata whose `issuer` omits the advertised tenant path; those two scenarios are listed as expected failures until the pin moves past conformance 0.1.15, where the mock was corrected. Differences from the main change: - RFC 9207 `iss` validation is not included; the callback contract on 1.x stays `tuple[str, str | None]`. - 1.x has no SEP-2350 scope union, so a step-up asks for the challenged scope as it did before. - The root-slash allowance lives in `validate_metadata_issuer` itself (new on this line) rather than only on the legacy path. --- .../actions/conformance/expected-failures.yml | 9 +- src/mcp/client/auth/oauth2.py | 196 +++-- src/mcp/client/auth/utils.py | 72 +- src/mcp/shared/auth.py | 3 + tests/client/test_auth.py | 772 +++++++++++++++++- 5 files changed, 965 insertions(+), 87 deletions(-) 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/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/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() From e5b671994da96010ad41b7925cd81e8da24c2a5d Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Wed, 2 Sep 2026 16:02:05 +0000 Subject: [PATCH 2/2] [v1.x] Let pre-provisioned OAuth clients name their authorization server Backport of the second half of #3398. ClientCredentialsOAuthProvider and PrivateKeyJWTOAuthProvider take an optional issuer keyword: the issuer identifier of the authorization server the fixed client_id (and secret) were issued by. When set, token requests are only built from discovered authorization server metadata whose issuer matches it; if discovery yields metadata for another server, or none at all, the flow stops with OAuthFlowError before the secret is attached or an assertion is minted, and the metadata and tokens held are dropped so the next request starts discovery again. When the resource advertises several authorization servers the one matching the configured issuer is used; the comparison is issuers_match (exact, root slash aside); a value that is not an http(s) URL is a ValueError. Omitting it keeps the current behaviour. Differences from the main change: the keyword follows `scopes` (the 1.x name); there is no docs page for these providers on 1.x, so the docstrings carry the description. --- .../auth/extensions/client_credentials.py | 67 +++++- .../extensions/test_client_credentials.py | 208 ++++++++++++++++++ 2 files changed, 274 insertions(+), 1 deletion(-) 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/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()