OAuth client: harden SEP-2352/SEP-2350 edge cases; fix conformance comment

.github/actions/conformance/client.py

@@ -7,8 +7,9 @@
     - MCP_CONFORMANCE_SCENARIO env var -> scenario name
     - MCP_CONFORMANCE_CONTEXT env var -> optional JSON (for client-credentials scenarios)
     - MCP_CONFORMANCE_PROTOCOL_VERSION env var -> spec version the harness mock
-      server is speaking (e.g. "2025-11-25", "2026-07-28"). Always set; defaults
-      to the harness's LATEST_SPEC_VERSION when --spec-version is omitted.
+      server is speaking (e.g. "2025-11-25", "2026-07-28"). Always set; when
+      --spec-version is omitted the harness picks per-scenario (LATEST_SPEC_VERSION
+      for active scenarios, DRAFT_PROTOCOL_VERSION for draft-only ones).
     - Server URL as last CLI argument (sys.argv[1])
     - Must exit 0 within 30 seconds
 
@@ -54,10 +55,11 @@
 logger = logging.getLogger(__name__)
 
 #: Spec version the harness is running this scenario at (e.g. "2025-11-25",
-#: "2026-07-28"). The harness always sets this (it falls back to its own
-#: LATEST_SPEC_VERSION when --spec-version is omitted), so None means we were
-#: invoked outside the harness. Handlers that need to take the stateless 2026
-#: path will branch on this once the SDK has one; today it is logged only.
+#: "2026-07-28"). The harness always sets this (when --spec-version is omitted
+#: it picks per-scenario: LATEST_SPEC_VERSION for active scenarios,
+#: DRAFT_PROTOCOL_VERSION for draft-only ones), so None means we were invoked
+#: outside the harness. Handlers that need to take the stateless 2026 path will
+#: branch on this once the SDK has one; today it is logged only.
 PROTOCOL_VERSION: str | None = os.environ.get("MCP_CONFORMANCE_PROTOCOL_VERSION")
 
 # Type for async scenario handler functions

src/mcp/client/auth/oauth2.py

@@ -424,6 +424,13 @@
         # Parse and validate response with scope validation
         token_response = await handle_token_response_scopes(response)
 
+        # RFC 6749 §5.1: an omitted scope means the granted scope equals the requested
+        # scope. Record it explicitly so the persisted token is self-describing — the
+        # SEP-2350 step-up union reads it after a restart, when client_metadata.scope
+        # has reverted to its constructor value.
+        if token_response.scope is None:
+            token_response.scope = self.context.client_metadata.scope
+
         # Store tokens in context
         self.context.current_tokens = token_response
         self.context.update_token_expiry(token_response)
@@ -470,6 +477,12 @@
             content = await response.aread()
             token_response = OAuthToken.model_validate_json(content)
 
+            # RFC 6749 §6: an omitted scope on refresh means the scope is unchanged from
+            # the prior access token. Carry it forward so the persisted token stays
+            # self-describing for the SEP-2350 step-up union after a restart.
+            if token_response.scope is None and self.context.current_tokens is not None:
+                token_response.scope = self.context.current_tokens.scope
+
             self.context.current_tokens = token_response
             self.context.update_token_expiry(token_response)
             await self.context.storage.set_tokens(token_response)
@@ -575,12 +588,15 @@
                             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
                         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)
@@ -600,6 +616,23 @@
                         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 3: Apply scope selection strategy
                     self.context.client_metadata.scope = get_client_metadata_scopes(
                         extract_scope_from_www_auth(response),

tests/client/test_auth.py

@@ -51,7 +51,7 @@ def __init__(self):
         self._client_info: OAuthClientInformationFull | None = None
 
     async def get_tokens(self) -> OAuthToken | None:
-        return self._tokens  # pragma: no cover
+        return self._tokens
 
     async def set_tokens(self, tokens: OAuthToken) -> None:
         self._tokens = tokens
@@ -2833,3 +2833,103 @@ def test_credentials_match_issuer_url_shaped_dcr_id_is_not_portable():
         issuer="https://as.example.com",
     )
     assert credentials_match_issuer(info, "https://other", "https://client.example/metadata.json") is False
+
+
+@pytest.mark.anyio
+async def test_handle_token_response_backfills_omitted_scope_from_request(
+    oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
+):
+    """RFC 6749 §5.1: an omitted token-response scope means granted == requested.
+
+    The token is stored with the requested scope filled in so it remains self-describing
+    after a restart, when the SEP-2350 step-up union reads it but ``client_metadata.scope``
+    has reverted to its constructor value.
+    """
+    oauth_provider.context.client_metadata.scope = "read admin"
+    response = httpx.Response(
+        200,
+        json={"access_token": "t", "token_type": "Bearer", "expires_in": 3600},
+        request=httpx.Request("POST", "https://auth.example.com/token"),
+    )
+    await oauth_provider._handle_token_response(response)
+
+    assert oauth_provider.context.current_tokens is not None
+    assert oauth_provider.context.current_tokens.scope == "read admin"
+    stored = await mock_storage.get_tokens()
+    assert stored is not None
+    assert stored.scope == "read admin"
+
+
+@pytest.mark.anyio
+async def test_handle_refresh_response_carries_prior_scope_when_response_omits_it(
+    oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
+):
+    """RFC 6749 §6: an omitted refresh-response scope means scope is unchanged from the prior token."""
+    oauth_provider.context.current_tokens = OAuthToken(access_token="old", scope="read write")
+    response = httpx.Response(
+        200,
+        json={"access_token": "new", "token_type": "Bearer", "expires_in": 3600},
+        request=httpx.Request("POST", "https://auth.example.com/token"),
+    )
+    ok = await oauth_provider._handle_refresh_response(response)
+
+    assert ok is True
+    assert oauth_provider.context.current_tokens is not None
+    assert oauth_provider.context.current_tokens.access_token == "new"
+    assert oauth_provider.context.current_tokens.scope == "read write"
+    stored = await mock_storage.get_tokens()
+    assert stored is not None
+    assert stored.scope == "read write"
+
+
+@pytest.mark.anyio
+async def test_issuer_binding_re_evaluated_after_asm_when_prm_discovery_failed(
+    oauth_provider: OAuthClientProvider,
+):
+    """SEP-2352: on the legacy no-PRM path the binding check uses the ASM-discovered issuer.
+
+    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.
+    """
+    oauth_provider.context.current_tokens = None
+    oauth_provider.context.token_expiry_time = None
+    oauth_provider._initialized = True
+    oauth_provider.context.client_info = OAuthClientInformationFull(
+        client_id="stale-client",
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        issuer="https://old-as.example.com",
+    )
+
+    auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+    response_401 = httpx.Response(401, request=request)
+
+    # PRM discovery: path-based then root, both 404.
+    prm_req = await auth_flow.asend(response_401)
+    assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
+    prm_req = await auth_flow.asend(httpx.Response(404, request=prm_req))
+    assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource"
+
+    # ASM discovery via root fallback (no auth_server_url) succeeds with a different issuer.
+    asm_req = await auth_flow.asend(httpx.Response(404, request=prm_req))
+    assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server"
+    asm_response = httpx.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,
+    )
+
+    # The stale bound credentials are discarded, so the next yield is a DCR request
+    # rather than the authorize redirect.
+    next_req = await auth_flow.asend(asm_response)
+    assert oauth_provider.context.auth_server_url is None
+    assert next_req.method == "POST"
+    assert str(next_req.url) == "https://api.example.com/register"
+    await auth_flow.aclose()