feat: enforce inbound DPoP in MCP adapters, bound TokenCache, add require_scopes (#12)

Adapters (authplane-mcp, authplane-fastmcp):
- Forward a DPoPRequestContext from verify_token to AuthplaneResource.verify
  so inbound_dpop=InboundDPoPOptions(required=True) enforces the proof check
  end-to-end. htu origin is always the operator-configured resource URI, never
  the inbound Host / X-Forwarded-Proto headers.
- Reconstruct htu from scope["raw_path"] to preserve percent-encoding (e.g.
  %2F) on the wire under ASGI, falling back to request.url.path.
- authplane-mcp adds AuthplaneRequestContextMiddleware, get_current_request(),
  and install_request_context(mcp) (idempotent) to publish the active request
  on a ContextVar so the verifier can build a DPoPRequestContext.
- Cache the in-flight verify task per request on request.state so a repeat
  verify_token within the same request reuses it instead of re-entering the
  inbound DPoP replay store. Cross-request replay protection is unaffected.

Core (authplane):
- Bound TokenCache with a configurable max_entries cap (default 10_000,
  exposed as TokenCache.DEFAULT_MAX_ENTRIES) and LRU eviction; plumbed
  through AuthplaneClient.create(cache_max_entries=...).
- Add VerifiedClaims.require_scopes(scopes) plural AND-style scope helper.
  Empty input is a no-op; raises InsufficientScopeError listing all missing
  scopes and the token's available scopes on failure.
- Export TokenCache from authplane.__init__.
- require_scope (singular) now renders an empty scope set as (none) instead
  of [], consistent with the plural helper.

Docs and demos run adapter setup, the async server entry point, and aclose()
in a single asyncio.run(main()), keeping the client's locks, HTTP pool, and
background refresh tasks on one event loop.

Author: muralx

CHANGELOG.md

@@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- `TokenCache` is now bounded by a configurable `max_entries` cap (default `10_000`, exposed as `TokenCache.DEFAULT_MAX_ENTRIES` and a read-only `cache.max_entries` property) and evicts the least-recently-used entry on overflow; both `get` and `set` bump the touched key to MRU. Plumbed through `AuthplaneClient.create(cache_max_entries=...)`. Token-exchange cache keys are high-cardinality (the subject token is part of the key), so the cap keeps long-lived clients bounded.
+- `VerifiedClaims.require_scopes(scopes: Iterable[str])` — plural AND-style helper that requires all listed scopes. Empty input is a no-op; on failure the raised `InsufficientScopeError` carries the full requested tuple on `required_scopes` and names every missing scope plus the token's available scopes in the message.
+- `authplane-mcp`: new public surface — `AuthplaneRequestContextMiddleware`, `get_current_request()`, `install_request_context(mcp)` — an ASGI middleware that publishes the active request on a `ContextVar` so the verifier can build a `DPoPRequestContext`.
+- `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
+- `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.
+
 ## [0.2.0] - 2026-05-20
 
 ### Security

authplane-fastmcp/authplane_fastmcp/auth.py

@@ -112,6 +112,7 @@ async def authplane_auth(
     metadata_refresh_seconds: int | None = None,
     cache_ttl_buffer_seconds: float | None = None,
     default_ttl_seconds: float | None = None,
+    cache_max_entries: int | None = None,
     circuit_breaker_threshold: int | None = None,
     circuit_breaker_cooldown_seconds: float | None = None,
     clock_skew_seconds: int | None = None,
@@ -158,6 +159,11 @@ async def authplane_auth(
             before cache expiry (default ``30.0``).
         default_ttl_seconds: Fallback token cache TTL used when token
             responses do not include expiry metadata (default ``3600.0``).
+        cache_max_entries: Maximum number of tokens kept in the LRU cache
+            before the least-recently-used entry is evicted (default
+            :attr:`TokenCache.DEFAULT_MAX_ENTRIES` = 10 000). Token-exchange
+            keys are high-cardinality; raise this cap for long-lived servers
+            with many distinct subjects.
         circuit_breaker_threshold: Number of transient failures before
             opening the AS circuit breaker (default ``5``).
         circuit_breaker_cooldown_seconds: Cooldown before allowing a
@@ -228,6 +234,7 @@ async def authplane_auth(
         "metadata_refresh_seconds": metadata_refresh_seconds,
         "cache_ttl_buffer_seconds": cache_ttl_buffer_seconds,
         "default_ttl_seconds": default_ttl_seconds,
+        "cache_max_entries": cache_max_entries,
         "circuit_breaker_threshold": circuit_breaker_threshold,
         "circuit_breaker_cooldown_seconds": circuit_breaker_cooldown_seconds,
     }

authplane-fastmcp/authplane_fastmcp/verifier.py

@@ -5,15 +5,29 @@
 to FastMCP's ``AccessToken`` with the full JWT payload in ``claims``.
 """
 
+import asyncio
 import logging
+from collections.abc import Callable
 from typing import Any, cast
-
-from authplane import AuthplaneError, AuthplaneResource
+from urllib.parse import urlsplit
+
+from authplane import AuthplaneError, AuthplaneResource, DPoPRequestContext
+from authplane._dpop_adapter import (
+    BuiltDPoPRequestContext,
+    get_or_create_verify_cache,
+    raw_request_path,
+    read_dpop_header,
+)
 from fastmcp.server.auth import AccessToken, TokenVerifier
+from fastmcp.server.dependencies import get_http_request as _default_get_http_request
+from starlette.requests import Request
 
 logger = logging.getLogger(__name__)
 
 
+__all__ = ["AuthplaneTokenVerifier"]
+
+
 class AuthplaneTokenVerifier(TokenVerifier):
     """FastMCP TokenVerifier backed by AuthplaneResource.
 
@@ -23,9 +37,39 @@ class AuthplaneTokenVerifier(TokenVerifier):
     tool handlers via FastMCP's native ``CurrentAccessToken()`` dependency
     or ``get_access_token()`` function.
 
-    All security-critical logic (signature verification, claim validation,
-    JWKS caching, SSRF protection) is handled by the core SDK. This class
-    is a thin adapter that maps between the two interfaces.
+    DPoP (RFC 9449)
+    ---------------
+
+    When the underlying :class:`AuthplaneResource` was created with
+    ``inbound_dpop=InboundDPoPOptions(...)``, this verifier pulls the
+    active HTTP request via
+    :func:`fastmcp.server.dependencies.get_http_request`, builds a
+    :class:`~authplane.DPoPRequestContext` (method + reconstructed
+    ``htu`` + proof header), and forwards it to
+    :meth:`AuthplaneResource.verify`. The ``htu`` origin
+    (scheme + host + port) is taken from the operator-configured
+    resource URI, never from the inbound ``Host`` /
+    ``X-Forwarded-Proto`` headers — letting an upstream decide which
+    ``htu`` the proof is checked against would neuter DPoP's
+    cross-endpoint anti-replay. Only the path varies per call.
+
+    Per-request verify cache
+    ------------------------
+
+    FastMCP's standard HTTP stack invokes ``verify_token`` exactly once
+    per request (Starlette ``AuthenticationMiddleware`` →
+    ``BearerAuthBackend`` → ``TokenVerifier.verify_token``). The first
+    call's in-flight verify task is stashed on ``request.state`` keyed by
+    the access token; any subsequent invocation within the same request
+    awaits the same task instead of re-entering the inbound DPoP replay
+    store. The cache is defensive: it mirrors the TS adapter's
+    ``AsyncLocalStorage`` pattern and pre-empts a class of regressions
+    where a future framework change (transport rewrite, custom auth
+    provider, ASGI wrapper) would silently double-call ``verify_token``
+    and the second call's proof would be rejected as
+    ``DPoPReplayDetected``. Different requests get distinct
+    ``request.state`` objects so cross-request replay protection is
+    preserved.
 
     Scope enforcement is FastMCP's responsibility via
     ``@mcp.tool(auth=require_scopes(...))``.
@@ -36,6 +80,8 @@ def __init__(
         verifier: AuthplaneResource,
         base_url: str | None = None,
         required_scopes: list[str] | None = None,
+        *,
+        get_http_request: Callable[[], Request] | None = None,
     ) -> None:
         """Initialize the token verifier.
 
@@ -47,9 +93,26 @@ def __init__(
                 ``TokenVerifier`` for PRM generation.
             required_scopes: Scopes required for all requests. Passed to
                 the parent ``TokenVerifier``.
+            get_http_request: Override for the active-request lookup
+                (defaults to
+                ``fastmcp.server.dependencies.get_http_request``). Tests
+                inject a fake to drive the DPoP / per-request-cache
+                paths without spinning up an ASGI app.
         """
         super().__init__(base_url=base_url, required_scopes=required_scopes)
         self._verifier = verifier
+        self._get_http_request = get_http_request or _default_get_http_request
+
+        # ``AuthplaneResource.resource`` is operator-configured and must be a
+        # string URI — guard against mis-wired mocks (a bare ``MagicMock`` with
+        # no ``resource`` set silently produces ``MagicMock://MagicMock`` here
+        # and would corrupt ``htu`` reconstruction in production).
+        if not isinstance(verifier.resource, str):
+            raise TypeError(
+                f"verifier.resource must be a str URI, got {type(verifier.resource).__name__}"
+            )
+        split = urlsplit(verifier.resource)
+        self._resource_origin = f"{split.scheme}://{split.netloc}"
 
     @property
     def verifier(self) -> AuthplaneResource:
@@ -67,25 +130,57 @@ def scopes_supported(self) -> list[str]:
         return list(self._verifier.scopes)
 
     async def verify_token(self, token: str) -> AccessToken | None:
-        """Validate a JWT and return a FastMCP AccessToken.
+        """Validate a JWT and return a FastMCP ``AccessToken``.
 
-        Called by FastMCP once per authenticated request. Delegates to
-        ``AuthplaneResource.verify()`` for all validation, then maps the
-        resulting ``VerifiedClaims`` to a FastMCP ``AccessToken`` with the
-        full JWT payload in ``claims``.
+        Pulls the active HTTP request to build a per-request
+        :class:`DPoPRequestContext` (RFC 9449 §7.1) and to scope the
+        verify-task cache. The cache hit path re-awaits the original
+        :class:`asyncio.Task`, so a cached :class:`AuthplaneError`
+        re-raises with the same type on every call within the request —
+        the ``AuthplaneError → None`` translation happens once here, not
+        inside the cached coroutine, which keeps the cached failure
+        diagnosable.
 
         Args:
-            token: The raw JWT string (FastMCP strips ``'Bearer '``
-                before calling this method).
+            token: The raw JWT string (FastMCP strips the ``Bearer ``
+                prefix before calling this method).
 
         Returns:
-            ``AccessToken`` on successful validation with ``token``,
-            ``client_id``, ``scopes``, ``expires_at``, and ``claims``
-            (full JWT payload dict) fields populated. Returns ``None``
-            on any validation failure (FastMCP responds with 401).
+            ``AccessToken`` on successful validation. ``None`` on any
+            ``AuthplaneError`` (FastMCP responds with 401).
         """
         try:
-            claims = await self._verifier.verify(token)
+            request: Request | None = self._get_http_request()
+        except RuntimeError as exc:
+            # ``fastmcp.server.dependencies.get_http_request`` raises a bare
+            # ``RuntimeError("No active HTTP request found.")`` when called
+            # outside an HTTP request context (unit tests, background tasks
+            # with no snapshotted request). Match the message to avoid
+            # silently degrading to ``dpop_request=None`` if a future
+            # FastMCP release surfaces an unrelated ``RuntimeError`` from
+            # this dependency — that would re-introduce the silent-pass
+            # this PR fixed (PRM advertising DPoP-required while the
+            # verifier never sees a proof). Drop this narrow when the
+            # upstream public surface adopts a typed exception.
+            if "No active HTTP request" not in str(exc):
+                raise
+            request = None
+
+        try:
+            if request is None:
+                claims = await self._verifier.verify(token, dpop_request=None)
+            else:
+                cache = get_or_create_verify_cache(request)
+                task = cache.get(token)
+                if task is None:
+                    # No await between cache miss and cache write — concurrent
+                    # verify_token(token) calls on the same loop cannot race.
+                    dpop_request = self._build_dpop_request_context(request)
+                    task = asyncio.create_task(
+                        self._verifier.verify(token, dpop_request=dpop_request)
+                    )
+                    cache[token] = task
+                claims = await task
         except AuthplaneError as error:
             logger.debug(
                 "authplane.token_verification_failed",
@@ -100,3 +195,36 @@ async def verify_token(self, token: str) -> AccessToken | None:
             expires_at=claims.expires_at,
             claims=cast("dict[str, Any]", claims.raw),  # full JWT payload
         )
+
+    def _build_dpop_request_context(self, request: Request) -> DPoPRequestContext:
+        """Build the per-request DPoP context.
+
+        Always returns a context — the resource verifier inspects the
+        access token's ``cnf`` claim plus the context's ``proof`` to
+        decide whether DPoP enforcement applies. When the resource is
+        not configured for inbound DPoP, the verifier's Mode-3 path
+        rejects any DPoP signal regardless of what is passed here.
+
+        Cross-SDK note: the TS sibling ``buildDpopRequestContext``
+        returns ``undefined`` when no ``DPoP`` header is present;
+        Python intentionally always builds the context with
+        ``proof=None``. Both shapes are behaviorally equivalent in
+        the core verifier (Mode 3 path treats absent and ``None``
+        proofs the same), but a DPoP-bound token with no proof
+        yields a more specific ``DPoPProofMissingError`` here
+        instead of ``DPoPBindingMismatchError``. The error-type
+        contract is pinned per language by design.
+        """
+        # ``raw_request_path`` reads ``scope["raw_path"]`` to preserve
+        # percent-encoding for DPoP ``htu`` parity with the TS sibling.
+        # ``request.url.query`` is sourced from ``scope["query_string"]``
+        # without percent-decoding, so it is already on-wire-safe.
+        url = f"{self._resource_origin}{raw_request_path(request)}"
+        query = request.url.query
+        if query:
+            url = f"{url}?{query}"
+        return BuiltDPoPRequestContext(
+            method=request.method.upper(),
+            url=url,
+            proof=read_dpop_header(request),
+        )

authplane-fastmcp/demo/mcpserver.py

@@ -16,11 +16,8 @@
 
 from authplane_fastmcp import authplane_auth
 
-if __name__ == "__main__":
-    logging.basicConfig(
-        level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)s %(message)s"
-    )
 
+async def main() -> None:
     load_dotenv(os.path.join(os.path.dirname(__file__), ".env"))
 
     resource = os.environ.get("RESOURCE_URL", "http://localhost:8080/mcp")
@@ -32,18 +29,16 @@
     client_id = os.environ.get("CLIENT_ID", resource)
     client_secret = os.environ["CLIENT_SECRET"]
 
-    auth_result = asyncio.run(
-        authplane_auth(
-            issuer=issuer,
-            base_url=base_url,
-            scopes=["tools/add", "tools/multiply"],
-            dev_mode=True,  # Enables local testing
-            as_credentials=ASCredentials(
-                client_id=client_id,
-                client_secret=client_secret,
-            ),
-            revocation_checker=IntrospectionRevocation(),
-        )
+    auth_result = await authplane_auth(
+        issuer=issuer,
+        base_url=base_url,
+        scopes=["tools/add", "tools/multiply"],
+        dev_mode=True,  # Enables local testing
+        as_credentials=ASCredentials(
+            client_id=client_id,
+            client_secret=client_secret,
+        ),
+        revocation_checker=IntrospectionRevocation(),
     )
 
     mcp = FastMCP("Calculator Service", **auth_result)
@@ -67,4 +62,19 @@ def multiply(a: float, b: float) -> float:
     # equivalent demo in ``authplane-mcp/demo/mcpserver.py``, which uses the
     # low-level MCP server and surfaces the elicitation correctly.
 
-    mcp.run(transport="http", port=port, log_level="DEBUG")
+    # The adapter setup, server, and aclose() must share one event loop —
+    # auth_result holds async resources (locks, httpx pool, background JWKS
+    # refresh tasks) bound to the running loop. ``run_async`` is FastMCP's
+    # async entry point and keeps everything on the same loop.
+    try:
+        await mcp.run_async(transport="http", port=port, log_level="DEBUG")
+    finally:
+        await auth_result.aclose()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)s %(message)s"
+    )
+
+    asyncio.run(main())

authplane-fastmcp/tests/conftest.py

@@ -63,8 +63,12 @@ def mock_verifier(valid_claims: VerifiedClaims) -> AsyncMock:
 
     mock = AsyncMock(spec=AuthplaneResource)
     type(mock).scopes = PropertyMock(return_value=["tools/query", "tools/write"])
+    type(mock).resource = PropertyMock(return_value="https://api.example.com/mcp")
 
-    async def verify_side_effect(token: str) -> VerifiedClaims:
+    async def verify_side_effect(
+        token: str, *, dpop_request: object | None = None
+    ) -> VerifiedClaims:
+        _ = dpop_request  # accept but ignore — covered by dedicated DPoP tests
         if token == "valid_token":
             return valid_claims
         raise AuthplaneError("Invalid token")