ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-21 13:04:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add verify_id_token option to OIDCProxy (#3248)

Browse source 
* Add verify_id_token option to OIDCProxy

Closes #3240

* chore: Update SDK documentation

* Preserve raw_token_data fields across token refresh

* chore: Update SDK documentation

* Use client_id as verifier audience in verify_id_token mode

* chore: Update SDK documentation

* Return upstream access_token in AccessToken when verifying id_token

* chore: Update SDK documentation

* Skip verifier scope checks in verify_id_token mode

* chore: Update SDK documentation

* Recompute derived scope state after restoring required_scopes in verify_id_token mode

* chore: Update SDK documentation

---------

Co-authored-by: marvin-context-protocol[bot] <225465937+marvin-context-protocol[bot]@users.noreply.github.com>
This commit is contained in:
Jeremiah Lowin 2026-02-20 14:20:47 -05:00 • committed by GitHub
commit f84b2dae04
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 375 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

6
docs/python-sdk/fastmcp-server-auth-oauth_proxy-proxy.mdx
View file

@ -273,7 +273,7 @@ Implements two-tier refresh:
6. Keep same FastMCP refresh token (unless upstream rotates)
#### `load_access_token` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oauth_proxy/proxy.py#L1370" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
#### `load_access_token` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oauth_proxy/proxy.py#L1384" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
```python
load_access_token(self, token: str) -> AccessToken | None
@ -292,7 +292,7 @@ The FastMCP JWT is a reference token - all authorization data comes
from validating the upstream token via the TokenVerifier.
#### `revoke_token` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oauth_proxy/proxy.py#L1429" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
#### `revoke_token` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oauth_proxy/proxy.py#L1460" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
```python
revoke_token(self, token: AccessToken | RefreshToken) -> None
@ -305,7 +305,7 @@ For all tokens, attempts upstream revocation if endpoint is configured.
Access token JTI mappings expire via TTL.
#### `get_routes` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oauth_proxy/proxy.py#L1462" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
#### `get_routes` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oauth_proxy/proxy.py#L1493" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
```python
get_routes(self, mcp_path: str | None = None) -> list[Route]

10
docs/python-sdk/fastmcp-server-auth-oidc_proxy.mdx
View file

@ -19,7 +19,7 @@ This implementation is based on:
## Classes
### `OIDCConfiguration` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oidc_proxy.py#L27" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
### `OIDCConfiguration` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oidc_proxy.py#L28" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
OIDC Configuration.
@ -27,7 +27,7 @@ OIDC Configuration.
**Methods:**
#### `get_oidc_configuration` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oidc_proxy.py#L142" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
#### `get_oidc_configuration` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oidc_proxy.py#L143" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
```python
get_oidc_configuration(cls, config_url: AnyHttpUrl) -> Self
@ -41,7 +41,7 @@ Get the OIDC configuration for the specified config URL.
- `timeout_seconds`: HTTP request timeout in seconds
### `OIDCProxy` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oidc_proxy.py#L172" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
### `OIDCProxy` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oidc_proxy.py#L173" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
OAuth provider that wraps OAuthProxy to provide configuration via an OIDC configuration URL.
@ -52,7 +52,7 @@ that is OIDC compliant.
**Methods:**
#### `get_oidc_configuration` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oidc_proxy.py#L386" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
#### `get_oidc_configuration` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oidc_proxy.py#L432" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
```python
get_oidc_configuration(self, config_url: AnyHttpUrl, strict: bool | None, timeout_seconds: int | None) -> OIDCConfiguration
@ -66,7 +66,7 @@ Gets the OIDC configuration for the specified configuration URL.
- `timeout_seconds`: HTTP request timeout in seconds
#### `get_token_verifier` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oidc_proxy.py#L403" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
#### `get_token_verifier` <sup><a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/fastmcp/server/auth/oidc_proxy.py#L449" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
```python
get_token_verifier(self) -> TokenVerifier

39
src/fastmcp/server/auth/oauth_proxy/proxy.py
View file

@ -1259,7 +1259,10 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
time.time() + new_refresh_expires_in
)
upstream_token_set.raw_token_data = token_response
upstream_token_set.raw_token_data = {
**upstream_token_set.raw_token_data,
**token_response,
}
# Calculate refresh TTL for storage
refresh_ttl = new_refresh_expires_in or (
int(upstream_token_set.refresh_token_expires_at - time.time())
@ -1367,6 +1370,17 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
# Token Validation
# -------------------------------------------------------------------------
def _get_verification_token(
self, upstream_token_set: UpstreamTokenSet
) -> str | None:
"""Get the token string to pass to the token verifier.
Returns the upstream access token by default. Subclasses can override
to verify a different token (e.g., the OIDC id_token for providers
that issue opaque access tokens).
"""
return upstream_token_set.access_token
async def load_access_token(self, token: str) -> AccessToken | None: # type: ignore[override]
"""Validate FastMCP JWT by swapping for upstream token.
@ -1405,14 +1419,31 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
# 3. Validate with upstream provider (delegated to TokenVerifier)
# This calls the real token validator (GitHub API, JWKS, etc.)
validated = await self._token_validator.verify_token(
upstream_token_set.access_token
)
verification_token = self._get_verification_token(upstream_token_set)
if verification_token is None:
logger.debug("No verification token available")
return None
validated = await self._token_validator.verify_token(verification_token)
if not validated:
logger.debug("Upstream token validation failed")
return None
# When the verification token differs from the access token
# (e.g., id_token verification), ensure the returned AccessToken
# carries the upstream access token and its scopes, not the
# verification token's values.
if verification_token != upstream_token_set.access_token:
validated = validated.model_copy(
update={
"token": upstream_token_set.access_token,
"scopes": upstream_token_set.scope.split()
if upstream_token_set.scope
else validated.scopes,
"expires_at": int(upstream_token_set.expires_at),
}
)
logger.debug(
"Token swap successful for JTI=%s (upstream validated)", jti[:8]
)

52
src/fastmcp/server/auth/oidc_proxy.py
View file

@ -18,6 +18,7 @@ from typing_extensions import Self
from fastmcp.server.auth import TokenVerifier
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
from fastmcp.utilities.logging import get_logger
@ -209,6 +210,7 @@ class OIDCProxy(OAuthProxy):
token_verifier: TokenVerifier | None = None,
algorithm: str | None = None,
required_scopes: list[str] | None = None,
verify_id_token: bool = False,
# FastMCP server configuration
base_url: AnyHttpUrl | str,
issuer_url: AnyHttpUrl | str | None = None,
@ -245,6 +247,9 @@ class OIDCProxy(OAuthProxy):
Cannot be used with algorithm or required_scopes parameters (configure these on your verifier instead).
algorithm: Token verifier algorithm (only used if token_verifier is not provided)
required_scopes: Required scopes for token validation (only used if token_verifier is not provided)
verify_id_token: If True, verify the OIDC id_token instead of the access_token.
Useful for providers that issue opaque (non-JWT) access tokens, since the
id_token is always a standard JWT verifiable via the provider's JWKS.
base_url: Public URL where OAuth endpoints will be accessible (includes any mount path)
issuer_url: Issuer URL for OAuth metadata (defaults to base_url). Use root-level URL
to avoid 404s during discovery when mounting under a path.
@ -330,14 +335,22 @@ class OIDCProxy(OAuthProxy):
# Use custom verifier if provided, otherwise create default JWTVerifier
if token_verifier is None:
# When verifying id_tokens:
# - aud is always the OAuth client_id (per OIDC Core §2), not
# the API audience, so use client_id for audience validation.
# - id_tokens don't carry scope/scp claims, so don't pass
# required_scopes to the verifier (scope enforcement happens
# at the FastMCP token level instead).
verifier_audience = client_id if verify_id_token else audience
verifier_scopes = None if verify_id_token else required_scopes
token_verifier = self.get_token_verifier(
algorithm=algorithm,
audience=audience,
required_scopes=required_scopes,
audience=verifier_audience,
required_scopes=verifier_scopes,
timeout_seconds=timeout_seconds,
)
init_kwargs = {
init_kwargs: dict[str, object] = {
"upstream_authorization_endpoint": str(
self.oidc_config.authorization_endpoint
),
@ -383,6 +396,39 @@ class OIDCProxy(OAuthProxy):
super().__init__(**init_kwargs) # ty: ignore[invalid-argument-type]
self._verify_id_token = verify_id_token
# When verify_id_token strips scopes from the verifier, restore
# them on the provider so they're still advertised to clients
# and enforced at the FastMCP token level. We also need to
# recompute derived state that OAuthProxy.__init__ already built
# from the (empty) verifier scopes.
if verify_id_token and required_scopes:
self.required_scopes = required_scopes
self._default_scope_str = " ".join(required_scopes)
if self.client_registration_options:
self.client_registration_options.valid_scopes = required_scopes
if self._cimd_manager is not None:
self._cimd_manager.default_scope = self._default_scope_str
def _get_verification_token(
self, upstream_token_set: UpstreamTokenSet
) -> str | None:
"""Get the token to verify from the upstream token set.
When verify_id_token is enabled, returns the id_token from the
upstream token response instead of the access_token.
"""
if self._verify_id_token:
id_token = upstream_token_set.raw_token_data.get("id_token")
if id_token is None:
logger.warning(
"verify_id_token is enabled but no id_token found in"
" upstream token response"
)
return id_token
return upstream_token_set.access_token
def get_oidc_configuration(
self,
config_url: AnyHttpUrl,

283
tests/server/auth/test_oidc_proxy.py
View file

@ -7,6 +7,7 @@ import pytest
from httpx import Response
from pydantic import AnyHttpUrl
from fastmcp.server.auth.oauth_proxy.models import UpstreamTokenSet
from fastmcp.server.auth.oidc_proxy import OIDCConfiguration, OIDCProxy
from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier
from fastmcp.server.auth.providers.jwt import JWTVerifier
@ -876,3 +877,285 @@ class TestOIDCProxyInitialization:
"audience": "original-audience",
"custom": "value",
}
def _make_upstream_token_set(*, id_token: str | None = None) -> UpstreamTokenSet:
"""Create an UpstreamTokenSet with optional id_token."""
raw_token_data: dict[str, str] = {"access_token": "opaque-access-token"}
if id_token is not None:
raw_token_data["id_token"] = id_token
return UpstreamTokenSet(
upstream_token_id="test-id",
access_token="opaque-access-token",
refresh_token=None,
refresh_token_expires_at=None,
expires_at=9999999999.0,
token_type="Bearer",
scope="openid",
client_id="test-client",
created_at=1000000000.0,
raw_token_data=raw_token_data,
)
class TestVerifyIdToken:
"""Tests for verify_id_token functionality."""
def test_verify_id_token_disabled_by_default(self, valid_oidc_configuration_dict):
"""Default behavior: verify the access_token."""
with patch(
"fastmcp.server.auth.oidc_proxy.OIDCConfiguration.get_oidc_configuration"
) as mock_get:
oidc_config = OIDCConfiguration.model_validate(
valid_oidc_configuration_dict
)
mock_get.return_value = oidc_config
proxy = OIDCProxy(
config_url=TEST_CONFIG_URL,
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
)
token_set = _make_upstream_token_set(id_token="jwt-id-token")
assert proxy._get_verification_token(token_set) == "opaque-access-token"
def test_verify_id_token_returns_id_token(self, valid_oidc_configuration_dict):
"""When enabled, verify the id_token instead of access_token."""
with patch(
"fastmcp.server.auth.oidc_proxy.OIDCConfiguration.get_oidc_configuration"
) as mock_get:
oidc_config = OIDCConfiguration.model_validate(
valid_oidc_configuration_dict
)
mock_get.return_value = oidc_config
proxy = OIDCProxy(
config_url=TEST_CONFIG_URL,
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
verify_id_token=True,
)
token_set = _make_upstream_token_set(id_token="jwt-id-token")
assert proxy._get_verification_token(token_set) == "jwt-id-token"
def test_verify_id_token_returns_none_when_missing(
self, valid_oidc_configuration_dict
):
"""When enabled but id_token is absent, return None."""
with patch(
"fastmcp.server.auth.oidc_proxy.OIDCConfiguration.get_oidc_configuration"
) as mock_get:
oidc_config = OIDCConfiguration.model_validate(
valid_oidc_configuration_dict
)
mock_get.return_value = oidc_config
proxy = OIDCProxy(
config_url=TEST_CONFIG_URL,
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
verify_id_token=True,
)
token_set = _make_upstream_token_set(id_token=None)
assert proxy._get_verification_token(token_set) is None
def test_verify_id_token_works_with_custom_verifier(
self, valid_oidc_configuration_dict
):
"""verify_id_token can be combined with a custom token_verifier."""
with patch(
"fastmcp.server.auth.oidc_proxy.OIDCConfiguration.get_oidc_configuration"
) as mock_get:
oidc_config = OIDCConfiguration.model_validate(
valid_oidc_configuration_dict
)
mock_get.return_value = oidc_config
custom_verifier = IntrospectionTokenVerifier(
introspection_url="https://example.com/oauth/introspect",
client_id="introspection-client",
client_secret="introspection-secret",
)
proxy = OIDCProxy(
config_url=TEST_CONFIG_URL,
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
verify_id_token=True,
token_verifier=custom_verifier,
)
token_set = _make_upstream_token_set(id_token="jwt-id-token")
assert proxy._get_verification_token(token_set) == "jwt-id-token"
assert proxy._token_validator is custom_verifier
def test_verify_id_token_survives_refresh_without_id_token(
self, valid_oidc_configuration_dict
):
"""id_token from original auth is preserved when refresh omits it."""
with patch(
"fastmcp.server.auth.oidc_proxy.OIDCConfiguration.get_oidc_configuration"
) as mock_get:
oidc_config = OIDCConfiguration.model_validate(
valid_oidc_configuration_dict
)
mock_get.return_value = oidc_config
proxy = OIDCProxy(
config_url=TEST_CONFIG_URL,
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
verify_id_token=True,
)
token_set = _make_upstream_token_set(id_token="original-id-token")
# Simulate a refresh response that omits id_token —
# the merge in exchange_refresh_token should preserve it
refresh_response = {
"access_token": "new-access-token",
"token_type": "Bearer",
}
token_set.raw_token_data = {**token_set.raw_token_data, **refresh_response}
token_set.access_token = "new-access-token"
assert proxy._get_verification_token(token_set) == "original-id-token"
def test_verify_id_token_updated_when_refresh_includes_it(
self, valid_oidc_configuration_dict
):
"""id_token is updated when refresh response includes a new one."""
with patch(
"fastmcp.server.auth.oidc_proxy.OIDCConfiguration.get_oidc_configuration"
) as mock_get:
oidc_config = OIDCConfiguration.model_validate(
valid_oidc_configuration_dict
)
mock_get.return_value = oidc_config
proxy = OIDCProxy(
config_url=TEST_CONFIG_URL,
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
verify_id_token=True,
)
token_set = _make_upstream_token_set(id_token="original-id-token")
# Simulate a refresh response that includes a new id_token
refresh_response = {
"access_token": "new-access-token",
"id_token": "refreshed-id-token",
}
token_set.raw_token_data = {**token_set.raw_token_data, **refresh_response}
token_set.access_token = "new-access-token"
assert proxy._get_verification_token(token_set) == "refreshed-id-token"
def test_verify_id_token_uses_client_id_as_verifier_audience(
self, valid_oidc_configuration_dict
):
"""When verify_id_token is enabled, the verifier audience should be
client_id (matching id_token.aud), not the API audience parameter."""
with patch(
"fastmcp.server.auth.oidc_proxy.OIDCConfiguration.get_oidc_configuration"
) as mock_get:
oidc_config = OIDCConfiguration.model_validate(
valid_oidc_configuration_dict
)
mock_get.return_value = oidc_config
proxy = OIDCProxy(
config_url=TEST_CONFIG_URL,
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
audience="https://api.example.com",
verify_id_token=True,
)
assert isinstance(proxy._token_validator, JWTVerifier)
assert proxy._token_validator.audience == TEST_CLIENT_ID
# The API audience should still be sent upstream
assert (
proxy._extra_authorize_params["audience"] == "https://api.example.com"
)
assert proxy._extra_token_params["audience"] == "https://api.example.com"
def test_verify_id_token_without_audience_uses_client_id(
self, valid_oidc_configuration_dict
):
"""When verify_id_token is enabled without an audience param,
the verifier audience should still be client_id."""
with patch(
"fastmcp.server.auth.oidc_proxy.OIDCConfiguration.get_oidc_configuration"
) as mock_get:
oidc_config = OIDCConfiguration.model_validate(
valid_oidc_configuration_dict
)
mock_get.return_value = oidc_config
proxy = OIDCProxy(
config_url=TEST_CONFIG_URL,
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
verify_id_token=True,
)
assert isinstance(proxy._token_validator, JWTVerifier)
assert proxy._token_validator.audience == TEST_CLIENT_ID
def test_verify_id_token_does_not_enforce_scopes_on_verifier(
self, valid_oidc_configuration_dict
):
"""When verify_id_token is enabled, required_scopes should not be
passed to the JWTVerifier since id_tokens don't carry scope claims."""
with patch(
"fastmcp.server.auth.oidc_proxy.OIDCConfiguration.get_oidc_configuration"
) as mock_get:
oidc_config = OIDCConfiguration.model_validate(
valid_oidc_configuration_dict
)
mock_get.return_value = oidc_config
proxy = OIDCProxy(
config_url=TEST_CONFIG_URL,
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
required_scopes=["read", "write"],
verify_id_token=True,
)
assert isinstance(proxy._token_validator, JWTVerifier)
assert proxy._token_validator.required_scopes == []
# Scopes should still be advertised via the proxy's required_scopes
assert proxy.required_scopes == ["read", "write"]
# Derived scope state should also be recomputed
assert proxy._default_scope_str == "read write"
assert proxy.client_registration_options is not None
assert proxy.client_registration_options.valid_scopes == [
"read",
"write",
]