From e331be658948f515273067300b01b1ee01de9c21 Mon Sep 17 00:00:00 2001 From: eaakun Date: Sun, 11 Oct 2026 01:27:41 +0800 Subject: [PATCH] Add DPoP (RFC 9449) support for token endpoint requests When the protected resource metadata advertises dpop_signing_alg_values_supported with ES256, attach DPoP proof JWTs to the authorization_code exchange and refresh_token grant, with automatic single retry on server nonce challenges. Accept token_type DPoP in token responses. Zero new dependencies. Closes #3671 --- docs/client/oauth-clients.md | 9 + src/mcp/client/auth/dpop.py | 105 ++++++++++ src/mcp/client/auth/oauth2.py | 61 +++++- src/mcp/shared/auth.py | 8 +- tests/client/auth/test_dpop.py | 127 ++++++++++++ tests/client/test_auth.py | 365 +++++++++++++++++++++++++++++++++ 6 files changed, 666 insertions(+), 9 deletions(-) create mode 100644 src/mcp/client/auth/dpop.py create mode 100644 tests/client/auth/test_dpop.py diff --git a/docs/client/oauth-clients.md b/docs/client/oauth-clients.md index 545784393d..ab6c7f0651 100644 --- a/docs/client/oauth-clients.md +++ b/docs/client/oauth-clients.md @@ -87,6 +87,15 @@ One transport rule applies to all of these requests: like the MCP request they r You wrote none of it. Two keyword arguments remain (`client_metadata_url` and `validate_resource_url`), and this file needs neither. `client_metadata_url` is the one worth knowing about; it gets its own section below. +## DPoP + +When the server's protected resource metadata advertises `dpop_signing_alg_values_supported` containing `ES256`, the provider speaks [DPoP](https://datatracker.ietf.org/doc/html/rfc9449): every token request (the authorization-code exchange and every refresh) carries a `DPoP` proof header, a short-lived JWT signed by a P-256 key the provider generates for this client. The issued tokens are bound to that key, so a stolen authorization code or refresh token cannot be redeemed by anyone else. Servers that don't advertise DPoP get byte-identical requests to before; nothing changes for them. + +Two behaviours to know about: + +* The provider answers a server nonce challenge (`DPoP-Nonce` on a `400`) by rebuilding the proof with that nonce and retrying once, by itself. +* The key lives in memory. Restart the process and the next refresh fails once, after which the provider re-runs the authorization flow with a fresh key. Persisting the key across restarts is planned; DPoP proofs on resource requests (the `DPoP` authorization scheme) are likewise future work — API calls still send `Authorization: Bearer ...`. + ### Try it The in-memory `Client(server)` your tests use is no help here: the whole point of the flow is an HTTP `401`, and there is no HTTP between an in-memory client and its server. diff --git a/src/mcp/client/auth/dpop.py b/src/mcp/client/auth/dpop.py new file mode 100644 index 0000000000..7ae0bc8fa6 --- /dev/null +++ b/src/mcp/client/auth/dpop.py @@ -0,0 +1,105 @@ +"""DPoP (RFC 9449) proofs for token endpoint requests. + +Attached when the authorization server advertises +``dpop_signing_alg_values_supported`` in its metadata. Binds issued tokens to +the client's key so stolen codes or refresh tokens cannot be redeemed +elsewhere. Token endpoint only; resource-server proofs (RFC 9449 §7) are +future work. Implemented on ``pyjwt[crypto]`` — no new dependencies. +""" + +import base64 +import hashlib +import json +import secrets +import time +from urllib.parse import urlsplit, urlunsplit + +import jwt +from cryptography.hazmat.primitives.asymmetric import ec + +from mcp.shared.auth import ProtectedResourceMetadata + + +def _b64u(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _b64u_decode(data: str) -> bytes: + return base64.urlsafe_b64decode(data + "=" * (-len(data) % 4)) + + +def _private_key(private_jwk_json: str) -> ec.EllipticCurvePrivateKey: + jwk = json.loads(private_jwk_json) + return ec.derive_private_key(int.from_bytes(_b64u_decode(jwk["d"]), "big"), ec.SECP256R1()) + + +def _normalize_htu(htu: str) -> str: + """RFC 9449 §4.2: the ``htu`` claim carries no query or fragment.""" + parts = urlsplit(htu) + return urlunsplit((parts.scheme, parts.netloc, parts.path, "", "")) + + +def generate_dpop_key() -> str: + """Generate a fresh P-256 keypair; returns the private JWK as JSON.""" + private_key = ec.generate_private_key(ec.SECP256R1()) + numbers = private_key.private_numbers() + public = numbers.public_numbers + return json.dumps( + { + "kty": "EC", + "crv": "P-256", + "x": _b64u(public.x.to_bytes(32, "big")), + "y": _b64u(public.y.to_bytes(32, "big")), + "d": _b64u(numbers.private_value.to_bytes(32, "big")), + } + ) + + +def dpop_public_jwk(private_jwk_json: str) -> dict[str, str]: + """Public JWK (no private material) from a private JWK JSON string.""" + jwk = json.loads(private_jwk_json) + return {k: jwk[k] for k in ("kty", "crv", "x", "y")} + + +def dpop_thumbprint(public_jwk: dict[str, str]) -> str: + """RFC 7638 JWK SHA-256 thumbprint (the ``jkt`` key identifier).""" + required = {k: public_jwk[k] for k in ("crv", "kty", "x", "y")} + canonical = json.dumps(required, separators=(",", ":"), sort_keys=True) + return _b64u(hashlib.sha256(canonical.encode("utf-8")).digest()) + + +def create_dpop_proof( + private_jwk_json: str, + *, + htm: str, + htu: str, + nonce: str | None = None, + iat: int | None = None, + jti: str | None = None, +) -> str: + """Create a DPoP proof JWT for one HTTP request (RFC 9449 §4).""" + public_jwk = dpop_public_jwk(private_jwk_json) + payload: dict[str, object] = { + "jti": jti or secrets.token_urlsafe(32), + "htm": htm.upper(), + "htu": _normalize_htu(htu), + "iat": iat if iat is not None else int(time.time()), + } + if nonce is not None: + payload["nonce"] = nonce + return jwt.encode( + payload, + _private_key(private_jwk_json), + algorithm="ES256", + headers={"typ": "dpop+jwt", "jwk": public_jwk}, + ) + + +def is_dpop_supported(resource_metadata: ProtectedResourceMetadata | None) -> bool: + """Whether the server advertises ES256 DPoP support. + + The SDK surfaces ``dpop_signing_alg_values_supported`` on the protected + resource metadata (RFC 9728 discovery document). + """ + algs = getattr(resource_metadata, "dpop_signing_alg_values_supported", None) + return bool(algs) and "ES256" in algs diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py index c513a99b46..69de90fbfb 100644 --- a/src/mcp/client/auth/oauth2.py +++ b/src/mcp/client/auth/oauth2.py @@ -19,6 +19,11 @@ from mcp_types.version import is_version_at_least from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, TypeAdapter, ValidationError +from mcp.client.auth.dpop import ( + create_dpop_proof, + generate_dpop_key, + is_dpop_supported, +) from mcp.client.auth.exceptions import OAuthFlowError, OAuthRegistrationError, OAuthTokenError from mcp.client.auth.utils import ( build_oauth_authorization_server_metadata_discovery_urls, @@ -171,6 +176,10 @@ class OAuthContext: current_tokens: OAuthToken | None = None token_expiry_time: float | None = None + # DPoP (RFC 9449): in-memory key. A restart drops it; the next refresh + # then fails once and the flow re-authorizes with a fresh key. + dpop_private_jwk: str | None = None + # State lock: anyio.Lock = field(default_factory=anyio.Lock) @@ -376,10 +385,10 @@ async def _handle_protected_resource_response(self, response: httpx2.Response) - f"Protected Resource Metadata request failed: {response.status_code}" ) # pragma: no cover - async def _perform_authorization(self) -> httpx2.Request: + async def _perform_authorization(self, dpop_nonce: str | None = None) -> httpx2.Request: """Perform the authorization flow.""" auth_code, code_verifier = await self._perform_authorization_code_grant() - token_request = await self._exchange_token_authorization_code(auth_code, code_verifier) + token_request = await self._exchange_token_authorization_code(auth_code, code_verifier, dpop_nonce) return token_request async def _perform_authorization_code_grant(self) -> tuple[str, str]: @@ -450,7 +459,40 @@ def _get_token_endpoint(self) -> str: token_url = urljoin(auth_base_url, "/token") return token_url - async def _exchange_token_authorization_code(self, auth_code: str, code_verifier: str) -> httpx2.Request: + def _dpop_active(self) -> bool: + """Whether DPoP proofs apply to token requests right now.""" + return is_dpop_supported(self.context.protected_resource_metadata) + + def _dpop_headers(self, token_url: str, dpop_nonce: str | None = None) -> dict[str, str]: + """DPoP header for a token request, or empty when DPoP is inactive.""" + if not self._dpop_active(): + return {} + if self.context.dpop_private_jwk is None: + self.context.dpop_private_jwk = generate_dpop_key() + proof = create_dpop_proof(self.context.dpop_private_jwk, htm="POST", htu=token_url, nonce=dpop_nonce) + return {"DPoP": proof} + + async def _dpop_nonce_retry( + self, response: httpx2.Response, build_request: Callable[..., Awaitable[httpx2.Request]] + ) -> httpx2.Request | None: + """Rebuild a token request when the AS answers with a DPoP nonce challenge. + + Returns the retry request, or None when no retry applies. Retries at most + once: the rebuilt request is not re-examined. + """ + if ( + response.status_code == 400 + and self._dpop_active() + and "DPoP" in response.request.headers + and "DPoP-Nonce" in response.headers + ): + logger.debug("Retrying token request with server-provided DPoP nonce") + return await build_request(dpop_nonce=response.headers["DPoP-Nonce"]) + return None + + async def _exchange_token_authorization_code( + self, auth_code: str, code_verifier: str, dpop_nonce: str | None = None + ) -> httpx2.Request: """Build token exchange request for authorization_code flow.""" if self.context.client_metadata.redirect_uris is None: raise OAuthFlowError("No redirect URIs provided for authorization code grant") # pragma: no cover @@ -473,6 +515,7 @@ async def _exchange_token_authorization_code(self, auth_code: str, code_verifier # Prepare authentication based on preferred method headers = {"Content-Type": "application/x-www-form-urlencoded"} token_data, headers = self.context.prepare_token_auth(token_data, headers) + headers.update(self._dpop_headers(token_url, dpop_nonce)) return httpx2.Request("POST", token_url, data=token_data, headers=headers) @@ -500,7 +543,7 @@ async def _handle_token_response(self, response: httpx2.Response) -> None: self.context.update_token_expiry(token_response) await self.context.storage.set_tokens(token_response) - async def _refresh_token(self) -> httpx2.Request: + async def _refresh_token(self, dpop_nonce: str | None = None) -> httpx2.Request: """Build token refresh request.""" if not self.context.current_tokens or not self.context.current_tokens.refresh_token: raise OAuthTokenError("No refresh token available") # pragma: no cover @@ -527,6 +570,7 @@ async def _refresh_token(self) -> httpx2.Request: # Prepare authentication based on preferred method headers = {"Content-Type": "application/x-www-form-urlencoded"} refresh_data, headers = self.context.prepare_token_auth(refresh_data, headers) + headers.update(self._dpop_headers(token_url, dpop_nonce)) return httpx2.Request("POST", token_url, data=refresh_data, headers=headers) @@ -614,6 +658,9 @@ async def _auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx2.Req # Try to refresh token refresh_request = await self._refresh_token() refresh_response = yield refresh_request + dpop_retry = await self._dpop_nonce_retry(refresh_response, self._refresh_token) + if dpop_retry is not None: + refresh_response = yield dpop_retry if not await self._handle_refresh_response(refresh_response): # Refresh failed, need full re-authentication @@ -782,7 +829,11 @@ async def _auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx2.Req await self.context.storage.set_client_info(client_information) # Step 5: Perform authorization and complete token exchange - token_response = yield await self._perform_authorization() + token_request = await self._perform_authorization() + token_response = yield token_request + dpop_retry = await self._dpop_nonce_retry(token_response, self._perform_authorization) + if dpop_retry is not None: + token_response = yield dpop_retry await self._handle_token_response(token_response) except Exception: logger.exception("OAuth flow error") diff --git a/src/mcp/shared/auth.py b/src/mcp/shared/auth.py index 881379d381..e47dd22bc6 100644 --- a/src/mcp/shared/auth.py +++ b/src/mcp/shared/auth.py @@ -27,7 +27,7 @@ class OAuthToken(BaseModel): """See https://datatracker.ietf.org/doc/html/rfc6749#section-5.1""" access_token: str - token_type: Literal["Bearer"] = "Bearer" + token_type: Literal["Bearer", "DPoP"] = "Bearer" expires_in: int | None = None scope: str | None = None refresh_token: str | None = None @@ -36,9 +36,9 @@ class OAuthToken(BaseModel): @classmethod def normalize_token_type(cls, v: str | None) -> str | None: if isinstance(v, str): - # Bearer is title-cased in the spec, so we normalize it - # https://datatracker.ietf.org/doc/html/rfc6750#section-4 - return v.title() + # Bearer is title-cased in the spec (RFC 6750 §4); + # DPoP is always exactly "DPoP" (RFC 9449 §5). + return "Bearer" if v.lower() == "bearer" else v return v # pragma: no cover diff --git a/tests/client/auth/test_dpop.py b/tests/client/auth/test_dpop.py new file mode 100644 index 0000000000..ba164a50c2 --- /dev/null +++ b/tests/client/auth/test_dpop.py @@ -0,0 +1,127 @@ +"""Tests for mcp.client.auth.dpop (RFC 9449 DPoP proofs).""" + +import json + +import jwt +import pytest +from jwt.algorithms import ECAlgorithm + +from mcp.client.auth.dpop import ( + create_dpop_proof, + dpop_public_jwk, + dpop_thumbprint, + generate_dpop_key, + is_dpop_supported, +) +from mcp.shared.auth import ProtectedResourceMetadata + +# RFC 9449, Figures 5/8/9: the example JWK and its published jkt. +RFC9449_EXAMPLE_JWK = { + "kty": "EC", + "crv": "P-256", + "x": "l8tFrhx-34tV3hRICRDY9zCkDlpBhF42UQUfWVAWBFs", + "y": "9VE4jf_Ok_o64zbTTlcuNJajHmt6v9TDVrU0CdvGRDA", +} +RFC9449_EXAMPLE_JKT = "0ZcOCORZNYy-DWpqq30jZyJGHTN0d2HglBV3uiguA4I" + + +def _decode_claims(proof: str) -> dict: + """Claims without signature verification (structure assertions only).""" + return jwt.decode(proof, options={"verify_signature": False}) + + +def _verify_signature(proof: str) -> dict: + """Full verification against the proof's embedded JWK.""" + header = jwt.get_unverified_header(proof) + key = ECAlgorithm.from_jwk(json.dumps(header["jwk"])) + return jwt.decode(proof, key, algorithms=["ES256"]) + + +def test_dpop_proof_header_advertises_dpop_jwt_type(): + """RFC 9449 §4.2 mandates typ=dpop+jwt and alg=ES256 with an embedded JWK.""" + proof = create_dpop_proof(generate_dpop_key(), htm="POST", htu="https://a.example/token") + header = jwt.get_unverified_header(proof) + assert header["typ"] == "dpop+jwt" + assert header["alg"] == "ES256" + assert header["jwk"]["kty"] == "EC" + assert "d" not in header["jwk"] + + +def test_dpop_proof_binds_method_and_uri_without_query_or_fragment(): + """RFC 9449 §4.2: htm/htu bind the request; htu carries no query/fragment.""" + proof = create_dpop_proof(generate_dpop_key(), htm="post", htu="https://a.example:8443/t/x?query=1#frag") + claims = _decode_claims(proof) + assert claims["htm"] == "POST" + assert claims["htu"] == "https://a.example:8443/t/x" + assert claims["iat"] > 0 + assert claims["jti"] + + +def test_dpop_proof_carries_server_nonce_only_when_challenged(): + """RFC 9449 §9.1: the nonce claim appears exactly when the server demanded one.""" + key = generate_dpop_key() + assert _decode_claims(create_dpop_proof(key, htm="POST", htu="https://a.example/t")).get("nonce") is None + assert ( + _decode_claims(create_dpop_proof(key, htm="POST", htu="https://a.example/t", nonce="srv-nonce"))["nonce"] + == "srv-nonce" + ) + + +def test_dpop_proof_signature_verifies_with_embedded_key(): + """A proof minted by create_dpop_proof verifies against its embedded JWK.""" + proof = create_dpop_proof(generate_dpop_key(), htm="POST", htu="https://a.example/token") + claims = _verify_signature(proof) + assert claims["htm"] == "POST" + assert claims["htu"] == "https://a.example/token" + + +def test_dpop_thumbprint_matches_rfc9449_published_value(): + """RFC 9449 Figures 8/9 publish the jkt of the Figure 5 example key.""" + assert dpop_thumbprint(RFC9449_EXAMPLE_JWK) == RFC9449_EXAMPLE_JKT + + +def test_dpop_proof_accepts_explicit_iat_and_jti(): + """Explicit iat/jti are honored instead of generated (deterministic tests).""" + proof = create_dpop_proof(generate_dpop_key(), htm="GET", htu="https://a.example/r", iat=1234567890, jti="fixed-id") + claims = _decode_claims(proof) + assert claims["iat"] == 1234567890 + assert claims["jti"] == "fixed-id" + + +def test_dpop_public_jwk_never_carries_private_material(): + """The embedded JWK must not leak the private key.""" + public = dpop_public_jwk(generate_dpop_key()) + assert set(public) == {"kty", "crv", "x", "y"} + + +@pytest.mark.parametrize( + ("algs", "expected"), + [ + (None, False), + ([], False), + (["EdDSA"], False), + (["ES256"], True), + (["ES256", "EdDSA"], True), + ], +) +def test_dpop_support_requires_es256_in_server_metadata(algs: list[str] | None, expected: bool): + """DPoP activates only when the server advertises ES256 in its metadata.""" + metadata = ( + ProtectedResourceMetadata( + resource="https://api.example.com/v1/mcp", + authorization_servers=["https://auth.example.com"], + dpop_signing_alg_values_supported=algs, + ) + if algs is not None + else None + ) + assert is_dpop_supported(metadata) is expected + + +def test_dpop_support_ignores_metadata_without_dpop_field(): + """Servers that predate DPoP metadata simply don't get proofs.""" + metadata = ProtectedResourceMetadata( + resource="https://api.example.com/v1/mcp", + authorization_servers=["https://auth.example.com"], + ) + assert is_dpop_supported(metadata) is False diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py index 18a1566705..edf995f85c 100644 --- a/tests/client/test_auth.py +++ b/tests/client/test_auth.py @@ -8,8 +8,10 @@ from urllib.parse import parse_qs, quote, unquote, urlparse import httpx2 +import jwt import pytest from inline_snapshot import Is, snapshot +from jwt.algorithms import ECAlgorithm from pydantic import AnyHttpUrl, AnyUrl from mcp.client.auth import OAuthClientProvider, PKCEParameters @@ -3667,3 +3669,366 @@ async def echo_callback() -> AuthorizationCodeResult: await auth_flow.asend(httpx2.Response(200, request=final_req)) except StopAsyncIteration: pass + + +# --- DPoP (RFC 9449) ---------------------------------------------------------- + + +def _dpop_prm() -> ProtectedResourceMetadata: + """PRM advertising ES256 DPoP support (the SDK surfaces the flag here).""" + return ProtectedResourceMetadata( + resource="https://api.example.com/v1/mcp", + authorization_servers=["https://auth.example.com"], + dpop_signing_alg_values_supported=["ES256"], + ) + + +def _dpop_oauth_metadata() -> OAuthMetadata: + return OAuthMetadata( + issuer="https://auth.example.com", + authorization_endpoint="https://auth.example.com/authorize", + token_endpoint="https://auth.example.com/token", + ) + + +def _dpop_client_info() -> OAuthClientInformationFull: + return OAuthClientInformationFull( + client_id="test_client", + client_secret="test_secret", + redirect_uris=[AnyUrl("http://localhost:3030/callback")], + token_endpoint_auth_method="client_secret_post", + ) + + +def _verified_dpop_claims(request: httpx2.Request) -> dict: + """Decode the DPoP proof on a request, verifying its signature.""" + proof = request.headers["DPoP"] + jwk = jwt.get_unverified_header(proof)["jwk"] + key = ECAlgorithm.from_jwk(json.dumps(jwk)) + return jwt.decode(proof, key, algorithms=["ES256"]) + + +@pytest.mark.anyio +async def test_token_exchange_attaches_dpop_proof_when_server_advertises_es256( + oauth_provider: OAuthClientProvider, +): + """The authorization_code exchange carries a DPoP proof bound to the token URL. + + Provenance: RFC 9449 §4.2 — proofs are required on token requests when the + server advertises DPoP support. + """ + oauth_provider.context.client_info = _dpop_client_info() + oauth_provider.context.protected_resource_metadata = _dpop_prm() + oauth_provider.context.oauth_metadata = _dpop_oauth_metadata() + + request = await oauth_provider._exchange_token_authorization_code("code", "verifier") + + claims = _verified_dpop_claims(request) + assert claims["htm"] == "POST" + assert claims["htu"] == "https://auth.example.com/token" + assert "nonce" not in claims + + +@pytest.mark.anyio +async def test_token_exchange_omits_dpop_header_when_server_silent( + oauth_provider: OAuthClientProvider, +): + """No DPoP advertisement means no DPoP header — plain servers are untouched. + + Provenance: SDK-defined back-compat; the feature must be purely additive. + """ + oauth_provider.context.client_info = _dpop_client_info() + oauth_provider.context.oauth_metadata = _dpop_oauth_metadata() + + request = await oauth_provider._exchange_token_authorization_code("code", "verifier") + + assert "DPoP" not in request.headers + + +@pytest.mark.anyio +async def test_dpop_key_is_reused_across_token_requests(oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken): + """One keypair per client: the embedded JWK is stable across requests. + + Provenance: SDK-defined; rotating the key per request would break the + server's key binding. + """ + oauth_provider.context.current_tokens = valid_tokens + oauth_provider.context.client_info = _dpop_client_info() + oauth_provider.context.protected_resource_metadata = _dpop_prm() + oauth_provider.context.oauth_metadata = _dpop_oauth_metadata() + + first = await oauth_provider._exchange_token_authorization_code("c1", "v1") + second = await oauth_provider._refresh_token() + + jwk1 = jwt.get_unverified_header(first.headers["DPoP"])["jwk"] + jwk2 = jwt.get_unverified_header(second.headers["DPoP"])["jwk"] + assert jwk1 == jwk2 + + +@pytest.mark.anyio +async def test_refresh_request_attaches_dpop_proof_when_server_advertises_es256( + oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken +): + """The refresh_token grant carries a DPoP proof like the initial exchange. + + Provenance: RFC 9449 §4.2 — every token request is proof-bound. + """ + oauth_provider.context.current_tokens = valid_tokens + oauth_provider.context.client_info = _dpop_client_info() + oauth_provider.context.protected_resource_metadata = _dpop_prm() + oauth_provider.context.oauth_metadata = _dpop_oauth_metadata() + + request = await oauth_provider._refresh_token() + + claims = _verified_dpop_claims(request) + assert claims["htm"] == "POST" + assert claims["htu"] == "https://auth.example.com/token" + + +@pytest.mark.anyio +async def test_dpop_nonce_challenge_rebuilds_request_with_nonce( + oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken +): + """A 400 + DPoP-Nonce answer rebuilds the request with a nonce-bound proof. + + Provenance: RFC 9449 §9.1 — the client retries once with the server nonce. + """ + oauth_provider.context.current_tokens = valid_tokens + oauth_provider.context.client_info = _dpop_client_info() + oauth_provider.context.protected_resource_metadata = _dpop_prm() + oauth_provider.context.oauth_metadata = _dpop_oauth_metadata() + + request = await oauth_provider._refresh_token() + challenge = httpx2.Response(400, headers={"DPoP-Nonce": "srv-nonce-1"}, request=request) + + retry = await oauth_provider._dpop_nonce_retry(challenge, oauth_provider._refresh_token) + + assert retry is not None + assert _verified_dpop_claims(retry)["nonce"] == "srv-nonce-1" + + +@pytest.mark.anyio +async def test_dpop_nonce_retry_ignores_non_challenge_responses( + oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken +): + """Only a real nonce challenge (400 + DPoP-Nonce on a DPoP request) retries. + + Provenance: SDK-defined; a stray header must never trigger a token-request retry. + """ + oauth_provider.context.current_tokens = valid_tokens + oauth_provider.context.client_info = _dpop_client_info() + oauth_provider.context.protected_resource_metadata = _dpop_prm() + oauth_provider.context.oauth_metadata = _dpop_oauth_metadata() + + dpop_request = await oauth_provider._refresh_token() + plain_request = httpx2.Request("POST", "https://auth.example.com/token") + build = oauth_provider._refresh_token + + assert await oauth_provider._dpop_nonce_retry(httpx2.Response(200, request=dpop_request), build) is None + assert await oauth_provider._dpop_nonce_retry(httpx2.Response(400, request=dpop_request), build) is None + assert ( + await oauth_provider._dpop_nonce_retry( + httpx2.Response(400, headers={"DPoP-Nonce": "n"}, request=plain_request), build + ) + is None + ) + + oauth_provider.context.protected_resource_metadata = None + assert ( + await oauth_provider._dpop_nonce_retry( + httpx2.Response(400, headers={"DPoP-Nonce": "n"}, request=dpop_request), build + ) + is None + ) + + +def _dpop_e2e_provider( + client_metadata: OAuthClientMetadata, + mock_storage: MockTokenStorage, + redirect_handler, + callback_handler, +) -> OAuthClientProvider: + return OAuthClientProvider( + server_url="https://api.example.com/v1/mcp", + client_metadata=client_metadata, + storage=mock_storage, + redirect_handler=redirect_handler, + callback_handler=callback_handler, + ) + + +@pytest.mark.anyio +async def test_token_flow_retries_with_server_provided_nonce( + client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage +): + """Full flow: the token endpoint's nonce challenge triggers one proof-bound retry. + + Provenance: RFC 9449 §9.1, driven through _auth_flow against an in-memory + mock AS that advertises DPoP in its protected resource metadata. + """ + captured: dict[str, str] = {} + + async def redirect_handler(url: str) -> None: + captured["state"] = parse_qs(urlparse(url).query)["state"][0] + + async def callback_handler() -> AuthorizationCodeResult: + return AuthorizationCodeResult(code="test_auth_code", state=captured["state"]) + + provider = _dpop_e2e_provider(client_metadata, mock_storage, redirect_handler, callback_handler) + provider._initialized = True + provider.context.client_info = _dpop_client_info() + + test_request = httpx2.Request("GET", "https://api.example.com/v1/mcp") + auth_flow = provider.async_auth_flow(test_request) + assert "Authorization" not in (await auth_flow.__anext__()).headers + + prm_request = await auth_flow.asend(httpx2.Response(401, headers={}, request=test_request)) + assert "oauth-protected-resource" in str(prm_request.url) + + prm_response = httpx2.Response( + 200, + content=json.dumps( + { + "resource": "https://api.example.com/v1/mcp", + "authorization_servers": ["https://auth.example.com"], + "dpop_signing_alg_values_supported": ["ES256"], + } + ).encode(), + request=prm_request, + ) + metadata_request = await auth_flow.asend(prm_response) + assert metadata_request.method == "GET" + + metadata_response = httpx2.Response( + 200, + content=json.dumps( + { + "issuer": "https://auth.example.com", + "authorization_endpoint": "https://auth.example.com/authorize", + "token_endpoint": "https://auth.example.com/token", + } + ).encode(), + request=metadata_request, + ) + token_request = await auth_flow.asend(metadata_response) + assert token_request.method == "POST" + assert "nonce" not in _verified_dpop_claims(token_request) + + retry_request = await auth_flow.asend( + httpx2.Response(400, headers={"DPoP-Nonce": "e2e-nonce"}, request=token_request) + ) + assert _verified_dpop_claims(retry_request)["nonce"] == "e2e-nonce" + assert ( + jwt.get_unverified_header(retry_request.headers["DPoP"])["jwk"] + == jwt.get_unverified_header(token_request.headers["DPoP"])["jwk"] + ) + + final_request = await auth_flow.asend( + httpx2.Response( + 200, + content=json.dumps({"access_token": "dpop-bound-at", "token_type": "DPoP", "expires_in": 3600}).encode(), + request=retry_request, + ) + ) + assert final_request.headers["Authorization"] == "Bearer dpop-bound-at" + + try: + await auth_flow.asend(httpx2.Response(200, request=final_request)) + except StopAsyncIteration: + pass + + +@pytest.mark.anyio +async def test_refresh_flow_retries_with_server_provided_nonce( + client_metadata: OAuthClientMetadata, + mock_storage: MockTokenStorage, + valid_tokens: OAuthToken, +): + """An expired DPoP-bound session refreshes through the same nonce retry. + + Provenance: RFC 9449 §9.1 applied to the refresh_token grant. + """ + + async def callback_handler() -> AuthorizationCodeResult: + raise AssertionError("full re-auth must not trigger here") + + async def redirect_handler(url: str) -> None: + pass + + provider = _dpop_e2e_provider(client_metadata, mock_storage, redirect_handler, callback_handler) + provider._initialized = True + provider.context.current_tokens = valid_tokens + provider.context.token_expiry_time = time.time() - 1 + provider.context.client_info = _dpop_client_info() + provider.context.protected_resource_metadata = _dpop_prm() + provider.context.oauth_metadata = _dpop_oauth_metadata() + + test_request = httpx2.Request("GET", "https://api.example.com/v1/mcp") + auth_flow = provider.async_auth_flow(test_request) + + refresh_request = await auth_flow.__anext__() + assert refresh_request.method == "POST" + assert "nonce" not in _verified_dpop_claims(refresh_request) + + retry_request = await auth_flow.asend( + httpx2.Response(400, headers={"DPoP-Nonce": "refresh-nonce"}, request=refresh_request) + ) + assert _verified_dpop_claims(retry_request)["nonce"] == "refresh-nonce" + + final_request = await auth_flow.asend( + httpx2.Response( + 200, + content=json.dumps({"access_token": "new-at", "token_type": "DPoP", "expires_in": 3600}).encode(), + request=retry_request, + ) + ) + assert final_request.headers["Authorization"] == "Bearer new-at" + + try: + await auth_flow.asend(httpx2.Response(200, request=final_request)) + except StopAsyncIteration: + pass + + +@pytest.mark.anyio +async def test_refresh_flow_succeeds_without_nonce_challenge( + client_metadata: OAuthClientMetadata, + mock_storage: MockTokenStorage, + valid_tokens: OAuthToken, +): + """A cooperative AS refreshes in one round trip — no retry is attempted. + + Provenance: SDK-defined; the retry path must stay dormant without a challenge. + """ + + async def callback_handler() -> AuthorizationCodeResult: + raise AssertionError("full re-auth must not trigger here") + + async def redirect_handler(url: str) -> None: + pass + + provider = _dpop_e2e_provider(client_metadata, mock_storage, redirect_handler, callback_handler) + provider._initialized = True + provider.context.current_tokens = valid_tokens + provider.context.token_expiry_time = time.time() - 1 + provider.context.client_info = _dpop_client_info() + provider.context.protected_resource_metadata = _dpop_prm() + provider.context.oauth_metadata = _dpop_oauth_metadata() + + test_request = httpx2.Request("GET", "https://api.example.com/v1/mcp") + auth_flow = provider.async_auth_flow(test_request) + + refresh_request = await auth_flow.__anext__() + final_request = await auth_flow.asend( + httpx2.Response( + 200, + content=json.dumps({"access_token": "new-at", "token_type": "Bearer", "expires_in": 3600}).encode(), + request=refresh_request, + ) + ) + assert final_request.headers["Authorization"] == "Bearer new-at" + + try: + await auth_flow.asend(httpx2.Response(200, request=final_request)) + except StopAsyncIteration: + pass