mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 07:09:11 +02:00
Add server-side identity assertion (SEP-990 ID-JAG) (#4483)
* Add server-side SEP-990 identity assertion (ID-JAG)
* Test SEP-990 identity assertion token endpoint
* Format identity assertion test
* Document SEP-990 identity assertion
* Thread identity_assertion through OIDCProxy
* Harden ID-JAG: authoritative scopes and grant-type enforcement
Scopes for the issued token now derive only from the signed assertion (or server policy when it omits scopes); the client-supplied request scope may narrow but never widen them. Enforce the registered grant-type constraint the SDK check bypassed, and have proxy DCR add the jwt-bearer grant to registered clients when identity assertion is enabled.
* Harden ID-JAG: honor nbf, reject non-object payload, bound jti cache, preserve required_scopes
* Document per-process ID-JAG replay limitation and nbf check
* Harden ID-JAG round 3: resource indicator, non-object header, algorithm config
- Honor RFC 8707 resource on the jwt-bearer grant (invalid_target on
mismatch), mirroring authorize()'s invariant incl. skip-when-unconfigured
- Reject JSON-array JOSE headers with invalid_grant instead of a 500
- Add IdentityAssertion.algorithm so ES256/PS256 issuers can be verified
(JWTVerifier otherwise defaults to RS256)
* Bind ID-JAG exchange to the assertion's signed client_id and resource
SEP-990: the IdP signs which client and which resource the assertion was
minted for. With public proxy clients the presented client_id is
self-asserted, so the signed binding is what stops client B redeeming
client A's leaked assertion — and the signed resource claim stops an
assertion for server A being redeemed at server B behind the same IdP.
* Harden ID-JAG round 4: check bindings before jti consumption; validate temporal claims, algorithm, and discovery body
- Move the client_id/resource binding checks into the validator itself,
before jti is recorded as consumed, so an assertion presented with the
wrong binding is rejected without burning replay protection for whoever
it actually belongs to
- Reject non-numeric exp/iat/nbf with invalid_grant instead of a 500
- Validate IdentityAssertion.algorithm at config time (must be an
asymmetric JWS algorithm verifiable via JWKS)
- Reject a non-object OIDC discovery body with invalid_grant instead of a 500
- Centralize the resource-URL comparison helpers used by both the
validator and OAuthProxy.authorize()
* Rebase onto httpx2/SDK b2 and harden ID-JAG round 5
- Migrate identity assertion + tests to httpx2 and the local httpx2_mock
(legacy httpx is now banned; pytest-httpx no longer intercepts)
- Add is_optional to the shared httpx2_mock, mirroring pytest-httpx
- Tighten the algorithm allowlist to JWTVerifier's exact supported set
(prefix check accepted typos like RS999 -> 500 on first exchange)
- Reject non-string jti before the cache lookup (unhashable -> 500)
- Track revocation for self-contained ID-JAG tokens: revoke_token records
the jti and load_access_token rejects it until natural expiry
- Dedupe resource-URL helpers: proxy now imports the shared
normalize_resource_url/server_url_has_query from identity_assertion
* Advertise 'none' token-endpoint auth method when ID-JAG is enabled without CIMD
DCR clients are public, so metadata consumers must see 'none' to use the
advertised jwt-bearer grant; previously only the CIMD path added it.
* Document 2026-07-28 protocol support as a distinct feature catalog
SEP-990 identity assertion leads: the SDK provides the wire contract and
provider hook; FastMCP provides the complete server-side implementation.
Inventories the full modern-era capability set for v4.
* Harden ID-JAG round 6: lazy re-export, dual-form audience, per-issuer algorithms, discovery backoff
- IdentityAssertion re-exported lazily from server.auth (the eager import
bypassed the package's documented lazy-import boundary)
- Accept the ID-JAG aud both with and without base_url's trailing slash;
metadata advertises the slashed form, so IdPs echoing it verbatim work
- algorithms={issuer: alg} per-issuer override, mirroring jwks_uris
- OIDC discovery serializes per-issuer and backs off 30s after a failure
(discovery runs pre-signature, so garbage could amplify into HTTP floods)
This commit is contained in:
Jeremiah Lowin
GitHub
parent
f018f68bbf
commit
998b37f32b
14 changed files with 2069 additions and 58 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
52
docs/development/v4-notes/protocol-2026.mdx
Normal file
52
docs/development/v4-notes/protocol-2026.mdx
Normal file
|
|
@ -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.
|
||||
|
|
@ -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"
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -653,6 +653,83 @@ auth = OAuthProxy(
|
|||
)
|
||||
```
|
||||
|
||||
## Identity Assertion (SEP-990)
|
||||
|
||||
<VersionBadge version="4.0.0" />
|
||||
|
||||
<Tip>
|
||||
Identity assertion is a beta feature. The API may change in a future release.
|
||||
</Tip>
|
||||
|
||||
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.
|
||||
|
||||
<Warning>
|
||||
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.
|
||||
</Warning>
|
||||
|
||||
## Security
|
||||
|
||||
### Key and Storage Management
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
531
fastmcp_slim/fastmcp/server/auth/identity_assertion.py
Normal file
531
fastmcp_slim/fastmcp/server/auth/identity_assertion.py
Normal file
|
|
@ -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))
|
||||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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"]
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
992
tests/server/auth/oauth_proxy/test_identity_assertion.py
Normal file
992
tests/server/auth/oauth_proxy/test_identity_assertion.py
Normal file
|
|
@ -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"]
|
||||
|
|
@ -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"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue