Harden SEP-2352 issuer binding (Codex review)

- Stamp the bound issuer from the discovered oauth_metadata.issuer when PRM did
  not advertise an authorization server (legacy no-PRM path), instead of leaving
  it None — otherwise migrated resources could reuse the old DCR client_id.
- Detect CIMD portability by the client_id equaling the configured
  client_metadata_url, not by URL shape, so a registration server that issues a
  URL-shaped client_id is still treated as bound to its issuer.

Author: Kludex

src/mcp/client/auth/oauth2.py

@@ -571,7 +571,9 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
                     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)
+                        and not credentials_match_issuer(
+                            self.context.client_info, self.context.auth_server_url, self.context.client_metadata_url
+                        )
                     ):
                         logger.debug("Authorization server changed; discarding bound credentials and re-registering")
                         self.context.client_info = None
@@ -608,6 +610,13 @@ 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: bind the credentials to the issuing AS. Prefer the PRM-advertised
+                        # authorization server; on the legacy no-PRM path fall back to the issuer from
+                        # the discovered metadata so the binding is still recorded.
+                        bound_issuer = self.context.auth_server_url
+                        if bound_issuer is None and self.context.oauth_metadata is not None:
+                            bound_issuer = str(self.context.oauth_metadata.issuer)
+
                         if should_use_client_metadata_url(
                             self.context.oauth_metadata, self.context.client_metadata_url
                         ):
@@ -617,8 +626,7 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
                                 self.context.client_metadata_url,  # type: ignore[arg-type]
                                 redirect_uris=self.context.client_metadata.redirect_uris,
                             )
-                            # SEP-2352: bind the credentials to the issuing authorization server
-                            client_information.issuer = self.context.auth_server_url
+                            client_information.issuer = bound_issuer
                             self.context.client_info = client_information
                             await self.context.storage.set_client_info(client_information)
                         else:
@@ -630,8 +638,7 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
                             )
                             registration_response = yield registration_request
                             client_information = await handle_registration_response(registration_response)
-                            # SEP-2352: bind the credentials to the issuing authorization server
-                            client_information.issuer = self.context.auth_server_url
+                            client_information.issuer = bound_issuer
                             self.context.client_info = client_information
                             await self.context.storage.set_client_info(client_information)
 

src/mcp/client/auth/utils.py

@@ -325,16 +325,20 @@ def is_valid_client_metadata_url(url: str | None) -> bool:
         return False
 
 
-def credentials_match_issuer(client_info: OAuthClientInformationFull, issuer: str) -> bool:
+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).
 
-    URL-based client IDs (CIMD) are portable across authorization servers — the same self-hosted
-    document is resolved by whichever server is in use — so they always match. Credentials with a
-    recorded issuer match only when it equals `issuer` (simple string comparison). Credentials
-    with no recorded issuer (pre-registered, or stored before issuer binding existed) carry no
-    binding to enforce and are left as-is.
+    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 equals `issuer` (simple string comparison). Credentials with no recorded
+    issuer (pre-registered, or stored before issuer binding existed) carry no binding to enforce
+    and are left as-is.
     """
-    if is_valid_client_metadata_url(client_info.client_id):
+    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

tests/client/test_auth.py

@@ -2798,26 +2798,37 @@ def test_union_scopes(previous: str | None, new: str | None, expected: str | Non
 
 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") is True
+    assert credentials_match_issuer(info, "https://as", None) is True
 
 
 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") is False
+    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") is True
+    assert credentials_match_issuer(info, "https://as", None) is True
 
 
 def test_credentials_match_issuer_cimd_is_portable():
-    """A URL-based client_id (CIMD) is portable across authorization servers."""
+    """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="https://client.example/metadata.json",
+        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") is True
+    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