Fold the SEP-2352 issuer-binding rules into one method used at both discovery points

The bound-credential discard and the CIMD keep-but-rebind case were spelled
out inline where PRM names the issuer, and only the former where the issuer
is first learned from AS metadata on the legacy no-PRM path. One method now
applies both rules wherever the issuer becomes known, so a CIMD record's
carried-over tokens are dropped on that path too. A unit test covers it.

Author: maxisbey

src/mcp/client/auth/oauth2.py

@@ -550,6 +550,29 @@ async def _handle_refresh_response(self, response: httpx2.Response) -> bool:
             self._initialized = False
             return False
 
+    async def _apply_issuer_binding(self, issuer: str) -> bool:
+        """Apply SEP-2352 to the held registration now that the authorization server's issuer is known.
+
+        Credentials bound to another issuer are discarded with their tokens so the flow re-registers.
+        A CIMD record is portable, so it is kept and re-stamped, but tokens it carried over from
+        another issuer (or of unknown origin, when the record is unstamped) are dropped. Returns
+        True when the held state was for a different issuer.
+        """
+        client_info = self.context.client_info
+        if client_info is None:
+            return False
+        if not credentials_match_issuer(client_info, 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()
+            return True
+        if client_info.client_id == self.context.client_metadata_url and client_info.issuer != issuer:
+            self.context.clear_tokens()
+            client_info.issuer = issuer
+            await self.context.storage.set_client_info(client_info)
+            return True
+        return False
+
     async def _initialize(self) -> None:
         """Load stored tokens and client info."""
         self.context.current_tokens = await self.context.storage.get_tokens()
@@ -636,35 +659,13 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                         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 and tokens belong to the issuer they came from.
+                    if self.context.auth_server_url is not None and await self._apply_issuer_binding(
+                        self.context.auth_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
-                    elif (
-                        self.context.client_info is not None
-                        and self.context.client_info.client_id == self.context.client_metadata_url
-                        and self.context.auth_server_url is not None
-                        and self.context.client_info.issuer != self.context.auth_server_url
-                    ):
-                        # A CIMD client_id is portable across authorization servers; the tokens issued
-                        # under it and the cached metadata are not. Keep the record, re-stamped. An
-                        # unstamped record's tokens have unknown provenance and are dropped the same way.
-                        self.context.clear_tokens()
+                        # rediscovery cannot leak the old endpoints into Steps 4-5.
                         self.context.oauth_metadata = None
-                        self.context.client_info.issuer = self.context.auth_server_url
-                        await self.context.storage.set_client_info(self.context.client_info)
 
                     asm_discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(
                         self.context.auth_server_url, self.context.server_url
@@ -688,21 +689,9 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                             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()
+                    # discovery (mirroring the bound_issuer fallback in Step 4).
+                    if self.context.auth_server_url is None and self.context.oauth_metadata is not None:
+                        await self._apply_issuer_binding(str(self.context.oauth_metadata.issuer))
 
                     # Step 3: Apply scope selection strategy
                     self.context.client_metadata.scope = get_client_metadata_scopes(

tests/client/test_auth.py

@@ -4,7 +4,7 @@
 import json
 import time
 from unittest import mock
-from urllib.parse import parse_qs, quote, unquote, urlparse
+from urllib.parse import parse_qs, parse_qsl, quote, unquote, urlparse
 
 import httpx2
 import pytest
@@ -3336,3 +3336,56 @@ async def test_cimd_record_is_restamped_and_its_tokens_and_cached_metadata_dropp
     )
     assert mock_storage._client_info is provider.context.client_info
     await auth_flow.aclose()
+
+
+@pytest.mark.anyio
+async def test_cimd_record_is_restamped_and_its_tokens_dropped_when_only_asm_reveals_a_new_issuer(
+    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage, valid_tokens: OAuthToken
+) -> None:
+    """The CIMD rebinding also applies on the legacy no-PRM path, where the issuer is learned from AS metadata.
+
+    PRM discovery 404s, so the issuer only becomes known from the root well-known metadata; it
+    differs from the record's stamp, so the tokens are dropped and the record re-stamped before
+    any refresh could be attempted, and the flow proceeds to authorize rather than refresh.
+    """
+    cimd_url = "https://client.example.com/.well-known/mcp-client"
+    provider = OAuthClientProvider(
+        server_url="https://api.example.com/v1/mcp",
+        client_metadata=client_metadata,
+        storage=mock_storage,
+        client_metadata_url=cimd_url,
+    )
+    provider.context.client_info = OAuthClientInformationFull(
+        client_id=cimd_url, token_endpoint_auth_method="none", issuer="https://old-as.example.com"
+    )
+    provider.context.current_tokens = valid_tokens
+    provider.context.token_expiry_time = time.time() + 1800
+    provider._initialized = True
+    provider._perform_authorization_code_grant = mock.AsyncMock(return_value=("auth-code", "verifier"))
+
+    auth_flow = provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/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))
+    assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server"
+    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'"client_id_metadata_document_supported": true}'
+        ),
+        request=asm_req,
+    )
+    next_req = await auth_flow.asend(asm_response)
+
+    assert dict(parse_qsl(next_req.content.decode()))["grant_type"] == "authorization_code"
+    assert provider.context.client_info is not None
+    assert (provider.context.client_info.client_id, provider.context.client_info.issuer) == (
+        cimd_url,
+        "https://api.example.com",
+    )
+    assert mock_storage._client_info is provider.context.client_info
+    await auth_flow.aclose()