fix(cache,verifier): expires_in tri-state, DPoP §4.3 cardinality, cnf_jkt round-trip (#15)

Bundles three related correctness fixes around token caching and inbound
DPoP enforcement.

* expires_in tri-state: TokenCache.set now distinguishes a missing
  expires_in from expires_in: 0. The previous branch collapsed both into
  default_ttl, so an AS-issued one-shot token (RFC 6749 §5.1 permits
  expires_in: 0) was cached for the default hour. The store now treats
  None as "apply default TTL", refuses to store when the value is 0 (the
  entry is born expired), and honors n seconds when positive.
  parse_token_response and the _optional_int sentinel carry the
  missing-vs-zero distinction through the parser end-to-end.

* DPoP §4.3 cardinality: inbound DPoP proof cardinality (RFC 9449 §4.3
  #1) is now enforced. read_dpop_header reads the full multi-value DPoP
  header list (and splits on "," defensively to catch proxies that
  pre-join repeated headers) and raises DPoPMultipleProofsError when more
  than one non-empty proof is present. www_authenticate maps this error
  to error="invalid_dpop_proof" per RFC 9449 §7.1; other DPoPError shapes
  keep the historical invalid_token mapping.

* cnf_jkt round-trip: CacheEntry now persists the DPoP key thumbprint
  (RFC 9449 §6.1), so a sender-constrained token retrieved from cache
  reports its binding to downstream callers instead of silently degrading
  to a bearer-only shape.

BREAKING CHANGE (pre-1.0): TokenResponse.expires_in and
CacheEntry.expires_in are now typed int | None (was int). A token
response that omits expires_in is None rather than 0 — the wire-level
prerequisite for the tri-state fix, since an absent field is otherwise
indistinguishable from expires_in: 0 once the JSON is decoded. Typed
downstream callers reading expires_in directly (arithmetic, comparison,
formatting) must guard for None: treat None as "apply your default" and
0 as "already expired".

Author: muralx

CHANGELOG.md

@@ -14,12 +14,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `authplane-fastmcp`, `authplane-mcp`: `AuthplaneTokenVerifier` caches the in-flight verify task per request (keyed by access token on `request.state`), so a repeat `verify_token` within the same HTTP request awaits the same task rather than re-entering the inbound DPoP replay store. Cross-request replay protection is unaffected (distinct requests get distinct caches).
 
 ### Fixed
+- `TokenCache.set` now distinguishes a missing `expires_in` from `expires_in: 0`. Both previously collapsed into `default_ttl`. The store now applies `default_ttl` only when `expires_in` is absent (`None`), treats `expires_in: 0` (RFC 6749 §5.1) as already expired and refuses to store it, and honors `n` seconds when positive. `parse_token_response` carries the missing-vs-zero distinction through the parser.
+- `authplane-fastmcp`, `authplane-mcp`: inbound DPoP cardinality (RFC 9449 §4.3 #1) is now enforced. `read_dpop_header` reads the full multi-value `DPoP` header list (and splits on `,` defensively to catch proxies that pre-join duplicate headers) and raises `DPoPMultipleProofsError` when more than one non-empty proof is present. `www_authenticate` maps this error to `error="invalid_dpop_proof"` per RFC 9449 §7.1.
 - `authplane-fastmcp`, `authplane-mcp`: inbound DPoP proof-of-possession is now enforced end-to-end. `AuthplaneTokenVerifier.verify_token` forwards a `DPoPRequestContext` (method + reconstructed `htu` + proof header) to `AuthplaneResource.verify`, so `inbound_dpop=InboundDPoPOptions(required=True)` checks the proof on every request. The `htu` origin is always the operator-configured resource URI, never the inbound `Host` / `X-Forwarded-Proto` headers. Operators using `required=True` with `authplane-mcp` should call `install_request_context(mcp)` after constructing `FastMCP` so the verifier can read the per-request context; if it is not installed the request fails closed (401) rather than skipping the check.
 - `authplane-fastmcp`, `authplane-mcp`: DPoP `htu` reconstruction reads `scope["raw_path"]` to preserve percent-encoding (e.g. `%2F`) on the wire under ASGI, falling back to `request.url.path` when the server omits `raw_path`.
 - `authplane-mcp`: `install_request_context(mcp)` is idempotent — repeated calls on the same `FastMCP` instance are no-ops.
 - `require_scope` (singular) now renders an empty token scope set as `(none)` instead of `[]`, matching the plural helper's output. Logging pipelines keyed on the old `Token has scopes: []` string should be updated.
 - Docs and demos now run adapter setup, the async server entry point (`run_streamable_http_async` / `run_async`), and `aclose()` in a single `asyncio.run(main())`, keeping the client's locks, HTTP pool, and background JWKS/metadata refresh tasks on one event loop.
 
+### Changed
+- **BREAKING (pre-1.0)** `TokenResponse.expires_in` and `CacheEntry.expires_in` are now typed `int | None` (was `int`), so a token response that omits `expires_in` is `None` rather than `0`. **Migration:** typed downstream callers reading `resp.expires_in` directly (arithmetic, comparison, formatting) must guard for `None`; treat `None` as "apply your default" and `0` as "already expired".
+
 ## [0.2.0] - 2026-05-20
 
 ### Security

authplane-fastmcp/tests/test_verifier_dpop_cache.py

@@ -201,13 +201,62 @@ async def test_dpop_context_header_lookup_is_case_insensitive() -> None:
     assert mock.verify.await_args.kwargs["dpop_request"].proof == "lower.case.proof"
 
 
+@pytest.mark.asyncio
+async def test_duplicate_dpop_headers_fail_auth() -> None:
+    """Two ``DPoP`` headers on one request violate RFC 9449 §4.3 #1.
+
+    The adapter must fail authentication (``verify_token`` returns
+    ``None``) without ever invoking the underlying ``verify`` — the
+    cardinality guard runs before the resource verifier sees the proof.
+    """
+    mock = _mock_verifier()
+    # Bypass _make_request to inject two DPoP headers; the helper's dict
+    # shape cannot represent the duplicate.
+    scope: dict[str, Any] = {
+        "type": "http",
+        "method": "POST",
+        "path": "/mcp",
+        "raw_path": b"/mcp",
+        "query_string": b"",
+        "headers": [
+            (b"dpop", b"proof.one.x"),
+            (b"dpop", b"proof.two.y"),
+        ],
+        "scheme": "http",
+        "server": ("testserver", 80),
+        "client": None,
+        "root_path": "",
+        "http_version": "1.1",
+    }
+    request = Request(scope)
+    verifier = _make_token_verifier(mock, request)
+
+    result = await verifier.verify_token("valid_token")
+    assert result is None
+    mock.verify.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_comma_joined_dpop_value_fails_auth() -> None:
+    """A single comma-joined ``DPoP`` value (the proxy-collapsed shape
+    permitted by RFC 9110 §5.3) is unambiguously two proofs — JWS compact
+    has no literal comma — and must trip the same §4.3 guard.
+    """
+    mock = _mock_verifier()
+    request = _make_request(headers={"DPoP": "a.b.c, x.y.z"})
+    verifier = _make_token_verifier(mock, request)
+
+    result = await verifier.verify_token("valid_token")
+    assert result is None
+    mock.verify.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_htu_origin_from_configured_resource_not_host_header() -> None:
     """htu's origin comes from the configured resource, never from Host.
 
-    Mirrors the TS sibling: an upstream that controls the Host /
-    X-Forwarded-Proto headers must not be able to decide which htu the
-    DPoP proof is validated against.
+    An upstream that controls the Host / X-Forwarded-Proto headers must
+    not be able to decide which htu the DPoP proof is validated against.
     """
     mock = _mock_verifier(resource="https://api.example.com/mcp")
     request = _make_request(

authplane-mcp/tests/test_verifier_dpop_cache.py

@@ -206,6 +206,56 @@ async def test_dpop_context_header_lookup_is_case_insensitive() -> None:
     assert mock.verify.await_args.kwargs["dpop_request"].proof == "lower.case.proof"
 
 
+@pytest.mark.asyncio
+async def test_duplicate_dpop_headers_fail_auth() -> None:
+    """Two ``DPoP`` headers on one request violate RFC 9449 §4.3 #1.
+
+    The adapter must fail authentication (``verify_token`` returns
+    ``None``) without ever invoking the underlying ``verify`` — the
+    cardinality guard runs before the resource verifier sees the proof.
+    """
+    mock = _mock_verifier()
+    # Bypass _make_request to inject two DPoP headers; the helper's dict
+    # shape cannot represent the duplicate.
+    scope: dict[str, Any] = {
+        "type": "http",
+        "method": "POST",
+        "path": "/mcp",
+        "raw_path": b"/mcp",
+        "query_string": b"",
+        "headers": [
+            (b"dpop", b"proof.one.x"),
+            (b"dpop", b"proof.two.y"),
+        ],
+        "scheme": "http",
+        "server": ("testserver", 80),
+        "client": None,
+        "root_path": "",
+        "http_version": "1.1",
+    }
+    request = Request(scope)
+    verifier = _make_token_verifier(mock, request)
+
+    result = await verifier.verify_token("valid_token")
+    assert result is None
+    mock.verify.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_comma_joined_dpop_value_fails_auth() -> None:
+    """A single comma-joined ``DPoP`` value (the proxy-collapsed shape
+    permitted by RFC 9110 §5.3) is unambiguously two proofs — JWS compact
+    has no literal comma — and must trip the same §4.3 guard.
+    """
+    mock = _mock_verifier()
+    request = _make_request(headers={"DPoP": "a.b.c, x.y.z"})
+    verifier = _make_token_verifier(mock, request)
+
+    result = await verifier.verify_token("valid_token")
+    assert result is None
+    mock.verify.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_htu_origin_from_configured_resource_not_host_header() -> None:
     """htu's origin comes from the configured resource, never from Host.

authplane/__init__.py

@@ -35,6 +35,7 @@
     ConsentRequiredError,
     DPoPBindingMismatchError,
     DPoPError,
+    DPoPMultipleProofsError,
     DPoPNotSupportedError,
     DPoPProofMissingError,
     DPoPReplayDetectedError,
@@ -84,6 +85,7 @@
     "DPoPBindingMismatchError",
     "DPoPError",
     "DPoPKeyMaterial",
+    "DPoPMultipleProofsError",
     "DPoPNonceStore",
     "DPoPNotSupportedError",
     "DPoPProofMissingError",

authplane/_dpop_adapter.py

@@ -28,6 +28,8 @@
 
 from typing import TYPE_CHECKING, Protocol
 
+from .errors import DPoPMultipleProofsError
+
 if TYPE_CHECKING:
     import asyncio
 
@@ -61,6 +63,8 @@ class _HeadersLike(Protocol):
 
     def get(self, key: str, default: str | None = ...) -> str | None: ...
 
+    def getlist(self, key: str) -> list[str]: ...
+
 
 class _URLLike(Protocol):
     """Minimal slice of ``starlette.datastructures.URL``."""
@@ -109,15 +113,44 @@ def __init__(self, method: str, url: str, proof: str | None) -> None:
 
 
 def read_dpop_header(request: _RequestLike) -> str | None:
-    """Read the ``DPoP`` request header (case-insensitive, first value).
+    """Read the ``DPoP`` request header, enforcing RFC 9449 §4.3 #1.
+
+    Returns the single proof JWT when exactly one non-empty ``DPoP``
+    header value is present, or ``None`` when no ``DPoP`` header is
+    present. Raises :class:`DPoPMultipleProofsError` when the request
+    carries more than one ``DPoP`` header value.
+
+    Two on-wire shapes are rejected:
+
+    1. Multiple ``DPoP`` headers on the request (``headers.getlist``
+       returns ≥ 2 non-empty entries).
+    2. A single ``DPoP`` header value pre-joined with ``,`` by an
+       upstream proxy or framework — RFC 9110 §5.3 permits combining
+       repeated headers this way. JWS compact serialization never
+       contains a literal comma, so split-on-comma is sound.
 
-    Starlette's ``Headers.get`` already does case-insensitive lookup
-    and returns the first occurrence when a header is repeated. Strict
-    rejection of repeated ``DPoP`` headers (RFC 9449 §4.3 #1) is
-    tracked separately and is intentionally not enforced here so this
-    layer does not overlap with that work.
+    Trimming and empty-piece filtering mirror the cross-language
+    cardinality boundary so a request carrying ``"DPoP: "`` (whitespace
+    only) is treated as header-absent rather than as one value.
     """
-    return request.headers.get("dpop")
+    raw_values = request.headers.getlist("dpop")
+    filtered: list[str] = []
+    for raw in raw_values:
+        trimmed = raw.strip()
+        if not trimmed:
+            continue
+        # ``split(",", 2)`` caps the allocation on an attacker-controlled
+        # header: we only need 0 / 1 / ≥ 2 non-blank pieces, and a third
+        # entry already trips the cardinality guard below.
+        for part in trimmed.split(",", 2):
+            piece = part.strip()
+            if piece:
+                filtered.append(piece)
+    if len(filtered) > 1:
+        raise DPoPMultipleProofsError(
+            f"request carries {len(filtered)} DPoP proofs (RFC 9449 §4.3 forbids it)"
+        )
+    return filtered[0] if filtered else None
 
 
 def raw_request_path(request: _RequestLike) -> str:

authplane/cache.py

@@ -11,9 +11,17 @@ class CacheEntry:
 
     access_token: str
     token_type: str
-    expires_in: int
+    # Mirrors the wire field's tri-state (``None`` = AS omitted it,
+    # positive int = AS-issued lifetime). ``0`` never reaches the cache —
+    # :meth:`TokenCache.set` refuses it — so callers reading from cache
+    # see a sentinel that round-trips to the AS-issued shape.
+    expires_in: int | None
     scope: str
     expires_at: float  # monotonic time
+    # DPoP key thumbprint (RFC 9449 §6.1). Persisted so a sender-constrained
+    # token retrieved from cache reports its binding to downstream callers
+    # instead of silently degrading to a bearer-only shape.
+    cnf_jkt: str = ""
 
 
 class TokenCache:
@@ -90,11 +98,30 @@ def set(
         key: str,
         access_token: str,
         token_type: str,
-        expires_in: int = 0,
+        expires_in: int | None = None,
         scope: str = "",
+        cnf_jkt: str = "",
     ) -> None:
-        """Cache a token. Skips caching if effective TTL <= 0."""
-        ttl = (expires_in if expires_in > 0 else self._default_ttl) - self._ttl_buffer
+        """Cache a token.
+
+        ``expires_in`` is tri-state:
+
+        * ``None`` ⇒ the AS omitted ``expires_in``; apply ``default_ttl``.
+        * ``0`` ⇒ RFC 6749 §5.1 explicit zero (one-shot, born-expired);
+          refuse to store so the next ``get`` is a miss instead of a
+          stale hit.
+        * positive ``n`` ⇒ honor ``n`` seconds.
+
+        The effective TTL (chosen value minus ``ttl_buffer_seconds``) must
+        be > 0; otherwise the entry is not stored.
+        """
+        if expires_in is None:
+            base_ttl = self._default_ttl
+        elif expires_in == 0:
+            return
+        else:
+            base_ttl = expires_in
+        ttl = base_ttl - self._ttl_buffer
         if ttl <= 0:
             return
         # `move_to_end` after insertion in case the key already exists —
@@ -106,6 +133,7 @@ def set(
             expires_in=expires_in,
             scope=scope,
             expires_at=time.monotonic() + ttl,
+            cnf_jkt=cnf_jkt,
         )
         self._entries.move_to_end(key)
         # Evict the LRU victim(s) until we're at-or-under the cap. The

authplane/client.py

@@ -308,6 +308,7 @@ async def client_credentials(
                 token_type=cached.token_type,
                 expires_in=cached.expires_in,
                 scope=cached.scope,
+                cnf_jkt=cached.cnf_jkt,
             )
 
         token_endpoint = await self._get_token_endpoint()
@@ -327,6 +328,7 @@ async def client_credentials(
                 result.token_type,
                 result.expires_in,
                 result.scope,
+                cnf_jkt=result.cnf_jkt,
             )
             return result
         except Exception as exc:

authplane/errors.py

@@ -122,6 +122,19 @@ class InvalidDPoPProofError(DPoPError):
     pass
 
 
+class DPoPMultipleProofsError(InvalidDPoPProofError):
+    """Raised when the inbound request carries more than one ``DPoP`` header
+    value (RFC 9449 §4.3 #1).
+
+    Surfaces with WWW-Authenticate ``error="invalid_dpop_proof"`` per
+    RFC 9449 §7.1. The other ``DPoPError`` subclasses keep the SDK's
+    historical ``invalid_token`` mapping; only this §4.3 cardinality
+    violation gets the proof-specific error code.
+    """
+
+    pass
+
+
 class DPoPReplayDetectedError(DPoPError):
     """Raised when a DPoP proof `jti` has already been seen."""
 
@@ -253,8 +266,11 @@ def www_authenticate(
 
     Maps SDK errors to the correct error code and authentication scheme:
     - ``InsufficientScopeError`` → ``insufficient_scope``
-    - ``DPoPError`` subclasses (except ``DPoPNotSupportedError``) → ``DPoP``
-      scheme with ``invalid_token``
+    - ``DPoPMultipleProofsError`` → ``DPoP`` scheme with
+      ``invalid_dpop_proof`` (RFC 9449 §7.1 prescribes this code for §4.3
+      cardinality rejections).
+    - Other ``DPoPError`` subclasses (except ``DPoPNotSupportedError``) →
+      ``DPoP`` scheme with ``invalid_token``
     - All other ``AuthplaneError`` → ``Bearer`` scheme with ``invalid_token``
 
     If ``scope`` is provided (or the error is an :class:`InsufficientScopeError`
@@ -272,6 +288,12 @@ def www_authenticate(
     """
     if isinstance(error, InsufficientScopeError):
         error_code = "insufficient_scope"
+    elif isinstance(error, DPoPMultipleProofsError):
+        # RFC 9449 §7.1 prescribes `invalid_dpop_proof` for §4.3
+        # cardinality rejections, not the SDK's historical `invalid_token`
+        # used by the other `DPoPError` shapes. Scoped to this error;
+        # a broader sweep is a separate change.
+        error_code = "invalid_dpop_proof"
     else:
         error_code = "invalid_token"
 

authplane/oauth/parsing.py

@@ -13,9 +13,14 @@ def _required_string(data: dict[str, Any], key: str) -> str:
     return value
 
 
-def _optional_int(data: dict[str, Any], key: str, *, default: int = 0) -> int:
-    value = data.get(key, default)
-    if value in ("", None):
+def _optional_int(data: dict[str, Any], key: str, *, default: int | None = 0) -> int | None:
+    """Parse an optional integer field. ``default`` is returned for absent
+    or empty-string values; ``None`` is a valid default so callers can
+    distinguish "field omitted on the wire" from "field present and zero".
+    """
+    sentinel = object()
+    value = data.get(key, sentinel)
+    if value is sentinel or value in ("", None):
         return default
     try:
         parsed = int(value)
@@ -74,7 +79,7 @@ def parse_token_response(
     return TokenResponse(
         access_token=access_token,
         token_type=token_type,
-        expires_in=_optional_int(data, "expires_in", default=0),
+        expires_in=_optional_int(data, "expires_in", default=None),
         scope=str(data.get("scope", "")),
         refresh_token=str(data.get("refresh_token", "")),
         issued_token_type=issued_token_type,

authplane/oauth/types.py

@@ -57,7 +57,12 @@ class TokenResponse:
 
     access_token: str
     token_type: str
-    expires_in: int
+    # ``expires_in`` is tri-state so the wire shape ``expires_in: 0``
+    # (RFC 6749 §5.1 — a deliberately-expired one-shot token) is
+    # distinguishable from the field being absent. Cache callers honor
+    # the AS's intent: ``None`` ⇒ apply the default TTL; ``0`` ⇒ refuse
+    # to store.
+    expires_in: int | None
     scope: str
     refresh_token: str = ""
     issued_token_type: str = ""