Add DPoP (RFC 9449) support for token endpoint requests

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.

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

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")

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
 
 

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