diff --git a/docs/docs.json b/docs/docs.json index 29d44c020..345050c8f 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -277,6 +277,7 @@ "integrations/github", "integrations/google", "integrations/oci", + "integrations/propelauth", "integrations/scalekit", "integrations/supabase", "integrations/workos", diff --git a/docs/integrations/propelauth.mdx b/docs/integrations/propelauth.mdx new file mode 100644 index 000000000..9b2838c3d --- /dev/null +++ b/docs/integrations/propelauth.mdx @@ -0,0 +1,164 @@ +--- +title: PropelAuth 🤝 FastMCP +sidebarTitle: PropelAuth +description: Secure your FastMCP server with PropelAuth +icon: shield-check +--- + +import { VersionBadge } from "/snippets/version-badge.mdx"; + + + +This guide shows you how to secure your FastMCP server using [**PropelAuth**](https://www.propelauth.com), a complete authentication and user management solution. This integration uses the [**Remote OAuth**](/servers/auth/remote-oauth) pattern, where PropelAuth handles user login, consent management, and your FastMCP server validates the tokens. + +## Configuration + +### Prerequisites + +Before you begin, you will need: + +1. A [PropelAuth](https://www.propelauth.com) account +2. Your FastMCP server's base URL (can be localhost for development, e.g., `http://localhost:8000`) + +### Step 1: Configure PropelAuth + + + + Navigate to the **MCP** section in your PropelAuth dashboard, click **Enable MCP**, and choose which environments to enable it for (Test, Staging, Prod). + + + + Under **MCP > Allowed MCP Clients**, add redirect URIs for each MCP client you want to allow. PropelAuth provides templates for popular clients like Claude, Cursor, and ChatGPT. + + + + Under **MCP > Scopes**, define the permissions available to MCP clients (e.g., `read:user_data`). + + + + Under **MCP > Settings > How Do Users Create OAuth Clients?**, you can optionally enable: + - **Dynamic Client Registration** — clients self-register automatically via the DCR protocol + - **Manually via Hosted Pages** — PropelAuth creates a UI for your users to register OAuth clients + + You can enable neither, one, or both. If you enable neither, you'll manage OAuth client creation yourself. + + + + Go to **MCP > Request Validation** and click **Create Credentials**. Note the **Client ID** and **Client Secret** - you'll need these to validate tokens. + + + + Find your Auth URL in the **Backend Integration** section of the dashboard (e.g., `https://auth.yourdomain.com`). + + + +For more details, see the [PropelAuth MCP documentation](https://docs.propelauth.com/mcp-authentication/overview). + +### Step 2: Environment Setup + +Create a `.env` file with your PropelAuth configuration: + +```bash +PROPELAUTH_AUTH_URL=https://auth.yourdomain.com # From Backend Integration page +PROPELAUTH_INTROSPECTION_CLIENT_ID=your-client-id # From MCP > Request Validation +PROPELAUTH_INTROSPECTION_CLIENT_SECRET=your-client-secret # From MCP > Request Validation +SERVER_URL=http://localhost:8000 # Your server's base URL +``` + +### Step 3: FastMCP Configuration + +Create your FastMCP server file and use the PropelAuthProvider to handle all the OAuth integration automatically: + +```python server.py +import os +from fastmcp import FastMCP +from fastmcp.server.auth.providers.propelauth import PropelAuthProvider + +auth_provider = PropelAuthProvider( + auth_url=os.environ["PROPELAUTH_AUTH_URL"], + introspection_client_id=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_ID"], + introspection_client_secret=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_SECRET"], + base_url=os.environ["SERVER_URL"], + required_scopes=["read:user_data"], # Optional scope enforcement +) + +mcp = FastMCP(name="My PropelAuth Protected Server", auth=auth_provider) +``` + +## Testing + +With your `.env` loaded, start the server: + +```bash +fastmcp run server.py --transport http --port 8000 +``` + +Then use a FastMCP client to verify authentication works: + +```python +from fastmcp import Client +import asyncio + +async def main(): + async with Client("http://localhost:8000/mcp", auth="oauth") as client: + assert await client.ping() + +if __name__ == "__main__": + asyncio.run(main()) +``` + +## Accessing User Information + +You can use `get_access_token()` inside your tools to identify the authenticated user: + +```python server.py +import os +from fastmcp import FastMCP +from fastmcp.server.auth.providers.propelauth import PropelAuthProvider +from fastmcp.server.dependencies import get_access_token + +auth = PropelAuthProvider( + auth_url=os.environ["PROPELAUTH_AUTH_URL"], + introspection_client_id=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_ID"], + introspection_client_secret=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_SECRET"], + base_url=os.environ["SERVER_URL"], + required_scopes=["read:user_data"], +) + +mcp = FastMCP(name="My PropelAuth Protected Server", auth=auth) + +@mcp.tool +def whoami() -> dict: + """Return the authenticated user's ID.""" + token = get_access_token() + if token is None: + return {"error": "Not authenticated"} + user_id = token.claims.get("sub") + return {"user_id": user_id} +``` + +## Advanced Configuration + +The `PropelAuthProvider` supports optional overrides for token introspection behavior, including caching and request timeouts: + +```python server.py +import os +from fastmcp import FastMCP +from fastmcp.server.auth.providers.propelauth import PropelAuthProvider + +auth = PropelAuthProvider( + auth_url=os.environ["PROPELAUTH_AUTH_URL"], + introspection_client_id=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_ID"], + introspection_client_secret=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_SECRET"], + base_url=os.environ.get("BASE_URL", "https://your-server.com"), + required_scopes=["read:user_data"], + resource="https://your-server.com/mcp", # Restrict to tokens intended for this server (RFC 8707) + token_introspection_overrides={ + "cache_ttl_seconds": 300, # Cache introspection results for 5 minutes + "max_cache_size": 1000, # Maximum cached tokens + "timeout_seconds": 15, # HTTP request timeout + }, +) + +mcp = FastMCP(name="My PropelAuth Protected Server", auth=auth) +``` diff --git a/examples/auth/propelauth_oauth/README.md b/examples/auth/propelauth_oauth/README.md new file mode 100644 index 000000000..575ea7314 --- /dev/null +++ b/examples/auth/propelauth_oauth/README.md @@ -0,0 +1,67 @@ +# PropelAuth OAuth Example + +Demonstrates FastMCP server protection with PropelAuth OAuth. + +## Setup + +### 1. Configure MCP Authentication in PropelAuth + +**Create a PropelAuth Account**: + +- Go to [PropelAuth Dashboard](https://www.propelauth.com) +- Navigate to the **MCP** section and click **Enable MCP** + +**Configure Allowed MCP Clients**: + +- Under **MCP > Allowed MCP Clients**, add redirect URIs for each MCP client you want to allow +- PropelAuth provides templates for popular clients like Claude, Cursor, and ChatGPT + +**Configure Scopes**: + +- Under **MCP > Scopes**, define the permissions available to MCP clients (e.g., `read:user_data`) + +**Generate Introspection Credentials**: + +- Go to **MCP > Request Validation** and click **Create Credentials** +- Note the **Client ID** and **Client Secret** + +**Note Your Auth URL**: + +- Find your Auth URL in the **Backend Integration** section (e.g., `https://auth.yourdomain.com`) + +Create a `.env` file: + +```bash +# Required PropelAuth credentials +PROPELAUTH_AUTH_URL=https://auth.yourdomain.com +PROPELAUTH_INTROSPECTION_CLIENT_ID=your-client-id +PROPELAUTH_INTROSPECTION_CLIENT_SECRET=your-client-secret +BASE_URL=http://localhost:8000/ +# Optional: additional scopes tokens must include (comma-separated) +# PROPELAUTH_REQUIRED_SCOPES=read:user_data +``` + +### 2. Run the Example + +Start the server: + +```bash +# From this directory +uv run python server.py +``` + +The server will start on `http://localhost:8000/mcp` with PropelAuth OAuth authentication enabled. + +Test with client: + +```bash +uv run python client.py +``` + +The `client.py` will: + +1. Attempt to connect to the server +2. Detect that OAuth authentication is required +3. Open a browser for PropelAuth authentication +4. Complete the OAuth flow and connect to the server +5. Demonstrate calling authenticated tools (echo and whoami) diff --git a/examples/auth/propelauth_oauth/client.py b/examples/auth/propelauth_oauth/client.py new file mode 100644 index 000000000..fd6a84b4e --- /dev/null +++ b/examples/auth/propelauth_oauth/client.py @@ -0,0 +1,43 @@ +"""OAuth client example for connecting to PropelAuth-protected FastMCP servers. + +This example demonstrates how to connect to a PropelAuth OAuth-protected FastMCP server. + +To run: + python client.py +""" + +import asyncio + +from fastmcp.client import Client + +SERVER_URL = "http://127.0.0.1:8000/mcp" + + +async def main(): + try: + async with Client(SERVER_URL, auth="oauth") as client: + assert await client.ping() + print("✅ Successfully authenticated with PropelAuth!") + + tools = await client.list_tools() + print(f"🔧 Available tools ({len(tools)}):") + for tool in tools: + print(f" - {tool.name}: {tool.description}") + + # Test calling a tool + result = await client.call_tool( + "echo", {"message": "Hello from PropelAuth!"} + ) + print(f"🎯 Echo result: {result}") + + # Test calling whoami tool + whoami = await client.call_tool("whoami", {}) + print(f"👤 Who am I: {whoami}") + + except Exception as e: + print(f"❌ Authentication failed: {e}") + raise + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/auth/propelauth_oauth/server.py b/examples/auth/propelauth_oauth/server.py new file mode 100644 index 000000000..8401882aa --- /dev/null +++ b/examples/auth/propelauth_oauth/server.py @@ -0,0 +1,54 @@ +"""PropelAuth OAuth server example for FastMCP. + +This example demonstrates how to protect a FastMCP server with PropelAuth OAuth. + +Required environment variables: +- PROPELAUTH_AUTH_URL: Your PropelAuth Auth URL (from Backend Integration page) +- PROPELAUTH_INTROSPECTION_CLIENT_ID: Introspection Client ID (from MCP > Request Validation) +- PROPELAUTH_INTROSPECTION_CLIENT_SECRET: Introspection Client Secret (from MCP > Request Validation) + +Optional: +- PROPELAUTH_REQUIRED_SCOPES: Comma-separated scopes tokens must include +- BASE_URL: Public URL where the FastMCP server is exposed (defaults to `http://localhost:8000/`) + +To run: + python server.py +""" + +import os + +from dotenv import load_dotenv + +from fastmcp import FastMCP +from fastmcp.server.auth.providers.propelauth import PropelAuthProvider +from fastmcp.server.dependencies import get_access_token + +load_dotenv() + +auth = PropelAuthProvider( + auth_url=os.environ["PROPELAUTH_AUTH_URL"], + introspection_client_id=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_ID"], + introspection_client_secret=os.environ["PROPELAUTH_INTROSPECTION_CLIENT_SECRET"], + base_url=os.getenv("BASE_URL", "http://localhost:8000/"), +) + +mcp = FastMCP("PropelAuth OAuth Example Server", auth=auth) + + +@mcp.tool +def echo(message: str) -> str: + """Echo the provided message.""" + return message + + +@mcp.tool +def whoami() -> dict: + """Return the authenticated user's ID.""" + token = get_access_token() + if token is None: + return {"error": "Not authenticated"} + return {"user_id": token.claims.get("sub")} + + +if __name__ == "__main__": + mcp.run(transport="http", port=8000) diff --git a/src/fastmcp/server/auth/providers/propelauth.py b/src/fastmcp/server/auth/providers/propelauth.py new file mode 100644 index 000000000..0183a4c31 --- /dev/null +++ b/src/fastmcp/server/auth/providers/propelauth.py @@ -0,0 +1,223 @@ +"""PropelAuth authentication provider for FastMCP. + +Example: + ```python + from fastmcp import FastMCP + from fastmcp.server.auth.providers.propelauth import PropelAuthProvider + + auth = PropelAuthProvider( + auth_url="https://auth.yourdomain.com", + introspection_client_id="your-client-id", + introspection_client_secret="your-client-secret", + base_url="https://your-fastmcp-server.com", + required_scopes=["read:user_data"], + ) + + mcp = FastMCP("My App", auth=auth) + ``` +""" + +from __future__ import annotations + +from typing import TypedDict + +import httpx +from pydantic import AnyHttpUrl, SecretStr +from starlette.responses import JSONResponse +from starlette.routing import Route + +from fastmcp.server.auth import AccessToken, RemoteAuthProvider +from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier +from fastmcp.utilities.logging import get_logger + +logger = get_logger(__name__) + + +class PropelAuthTokenIntrospectionOverrides(TypedDict, total=False): + timeout_seconds: int + cache_ttl_seconds: int | None + max_cache_size: int | None + http_client: httpx.AsyncClient | None + + +class PropelAuthProvider(RemoteAuthProvider): + """PropelAuth resource server provider using OAuth 2.1 token introspection. + + This provider validates access tokens via PropelAuth's introspection endpoint + and forwards authorization server metadata for OAuth discovery. + + Setup: + 1. Enable MCP authentication in the PropelAuth Dashboard + 2. Configure scopes on the MCP page + 3. Select which redirect URIs to enable by picking which clients you support + 4. Generate introspection credentials (Client ID + Client Secret) + + For detailed setup instructions, see: + https://docs.propelauth.com/mcp-authentication/overview + + Example: + ```python + from fastmcp import FastMCP + from fastmcp.server.auth.providers.propelauth import PropelAuthProvider + + auth = PropelAuthProvider( + auth_url="https://auth.yourdomain.com", + introspection_client_id="your-client-id", + introspection_client_secret="your-client-secret", + base_url="https://your-fastmcp-server.com", + required_scopes=["read:user_data"], + ) + + mcp = FastMCP("My App", auth=auth) + ``` + """ + + def __init__( + self, + *, + auth_url: AnyHttpUrl | str, + introspection_client_id: str, + introspection_client_secret: str | SecretStr, + base_url: AnyHttpUrl | str, + required_scopes: list[str] | None = None, + resource: AnyHttpUrl | str | None = None, + token_introspection_overrides: ( + PropelAuthTokenIntrospectionOverrides | None + ) = None, + ): + """Initialize PropelAuth provider. + + Args: + auth_url: Your PropelAuth Auth URL (from the Backend Integration page) + introspection_client_id: Introspection Client ID from the PropelAuth Dashboard + introspection_client_secret: Introspection Client Secret from the PropelAuth Dashboard + base_url: Public URL of this FastMCP server + required_scopes: Optional list of scopes that must be present in tokens + resource: Optional resource URI (RFC 8707) identifying this MCP server. + Use this when multiple MCP servers share the same PropelAuth + authorization server (e.g. ``resource="https://api.example.com/mcp"``), + so only tokens intended for this MCP server are accepted. + token_introspection_overrides: Optional overrides for the underlying + IntrospectionTokenVerifier (timeout, caching, http_client) + """ + normalized_auth_url = str(auth_url).rstrip("/") + introspection_url = f"{normalized_auth_url}/oauth/2.1/introspect" + authorization_server_url = AnyHttpUrl(f"{normalized_auth_url}/oauth/2.1") + + if resource is None: + self._resource = None + logger.debug( + "PropelAuthProvider: no resource configured, audience checking disabled" + ) + else: + self._resource = str(resource) + + token_verifier = self._create_token_verifier( + introspection_url=introspection_url, + client_id=introspection_client_id, + client_secret=introspection_client_secret, + required_scopes=required_scopes, + introspection_overrides=token_introspection_overrides, + ) + + self._normalized_auth_url = normalized_auth_url + super().__init__( + token_verifier=token_verifier, + authorization_servers=[authorization_server_url], + base_url=base_url, + ) + + def get_routes( + self, + mcp_path: str | None = None, + ) -> list[Route]: + """Get routes for this provider. + + Includes the standard routes from the RemoteAuthProvider (protected resource metadata routes (RFC 9728)), + and creates an authorization server metadata route that forwards to PropelAuth's route + + Args: + mcp_path: The path where the MCP endpoint is mounted (e.g., "/mcp") + This is used to advertise the resource URL in metadata. + """ + routes = super().get_routes(mcp_path) + + async def oauth_authorization_server_metadata(request): + """Forward PropelAuth OAuth authorization server metadata""" + try: + async with httpx.AsyncClient() as client: + response = await client.get( + f"{self._normalized_auth_url}/.well-known/oauth-authorization-server/oauth/2.1" + ) + response.raise_for_status() + metadata = response.json() + return JSONResponse(metadata) + except Exception as e: + return JSONResponse( + { + "error": "server_error", + "error_description": f"Failed to fetch PropelAuth metadata: {e}", + }, + status_code=500, + ) + + routes.append( + Route( + "/.well-known/oauth-authorization-server", + endpoint=oauth_authorization_server_metadata, + methods=["GET"], + ) + ) + + return routes + + async def verify_token(self, token: str) -> AccessToken | None: + """Verify token and check the ``aud`` claim against the configured resource.""" + result = await super().verify_token(token) + if result is None or self._resource is None: + return result + + aud = result.claims.get("aud") + if aud != self._resource: + logger.debug( + "PropelAuthProvider: token audience %r does not match resource %s", + aud, + self._resource, + ) + return None + + return result + + def _create_token_verifier( + self, + introspection_url: str, + client_id: str, + client_secret: str | SecretStr, + required_scopes: list[str] | None, + introspection_overrides: PropelAuthTokenIntrospectionOverrides | None, + ) -> IntrospectionTokenVerifier: + # Being defensive here, check for only the fields we are expecting + safe_overrides: PropelAuthTokenIntrospectionOverrides = {} + if introspection_overrides is not None: + if "timeout_seconds" in introspection_overrides: + safe_overrides["timeout_seconds"] = introspection_overrides[ + "timeout_seconds" + ] + if "cache_ttl_seconds" in introspection_overrides: + safe_overrides["cache_ttl_seconds"] = introspection_overrides[ + "cache_ttl_seconds" + ] + if "max_cache_size" in introspection_overrides: + safe_overrides["max_cache_size"] = introspection_overrides[ + "max_cache_size" + ] + if "http_client" in introspection_overrides: + safe_overrides["http_client"] = introspection_overrides["http_client"] + + return IntrospectionTokenVerifier( + introspection_url=introspection_url, + client_id=client_id, + client_secret=client_secret, + required_scopes=required_scopes, + **safe_overrides, + ) diff --git a/tests/server/auth/providers/test_propelauth.py b/tests/server/auth/providers/test_propelauth.py new file mode 100644 index 000000000..9d70496fc --- /dev/null +++ b/tests/server/auth/providers/test_propelauth.py @@ -0,0 +1,336 @@ +"""Tests for PropelAuthProvider.""" + +from typing import cast +from unittest.mock import AsyncMock + +import httpx +import pytest +from pydantic import SecretStr + +from fastmcp import Client, FastMCP +from fastmcp.server.auth import AccessToken +from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier +from fastmcp.server.auth.providers.propelauth import ( + PropelAuthProvider, + PropelAuthTokenIntrospectionOverrides, +) +from fastmcp.utilities.tests import run_server_async + + +class TestPropelAuthProvider: + """Test PropelAuth's auth provider.""" + + def test_init_with_only_required_params(self): + """Test PropelAuthProvider initialization with only required params.""" + provider = PropelAuthProvider( + auth_url="https://auth.example.com", + introspection_client_id="client_id_123", + introspection_client_secret="client_secret_123", + base_url="https://example.com", + ) + + # Verify the provider is configured correctly + assert len(provider.authorization_servers) == 1 + assert ( + str(provider.authorization_servers[0]) + == "https://auth.example.com/oauth/2.1" + ) + assert str(provider.base_url) == "https://example.com/" + + # Verify token verifier is configured correctly + assert isinstance(provider.token_verifier, IntrospectionTokenVerifier) + assert ( + provider.token_verifier.introspection_url + == "https://auth.example.com/oauth/2.1/introspect" + ) + assert provider.token_verifier.client_id == "client_id_123" + assert provider.token_verifier.client_secret == "client_secret_123" + + def test_auth_url_trailing_slash_normalization(self): + """Test that trailing slash on auth_url is stripped before building URLs.""" + provider = PropelAuthProvider( + auth_url="https://auth.example.com/", + introspection_client_id="client_id_123", + introspection_client_secret="client_secret_123", + base_url="https://example.com", + ) + + assert isinstance(provider.token_verifier, IntrospectionTokenVerifier) + assert len(provider.authorization_servers) == 1 + assert ( + str(provider.authorization_servers[0]) + == "https://auth.example.com/oauth/2.1" + ) + assert ( + provider.token_verifier.introspection_url + == "https://auth.example.com/oauth/2.1/introspect" + ) + + def test_required_scopes_passed_to_verifier(self): + """Test that required_scopes are passed through to the token verifier.""" + provider = PropelAuthProvider( + auth_url="https://auth.example.com", + introspection_client_id="client_id_123", + introspection_client_secret="client_secret_123", + base_url="https://example.com", + required_scopes=["read", "write"], + ) + + assert isinstance(provider.token_verifier, IntrospectionTokenVerifier) + assert provider.token_verifier.required_scopes == ["read", "write"] + + def test_introspection_client_secret_as_secret_str(self): + """Test that SecretStr client_secret is unwrapped correctly.""" + provider = PropelAuthProvider( + auth_url="https://auth.example.com", + introspection_client_id="client_id_123", + introspection_client_secret=SecretStr("my_secret"), + base_url="https://example.com", + ) + + assert isinstance(provider.token_verifier, IntrospectionTokenVerifier) + assert provider.token_verifier.client_secret == "my_secret" + + def test_authorization_servers_configuration(self): + """Test that authorization_servers contains the correct PropelAuth URL.""" + provider = PropelAuthProvider( + auth_url="https://auth.propelauth.com", + introspection_client_id="client_id_123", + introspection_client_secret="client_secret_123", + base_url="https://example.com", + ) + + assert len(provider.authorization_servers) == 1 + assert ( + str(provider.authorization_servers[0]) + == "https://auth.propelauth.com/oauth/2.1" + ) + + def test_token_introspection_overrides_timeout(self): + """Test that timeout_seconds override is passed to the verifier.""" + provider = PropelAuthProvider( + auth_url="https://auth.example.com", + introspection_client_id="client_id_123", + introspection_client_secret="client_secret_123", + base_url="https://example.com", + token_introspection_overrides={"timeout_seconds": 30}, + ) + + assert isinstance(provider.token_verifier, IntrospectionTokenVerifier) + assert provider.token_verifier.timeout_seconds == 30 + + def test_token_introspection_overrides_cache(self): + """Test that cache overrides are passed to the verifier.""" + provider = PropelAuthProvider( + auth_url="https://auth.example.com", + introspection_client_id="client_id_123", + introspection_client_secret="client_secret_123", + base_url="https://example.com", + token_introspection_overrides={ + "cache_ttl_seconds": 300, + "max_cache_size": 500, + }, + ) + + assert isinstance(provider.token_verifier, IntrospectionTokenVerifier) + assert provider.token_verifier._cache_ttl == 300 + assert provider.token_verifier._max_cache_size == 500 + + def test_token_introspection_overrides_http_client(self): + """Test that http_client override is passed to the verifier.""" + client = httpx.AsyncClient() + provider = PropelAuthProvider( + auth_url="https://auth.example.com", + introspection_client_id="client_id_123", + introspection_client_secret="client_secret_123", + base_url="https://example.com", + token_introspection_overrides={"http_client": client}, + ) + + assert isinstance(provider.token_verifier, IntrospectionTokenVerifier) + assert provider.token_verifier._http_client is client + + def test_token_introspection_overrides_ignores_unknown_keys(self): + """Test that unknown override keys are silently ignored.""" + provider = PropelAuthProvider( + auth_url="https://auth.example.com", + introspection_client_id="client_id_123", + introspection_client_secret="client_secret_123", + base_url="https://example.com", + # This won't typecheck without casting, since it shouldn't be allowed + token_introspection_overrides=cast( + PropelAuthTokenIntrospectionOverrides, {"unknown_key": "value"} + ), + ) + + assert isinstance(provider.token_verifier, IntrospectionTokenVerifier) + assert provider.token_verifier.timeout_seconds == 10 + + def test_token_introspection_overrides_ignores_disallowed_known_keys(self): + """Test that known IntrospectionTokenVerifier keys not in the allow list are ignored.""" + provider = PropelAuthProvider( + auth_url="https://auth.example.com", + introspection_client_id="client_id_123", + introspection_client_secret="client_secret_123", + base_url="https://example.com", + # This won't typecheck without casting, since it shouldn't be allowed + token_introspection_overrides=cast( + PropelAuthTokenIntrospectionOverrides, {"client_id": "sneaky_override"} + ), + ) + + assert isinstance(provider.token_verifier, IntrospectionTokenVerifier) + assert provider.token_verifier.client_id == "client_id_123" + + +class TestPropelAuthResourceChecking: + """Test audience (aud) checking when resource is configured.""" + + def _make_provider(self, resource: str | None = None) -> PropelAuthProvider: + return PropelAuthProvider( + auth_url="https://auth.example.com", + introspection_client_id="client_id_123", + introspection_client_secret="client_secret_123", + base_url="https://example.com", + resource=resource, + ) + + def _make_access_token(self, aud: str) -> AccessToken: + return AccessToken( + token="test-token", + client_id="client_id_123", + scopes=[], + claims={"active": True, "sub": "user-1", "aud": aud}, + ) + + async def test_no_resource_skips_aud_check(self, monkeypatch: pytest.MonkeyPatch): + """When resource is not configured, tokens are accepted without aud checking.""" + provider = self._make_provider(resource=None) + token = self._make_access_token(aud="https://anything.example.com") + monkeypatch.setattr( + provider.token_verifier, "verify_token", AsyncMock(return_value=token) + ) + + result = await provider.verify_token("test-token") + assert result is token + + async def test_aud_matches_resource(self, monkeypatch: pytest.MonkeyPatch): + """Token is accepted when aud matches the configured resource.""" + provider = self._make_provider(resource="https://api.example.com/mcp") + token = self._make_access_token(aud="https://api.example.com/mcp") + monkeypatch.setattr( + provider.token_verifier, "verify_token", AsyncMock(return_value=token) + ) + + result = await provider.verify_token("test-token") + assert result is token + + async def test_aud_does_not_match_resource(self, monkeypatch: pytest.MonkeyPatch): + """Token is rejected when aud doesn't match the configured resource.""" + provider = self._make_provider(resource="https://api.example.com/mcp") + token = self._make_access_token(aud="https://other-server.example.com/mcp") + monkeypatch.setattr( + provider.token_verifier, "verify_token", AsyncMock(return_value=token) + ) + + result = await provider.verify_token("test-token") + assert result is None + + async def test_inner_verifier_returns_none(self, monkeypatch: pytest.MonkeyPatch): + """When the inner verifier rejects the token, None is returned without aud checking.""" + provider = self._make_provider(resource="https://api.example.com/mcp") + monkeypatch.setattr( + provider.token_verifier, "verify_token", AsyncMock(return_value=None) + ) + + result = await provider.verify_token("test-token") + assert result is None + + +@pytest.fixture +async def mcp_server_url(): + """Start MCP server with PropelAuth authentication.""" + mcp = FastMCP( + auth=PropelAuthProvider( + auth_url="https://auth.example.com", + introspection_client_id="client_id_123", + introspection_client_secret="client_secret_123", + base_url="http://localhost:4321", + ) + ) + + @mcp.tool + def add(a: int, b: int) -> int: + return a + b + + async with run_server_async(mcp, transport="http") as url: + yield url + + +class TestPropelAuthProviderIntegration: + async def test_unauthorized_access(self, mcp_server_url: str): + with pytest.raises(httpx.HTTPStatusError) as exc_info: + async with Client(mcp_server_url) as client: + tools = await client.list_tools() # noqa: F841 + + assert isinstance(exc_info.value, httpx.HTTPStatusError) + assert exc_info.value.response.status_code == 401 + assert "tools" not in locals() + + async def test_metadata_route_forwards_propelauth_response( + self, + monkeypatch: pytest.MonkeyPatch, + mcp_server_url: str, + ) -> None: + """Ensure PropelAuth metadata route proxies upstream JSON.""" + + metadata_payload = { + "issuer": "https://auth.example.com", + "token_endpoint": "https://auth.example.com/oauth/2.1/token", + "authorization_endpoint": "https://auth.example.com/oauth/2.1/authorize", + } + + class DummyResponse: + status_code = 200 + + def __init__(self, data: dict[str, str]): + self._data = data + + def json(self): + return self._data + + def raise_for_status(self): + return None + + class DummyAsyncClient: + last_url: str | None = None + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def get(self, url: str): + DummyAsyncClient.last_url = url + return DummyResponse(metadata_payload) + + real_httpx_client = httpx.AsyncClient + + monkeypatch.setattr( + "fastmcp.server.auth.providers.propelauth.httpx.AsyncClient", + DummyAsyncClient, + ) + + base_url = mcp_server_url.rsplit("/mcp", 1)[0] + async with real_httpx_client() as client: + response = await client.get( + f"{base_url}/.well-known/oauth-authorization-server" + ) + + assert response.status_code == 200 + assert response.json() == metadata_payload + assert ( + DummyAsyncClient.last_url + == "https://auth.example.com/.well-known/oauth-authorization-server/oauth/2.1" + )