Extend the interaction suite to the 2026-07-28 spec

tests/interaction/auth/_harness.py

@@ -132,25 +132,44 @@ class HeadlessOAuth:
     `redirect_handler` GETs the authorize URL on the bound client (with `auth=None` so the
     request does not re-enter the locked auth flow), parses `code` and `state` from the 302
     `Location`, and stashes them; `callback_handler` returns the stashed pair. Tests inspect
-    `authorize_url` to assert what the SDK put on the authorize request.
+    `authorize_url` to assert what the SDK put on the authorize request, and `iss`/`error` to
+    assert what the redirect carried — both record the redirect regardless of the
+    callback-boundary levers below.
 
     `state_override`: when set, `callback_handler` returns this value as the state instead of
     the one parsed from the redirect, so tests can drive the state-mismatch path.
 
     `iss_override`: when set, `callback_handler` returns this value as the RFC 9207 issuer
     instead of the one parsed from the redirect, so tests can drive the iss-mismatch path.
+
+    `code_override`: when set, `callback_handler` returns this value as the authorization code
+    instead of the one parsed from the redirect, so tests can drive the token-endpoint
+    rejection path.
+
+    `omit_iss`: when set, `callback_handler` returns no iss regardless of what the redirect
+    carried or what `iss_override` supplies (omission wins when both are set), so tests can
+    drive the missing-iss paths (the `iss_override` sentinel cannot express absence).
     """
 
-    def __init__(self, *, state_override: str | None = None, iss_override: str | None = None) -> None:
+    def __init__(
+        self,
+        *,
+        state_override: str | None = None,
+        iss_override: str | None = None,
+        code_override: str | None = None,
+        omit_iss: bool = False,
+    ) -> None:
         self.authorize_url: str | None = None
         self.authorize_urls: list[str] = []
         self.error: str | None = None
+        self.iss: str | None = None
         self._state_override = state_override
         self._iss_override = iss_override
+        self._code_override = code_override
+        self._omit_iss = omit_iss
         self._http: httpx.AsyncClient | None = None
         self._code: str = ""
         self._state: str | None = None
-        self._iss: str | None = None
 
     def bind(self, http_client: httpx.AsyncClient) -> None:
         self._http = http_client
@@ -166,14 +185,15 @@ async def redirect_handler(self, authorization_url: str) -> None:
         params = parse_qs(urlsplit(response.headers["location"]).query)
         self._code = params.get("code", [""])[0]
         self._state = params.get("state", [None])[0]
-        self._iss = params.get("iss", [None])[0]
+        self.iss = params.get("iss", [None])[0]
         self.error = params.get("error", [None])[0]
 
     async def callback_handler(self) -> AuthorizationCodeResult:
+        iss = self._iss_override if self._iss_override is not None else self.iss
         return AuthorizationCodeResult(
-            code=self._code,
+            code=self._code_override if self._code_override is not None else self._code,
             state=self._state_override if self._state_override is not None else self._state,
-            iss=self._iss_override if self._iss_override is not None else self._iss,
+            iss=None if self._omit_iss else iss,
         )
 
 
@@ -308,7 +328,7 @@ def first_challenge_shim(www_authenticate: str, *, path: str = "/mcp") -> Callab
     return lambda app: _FirstChallenge(app, path, www_authenticate)
 
 
-def step_up_shim(www_authenticate: str, *, on_nth_authenticated_post: int = 2) -> AppShim:
+def step_up_shim(www_authenticate: str, *, on_nth_authenticated_post: int = 2, persist: bool = False) -> AppShim:
     """Build an `app_shim` that 403s the Nth authenticated POST to `/mcp` with the given challenge.
 
     Subsequent requests pass through. Used to drive the client's `insufficient_scope` step-up
@@ -320,6 +340,10 @@ def step_up_shim(www_authenticate: str, *, on_nth_authenticated_post: int = 2) -
     first authenticated POST is the auth flow's retry of the original initialize request (yielded
     after the 401 branch, where the generator ends without inspecting the response), so a 403
     there would not reach the step-up handler.
+
+    `persist`: when set, every authenticated POST from the Nth onward receives the 403 challenge
+    instead of only the Nth, so tests can drive a further `insufficient_scope` challenge on the
+    request retried after a step-up.
     """
     seen = 0
     fired = False
@@ -328,7 +352,7 @@ def factory(app: ASGIApp) -> ASGIApp:
         async def wrapped(scope: Scope, receive: Receive, send: Send) -> None:
             nonlocal seen, fired
             if (
-                not fired
+                (persist or not fired)
                 and scope["type"] == "http"
                 and scope["path"] == "/mcp"
                 and scope["method"] == "POST"

tests/interaction/auth/_provider.py

@@ -49,6 +49,9 @@ class InMemoryAuthorizationServerProvider(
         `fail_next_refresh`: the next refresh-token exchange raises `invalid_grant` once.
         `reject_all_tokens`: `load_access_token` returns None for every token, so the bearer
             middleware 401s every authenticated request.
+        `rotate_refresh_tokens`: when False, the refresh exchange issues only a new access
+            token (the response carries no `refresh_token` and the presented one stays valid),
+            modelling an RFC 6749 §6 non-rotating authorization server.
     """
 
     def __init__(
@@ -59,6 +62,7 @@ def __init__(
         issue_expired_first: bool = False,
         fail_next_refresh: bool = False,
         reject_all_tokens: bool = False,
+        rotate_refresh_tokens: bool = True,
         issuer: str | None = None,
     ) -> None:
         self._default_scopes = list(default_scopes) if default_scopes is not None else ["mcp"]
@@ -71,6 +75,7 @@ def __init__(
         self._issue_expired_first = issue_expired_first
         self._fail_next_refresh = fail_next_refresh
         self._reject_all_tokens = reject_all_tokens
+        self._rotate_refresh_tokens = rotate_refresh_tokens
         self._tokens_issued = 0
         self.clients: dict[str, OAuthClientInformationFull] = {}
         self.codes: dict[str, AuthorizationCode] = {}
@@ -178,11 +183,23 @@ async def load_refresh_token(self, client: OAuthClientInformationFull, refresh_t
     async def exchange_refresh_token(
         self, client: OAuthClientInformationFull, refresh_token: RefreshToken, scopes: list[str]
     ) -> OAuthToken:
-        """Mint a new access token and rotate the refresh token, consuming the old one."""
+        """Mint a new access token and rotate the refresh token, consuming the old one.
+
+        Unless `rotate_refresh_tokens` is off: then only a new access token is minted, the
+        response carries no `refresh_token`, and the presented one stays valid.
+        """
         assert client.client_id is not None
         if self._fail_next_refresh:
             self._fail_next_refresh = False
             raise TokenError(error="invalid_grant", error_description="refresh denied by harness")
+        if not self._rotate_refresh_tokens:
+            access = self.mint_access_token(client_id=client.client_id, scopes=scopes)
+            return OAuthToken(
+                access_token=access,
+                token_type="Bearer",
+                expires_in=self._next_expires_in(),
+                scope=" ".join(scopes),
+            )
         access = self.mint_access_token(client_id=client.client_id, scopes=scopes)
         new_refresh = f"refresh_{secrets.token_hex(16)}"
         self.refresh_tokens[new_refresh] = RefreshToken(token=new_refresh, client_id=client.client_id, scopes=scopes)

tests/interaction/auth/test_as_handlers.py

@@ -16,10 +16,11 @@
 import httpx
 import pytest
 from inline_snapshot import snapshot
+from pydantic import AnyUrl
 
 from mcp.server import Server
 from mcp.server.auth.provider import ProviderTokenVerifier
-from mcp.shared.auth import OAuthClientInformationFull
+from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata
 from tests.interaction._connect import mounted_app
 from tests.interaction._requirements import requirement
 from tests.interaction.auth._harness import REDIRECT_URI, auth_settings, oauth_client_metadata
@@ -298,3 +299,34 @@ async def test_a_non_loopback_http_redirect_uri_is_accepted_at_registration(
     info = OAuthClientInformationFull.model_validate_json(response.content)
     assert [str(u) for u in (info.redirect_uris or [])] == ["http://evil.example/callback"]
     assert info.client_id in provider.clients
+
+
+@requirement("hosting:auth:as:register-echo-application-type")
+async def test_register_echoes_native_for_a_client_that_registered_application_type_web(
+    as_app: tuple[httpx.AsyncClient, InMemoryAuthorizationServerProvider],
+) -> None:
+    """A client registering `application_type: "web"` is told `"native"` in the registration echo.
+
+    Pins the known gap recorded on the requirement (divergence): the registration handler's
+    field-by-field passthrough omits `application_type`, so the model default fills the echo
+    where RFC 7591 §3.2.1 requires the registered value -- and the SDK OAuth client adopts the
+    echo into persisted storage, so the corruption is client-visible end to end. When the
+    one-line passthrough fix lands this test fails: re-pin the echo to `"web"`, delete the
+    Divergence, and add the echo assertion to
+    `test_dcr_sends_a_consumer_set_application_type_verbatim` (test_flow.py) per the
+    requirement's note.
+    """
+    http, _ = as_app
+    metadata = OAuthClientMetadata(
+        client_name="interaction-suite", redirect_uris=[AnyUrl(REDIRECT_URI)], application_type="web"
+    )
+
+    response = await http.post("/register", content=metadata.model_dump_json())
+
+    # Registration itself succeeds: the divergence is in the echo, not in acceptance.
+    assert response.status_code == 201
+    body = response.json()
+    # The request carried "web" (the metadata above); the echo says "native" -- the pinned gap.
+    assert body["application_type"] == "native"
+    # The omission is specific to application_type, not a generally lossy echo.
+    assert body["client_name"] == "interaction-suite"