From f8586d7037f9715663f7c97442f7990a2b592d7f Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Wed, 26 Aug 2026 15:50:49 +0000 Subject: [PATCH 1/2] Derive the expected issuer before fetching authorization server metadata 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). The metadata issuer check runs on both paths instead of only when PRM was found (on the no-PRM path a root issuer rendered with its trailing slash still names the origin, and the origin is written the way metadata issuers are, so host case or an explicit default port in `server_url` do not matter). The SEP-2352 stored-credential binding is evaluated once against that same value before metadata discovery, so it also applies when no metadata is served and the default endpoints are used, and the special case that re-evaluated it against the served issuer goes away; newly registered clients are bound to it when metadata for it was found. When no protected resource metadata location answers and one of them failed with a server error (or 429), the flow now stops rather than reading that as "no such metadata", so a transient failure cannot send a client with a bound registration down the legacy path. A 403 insufficient_scope step-up takes the same path as a 401. Until now it re-authorized with whatever metadata was in memory, and with none after a restart, in which case it used the 2025-03-26 default endpoints on the resource server's origin regardless of what the server advertises and never consulted the binding. It now discovers first when no metadata is held (extract_resource_metadata_from_www_auth also reads the `resource_metadata` hint from a 403 challenge), then re-authorizes with the SEP-2350 scope union, keeping the granted scope in the union even if discovery dropped the old token; 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, as IdentityAssertionOAuthProvider already does. That root-slash allowance is one helper, issuers_match, which credentials_match_issuer uses as well, so records stamped in either form keep matching on both paths. This brings the no-PRM path in line with the TypeScript client, which applies the section 3.3 check on every discovery path (that client also offers an opt-out; this one does not, as on the PRM path since 2.0). --- docs/client/oauth-clients.md | 2 +- docs/migration.md | 6 +- src/mcp/client/auth/oauth2.py | 231 +++++++++++----------- src/mcp/client/auth/utils.py | 18 +- tests/client/test_auth.py | 347 ++++++++++++++++++++++++++++++++-- 5 files changed, 473 insertions(+), 131 deletions(-) diff --git a/docs/client/oauth-clients.md b/docs/client/oauth-clients.md index cd7de35626..6dea60d32d 100644 --- a/docs/client/oauth-clients.md +++ b/docs/client/oauth-clients.md @@ -76,7 +76,7 @@ Look at `main()`. The provider goes on the **httpx2 client**, the httpx2 client The first time `Client` sends a request, the server answers `401`. The provider takes over: -1. **Discovery.** It reads the `WWW-Authenticate` header, fetches the server's Protected Resource Metadata from `/.well-known/oauth-protected-resource`, learns which authorization server protects this resource, and fetches *that* server's metadata. +1. **Discovery.** It reads the `WWW-Authenticate` header, fetches the server's Protected Resource Metadata from `/.well-known/oauth-protected-resource`, learns which authorization server protects this resource, and fetches *that* server's metadata. (An older server that publishes no resource metadata is asked for authorization server metadata at its own origin instead.) Either way the metadata must name, as its `issuer`, the server it was fetched for; anything else is refused. 2. **Registration.** Nothing in storage? It registers you dynamically with your `OAuthClientMetadata` and stores the result. 3. **Authorization.** It generates the PKCE pair and a `state`, builds the authorization URL, awaits your `redirect_handler`, then awaits your `callback_handler` for the code. 4. **Exchange.** It trades the code for an `OAuthToken`, stores it, and replays your original request with `Authorization: Bearer ...`. diff --git a/docs/migration.md b/docs/migration.md index 7927c60611..e49cb3b8d3 100644 --- a/docs/migration.md +++ b/docs/migration.md @@ -2499,7 +2499,9 @@ metadata's `issuer` exactly matches the authorization server URL advertised in t resource metadata, as required by [RFC 8414](https://datatracker.ietf.org/doc/html/rfc8414) section 3.3 ([SEP-2468](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2468)). The comparison is a simple string comparison ([RFC 3986](https://datatracker.ietf.org/doc/html/rfc3986) -section 6.2.1), so even a trailing-slash disagreement counts as a mismatch. v1 accepted the +section 6.2.1), so even a trailing-slash disagreement counts as a mismatch. (For an older server +that publishes no protected resource metadata the expected value is the MCP server's own origin, +and there a root issuer with a trailing slash is accepted too.) v1 accepted the metadata without checking, so a server pairing whose two values disagree authenticated fine under v1 and now fails the entire flow. For example, when the MCP server's protected resource metadata advertises @@ -2517,7 +2519,7 @@ OAuthFlowError: Authorization server metadata issuer mismatch: https://as.exampl There is no client-side override. Fix the deployment instead: make the authorization server's `issuer` string-equal the URL in the protected resource metadata's `authorization_servers` -list. See [OAuth metadata URLs no longer gain a trailing slash](#oauth-metadata-urls-no-longer-gain-a-trailing-slash) +list (or the MCP server's origin, without protected resource metadata). See [OAuth metadata URLs no longer gain a trailing slash](#oauth-metadata-urls-no-longer-gain-a-trailing-slash) for how v2 preserves the exact string form of these URLs. ### OAuth client requests `offline_access` and adds `prompt=consent` when the authorization server supports it ([SEP-2207](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2207)) diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py index 7dc62b52b9..78d6f62eca 100644 --- a/src/mcp/client/auth/oauth2.py +++ b/src/mcp/client/auth/oauth2.py @@ -17,7 +17,7 @@ import anyio import httpx2 from mcp_types.version import is_version_at_least -from pydantic import BaseModel, Field, ValidationError +from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, TypeAdapter, ValidationError from mcp.client.auth.exceptions import OAuthFlowError, OAuthRegistrationError, OAuthTokenError from mcp.client.auth.utils import ( @@ -36,6 +36,7 @@ handle_registration_response, handle_token_response_scopes, is_valid_client_metadata_url, + issuers_match, should_use_client_metadata_url, union_scopes, validate_authorization_response_iss, @@ -276,6 +277,16 @@ def prepare_token_auth( return data, headers +_ORIGIN_URL = TypeAdapter(AnyHttpUrl, config=ConfigDict(url_preserve_empty_path=True)) + + +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) so the two compare as strings.""" + parsed = urlparse(server_url) + return str(_ORIGIN_URL.validate_python(f"{parsed.scheme}://{parsed.netloc}")) + + class OAuthClientProvider(httpx2.Auth): """OAuth2 authentication for httpx2. @@ -577,6 +588,16 @@ async def _validate_resource_match(self, prm: ProtectedResourceMetadata) -> None if not check_resource_allowed(requested_resource=default_resource, configured_resource=prm_resource): raise OAuthFlowError(f"Protected resource {prm_resource} does not match expected {default_resource}") + 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: httpx2.Request) -> AsyncGenerator[httpx2.Request, httpx2.Response]: """httpx2 auth flow integration.""" async with self.context.lock: @@ -600,107 +621,122 @@ async def async_auth_flow(self, request: httpx2.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) + # Read before discovery, which may clear the tokens: on a restart the stored + # token's scope is the only record of what was granted (see Step 3). + granted_scope = self.context.current_tokens.scope if self.context.current_tokens else None + + # 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 + ) - # 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 - ) + prm_request_failed: int | None = None + for url in prm_discovery_urls: + discovery_request = create_oauth_metadata_request(url) - for url in prm_discovery_urls: # pragma: no branch - discovery_request = create_oauth_metadata_request(url) + discovery_response = yield discovery_request # sending request - 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 - 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: + 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}" + ) - # 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 + expected_issuer = self._expected_issuer() - self.context.auth_server_url = str(prm.authorization_servers[0]) - break - else: - logger.debug(f"Protected resource metadata discovery failed: {url}") - - # SEP-2352: stored credentials are bound to the issuer that registered them. - # If the authorization server changed, 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 self.context.auth_server_url is not None - and not credentials_match_issuer( - self.context.client_info, self.context.auth_server_url, self.context.client_metadata_url + # 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 ) - ): - 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 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: metadata issuer must match the discovery issuer - if self.context.auth_server_url is not None: - validate_metadata_issuer(asm, self.context.auth_server_url) - self.context.oauth_metadata = asm - break - else: - logger.debug(f"OAuth metadata discovery failed: {url}") - - # SEP-2352: on the legacy no-PRM path the issuer is only known after ASM - # discovery, so re-evaluate the binding here using the discovered metadata - # issuer (mirroring the bound_issuer fallback in Step 4). - if ( - self.context.client_info is not None - and self.context.auth_server_url is None - and self.context.oauth_metadata is not None - and not credentials_match_issuer( - self.context.client_info, - str(self.context.oauth_metadata.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() + # 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 §3.3: the metadata must name the expected issuer. + # On the legacy path a root issuer rendered with its trailing slash + # names the same origin. + if self.context.auth_server_url is None and issuers_match( + str(asm.issuer), expected_issuer + ): + expected_issuer = str(asm.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( + challenged_scope = get_client_metadata_scopes( extract_scope_from_www_auth(response), self.context.protected_resource_metadata, self.context.oauth_metadata, self.context.client_metadata.grant_types, ) + if step_up: + # SEP-2350: union previously requested scopes with the newly challenged ones so + # escalating one operation keeps the others' grants, folding in the granted + # scope read above since client_metadata.scope is not reloaded on a restart. + prior_scope = union_scopes(self.context.client_metadata.scope, granted_scope) + self.context.client_metadata.scope = union_scopes(prior_scope, challenged_scope) + else: + self.context.client_metadata.scope = challenged_scope # 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, when known. - discovered_issuer: str | None = None - if self.context.oauth_metadata is not None: - discovered_issuer = self.context.auth_server_url or str(self.context.oauth_metadata.issuer) + # 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 @@ -752,34 +788,3 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx # 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: Union previously requested scopes with the newly challenged - # scopes (SEP-2350) so escalating one operation keeps the others' grants. - # Fold in the stored token's scope too: on a restart the token is reloaded - # but client_metadata.scope is not, so it would otherwise be the only basis. - challenged_scope = get_client_metadata_scopes( - extract_scope_from_www_auth(response), - self.context.protected_resource_metadata, - self.context.oauth_metadata, - self.context.client_metadata.grant_types, - ) - granted_scope = self.context.current_tokens.scope if self.context.current_tokens else None - prior_scope = union_scopes(self.context.client_metadata.scope, granted_scope) - self.context.client_metadata.scope = union_scopes(prior_scope, challenged_scope) - - # 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 31e2e5cade..30d420a581 100644 --- a/src/mcp/client/auth/utils.py +++ b/src/mcp/client/auth/utils.py @@ -54,7 +54,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") @@ -343,7 +343,8 @@ def credentials_match_issuer( 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 equals `issuer` (simple string comparison). Credentials with no recorded + match only when it equals `issuer` (simple string comparison; a root issuer with and without + its trailing slash count as equal). Credentials with no recorded issuer (pre-registered, or stored before issuer binding existed) carry no binding to enforce and are left as-is. """ @@ -351,7 +352,18 @@ def credentials_match_issuer( return True if client_info.issuer is None: return True - return client_info.issuer == issuer + return issuers_match(client_info.issuer, 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 should_use_client_metadata_url( diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py index be96cc8eec..eb168e6e94 100644 --- a/tests/client/test_auth.py +++ b/tests/client/test_auth.py @@ -1502,8 +1502,11 @@ async def mock_callback() -> AuthorizationCodeResult: 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(httpx2.Response(404, request=prm_request)) + asm_request = await auth_flow.asend(httpx2.Response(404, request=prm_request)) + token_exchange_request = await auth_flow.asend(httpx2.Response(404, request=asm_request)) # Verify scope was updated to the union of prior and challenged scopes (SEP-2350) assert oauth_provider.context.client_metadata.scope == "read write admin:write admin:delete" @@ -1576,7 +1579,10 @@ async def mock_callback() -> AuthorizationCodeResult: headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'}, request=request, ) - token_exchange_request = await auth_flow.asend(response_403) + prm_request = await auth_flow.asend(response_403) + prm_request = await auth_flow.asend(httpx2.Response(404, request=prm_request)) + asm_request = await auth_flow.asend(httpx2.Response(404, request=prm_request)) + token_exchange_request = await auth_flow.asend(httpx2.Response(404, request=asm_request)) assert reauthorize_scope == "read write" @@ -1593,6 +1599,101 @@ async def mock_callback() -> AuthorizationCodeResult: pass +@pytest.mark.anyio +async def test_scope_step_up_discovers_the_authorization_server_before_reauthorizing( + oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage, 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 + the requested scope is the SEP-2350 union. + """ + await mock_storage.set_tokens(valid_tokens) + await mock_storage.set_client_info( + OAuthClientInformationFull( + client_id="registered-client", redirect_uris=[AnyUrl("http://localhost:3030/callback")] + ) + ) + redirects: list[str] = [] + + async def record_redirect(url: str) -> None: + redirects.append(url) + + async def echo_callback() -> AuthorizationCodeResult: + state = parse_qs(urlparse(redirects[-1]).query)["state"][0] + return AuthorizationCodeResult(code="auth_code", state=state) + + oauth_provider.context.redirect_handler = record_redirect + oauth_provider.context.callback_handler = echo_callback + + auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + response_403 = httpx2.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 == "GET" + assert str(prm_request.url) == "https://api.example.com/v1/mcp/resource-metadata" + prm_response = httpx2.Response( + 200, + content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}', + request=prm_request, + ) + asm_request = await auth_flow.asend(prm_response) + assert str(asm_request.url) == "https://auth.example.com/.well-known/oauth-authorization-server" + asm_response = httpx2.Response( + 200, + content=( + b'{"issuer": "https://auth.example.com",' + b' "authorization_endpoint": "https://auth.example.com/authorize",' + b' "token_endpoint": "https://auth.example.com/token"}' + ), + request=asm_request, + ) + + token_request = await auth_flow.asend(asm_response) + assert redirects[-1].startswith("https://auth.example.com/authorize?") + assert parse_qs(urlparse(redirects[-1]).query)["scope"] == ["read write admin"] + assert token_request.method == "POST" + assert str(token_request.url) == "https://auth.example.com/token" + + final_request = await auth_flow.asend( + httpx2.Response( + 200, json={"access_token": "stepped_up", "token_type": "Bearer", "expires_in": 3600}, request=token_request + ) + ) + assert final_request.headers["Authorization"] == "Bearer stepped_up" + with pytest.raises(StopAsyncIteration): + await auth_flow.asend(httpx2.Response(200, request=final_request)) + + +@pytest.mark.anyio +async def test_403_without_a_scope_challenge_is_returned_to_the_caller( + oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage, valid_tokens: OAuthToken +): + """SDK-defined: a 403 that is not an insufficient_scope challenge ends the flow; the request is + not retried.""" + await mock_storage.set_tokens(valid_tokens) + + auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + response_403 = httpx2.Response(403, headers={"WWW-Authenticate": 'Bearer error="access_denied"'}, request=request) + + with pytest.raises(StopAsyncIteration): + await auth_flow.asend(response_403) + + @pytest.mark.parametrize( ( "issuer_url", @@ -1818,8 +1919,8 @@ async def callback_handler() -> AuthorizationCodeResult: 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 = httpx2.Response(500, request=prm_request_1) + # Not served there + prm_response_1 = httpx2.Response(404, request=prm_request_1) # Try path-based fallback prm_request_2 = await auth_flow.asend(prm_response_1) @@ -2891,6 +2992,25 @@ def test_credentials_match_issuer_different_issuer(): assert credentials_match_issuer(info, "https://other", None) is False +@pytest.mark.parametrize( + ("recorded", "current"), + [("https://as.example.com", "https://as.example.com/"), ("https://as.example.com/", "https://as.example.com")], + ids=["current-has-root-slash", "recorded-has-root-slash"], +) +def test_credentials_match_issuer_treats_root_slash_as_the_same_issuer(recorded: str, current: str): + """SDK-defined: a root issuer recorded with and without its trailing slash names the same server.""" + info = OAuthClientInformationFull(client_id="c", redirect_uris=[AnyUrl("http://localhost/cb")], issuer=recorded) + assert credentials_match_issuer(info, current, 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_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")]) @@ -3008,15 +3128,14 @@ async def test_handle_refresh_response_adopts_rotated_refresh_token_when_returne @pytest.mark.anyio -async def test_issuer_binding_re_evaluated_after_asm_when_prm_discovery_failed( +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 ASM-discovered issuer. + """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`` and the post-PRM check is - skipped; when ASM discovery then succeeds via the root well-known fallback, the discovered - metadata's issuer is compared against the stored credentials' bound issuer and a mismatch - triggers re-registration. + PRM discovery fails (404) so ``auth_server_url`` stays ``None``; the expected issuer is then the + origin the legacy well-known URL is built from, so stored credentials bound to another issuer are + discarded before ASM discovery runs, and re-registration follows. """ oauth_provider.context.current_tokens = None oauth_provider.context.token_expiry_time = None @@ -3037,9 +3156,11 @@ async def test_issuer_binding_re_evaluated_after_asm_when_prm_discovery_failed( prm_req = await auth_flow.asend(httpx2.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) succeeds with a different issuer. + # 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(httpx2.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 asm_response = httpx2.Response( 200, content=( @@ -3060,6 +3181,208 @@ async def test_issuer_binding_re_evaluated_after_asm_when_prm_discovery_failed( await auth_flow.aclose() +@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; anything else stops the flow before any + authorization or token request is built.""" + auth_flow = oauth_provider.async_auth_flow(httpx2.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(httpx2.Response(401, request=request)) + prm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req)) + asm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req)) + assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server" + + asm_response = httpx2.Response( + 200, + content=( + b'{"issuer": "https://other-as.example.com", ' + b'"authorization_endpoint": "https://other-as.example.com/authorize", ' + b'"token_endpoint": "https://other-as.example.com/token"}' + ), + 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" + ) + + +@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, + mock_storage: MockTokenStorage, + 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. + """ + await mock_storage.set_tokens(valid_tokens) + await mock_storage.set_client_info( + 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", + ) + ) + redirects: list[str] = [] + + async def record_redirect(url: str) -> None: + redirects.append(url) + + async def echo_callback() -> AuthorizationCodeResult: + state = parse_qs(urlparse(redirects[-1]).query)["state"][0] + return AuthorizationCodeResult(code="auth_code", state=state) + + oauth_provider.context.redirect_handler = record_redirect + oauth_provider.context.callback_handler = echo_callback + + auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp")) + request = await auth_flow.__anext__() + + status, headers = first_response + prm_req = await auth_flow.asend(httpx2.Response(status, headers=headers, request=request)) + prm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req)) + asm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req)) + assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server" + + # No metadata at the origin either: register at the origin's default endpoint. + register_req = await auth_flow.asend(httpx2.Response(404, request=asm_req)) + assert register_req.method == "POST" + assert str(register_req.url) == "https://api.example.com/register" + register_response = httpx2.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 +@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, mock_storage: MockTokenStorage, 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", + ) + await mock_storage.set_client_info(bound) + await mock_storage.set_tokens(valid_tokens) + flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp")) + request = await flow.__anext__() + prm_request = await flow.asend(httpx2.Response(401, request=request)) + root_prm_request = await flow.asend(httpx2.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(httpx2.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 + assert await mock_storage.get_client_info() == bound + + +@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.""" + flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp")) + request = await flow.__anext__() + prm_request = await flow.asend(httpx2.Response(401, request=request)) + root_prm_request = await flow.asend(httpx2.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(httpx2.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 +async def test_legacy_fallback_accepts_a_root_slash_issuer_for_a_server_url_in_any_spelling( + client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage +): + """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 the trailing slash a root issuer is often rendered with, and + the flow proceeds to registration.""" + + async def redirect_handler(url: str) -> None: + raise NotImplementedError + + async def callback_handler() -> AuthorizationCodeResult: + 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(httpx2.Request("GET", "https://API.Example.com:443/v1/mcp")) + request = await auth_flow.__anext__() + prm_req = await auth_flow.asend(httpx2.Response(401, request=request)) + prm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req)) + asm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req)) + asm_response = httpx2.Response( + 200, + content=( + b'{"issuer": "https://api.example.com/", ' + b'"authorization_endpoint": "https://api.example.com/authorize", ' + b'"token_endpoint": "https://api.example.com/token", ' + b'"registration_endpoint": "https://api.example.com/register"}' + ), + request=asm_req, + ) + + register_req = await auth_flow.asend(asm_response) + + assert register_req.method == "POST" + assert str(register_req.url) == "https://api.example.com/register" + await auth_flow.aclose() + + @pytest.mark.anyio @pytest.mark.parametrize( "asm_responses", From 11baaf30b3108be4294677e3c5c1d93562d876bf Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Wed, 26 Aug 2026 16:04:46 +0000 Subject: [PATCH 2/2] Let pre-provisioned OAuth clients name their authorization server 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, the client_credentials exchange and any refresh, 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. This is the same "the authorization server is configuration" model that IdentityAssertionOAuthProvider already uses, made available to the two older machine-to-machine providers without changing their defaults. --- docs/client/oauth-clients.md | 5 +- docs_src/oauth_clients/tutorial002.py | 1 + .../auth/extensions/client_credentials.py | 67 +++++- .../extensions/test_client_credentials.py | 209 +++++++++++++++++- tests/docs_src/test_oauth_clients.py | 9 + 5 files changed, 287 insertions(+), 4 deletions(-) diff --git a/docs/client/oauth-clients.md b/docs/client/oauth-clients.md index 6dea60d32d..3954fd539b 100644 --- a/docs/client/oauth-clients.md +++ b/docs/client/oauth-clients.md @@ -105,13 +105,14 @@ A nightly job, a CI step, another service. There is no browser and nobody to cli `ClientCredentialsOAuthProvider` is the same `httpx2.Auth`, minus the human: -```python title="client.py" hl_lines="4 27-33" +```python title="client.py" hl_lines="4 27-34" --8<-- "docs_src/oauth_clients/tutorial002.py" ``` What changed: * No `OAuthClientMetadata`, no handlers. You pass `client_id` and `client_secret`; the provider builds a minimal `client_credentials` registration around them and skips dynamic registration entirely. +* `issuer` names the authorization server that issued those credentials; use the `issuer` value its `/.well-known/oauth-authorization-server` document returns. Discovery still runs as above, but token requests are only ever built from metadata for *that* issuer; if the MCP server points anywhere else, the flow stops with an `OAuthFlowError` instead. Leave it out and the provider uses whichever authorization server discovery finds. * `scope` is a space-separated string, the OAuth wire format. * Everything downstream is identical: the same `TokenStorage`, the same `httpx2.AsyncClient(auth=...)`, the same `streamable_http_client`. @@ -124,7 +125,7 @@ By default the secret travels as HTTP Basic auth on the token request (`client_s One more provider lives in `mcp.client.auth.extensions.client_credentials`: **`PrivateKeyJWTOAuthProvider`**, for clients that authenticate with a JWT instead of a shared secret (`private_key_jwt`, the key-pair and workload-identity flavour). It follows - the same pattern: construct one, put it on `auth=`. The same module ships + the same pattern: construct one (it takes the same optional `issuer`), put it on `auth=`. The same module ships `SignedJWTParameters` and `static_assertion_provider`, two helpers that build its assertion. There is one more no-human situation: the client belongs to an enterprise whose identity provider, not the user, decides which MCP servers it may reach. That is a different grant with its own trust model and its own page, **[Identity assertion](identity-assertion.md)**. diff --git a/docs_src/oauth_clients/tutorial002.py b/docs_src/oauth_clients/tutorial002.py index dd4105f937..d6323c8291 100644 --- a/docs_src/oauth_clients/tutorial002.py +++ b/docs_src/oauth_clients/tutorial002.py @@ -30,6 +30,7 @@ async def set_client_info(self, client_info: OAuthClientInformationFull) -> None client_id="reporting-agent", client_secret="...", scope="user", + issuer="http://localhost:9000", ) diff --git a/src/mcp/client/auth/extensions/client_credentials.py b/src/mcp/client/auth/extensions/client_credentials.py index 29197bb504..5cdefad1f8 100644 --- a/src/mcp/client/auth/extensions/client_credentials.py +++ b/src/mcp/client/auth/extensions/client_credentials.py @@ -9,6 +9,7 @@ import time from collections.abc import Awaitable, Callable from typing import Any, Literal +from urllib.parse import urlparse from uuid import uuid4 import httpx2 @@ -16,14 +17,50 @@ from pydantic import BaseModel, Field from mcp.client.auth import OAuthClientProvider, OAuthFlowError, 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 @@ -32,6 +69,7 @@ class ClientCredentialsOAuthProvider(OAuthClientProvider): storage=my_token_storage, client_id="my-client-id", client_secret="my-client-secret", + issuer="https://auth.example.com", ) ``` """ @@ -44,6 +82,7 @@ def __init__( client_secret: str, token_endpoint_auth_method: Literal["client_secret_basic", "client_secret_post"] = "client_secret_basic", scope: str | None = None, + issuer: str | None = None, ) -> None: """Initialize client_credentials OAuth provider. @@ -55,6 +94,11 @@ def __init__( token_endpoint_auth_method: Authentication method for token endpoint. Either "client_secret_basic" (default) or "client_secret_post". scope: 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( @@ -64,6 +108,7 @@ def __init__( scope=scope, ) super().__init__(server_url, client_metadata, storage, None, None) + 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, @@ -80,12 +125,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) -> httpx2.Request: """Perform client_credentials authorization.""" return await self._exchange_token_client_credentials() async def _exchange_token_client_credentials(self) -> httpx2.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", } @@ -196,7 +246,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** @@ -256,6 +309,7 @@ def __init__( client_id: str, assertion_provider: Callable[[str], Awaitable[str]], scope: str | None = None, + issuer: str | None = None, ) -> None: """Initialize private_key_jwt OAuth provider. @@ -269,6 +323,11 @@ def __init__( `static_assertion_provider()` for pre-built JWTs, or provide your own callback for workload identity federation. scope: 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( @@ -279,6 +338,7 @@ def __init__( ) super().__init__(server_url, client_metadata, storage, None, None) 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, @@ -294,6 +354,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) -> httpx2.Request: """Perform client_credentials authorization with private_key_jwt.""" return await self._exchange_token_client_credentials() @@ -314,6 +377,8 @@ async def _add_client_authentication_jwt(self, *, token_data: dict[str, Any]) -> async def _exchange_token_client_credentials(self) -> httpx2.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 16336f8002..5933604eeb 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 httpx2 import jwt import pytest +from inline_snapshot import snapshot from pydantic import AnyHttpUrl +from mcp.client.auth import OAuthClientProvider, OAuthFlowError from mcp.client.auth.extensions.client_credentials import ( ClientCredentialsOAuthProvider, PrivateKeyJWTOAuthProvider, @@ -27,7 +31,7 @@ def __init__(self): async def get_tokens(self) -> OAuthToken | None: return self._tokens - async def set_tokens(self, tokens: OAuthToken) -> None: # pragma: no cover + async def set_tokens(self, tokens: OAuthToken) -> None: self._tokens = tokens async def get_client_info(self) -> OAuthClientInformationFull | None: # pragma: no cover @@ -325,3 +329,206 @@ 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[httpx2.Request, httpx2.Response], + *, + authorization_server: str | list[str] | None, + metadata: dict[str, str] | None, +) -> httpx2.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(httpx2.Response(401, request=request)) + while "/.well-known/oauth-protected-resource" in str(request.url): + if authorization_server is None: + response = httpx2.Response(404, request=request) + else: + advertised = authorization_server if isinstance(authorization_server, list) else [authorization_server] + prm = {"resource": _SERVER_URL, "authorization_servers": advertised} + response = httpx2.Response(200, json=prm, request=request) + request = await flow.asend(response) + while "/.well-known/" in str(request.url): + if metadata is None: + response = httpx2.Response(404, request=request) + else: + response = httpx2.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(httpx2.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") + assert audiences == ([] if kind == "secret" else [served_issuer]) + 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(httpx2.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 == _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(httpx2.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(httpx2.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(httpx2.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(httpx2.Response(200, json=token, request=token_request)) + with pytest.raises(StopAsyncIteration): + await flow.asend(httpx2.Response(200, request=retried)) + + flow = provider.async_auth_flow(httpx2.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(httpx2.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/docs_src/test_oauth_clients.py b/tests/docs_src/test_oauth_clients.py index 801fbb2ca1..a4ec05d9fe 100644 --- a/tests/docs_src/test_oauth_clients.py +++ b/tests/docs_src/test_oauth_clients.py @@ -9,6 +9,7 @@ from docs_src.oauth_clients import tutorial001, tutorial002 from mcp.client.auth import OAuthClientProvider, OAuthFlowError, OAuthRegistrationError, OAuthTokenError, TokenStorage from mcp.client.auth.extensions.client_credentials import ( + ClientCredentialsOAuthProvider, PrivateKeyJWTOAuthProvider, static_assertion_provider, ) @@ -80,6 +81,14 @@ async def test_client_credentials_provider_builds_its_own_metadata() -> None: assert metadata.scope == "user" +@pytest.mark.parametrize("provider_class", [ClientCredentialsOAuthProvider, PrivateKeyJWTOAuthProvider]) +async def test_issuer_is_an_optional_keyword_on_both_machine_to_machine_providers(provider_class: type) -> None: + """tutorial002 passes `issuer=`; the page says leaving it out is allowed, on either provider.""" + issuer = inspect.signature(provider_class.__init__).parameters["issuer"] + assert issuer.kind is inspect.Parameter.POSITIONAL_OR_KEYWORD + assert issuer.default is None + + async def test_the_two_remaining_keyword_arguments_have_defaults() -> None: """The page names `client_metadata_url` and `validate_resource_url` as the remainder.""" parameters = inspect.signature(OAuthClientProvider.__init__).parameters