From 654335607c512ce8b4993624fce650f70dfdcfe4 Mon Sep 17 00:00:00 2001 From: vijaydeepsinha Date: Sun, 19 Jul 2026 06:45:02 +0530 Subject: [PATCH] Fix-issue-4284 : Add Auth0MCPProvider for Auth0 Auth for MCP (#4411) * Add Auth0MCPProvider for Auth0 Auth for MCP * Document Auth0 MCP provider integration * Add Auth0MCPProvider scope and auth rejection tests Cover permissions-based required_scopes enforcement and unauthenticated MCP 401 responses. * fixed documentation * Narrow Auth0 docs to integration guide only * Fix Auth0 provider: use httpx2 instead of httpx httpx is a dev-only transitive dependency in this repo; runtime installs declare httpx2 exclusively. The module-level 'import httpx' in auth0.py broke import on a clean install of fastmcp or fastmcp-slim[server]. --------- Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> --- docs/integrations/auth0.mdx | 49 ++- examples/auth/auth0_mcp/README.md | 28 ++ examples/auth/auth0_mcp/client.py | 21 + examples/auth/auth0_mcp/server.py | 39 ++ .../fastmcp/server/auth/providers/auth0.py | 179 +++++++- tests/server/auth/providers/test_auth0_mcp.py | 392 ++++++++++++++++++ 6 files changed, 700 insertions(+), 8 deletions(-) create mode 100644 examples/auth/auth0_mcp/README.md create mode 100644 examples/auth/auth0_mcp/client.py create mode 100644 examples/auth/auth0_mcp/server.py create mode 100644 tests/server/auth/providers/test_auth0_mcp.py diff --git a/docs/integrations/auth0.mdx b/docs/integrations/auth0.mdx index 65f9d3873..6ca28b88c 100644 --- a/docs/integrations/auth0.mdx +++ b/docs/integrations/auth0.mdx @@ -9,9 +9,54 @@ import { VersionBadge } from "/snippets/version-badge.mdx" -This guide shows you how to secure your FastMCP server using **Auth0 OAuth**. While Auth0 does have support for Dynamic Client Registration, it is not enabled by default so this integration uses the [**OIDC Proxy**](/servers/auth/oidc-proxy) pattern to bridge Auth0's dynamic OIDC configuration with MCP's authentication requirements. +FastMCP supports two Auth0 integration paths: -## Configuration +- **[Auth for MCP](#auth-for-mcp-dcr)** — Auth0 handles OAuth, DCR, and CIMD; FastMCP validates tokens (`Auth0MCPProvider`). Use this for MCP-native clients and Auth0's [Auth for MCP](https://auth0.com/ai/docs/mcp/intro/overview) setup. +- **[OIDC Proxy](#oidc-proxy-fixed-credentials)** — FastMCP proxies OAuth with fixed application credentials (`Auth0Provider`). Use this when you manage an Auth0 application manually and do not need tenant-level DCR. + +## Auth for MCP (DCR) + + + +This path uses the [**Remote OAuth**](/servers/auth/remote-oauth) pattern. Auth0 acts as the authorization server; FastMCP is the resource server. + +### Prerequisites + +1. An **[Auth0 account](https://auth0.com/)** with **Auth for MCP** enabled +2. **Resource Parameter Compatibility Profile** enabled (Settings → Advanced) +3. Your FastMCP server URL (use `http://127.0.0.1:8000` in development — not `localhost`) + +See Auth0's [authorization quickstart](https://auth0.com/ai/docs/mcp/get-started/authorization-for-your-mcp-server) for tenant setup (API identifier, domain-level connections, CIMD approval). + +### Step 1: Create an Auth0 API + +Create an API (Resource Server) whose **identifier** is your MCP resource URL, for example `http://127.0.0.1:8000/mcp`. Use `RS256` signing and the `rfc9068_profile_authz` token dialect if you need `permissions` claims on tokens. + +When the server starts, it logs the exact `aud` value it validates — your API identifier must match. + +### Step 2: FastMCP configuration + +```python server_mcp.py +from fastmcp import FastMCP +from fastmcp.server.auth.providers.auth0 import Auth0MCPProvider + +auth_provider = Auth0MCPProvider( + config_url="https://YOUR_TENANT.auth0.com/.well-known/openid-configuration", + base_url="http://127.0.0.1:8000", +) + +mcp = FastMCP(name="Auth0 MCP Server", auth=auth_provider) +``` + +No `client_id` or `client_secret` is required on the FastMCP side — MCP clients register with Auth0 directly. + +### Testing + +See `examples/auth/auth0_mcp/` for a runnable server and DCR client. Set `AUTH0_CONFIG_URL` to your tenant's OIDC discovery URL before starting the server. + +## OIDC Proxy (fixed credentials) + +This integration uses the [**OIDC Proxy**](/servers/auth/oidc-proxy) pattern when you use a fixed Auth0 application instead of tenant-level DCR. ### Prerequisites diff --git a/examples/auth/auth0_mcp/README.md b/examples/auth/auth0_mcp/README.md new file mode 100644 index 000000000..e8e0e6bd6 --- /dev/null +++ b/examples/auth/auth0_mcp/README.md @@ -0,0 +1,28 @@ +# Auth0 Auth for MCP Example + +Protects a FastMCP server with Auth0 [Auth for MCP](https://auth0.com/ai/docs/mcp/intro/overview). Auth0 handles OAuth and client registration; FastMCP validates access tokens. + +## Auth0 setup + +1. Enable **Resource Parameter Compatibility Profile** (Settings → Advanced). +2. Create an API whose identifier is `http://127.0.0.1:8000/mcp` (must match the URL logged at server startup). +3. Promote your login connections to domain-level (required for third-party DCR clients). + +See Auth0's [authorization quickstart](https://auth0.com/ai/docs/mcp/get-started/authorization-for-your-mcp-server) for details. + +## Running + +```bash +export AUTH0_CONFIG_URL="https://YOUR_TENANT.auth0.com/.well-known/openid-configuration" +python server.py +``` + +In another terminal: + +```bash +python client.py +``` + +Use `127.0.0.1` consistently — mixing `localhost` and `127.0.0.1` breaks audience validation. + +For troubleshooting (DCR grants, token exchange errors, MCP Inspector), see the [Auth0 integration guide](https://gofastmcp.com/integrations/auth0). diff --git a/examples/auth/auth0_mcp/client.py b/examples/auth/auth0_mcp/client.py new file mode 100644 index 000000000..24985ae9c --- /dev/null +++ b/examples/auth/auth0_mcp/client.py @@ -0,0 +1,21 @@ +"""Auth0 Auth for MCP client example.""" + +import asyncio + +from fastmcp import Client +from fastmcp.client.auth import OAuth + +auth = OAuth( + additional_client_metadata={"token_endpoint_auth_method": "none"}, + callback_host="127.0.0.1", +) + + +async def main() -> None: + async with Client("http://127.0.0.1:8000/mcp", auth=auth) as client: + result = await client.call_tool("echo", {"message": "hello"}) + print(result) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/auth/auth0_mcp/server.py b/examples/auth/auth0_mcp/server.py new file mode 100644 index 000000000..ac08c1c43 --- /dev/null +++ b/examples/auth/auth0_mcp/server.py @@ -0,0 +1,39 @@ +"""Auth0 Auth for MCP server example. + +Required environment variables: +- AUTH0_CONFIG_URL: OIDC discovery URL for your Auth0 tenant + +To run: + export AUTH0_CONFIG_URL="https://YOUR_TENANT.auth0.com/.well-known/openid-configuration" + python server.py +""" + +import os +import sys + +from fastmcp import FastMCP +from fastmcp.server.auth.providers.auth0 import Auth0MCPProvider + +config_url = os.getenv("AUTH0_CONFIG_URL") +if not config_url: + sys.exit( + "AUTH0_CONFIG_URL must be set to your Auth0 OIDC discovery URL, " + 'e.g. "https://YOUR_TENANT.auth0.com/.well-known/openid-configuration"' + ) + +auth = Auth0MCPProvider( + config_url=config_url, + base_url="http://127.0.0.1:8000", +) + +mcp = FastMCP("Auth0 MCP Example Server", auth=auth) + + +@mcp.tool +def echo(message: str) -> str: + """Echo the provided message.""" + return message + + +if __name__ == "__main__": + mcp.run(transport="http", port=8000) diff --git a/fastmcp_slim/fastmcp/server/auth/providers/auth0.py b/fastmcp_slim/fastmcp/server/auth/providers/auth0.py index 16ad68452..e6a939b51 100644 --- a/fastmcp_slim/fastmcp/server/auth/providers/auth0.py +++ b/fastmcp_slim/fastmcp/server/auth/providers/auth0.py @@ -1,14 +1,15 @@ -"""Auth0 OAuth provider for FastMCP. +"""Auth0 OAuth providers for FastMCP. -This module provides a complete Auth0 integration that's ready to use with -just the configuration URL, client ID, client secret, audience, and base URL. +This module provides two Auth0 integrations: -Example: +- ``Auth0Provider`` — OAuth proxy for fixed Auth0 application credentials +- ``Auth0MCPProvider`` — resource server for Auth0 Auth for MCP (DCR/CIMD) + +Example (OAuth proxy): ```python from fastmcp import FastMCP from fastmcp.server.auth.providers.auth0 import Auth0Provider - # Simple Auth0 OAuth protection auth = Auth0Provider( config_url="https://auth0.config.url", client_id="your-auth0-client-id", @@ -19,17 +20,38 @@ Example: mcp = FastMCP("My Protected Server", auth=auth) ``` + +Example (Auth for MCP): + ```python + from fastmcp import FastMCP + from fastmcp.server.auth.providers.auth0 import Auth0MCPProvider + + auth = Auth0MCPProvider( + config_url="https://your-tenant.auth0.com/.well-known/openid-configuration", + base_url="http://127.0.0.1:8000", + ) + + mcp = FastMCP("My MCP Server", auth=auth) + ``` """ -from typing import Literal +from __future__ import annotations +from typing import Any, Literal + +import httpx2 from key_value.aio.protocols import AsyncKeyValue from pydantic import AnyHttpUrl +from starlette.responses import JSONResponse +from starlette.routing import Route +from fastmcp.server.auth import RemoteAuthProvider, TokenVerifier from fastmcp.server.auth.oidc_proxy import ( DEFAULT_OIDC_DISCOVERY_TIMEOUT_SECONDS, + OIDCConfiguration, OIDCProxy, ) +from fastmcp.server.auth.providers.jwt import JWTVerifier from fastmcp.utilities.auth import parse_scopes from fastmcp.utilities.logging import get_logger @@ -159,3 +181,148 @@ class Auth0Provider(OIDCProxy): client_id, auth0_required_scopes, ) + + +class Auth0JWTVerifier(JWTVerifier): + """JWT verifier for Auth0 MCP access tokens. + + Auth0's ``rfc9068_profile_authz`` token dialect exposes API permissions in + the ``permissions`` claim. Standard OAuth ``scope``/``scp`` claims are checked + first; ``permissions`` is included when present. + """ + + def _extract_scopes(self, claims: dict[str, Any]) -> list[str]: + scopes = super()._extract_scopes(claims) + permissions = claims.get("permissions") + if isinstance(permissions, str): + return scopes + permissions.split() + if isinstance(permissions, list): + return scopes + [str(permission) for permission in permissions] + return scopes + + +class Auth0MCPProvider(RemoteAuthProvider): + """Auth0 resource server provider for Auth for MCP (DCR/CIMD). + + FastMCP validates access tokens issued by Auth0 while Auth0 handles OAuth, + dynamic client registration, and CIMD approval in the tenant dashboard. + + Enable the Resource Parameter Compatibility Profile in Auth0 and create an + API whose identifier matches this server's resource URL (logged at startup). + + Example: + ```python + from fastmcp.server.auth.providers.auth0 import Auth0MCPProvider + + auth = Auth0MCPProvider( + config_url="https://your-tenant.auth0.com/.well-known/openid-configuration", + base_url="http://127.0.0.1:8000", + ) + ``` + """ + + def __init__( + self, + *, + config_url: AnyHttpUrl | str, + base_url: AnyHttpUrl | str, + resource_base_url: AnyHttpUrl | str | None = None, + required_scopes: list[str] | None = None, + scopes_supported: list[str] | None = None, + resource_name: str | None = None, + resource_documentation: AnyHttpUrl | None = None, + token_verifier: TokenVerifier | None = None, + timeout_seconds: int | None = DEFAULT_OIDC_DISCOVERY_TIMEOUT_SECONDS, + ) -> None: + """Initialize Auth0 MCP resource server provider. + + Args: + config_url: Auth0 OIDC discovery URL + base_url: Public URL of this FastMCP server + resource_base_url: Optional public base URL for protected resource metadata + required_scopes: Scopes or permissions required on every token + scopes_supported: Scopes advertised in OAuth metadata + resource_name: Optional protected resource name + resource_documentation: Optional protected resource documentation URL + token_verifier: Optional custom verifier (skips audience auto-binding) + timeout_seconds: OIDC discovery timeout during construction + """ + oidc_config = OIDCConfiguration.get_oidc_configuration( + AnyHttpUrl(str(config_url)), + strict=None, + timeout_seconds=timeout_seconds, + ) + self.issuer = str(oidc_config.issuer).rstrip("/") + self.base_url = AnyHttpUrl(str(base_url).rstrip("/")) + + parsed_scopes = ( + parse_scopes(required_scopes) if required_scopes is not None else None + ) + + self._auto_bind_audience = token_verifier is None + if token_verifier is None: + token_verifier = Auth0JWTVerifier( + jwks_uri=str(oidc_config.jwks_uri), + issuer=str(oidc_config.issuer), + algorithm="RS256", + required_scopes=parsed_scopes, + ) + + super().__init__( + token_verifier=token_verifier, + authorization_servers=[AnyHttpUrl(self.issuer)], + base_url=self.base_url, + resource_base_url=resource_base_url, + scopes_supported=scopes_supported, + resource_name=resource_name, + resource_documentation=resource_documentation, + ) + + def set_mcp_path(self, mcp_path: str | None) -> None: + """Bind the default verifier's audience to this server's resource URL.""" + super().set_mcp_path(mcp_path) + if ( + self._auto_bind_audience + and self._resource_url is not None + and isinstance(self.token_verifier, JWTVerifier) + ): + resource_url = str(self._resource_url) + self.token_verifier.audience = resource_url + logger.info( + "Auth0 tokens will be validated against aud=%s. " + "Set your Auth0 API identifier to this URL and enable the " + "Resource Parameter Compatibility Profile.", + resource_url, + ) + + def get_routes( + self, + mcp_path: str | None = None, + ) -> list[Route]: + """Protected resource routes plus Auth0 authorization server metadata.""" + routes = super().get_routes(mcp_path) + metadata_url = f"{self.issuer}/.well-known/oauth-authorization-server" + + async def oauth_authorization_server_metadata(request): + try: + async with httpx2.AsyncClient() as client: + response = await client.get(metadata_url) + response.raise_for_status() + return JSONResponse(response.json()) + except Exception as e: + return JSONResponse( + { + "error": "server_error", + "error_description": f"Failed to fetch Auth0 metadata: {e}", + }, + status_code=500, + ) + + routes.append( + Route( + "/.well-known/oauth-authorization-server", + endpoint=oauth_authorization_server_metadata, + methods=["GET"], + ) + ) + return routes diff --git a/tests/server/auth/providers/test_auth0_mcp.py b/tests/server/auth/providers/test_auth0_mcp.py new file mode 100644 index 000000000..01aee0e8d --- /dev/null +++ b/tests/server/auth/providers/test_auth0_mcp.py @@ -0,0 +1,392 @@ +"""Tests for Auth0 MCP resource server provider.""" + +from unittest.mock import patch + +import httpx2 +import pytest + +from fastmcp import FastMCP +from fastmcp.server.auth.oidc_proxy import OIDCConfiguration +from fastmcp.server.auth.providers.auth0 import Auth0JWTVerifier, Auth0MCPProvider +from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair + +TEST_CONFIG_URL = "https://example.us.auth0.com/.well-known/openid-configuration" +TEST_BASE_URL = "http://127.0.0.1:8000" +TEST_ISSUER = "https://example.us.auth0.com/" +TEST_JWKS_URI = "https://example.us.auth0.com/.well-known/jwks.json" + + +@pytest.fixture +def valid_oidc_configuration_dict(): + return { + "issuer": TEST_ISSUER, + "authorization_endpoint": "https://example.us.auth0.com/authorize", + "token_endpoint": "https://example.us.auth0.com/oauth/token", + "jwks_uri": TEST_JWKS_URI, + "registration_endpoint": "https://example.us.auth0.com/oidc/register", + "response_types_supported": ["code"], + "subject_types_supported": ["public"], + "id_token_signing_alg_values_supported": ["RS256"], + } + + +class TestAuth0JWTVerifier: + def test_extract_scopes_includes_permissions(self): + verifier = Auth0JWTVerifier( + jwks_uri=TEST_JWKS_URI, + issuer=TEST_ISSUER, + ) + scopes = verifier._extract_scopes( + {"scope": "openid", "permissions": ["tool:whoami", "tool:greet"]} + ) + assert scopes == ["openid", "tool:whoami", "tool:greet"] + + def test_extract_scopes_permissions_string(self): + verifier = Auth0JWTVerifier( + jwks_uri=TEST_JWKS_URI, + issuer=TEST_ISSUER, + ) + scopes = verifier._extract_scopes({"permissions": "tool:whoami tool:greet"}) + assert scopes == ["tool:whoami", "tool:greet"] + + async def test_verify_token_accepts_permissions_as_required_scopes(self): + key_pair = RSAKeyPair.generate() + verifier = Auth0JWTVerifier( + public_key=key_pair.public_key, + issuer=TEST_ISSUER, + required_scopes=["tool:echo"], + ) + token = key_pair.create_token( + subject="user_123", + issuer=TEST_ISSUER, + additional_claims={"permissions": ["tool:echo"]}, + ) + + access_token = await verifier.load_access_token(token) + assert access_token is not None + assert access_token.client_id == "user_123" + + async def test_verify_token_rejects_missing_permissions(self): + key_pair = RSAKeyPair.generate() + verifier = Auth0JWTVerifier( + public_key=key_pair.public_key, + issuer=TEST_ISSUER, + required_scopes=["tool:echo"], + ) + token = key_pair.create_token( + subject="user_123", + issuer=TEST_ISSUER, + additional_claims={"permissions": ["tool:other"]}, + ) + + access_token = await verifier.load_access_token(token) + assert access_token is None + + +class TestAuth0MCPProviderInit: + def test_init_from_oidc_discovery(self, valid_oidc_configuration_dict): + with patch( + "fastmcp.server.auth.providers.auth0.OIDCConfiguration.get_oidc_configuration" + ) as mock_get: + mock_get.return_value = OIDCConfiguration.model_validate( + valid_oidc_configuration_dict + ) + + provider = Auth0MCPProvider( + config_url=TEST_CONFIG_URL, + base_url=TEST_BASE_URL, + ) + + mock_get.assert_called_once() + assert provider.issuer == "https://example.us.auth0.com" + assert str(provider.base_url) == f"{TEST_BASE_URL}/" + verifier = provider.token_verifier + assert isinstance(verifier, Auth0JWTVerifier) + assert verifier.jwks_uri == TEST_JWKS_URI + assert verifier.issuer == TEST_ISSUER + assert len(provider.authorization_servers) == 1 + assert ( + str(provider.authorization_servers[0]).rstrip("/") + == "https://example.us.auth0.com" + ) + + def test_custom_token_verifier_not_replaced(self, valid_oidc_configuration_dict): + custom = JWTVerifier( + jwks_uri=TEST_JWKS_URI, + issuer=TEST_ISSUER, + audience="https://custom.example.com/mcp", + ) + with patch( + "fastmcp.server.auth.providers.auth0.OIDCConfiguration.get_oidc_configuration" + ) as mock_get: + mock_get.return_value = OIDCConfiguration.model_validate( + valid_oidc_configuration_dict + ) + provider = Auth0MCPProvider( + config_url=TEST_CONFIG_URL, + base_url=TEST_BASE_URL, + token_verifier=custom, + ) + + assert provider.token_verifier is custom + assert provider._auto_bind_audience is False + + +class TestAuth0MCPAudienceBinding: + def test_audience_binds_on_set_mcp_path(self, valid_oidc_configuration_dict): + with patch( + "fastmcp.server.auth.providers.auth0.OIDCConfiguration.get_oidc_configuration" + ) as mock_get: + mock_get.return_value = OIDCConfiguration.model_validate( + valid_oidc_configuration_dict + ) + provider = Auth0MCPProvider( + config_url=TEST_CONFIG_URL, + base_url=TEST_BASE_URL, + ) + + verifier = provider.token_verifier + assert isinstance(verifier, Auth0JWTVerifier) + assert verifier.audience is None + + provider.set_mcp_path("/mcp") + assert verifier.audience == "http://127.0.0.1:8000/mcp" + + def test_audience_respects_resource_base_url(self, valid_oidc_configuration_dict): + with patch( + "fastmcp.server.auth.providers.auth0.OIDCConfiguration.get_oidc_configuration" + ) as mock_get: + mock_get.return_value = OIDCConfiguration.model_validate( + valid_oidc_configuration_dict + ) + provider = Auth0MCPProvider( + config_url=TEST_CONFIG_URL, + base_url="https://oauth.example.com", + resource_base_url="https://api.example.com", + ) + + provider.set_mcp_path("/mcp") + verifier = provider.token_verifier + assert isinstance(verifier, Auth0JWTVerifier) + assert verifier.audience == "https://api.example.com/mcp" + + def test_custom_verifier_audience_not_overwritten( + self, valid_oidc_configuration_dict + ): + custom_audience = "https://other.example.com" + custom = JWTVerifier( + jwks_uri=TEST_JWKS_URI, + issuer=TEST_ISSUER, + audience=custom_audience, + ) + with patch( + "fastmcp.server.auth.providers.auth0.OIDCConfiguration.get_oidc_configuration" + ) as mock_get: + mock_get.return_value = OIDCConfiguration.model_validate( + valid_oidc_configuration_dict + ) + provider = Auth0MCPProvider( + config_url=TEST_CONFIG_URL, + base_url=TEST_BASE_URL, + token_verifier=custom, + ) + provider.set_mcp_path("/mcp") + + assert custom.audience == custom_audience + + def test_set_mcp_path_none_binds_to_base_url(self, valid_oidc_configuration_dict): + with patch( + "fastmcp.server.auth.providers.auth0.OIDCConfiguration.get_oidc_configuration" + ) as mock_get: + mock_get.return_value = OIDCConfiguration.model_validate( + valid_oidc_configuration_dict + ) + provider = Auth0MCPProvider( + config_url=TEST_CONFIG_URL, + base_url=TEST_BASE_URL, + ) + + provider.set_mcp_path(None) + + verifier = provider.token_verifier + assert isinstance(verifier, Auth0JWTVerifier) + assert verifier.audience == "http://127.0.0.1:8000/" + + def test_audience_binds_through_http_app(self, valid_oidc_configuration_dict): + with patch( + "fastmcp.server.auth.providers.auth0.OIDCConfiguration.get_oidc_configuration" + ) as mock_get: + mock_get.return_value = OIDCConfiguration.model_validate( + valid_oidc_configuration_dict + ) + auth = Auth0MCPProvider( + config_url=TEST_CONFIG_URL, + base_url=TEST_BASE_URL, + ) + mcp = FastMCP("test", auth=auth) + mcp.http_app(path="/mcp") + + verifier = auth.token_verifier + assert isinstance(verifier, Auth0JWTVerifier) + assert verifier.audience == "http://127.0.0.1:8000/mcp" + + +class TestAuth0MCPMetadataForwarding: + async def test_forwards_authorization_server_metadata( + self, valid_oidc_configuration_dict, monkeypatch + ): + metadata_payload = { + "issuer": TEST_ISSUER, + "authorization_endpoint": "https://example.us.auth0.com/authorize", + "token_endpoint": "https://example.us.auth0.com/oauth/token", + "registration_endpoint": "https://example.us.auth0.com/oidc/register", + } + + class DummyResponse: + def __init__(self, payload): + self._payload = payload + + def raise_for_status(self): + return None + + def json(self): + return self._payload + + class DummyAsyncClient: + last_url: str | None = None + + def __init__(self, *args, **kwargs): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return None + + async def get(self, url): + DummyAsyncClient.last_url = url + return DummyResponse(metadata_payload) + + real_httpx_client = httpx2.AsyncClient + + monkeypatch.setattr( + "fastmcp.server.auth.providers.auth0.httpx2.AsyncClient", + DummyAsyncClient, + ) + + with patch( + "fastmcp.server.auth.providers.auth0.OIDCConfiguration.get_oidc_configuration" + ) as mock_get: + mock_get.return_value = OIDCConfiguration.model_validate( + valid_oidc_configuration_dict + ) + provider = Auth0MCPProvider( + config_url=TEST_CONFIG_URL, + base_url=TEST_BASE_URL, + ) + + mcp = FastMCP("test", auth=provider) + app = mcp.http_app() + + async with real_httpx_client( + transport=httpx2.ASGITransport(app=app), + base_url=TEST_BASE_URL, + ) as client: + response = await client.get("/.well-known/oauth-authorization-server") + + assert response.status_code == 200 + assert response.json() == metadata_payload + assert ( + DummyAsyncClient.last_url + == "https://example.us.auth0.com/.well-known/oauth-authorization-server" + ) + + +class TestAuth0MCPIntegration: + async def test_unauthenticated_mcp_request_returns_401( + self, valid_oidc_configuration_dict + ): + with patch( + "fastmcp.server.auth.providers.auth0.OIDCConfiguration.get_oidc_configuration" + ) as mock_get: + mock_get.return_value = OIDCConfiguration.model_validate( + valid_oidc_configuration_dict + ) + provider = Auth0MCPProvider( + config_url=TEST_CONFIG_URL, + base_url=TEST_BASE_URL, + ) + + mcp = FastMCP("test-server", auth=provider) + + @mcp.tool + def echo(message: str) -> str: + return message + + app = mcp.http_app() + + async with httpx2.AsyncClient( + transport=httpx2.ASGITransport(app=app), + base_url=TEST_BASE_URL, + ) as client: + response = await client.post( + "/mcp", + json={"jsonrpc": "2.0", "method": "tools/list", "id": 1}, + headers={"Content-Type": "application/json"}, + ) + + assert response.status_code == 401 + + async def test_no_register_proxy_route(self, valid_oidc_configuration_dict): + with patch( + "fastmcp.server.auth.providers.auth0.OIDCConfiguration.get_oidc_configuration" + ) as mock_get: + mock_get.return_value = OIDCConfiguration.model_validate( + valid_oidc_configuration_dict + ) + provider = Auth0MCPProvider( + config_url=TEST_CONFIG_URL, + base_url=TEST_BASE_URL, + ) + + mcp = FastMCP("test-server", auth=provider) + app = mcp.http_app() + + async with httpx2.AsyncClient( + transport=httpx2.ASGITransport(app=app), + base_url=TEST_BASE_URL, + ) as client: + response = await client.post( + "/register", + json={"client_name": "Test", "redirect_uris": ["http://localhost/cb"]}, + headers={"Content-Type": "application/json"}, + ) + + assert response.status_code == 404 + + async def test_protected_resource_metadata(self, valid_oidc_configuration_dict): + with patch( + "fastmcp.server.auth.providers.auth0.OIDCConfiguration.get_oidc_configuration" + ) as mock_get: + mock_get.return_value = OIDCConfiguration.model_validate( + valid_oidc_configuration_dict + ) + provider = Auth0MCPProvider( + config_url=TEST_CONFIG_URL, + base_url=TEST_BASE_URL, + ) + + mcp = FastMCP("test-server", auth=provider) + app = mcp.http_app() + + async with httpx2.AsyncClient( + transport=httpx2.ASGITransport(app=app), + base_url=TEST_BASE_URL, + ) as client: + response = await client.get("/.well-known/oauth-protected-resource/mcp") + + assert response.status_code == 200 + data = response.json() + assert data["resource"] == f"{TEST_BASE_URL}/mcp" + assert data["authorization_servers"] == [TEST_ISSUER]