Address review: refresh-failure parity, CIMD token binding, pinning tests

- A failed refresh resets _initialized inside _handle_refresh_response, so
  the 401-path refresh site behaves like the pre-request one: the next
  request re-reads storage and can retry a refresh token that is still good
  instead of leaving a long-lived headless provider with no tokens.
- A CIMD client_id is portable across authorization servers but its tokens
  are not: when the discovered issuer differs from the record's stamp, keep
  the record, drop the tokens, and re-stamp before deciding to refresh.
- Tests pinning: rejected 401-path refresh falls back to authorization; a
  headless provider retries the refresh on its next connection; a fresh
  registration does not present tokens left from a previous client; the
  CIMD issuer-change case. Manifest wording for the discovered-endpoint
  entry no longer overclaims about the no-metadata fallback.

Author: maxisbey

src/mcp/client/auth/oauth2.py

@@ -374,7 +374,7 @@ async def _perform_authorization_code_grant(self) -> tuple[str, str]:
         if self.context.client_metadata.redirect_uris is None:
             raise OAuthFlowError("No redirect URIs provided for authorization code grant")  # pragma: no cover
         if not self.context.redirect_handler:
-            raise OAuthFlowError("No redirect handler provided for authorization code grant")  # pragma: no cover
+            raise OAuthFlowError("No redirect handler provided for authorization code grant")
         if not self.context.callback_handler:
             raise OAuthFlowError("No callback handler provided for authorization code grant")  # pragma: no cover
 
@@ -521,6 +521,8 @@ async def _handle_refresh_response(self, response: httpx2.Response) -> bool:
         if response.status_code != 200:
             logger.warning(f"Token refresh failed: {response.status_code}")
             self.context.clear_tokens()
+            # Re-read storage on the next request: the failure may have been transient.
+            self._initialized = False
             return False
 
         try:
@@ -545,6 +547,7 @@ async def _handle_refresh_response(self, response: httpx2.Response) -> bool:
         except ValidationError:  # pragma: no cover
             logger.exception("Invalid refresh response")
             self.context.clear_tokens()
+            self._initialized = False
             return False
 
     async def _initialize(self) -> None:
@@ -593,12 +596,8 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                 and self.context.can_refresh_token()
                 and self.context.oauth_metadata is not None
             ):
-                refresh_request = await self._refresh_token()
-                refresh_response = yield refresh_request
-
-                if not await self._handle_refresh_response(refresh_response):
-                    # Refresh failed, need full re-authentication
-                    self._initialized = False
+                refresh_response = yield await self._refresh_token()
+                await self._handle_refresh_response(refresh_response)
 
             if self.context.is_token_valid():
                 self._add_auth_header(request)
@@ -749,6 +748,21 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                             # Held tokens belong to a previous client and cannot be refreshed by this one.
                             self.context.clear_tokens()
 
+                    # A CIMD client_id is portable across authorization servers (SEP-2352) but tokens
+                    # issued under it are not: on an issuer change keep the record, drop the tokens.
+                    client_info = self.context.client_info
+                    current_issuer = self.context.auth_server_url or (
+                        str(self.context.oauth_metadata.issuer) if self.context.oauth_metadata else None
+                    )
+                    if (
+                        client_info.client_id == self.context.client_metadata_url
+                        and current_issuer is not None
+                        and client_info.issuer not in (None, current_issuer)
+                    ):
+                        self.context.clear_tokens()
+                        client_info.issuer = current_issuer
+                        await self.context.storage.set_client_info(client_info)
+
                     # Step 5: Refresh with the stored refresh token first (RFC 6749 §6); run the full
                     # authorization only when there is none or the server rejects it.
                     refreshed = False

tests/interaction/_requirements.py

@@ -3897,11 +3897,14 @@ def __post_init__(self) -> None:
         source="issue:#3240",
         behavior=(
             "A refresh in a process that has not yet discovered the authorization server happens only after "
-            "protected-resource and authorization-server metadata discovery and posts to the advertised "
-            "token endpoint, never to a path guessed from the server origin."
+            "protected-resource and authorization-server metadata discovery and posts to the token endpoint "
+            "that metadata advertises."
         ),
         transports=("streamable-http",),
-        note="OAuth is HTTP-only.",
+        note=(
+            "OAuth is HTTP-only. When discovery yields no AS metadata at all, the 2025-03-26 origin-derived "
+            "fallback endpoint is still used, as it is for the authorization itself."
+        ),
     ),
     "client-auth:resource-parameter": Requirement(
         source=f"{SPEC_BASE_URL}/basic/authorization#resource-parameter-implementation",

tests/interaction/auth/test_lifecycle.py

@@ -18,15 +18,16 @@
 from pydantic import AnyHttpUrl, AnyUrl
 
 from mcp import MCPError
-from mcp.client.auth import OAuthClientProvider
+from mcp.client.auth import OAuthClientProvider, OAuthFlowError
 from mcp.client.auth.extensions.client_credentials import ClientCredentialsOAuthProvider, PrivateKeyJWTOAuthProvider
 from mcp.server import Server, ServerRequestContext
-from mcp.shared.auth import OAuthClientInformationFull, OAuthMetadata
+from mcp.shared.auth import OAuthClientInformationFull, OAuthMetadata, OAuthToken
 from tests.interaction._connect import BASE_URL
 from tests.interaction._requirements import requirement
 from tests.interaction.auth._harness import (
     REDIRECT_URI,
     AppShim,
+    HeadlessOAuth,
     InMemoryTokenStorage,
     RecordedRequest,
     auth_settings,
@@ -125,6 +126,17 @@ def restarted_headless_provider(storage: InMemoryTokenStorage) -> OAuthClientPro
     return OAuthClientProvider(server_url=f"{BASE_URL}/mcp", client_metadata=oauth_client_metadata(), storage=storage)
 
 
+def restarted_interactive_provider(storage: InMemoryTokenStorage, headless: HeadlessOAuth) -> OAuthClientProvider:
+    """A provider as a second process with a browser available constructs it: same storage, fresh state."""
+    return OAuthClientProvider(
+        server_url=f"{BASE_URL}/mcp",
+        client_metadata=oauth_client_metadata(),
+        storage=storage,
+        redirect_handler=headless.redirect_handler,
+        callback_handler=headless.callback_handler,
+    )
+
+
 @requirement("client-auth:refresh:transparent")
 async def test_an_expired_access_token_is_transparently_refreshed_before_the_next_request() -> None:
     """An access token the client considers expired is refreshed and the new bearer is used.
@@ -466,6 +478,154 @@ async def test_a_restarted_client_refreshes_at_the_token_endpoint_advertised_und
     assert not any(r.path.endswith("/authorize") for r in recorded)
 
 
+@requirement("client-auth:invalid-grant-clears-tokens")
+async def test_a_refresh_the_server_rejects_on_the_401_path_falls_back_to_authorization() -> None:
+    """When the 401 path's refresh is rejected, the flow runs the full authorization instead of giving up.
+
+    Second process with a browser; the harness denies the one refresh with `invalid_grant`. The
+    recording proves the refresh was tried first, then exactly one authorize and code exchange,
+    with no re-registration.
+    """
+    provider = InMemoryAuthorizationServerProvider(fail_next_refresh=True)
+    storage = InMemoryTokenStorage()
+    with anyio.fail_after(5):
+        provider.expire_access_token(await first_process_login(provider, storage))
+
+    recorded, on_request = record_requests()
+    headless = HeadlessOAuth()
+    server = Server("guarded", on_list_tools=list_tools)
+    with anyio.fail_after(5):
+        async with connect_with_oauth(
+            server,
+            provider=provider,
+            auth=restarted_interactive_provider(storage, headless),
+            headless=headless,
+            on_request=on_request,
+        ) as (client, _):
+            result = await client.list_tools()
+
+    assert result.tools[0].name == "echo"
+    assert [form_body(r)["grant_type"] for r in find(recorded, "POST", "/token")] == snapshot(
+        ["refresh_token", "authorization_code"]
+    )
+    counts = path_counts(recorded)
+    assert counts[("GET", "/authorize")] == 1
+    assert counts[("POST", "/register")] == 0
+
+
+@requirement("client-auth:refresh:on-401")
+async def test_a_headless_client_whose_refresh_failed_retries_it_from_storage_on_the_next_connection() -> None:
+    """A failed refresh does not wedge a long-lived headless provider: the next connection reloads and refreshes.
+
+    The harness denies the first refresh with `invalid_grant` but leaves the refresh token valid,
+    standing in for a transient token-endpoint failure. The first connect raises (no browser to
+    fall back to); the second, through the same provider instance, refreshes and succeeds.
+    """
+    provider = InMemoryAuthorizationServerProvider(fail_next_refresh=True)
+    storage = InMemoryTokenStorage()
+    with anyio.fail_after(5):
+        provider.expire_access_token(await first_process_login(provider, storage))
+    daemon = restarted_headless_provider(storage)
+
+    with anyio.fail_after(5):
+        with pytest.RaisesGroup(pytest.RaisesExc(OAuthFlowError), flatten_subgroups=True):
+            await connect_with_oauth(
+                Server("guarded", on_list_tools=list_tools), provider=provider, auth=daemon
+            ).__aenter__()
+
+    recorded, on_request = record_requests()
+    with anyio.fail_after(5):
+        async with connect_with_oauth(
+            Server("guarded", on_list_tools=list_tools), provider=provider, auth=daemon, on_request=on_request
+        ) as (client, _):
+            result = await client.list_tools()
+
+    assert result.tools[0].name == "echo"
+    assert [form_body(r)["grant_type"] for r in find(recorded, "POST", "/token")] == ["refresh_token"]
+    assert find(recorded, "GET", "/authorize") == []
+
+
+@requirement("client-auth:refresh:on-401")
+async def test_a_fresh_registration_does_not_present_tokens_left_from_a_previous_client() -> None:
+    """Tokens found in storage without their registration are not refreshed under a newly registered client.
+
+    The second process finds tokens but no `client_info` (lost, or never persisted), so the 401
+    path registers a new client; the refresh token belonged to the old one and is dropped rather
+    than presented, and the flow authorizes once. RFC 6749 §6 binds refresh tokens to their client.
+    """
+    provider = InMemoryAuthorizationServerProvider()
+    storage = InMemoryTokenStorage()
+    with anyio.fail_after(5):
+        provider.expire_access_token(await first_process_login(provider, storage))
+    storage.client_info = None
+
+    recorded, on_request = record_requests()
+    headless = HeadlessOAuth()
+    server = Server("guarded", on_list_tools=list_tools)
+    with anyio.fail_after(5):
+        async with connect_with_oauth(
+            server,
+            provider=provider,
+            auth=restarted_interactive_provider(storage, headless),
+            headless=headless,
+            on_request=on_request,
+        ) as (client, _):
+            result = await client.list_tools()
+
+    assert result.tools[0].name == "echo"
+    assert [(r.method, r.path) for r in recorded[:6]] == snapshot(
+        [
+            ("POST", "/mcp"),
+            ("GET", "/.well-known/oauth-protected-resource/mcp"),
+            ("GET", "/.well-known/oauth-authorization-server"),
+            ("POST", "/register"),
+            ("GET", "/authorize"),
+            ("POST", "/token"),
+        ]
+    )
+    assert [form_body(r)["grant_type"] for r in find(recorded, "POST", "/token")] == ["authorization_code"]
+
+
+@requirement("client-auth:as-binding")
+async def test_a_cimd_client_keeps_its_id_but_drops_its_tokens_when_the_authorization_server_changes() -> None:
+    """A CIMD registration is portable across authorization servers; the tokens issued under it are not.
+
+    Storage holds a CIMD record stamped with a previous issuer and a refresh token minted there.
+    On the 401 the discovered issuer differs, so the flow keeps the URL client_id, re-stamps it,
+    and authorizes afresh instead of presenting the old refresh token to the new server (SEP-2352;
+    the TypeScript SDK discards tokens on the same mismatch).
+    """
+    recorded, on_request = record_requests()
+    provider = InMemoryAuthorizationServerProvider()
+    seeded_client(provider, client_id=CIMD_URL)
+    stale = OAuthClientInformationFull(
+        client_id=CIMD_URL,
+        token_endpoint_auth_method="none",
+        redirect_uris=[AnyUrl(REDIRECT_URI)],
+        issuer="https://old-as.example.com",
+    )
+    storage = InMemoryTokenStorage(client_info=stale)
+    storage.tokens = OAuthToken(access_token="issued-by-old-as", refresh_token="refresh-from-old-as", expires_in=3600)
+    server = Server("guarded", on_list_tools=list_tools)
+
+    with anyio.fail_after(5):
+        async with connect_with_oauth(
+            server,
+            provider=provider,
+            storage=storage,
+            client_metadata_url=CIMD_URL,
+            app_shim=shim(serve={ASM_PATH: cimd_supported_metadata()}),
+            on_request=on_request,
+        ) as (client, _):
+            await client.list_tools()
+
+    assert [form_body(r)["grant_type"] for r in find(recorded, "POST", "/token")] == ["authorization_code"]
+    assert all(b"refresh-from-old-as" not in r.content for r in recorded)
+    assert path_counts(recorded)[("POST", "/register")] == 0
+    assert storage.client_info is not None
+    assert (storage.client_info.client_id, storage.client_info.issuer) == (CIMD_URL, f"{BASE_URL}/")
+
+
 @requirement("client-auth:client-credentials")
 async def test_client_credentials_provider_obtains_a_token_without_an_authorize_step() -> None:
     """The client-credentials provider connects with no authorize step and a `client_credentials` grant.