From deb0c3ea95e6af64fe23d9b5d475d0f30d694f82 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Mon, 20 Oct 2025 16:01:36 -0400 Subject: [PATCH] Rename OIDCProxy -> OIDCDCRProxy --- .../fastmcp-server-auth-oidc_proxy.mdx | 2 +- docs/servers/auth/oidc-proxy.mdx | 8 +- src/fastmcp/server/auth/oidc_dcr_proxy.py | 350 +++++ src/fastmcp/server/auth/oidc_proxy.py | 360 +---- src/fastmcp/server/auth/providers/auth0.py | 4 +- src/fastmcp/server/auth/providers/aws.py | 4 +- .../auth/oauth_dcr_proxy/test_oidc_proxy.py | 22 +- tests/server/auth/providers/test_auth0.py | 2 +- tests/server/auth/test_oauth_proxy.py | 1297 ----------------- 9 files changed, 388 insertions(+), 1661 deletions(-) create mode 100644 src/fastmcp/server/auth/oidc_dcr_proxy.py delete mode 100644 tests/server/auth/test_oauth_proxy.py diff --git a/docs/python-sdk/fastmcp-server-auth-oidc_proxy.mdx b/docs/python-sdk/fastmcp-server-auth-oidc_proxy.mdx index f0124db1e..39360e222 100644 --- a/docs/python-sdk/fastmcp-server-auth-oidc_proxy.mdx +++ b/docs/python-sdk/fastmcp-server-auth-oidc_proxy.mdx @@ -41,7 +41,7 @@ Get the OIDC configuration for the specified config URL. - `timeout_seconds`: HTTP request timeout in seconds -### `OIDCProxy` +### `OIDCDCRProxy` OAuth provider that wraps OAuthProxy to provide configuration via an OIDC configuration URL. diff --git a/docs/servers/auth/oidc-proxy.mdx b/docs/servers/auth/oidc-proxy.mdx index 22f5e2ee8..bf0b130e9 100644 --- a/docs/servers/auth/oidc-proxy.mdx +++ b/docs/servers/auth/oidc-proxy.mdx @@ -39,10 +39,10 @@ Here's how to implement the OIDC proxy with any provider: ```python from fastmcp import FastMCP -from fastmcp.server.auth.oidc_proxy import OIDCProxy +from fastmcp.server.auth.oidc_dcr_proxy import OIDCDCRProxy # Create the OIDC proxy -auth = OIDCProxy( +auth = OIDCDCRProxy( # Provider's configuration URL config_url="https://provider.com/.well-known/openid-configuration", @@ -62,7 +62,7 @@ mcp = FastMCP(name="My Server", auth=auth) ### Configuration Parameters - + URL of your OAuth provider's OIDC configuration @@ -136,7 +136,7 @@ Set this if your provider requires a specific authentication method and the defa from fastmcp.utilities.storage import InMemoryStorage # Use in-memory storage for testing (clients lost on restart) -auth = OIDCProxy(..., client_storage=InMemoryStorage()) +auth = OIDCDCRProxy(..., client_storage=InMemoryStorage()) ``` diff --git a/src/fastmcp/server/auth/oidc_dcr_proxy.py b/src/fastmcp/server/auth/oidc_dcr_proxy.py new file mode 100644 index 000000000..90e5c4816 --- /dev/null +++ b/src/fastmcp/server/auth/oidc_dcr_proxy.py @@ -0,0 +1,350 @@ +"""OIDC Proxy Provider for FastMCP. + +This provider acts as a transparent proxy to an upstream OIDC compliant Authorization +Server. It leverages the OAuthProxy class to handle Dynamic Client Registration and +forwarding of all OAuth flows. + +This implementation is based on: + OpenID Connect Discovery 1.0 - https://openid.net/specs/openid-connect-discovery-1_0.html + OAuth 2.0 Authorization Server Metadata - https://datatracker.ietf.org/doc/html/rfc8414 +""" + +from collections.abc import Sequence + +import httpx +from key_value.aio.protocols import AsyncKeyValue +from pydantic import AnyHttpUrl, BaseModel, model_validator +from typing_extensions import Self + +from fastmcp.server.auth import TokenVerifier +from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy +from fastmcp.server.auth.providers.jwt import JWTVerifier +from fastmcp.utilities.logging import get_logger + +logger = get_logger(__name__) + + +class OIDCConfiguration(BaseModel): + """OIDC Configuration. + + See: + https://openid.net/specs/openid-connect-discovery-1_0.html#ProviderMetadata + https://datatracker.ietf.org/doc/html/rfc8414#section-2 + """ + + strict: bool = True + + # OpenID Connect Discovery 1.0 + issuer: AnyHttpUrl | str | None = None # Strict + + authorization_endpoint: AnyHttpUrl | str | None = None # Strict + token_endpoint: AnyHttpUrl | str | None = None # Strict + userinfo_endpoint: AnyHttpUrl | str | None = None + + jwks_uri: AnyHttpUrl | str | None = None # Strict + + registration_endpoint: AnyHttpUrl | str | None = None + + scopes_supported: Sequence[str] | None = None + + response_types_supported: Sequence[str] | None = None # Strict + response_modes_supported: Sequence[str] | None = None + + grant_types_supported: Sequence[str] | None = None + + acr_values_supported: Sequence[str] | None = None + + subject_types_supported: Sequence[str] | None = None # Strict + + id_token_signing_alg_values_supported: Sequence[str] | None = None # Strict + id_token_encryption_alg_values_supported: Sequence[str] | None = None + id_token_encryption_enc_values_supported: Sequence[str] | None = None + + userinfo_signing_alg_values_supported: Sequence[str] | None = None + userinfo_encryption_alg_values_supported: Sequence[str] | None = None + userinfo_encryption_enc_values_supported: Sequence[str] | None = None + + request_object_signing_alg_values_supported: Sequence[str] | None = None + request_object_encryption_alg_values_supported: Sequence[str] | None = None + request_object_encryption_enc_values_supported: Sequence[str] | None = None + + token_endpoint_auth_methods_supported: Sequence[str] | None = None + token_endpoint_auth_signing_alg_values_supported: Sequence[str] | None = None + + display_values_supported: Sequence[str] | None = None + + claim_types_supported: Sequence[str] | None = None + claims_supported: Sequence[str] | None = None + + service_documentation: AnyHttpUrl | str | None = None + + claims_locales_supported: Sequence[str] | None = None + ui_locales_supported: Sequence[str] | None = None + + claims_parameter_supported: bool | None = None + request_parameter_supported: bool | None = None + request_uri_parameter_supported: bool | None = None + + require_request_uri_registration: bool | None = None + + op_policy_uri: AnyHttpUrl | str | None = None + op_tos_uri: AnyHttpUrl | str | None = None + + # OAuth 2.0 Authorization Server Metadata + revocation_endpoint: AnyHttpUrl | str | None = None + revocation_endpoint_auth_methods_supported: Sequence[str] | None = None + revocation_endpoint_auth_signing_alg_values_supported: Sequence[str] | None = None + + introspection_endpoint: AnyHttpUrl | str | None = None + introspection_endpoint_auth_methods_supported: Sequence[str] | None = None + introspection_endpoint_auth_signing_alg_values_supported: Sequence[str] | None = ( + None + ) + + code_challenge_methods_supported: Sequence[str] | None = None + + signed_metadata: str | None = None + + @model_validator(mode="after") + def _enforce_strict(self) -> Self: + """Enforce strict rules.""" + if not self.strict: + return self + + def enforce(attr: str, is_url: bool = False) -> None: + value = getattr(self, attr, None) + if not value: + message = f"Missing required configuration metadata: {attr}" + logger.error(message) + raise ValueError(message) + + if not is_url or isinstance(value, AnyHttpUrl): + return + + try: + AnyHttpUrl(value) + except Exception: + message = f"Invalid URL for configuration metadata: {attr}" + logger.error(message) + raise ValueError(message) + + enforce("issuer", True) + enforce("authorization_endpoint", True) + enforce("token_endpoint", True) + enforce("jwks_uri", True) + enforce("response_types_supported") + enforce("subject_types_supported") + enforce("id_token_signing_alg_values_supported") + + return self + + @classmethod + def get_oidc_configuration( + cls, config_url: AnyHttpUrl, *, strict: bool | None, timeout_seconds: int | None + ) -> Self: + """Get the OIDC configuration for the specified config URL. + + Args: + config_url: The OIDC config URL + strict: The strict flag for the configuration + timeout_seconds: HTTP request timeout in seconds + """ + get_kwargs = {} + if timeout_seconds is not None: + get_kwargs["timeout"] = timeout_seconds + + try: + response = httpx.get(str(config_url), **get_kwargs) + response.raise_for_status() + + config_data = response.json() + if strict is not None: + config_data["strict"] = strict + + return cls.model_validate(config_data) + except Exception: + logger.exception( + f"Unable to get OIDC configuration for config url: {config_url}" + ) + raise + + +class OIDCDCRProxy(OAuthDCRProxy): + """OAuth provider that wraps OAuthDCRProxy to provide configuration via an OIDC configuration URL. + + This provider makes it easier to add OAuth protection for any upstream provider + that is OIDC compliant. + + Example: + ```python + from fastmcp import FastMCP + from fastmcp.server.auth.oidc_dcr_proxy import OIDCDCRProxy + + # Simple OIDC based protection + auth = OIDCDCRProxy( + config_url="https://oidc.config.url", + client_id="your-oidc-client-id", + client_secret="your-oidc-client-secret", + base_url="https://your.server.url", + ) + + mcp = FastMCP("My Protected Server", auth=auth) + ``` + """ + + oidc_config: OIDCConfiguration + + def __init__( + self, + *, + # OIDC configuration + config_url: AnyHttpUrl | str, + strict: bool | None = None, + # Upstream server configuration + client_id: str, + client_secret: str, + audience: str | None = None, + timeout_seconds: int | None = None, + # Token verifier + algorithm: str | None = None, + required_scopes: list[str] | None = None, + # FastMCP server configuration + base_url: AnyHttpUrl | str, + issuer_url: AnyHttpUrl | str | None = None, + redirect_path: str | None = None, + # Client configuration + allowed_client_redirect_uris: list[str] | None = None, + client_storage: AsyncKeyValue | None = None, + # Token validation configuration + token_endpoint_auth_method: str | None = None, + ) -> None: + """Initialize the OIDC proxy provider. + + Args: + config_url: URL of upstream configuration + strict: Optional strict flag for the configuration + client_id: Client ID registered with upstream server + 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 + 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. + redirect_path: Redirect path configured in upstream OAuth app (defaults to "/auth/callback") + allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients. + Patterns support wildcards (e.g., "http://localhost:*", "https://*.example.com/*"). + If None (default), only localhost redirect URIs are allowed. + If empty list, all redirect URIs are allowed (not recommended for production). + These are for MCP clients performing loopback redirects, NOT for the upstream OAuth app. + client_storage: An AsyncKeyValue-compatible store for client registrations, registrations are stored in memory if not provided + token_endpoint_auth_method: Token endpoint authentication method for upstream server. + Common values: "client_secret_basic", "client_secret_post", "none". + If None, authlib will use its default (typically "client_secret_basic"). + """ + if not config_url: + raise ValueError("Missing required config URL") + + if not client_id: + raise ValueError("Missing required client id") + + if not client_secret: + raise ValueError("Missing required client secret") + + if not base_url: + raise ValueError("Missing required base URL") + + if isinstance(config_url, str): + config_url = AnyHttpUrl(config_url) + + self.oidc_config = self.get_oidc_configuration( + config_url, strict, timeout_seconds + ) + if ( + not self.oidc_config.authorization_endpoint + or not self.oidc_config.token_endpoint + ): + logger.debug(f"Invalid OIDC Configuration: {self.oidc_config}") + raise ValueError("Missing required OIDC endpoints") + + revocation_endpoint = ( + str(self.oidc_config.revocation_endpoint) + if self.oidc_config.revocation_endpoint + else 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( + self.oidc_config.authorization_endpoint + ), + "upstream_token_endpoint": str(self.oidc_config.token_endpoint), + "upstream_client_id": client_id, + "upstream_client_secret": client_secret, + "upstream_revocation_endpoint": revocation_endpoint, + "token_verifier": token_verifier, + "base_url": base_url, + "issuer_url": issuer_url or base_url, + "service_documentation_url": self.oidc_config.service_documentation, + "allowed_client_redirect_uris": allowed_client_redirect_uris, + "client_storage": client_storage, + "token_endpoint_auth_method": token_endpoint_auth_method, + } + + if redirect_path: + init_kwargs["redirect_path"] = redirect_path + + if audience: + extra_params = {"audience": audience} + init_kwargs["extra_authorize_params"] = extra_params + init_kwargs["extra_token_params"] = extra_params + + super().__init__(**init_kwargs) + + def get_oidc_configuration( + self, + config_url: AnyHttpUrl, + strict: bool | None, + timeout_seconds: int | None, + ) -> OIDCConfiguration: + """Gets the OIDC configuration for the specified configuration URL. + + Args: + config_url: The OIDC configuration URL + strict: The strict flag for the configuration + timeout_seconds: HTTP request timeout in seconds + """ + return OIDCConfiguration.get_oidc_configuration( + config_url, strict=strict, timeout_seconds=timeout_seconds + ) + + def get_token_verifier( + self, + *, + algorithm: str | None = None, + audience: str | None = None, + required_scopes: list[str] | None = None, + timeout_seconds: int | None = None, + ) -> TokenVerifier: + """Creates the token verifier for the specified OIDC configuration and arguments. + + Args: + algorithm: Optional token verifier algorithm + audience: Optional token verifier audience + required_scopes: Optional token verifier required_scopes + timeout_seconds: HTTP request timeout in seconds + """ + return JWTVerifier( + jwks_uri=str(self.oidc_config.jwks_uri), + issuer=str(self.oidc_config.issuer), + algorithm=algorithm, + audience=audience, + required_scopes=required_scopes, + ) diff --git a/src/fastmcp/server/auth/oidc_proxy.py b/src/fastmcp/server/auth/oidc_proxy.py index 7084d1b02..529a63987 100644 --- a/src/fastmcp/server/auth/oidc_proxy.py +++ b/src/fastmcp/server/auth/oidc_proxy.py @@ -1,350 +1,24 @@ -"""OIDC Proxy Provider for FastMCP. +"""Backwards compatibility shim for oidc_proxy.py -This provider acts as a transparent proxy to an upstream OIDC compliant Authorization -Server. It leverages the OAuthProxy class to handle Dynamic Client Registration and -forwarding of all OAuth flows. - -This implementation is based on: - OpenID Connect Discovery 1.0 - https://openid.net/specs/openid-connect-discovery-1_0.html - OAuth 2.0 Authorization Server Metadata - https://datatracker.ietf.org/doc/html/rfc8414 +The OIDCProxy class has been moved to fastmcp.server.auth.oidc_dcr_proxy.OIDCDCRProxy +for better organization. This module provides a backwards-compatible import. """ -from collections.abc import Sequence +import warnings -import httpx -from key_value.aio.protocols import AsyncKeyValue -from pydantic import AnyHttpUrl, BaseModel, model_validator -from typing_extensions import Self +import fastmcp +from fastmcp.server.auth.oidc_dcr_proxy import OIDCDCRProxy as OIDCProxy -from fastmcp.server.auth import TokenVerifier -from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy -from fastmcp.server.auth.providers.jwt import JWTVerifier -from fastmcp.utilities.logging import get_logger +# Re-export for backwards compatibility +__all__ = ["OIDCProxy"] -logger = get_logger(__name__) - - -class OIDCConfiguration(BaseModel): - """OIDC Configuration. - - See: - https://openid.net/specs/openid-connect-discovery-1_0.html#ProviderMetadata - https://datatracker.ietf.org/doc/html/rfc8414#section-2 - """ - - strict: bool = True - - # OpenID Connect Discovery 1.0 - issuer: AnyHttpUrl | str | None = None # Strict - - authorization_endpoint: AnyHttpUrl | str | None = None # Strict - token_endpoint: AnyHttpUrl | str | None = None # Strict - userinfo_endpoint: AnyHttpUrl | str | None = None - - jwks_uri: AnyHttpUrl | str | None = None # Strict - - registration_endpoint: AnyHttpUrl | str | None = None - - scopes_supported: Sequence[str] | None = None - - response_types_supported: Sequence[str] | None = None # Strict - response_modes_supported: Sequence[str] | None = None - - grant_types_supported: Sequence[str] | None = None - - acr_values_supported: Sequence[str] | None = None - - subject_types_supported: Sequence[str] | None = None # Strict - - id_token_signing_alg_values_supported: Sequence[str] | None = None # Strict - id_token_encryption_alg_values_supported: Sequence[str] | None = None - id_token_encryption_enc_values_supported: Sequence[str] | None = None - - userinfo_signing_alg_values_supported: Sequence[str] | None = None - userinfo_encryption_alg_values_supported: Sequence[str] | None = None - userinfo_encryption_enc_values_supported: Sequence[str] | None = None - - request_object_signing_alg_values_supported: Sequence[str] | None = None - request_object_encryption_alg_values_supported: Sequence[str] | None = None - request_object_encryption_enc_values_supported: Sequence[str] | None = None - - token_endpoint_auth_methods_supported: Sequence[str] | None = None - token_endpoint_auth_signing_alg_values_supported: Sequence[str] | None = None - - display_values_supported: Sequence[str] | None = None - - claim_types_supported: Sequence[str] | None = None - claims_supported: Sequence[str] | None = None - - service_documentation: AnyHttpUrl | str | None = None - - claims_locales_supported: Sequence[str] | None = None - ui_locales_supported: Sequence[str] | None = None - - claims_parameter_supported: bool | None = None - request_parameter_supported: bool | None = None - request_uri_parameter_supported: bool | None = None - - require_request_uri_registration: bool | None = None - - op_policy_uri: AnyHttpUrl | str | None = None - op_tos_uri: AnyHttpUrl | str | None = None - - # OAuth 2.0 Authorization Server Metadata - revocation_endpoint: AnyHttpUrl | str | None = None - revocation_endpoint_auth_methods_supported: Sequence[str] | None = None - revocation_endpoint_auth_signing_alg_values_supported: Sequence[str] | None = None - - introspection_endpoint: AnyHttpUrl | str | None = None - introspection_endpoint_auth_methods_supported: Sequence[str] | None = None - introspection_endpoint_auth_signing_alg_values_supported: Sequence[str] | None = ( - None +# Deprecated in 2.13 +if fastmcp.settings.deprecation_warnings: + warnings.warn( + "The `fastmcp.server.auth.oidc_proxy` module is deprecated " + "and will be removed in a future version. " + "Please use `fastmcp.server.auth.oidc_dcr_proxy.OIDCDCRProxy` " + "instead of this module's OIDCProxy.", + DeprecationWarning, + stacklevel=2, ) - - code_challenge_methods_supported: Sequence[str] | None = None - - signed_metadata: str | None = None - - @model_validator(mode="after") - def _enforce_strict(self) -> Self: - """Enforce strict rules.""" - if not self.strict: - return self - - def enforce(attr: str, is_url: bool = False) -> None: - value = getattr(self, attr, None) - if not value: - message = f"Missing required configuration metadata: {attr}" - logger.error(message) - raise ValueError(message) - - if not is_url or isinstance(value, AnyHttpUrl): - return - - try: - AnyHttpUrl(value) - except Exception: - message = f"Invalid URL for configuration metadata: {attr}" - logger.error(message) - raise ValueError(message) - - enforce("issuer", True) - enforce("authorization_endpoint", True) - enforce("token_endpoint", True) - enforce("jwks_uri", True) - enforce("response_types_supported") - enforce("subject_types_supported") - enforce("id_token_signing_alg_values_supported") - - return self - - @classmethod - def get_oidc_configuration( - cls, config_url: AnyHttpUrl, *, strict: bool | None, timeout_seconds: int | None - ) -> Self: - """Get the OIDC configuration for the specified config URL. - - Args: - config_url: The OIDC config URL - strict: The strict flag for the configuration - timeout_seconds: HTTP request timeout in seconds - """ - get_kwargs = {} - if timeout_seconds is not None: - get_kwargs["timeout"] = timeout_seconds - - try: - response = httpx.get(str(config_url), **get_kwargs) - response.raise_for_status() - - config_data = response.json() - if strict is not None: - config_data["strict"] = strict - - return cls.model_validate(config_data) - except Exception: - logger.exception( - f"Unable to get OIDC configuration for config url: {config_url}" - ) - raise - - -class OIDCProxy(OAuthDCRProxy): - """OAuth provider that wraps OAuthProxy to provide configuration via an OIDC configuration URL. - - This provider makes it easier to add OAuth protection for any upstream provider - that is OIDC compliant. - - Example: - ```python - from fastmcp import FastMCP - from fastmcp.server.auth.oidc_proxy import OIDCProxy - - # Simple OIDC based protection - auth = OIDCProxy( - config_url="https://oidc.config.url", - client_id="your-oidc-client-id", - client_secret="your-oidc-client-secret", - base_url="https://your.server.url", - ) - - mcp = FastMCP("My Protected Server", auth=auth) - ``` - """ - - oidc_config: OIDCConfiguration - - def __init__( - self, - *, - # OIDC configuration - config_url: AnyHttpUrl | str, - strict: bool | None = None, - # Upstream server configuration - client_id: str, - client_secret: str, - audience: str | None = None, - timeout_seconds: int | None = None, - # Token verifier - algorithm: str | None = None, - required_scopes: list[str] | None = None, - # FastMCP server configuration - base_url: AnyHttpUrl | str, - issuer_url: AnyHttpUrl | str | None = None, - redirect_path: str | None = None, - # Client configuration - allowed_client_redirect_uris: list[str] | None = None, - client_storage: AsyncKeyValue | None = None, - # Token validation configuration - token_endpoint_auth_method: str | None = None, - ) -> None: - """Initialize the OIDC proxy provider. - - Args: - config_url: URL of upstream configuration - strict: Optional strict flag for the configuration - client_id: Client ID registered with upstream server - 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 - 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. - redirect_path: Redirect path configured in upstream OAuth app (defaults to "/auth/callback") - allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients. - Patterns support wildcards (e.g., "http://localhost:*", "https://*.example.com/*"). - If None (default), only localhost redirect URIs are allowed. - If empty list, all redirect URIs are allowed (not recommended for production). - These are for MCP clients performing loopback redirects, NOT for the upstream OAuth app. - client_storage: An AsyncKeyValue-compatible store for client registrations, registrations are stored in memory if not provided - token_endpoint_auth_method: Token endpoint authentication method for upstream server. - Common values: "client_secret_basic", "client_secret_post", "none". - If None, authlib will use its default (typically "client_secret_basic"). - """ - if not config_url: - raise ValueError("Missing required config URL") - - if not client_id: - raise ValueError("Missing required client id") - - if not client_secret: - raise ValueError("Missing required client secret") - - if not base_url: - raise ValueError("Missing required base URL") - - if isinstance(config_url, str): - config_url = AnyHttpUrl(config_url) - - self.oidc_config = self.get_oidc_configuration( - config_url, strict, timeout_seconds - ) - if ( - not self.oidc_config.authorization_endpoint - or not self.oidc_config.token_endpoint - ): - logger.debug(f"Invalid OIDC Configuration: {self.oidc_config}") - raise ValueError("Missing required OIDC endpoints") - - revocation_endpoint = ( - str(self.oidc_config.revocation_endpoint) - if self.oidc_config.revocation_endpoint - else 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( - self.oidc_config.authorization_endpoint - ), - "upstream_token_endpoint": str(self.oidc_config.token_endpoint), - "upstream_client_id": client_id, - "upstream_client_secret": client_secret, - "upstream_revocation_endpoint": revocation_endpoint, - "token_verifier": token_verifier, - "base_url": base_url, - "issuer_url": issuer_url or base_url, - "service_documentation_url": self.oidc_config.service_documentation, - "allowed_client_redirect_uris": allowed_client_redirect_uris, - "client_storage": client_storage, - "token_endpoint_auth_method": token_endpoint_auth_method, - } - - if redirect_path: - init_kwargs["redirect_path"] = redirect_path - - if audience: - extra_params = {"audience": audience} - init_kwargs["extra_authorize_params"] = extra_params - init_kwargs["extra_token_params"] = extra_params - - super().__init__(**init_kwargs) - - def get_oidc_configuration( - self, - config_url: AnyHttpUrl, - strict: bool | None, - timeout_seconds: int | None, - ) -> OIDCConfiguration: - """Gets the OIDC configuration for the specified configuration URL. - - Args: - config_url: The OIDC configuration URL - strict: The strict flag for the configuration - timeout_seconds: HTTP request timeout in seconds - """ - return OIDCConfiguration.get_oidc_configuration( - config_url, strict=strict, timeout_seconds=timeout_seconds - ) - - def get_token_verifier( - self, - *, - algorithm: str | None = None, - audience: str | None = None, - required_scopes: list[str] | None = None, - timeout_seconds: int | None = None, - ) -> TokenVerifier: - """Creates the token verifier for the specified OIDC configuration and arguments. - - Args: - algorithm: Optional token verifier algorithm - audience: Optional token verifier audience - required_scopes: Optional token verifier required_scopes - timeout_seconds: HTTP request timeout in seconds - """ - return JWTVerifier( - jwks_uri=str(self.oidc_config.jwks_uri), - issuer=str(self.oidc_config.issuer), - algorithm=algorithm, - audience=audience, - required_scopes=required_scopes, - ) diff --git a/src/fastmcp/server/auth/providers/auth0.py b/src/fastmcp/server/auth/providers/auth0.py index 4d994ce98..71b4e63ac 100644 --- a/src/fastmcp/server/auth/providers/auth0.py +++ b/src/fastmcp/server/auth/providers/auth0.py @@ -25,7 +25,7 @@ from key_value.aio.protocols import AsyncKeyValue from pydantic import AnyHttpUrl, SecretStr, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict -from fastmcp.server.auth.oidc_proxy import OIDCProxy +from fastmcp.server.auth.oidc_dcr_proxy import OIDCDCRProxy from fastmcp.settings import ENV_FILE from fastmcp.utilities.auth import parse_scopes from fastmcp.utilities.logging import get_logger @@ -59,7 +59,7 @@ class Auth0ProviderSettings(BaseSettings): return parse_scopes(v) -class Auth0Provider(OIDCProxy): +class Auth0Provider(OIDCDCRProxy): """An Auth0 provider implementation for FastMCP. This provider is a complete Auth0 integration that's ready to use with diff --git a/src/fastmcp/server/auth/providers/aws.py b/src/fastmcp/server/auth/providers/aws.py index 31de6c9a0..84fe55eef 100644 --- a/src/fastmcp/server/auth/providers/aws.py +++ b/src/fastmcp/server/auth/providers/aws.py @@ -29,7 +29,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict from fastmcp.server.auth import TokenVerifier from fastmcp.server.auth.auth import AccessToken -from fastmcp.server.auth.oidc_proxy import OIDCProxy +from fastmcp.server.auth.oidc_dcr_proxy import OIDCDCRProxy from fastmcp.server.auth.providers.jwt import JWTVerifier from fastmcp.settings import ENV_FILE from fastmcp.utilities.auth import parse_scopes @@ -91,7 +91,7 @@ class AWSCognitoTokenVerifier(JWTVerifier): ) -class AWSCognitoProvider(OIDCProxy): +class AWSCognitoProvider(OIDCDCRProxy): """Complete AWS Cognito OAuth provider for FastMCP. This provider makes it trivial to add AWS Cognito OAuth protection to any diff --git a/tests/server/auth/oauth_dcr_proxy/test_oidc_proxy.py b/tests/server/auth/oauth_dcr_proxy/test_oidc_proxy.py index f45f835be..86384af3d 100644 --- a/tests/server/auth/oauth_dcr_proxy/test_oidc_proxy.py +++ b/tests/server/auth/oauth_dcr_proxy/test_oidc_proxy.py @@ -7,7 +7,7 @@ import pytest from httpx import Response from pydantic import AnyHttpUrl -from fastmcp.server.auth.oidc_proxy import OIDCConfiguration, OIDCProxy +from fastmcp.server.auth.oidc_dcr_proxy import OIDCConfiguration, OIDCDCRProxy from fastmcp.server.auth.providers.jwt import JWTVerifier TEST_ISSUER = "https://example.com" @@ -440,7 +440,7 @@ def validate_proxy(mock_get, proxy, oidc_config): assert proxy.oidc_config == oidc_config -class TestOIDCProxyInitialization: +class TestOIDCDCRProxyInitialization: """Tests for OIDC proxy initialization.""" def test_default_initialization(self, valid_oidc_configuration_dict): @@ -453,7 +453,7 @@ class TestOIDCProxyInitialization: ) mock_get.return_value = oidc_config - proxy = OIDCProxy( + proxy = OIDCDCRProxy( config_url=TEST_CONFIG_URL, client_id=TEST_CLIENT_ID, client_secret=TEST_CLIENT_SECRET, @@ -472,7 +472,7 @@ class TestOIDCProxyInitialization: ) mock_get.return_value = oidc_config - proxy = OIDCProxy( + proxy = OIDCDCRProxy( config_url=TEST_CONFIG_URL, client_id=TEST_CLIENT_ID, client_secret=TEST_CLIENT_SECRET, @@ -495,7 +495,7 @@ class TestOIDCProxyInitialization: ) mock_get.return_value = oidc_config - proxy = OIDCProxy( + proxy = OIDCDCRProxy( config_url=TEST_CONFIG_URL, client_id=TEST_CLIENT_ID, client_secret=TEST_CLIENT_SECRET, @@ -523,7 +523,7 @@ class TestOIDCProxyInitialization: ) mock_get.return_value = oidc_config - proxy = OIDCProxy( + proxy = OIDCDCRProxy( config_url=TEST_CONFIG_URL, client_id=TEST_CLIENT_ID, client_secret=TEST_CLIENT_SECRET, @@ -548,7 +548,7 @@ class TestOIDCProxyInitialization: ) mock_get.return_value = oidc_config - proxy = OIDCProxy( + proxy = OIDCDCRProxy( config_url=TEST_CONFIG_URL, client_id=TEST_CLIENT_ID, client_secret=TEST_CLIENT_SECRET, @@ -577,7 +577,7 @@ class TestOIDCProxyInitialization: mock_get.return_value = oidc_config with pytest.raises(ValueError, match="Missing required config URL"): - OIDCProxy( + OIDCDCRProxy( config_url=None, # type: ignore client_id=TEST_CLIENT_ID, client_secret=TEST_CLIENT_SECRET, @@ -597,7 +597,7 @@ class TestOIDCProxyInitialization: mock_get.return_value = oidc_config with pytest.raises(ValueError, match="Missing required client id"): - OIDCProxy( + OIDCDCRProxy( config_url=TEST_CONFIG_URL, client_id=None, # type: ignore client_secret=TEST_CLIENT_SECRET, @@ -617,7 +617,7 @@ class TestOIDCProxyInitialization: mock_get.return_value = oidc_config with pytest.raises(ValueError, match="Missing required client secret"): - OIDCProxy( + OIDCDCRProxy( config_url=TEST_CONFIG_URL, client_id=TEST_CLIENT_ID, client_secret=None, # type: ignore @@ -637,7 +637,7 @@ class TestOIDCProxyInitialization: mock_get.return_value = oidc_config with pytest.raises(ValueError, match="Missing required base URL"): - OIDCProxy( + OIDCDCRProxy( config_url=TEST_CONFIG_URL, client_id=TEST_CLIENT_ID, client_secret=TEST_CLIENT_SECRET, diff --git a/tests/server/auth/providers/test_auth0.py b/tests/server/auth/providers/test_auth0.py index 01f60e54b..87b6b2424 100644 --- a/tests/server/auth/providers/test_auth0.py +++ b/tests/server/auth/providers/test_auth0.py @@ -5,7 +5,7 @@ from unittest.mock import patch import pytest -from fastmcp.server.auth.oidc_proxy import OIDCConfiguration +from fastmcp.server.auth.oidc_dcr_proxy import OIDCConfiguration from fastmcp.server.auth.providers.auth0 import Auth0Provider, Auth0ProviderSettings from fastmcp.server.auth.providers.jwt import JWTVerifier diff --git a/tests/server/auth/test_oauth_proxy.py b/tests/server/auth/test_oauth_proxy.py deleted file mode 100644 index 89b4ac63b..000000000 --- a/tests/server/auth/test_oauth_proxy.py +++ /dev/null @@ -1,1297 +0,0 @@ -"""Comprehensive tests for OAuth Proxy Provider functionality. - -This test suite covers: -1. Initialization and configuration -2. Client registration (DCR) -3. Authorization flow -4. Token management -5. PKCE forwarding -6. Token endpoint authentication methods -7. E2E testing with mock OAuth provider -""" - -import asyncio -import secrets -import time -from unittest.mock import AsyncMock, Mock, patch -from urllib.parse import parse_qs, urlencode, urlparse - -import httpx -import pytest -from mcp.server.auth.provider import AuthorizationParams -from mcp.shared.auth import OAuthClientInformationFull -from pydantic import AnyUrl -from starlette.applications import Starlette -from starlette.responses import JSONResponse -from starlette.routing import Route - -from fastmcp import FastMCP -from fastmcp.server.auth.auth import AccessToken, RefreshToken, TokenVerifier -from fastmcp.server.auth.oauth_dcr_proxy import OAuthDCRProxy -from fastmcp.server.auth.providers.jwt import JWTVerifier - -# ============================================================================= -# Mock OAuth Provider for E2E Testing -# ============================================================================= - - -class MockOAuthProvider: - """Mock OAuth provider for testing OAuth proxy E2E flows. - - This provider simulates a complete OAuth server without requiring: - - Real authentication credentials - - Browser automation - - Network calls to external services - """ - - def __init__(self, port: int = 0): - self.port = port - self.base_url = f"http://localhost:{port}" - self.app = None - self.server = None - - # Storage for OAuth state - self.authorization_codes = {} - self.access_tokens = {} - self.refresh_tokens = {} - self.revoked_tokens = set() - - # Tracking for assertions - self.authorize_called = False - self.token_called = False - self.refresh_called = False - self.revoke_called = False - - # Configuration - self.require_pkce = False - self.token_endpoint_auth_method = "client_secret_basic" - - @property - def authorize_endpoint(self) -> str: - return f"{self.base_url}/authorize" - - @property - def token_endpoint(self) -> str: - return f"{self.base_url}/token" - - @property - def revocation_endpoint(self) -> str: - return f"{self.base_url}/revoke" - - def create_app(self) -> Starlette: - """Create the mock OAuth server application.""" - return Starlette( - routes=[ - Route("/authorize", self.handle_authorize), - Route("/token", self.handle_token, methods=["POST"]), - Route("/revoke", self.handle_revoke, methods=["POST"]), - ] - ) - - async def handle_authorize(self, request): - """Handle authorization requests.""" - self.authorize_called = True - query = dict(request.query_params) - - # Validate PKCE if required - if self.require_pkce and "code_challenge" not in query: - return JSONResponse( - {"error": "invalid_request", "error_description": "PKCE required"}, - status_code=400, - ) - - # Generate authorization code - code = secrets.token_urlsafe(32) - self.authorization_codes[code] = { - "client_id": query.get("client_id"), - "redirect_uri": query.get("redirect_uri"), - "state": query.get("state"), - "code_challenge": query.get("code_challenge"), - "code_challenge_method": query.get("code_challenge_method", "S256"), - "scope": query.get("scope"), - "created_at": time.time(), - } - - # Redirect back to callback - redirect_uri = query["redirect_uri"] - params = {"code": code} - if query.get("state"): - params["state"] = query["state"] - - redirect_url = f"{redirect_uri}?{urlencode(params)}" - return JSONResponse( - content={}, status_code=302, headers={"Location": redirect_url} - ) - - async def handle_token(self, request): - """Handle token requests.""" - self.token_called = True - form = await request.form() - grant_type = form.get("grant_type") - - if grant_type == "authorization_code": - code = form.get("code") - if code not in self.authorization_codes: - return JSONResponse( - {"error": "invalid_grant", "error_description": "Invalid code"}, - status_code=400, - ) - - # Validate PKCE if it was used - auth_data = self.authorization_codes[code] - if auth_data.get("code_challenge"): - verifier = form.get("code_verifier") - if not verifier: - return JSONResponse( - { - "error": "invalid_request", - "error_description": "Missing code_verifier", - }, - status_code=400, - ) - # In a real implementation, we'd validate the verifier - - # Generate tokens - access_token = f"mock_access_{secrets.token_hex(16)}" - refresh_token = f"mock_refresh_{secrets.token_hex(16)}" - - self.access_tokens[access_token] = { - "client_id": auth_data["client_id"], - "scope": auth_data.get("scope"), - "expires_at": time.time() + 3600, - } - self.refresh_tokens[refresh_token] = { - "client_id": auth_data["client_id"], - "scope": auth_data.get("scope"), - } - - # Clean up used code - del self.authorization_codes[code] - - return JSONResponse( - { - "access_token": access_token, - "token_type": "Bearer", - "expires_in": 3600, - "refresh_token": refresh_token, - "scope": auth_data.get("scope"), - } - ) - - elif grant_type == "refresh_token": - self.refresh_called = True - refresh_token = form.get("refresh_token") - - if refresh_token not in self.refresh_tokens: - return JSONResponse( - { - "error": "invalid_grant", - "error_description": "Invalid refresh token", - }, - status_code=400, - ) - - # Generate new access token - new_access = f"mock_access_{secrets.token_hex(16)}" - token_data = self.refresh_tokens[refresh_token] - - self.access_tokens[new_access] = { - "client_id": token_data["client_id"], - "scope": token_data.get("scope"), - "expires_at": time.time() + 3600, - } - - return JSONResponse( - { - "access_token": new_access, - "token_type": "Bearer", - "expires_in": 3600, - "refresh_token": refresh_token, # Same refresh token - "scope": token_data.get("scope"), - } - ) - - return JSONResponse({"error": "unsupported_grant_type"}, status_code=400) - - async def handle_revoke(self, request): - """Handle token revocation.""" - self.revoke_called = True - form = await request.form() - token = form.get("token") - - if token: - self.revoked_tokens.add(token) - # Remove from active tokens - self.access_tokens.pop(token, None) - self.refresh_tokens.pop(token, None) - - return JSONResponse({}) - - async def start(self): - """Start the mock OAuth server.""" - import socket - - from uvicorn import Config, Server - - self.app = self.create_app() - - # If port is 0, find an available port - if self.port == 0: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.bind(("127.0.0.1", 0)) - s.listen(1) - self.port = s.getsockname()[1] - - self.base_url = f"http://localhost:{self.port}" - config = Config( - self.app, - host="localhost", - port=self.port, - log_level="error", - ws="websockets-sansio", - ) - self.server = Server(config) - - # Start server in background - asyncio.create_task(self.server.serve()) - - # Wait for server to be ready - await asyncio.sleep(0.05) - - async def stop(self): - """Stop the mock OAuth server.""" - if self.server: - self.server.should_exit = True - await asyncio.sleep(0.01) - - def reset(self): - """Reset all state for next test.""" - self.authorization_codes.clear() - self.access_tokens.clear() - self.refresh_tokens.clear() - self.revoked_tokens.clear() - self.authorize_called = False - self.token_called = False - self.refresh_called = False - self.revoke_called = False - - -class MockTokenVerifier(TokenVerifier): - """Mock token verifier for testing.""" - - def __init__(self, required_scopes=None): - self.required_scopes = required_scopes or ["read", "write"] - self.verify_called = False - - async def verify_token(self, token: str) -> AccessToken: - """Mock token verification.""" - self.verify_called = True - return AccessToken( - token=token, - client_id="mock-client", - scopes=self.required_scopes, - expires_at=int(time.time() + 3600), - ) - - -# ============================================================================= -# Test Fixtures -# ============================================================================= - - -@pytest.fixture -def jwt_verifier(): - """Create a mock JWT verifier for testing.""" - verifier = Mock(spec=JWTVerifier) - verifier.required_scopes = ["read", "write"] - verifier.verify_token = Mock(return_value=None) - return verifier - - -@pytest.fixture -def oauth_proxy(jwt_verifier): - """Create a standard OAuthProxy instance for testing.""" - return OAuthDCRProxy( - upstream_authorization_endpoint="https://github.com/login/oauth/authorize", - upstream_token_endpoint="https://github.com/login/oauth/access_token", - upstream_client_id="test-client-id", - upstream_client_secret="test-client-secret", - token_verifier=jwt_verifier, - base_url="https://myserver.com", - redirect_path="/auth/callback", - ) - - -@pytest.fixture -async def mock_oauth_provider(): - """Create and start a mock OAuth provider.""" - provider = MockOAuthProvider() - await provider.start() - yield provider - await provider.stop() - - -# ============================================================================= -# Test Classes -# ============================================================================= - - -class TestOAuthProxyInitialization: - """Tests for OAuth proxy initialization and configuration.""" - - def test_basic_initialization(self, jwt_verifier): - """Test basic proxy initialization with required parameters.""" - proxy = OAuthDCRProxy( - upstream_authorization_endpoint="https://auth.example.com/authorize", - upstream_token_endpoint="https://auth.example.com/token", - upstream_client_id="client-123", - upstream_client_secret="secret-456", - token_verifier=jwt_verifier, - base_url="https://api.example.com", - ) - - assert ( - proxy._upstream_authorization_endpoint - == "https://auth.example.com/authorize" - ) - assert proxy._upstream_token_endpoint == "https://auth.example.com/token" - assert proxy._upstream_client_id == "client-123" - assert proxy._upstream_client_secret.get_secret_value() == "secret-456" - assert str(proxy.base_url) == "https://api.example.com/" - - def test_all_optional_parameters(self, jwt_verifier): - """Test initialization with all optional parameters.""" - proxy = OAuthDCRProxy( - upstream_authorization_endpoint="https://auth.example.com/authorize", - upstream_token_endpoint="https://auth.example.com/token", - upstream_client_id="client-123", - upstream_client_secret="secret-456", - upstream_revocation_endpoint="https://auth.example.com/revoke", - token_verifier=jwt_verifier, - base_url="https://api.example.com", - redirect_path="/custom/callback", - issuer_url="https://issuer.example.com", - service_documentation_url="https://docs.example.com", - allowed_client_redirect_uris=["http://localhost:*"], - valid_scopes=["custom", "scopes"], - forward_pkce=False, - token_endpoint_auth_method="client_secret_post", - ) - - assert proxy._upstream_revocation_endpoint == "https://auth.example.com/revoke" - assert proxy._redirect_path == "/custom/callback" - assert proxy._forward_pkce is False - assert proxy._token_endpoint_auth_method == "client_secret_post" - assert proxy.client_registration_options is not None - assert proxy.client_registration_options.valid_scopes == ["custom", "scopes"] - - def test_redirect_path_normalization(self, jwt_verifier): - """Test that redirect_path is normalized with leading slash.""" - proxy = OAuthDCRProxy( - upstream_authorization_endpoint="https://auth.com/authorize", - upstream_token_endpoint="https://auth.com/token", - upstream_client_id="client", - upstream_client_secret="secret", - token_verifier=jwt_verifier, - base_url="https://api.com", - redirect_path="auth/callback", # No leading slash - ) - assert proxy._redirect_path == "/auth/callback" - - -class TestOAuthProxyClientRegistration: - """Tests for OAuth proxy client registration (DCR).""" - - async def test_register_client(self, oauth_proxy): - """Test client registration creates ProxyDCRClient.""" - client_info = OAuthClientInformationFull( - client_id="original-client", - client_secret="original-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - await oauth_proxy.register_client(client_info) - - # Client should be retrievable with original credentials - stored = await oauth_proxy.get_client("original-client") - assert stored is not None - assert stored.client_id == "original-client" - assert stored.client_secret == "original-secret" - - async def test_get_registered_client(self, oauth_proxy): - """Test retrieving a registered client.""" - client_info = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:8080/callback")], - ) - await oauth_proxy.register_client(client_info) - - retrieved = await oauth_proxy.get_client("test-client") - assert retrieved is not None - assert retrieved.client_id == "test-client" - - async def test_get_unregistered_client_returns_none(self, oauth_proxy): - """Test that unregistered clients return None.""" - client = await oauth_proxy.get_client("unknown-client") - assert client is None - - -class TestOAuthProxyAuthorization: - """Tests for OAuth proxy authorization flow.""" - - async def test_authorize_creates_transaction(self, oauth_proxy): - """Test that authorize creates transaction and redirects to consent.""" - client = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:54321/callback")], - ) - - # Register client first (required for consent flow) - await oauth_proxy.register_client(client) - - params = AuthorizationParams( - redirect_uri=AnyUrl("http://localhost:54321/callback"), - redirect_uri_provided_explicitly=True, - state="client-state-123", - code_challenge="challenge-abc", - code_challenge_method="S256", - scopes=["read", "write"], - ) - - redirect_url = await oauth_proxy.authorize(client, params) - - # Parse the redirect URL - parsed = urlparse(redirect_url) - query_params = parse_qs(parsed.query) - - # Should redirect to consent page - assert "/consent" in redirect_url - assert "txn_id" in query_params - - # Verify transaction was stored with correct data - txn_id = query_params["txn_id"][0] - transaction = await oauth_proxy._transaction_store.get(key=txn_id) - assert transaction is not None - assert transaction.client_id == "test-client" - assert transaction.code_challenge == "challenge-abc" - assert transaction.client_state == "client-state-123" - assert transaction.scopes == ["read", "write"] - - -class TestOAuthProxyPKCE: - """Tests for OAuth proxy PKCE forwarding.""" - - @pytest.fixture - def proxy_with_pkce(self, jwt_verifier): - return OAuthDCRProxy( - upstream_authorization_endpoint="https://oauth.example.com/authorize", - upstream_token_endpoint="https://oauth.example.com/token", - upstream_client_id="upstream-client", - upstream_client_secret="upstream-secret", - token_verifier=jwt_verifier, - base_url="https://proxy.example.com", - forward_pkce=True, - ) - - @pytest.fixture - def proxy_without_pkce(self, jwt_verifier): - return OAuthDCRProxy( - upstream_authorization_endpoint="https://oauth.example.com/authorize", - upstream_token_endpoint="https://oauth.example.com/token", - upstream_client_id="upstream-client", - upstream_client_secret="upstream-secret", - token_verifier=jwt_verifier, - base_url="https://proxy.example.com", - forward_pkce=False, - ) - - async def test_pkce_forwarding_enabled(self, proxy_with_pkce): - """Test that proxy generates and forwards its own PKCE.""" - client = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - # Register client first - await proxy_with_pkce.register_client(client) - - params = AuthorizationParams( - redirect_uri=AnyUrl("http://localhost:12345/callback"), - redirect_uri_provided_explicitly=True, - state="client-state", - code_challenge="client_challenge", - scopes=["read"], - ) - - redirect_url = await proxy_with_pkce.authorize(client, params) - query_params = parse_qs(urlparse(redirect_url).query) - - # Should redirect to consent page - assert "/consent" in redirect_url - assert "txn_id" in query_params - - # Transaction should store both challenges - txn_id = query_params["txn_id"][0] - transaction = await proxy_with_pkce._transaction_store.get(key=txn_id) - assert transaction is not None - assert transaction.code_challenge == "client_challenge" # Client's - assert transaction.proxy_code_verifier is not None # Proxy's verifier - # Proxy code challenge is computed from verifier when building upstream URL - # Just verify the verifier exists and is different from client's challenge - assert len(transaction.proxy_code_verifier) > 0 - - async def test_pkce_forwarding_disabled(self, proxy_without_pkce): - """Test that PKCE is not forwarded when disabled.""" - client = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - # Register client first - await proxy_without_pkce.register_client(client) - - params = AuthorizationParams( - redirect_uri=AnyUrl("http://localhost:12345/callback"), - redirect_uri_provided_explicitly=True, - state="client-state", - code_challenge="client_challenge", - scopes=["read"], - ) - - redirect_url = await proxy_without_pkce.authorize(client, params) - query_params = parse_qs(urlparse(redirect_url).query) - - # Should redirect to consent page - assert "/consent" in redirect_url - assert "txn_id" in query_params - - # Client's challenge still stored, but no proxy PKCE - txn_id = query_params["txn_id"][0] - transaction = await proxy_without_pkce._transaction_store.get(key=txn_id) - assert transaction is not None - assert transaction.code_challenge == "client_challenge" - assert transaction.proxy_code_verifier is None # No proxy PKCE when disabled - - -class TestOAuthProxyTokenEndpointAuth: - """Tests for token endpoint authentication methods.""" - - def test_token_auth_method_initialization(self, jwt_verifier): - """Test different token endpoint auth methods.""" - # client_secret_post - proxy_post = OAuthDCRProxy( - upstream_authorization_endpoint="https://oauth.example.com/authorize", - upstream_token_endpoint="https://oauth.example.com/token", - upstream_client_id="client", - upstream_client_secret="secret", - token_verifier=jwt_verifier, - base_url="https://proxy.example.com", - token_endpoint_auth_method="client_secret_post", - ) - assert proxy_post._token_endpoint_auth_method == "client_secret_post" - - # client_secret_basic (default) - proxy_basic = OAuthDCRProxy( - upstream_authorization_endpoint="https://oauth.example.com/authorize", - upstream_token_endpoint="https://oauth.example.com/token", - upstream_client_id="client", - upstream_client_secret="secret", - token_verifier=jwt_verifier, - base_url="https://proxy.example.com", - token_endpoint_auth_method="client_secret_basic", - ) - assert proxy_basic._token_endpoint_auth_method == "client_secret_basic" - - # None (use authlib default) - proxy_default = OAuthDCRProxy( - upstream_authorization_endpoint="https://oauth.example.com/authorize", - upstream_token_endpoint="https://oauth.example.com/token", - upstream_client_id="client", - upstream_client_secret="secret", - token_verifier=jwt_verifier, - base_url="https://proxy.example.com", - ) - assert proxy_default._token_endpoint_auth_method is None - - async def test_token_auth_method_passed_to_client(self, jwt_verifier): - """Test that auth method is passed to AsyncOAuth2Client.""" - proxy = OAuthDCRProxy( - upstream_authorization_endpoint="https://oauth.example.com/authorize", - upstream_token_endpoint="https://oauth.example.com/token", - upstream_client_id="client-id", - upstream_client_secret="client-secret", - token_verifier=jwt_verifier, - base_url="https://proxy.example.com", - token_endpoint_auth_method="client_secret_post", - ) - - # First, create a valid FastMCP token via full OAuth flow - client = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - # Mock the upstream OAuth provider response - with patch( - "fastmcp.server.auth.oauth_dcr_proxy.AsyncOAuth2Client" - ) as MockClient: - mock_client = AsyncMock() - - # Mock initial token exchange (authorization code flow) - mock_client.fetch_token = AsyncMock( - return_value={ - "access_token": "upstream-access-token", - "refresh_token": "upstream-refresh-token", - "expires_in": 3600, - "token_type": "Bearer", - } - ) - - # Mock token refresh - mock_client.refresh_token = AsyncMock( - return_value={ - "access_token": "new-upstream-token", - "refresh_token": "new-upstream-refresh", - "expires_in": 3600, - "token_type": "Bearer", - } - ) - MockClient.return_value = mock_client - - # Register client and do initial OAuth flow to get valid FastMCP tokens - await proxy.register_client(client) - - # Store client code that would be created during OAuth callback - from fastmcp.server.auth.oauth_dcr_proxy import ClientCode - - client_code = ClientCode( - code="test-auth-code", - client_id="test-client", - redirect_uri="http://localhost:12345/callback", - code_challenge="", - code_challenge_method="S256", - scopes=["read"], - idp_tokens={ - "access_token": "upstream-access-token", - "refresh_token": "upstream-refresh-token", - "expires_in": 3600, - "token_type": "Bearer", - }, - expires_at=time.time() + 300, - created_at=time.time(), - ) - await proxy._code_store.put(key=client_code.code, value=client_code) - - # Exchange authorization code to get FastMCP tokens - from mcp.server.auth.provider import AuthorizationCode - - auth_code = AuthorizationCode( - code="test-auth-code", - scopes=["read"], - expires_at=time.time() + 300, - client_id="test-client", - code_challenge="", - redirect_uri=AnyUrl("http://localhost:12345/callback"), - redirect_uri_provided_explicitly=True, - ) - result = await proxy.exchange_authorization_code( - client=client, - authorization_code=auth_code, - ) - - # Now test refresh with the valid FastMCP refresh token - assert result.refresh_token is not None - fastmcp_refresh = RefreshToken( - token=result.refresh_token, - client_id="test-client", - scopes=["read"], - expires_at=None, - ) - - # Reset mock to check refresh call - MockClient.reset_mock() - mock_client.refresh_token = AsyncMock( - return_value={ - "access_token": "new-upstream-token-2", - "refresh_token": "new-upstream-refresh-2", - "expires_in": 3600, - "token_type": "Bearer", - } - ) - MockClient.return_value = mock_client - - await proxy.exchange_refresh_token(client, fastmcp_refresh, ["read"]) - - # Verify auth method was passed to OAuth client - MockClient.assert_called_with( - client_id="client-id", - client_secret="client-secret", - token_endpoint_auth_method="client_secret_post", - timeout=30.0, - ) - - -class TestOAuthProxyE2E: - """End-to-end tests using mock OAuth provider.""" - - async def test_full_oauth_flow_with_mock_provider(self, mock_oauth_provider): - """Test complete OAuth flow with mock provider.""" - # Create proxy pointing to mock provider - proxy = OAuthDCRProxy( - upstream_authorization_endpoint=mock_oauth_provider.authorize_endpoint, - upstream_token_endpoint=mock_oauth_provider.token_endpoint, - upstream_client_id="mock-client", - upstream_client_secret="mock-secret", - token_verifier=MockTokenVerifier(), - base_url="http://localhost:8000", - ) - - # Create FastMCP server with proxy - server = FastMCP("Test Server", auth=proxy) - - @server.tool - def protected_tool() -> str: - return "Protected data" - - # Start authorization flow - client_info = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - # Register client first - await proxy.register_client(client_info) - - params = AuthorizationParams( - redirect_uri=AnyUrl("http://localhost:12345/callback"), - redirect_uri_provided_explicitly=True, - state="client-state", - code_challenge="", # Empty string for no PKCE - scopes=["read"], - ) - - # Get authorization URL (now returns consent redirect) - auth_url = await proxy.authorize(client_info, params) - - # Should redirect to consent page - assert "/consent" in auth_url - query_params = parse_qs(urlparse(auth_url).query) - assert "txn_id" in query_params - - # Verify transaction was created with correct configuration - txn_id = query_params["txn_id"][0] - transaction = await proxy._transaction_store.get(key=txn_id) - assert transaction is not None - assert transaction.client_id == "test-client" - assert transaction.scopes == ["read"] - # Transaction ID itself is used as upstream state parameter - assert transaction.txn_id == txn_id - - async def test_token_refresh_with_mock_provider(self, mock_oauth_provider): - """Test token refresh flow with mock provider.""" - proxy = OAuthDCRProxy( - upstream_authorization_endpoint=mock_oauth_provider.authorize_endpoint, - upstream_token_endpoint=mock_oauth_provider.token_endpoint, - upstream_client_id="mock-client", - upstream_client_secret="mock-secret", - token_verifier=MockTokenVerifier(), - base_url="http://localhost:8000", - ) - - client = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - # Register client first - await proxy.register_client(client) - - # Set up initial upstream tokens in mock provider - upstream_refresh_token = "mock_refresh_initial" - mock_oauth_provider.refresh_tokens[upstream_refresh_token] = { - "client_id": "mock-client", - "scope": "read write", - } - - with patch( - "fastmcp.server.auth.oauth_dcr_proxy.AsyncOAuth2Client" - ) as MockClient: - mock_client = AsyncMock() - - # Mock initial token exchange to get FastMCP tokens - mock_client.fetch_token = AsyncMock( - return_value={ - "access_token": "upstream-access-initial", - "refresh_token": upstream_refresh_token, - "expires_in": 3600, - "token_type": "Bearer", - } - ) - - # Configure mock to call real provider for refresh - async def mock_refresh(*args, **kwargs): - async with httpx.AsyncClient() as http: - response = await http.post( - mock_oauth_provider.token_endpoint, - data={ - "grant_type": "refresh_token", - "refresh_token": upstream_refresh_token, - }, - ) - return response.json() - - mock_client.refresh_token = mock_refresh - MockClient.return_value = mock_client - - # Store client code that would be created during OAuth callback - from fastmcp.server.auth.oauth_dcr_proxy import ClientCode - - client_code = ClientCode( - code="test-auth-code", - client_id="test-client", - redirect_uri="http://localhost:12345/callback", - code_challenge="", - code_challenge_method="S256", - scopes=["read", "write"], - idp_tokens={ - "access_token": "upstream-access-initial", - "refresh_token": upstream_refresh_token, - "expires_in": 3600, - "token_type": "Bearer", - }, - expires_at=time.time() + 300, - created_at=time.time(), - ) - await proxy._code_store.put(key=client_code.code, value=client_code) - - # Exchange authorization code to get FastMCP tokens - from mcp.server.auth.provider import AuthorizationCode - - auth_code = AuthorizationCode( - code="test-auth-code", - scopes=["read", "write"], - expires_at=time.time() + 300, - client_id="test-client", - code_challenge="", - redirect_uri=AnyUrl("http://localhost:12345/callback"), - redirect_uri_provided_explicitly=True, - ) - initial_result = await proxy.exchange_authorization_code( - client=client, - authorization_code=auth_code, - ) - - # Now test refresh with the valid FastMCP refresh token - assert initial_result.refresh_token is not None - fastmcp_refresh = RefreshToken( - token=initial_result.refresh_token, - client_id="test-client", - scopes=["read"], - expires_at=None, - ) - - result = await proxy.exchange_refresh_token( - client, fastmcp_refresh, ["read"] - ) - - # Should return new FastMCP tokens (not upstream tokens) - assert result.access_token != "upstream-access-initial" - # FastMCP tokens are JWTs (have 3 segments) - assert len(result.access_token.split(".")) == 3 - assert mock_oauth_provider.refresh_called - - async def test_pkce_validation_with_mock_provider(self, mock_oauth_provider): - """Test PKCE validation with mock provider.""" - mock_oauth_provider.require_pkce = True - - proxy = OAuthDCRProxy( - upstream_authorization_endpoint=mock_oauth_provider.authorize_endpoint, - upstream_token_endpoint=mock_oauth_provider.token_endpoint, - upstream_client_id="mock-client", - upstream_client_secret="mock-secret", - token_verifier=MockTokenVerifier(), - base_url="http://localhost:8000", - forward_pkce=True, # Enable PKCE forwarding - ) - - client = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - # Register client first - await proxy.register_client(client) - - params = AuthorizationParams( - redirect_uri=AnyUrl("http://localhost:12345/callback"), - redirect_uri_provided_explicitly=True, - state="client-state", - code_challenge="client_challenge_value", - code_challenge_method="S256", - scopes=["read"], - ) - - # Start authorization with PKCE - auth_url = await proxy.authorize(client, params) - query_params = parse_qs(urlparse(auth_url).query) - - # Should redirect to consent page - assert "/consent" in auth_url - assert "txn_id" in query_params - - # Transaction should have proxy's PKCE verifier (different from client's) - txn_id = query_params["txn_id"][0] - transaction = await proxy._transaction_store.get(key=txn_id) - assert transaction is not None - assert ( - transaction.code_challenge == "client_challenge_value" - ) # Client's challenge - assert transaction.proxy_code_verifier is not None # Proxy generated its own - # Proxy code challenge is computed from verifier when needed - assert len(transaction.proxy_code_verifier) > 0 - - -class TestParameterForwarding: - """Tests for forwarding custom parameters to upstream OAuth provider.""" - - @pytest.fixture - def proxy_with_extra_params(self, jwt_verifier): - """Create OAuthProxy with extra parameters configured.""" - return OAuthDCRProxy( - upstream_authorization_endpoint="https://oauth.example.com/authorize", - upstream_token_endpoint="https://oauth.example.com/token", - upstream_client_id="upstream-client", - upstream_client_secret="upstream-secret", - token_verifier=jwt_verifier, - base_url="https://proxy.example.com", - extra_authorize_params={"audience": "https://api.example.com"}, - extra_token_params={"audience": "https://api.example.com"}, - ) - - @pytest.fixture - def proxy_without_extra_params(self, jwt_verifier): - """Create OAuthProxy without extra parameters.""" - return OAuthDCRProxy( - upstream_authorization_endpoint="https://oauth.example.com/authorize", - upstream_token_endpoint="https://oauth.example.com/token", - upstream_client_id="upstream-client", - upstream_client_secret="upstream-secret", - token_verifier=jwt_verifier, - base_url="https://proxy.example.com", - ) - - async def test_resource_parameter_forwarding(self, proxy_without_extra_params): - """Test that RFC 8707 resource parameter is forwarded from client request.""" - client = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - # Register client first - await proxy_without_extra_params.register_client(client) - - params = AuthorizationParams( - redirect_uri=AnyUrl("http://localhost:12345/callback"), - redirect_uri_provided_explicitly=True, - state="client-state", - code_challenge="client_challenge", - scopes=["read"], - resource="https://api.example.com/v1", # RFC 8707 resource indicator - ) - - redirect_url = await proxy_without_extra_params.authorize(client, params) - query_params = parse_qs(urlparse(redirect_url).query) - - # Should redirect to consent page - assert "/consent" in redirect_url - assert "txn_id" in query_params - - # Resource parameter should be stored in transaction for upstream forwarding - txn_id = query_params["txn_id"][0] - transaction = await proxy_without_extra_params._transaction_store.get( - key=txn_id - ) - assert transaction is not None - assert transaction.resource == "https://api.example.com/v1" - - async def test_extra_authorize_params(self, proxy_with_extra_params): - """Test that extra authorization parameters are included.""" - client = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - # Register client first - await proxy_with_extra_params.register_client(client) - - params = AuthorizationParams( - redirect_uri=AnyUrl("http://localhost:12345/callback"), - redirect_uri_provided_explicitly=True, - state="client-state", - code_challenge="client_challenge", - scopes=["read"], - ) - - redirect_url = await proxy_with_extra_params.authorize(client, params) - query_params = parse_qs(urlparse(redirect_url).query) - - # Should redirect to consent page - assert "/consent" in redirect_url - assert "txn_id" in query_params - - # Extra audience parameter is configured at proxy level (not per-transaction) - txn_id = query_params["txn_id"][0] - transaction = await proxy_with_extra_params._transaction_store.get(key=txn_id) - assert transaction is not None - # Verify proxy has extra params configured - assert ( - proxy_with_extra_params._extra_authorize_params.get("audience") - == "https://api.example.com" - ) - - async def test_resource_and_extra_params_together(self, proxy_with_extra_params): - """Test that both resource and extra params can be used together.""" - client = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - # Register client first - await proxy_with_extra_params.register_client(client) - - params = AuthorizationParams( - redirect_uri=AnyUrl("http://localhost:12345/callback"), - redirect_uri_provided_explicitly=True, - state="client-state", - code_challenge="client_challenge", - scopes=["read"], - resource="https://resource.example.com", # Client-specified resource - ) - - redirect_url = await proxy_with_extra_params.authorize(client, params) - query_params = parse_qs(urlparse(redirect_url).query) - - # Should redirect to consent page - assert "/consent" in redirect_url - assert "txn_id" in query_params - - # Resource stored in transaction, extra params configured at proxy level - txn_id = query_params["txn_id"][0] - transaction = await proxy_with_extra_params._transaction_store.get(key=txn_id) - assert transaction is not None - assert transaction.resource == "https://resource.example.com" - assert ( - proxy_with_extra_params._extra_authorize_params.get("audience") - == "https://api.example.com" - ) - - async def test_no_extra_params_when_not_configured( - self, proxy_without_extra_params - ): - """Test that no extra params are added when not configured.""" - client = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - params = AuthorizationParams( - redirect_uri=AnyUrl("http://localhost:12345/callback"), - redirect_uri_provided_explicitly=True, - state="client-state", - code_challenge="client_challenge", - scopes=["read"], - # No resource parameter - ) - - redirect_url = await proxy_without_extra_params.authorize(client, params) - query_params = parse_qs(urlparse(redirect_url).query) - - # No audience parameter should be present (not configured) - assert "audience" not in query_params - # No resource parameter should be present (not provided by client) - assert "resource" not in query_params - - async def test_multiple_extra_params(self, jwt_verifier): - """Test multiple extra parameters can be configured and forwarded.""" - proxy = OAuthDCRProxy( - upstream_authorization_endpoint="https://oauth.example.com/authorize", - upstream_token_endpoint="https://oauth.example.com/token", - upstream_client_id="upstream-client", - upstream_client_secret="upstream-secret", - token_verifier=jwt_verifier, - base_url="https://proxy.example.com", - extra_authorize_params={ - "audience": "https://api.example.com", - "prompt": "consent", - "max_age": "3600", - }, - ) - - client = OAuthClientInformationFull( - client_id="test-client", - client_secret="test-secret", - redirect_uris=[AnyUrl("http://localhost:12345/callback")], - ) - - # Register client first - await proxy.register_client(client) - - params = AuthorizationParams( - redirect_uri=AnyUrl("http://localhost:12345/callback"), - redirect_uri_provided_explicitly=True, - state="client-state", - code_challenge="client_challenge", - scopes=["read"], - ) - - redirect_url = await proxy.authorize(client, params) - query_params = parse_qs(urlparse(redirect_url).query) - - # Should redirect to consent page - assert "/consent" in redirect_url - assert "txn_id" in query_params - - # All extra parameters configured at proxy level - txn_id = query_params["txn_id"][0] - transaction = await proxy._transaction_store.get(key=txn_id) - assert transaction is not None - # Verify proxy has all extra params configured - assert ( - proxy._extra_authorize_params.get("audience") == "https://api.example.com" - ) - assert proxy._extra_authorize_params.get("prompt") == "consent" - assert proxy._extra_authorize_params.get("max_age") == "3600" - - async def test_token_endpoint_invalid_client_error(self, jwt_verifier): - """Test that invalid client_id returns OAuth 2.1 compliant error response. - - When a client ID is not found during token exchange, the proxy should: - 1. Return HTTP 401 status code - 2. Use 'invalid_client' error code instead of 'unauthorized_client' - - This aligns with OAuth 2.1 spec and enables Claude's automatic client re-registration. - """ - from starlette.applications import Starlette - from starlette.testclient import TestClient - - proxy = OAuthDCRProxy( - upstream_authorization_endpoint="https://oauth.example.com/authorize", - upstream_token_endpoint="https://oauth.example.com/token", - upstream_client_id="upstream-client", - upstream_client_secret="upstream-secret", - token_verifier=jwt_verifier, - base_url="https://proxy.example.com", - ) - - # Create a test app with OAuth routes - app = Starlette(routes=proxy.get_routes()) - - # Test the token endpoint with an invalid (non-existent) client_id - with TestClient(app) as client: - response = client.post( - "/token", - data={ - "grant_type": "authorization_code", - "code": "test-auth-code", - "client_id": "non-existent-client-id", - "code_verifier": "test-code-verifier", - "redirect_uri": "http://localhost:12345/callback", - }, - headers={ - "Content-Type": "application/x-www-form-urlencoded", - }, - ) - - # Verify OAuth 2.1 compliant error response - assert response.status_code == 401, ( - f"Expected 401 but got {response.status_code}" - ) - - error_data = response.json() - assert error_data["error"] == "invalid_client", ( - f"Expected 'invalid_client' but got '{error_data.get('error')}'" - ) - assert "Invalid client_id" in error_data["error_description"] - - # Verify proper cache headers are set - assert response.headers.get("Cache-Control") == "no-store" - assert response.headers.get("Pragma") == "no-cache" - - -class TestTokenHandlerErrorTransformation: - """Tests for TokenHandler's OAuth 2.1 compliant error transformation.""" - - def test_transforms_client_auth_failure_to_invalid_client_401(self): - """Test that client authentication failures return invalid_client with 401.""" - from mcp.server.auth.handlers.token import TokenErrorResponse - - from fastmcp.server.auth.oauth_dcr_proxy import TokenHandler - - handler = TokenHandler(provider=Mock(), client_authenticator=Mock()) - - # Simulate error from ClientAuthenticator.authenticate() failure - error_response = TokenErrorResponse( - error="unauthorized_client", - error_description="Invalid client_id 'test-client-id'", - ) - - response = handler.response(error_response) - - # Should transform to OAuth 2.1 compliant response - assert response.status_code == 401 - assert b'"error":"invalid_client"' in response.body - assert ( - b'"error_description":"Invalid client_id \'test-client-id\'"' - in response.body - ) - - def test_does_not_transform_grant_type_unauthorized_to_invalid_client(self): - """Test that grant type authorization errors stay as unauthorized_client with 400.""" - from mcp.server.auth.handlers.token import TokenErrorResponse - - from fastmcp.server.auth.oauth_dcr_proxy import TokenHandler - - handler = TokenHandler(provider=Mock(), client_authenticator=Mock()) - - # Simulate error from grant_type not in client_info.grant_types - error_response = TokenErrorResponse( - error="unauthorized_client", - error_description="Client not authorized for this grant type", - ) - - response = handler.response(error_response) - - # Should NOT transform - keep as 400 unauthorized_client - assert response.status_code == 400 - assert b'"error":"unauthorized_client"' in response.body - - def test_does_not_transform_other_errors(self): - """Test that other error types pass through unchanged.""" - from mcp.server.auth.handlers.token import TokenErrorResponse - - from fastmcp.server.auth.oauth_dcr_proxy import TokenHandler - - handler = TokenHandler(provider=Mock(), client_authenticator=Mock()) - - error_response = TokenErrorResponse( - error="invalid_grant", - error_description="Authorization code has expired", - ) - - response = handler.response(error_response) - - # Should pass through unchanged - assert response.status_code == 400 - assert b'"error":"invalid_grant"' in response.body