From f84b2dae0469e6fd29c51a2ecb07205eb3b7f738 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Fri, 20 Feb 2026 14:20:47 -0500 Subject: [PATCH] Add verify_id_token option to OIDCProxy (#3248) * 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> --- .../fastmcp-server-auth-oauth_proxy-proxy.mdx | 6 +- .../fastmcp-server-auth-oidc_proxy.mdx | 10 +- src/fastmcp/server/auth/oauth_proxy/proxy.py | 39 ++- src/fastmcp/server/auth/oidc_proxy.py | 52 +++- tests/server/auth/test_oidc_proxy.py | 283 ++++++++++++++++++ 5 files changed, 375 insertions(+), 15 deletions(-) diff --git a/docs/python-sdk/fastmcp-server-auth-oauth_proxy-proxy.mdx b/docs/python-sdk/fastmcp-server-auth-oauth_proxy-proxy.mdx index dcb22c638..dd1400086 100644 --- a/docs/python-sdk/fastmcp-server-auth-oauth_proxy-proxy.mdx +++ b/docs/python-sdk/fastmcp-server-auth-oauth_proxy-proxy.mdx @@ -273,7 +273,7 @@ Implements two-tier refresh: 6. Keep same FastMCP refresh token (unless upstream rotates) -#### `load_access_token` +#### `load_access_token` ```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` +#### `revoke_token` ```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` +#### `get_routes` ```python get_routes(self, mcp_path: str | None = None) -> list[Route] diff --git a/docs/python-sdk/fastmcp-server-auth-oidc_proxy.mdx b/docs/python-sdk/fastmcp-server-auth-oidc_proxy.mdx index c9339aa72..183380ed9 100644 --- a/docs/python-sdk/fastmcp-server-auth-oidc_proxy.mdx +++ b/docs/python-sdk/fastmcp-server-auth-oidc_proxy.mdx @@ -19,7 +19,7 @@ This implementation is based on: ## Classes -### `OIDCConfiguration` +### `OIDCConfiguration` OIDC Configuration. @@ -27,7 +27,7 @@ OIDC Configuration. **Methods:** -#### `get_oidc_configuration` +#### `get_oidc_configuration` ```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` +### `OIDCProxy` 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` +#### `get_oidc_configuration` ```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` +#### `get_token_verifier` ```python get_token_verifier(self) -> TokenVerifier diff --git a/src/fastmcp/server/auth/oauth_proxy/proxy.py b/src/fastmcp/server/auth/oauth_proxy/proxy.py index cca772db3..0a72ae9f9 100644 --- a/src/fastmcp/server/auth/oauth_proxy/proxy.py +++ b/src/fastmcp/server/auth/oauth_proxy/proxy.py @@ -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] ) diff --git a/src/fastmcp/server/auth/oidc_proxy.py b/src/fastmcp/server/auth/oidc_proxy.py index 3a000129f..418bea34c 100644 --- a/src/fastmcp/server/auth/oidc_proxy.py +++ b/src/fastmcp/server/auth/oidc_proxy.py @@ -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, diff --git a/tests/server/auth/test_oidc_proxy.py b/tests/server/auth/test_oidc_proxy.py index e3dd95e1c..b4fbd32a5 100644 --- a/tests/server/auth/test_oidc_proxy.py +++ b/tests/server/auth/test_oidc_proxy.py @@ -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", + ]