mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-21 04:54:17 +02:00
Add custom token verifier support to OIDCProxy (#2279)
* Add custom token verifier support to OIDCProxy OIDCProxy now accepts an optional token_verifier parameter to support non-JWT token formats like opaque tokens from providers such as Clerk. When provided, the custom verifier is used instead of creating a default JWTVerifier. Parameters that only apply to JWTVerifier creation (algorithm, required_scopes) raise clear errors when specified alongside a custom verifier. Parameters with other purposes (audience for OAuth flow, timeout_seconds for config fetch) remain allowed. The custom verifier's required_scopes are automatically loaded and advertised through OAuth discovery endpoints. * Document custom token verifier support in OIDC proxy
This commit is contained in:
parent
6d600e36db
commit
15dbe7ecf0
3 changed files with 172 additions and 11 deletions
|
|
@ -93,14 +93,22 @@ mcp = FastMCP(name="My Server", auth=auth)
|
|||
HTTP request timeout in seconds for fetching OIDC configuration
|
||||
</ParamField>
|
||||
|
||||
<ParamField body="token_verifier" type="TokenVerifier | None">
|
||||
|
||||
<VersionBadge version="2.13.1" />
|
||||
Custom token verifier for validating tokens. When provided, FastMCP uses your custom verifier instead of creating a default `JWTVerifier`.
|
||||
|
||||
Cannot be used with `algorithm` or `required_scopes` parameters - configure these on your verifier instead. The verifier's `required_scopes` are automatically loaded and advertised.
|
||||
</ParamField>
|
||||
|
||||
<ParamField body="algorithm" type="str | None">
|
||||
JWT algorithm to use for token verification (e.g., "RS256"). If not specified,
|
||||
uses the provider's default.
|
||||
uses the provider's default. Only used when `token_verifier` is not provided.
|
||||
</ParamField>
|
||||
|
||||
<ParamField body="required_scopes" type="list[str] | None">
|
||||
List of OAuth scopes to request from the provider. These are automatically
|
||||
included in authorization requests.
|
||||
List of OAuth scopes for token validation. These are automatically
|
||||
included in authorization requests. Only used when `token_verifier` is not provided.
|
||||
</ParamField>
|
||||
|
||||
<ParamField body="redirect_path" type="str" default="/auth/callback">
|
||||
|
|
|
|||
|
|
@ -206,6 +206,7 @@ class OIDCProxy(OAuthProxy):
|
|||
audience: str | None = None,
|
||||
timeout_seconds: int | None = None,
|
||||
# Token verifier
|
||||
token_verifier: TokenVerifier | None = None,
|
||||
algorithm: str | None = None,
|
||||
required_scopes: list[str] | None = None,
|
||||
# FastMCP server configuration
|
||||
|
|
@ -231,8 +232,11 @@ class OIDCProxy(OAuthProxy):
|
|||
client_secret: Client secret for upstream server
|
||||
audience: Audience for upstream server
|
||||
timeout_seconds: HTTP request timeout in seconds
|
||||
algorithm: Token verifier algorithm
|
||||
required_scopes: Required OAuth scopes
|
||||
token_verifier: Optional custom token verifier (e.g., IntrospectionTokenVerifier for opaque tokens).
|
||||
If not provided, a JWTVerifier will be created using the OIDC configuration.
|
||||
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)
|
||||
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.
|
||||
|
|
@ -268,6 +272,19 @@ class OIDCProxy(OAuthProxy):
|
|||
if not base_url:
|
||||
raise ValueError("Missing required base URL")
|
||||
|
||||
# Validate that verifier-specific parameters are not used with custom verifier
|
||||
if token_verifier is not None:
|
||||
if algorithm is not None:
|
||||
raise ValueError(
|
||||
"Cannot specify 'algorithm' when providing a custom token_verifier. "
|
||||
"Configure the algorithm on your token verifier instead."
|
||||
)
|
||||
if required_scopes is not None:
|
||||
raise ValueError(
|
||||
"Cannot specify 'required_scopes' when providing a custom token_verifier. "
|
||||
"Configure required scopes on your token verifier instead."
|
||||
)
|
||||
|
||||
if isinstance(config_url, str):
|
||||
config_url = AnyHttpUrl(config_url)
|
||||
|
||||
|
|
@ -287,12 +304,14 @@ class OIDCProxy(OAuthProxy):
|
|||
else None
|
||||
)
|
||||
|
||||
token_verifier = self.get_token_verifier(
|
||||
algorithm=algorithm,
|
||||
audience=audience,
|
||||
required_scopes=required_scopes,
|
||||
timeout_seconds=timeout_seconds,
|
||||
)
|
||||
# Use custom verifier if provided, otherwise create default JWTVerifier
|
||||
if token_verifier is None:
|
||||
token_verifier = self.get_token_verifier(
|
||||
algorithm=algorithm,
|
||||
audience=audience,
|
||||
required_scopes=required_scopes,
|
||||
timeout_seconds=timeout_seconds,
|
||||
)
|
||||
|
||||
init_kwargs = {
|
||||
"upstream_authorization_endpoint": str(
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ from httpx import Response
|
|||
from pydantic import AnyHttpUrl
|
||||
|
||||
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
|
||||
|
||||
TEST_ISSUER = "https://example.com"
|
||||
|
|
@ -649,3 +650,136 @@ class TestOIDCProxyInitialization:
|
|||
client_secret=TEST_CLIENT_SECRET,
|
||||
base_url=None, # type: ignore
|
||||
)
|
||||
|
||||
def test_custom_token_verifier_initialization(self, valid_oidc_configuration_dict):
|
||||
"""Test initialization with 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
|
||||
|
||||
# Create custom verifier for opaque tokens
|
||||
custom_verifier = IntrospectionTokenVerifier(
|
||||
introspection_url="https://example.com/oauth/introspect",
|
||||
client_id="introspection-client",
|
||||
client_secret="introspection-secret",
|
||||
required_scopes=["custom", "scopes"],
|
||||
)
|
||||
|
||||
proxy = OIDCProxy(
|
||||
config_url=TEST_CONFIG_URL,
|
||||
client_id=TEST_CLIENT_ID,
|
||||
client_secret=TEST_CLIENT_SECRET,
|
||||
base_url=TEST_BASE_URL,
|
||||
token_verifier=custom_verifier,
|
||||
jwt_signing_key="test-secret",
|
||||
)
|
||||
|
||||
validate_proxy(mock_get, proxy, oidc_config)
|
||||
|
||||
# Verify the custom verifier is used
|
||||
assert proxy._token_validator is custom_verifier
|
||||
assert isinstance(proxy._token_validator, IntrospectionTokenVerifier)
|
||||
|
||||
# Verify required_scopes are properly loaded from the custom verifier
|
||||
assert proxy.required_scopes == ["custom", "scopes"]
|
||||
|
||||
def test_custom_token_verifier_with_algorithm_raises_error(
|
||||
self, valid_oidc_configuration_dict
|
||||
):
|
||||
"""Test that providing algorithm with custom verifier raises error."""
|
||||
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",
|
||||
)
|
||||
|
||||
with pytest.raises(
|
||||
ValueError,
|
||||
match="Cannot specify 'algorithm' when providing a custom token_verifier",
|
||||
):
|
||||
OIDCProxy(
|
||||
config_url=TEST_CONFIG_URL,
|
||||
client_id=TEST_CLIENT_ID,
|
||||
client_secret=TEST_CLIENT_SECRET,
|
||||
base_url=TEST_BASE_URL,
|
||||
token_verifier=custom_verifier,
|
||||
algorithm="RS256", # This should cause an error
|
||||
jwt_signing_key="test-secret",
|
||||
)
|
||||
|
||||
def test_custom_token_verifier_with_required_scopes_raises_error(
|
||||
self, valid_oidc_configuration_dict
|
||||
):
|
||||
"""Test that providing required_scopes with custom verifier raises error."""
|
||||
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",
|
||||
)
|
||||
|
||||
with pytest.raises(
|
||||
ValueError,
|
||||
match="Cannot specify 'required_scopes' when providing a custom token_verifier",
|
||||
):
|
||||
OIDCProxy(
|
||||
config_url=TEST_CONFIG_URL,
|
||||
client_id=TEST_CLIENT_ID,
|
||||
client_secret=TEST_CLIENT_SECRET,
|
||||
base_url=TEST_BASE_URL,
|
||||
token_verifier=custom_verifier,
|
||||
required_scopes=["read", "write"], # This should cause an error
|
||||
jwt_signing_key="test-secret",
|
||||
)
|
||||
|
||||
def test_custom_token_verifier_with_audience_allowed(
|
||||
self, valid_oidc_configuration_dict
|
||||
):
|
||||
"""Test that providing audience with custom verifier is allowed (for OAuth flow)."""
|
||||
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",
|
||||
)
|
||||
|
||||
# This should NOT raise an error - audience is for OAuth flow
|
||||
proxy = OIDCProxy(
|
||||
config_url=TEST_CONFIG_URL,
|
||||
client_id=TEST_CLIENT_ID,
|
||||
client_secret=TEST_CLIENT_SECRET,
|
||||
base_url=TEST_BASE_URL,
|
||||
token_verifier=custom_verifier,
|
||||
audience="test-audience", # Should be allowed for OAuth flow
|
||||
jwt_signing_key="test-secret",
|
||||
)
|
||||
|
||||
validate_proxy(mock_get, proxy, oidc_config)
|
||||
assert proxy._extra_authorize_params == {"audience": "test-audience"}
|
||||
assert proxy._extra_token_params == {"audience": "test-audience"}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue