diff --git a/docs/development/v4-notes/change-register.mdx b/docs/development/v4-notes/change-register.mdx
index 7bd6c8b64..2a9aa3bd7 100644
--- a/docs/development/v4-notes/change-register.mdx
+++ b/docs/development/v4-notes/change-register.mdx
@@ -341,6 +341,12 @@ FastMCP keeps its own DCR redirect-URI hardening (PRs #4419, #4408) regardless o
*Verify:* recent commits `67527c1f` (block unsafe OAuth redirect schemes), `57a27992` (DNS rebinding), `cccb529f` (DCR redirect URI validation) on `main`.
+### Identity assertion (SEP-990 ID-JAG) — Added (beta)
+
+`OAuthProxy` (and `OIDCProxy`, which inherits it) accepts an optional `identity_assertion=IdentityAssertion(trusted_issuers=[...])`. When configured, the token endpoint accepts the RFC 7523 `urn:ietf:params:oauth:grant-type:jwt-bearer` grant carrying an enterprise IdP-issued ID-JAG, validates it (signature against the trusted issuer's JWKS, `iss`/`aud`/`exp`, `typ` of `oauth-id-jag+jwt`, mandatory `sub`, signed `client_id`/`resource` binding, and `jti` replay rejection), and mints a short-lived FastMCP access token carrying the asserted subject with no refresh token. Authorization server metadata advertises the `jwt-bearer` grant type and the `urn:ietf:params:oauth:grant-profile:id-jag` profile when enabled. This is server-side only; the client-side wrapper ships separately. See [Identity Assertion](/servers/auth/oauth-proxy#identity-assertion-sep-990).
+
+*Verify:* `fastmcp_slim/fastmcp/server/auth/identity_assertion.py`, the `exchange_identity_assertion` and `get_routes` changes in `fastmcp_slim/fastmcp/server/auth/oauth_proxy/proxy.py`, and the jwt-bearer dispatch in `fastmcp_slim/fastmcp/server/auth/auth.py` (`TokenHandler._maybe_handle_id_jag`).
+
### Templated resource parameters are path-screened by default — Breaking (behavior)
Every templated resource now has its extracted parameter values screened for path-traversal (`..` segments), absolute paths, and null bytes **before the handler runs** — on by default, at the server's read chokepoint, covering local and provider-sourced (mounted/proxied) templates alike. Previously these payloads reached handlers raw; a template whose parameter flowed into a filesystem path or upstream URL was exposed unless the author added their own check. A rejected read now surfaces a non-leaky "resource not found" error (`-32602`) and a debug log.
diff --git a/docs/development/v4-notes/index.mdx b/docs/development/v4-notes/index.mdx
index 1cbebfdcd..4a5959545 100644
--- a/docs/development/v4-notes/index.mdx
+++ b/docs/development/v4-notes/index.mdx
@@ -5,7 +5,7 @@ title: v4.0 Development Notes
This directory is the working map of FastMCP v4.0: the complete register of user-facing changes from the MCP Python SDK v2 migration ([PR #4437](https://github.com/PrefectHQ/fastmcp/pull/4437)), plus the forward v4 feature program. It plays three roles at once.
1. **A change register.** Every user-visible change from the migration, organized by subsystem, with a note on how FastMCP handles it (absorbed, bridged, breaking, or deprecated) and where to find it in the diff. This is the [Change Register](/development/v4-notes/change-register).
-2. **A feature program.** The forward v4 work — sampling removal, MRTR elicitation, the first-class 2026 client, and the SDK-delegation round-two convergence — each with an explicit status. This is the [Feature Program](/development/v4-notes/feature-program).
+2. **A feature program.** The forward v4 work — sampling removal, MRTR elicitation, the first-class 2026 client, and the SDK-delegation round-two convergence — each with an explicit status. This is the [Feature Program](/development/v4-notes/feature-program). The shipped side of that program — what a v4 deployment provides on the modern protocol today, including the complete server-side SEP-990 identity assertion implementation — is cataloged in [2026-07-28 Protocol Support](/development/v4-notes/protocol-2026).
3. **A review lens.** Because the migration PR is too large to review line by line, the change register is organized so a reviewer can take one subsystem, read its claimed changes, and verify each against the diff. The [Known Gaps](/development/v4-notes/known-gaps) page collects the deliberate xfails and the upstream dependencies that gate the follow-up work.
## Why v4 exists
diff --git a/docs/development/v4-notes/protocol-2026.mdx b/docs/development/v4-notes/protocol-2026.mdx
new file mode 100644
index 000000000..709e0fd0d
--- /dev/null
+++ b/docs/development/v4-notes/protocol-2026.mdx
@@ -0,0 +1,52 @@
+---
+title: 2026-07-28 Protocol Support
+---
+
+FastMCP v4 serves the sessionless `2026-07-28` protocol era and the session-based handshake eras from a single server, with per-connection auto-detection. This page catalogs what FastMCP provides for the modern era — both the protocol machinery it inherits from the MCP Python SDK and the capabilities FastMCP implements itself on top of that layer. It is the reference for what a v4 deployment can actually do on the modern protocol today.
+
+## Identity assertion (SEP-990): a complete server-side implementation
+
+SEP-990 defines enterprise "on-behalf-of" access: a corporate identity provider (Okta, Microsoft Entra, etc.) issues a signed *ID-JAG* asserting an employee's identity, the employee's agent presents it at the MCP authorization server's token endpoint via the RFC 7523 `jwt-bearer` grant, and receives a short-lived access token — no browser login, no per-user consent screen, and revocation lives at the IdP.
+
+The protocol layer for this flow — grant parsing, the `exchange_identity_assertion` provider hook, and metadata advertisement — comes from the SDK. The validation and issuance logic that makes the flow actually work is FastMCP's implementation, and enabling it is one parameter on the existing auth providers:
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.auth import OAuthProxy, IdentityAssertion
+
+auth = OAuthProxy(
+ ..., # existing upstream configuration unchanged
+ identity_assertion=IdentityAssertion(
+ trusted_issuers=["https://login.acme-corp.com"],
+ ),
+)
+mcp = FastMCP("Internal API", auth=auth)
+```
+
+Behind that one parameter, FastMCP performs the full SEP-990 §5.1 / RFC 7523 §3 processing: JWKS-based signature verification with automatic OIDC discovery of issuer keys, `typ`/`iss`/`aud`/`sub` validation, temporal checks (`exp`, `iat`, `nbf`, maximum assertion lifetime), enforcement of the assertion's signed `client_id` and `resource` bindings, `jti` replay rejection, scope derivation from the signed assertion (client requests can narrow but never widen), short-lived token issuance with no refresh token, and revocation tracking for the issued tokens. The asserted subject flows into the normal FastMCP auth context, so tools read it through `get_access_token()` like any other identity. See [Identity Assertion](/servers/auth/oauth-proxy#identity-assertion-sep-990) for the full documentation.
+
+This slots into FastMCP's existing authorization-server stack — the OAuth proxy's dynamic client registration, the consent flow, and self-issued JWTs — which is what makes a one-parameter enterprise deployment possible.
+
+## Modern-era capability inventory
+
+The complete picture of what a FastMCP v4 server and client provide on the `2026-07-28` era:
+
+| Capability | What FastMCP provides |
+| --- | --- |
+| **Dual-era serving** | One server answers both `server/discover` (modern, sessionless) and `initialize` (handshake) connections, auto-detected per connection. Any replica behind a plain load balancer can answer a modern request. |
+| **Identity assertion (SEP-990)** | Complete server-side implementation, one parameter to enable (above). |
+| **Authorization server** | Full AS stack: `OAuthProxy` bridges DCR-expecting MCP clients to non-DCR enterprise IdPs, ~18 built-in providers, consent UI, self-issued JWTs, protected-resource metadata (RFC 9728). |
+| **Cache hints (SEP-2549)** | Server-level authoring (`FastMCP(cache_ttl=..., cache_scope=...)`) stamps every cacheable result; the FastMCP client honors hints with an opt-in response cache. |
+| **Distributed response caching** | `KeyValueResponseCacheStore` backs the client cache with any key-value store (Redis, memory, filetree), so a fleet of clients or proxy replicas shares cache fills across processes. |
+| **Resource path security** | Templated resource parameters are screened for traversal, absolute paths, and null bytes before handlers run — on by default, including provider-sourced and mounted templates. |
+| **Client protocol negotiation** | `Client(mode="auto")` probes `server/discover` and falls back to the classic handshake; the client answers multi-round-trip `input_required` requests through its existing handlers. |
+| **Spec-standard errors (SEP-2164)** | Missing-resource reads return `-32602`; push-feature calls on modern connections fail with clear era-specific errors rather than generic method-not-found. |
+| **Background tasks** | `@mcp.tool(task=True)` runs on a Redis-backed distributed runtime (Docket) with cross-replica notifications — execution infrastructure that is FastMCP's own, independent of the protocol-era task surface. |
+| **Middleware** | Typed per-method hooks (`on_call_tool`, `on_list_tools`, …) and a suite of built-ins (auth, rate limiting, caching, error handling, logging, timing, and more). |
+| **Composition** | `mount()`, providers, proxying, and tool transforms compose servers dynamically at runtime, with lifespans and middleware driven through the SDK session manager. |
+| **Pagination** | Declarative `FastMCP(list_page_size=...)` paginates all list operations in the high-level server; the client auto-paginates with cycle detection. |
+| **Telemetry** | OpenTelemetry spans on by default (no-op without an exporter), SDK-aligned attributes (`mcp.method.name`, `mcp.protocol.version`, `gen_ai.*`), plus auth and provider-delegation spans; `FASTMCP_ENABLE_TELEMETRY=false` disables cleanly. |
+
+## Still in the program
+
+Two modern-era surfaces are deliberately staged rather than shipped, tracked in the [Feature Program](/development/v4-notes/feature-program): server-side multi-round-trip elicitation (the replacement for push-based `ctx.elicit` on modern connections — the client half already answers `input_required` requests) and the unified `subscriptions/listen` stream. The [Known Gaps](/development/v4-notes/known-gaps) page tracks the upstream dependencies that gate them.
diff --git a/docs/docs.json b/docs/docs.json
index 4eb3cd51b..17e2abc3c 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -376,6 +376,7 @@
"development/v4-notes/index",
"development/v4-notes/change-register",
"development/v4-notes/feature-program",
+ "development/v4-notes/protocol-2026",
"development/v4-notes/known-gaps"
]
}
diff --git a/docs/servers/auth/oauth-proxy.mdx b/docs/servers/auth/oauth-proxy.mdx
index e35f244eb..c65ba20f5 100644
--- a/docs/servers/auth/oauth-proxy.mdx
+++ b/docs/servers/auth/oauth-proxy.mdx
@@ -653,6 +653,83 @@ auth = OAuthProxy(
)
```
+## Identity Assertion (SEP-990)
+
+
+
+
+Identity assertion is a beta feature. The API may change in a future release.
+
+
+Identity assertion enables an enterprise "on-behalf-of" flow. A corporate identity provider (Okta, Microsoft Entra, etc.) issues an *ID-JAG* — a signed JWT that asserts an employee's identity to a specific MCP authorization server. The client presents that ID-JAG at the token endpoint using the RFC 7523 `jwt-bearer` grant, and the proxy validates it and mints a short-lived access token for the asserted user. No refresh token is issued: the identity provider controls session lifetime, and the client re-exchanges a fresh ID-JAG when its access token expires. This lets a workforce reach your MCP server with corporate-managed identity and centralized revocation, without each user running an interactive browser login.
+
+To enable it, pass an `IdentityAssertion` configuration listing the issuers you trust:
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.auth import OAuthProxy, IdentityAssertion
+
+auth = OAuthProxy(
+ upstream_authorization_endpoint="https://accounts.example.com/authorize",
+ upstream_token_endpoint="https://accounts.example.com/token",
+ upstream_client_id="your-client-id",
+ upstream_client_secret="your-client-secret",
+ base_url="https://your-server.com",
+ identity_assertion=IdentityAssertion(
+ trusted_issuers=["https://login.acme-corp.com"],
+ ),
+)
+
+mcp = FastMCP("Internal API", auth=auth)
+
+@mcp.tool
+def whoami() -> str:
+ from fastmcp.server.dependencies import get_access_token
+
+ token = get_access_token()
+ return token.subject or "unknown"
+```
+
+When identity assertion is configured, the proxy advertises the `urn:ietf:params:oauth:grant-type:jwt-bearer` grant type and the `urn:ietf:params:oauth:grant-profile:id-jag` grant profile in its authorization server metadata, so compatible clients can discover the capability. When it is not configured, the grant is rejected as unsupported.
+
+### How Validation Works
+
+For each ID-JAG presented at the token endpoint, the proxy checks that:
+
+- the JOSE header `typ` is `oauth-id-jag+jwt`;
+- the `iss` claim is one of the configured `trusted_issuers`;
+- the signature verifies against the issuer's published keys;
+- the `aud` claim identifies this authorization server;
+- the signed `client_id` claim matches the client presenting the assertion — an assertion the IdP minted for one client cannot be redeemed by another;
+- the signed `resource` claim names this server — an assertion minted for a different MCP server behind the same IdP is rejected;
+- `exp` (and `iat`/`nbf`, when present) place the assertion within a short lifetime and its validity window; and
+- the `jti` has not been seen before, preventing replay.
+
+The issuer's signing keys are discovered automatically via OIDC (`{issuer}/.well-known/openid-configuration`). For issuers that do not publish a discovery document, provide the JWKS URI explicitly per issuer:
+
+```python
+identity_assertion=IdentityAssertion(
+ trusted_issuers=["https://login.acme-corp.com"],
+ jwks_uris={"https://login.acme-corp.com": "https://login.acme-corp.com/keys"},
+)
+```
+
+Verification assumes `RS256` unless the issuer signs with another algorithm, in which case set `algorithm` explicitly (any asymmetric JWS algorithm — `RS*`, `PS*`, or `ES*` — since assertions are verified against a published JWKS, not a shared secret). When trusted issuers use different algorithms, override per issuer with `algorithms`, keyed the same way as `jwks_uris`:
+
+```python
+identity_assertion=IdentityAssertion(
+ trusted_issuers=["https://login.acme-corp.com", "https://sso.other-corp.com"],
+ algorithm="ES256",
+ algorithms={"https://sso.other-corp.com": "RS256"},
+)
+```
+
+The subject asserted in the ID-JAG flows into the normal FastMCP auth context. Tools read it through `get_access_token()` exactly as they would for any other token, because the proxy issues the access token through its own token factory.
+
+
+Replay protection is per-process. Each server process tracks seen `jti` values in memory, so a horizontally-scaled deployment running multiple workers or replicas could accept the same assertion once per process. The same applies to revocation of ID-JAG access tokens: they are self-contained, so revocation is tracked in-process until the token's (short, 5-minute default) natural expiry. For deployments that require strict single-use enforcement across replicas, configure sticky routing so a given client's requests reach the same process, or place a shared store in front of the token endpoint. This mirrors the posture of CIMD `private_key_jwt` replay protection, which is also per-process.
+
+
## Security
### Key and Storage Management
diff --git a/fastmcp_slim/fastmcp/server/auth/__init__.py b/fastmcp_slim/fastmcp/server/auth/__init__.py
index cd6a300ad..67ee25915 100644
--- a/fastmcp_slim/fastmcp/server/auth/__init__.py
+++ b/fastmcp_slim/fastmcp/server/auth/__init__.py
@@ -17,6 +17,7 @@ from .authorization import (
)
if TYPE_CHECKING:
+ from .identity_assertion import IdentityAssertion as IdentityAssertion
from .oauth_proxy import OAuthProxy as OAuthProxy
from .oidc_proxy import OIDCProxy as OIDCProxy
from .providers.debug import DebugTokenVerifier as DebugTokenVerifier
@@ -44,6 +45,10 @@ def __getattr__(name: str) -> object:
from .providers.jwt import StaticTokenVerifier
return StaticTokenVerifier
+ if name == "IdentityAssertion":
+ from .identity_assertion import IdentityAssertion
+
+ return IdentityAssertion
if name == "OAuthProxy":
from .oauth_proxy import OAuthProxy
@@ -61,6 +66,7 @@ __all__ = [
"AuthContext",
"AuthProvider",
"DebugTokenVerifier",
+ "IdentityAssertion",
"JWTVerifier",
"MultiAuth",
"OAuthProvider",
diff --git a/fastmcp_slim/fastmcp/server/auth/auth.py b/fastmcp_slim/fastmcp/server/auth/auth.py
index dc203fbf4..0b3e6abfe 100644
--- a/fastmcp_slim/fastmcp/server/auth/auth.py
+++ b/fastmcp_slim/fastmcp/server/auth/auth.py
@@ -21,8 +21,10 @@ from mcp.server.auth.provider import (
)
from mcp.server.auth.provider import (
AuthorizationCode,
+ IdentityAssertionParams,
OAuthAuthorizationServerProvider,
RefreshToken,
+ TokenError,
)
from mcp.server.auth.provider import (
TokenVerifier as TokenVerifierProtocol,
@@ -36,7 +38,7 @@ from mcp.server.auth.settings import (
ClientRegistrationOptions,
RevocationOptions,
)
-from mcp.shared.auth import OAuthClientInformationFull
+from mcp.shared.auth import JWT_BEARER_GRANT_TYPE, OAuthClientInformationFull
from pydantic import AnyHttpUrl, Field
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
@@ -74,7 +76,22 @@ class TokenHandler(_SDKTokenHandler):
"""
async def handle(self, request: Any):
- """Wrap SDK handle() and transform auth error responses."""
+ """Wrap SDK handle() and transform auth error responses.
+
+ The SEP-990 jwt-bearer (ID-JAG) grant is dispatched here rather than by
+ the SDK. The SDK requires a confidential client (a stored client_secret)
+ before it will call `exchange_identity_assertion`. FastMCP OAuth-proxy
+ clients are always public (`token_endpoint_auth_method="none"`, no stored
+ secret), and in the proxy trust model the ID-JAG — validated against a
+ trusted issuer — is the authoritative grant, not a per-client secret.
+ So when identity assertion is enabled we authenticate the client and
+ dispatch the grant ourselves, without the SDK's confidential precondition.
+ """
+ if self.identity_assertion_enabled:
+ id_jag_response = await self._maybe_handle_id_jag(request)
+ if id_jag_response is not None:
+ return id_jag_response
+
response = await super().handle(request)
# Transform 401 unauthorized_client -> invalid_client
@@ -118,6 +135,80 @@ class TokenHandler(_SDKTokenHandler):
return response
+ async def _maybe_handle_id_jag(self, request: Request):
+ """Dispatch the SEP-990 jwt-bearer grant, or None to fall through.
+
+ Returns a response for the jwt-bearer grant (ID-JAG), or None when the
+ request is not a jwt-bearer grant so the SDK handler runs normally.
+ """
+ form_data = await request.form()
+ if form_data.get("grant_type") != JWT_BEARER_GRANT_TYPE:
+ return None
+
+ try:
+ client_info = await self.client_authenticator.authenticate_request(request)
+ except AuthenticationError as e:
+ return PydanticJSONResponse(
+ content=TokenErrorResponse(
+ error="invalid_client",
+ error_description=e.message,
+ ),
+ status_code=401,
+ headers={"Cache-Control": "no-store", "Pragma": "no-cache"},
+ )
+
+ # Dispatching the jwt-bearer grant ourselves bypasses the SDK's
+ # `grant_type not in client_info.grant_types` check, so enforce it here:
+ # a client may only use the ID-JAG grant if it registered for it. On the
+ # proxy, DCR adds this grant type to registered clients when identity
+ # assertion is enabled, so legitimately-registered clients are accepted
+ # while clients registered only for authorization_code/refresh_token are not.
+ if JWT_BEARER_GRANT_TYPE not in client_info.grant_types:
+ return self.response(
+ TokenErrorResponse(
+ error="unsupported_grant_type",
+ error_description=(
+ "Unsupported grant type (supported grant types are "
+ f"{client_info.grant_types})"
+ ),
+ )
+ )
+
+ assertion = form_data.get("assertion")
+ if not isinstance(assertion, str) or not assertion:
+ return self.response(
+ TokenErrorResponse(
+ error="invalid_request",
+ error_description="Missing assertion",
+ )
+ )
+
+ scope = form_data.get("scope")
+ resource = form_data.get("resource")
+ params = IdentityAssertionParams(
+ assertion=assertion,
+ scopes=scope.split(" ") if isinstance(scope, str) and scope else None,
+ resource=resource if isinstance(resource, str) else None,
+ )
+
+ try:
+ tokens = await self.provider.exchange_identity_assertion(
+ client_info, params
+ )
+ except TokenError as e:
+ # Per MCP spec, invalid/expired grants MUST return 401 (the SDK path is
+ # transformed the same way in handle()).
+ status_code = 401 if e.error == "invalid_grant" else 400
+ return PydanticJSONResponse(
+ content=TokenErrorResponse(
+ error=e.error, error_description=e.error_description
+ ),
+ status_code=status_code,
+ headers={"Cache-Control": "no-store", "Pragma": "no-cache"},
+ )
+
+ return self.response(tokens)
+
# Expected assertion type for private_key_jwt
JWT_BEARER_ASSERTION_TYPE = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
diff --git a/fastmcp_slim/fastmcp/server/auth/identity_assertion.py b/fastmcp_slim/fastmcp/server/auth/identity_assertion.py
new file mode 100644
index 000000000..b86eceb34
--- /dev/null
+++ b/fastmcp_slim/fastmcp/server/auth/identity_assertion.py
@@ -0,0 +1,531 @@
+"""Server-side identity assertion (ID-JAG) support for FastMCP (SEP-990).
+
+.. warning::
+ **Beta Feature**: Identity assertion support is currently in beta. The API
+ may change in future releases. Please report any issues you encounter.
+
+SEP-990 defines an enterprise "on-behalf-of" flow. A corporate identity provider
+(Okta, Entra, etc.) issues an *ID-JAG* (Identity Assertion JWT Authorization
+Grant) that asserts an employee's identity to a specific MCP authorization
+server. The client presents that ID-JAG at the token endpoint using the RFC 7523
+``urn:ietf:params:oauth:grant-type:jwt-bearer`` grant (the RFC 8693 token-exchange
+profile). This module validates the assertion and lets the authorization server
+mint a short-lived access token carrying the asserted subject, with no refresh
+token — the client re-exchanges a fresh ID-JAG instead, and revocation lives at
+the IdP.
+
+This module provides:
+
+- ``IdentityAssertion``: a small pydantic config model attached to ``OAuthProxy``
+ via the ``identity_assertion`` parameter.
+- ``IdentityAssertionValidator``: validates an ID-JAG per RFC 7523 §3 and the
+ SEP-990 processing rules, reusing FastMCP's :class:`JWTVerifier` for signature,
+ issuer, audience, and expiry checks, and enforcing ``typ``, ``sub`` presence,
+ and ``jti`` replay protection on top.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from typing import TYPE_CHECKING
+from urllib.parse import urlparse, urlunparse
+
+import httpx2
+from pydantic import BaseModel, Field, field_validator
+
+from fastmcp.utilities.auth import decode_jwt_header
+from fastmcp.utilities.logging import get_logger
+
+if TYPE_CHECKING:
+ from fastmcp.server.auth.providers.jwt import JWTVerifier
+
+logger = get_logger(__name__)
+
+#: RFC 7523 §2.1 authorization grant used to present the ID-JAG.
+JWT_BEARER_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:jwt-bearer"
+
+#: SEP-990 grant profile advertised in authorization server metadata.
+ID_JAG_GRANT_PROFILE = "urn:ietf:params:oauth:grant-profile:id-jag"
+
+#: SEP-990 §5.1: the ID-JAG's JOSE header ``typ`` MUST be this media type.
+ID_JAG_TYP = "oauth-id-jag+jwt"
+
+#: Asymmetric JWS algorithms JWTVerifier supports for JWKS-based verification.
+SUPPORTED_ASSERTION_ALGORITHMS = frozenset(
+ {"RS256", "RS384", "RS512", "PS256", "PS384", "PS512", "ES256", "ES384", "ES512"}
+)
+
+
+class IdentityAssertion(BaseModel):
+ """Configuration for server-side identity assertion (ID-JAG) support.
+
+ When attached to an :class:`~fastmcp.server.auth.oauth_proxy.OAuthProxy` via the
+ ``identity_assertion`` parameter, the proxy's token endpoint accepts the RFC 7523
+ ``jwt-bearer`` grant carrying an ID-JAG issued by one of the ``trusted_issuers``,
+ and mints a short-lived FastMCP access token for the asserted subject.
+
+ Example:
+ ```python
+ from fastmcp.server.auth import OAuthProxy, IdentityAssertion
+
+ auth = OAuthProxy(
+ ...,
+ identity_assertion=IdentityAssertion(
+ trusted_issuers=["https://login.acme-corp.com"],
+ ),
+ )
+ ```
+ """
+
+ trusted_issuers: list[str] = Field(
+ ...,
+ description=(
+ "Issuer (`iss`) values the authorization server accepts on an ID-JAG. "
+ "Each must exactly match the assertion's `iss` claim. For each issuer, "
+ "the JWKS used to verify the assertion signature is discovered via OIDC "
+ "(`{issuer}/.well-known/openid-configuration`) unless overridden in "
+ "`jwks_uris`."
+ ),
+ )
+ jwks_uris: dict[str, str] | None = Field(
+ default=None,
+ description=(
+ "Optional explicit JWKS URI per issuer, keyed by the issuer string. When "
+ "an issuer is absent here, its JWKS URI is discovered via OIDC. Provide "
+ "this for issuers that do not publish an OIDC discovery document."
+ ),
+ )
+ audience: str | None = Field(
+ default=None,
+ description=(
+ "Expected `aud` value on the ID-JAG. When omitted, the audience is the "
+ "authorization server's own issuer URL (its base URL), which is where the "
+ "ID-JAG's `aud` must point per SEP-990. Override only when the IdP mints "
+ "assertions bound to a different audience identifier."
+ ),
+ )
+ required_scopes: list[str] | None = Field(
+ default=None,
+ description="Scopes that must be present on the issued access token.",
+ )
+ algorithm: str | None = Field(
+ default=None,
+ description=(
+ "JWS signing algorithm the trusted issuers use (e.g. `ES256`, "
+ "`PS256`). When omitted, verification defaults to `RS256`; IdPs "
+ "signing with another algorithm must set this explicitly. When "
+ "issuers use different algorithms, override per issuer with "
+ "`algorithms`."
+ ),
+ )
+ algorithms: dict[str, str] | None = Field(
+ default=None,
+ description=(
+ "Optional per-issuer signing-algorithm override, keyed by the "
+ "issuer string (mirroring `jwks_uris`). Issuers absent here fall "
+ "back to `algorithm`."
+ ),
+ )
+ access_token_expiry_seconds: int = Field(
+ default=300,
+ gt=0,
+ description=(
+ "Lifetime, in seconds, of the short-lived access token minted from an "
+ "ID-JAG. SEP-990 relies on the client re-exchanging a fresh assertion, so "
+ "this is intentionally short and no refresh token is issued."
+ ),
+ )
+
+ @field_validator("trusted_issuers")
+ @classmethod
+ def _validate_trusted_issuers(cls, v: list[str]) -> list[str]:
+ if not v:
+ raise ValueError("identity_assertion.trusted_issuers must not be empty")
+ for issuer in v:
+ if not issuer or not issuer.strip():
+ raise ValueError("trusted_issuers entries must be non-empty strings")
+ return v
+
+ @field_validator("algorithm")
+ @classmethod
+ def _validate_algorithm(cls, v: str | None) -> str | None:
+ # Trusted issuers are verified via JWKS (public keys only), so the
+ # algorithm must be one of the asymmetric JWS algorithms JWTVerifier
+ # actually supports — HS* (shared-secret) has no JWKS equivalent, and
+ # anything else (EdDSA, or a typo like RS999) would otherwise surface
+ # as a 500 on the first exchange instead of a clean config error now.
+ if v is not None and v not in SUPPORTED_ASSERTION_ALGORITHMS:
+ supported = ", ".join(sorted(SUPPORTED_ASSERTION_ALGORITHMS))
+ raise ValueError(
+ f"Unsupported algorithm {v!r} for identity assertion: trusted "
+ f"issuers are verified via JWKS, so algorithm must be one of "
+ f"{supported}"
+ )
+ return v
+
+ @field_validator("algorithms")
+ @classmethod
+ def _validate_algorithms(cls, v: dict[str, str] | None) -> dict[str, str] | None:
+ if v is not None:
+ for issuer, algorithm in v.items():
+ if algorithm not in SUPPORTED_ASSERTION_ALGORITHMS:
+ supported = ", ".join(sorted(SUPPORTED_ASSERTION_ALGORITHMS))
+ raise ValueError(
+ f"Unsupported algorithm {algorithm!r} for issuer "
+ f"{issuer!r}: must be one of {supported}"
+ )
+ return v
+
+
+class IdentityAssertionError(Exception):
+ """Raised when an ID-JAG fails validation.
+
+ The message is for server-side logging only; the token endpoint maps this to a
+ generic OAuth error response and does not leak the detail to the client.
+ """
+
+
+class IdentityAssertionValidator:
+ """Validates ID-JAG assertions for the SEP-990 jwt-bearer grant.
+
+ Reuses :class:`JWTVerifier` for signature, issuer, audience, and expiry checks
+ (with JWKS fetching and caching), and layers on the SEP-990 processing rules
+ that the generic verifier does not cover: the ``typ`` JOSE header, a mandatory
+ ``sub``, and ``jti`` replay rejection.
+
+ JTI replay protection mirrors :class:`CIMDAssertionValidator`: seen ``jti``
+ values are cached until the assertion would expire anyway, with periodic
+ cleanup and an emergency size cap. Like CIMD, the cache is per-process, so
+ replay protection is not shared across horizontally-scaled workers or
+ replicas; see the identity-assertion docs for the deployment caveat.
+ """
+
+ #: RFC 7523 recommends short-lived assertions; reject anything longer.
+ MAX_ASSERTION_LIFETIME = 300 # 5 minutes
+ #: Clock-skew tolerance for exp/iat checks.
+ CLOCK_SKEW_SECONDS = 30
+
+ def __init__(self, config: IdentityAssertion, audience: str):
+ """Initialize the validator.
+
+ Args:
+ config: The identity assertion configuration.
+ audience: The authorization server's own issuer URL; the ID-JAG's `aud`
+ must match this unless `config.audience` overrides it.
+ """
+ self.config = config
+ # Accept the audience both with and without a trailing slash: metadata
+ # advertises the issuer exactly as pydantic renders base_url (a bare
+ # domain gains a trailing slash), so an IdP that sets `aud` to the
+ # advertised value verbatim must match, and so must one that strips it.
+ if config.audience:
+ self.audience: str | list[str] = config.audience
+ else:
+ base = audience.rstrip("/")
+ self.audience = [base, base + "/"]
+
+ self._jti_cache: dict[str, float] = {}
+ self._jti_cache_max_size = 10000
+ self._last_cleanup = time.monotonic()
+ self._cleanup_interval = 60
+ # One JWTVerifier per issuer, created lazily once the JWKS URI is known.
+ self._verifiers: dict[str, JWTVerifier] = {}
+ # OIDC discovery hardening: discovery runs before signature verification,
+ # so a malformed-but-trusted-iss assertion can trigger an outbound HTTP
+ # call. Serialize per-issuer lookups and back off after a failure so
+ # concurrent or repeated garbage cannot amplify into request floods.
+ self._discovery_locks: dict[str, asyncio.Lock] = {}
+ self._discovery_failures: dict[str, float] = {}
+ self._discovery_failure_cooldown = 30.0
+
+ def _cleanup_expired_jtis(self) -> None:
+ now = time.time()
+ expired = [jti for jti, exp in self._jti_cache.items() if exp < now]
+ for jti in expired:
+ del self._jti_cache[jti]
+ if expired:
+ logger.debug("Cleaned up %d expired ID-JAG jtis from cache", len(expired))
+
+ def _maybe_cleanup(self) -> None:
+ now = time.monotonic()
+ if now - self._last_cleanup > self._cleanup_interval:
+ self._cleanup_expired_jtis()
+ self._last_cleanup = now
+
+ async def _discover_jwks_uri(self, issuer: str) -> str:
+ """Discover an issuer's JWKS URI via OIDC discovery.
+
+ Fetches ``{issuer}/.well-known/openid-configuration`` and returns its
+ ``jwks_uri``. Trusted issuers are operator-configured, so this uses a
+ plain fetch (consistent with how operator-configured JWKS URIs are
+ treated elsewhere, including localhost issuers in development).
+ """
+ lock = self._discovery_locks.setdefault(issuer, asyncio.Lock())
+ async with lock:
+ failed_at = self._discovery_failures.get(issuer)
+ if (
+ failed_at is not None
+ and time.monotonic() - failed_at < self._discovery_failure_cooldown
+ ):
+ raise IdentityAssertionError(
+ f"OIDC discovery for issuer {issuer!r} recently failed; backing off"
+ )
+ return await self._fetch_discovery(issuer)
+
+ async def _fetch_discovery(self, issuer: str) -> str:
+ """Perform the actual discovery fetch; caller holds the issuer lock."""
+ config_url = issuer.rstrip("/") + "/.well-known/openid-configuration"
+ try:
+ async with httpx2.AsyncClient() as client:
+ response = await client.get(config_url, timeout=10.0)
+ response.raise_for_status()
+ body = response.json()
+ except (httpx2.HTTPError, ValueError) as e:
+ self._discovery_failures[issuer] = time.monotonic()
+ raise IdentityAssertionError(
+ f"OIDC discovery for issuer {issuer!r} failed: {e}"
+ ) from e
+ if not isinstance(body, dict):
+ # Valid JSON that isn't an object (e.g. `[]` or a bare string) —
+ # guard before .get() so a misbehaving discovery endpoint maps to
+ # invalid_grant, not a 500 on every subsequent exchange.
+ raise IdentityAssertionError(
+ f"OIDC discovery document for issuer {issuer!r} is not a JSON object"
+ )
+
+ jwks_uri = body.get("jwks_uri")
+ if not jwks_uri or not isinstance(jwks_uri, str):
+ raise IdentityAssertionError(
+ f"OIDC discovery document for issuer {issuer!r} has no jwks_uri"
+ )
+ return jwks_uri
+
+ async def _get_verifier(self, issuer: str) -> JWTVerifier:
+ from fastmcp.server.auth.providers.jwt import JWTVerifier as _JWTVerifier
+
+ verifier = self._verifiers.get(issuer)
+ if verifier is not None:
+ return verifier
+
+ jwks_uri = (self.config.jwks_uris or {}).get(issuer)
+ if not jwks_uri:
+ jwks_uri = await self._discover_jwks_uri(issuer)
+
+ algorithm = (self.config.algorithms or {}).get(issuer, self.config.algorithm)
+ verifier = _JWTVerifier(
+ jwks_uri=jwks_uri,
+ issuer=issuer,
+ audience=self.audience,
+ algorithm=algorithm,
+ )
+ self._verifiers[issuer] = verifier
+ return verifier
+
+ async def validate(
+ self, assertion: str, *, client_id: str, resource_url: str | None
+ ) -> dict:
+ """Validate an ID-JAG and return its claims.
+
+ Args:
+ assertion: The compact-serialized ID-JAG JWT.
+ client_id: The authenticated client presenting the assertion. Must
+ match the assertion's signed `client_id` claim — checked before
+ the jti is recorded as consumed, so an assertion presented by
+ the wrong client is rejected without burning it for the right
+ one.
+ resource_url: This server's resource URL, if configured. Must match
+ the assertion's signed `resource` claim, for the same reason.
+
+ Returns:
+ The verified claims (including `sub`, `iss`, and any `resource`/`scope`).
+
+ Raises:
+ IdentityAssertionError: If the assertion is invalid for any reason.
+ """
+ self._maybe_cleanup()
+
+ # 1. typ header MUST be oauth-id-jag+jwt (SEP-990 §5.1).
+ try:
+ header = decode_jwt_header(assertion)
+ except (ValueError, KeyError, IndexError) as e:
+ raise IdentityAssertionError(f"Malformed assertion header: {e}") from e
+ if not isinstance(header, dict):
+ # A JSON-array/scalar header is valid JSON but not a JOSE header;
+ # guard before .get() so this maps to invalid_grant, not a 500.
+ raise IdentityAssertionError("Assertion JOSE header must be a JSON object")
+ if header.get("typ") != ID_JAG_TYP:
+ raise IdentityAssertionError(
+ f"Assertion typ must be {ID_JAG_TYP!r}, got {header.get('typ')!r}"
+ )
+
+ # 2. iss must be a trusted issuer before we fetch any keys for it.
+ try:
+ unverified_claims = _decode_unverified_claims(assertion)
+ except (ValueError, KeyError, IndexError) as e:
+ raise IdentityAssertionError(f"Malformed assertion payload: {e}") from e
+ if not isinstance(unverified_claims, dict):
+ raise IdentityAssertionError("Assertion payload is not a JSON object")
+ iss = unverified_claims.get("iss")
+ if not iss or iss not in self.config.trusted_issuers:
+ raise IdentityAssertionError(f"Untrusted assertion issuer: {iss!r}")
+
+ # 3. Verify signature, iss, aud, and exp via JWTVerifier.
+ verifier = await self._get_verifier(iss)
+ access_token = await verifier.load_access_token(assertion)
+ if access_token is None:
+ raise IdentityAssertionError(
+ "Assertion failed signature/issuer/audience/expiry validation"
+ )
+ claims = access_token.claims
+
+ now = time.time()
+ exp = _numeric_date_claim(claims, "exp")
+ iat = _numeric_date_claim(claims, "iat")
+ nbf = _numeric_date_claim(claims, "nbf")
+ if exp is None:
+ raise IdentityAssertionError("Assertion must include exp claim")
+ if nbf is not None and nbf > now + self.CLOCK_SKEW_SECONDS:
+ raise IdentityAssertionError("Assertion is not yet valid (nbf in future)")
+ if iat is not None:
+ if iat > now + self.CLOCK_SKEW_SECONDS:
+ raise IdentityAssertionError("Assertion iat is in the future")
+ if exp - iat > self.MAX_ASSERTION_LIFETIME:
+ raise IdentityAssertionError(
+ f"Assertion lifetime too long (max {self.MAX_ASSERTION_LIFETIME}s)"
+ )
+ elif exp > now + self.MAX_ASSERTION_LIFETIME:
+ raise IdentityAssertionError(
+ f"Assertion exp too far in future (max {self.MAX_ASSERTION_LIFETIME}s)"
+ )
+
+ # 4. sub is mandatory (RFC 7523 §3) — it identifies the end user.
+ sub = claims.get("sub")
+ if not sub:
+ raise IdentityAssertionError("Assertion must include sub claim")
+
+ # 5. Required scopes on the issued access token derive from the assertion.
+ if self.config.required_scopes:
+ granted = set(_assertion_scopes(claims))
+ missing = set(self.config.required_scopes) - granted
+ if missing:
+ raise IdentityAssertionError(
+ f"Assertion missing required scopes: {sorted(missing)}"
+ )
+
+ # 6. The signed client_id and resource claims bind the assertion to the
+ # presenting client and this server. Checked here — before jti is
+ # recorded as consumed below — so an assertion presented with the
+ # wrong binding is rejected without burning replay protection for
+ # whichever client/server it actually belongs to.
+ assertion_client_id = claims.get("client_id")
+ if not assertion_client_id or assertion_client_id != client_id:
+ raise IdentityAssertionError(
+ f"Assertion client_id {assertion_client_id!r} does not match "
+ f"authenticated client {client_id!r}"
+ )
+ if resource_url is not None:
+ assertion_resource = claims.get("resource")
+ if not isinstance(assertion_resource, str) or not assertion_resource:
+ raise IdentityAssertionError("Assertion is missing resource claim")
+ if server_url_has_query(resource_url):
+ claim_matches = assertion_resource.rstrip("/") == resource_url.rstrip(
+ "/"
+ )
+ else:
+ claim_matches = normalize_resource_url(
+ assertion_resource
+ ) == normalize_resource_url(resource_url)
+ if not claim_matches:
+ raise IdentityAssertionError(
+ f"Assertion resource {assertion_resource!r} does not match "
+ f"this server {resource_url!r}"
+ )
+
+ # 7. jti replay rejection (RFC 7523 §3). Must be a non-empty string —
+ # an array/object jti is unhashable and would raise TypeError on the
+ # cache lookup (a 500) instead of a clean invalid_grant.
+ jti = claims.get("jti")
+ if not jti or not isinstance(jti, str):
+ raise IdentityAssertionError("Assertion must include a string jti claim")
+ cached_exp = self._jti_cache.get(jti)
+ if cached_exp is not None and cached_exp > now:
+ raise IdentityAssertionError(f"Assertion replay detected: jti {jti} reused")
+
+ # Enforce the cap BEFORE inserting so a rejected assertion never grows the
+ # cache. A fresh jti that would exceed capacity is rejected outright (after
+ # a cleanup pass to reclaim any expired entries first).
+ if (
+ jti not in self._jti_cache
+ and len(self._jti_cache) >= self._jti_cache_max_size
+ ):
+ self._cleanup_expired_jtis()
+ if len(self._jti_cache) >= self._jti_cache_max_size:
+ logger.warning("ID-JAG jti cache at capacity, possible attack")
+ raise IdentityAssertionError("Server overloaded, please retry")
+ self._jti_cache[jti] = exp
+
+ logger.debug("ID-JAG validated for subject=%s issuer=%s", sub, iss)
+ return claims
+
+
+def normalize_resource_url(url: str) -> str:
+ """Normalize a resource URL by removing query parameters and trailing slashes.
+
+ RFC 8707 allows clients to include query parameters in resource URLs, but
+ the server's configured resource URL typically doesn't include them. This
+ normalizes both sides for comparison by stripping query and fragment.
+ """
+ parsed = urlparse(str(url))
+ return urlunparse(
+ (parsed.scheme, parsed.netloc, parsed.path.rstrip("/"), "", "", "")
+ )
+
+
+def server_url_has_query(url: str) -> bool:
+ """Check if a URL has query parameters."""
+ return bool(urlparse(str(url)).query)
+
+
+def _numeric_date_claim(claims: dict, name: str) -> float | None:
+ """Read a NumericDate claim (RFC 7519 §2), rejecting non-numeric values.
+
+ A validly-signed assertion could still carry a malformed `exp`/`iat`/`nbf`
+ (e.g. a string, from a misbehaving IdP); comparing against it directly
+ would raise `TypeError` outside the validation-error path. `bool` is
+ excluded even though it subclasses `int` in Python — `true`/`false` are
+ not timestamps.
+ """
+ value = claims.get(name)
+ if value is None:
+ return None
+ if isinstance(value, bool) or not isinstance(value, (int, float)):
+ raise IdentityAssertionError(f"Assertion {name} claim must be a number")
+ return float(value)
+
+
+def _assertion_scopes(claims: dict) -> list[str]:
+ """Extract the scopes an ID-JAG grants, from `scope` or `scp`."""
+ scope = claims.get("scope")
+ if isinstance(scope, str):
+ return scope.split()
+ scp = claims.get("scp")
+ if isinstance(scp, list):
+ return [str(s) for s in scp]
+ if isinstance(scp, str):
+ return scp.split()
+ return []
+
+
+def _decode_unverified_claims(token: str) -> dict:
+ """Decode a JWT payload without verifying the signature.
+
+ Used only to read the `iss` claim so we can select the trusted issuer's key
+ before performing the real, signature-verifying decode.
+ """
+ import base64
+ import json
+
+ payload_b64 = token.split(".")[1]
+ payload_b64 += "=" * (-len(payload_b64) % 4)
+ return json.loads(base64.urlsafe_b64decode(payload_b64))
diff --git a/fastmcp_slim/fastmcp/server/auth/jwt_issuer.py b/fastmcp_slim/fastmcp/server/auth/jwt_issuer.py
index ae5c53715..6715e4083 100644
--- a/fastmcp_slim/fastmcp/server/auth/jwt_issuer.py
+++ b/fastmcp_slim/fastmcp/server/auth/jwt_issuer.py
@@ -109,6 +109,8 @@ class JWTIssuer:
jti: str,
expires_in: int = 3600,
upstream_claims: dict[str, Any] | None = None,
+ subject: str | None = None,
+ extra_claims: dict[str, Any] | None = None,
) -> str:
"""Issue a minimal FastMCP access token.
@@ -122,6 +124,12 @@ class JWTIssuer:
jti: Unique token identifier (maps to upstream token)
expires_in: Token lifetime in seconds
upstream_claims: Optional claims from upstream IdP token to include
+ subject: Optional `sub` claim. Set for self-contained tokens (e.g.
+ minted from an ID-JAG) where the subject is carried directly in
+ the token rather than looked up via a JTI mapping.
+ extra_claims: Optional additional top-level claims to embed. Used to
+ mark self-contained tokens (e.g. the ID-JAG issuer/marker) so
+ `load_access_token` can validate them without a JTI mapping.
Returns:
Signed JWT token
@@ -139,6 +147,12 @@ class JWTIssuer:
"jti": jti,
}
+ if subject is not None:
+ payload["sub"] = subject
+
+ if extra_claims:
+ payload.update(extra_claims)
+
if upstream_claims:
payload["upstream_claims"] = upstream_claims
diff --git a/fastmcp_slim/fastmcp/server/auth/oauth_proxy/proxy.py b/fastmcp_slim/fastmcp/server/auth/oauth_proxy/proxy.py
index 1c471db6b..fd6a4f3dc 100644
--- a/fastmcp_slim/fastmcp/server/auth/oauth_proxy/proxy.py
+++ b/fastmcp_slim/fastmcp/server/auth/oauth_proxy/proxy.py
@@ -26,12 +26,13 @@ from collections import OrderedDict
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any, Literal
-from urllib.parse import urlencode, urlparse, urlunparse
+from urllib.parse import urlencode
import anyio
import httpx2
from authlib.common.security import generate_token
from cryptography.fernet import Fernet
+from joserfc.errors import JoseError
from key_value.aio.adapters.pydantic import PydanticAdapter
from key_value.aio.protocols import AsyncKeyValue
from key_value.aio.stores.filetree import (
@@ -41,11 +42,13 @@ from key_value.aio.stores.filetree import (
)
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from mcp.server.auth.handlers.metadata import MetadataHandler
+from mcp.server.auth.middleware.client_auth import ClientAuthenticator
from mcp.server.auth.provider import (
AccessToken,
AuthorizationCode,
AuthorizationParams,
AuthorizeError,
+ IdentityAssertionParams,
RefreshToken,
RegistrationError,
TokenError,
@@ -71,6 +74,14 @@ from fastmcp.server.auth.auth import (
)
from fastmcp.server.auth.cimd import CIMDClientManager
from fastmcp.server.auth.handlers.authorize import AuthorizationHandler
+from fastmcp.server.auth.identity_assertion import (
+ JWT_BEARER_GRANT_TYPE,
+ IdentityAssertion,
+ IdentityAssertionError,
+ IdentityAssertionValidator,
+ normalize_resource_url,
+ server_url_has_query,
+)
from fastmcp.server.auth.jwt_issuer import (
JWTIssuer,
derive_jwt_key,
@@ -100,29 +111,23 @@ logger = get_logger(__name__)
_REFRESH_LOCK_CACHE_SIZE = 10_000
-
-def _normalize_resource_url(url: str) -> str:
- """Normalize a resource URL by removing query parameters and trailing slashes.
-
- RFC 8707 allows clients to include query parameters in resource URLs, but the
- server's configured resource URL typically doesn't include them. This function
- normalizes URLs for comparison by stripping query params and fragments.
-
- Args:
- url: The URL to normalize
-
- Returns:
- Normalized URL with scheme, host, and path only (no query/fragment)
- """
- parsed = urlparse(str(url))
- return urlunparse(
- (parsed.scheme, parsed.netloc, parsed.path.rstrip("/"), "", "", "")
- )
+#: Marker claim identifying a FastMCP access token minted from a SEP-990 ID-JAG.
+#: These tokens are self-contained (they carry the asserted subject directly) and
+#: are validated without the upstream token-swap that regular proxy tokens use.
+_ID_JAG_GRANT_MARKER = "id_jag"
-def _server_url_has_query(url: str) -> bool:
- """Check if a URL has query parameters."""
- return bool(urlparse(str(url)).query)
+def _assertion_granted_scopes(claims: dict[str, Any]) -> list[str]:
+ """Scopes granted by an ID-JAG, from its `scope` or `scp` claim."""
+ scope = claims.get("scope")
+ if isinstance(scope, str):
+ return scope.split()
+ scp = claims.get("scp")
+ if isinstance(scp, list):
+ return [str(s) for s in scp]
+ if isinstance(scp, str):
+ return scp.split()
+ return []
class OAuthProxy(OAuthProvider, ConsentMixin):
@@ -281,6 +286,8 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
token_expiry_threshold_seconds: int = 0,
# CIMD (Client ID Metadata Document) support
enable_cimd: bool = True,
+ # Identity assertion (SEP-990 ID-JAG) support
+ identity_assertion: IdentityAssertion | None = None,
):
"""Initialize the OAuth proxy provider.
@@ -377,6 +384,11 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
enable_cimd: Enable CIMD (Client ID Metadata Document) support for URL-based
client IDs. When True, clients can authenticate using HTTPS URLs as client
IDs, with metadata fetched from the URL. Supports private_key_jwt auth.
+ identity_assertion: Optional SEP-990 identity assertion (ID-JAG) configuration.
+ When provided, the token endpoint accepts the RFC 7523 jwt-bearer grant
+ carrying an ID-JAG issued by one of the configured trusted issuers, and
+ mints a short-lived access token (no refresh token) for the asserted
+ subject. When omitted, the grant is rejected as unsupported.
"""
default_scopes = valid_scopes or token_verifier.required_scopes
@@ -620,6 +632,20 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
allowed_redirect_uri_patterns=self._allowed_client_redirect_uris,
)
+ # Identity assertion (SEP-990 ID-JAG): the audience the ID-JAG must be
+ # bound to is this authorization server's own issuer URL (base_url).
+ self._identity_assertion: IdentityAssertion | None = identity_assertion
+ self._identity_assertion_validator: IdentityAssertionValidator | None = None
+ if identity_assertion is not None:
+ self._identity_assertion_validator = IdentityAssertionValidator(
+ config=identity_assertion,
+ audience=str(self.base_url),
+ )
+ # ID-JAG access tokens are self-contained (no upstream token or JTI
+ # mapping to delete), so revocation tracks their jtis here until the
+ # token would expire anyway. Per-process, like ID-JAG replay tracking.
+ self._revoked_id_jag_jtis: dict[str, float] = {}
+
# Advisory locks for transparent upstream token refresh, keyed by
# upstream_token_id. Prevents concurrent async tasks from racing to
# refresh the same token within a single process. Does not protect
@@ -815,11 +841,14 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
"Client %s matched upstream client_id — synthesizing client without DCR",
client_id,
)
+ synthesized_grant_types = ["authorization_code", "refresh_token"]
+ if self._identity_assertion is not None:
+ synthesized_grant_types.append(JWT_BEARER_GRANT_TYPE)
return ProxyDCRClient(
client_id=client_id,
client_secret=None,
redirect_uris=[AnyUrl("http://localhost")],
- grant_types=["authorization_code", "refresh_token"],
+ grant_types=synthesized_grant_types,
scope=self._default_scope_str,
token_endpoint_auth_method="none",
allowed_redirect_uri_patterns=self._allowed_client_redirect_uris,
@@ -854,6 +883,20 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
redirect_uris = client_info.redirect_uris or [AnyUrl("http://localhost")]
+ # When identity assertion is enabled, registered clients are allowed to
+ # present the SEP-990 jwt-bearer (ID-JAG) grant. Add it to the client's
+ # registered grant types so the token endpoint's grant-type check accepts
+ # it — clients that never register (or register without it while identity
+ # assertion is disabled) remain unable to use the grant.
+ registered_grant_types = list(
+ client_info.grant_types or ["authorization_code", "refresh_token"]
+ )
+ if (
+ self._identity_assertion is not None
+ and JWT_BEARER_GRANT_TYPE not in registered_grant_types
+ ):
+ registered_grant_types.append(JWT_BEARER_GRANT_TYPE)
+
# We use token_endpoint_auth_method="none" because the proxy handles
# all upstream authentication. The client_secret must also be None
# because the SDK requires secrets to be provided if they're set,
@@ -862,8 +905,7 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
client_id=client_info.client_id,
client_secret=None,
redirect_uris=redirect_uris,
- grant_types=client_info.grant_types
- or ["authorization_code", "refresh_token"],
+ grant_types=registered_grant_types,
scope=client_info.scope or self._default_scope_str,
token_endpoint_auth_method="none",
allowed_redirect_uri_patterns=self._allowed_client_redirect_uris,
@@ -931,14 +973,14 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
server_url = str(self._resource_url)
client_url = str(client_resource)
- if _server_url_has_query(server_url):
+ if server_url_has_query(server_url):
# Server has query params - require exact match for security
urls_match = client_url.rstrip("/") == server_url.rstrip("/")
else:
# Server has no query params - normalize both for comparison
- urls_match = _normalize_resource_url(
+ urls_match = normalize_resource_url(
client_url
- ) == _normalize_resource_url(server_url)
+ ) == normalize_resource_url(server_url)
if not urls_match:
logger.warning(
@@ -1284,6 +1326,131 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
scope=" ".join(granted_scopes),
)
+ # -------------------------------------------------------------------------
+ # Identity Assertion Flow (SEP-990 ID-JAG)
+ # -------------------------------------------------------------------------
+
+ async def exchange_identity_assertion(
+ self,
+ client: OAuthClientInformationFull,
+ params: IdentityAssertionParams,
+ ) -> OAuthToken:
+ """Exchange a SEP-990 ID-JAG for a short-lived FastMCP access token.
+
+ Validates the ID-JAG against the configured trusted issuers (signature,
+ `iss`, `aud`, `exp`, `typ`, `sub`, and `jti` replay), then mints a
+ self-contained FastMCP access token carrying the asserted subject. No
+ refresh token is issued — the client re-exchanges a fresh assertion.
+
+ Raises:
+ TokenError: ``invalid_grant`` if the assertion is rejected, or
+ ``unsupported_grant_type`` if identity assertion is not configured.
+ """
+ if (
+ self._identity_assertion is None
+ or self._identity_assertion_validator is None
+ ):
+ raise TokenError(
+ "unsupported_grant_type",
+ "The JWT bearer grant is not supported by this authorization server",
+ )
+
+ # RFC 8707: when the request names a resource, it must be this server —
+ # the same invariant (and the same skip-when-unconfigured behavior)
+ # authorize() enforces for authorization requests.
+ if params.resource and self._resource_url:
+ server_url = str(self._resource_url)
+ client_url = str(params.resource)
+ if server_url_has_query(server_url):
+ # Server has query params - require exact match for security
+ resource_matches = client_url.rstrip("/") == server_url.rstrip("/")
+ else:
+ resource_matches = normalize_resource_url(
+ client_url
+ ) == normalize_resource_url(server_url)
+ if not resource_matches:
+ logger.warning(
+ "ID-JAG resource mismatch: client requested %s but server is %s",
+ client_url,
+ self._resource_url,
+ )
+ raise TokenError(
+ "invalid_target", "Resource does not match this server"
+ )
+
+ # SEP-990: the assertion's signed client_id and resource claims (checked
+ # against the authenticated client and this server) bind the assertion
+ # before its jti is recorded as consumed — passed into validate() itself
+ # so that binding happens ahead of replay-tracking, not after.
+ try:
+ claims = await self._identity_assertion_validator.validate(
+ params.assertion,
+ client_id=client.client_id or "",
+ resource_url=str(self._resource_url) if self._resource_url else None,
+ )
+ except IdentityAssertionError as e:
+ # Log detail server-side; return a generic error to the client so we
+ # do not leak which validation step failed.
+ logger.info("ID-JAG rejected: %s", e)
+ raise TokenError("invalid_grant", "Invalid identity assertion") from e
+
+ subject = str(claims["sub"])
+
+ # Granted scopes are authoritative from the signed assertion (or, when the
+ # assertion omits them, from explicit server policy). The client-supplied
+ # request `scope` (`params.scopes`) is NOT covered by the signed assertion,
+ # so it may only NARROW the granted set — never widen it. A client cannot
+ # obtain a scope the assertion did not grant by asking for it at the token
+ # endpoint (e.g. an assertion granting `read` requesting `admin` gets nothing
+ # extra).
+ authoritative_scopes = _assertion_granted_scopes(claims)
+ if not authoritative_scopes:
+ authoritative_scopes = list(self._identity_assertion.required_scopes or [])
+
+ # Configured mandatory scopes that the assertion actually grants must always
+ # ride on the issued token — the client request may only narrow the
+ # remaining, optional scopes. Otherwise a request like `scope=read` could
+ # silently drop a required `admin` scope the assertion authorized.
+ required = set(self._identity_assertion.required_scopes or [])
+
+ if params.scopes:
+ requested = set(params.scopes)
+ granted_scopes = [
+ s for s in authoritative_scopes if s in required or s in requested
+ ]
+ else:
+ granted_scopes = list(authoritative_scopes)
+
+ expires_in = self._identity_assertion.access_token_expiry_seconds
+ access_jti = secrets.token_urlsafe(32)
+
+ access_token = self.jwt_issuer.issue_access_token(
+ client_id=client.client_id or "",
+ scopes=granted_scopes,
+ jti=access_jti,
+ expires_in=expires_in,
+ subject=subject,
+ extra_claims={
+ "fastmcp_grant": _ID_JAG_GRANT_MARKER,
+ "assertion_iss": str(claims.get("iss")),
+ },
+ )
+
+ logger.debug(
+ "Issued ID-JAG access token for subject=%s client=%s jti=%s",
+ subject,
+ client.client_id,
+ access_jti[:8],
+ )
+
+ # SEP-990: no refresh token — the IdP controls session lifetime.
+ return OAuthToken(
+ access_token=access_token,
+ token_type="Bearer",
+ expires_in=expires_in,
+ scope=" ".join(granted_scopes),
+ )
+
# -------------------------------------------------------------------------
# Refresh Token Flow
# -------------------------------------------------------------------------
@@ -1808,6 +1975,24 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
jti = payload["jti"]
upstream_claims = payload.get("upstream_claims")
+ # SEP-990: ID-JAG tokens are self-contained — the asserted subject
+ # is carried in the token itself, and there is no upstream token to
+ # swap for. Return directly from the verified claims, unless the
+ # token was revoked (tracked by jti until natural expiry).
+ if payload.get("fastmcp_grant") == _ID_JAG_GRANT_MARKER:
+ if jti in self._revoked_id_jag_jtis:
+ logger.info("Rejected revoked ID-JAG access token jti=%s", jti[:16])
+ return None
+ scope = payload.get("scope", "")
+ return AccessToken(
+ token=token,
+ client_id=str(payload.get("client_id", "")),
+ scopes=scope.split() if scope else [],
+ expires_at=int(payload["exp"]) if payload.get("exp") else None,
+ subject=str(payload["sub"]) if payload.get("sub") else None,
+ claims=payload,
+ )
+
# 2. Look up upstream token via JTI mapping
jti_mapping = await self._jti_mapping_store.get(key=jti)
if not jti_mapping:
@@ -1964,6 +2149,28 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
if isinstance(token, RefreshToken):
await self._refresh_token_store.delete(key=_hash_token(token.token))
+ # ID-JAG access tokens are self-contained and never known upstream, so
+ # upstream revocation cannot invalidate them. Track the jti locally so
+ # load_access_token() rejects the token for its remaining lifetime.
+ try:
+ payload = self.jwt_issuer.verify_token(token.token)
+ except (JoseError, ValueError, KeyError):
+ # Not a (valid) FastMCP-issued JWT — nothing to track locally.
+ payload = None
+ if payload is not None and payload.get("fastmcp_grant") == _ID_JAG_GRANT_MARKER:
+ now = time.time()
+ self._revoked_id_jag_jtis = {
+ jti: exp for jti, exp in self._revoked_id_jag_jtis.items() if exp > now
+ }
+ exp = payload.get("exp")
+ jti = payload.get("jti")
+ if isinstance(jti, str) and jti:
+ self._revoked_id_jag_jtis[jti] = (
+ float(exp) if isinstance(exp, (int, float)) else now + 3600
+ )
+ logger.debug("Revoked ID-JAG access token jti=%s", jti[:16])
+ return
+
# Attempt upstream revocation if endpoint is configured
if self._upstream_revocation_endpoint:
try:
@@ -2050,22 +2257,30 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
)
)
elif (
- self._cimd_manager is not None
+ (self._cimd_manager is not None or self._identity_assertion is not None)
and isinstance(route, Route)
and route.path == "/token"
and route.methods is not None
and "POST" in route.methods
):
- # Replace the token endpoint authenticator with one that supports
- # private_key_jwt for CIMD clients
+ # Replace the token endpoint so it can (a) authenticate CIMD
+ # private_key_jwt clients and (b) accept the SEP-990 jwt-bearer
+ # grant when identity assertion is enabled.
token_endpoint_url = f"{self.base_url}/token"
- cimd_authenticator = PrivateKeyJWTClientAuthenticator(
- provider=self,
- cimd_manager=self._cimd_manager,
- token_endpoint_url=token_endpoint_url,
- )
+ if self._cimd_manager is not None:
+ authenticator: ClientAuthenticator = (
+ PrivateKeyJWTClientAuthenticator(
+ provider=self,
+ cimd_manager=self._cimd_manager,
+ token_endpoint_url=token_endpoint_url,
+ )
+ )
+ else:
+ authenticator = ClientAuthenticator(self)
token_handler = TokenHandler(
- provider=self, client_authenticator=cimd_authenticator
+ provider=self,
+ client_authenticator=authenticator,
+ identity_assertion_enabled=self._identity_assertion is not None,
)
custom_routes.append(
Route(
@@ -2077,7 +2292,7 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
)
)
elif (
- self._cimd_manager is not None
+ (self._cimd_manager is not None or self._identity_assertion is not None)
and isinstance(route, Route)
and route.path.startswith("/.well-known/oauth-authorization-server")
):
@@ -2090,14 +2305,28 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
self.service_documentation_url,
client_registration_options,
revocation_options,
+ supports_identity_assertion=self._identity_assertion is not None,
)
- metadata.client_id_metadata_document_supported = True
- existing = metadata.token_endpoint_auth_methods_supported or []
- metadata.token_endpoint_auth_methods_supported = [
- *existing,
- "private_key_jwt",
- "none",
- ]
+ if self._cimd_manager is not None:
+ metadata.client_id_metadata_document_supported = True
+ existing = metadata.token_endpoint_auth_methods_supported or []
+ metadata.token_endpoint_auth_methods_supported = [
+ *existing,
+ "private_key_jwt",
+ "none",
+ ]
+ if self._identity_assertion is not None:
+ # DCR clients are public (`token_endpoint_auth_method="none"`),
+ # so a metadata consumer must see `none` advertised to use the
+ # jwt-bearer grant — even when CIMD (which also adds it) is off.
+ methods_supported = (
+ metadata.token_endpoint_auth_methods_supported or []
+ )
+ if "none" not in methods_supported:
+ metadata.token_endpoint_auth_methods_supported = [
+ *methods_supported,
+ "none",
+ ]
handler = MetadataHandler(metadata)
methods = route.methods or ["GET", "OPTIONS"]
diff --git a/fastmcp_slim/fastmcp/server/auth/oidc_proxy.py b/fastmcp_slim/fastmcp/server/auth/oidc_proxy.py
index c96e7bdb8..8f17c75b8 100644
--- a/fastmcp_slim/fastmcp/server/auth/oidc_proxy.py
+++ b/fastmcp_slim/fastmcp/server/auth/oidc_proxy.py
@@ -18,6 +18,7 @@ from pydantic import AnyHttpUrl, BaseModel, model_validator
from typing_extensions import Self
from fastmcp.server.auth import TokenVerifier
+from fastmcp.server.auth.identity_assertion import IdentityAssertion
from fastmcp.server.auth.oauth_proxy import OAuthProxy
from fastmcp.server.auth.oauth_proxy.models import UpstreamTokenSet
from fastmcp.server.auth.providers.jwt import JWTVerifier
@@ -246,6 +247,8 @@ class OIDCProxy(OAuthProxy):
token_expiry_threshold_seconds: int = 0,
# CIMD configuration
enable_cimd: bool = True,
+ # Identity assertion (SEP-990 ID-JAG) support
+ identity_assertion: IdentityAssertion | None = None,
) -> None:
"""Initialize the OIDC proxy provider.
@@ -328,6 +331,9 @@ class OIDCProxy(OAuthProxy):
enable_cimd: Whether to enable CIMD (Client ID Metadata Document) client support.
When True, clients can use their metadata document URL as client_id instead of
Dynamic Client Registration. Default is True.
+ identity_assertion: Optional SEP-990 identity assertion (ID-JAG) configuration.
+ When provided, the token endpoint accepts the RFC 7523 jwt-bearer grant
+ carrying an ID-JAG issued by one of the configured trusted issuers.
"""
if not config_url:
raise ValueError("Missing required config URL")
@@ -419,6 +425,7 @@ class OIDCProxy(OAuthProxy):
"fastmcp_access_token_expiry_seconds": fastmcp_access_token_expiry_seconds,
"token_expiry_threshold_seconds": token_expiry_threshold_seconds,
"enable_cimd": enable_cimd,
+ "identity_assertion": identity_assertion,
}
if redirect_path:
diff --git a/tests/server/auth/oauth_proxy/test_config.py b/tests/server/auth/oauth_proxy/test_config.py
index a69330a3e..cc9b2bff6 100644
--- a/tests/server/auth/oauth_proxy/test_config.py
+++ b/tests/server/auth/oauth_proxy/test_config.py
@@ -6,15 +6,15 @@ from mcp.server.auth.provider import AuthorizationParams, AuthorizeError
from mcp.shared.auth import OAuthClientInformationFull
from pydantic import AnyHttpUrl, AnyUrl
-from fastmcp.server.auth.oauth_proxy import OAuthProxy
-from fastmcp.server.auth.oauth_proxy.proxy import (
- _normalize_resource_url,
- _server_url_has_query,
+from fastmcp.server.auth.identity_assertion import (
+ normalize_resource_url,
+ server_url_has_query,
)
+from fastmcp.server.auth.oauth_proxy import OAuthProxy
class TestNormalizeResourceUrl:
- """Unit tests for the _normalize_resource_url helper function."""
+ """Unit tests for the normalize_resource_url helper function."""
@pytest.mark.parametrize(
"url,expected",
@@ -46,7 +46,7 @@ class TestNormalizeResourceUrl:
)
def test_normalizes_urls_correctly(self, url: str, expected: str):
"""Test that URLs are normalized by stripping query params, fragments, and trailing slashes."""
- assert _normalize_resource_url(url) == expected
+ assert normalize_resource_url(url) == expected
@pytest.mark.parametrize(
"url,has_query",
@@ -58,9 +58,9 @@ class TestNormalizeResourceUrl:
("https://example.com/mcp?a=1&b=2", True),
],
)
- def test_server_url_has_query(self, url: str, has_query: bool):
+ def testserver_url_has_query(self, url: str, has_query: bool):
"""Test detection of query parameters in server URLs."""
- assert _server_url_has_query(url) == has_query
+ assert server_url_has_query(url) == has_query
class TestResourceURLValidation:
diff --git a/tests/server/auth/oauth_proxy/test_identity_assertion.py b/tests/server/auth/oauth_proxy/test_identity_assertion.py
new file mode 100644
index 000000000..6a9db3ddb
--- /dev/null
+++ b/tests/server/auth/oauth_proxy/test_identity_assertion.py
@@ -0,0 +1,992 @@
+"""Tests for server-side SEP-990 identity assertion (ID-JAG) support.
+
+These exercise the OAuthProxy token endpoint end-to-end using a locally-minted
+fake IdP JWT (a keypair is generated per test). The proxy's JWKS lookup is
+served via httpx_mock, so no real network calls are made.
+"""
+
+import subprocess
+import sys
+import time
+
+import httpx2
+import pytest
+from joserfc import jwk, jwt
+from key_value.aio.stores.memory import MemoryStore
+from mcp.shared.auth import OAuthClientInformationFull
+from pydantic import AnyUrl
+
+from fastmcp import FastMCP
+from fastmcp.server.auth import IdentityAssertion
+from fastmcp.server.auth.identity_assertion import (
+ ID_JAG_GRANT_PROFILE,
+ ID_JAG_TYP,
+ JWT_BEARER_GRANT_TYPE,
+)
+from fastmcp.server.auth.oauth_proxy import OAuthProxy
+from fastmcp.server.auth.oauth_proxy.models import ProxyDCRClient
+from fastmcp.server.auth.providers.jwt import RSAKeyPair
+from tests.server.auth.oauth_proxy.conftest import MockTokenVerifier
+from tests.utilities.httpx2_mock import HTTPXMock
+
+BASE_URL = "https://myserver.com"
+ISSUER = "https://login.acme-corp.com"
+JWKS_URI = "https://login.acme-corp.com/jwks"
+RESOURCE = f"{BASE_URL}/mcp"
+
+
+def _b64url_json(value: object) -> str:
+ """Base64url-encode a JSON value as a JWT segment (no padding)."""
+ import base64
+ import json
+
+ raw = json.dumps(value).encode()
+ return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
+
+
+def _idp_jwks(key_pair: RSAKeyPair) -> dict:
+ """Build a JWKS document from an RSA key pair's public key."""
+ public_key = jwk.import_key(key_pair.public_key, "RSA")
+ data = public_key.as_dict()
+ data["kid"] = "idp-key-1"
+ data["alg"] = "RS256"
+ return {"keys": [data]}
+
+
+def _mint_id_jag(
+ key_pair: RSAKeyPair,
+ *,
+ issuer: str = ISSUER,
+ audience: str = BASE_URL,
+ subject: str = "employee@acme-corp.com",
+ typ: str = ID_JAG_TYP,
+ jti: str = "jti-1",
+ expires_in: int = 120,
+ scope: str | None = None,
+ include_iat: bool = True,
+ nbf_offset: int | None = None,
+ client_id: str | None = "mcp-client",
+ resource: str | None = RESOURCE,
+ claim_overrides: dict | None = None,
+) -> str:
+ """Mint a fake ID-JAG JWT with full control over header and claims.
+
+ `client_id` and `resource` default to values matching the standard test
+ client and this server's resource URL — SEP-990 binds the assertion to
+ both, and the exchange enforces the bindings. Pass `None` to omit.
+ `claim_overrides` is merged in last, for injecting malformed values.
+ """
+ now = int(time.time())
+ header = {"alg": "RS256", "typ": typ, "kid": "idp-key-1"}
+ payload: dict = {
+ "iss": issuer,
+ "aud": audience,
+ "sub": subject,
+ "exp": now + expires_in,
+ "jti": jti,
+ }
+ if client_id is not None:
+ payload["client_id"] = client_id
+ if resource is not None:
+ payload["resource"] = resource
+ if include_iat:
+ payload["iat"] = now
+ if nbf_offset is not None:
+ payload["nbf"] = now + nbf_offset
+ if scope is not None:
+ payload["scope"] = scope
+ if claim_overrides:
+ payload.update(claim_overrides)
+ signing_key = jwk.import_key(key_pair.private_key.get_secret_value(), "RSA")
+ return jwt.encode(header, payload, signing_key, algorithms=["RS256"])
+
+
+def _make_proxy(identity_assertion: IdentityAssertion | None) -> OAuthProxy:
+ return OAuthProxy(
+ upstream_authorization_endpoint="https://login.acme-corp.com/authorize",
+ upstream_token_endpoint="https://login.acme-corp.com/token",
+ upstream_client_id="upstream-client",
+ upstream_client_secret="upstream-secret",
+ token_verifier=MockTokenVerifier(),
+ base_url=BASE_URL,
+ jwt_signing_key="test-signing-key",
+ client_storage=MemoryStore(),
+ identity_assertion=identity_assertion,
+ )
+
+
+@pytest.fixture
+def idp_key() -> RSAKeyPair:
+ return RSAKeyPair.generate()
+
+
+@pytest.fixture
+def config() -> IdentityAssertion:
+ # Explicit jwks_uris avoids OIDC discovery so only the JWKS fetch is mocked.
+ return IdentityAssertion(
+ trusted_issuers=[ISSUER],
+ jwks_uris={ISSUER: JWKS_URI},
+ )
+
+
+async def _register_client(proxy: OAuthProxy) -> None:
+ """Register the MCP client so the token endpoint can authenticate it."""
+ await proxy.register_client(
+ OAuthClientInformationFull(
+ client_id="mcp-client",
+ client_secret="mcp-secret",
+ redirect_uris=[AnyUrl("http://localhost/callback")],
+ grant_types=[JWT_BEARER_GRANT_TYPE],
+ )
+ )
+
+
+async def _post_token(
+ proxy: OAuthProxy,
+ assertion: str,
+ *,
+ request_scope: str | None = None,
+ resource: str | None = None,
+ register: bool = True,
+) -> httpx2.Response:
+ """POST a jwt-bearer grant to the proxy's /token endpoint via an ASGI app.
+
+ When ``register`` is True the standard test client is registered first; pass
+ ``register=False`` to exercise a client the caller has already stored.
+ """
+ if register:
+ await _register_client(proxy)
+ app = FastMCP("ID-JAG Server", auth=proxy).http_app()
+ transport = httpx2.ASGITransport(app=app)
+ data = {
+ "grant_type": JWT_BEARER_GRANT_TYPE,
+ "assertion": assertion,
+ "client_id": "mcp-client",
+ "client_secret": "mcp-secret",
+ }
+ if request_scope is not None:
+ data["scope"] = request_scope
+ if resource is not None:
+ data["resource"] = resource
+ async with httpx2.AsyncClient(transport=transport, base_url=BASE_URL) as client:
+ return await client.post("/token", data=data)
+
+
+class TestIdentityAssertionConfig:
+ def test_requires_trusted_issuers(self):
+ with pytest.raises(ValueError):
+ IdentityAssertion(trusted_issuers=[])
+
+ def test_rejects_blank_issuer(self):
+ with pytest.raises(ValueError):
+ IdentityAssertion(trusted_issuers=[" "])
+
+ @pytest.mark.parametrize("algorithm", ["ES256", "PS256", "RS384"])
+ def test_accepts_asymmetric_algorithm(self, algorithm: str):
+ IdentityAssertion(trusted_issuers=[ISSUER], algorithm=algorithm)
+
+ @pytest.mark.parametrize(
+ "algorithm", ["HS256", "EdDSA", "none", "", "RS999", "ES999"]
+ )
+ def test_rejects_incompatible_or_unsupported_algorithm(self, algorithm: str):
+ # HS* has no JWKS equivalent (shared secret, not a public key); EdDSA,
+ # typo'd variants like RS999, and other unimportable algorithms would
+ # otherwise surface as a 500 on the first exchange rather than a clean
+ # config error now. The allowlist is exactly what JWTVerifier supports.
+ with pytest.raises(ValueError):
+ IdentityAssertion(trusted_issuers=[ISSUER], algorithm=algorithm)
+
+ def test_defaults(self):
+ cfg = IdentityAssertion(trusted_issuers=[ISSUER])
+ assert cfg.access_token_expiry_seconds == 300
+ assert cfg.audience is None
+ assert cfg.jwks_uris is None
+
+ def test_per_issuer_algorithms_validated(self):
+ IdentityAssertion(trusted_issuers=[ISSUER], algorithms={ISSUER: "ES256"})
+ with pytest.raises(ValueError):
+ IdentityAssertion(trusted_issuers=[ISSUER], algorithms={ISSUER: "HS256"})
+
+ def test_lazy_reexport_does_not_import_module(self):
+ # fastmcp.server.auth must not load identity_assertion (and its
+ # httpx2 dependency) eagerly — the re-export is lazy via __getattr__.
+ code = (
+ "import sys\n"
+ "import fastmcp.server.auth\n"
+ "loaded = [m for m in sys.modules if 'identity_assertion' in m]\n"
+ "assert not loaded, f'eagerly loaded: {loaded}'\n"
+ "from fastmcp.server.auth import IdentityAssertion\n"
+ "print('OK')\n"
+ )
+ result = subprocess.run(
+ [sys.executable, "-c", code], capture_output=True, text=True
+ )
+ assert result.returncode == 0, result.stderr
+ assert "OK" in result.stdout
+
+
+class TestMetadataAdvertisement:
+ async def _metadata(self, proxy: OAuthProxy) -> dict:
+ app = FastMCP("ID-JAG Server", auth=proxy).http_app()
+ transport = httpx2.ASGITransport(app=app)
+ async with httpx2.AsyncClient(transport=transport, base_url=BASE_URL) as client:
+ resp = await client.get("/.well-known/oauth-authorization-server")
+ return resp.json()
+
+ async def test_advertises_grant_when_enabled(self, config: IdentityAssertion):
+ proxy = _make_proxy(config)
+ metadata = await self._metadata(proxy)
+ assert JWT_BEARER_GRANT_TYPE in metadata["grant_types_supported"]
+ assert (
+ ID_JAG_GRANT_PROFILE in metadata["authorization_grant_profiles_supported"]
+ )
+
+ async def test_not_advertised_when_disabled(self):
+ proxy = _make_proxy(None)
+ metadata = await self._metadata(proxy)
+ assert JWT_BEARER_GRANT_TYPE not in metadata["grant_types_supported"]
+ assert metadata.get("authorization_grant_profiles_supported") is None
+
+ async def test_advertises_none_auth_method_without_cimd(
+ self, config: IdentityAssertion
+ ):
+ # DCR clients are public (token_endpoint_auth_method="none"), so when
+ # the jwt-bearer grant is advertised, `none` must be advertised too —
+ # even with CIMD (which also adds it) disabled.
+ proxy = OAuthProxy(
+ upstream_authorization_endpoint="https://login.acme-corp.com/authorize",
+ upstream_token_endpoint="https://login.acme-corp.com/token",
+ upstream_client_id="upstream-client",
+ upstream_client_secret="upstream-secret",
+ token_verifier=MockTokenVerifier(),
+ base_url=BASE_URL,
+ jwt_signing_key="test-signing-key",
+ client_storage=MemoryStore(),
+ identity_assertion=config,
+ enable_cimd=False,
+ )
+ metadata = await self._metadata(proxy)
+ assert JWT_BEARER_GRANT_TYPE in metadata["grant_types_supported"]
+ assert "none" in metadata["token_endpoint_auth_methods_supported"]
+
+
+class TestTokenEndpoint:
+ async def test_happy_path_issues_token(
+ self,
+ idp_key: RSAKeyPair,
+ config: IdentityAssertion,
+ httpx_mock: HTTPXMock,
+ ):
+ httpx_mock.add_response(url=JWKS_URI, json=_idp_jwks(idp_key))
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, scope="read write")
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 200
+ body = resp.json()
+ assert body["token_type"] == "Bearer"
+ assert body["access_token"]
+ # SEP-990: no refresh token is issued.
+ assert body.get("refresh_token") is None
+ assert body["expires_in"] == config.access_token_expiry_seconds
+
+ async def test_issued_token_carries_subject(
+ self,
+ idp_key: RSAKeyPair,
+ config: IdentityAssertion,
+ httpx_mock: HTTPXMock,
+ ):
+ httpx_mock.add_response(url=JWKS_URI, json=_idp_jwks(idp_key))
+ proxy = _make_proxy(config)
+ proxy.set_mcp_path("/mcp")
+ assertion = _mint_id_jag(idp_key, subject="alice@acme-corp.com")
+
+ resp = await _post_token(proxy, assertion)
+ access_token = resp.json()["access_token"]
+
+ # The FastMCP-issued token validates via the proxy and exposes the subject.
+ loaded = await proxy.load_access_token(access_token)
+ assert loaded is not None
+ assert loaded.subject == "alice@acme-corp.com"
+
+ async def test_asserted_subject_flows_into_auth_context(
+ self,
+ idp_key: RSAKeyPair,
+ config: IdentityAssertion,
+ httpx_mock: HTTPXMock,
+ ):
+ """The issued token, verified via the same path the bearer-auth middleware
+ uses (`verify_token` -> `load_access_token`), exposes the asserted subject —
+ which is exactly what `get_access_token()` returns to a tool."""
+ httpx_mock.add_response(url=JWKS_URI, json=_idp_jwks(idp_key))
+ proxy = _make_proxy(config)
+ proxy.set_mcp_path("/mcp")
+ assertion = _mint_id_jag(
+ idp_key, subject="carol@acme-corp.com", scope="read write"
+ )
+
+ resp = await _post_token(proxy, assertion)
+ access_token = resp.json()["access_token"]
+
+ verified = await proxy.verify_token(access_token)
+ assert verified is not None
+ assert verified.subject == "carol@acme-corp.com"
+ assert "read" in verified.scopes and "write" in verified.scopes
+
+ async def test_grant_rejected_when_not_configured(
+ self,
+ idp_key: RSAKeyPair,
+ ):
+ proxy = _make_proxy(None)
+ assertion = _mint_id_jag(idp_key)
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 400
+ assert resp.json()["error"] == "unsupported_grant_type"
+
+ async def test_request_scope_cannot_widen_assertion_scope(
+ self,
+ idp_key: RSAKeyPair,
+ config: IdentityAssertion,
+ httpx_mock: HTTPXMock,
+ ):
+ """A client whose assertion grants only `readonly` cannot obtain `admin`
+ by asking for it at the token endpoint. The request `scope` is not covered
+ by the signed assertion, so it may only narrow the granted set."""
+ httpx_mock.add_response(url=JWKS_URI, json=_idp_jwks(idp_key))
+ proxy = _make_proxy(config)
+ proxy.set_mcp_path("/mcp")
+ assertion = _mint_id_jag(idp_key, scope="readonly")
+
+ resp = await _post_token(proxy, assertion, request_scope="admin")
+
+ assert resp.status_code == 200
+ verified = await proxy.verify_token(resp.json()["access_token"])
+ assert verified is not None
+ assert "admin" not in verified.scopes
+
+ async def test_request_scope_narrows_assertion_scope(
+ self,
+ idp_key: RSAKeyPair,
+ config: IdentityAssertion,
+ httpx_mock: HTTPXMock,
+ ):
+ """When the request `scope` is a subset of the assertion's granted scopes,
+ the issued token carries only the intersection."""
+ httpx_mock.add_response(url=JWKS_URI, json=_idp_jwks(idp_key))
+ proxy = _make_proxy(config)
+ proxy.set_mcp_path("/mcp")
+ assertion = _mint_id_jag(idp_key, scope="read write")
+
+ resp = await _post_token(proxy, assertion, request_scope="read")
+
+ assert resp.status_code == 200
+ verified = await proxy.verify_token(resp.json()["access_token"])
+ assert verified is not None
+ assert "read" in verified.scopes
+ assert "write" not in verified.scopes
+
+ async def test_request_narrowing_preserves_required_scope(
+ self,
+ idp_key: RSAKeyPair,
+ httpx_mock: HTTPXMock,
+ ):
+ """A configured `required_scope` the assertion grants must always ride on
+ the issued token; the request `scope` may only narrow the optional
+ remainder. Requesting `read` must not drop the mandatory `admin`."""
+ httpx_mock.add_response(url=JWKS_URI, json=_idp_jwks(idp_key))
+ config = IdentityAssertion(
+ trusted_issuers=[ISSUER],
+ jwks_uris={ISSUER: JWKS_URI},
+ required_scopes=["admin"],
+ )
+ proxy = _make_proxy(config)
+ proxy.set_mcp_path("/mcp")
+ assertion = _mint_id_jag(idp_key, scope="admin read")
+
+ resp = await _post_token(proxy, assertion, request_scope="read")
+
+ assert resp.status_code == 200
+ verified = await proxy.verify_token(resp.json()["access_token"])
+ assert verified is not None
+ assert "admin" in verified.scopes
+ assert "read" in verified.scopes
+
+ async def test_request_for_required_scope_only(
+ self,
+ idp_key: RSAKeyPair,
+ httpx_mock: HTTPXMock,
+ ):
+ """Requesting only the required scope keeps it and narrows away the rest."""
+ httpx_mock.add_response(url=JWKS_URI, json=_idp_jwks(idp_key))
+ config = IdentityAssertion(
+ trusted_issuers=[ISSUER],
+ jwks_uris={ISSUER: JWKS_URI},
+ required_scopes=["admin"],
+ )
+ proxy = _make_proxy(config)
+ proxy.set_mcp_path("/mcp")
+ assertion = _mint_id_jag(idp_key, scope="admin read")
+
+ resp = await _post_token(proxy, assertion, request_scope="admin")
+
+ assert resp.status_code == 200
+ verified = await proxy.verify_token(resp.json()["access_token"])
+ assert verified is not None
+ assert "admin" in verified.scopes
+ assert "read" not in verified.scopes
+
+
+@pytest.mark.httpx_mock(assert_all_responses_were_requested=False)
+class TestValidationMatrix:
+ @pytest.fixture(autouse=True)
+ def _mock_jwks(self, idp_key: RSAKeyPair, httpx_mock: HTTPXMock):
+ # Optional: several matrix tests reject before any JWKS fetch happens.
+ httpx_mock.add_response(url=JWKS_URI, json=_idp_jwks(idp_key), is_optional=True)
+
+ async def test_untrusted_issuer_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, issuer="https://evil.example.com")
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_wrong_audience_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, audience="https://other-server.com")
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_wrong_typ_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, typ="JWT")
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_expired_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, expires_in=-10)
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_replayed_jti_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, jti="replay-me")
+
+ first = await _post_token(proxy, assertion)
+ second = await _post_token(proxy, assertion)
+
+ assert first.status_code == 200
+ assert second.status_code == 401
+ assert second.json()["error"] == "invalid_grant"
+
+ async def test_wrong_signature_rejected(self, config: IdentityAssertion):
+ # Sign with a different key than the one served in the JWKS.
+ other_key = RSAKeyPair.generate()
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(other_key)
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_missing_required_scope_rejected(self, idp_key: RSAKeyPair):
+ config = IdentityAssertion(
+ trusted_issuers=[ISSUER],
+ jwks_uris={ISSUER: JWKS_URI},
+ required_scopes=["admin"],
+ )
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, scope="read")
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_future_nbf_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ # An ID-JAG whose not-before (`nbf`) claim is in the future is not yet
+ # valid and must be rejected.
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, nbf_offset=300)
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_past_nbf_accepted(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ # An `nbf` in the past means the assertion is already valid.
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, nbf_offset=-60)
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 200
+ assert resp.json()["access_token"]
+
+ async def test_aud_matching_advertised_issuer_with_trailing_slash(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ # Metadata advertises the issuer exactly as pydantic renders base_url —
+ # a bare domain gains a trailing slash. An IdP that sets `aud` to that
+ # advertised value verbatim must be accepted, not rejected on the slash.
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, audience=f"{BASE_URL}/")
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 200
+
+ @pytest.mark.parametrize("claim", ["exp", "iat", "nbf"])
+ @pytest.mark.parametrize("bad_value", ["not-a-number", [], {}, True])
+ async def test_non_numeric_temporal_claim_rejected(
+ self,
+ idp_key: RSAKeyPair,
+ config: IdentityAssertion,
+ claim: str,
+ bad_value: object,
+ ):
+ # A validly-signed assertion could still carry a malformed exp/iat/nbf
+ # (a misbehaving IdP); comparing against it must map to invalid_grant,
+ # not an unhandled TypeError. `True`/`False` are excluded even though
+ # bool subclasses int in Python.
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, claim_overrides={claim: bad_value})
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ @pytest.mark.parametrize("bad_jti", [["a", "b"], {"x": 1}, 42])
+ async def test_non_string_jti_rejected(
+ self,
+ idp_key: RSAKeyPair,
+ config: IdentityAssertion,
+ bad_jti: object,
+ ):
+ # An array/object jti is unhashable — the cache lookup would raise
+ # TypeError (a 500) instead of a clean invalid_grant.
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key, claim_overrides={"jti": bad_jti})
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_non_object_payload_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ # A JWT with a valid typ header but a JSON-array payload must map to a
+ # clean invalid_grant, not an unhandled 500 from calling `.get()` on a list.
+ header = _b64url_json({"alg": "RS256", "typ": ID_JAG_TYP, "kid": "idp-key-1"})
+ payload = _b64url_json([])
+ assertion = f"{header}.{payload}.signature"
+
+ resp = await _post_token(proxy=_make_proxy(config), assertion=assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_non_object_header_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ # Same class of bug as the payload case: a JSON-array JOSE header must
+ # map to invalid_grant, not a 500 from `.get()` on a list.
+ header = _b64url_json([])
+ payload = _b64url_json({"iss": ISSUER, "sub": "x"})
+ assertion = f"{header}.{payload}.signature"
+
+ resp = await _post_token(proxy=_make_proxy(config), assertion=assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_assertion_for_other_client_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ # SEP-990: the IdP signs which client the assertion was minted for.
+ # A registered client must not be able to redeem an assertion minted
+ # for a different client — with public clients, this signed binding is
+ # the control that stops cross-client redemption of leaked assertions.
+ assertion = _mint_id_jag(idp_key, client_id="some-other-client")
+
+ resp = await _post_token(proxy=_make_proxy(config), assertion=assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_wrong_client_binding_does_not_consume_jti(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ # An assertion presented by the wrong client must be rejected WITHOUT
+ # its jti being recorded as consumed -- otherwise the client it
+ # actually belongs to would find that same jti already "replayed"
+ # when it (or a retry) presents a correctly-bound assertion. Two
+ # distinct, validly-signed tokens sharing one jti value is exactly
+ # the scenario jti replay tracking cares about, regardless of what
+ # else differs between them.
+ proxy = _make_proxy(config)
+ shared_jti = "jti-shared-client"
+ wrong_client = _mint_id_jag(
+ idp_key, client_id="some-other-client", jti=shared_jti
+ )
+
+ rejected = await _post_token(proxy, wrong_client)
+ assert rejected.status_code == 401
+ assert rejected.json()["error"] == "invalid_grant"
+
+ correct_client = _mint_id_jag(idp_key, client_id="mcp-client", jti=shared_jti)
+ accepted = await _post_token(proxy, correct_client, register=False)
+ assert accepted.status_code == 200
+
+ async def test_wrong_resource_binding_does_not_consume_jti(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ proxy = _make_proxy(config)
+ shared_jti = "jti-shared-resource"
+ wrong_resource = _mint_id_jag(
+ idp_key,
+ resource="https://other-server.example.com/mcp",
+ jti=shared_jti,
+ )
+
+ rejected = await _post_token(proxy, wrong_resource)
+ assert rejected.status_code == 401
+ assert rejected.json()["error"] == "invalid_grant"
+
+ correct_resource = _mint_id_jag(idp_key, jti=shared_jti) # default = RESOURCE
+ accepted = await _post_token(proxy, correct_resource, register=False)
+ assert accepted.status_code == 200
+
+ async def test_assertion_without_client_id_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ assertion = _mint_id_jag(idp_key, client_id=None)
+
+ resp = await _post_token(proxy=_make_proxy(config), assertion=assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_assertion_for_other_resource_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ # The signed resource claim governs: an assertion minted for server A
+ # must not be redeemable at server B behind the same IdP.
+ assertion = _mint_id_jag(
+ idp_key, resource="https://other-server.example.com/mcp"
+ )
+
+ resp = await _post_token(proxy=_make_proxy(config), assertion=assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_assertion_without_resource_rejected(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ # When the proxy knows its resource URL, an assertion that names no
+ # resource cannot be audience-restricted per SEP-990 and is rejected.
+ assertion = _mint_id_jag(idp_key, resource=None)
+
+ resp = await _post_token(proxy=_make_proxy(config), assertion=assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_resource_mismatch_rejected(
+ self,
+ idp_key: RSAKeyPair,
+ config: IdentityAssertion,
+ ):
+ # RFC 8707: a token request naming a different resource must get
+ # invalid_target (mirrors the authorize() invariant), not a token
+ # for this server. Rejection happens before any JWKS fetch.
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key)
+
+ resp = await _post_token(
+ proxy, assertion, resource="https://other-server.example.com/mcp"
+ )
+
+ assert resp.status_code == 400
+ assert resp.json()["error"] == "invalid_target"
+
+ async def test_matching_resource_accepted(
+ self,
+ idp_key: RSAKeyPair,
+ config: IdentityAssertion,
+ ):
+ # JWKS served by the class's autouse _mock_jwks fixture.
+ proxy = _make_proxy(config)
+ assertion = _mint_id_jag(idp_key)
+
+ resp = await _post_token(proxy, assertion, resource=f"{BASE_URL}/mcp")
+
+ assert resp.status_code == 200
+
+ async def test_jti_cache_does_not_grow_past_capacity(
+ self, idp_key: RSAKeyPair, config: IdentityAssertion
+ ):
+ # Once the JTI cache is full of still-valid entries, further fresh
+ # assertions are rejected as overloaded WITHOUT being inserted, so the
+ # cache never grows beyond its cap.
+ proxy = _make_proxy(config)
+ validator = proxy._identity_assertion_validator
+ assert validator is not None
+ validator._jti_cache_max_size = 2
+ future = time.time() + 120
+ validator._jti_cache = {"filler-a": future, "filler-b": future}
+
+ for i in range(3):
+ assertion = _mint_id_jag(idp_key, jti=f"fresh-{i}")
+ resp = await _post_token(proxy, assertion)
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ assert len(validator._jti_cache) == 2
+
+
+class TestAlgorithmConfig:
+ """Non-RS256 issuers work when `algorithm` is configured.
+
+ Lives outside TestValidationMatrix because that class's autouse
+ `_mock_jwks` fixture pre-registers an RSA JWKS for the same URL, and
+ pytest-httpx serves first-registered responses first.
+ """
+
+ async def test_es256_issuer_supported_via_algorithm_config(
+ self, httpx_mock: HTTPXMock
+ ):
+ ec_key = jwk.ECKey.generate_key("P-256")
+ jwks_entry = ec_key.as_dict(private=False)
+ jwks_entry["kid"] = "idp-ec-1"
+ jwks_entry["alg"] = "ES256"
+ httpx_mock.add_response(url=JWKS_URI, json={"keys": [jwks_entry]})
+
+ now = int(time.time())
+ header = {"alg": "ES256", "typ": ID_JAG_TYP, "kid": "idp-ec-1"}
+ payload = {
+ "iss": ISSUER,
+ "aud": BASE_URL,
+ "sub": "employee@acme-corp.com",
+ "exp": now + 120,
+ "iat": now,
+ "jti": "jti-es256-1",
+ "client_id": "mcp-client",
+ "resource": RESOURCE,
+ }
+ assertion = jwt.encode(header, payload, ec_key, algorithms=["ES256"])
+
+ es_config = IdentityAssertion(
+ trusted_issuers=[ISSUER],
+ jwks_uris={ISSUER: JWKS_URI},
+ algorithm="ES256",
+ )
+ resp = await _post_token(proxy=_make_proxy(es_config), assertion=assertion)
+ assert resp.status_code == 200
+
+
+class TestIssuerKeyDiscovery:
+ async def test_jwks_discovered_via_oidc(
+ self, idp_key: RSAKeyPair, httpx_mock: HTTPXMock
+ ):
+ # No explicit jwks_uris: the validator must discover the JWKS URI from
+ # the issuer's OIDC configuration document.
+ oidc_config_url = ISSUER.rstrip("/") + "/.well-known/openid-configuration"
+ httpx_mock.add_response(url=oidc_config_url, json={"jwks_uri": JWKS_URI})
+ httpx_mock.add_response(url=JWKS_URI, json=_idp_jwks(idp_key))
+
+ proxy = _make_proxy(IdentityAssertion(trusted_issuers=[ISSUER]))
+ assertion = _mint_id_jag(idp_key)
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 200
+ assert resp.json()["access_token"]
+
+ async def test_non_object_discovery_body_rejected(
+ self, idp_key: RSAKeyPair, httpx_mock: HTTPXMock
+ ):
+ # A discovery endpoint returning valid JSON that isn't an object (e.g.
+ # a bare array) must map to invalid_grant, not a 500 on `.get()`.
+ oidc_config_url = ISSUER.rstrip("/") + "/.well-known/openid-configuration"
+ httpx_mock.add_response(url=oidc_config_url, json=[])
+
+ proxy = _make_proxy(IdentityAssertion(trusted_issuers=[ISSUER]))
+ assertion = _mint_id_jag(idp_key)
+
+ resp = await _post_token(proxy, assertion)
+
+ assert resp.status_code == 401
+ assert resp.json()["error"] == "invalid_grant"
+
+ async def test_failed_discovery_backs_off(
+ self, idp_key: RSAKeyPair, httpx_mock: HTTPXMock
+ ):
+ # Discovery runs before signature verification, so repeated garbage
+ # with a trusted iss must not turn into an outbound HTTP call per
+ # request: after a failure, subsequent requests fast-fail without
+ # fetching until the cooldown elapses.
+ oidc_config_url = ISSUER.rstrip("/") + "/.well-known/openid-configuration"
+ httpx_mock.add_exception(
+ httpx2.ConnectError("connection refused"), url=oidc_config_url
+ )
+
+ proxy = _make_proxy(IdentityAssertion(trusted_issuers=[ISSUER]))
+
+ first = await _post_token(proxy, _mint_id_jag(idp_key, jti="jti-d1"))
+ assert first.status_code == 401
+
+ second = await _post_token(
+ proxy, _mint_id_jag(idp_key, jti="jti-d2"), register=False
+ )
+ assert second.status_code == 401
+ # Only the FIRST request hit the network; the second fast-failed
+ # inside the cooldown window.
+ assert len(httpx_mock.get_requests()) == 1
+
+
+class TestRevocation:
+ async def test_revoked_id_jag_token_rejected(
+ self,
+ idp_key: RSAKeyPair,
+ httpx_mock: HTTPXMock,
+ ):
+ # ID-JAG access tokens are self-contained — nothing upstream knows
+ # them, so revocation must be tracked locally. After revoke_token,
+ # load_access_token rejects the token for its remaining lifetime.
+ httpx_mock.add_response(url=JWKS_URI, json=_idp_jwks(idp_key))
+ proxy = _make_proxy(
+ IdentityAssertion(trusted_issuers=[ISSUER], jwks_uris={ISSUER: JWKS_URI})
+ )
+ resp = await _post_token(proxy, _mint_id_jag(idp_key))
+ assert resp.status_code == 200
+ issued = resp.json()["access_token"]
+
+ loaded = await proxy.load_access_token(issued)
+ assert loaded is not None
+
+ await proxy.revoke_token(loaded)
+
+ assert await proxy.load_access_token(issued) is None
+
+
+class TestGrantTypeEnforcement:
+ """The proxy dispatches the jwt-bearer grant itself, so it must enforce the
+ registered-grant-type constraint the SDK would otherwise apply: only clients
+ registered for the jwt-bearer grant may present an ID-JAG. DCR adds the grant
+ to registered clients when identity assertion is enabled."""
+
+ async def test_client_not_registered_for_jwt_bearer_rejected(
+ self,
+ idp_key: RSAKeyPair,
+ config: IdentityAssertion,
+ ):
+ # The grant-type check rejects before any assertion validation, so no
+ # JWKS fetch occurs.
+ proxy = _make_proxy(config)
+ # A client registered only for the standard grants (e.g. before identity
+ # assertion was enabled). Stored directly so the DCR enabled-path does not
+ # add jwt-bearer for us.
+ await proxy._client_store.put(
+ key="mcp-client",
+ value=ProxyDCRClient(
+ client_id="mcp-client",
+ client_secret=None,
+ redirect_uris=[AnyUrl("http://localhost/callback")],
+ grant_types=["authorization_code", "refresh_token"],
+ token_endpoint_auth_method="none",
+ ),
+ )
+ assertion = _mint_id_jag(idp_key, scope="read")
+
+ resp = await _post_token(proxy, assertion, register=False)
+
+ assert resp.status_code == 400
+ assert resp.json()["error"] == "unsupported_grant_type"
+
+ async def test_dcr_adds_jwt_bearer_when_enabled(self, config: IdentityAssertion):
+ proxy = _make_proxy(config)
+ await proxy.register_client(
+ OAuthClientInformationFull(
+ client_id="dcr-client",
+ redirect_uris=[AnyUrl("http://localhost/callback")],
+ grant_types=["authorization_code", "refresh_token"],
+ )
+ )
+
+ client = await proxy.get_client("dcr-client")
+
+ assert client is not None
+ assert JWT_BEARER_GRANT_TYPE in client.grant_types
+
+ async def test_dcr_does_not_add_jwt_bearer_when_disabled(self):
+ proxy = _make_proxy(None)
+ await proxy.register_client(
+ OAuthClientInformationFull(
+ client_id="dcr-client",
+ redirect_uris=[AnyUrl("http://localhost/callback")],
+ grant_types=["authorization_code", "refresh_token"],
+ )
+ )
+
+ client = await proxy.get_client("dcr-client")
+
+ assert client is not None
+ assert JWT_BEARER_GRANT_TYPE not in client.grant_types
+
+ async def test_dcr_registered_client_can_exchange(
+ self,
+ idp_key: RSAKeyPair,
+ config: IdentityAssertion,
+ httpx_mock: HTTPXMock,
+ ):
+ """A client that registers via DCR without the jwt-bearer grant can still
+ exchange an ID-JAG, because the enabled-path adds the grant on registration."""
+ httpx_mock.add_response(url=JWKS_URI, json=_idp_jwks(idp_key))
+ proxy = _make_proxy(config)
+ await proxy.register_client(
+ OAuthClientInformationFull(
+ client_id="mcp-client",
+ redirect_uris=[AnyUrl("http://localhost/callback")],
+ grant_types=["authorization_code", "refresh_token"],
+ )
+ )
+ assertion = _mint_id_jag(idp_key, scope="read")
+
+ resp = await _post_token(proxy, assertion, register=False)
+
+ assert resp.status_code == 200
+ assert resp.json()["access_token"]
diff --git a/tests/utilities/httpx2_mock.py b/tests/utilities/httpx2_mock.py
index 911b29a9d..b565a849f 100644
--- a/tests/utilities/httpx2_mock.py
+++ b/tests/utilities/httpx2_mock.py
@@ -55,9 +55,11 @@ class _Matcher:
self,
url: str | re.Pattern[str] | httpx2.URL | None,
method: str | None,
+ is_optional: bool = False,
) -> None:
self.url = httpx2.URL(url) if isinstance(url, str) else url
self.method = method.upper() if method else method
+ self.is_optional = is_optional
self.nb_calls = 0
def match(self, request: httpx2.Request) -> bool:
@@ -105,6 +107,7 @@ class HTTPXMock:
*,
url: str | re.Pattern[str] | httpx2.URL | None = None,
method: str | None = None,
+ is_optional: bool = False,
) -> None:
json = copy.deepcopy(json) if json is not None else None
@@ -119,7 +122,7 @@ class HTTPXMock:
stream=stream,
)
- self._callbacks.append((_Matcher(url, method), callback))
+ self._callbacks.append((_Matcher(url, method, is_optional), callback))
def add_exception(
self,
@@ -203,7 +206,9 @@ class HTTPXMock:
def _assert_options(self) -> None:
not_requested = [
- str(matcher) for matcher, _ in self._callbacks if not matcher.nb_calls
+ str(matcher)
+ for matcher, _ in self._callbacks
+ if not matcher.nb_calls and not matcher.is_optional
]
assert not not_requested, (
"The following responses are mocked but not requested:\n"