diff --git a/docs/servers/auth/token-verification.mdx b/docs/servers/auth/token-verification.mdx index 89c48b07a..890214c6d 100644 --- a/docs/servers/auth/token-verification.mdx +++ b/docs/servers/auth/token-verification.mdx @@ -171,7 +171,25 @@ verifier = IntrospectionTokenVerifier( mcp = FastMCP(name="Protected API", auth=verifier) ``` -The verifier authenticates to the introspection endpoint using HTTP Basic Auth with your client credentials. When a request arrives with a bearer token, FastMCP queries the introspection endpoint to determine if the token is active and has sufficient scopes. +The verifier authenticates to the introspection endpoint using client credentials and queries it whenever a bearer token arrives. FastMCP checks whether the token is active and has sufficient scopes before allowing access. + +Two standard client authentication methods are supported, both defined in RFC 6749: + +- **`client_secret_basic`** (default): Sends credentials via HTTP Basic Auth header +- **`client_secret_post`**: Sends credentials in the POST request body + +Most OAuth providers support both methods, though some may require one specifically. Configure the authentication method with the `client_auth_method` parameter: + +```python +# Use POST body authentication instead of Basic Auth +verifier = IntrospectionTokenVerifier( + introspection_url="https://auth.yourcompany.com/oauth/introspect", + client_id="mcp-resource-server", + client_secret="your-client-secret", + client_auth_method="client_secret_post", + required_scopes=["api:read", "api:write"] +) +``` ## Development and Testing diff --git a/src/fastmcp/server/auth/providers/introspection.py b/src/fastmcp/server/auth/providers/introspection.py index ec1ad6b95..707e25471 100644 --- a/src/fastmcp/server/auth/providers/introspection.py +++ b/src/fastmcp/server/auth/providers/introspection.py @@ -25,7 +25,7 @@ from __future__ import annotations import base64 import time -from typing import Any +from typing import Any, Literal, get_args import httpx from pydantic import AnyHttpUrl, SecretStr @@ -36,6 +36,8 @@ from fastmcp.utilities.logging import get_logger logger = get_logger(__name__) +ClientAuthMethod = Literal["client_secret_basic", "client_secret_post"] + class IntrospectionTokenVerifier(TokenVerifier): """ @@ -45,8 +47,11 @@ class IntrospectionTokenVerifier(TokenVerifier): endpoint. Unlike JWT verification which is stateless, token introspection requires a network call to the authorization server for each token validation. - The verifier authenticates to the introspection endpoint using HTTP Basic Auth - with the provided client_id and client_secret, as specified in RFC 7662. + The verifier authenticates to the introspection endpoint using either: + - HTTP Basic Auth (client_secret_basic, default): credentials in Authorization header + - POST body authentication (client_secret_post): credentials in request body + + Both methods are specified in RFC 6749 (OAuth 2.0) and RFC 7662 (Token Introspection). Use this when: - Your authorization server issues opaque (non-JWT) tokens @@ -71,6 +76,7 @@ class IntrospectionTokenVerifier(TokenVerifier): introspection_url: str, client_id: str, client_secret: str | SecretStr, + client_auth_method: ClientAuthMethod = "client_secret_basic", timeout_seconds: int = 10, required_scopes: list[str] | None = None, base_url: AnyHttpUrl | str | None = None, @@ -82,6 +88,8 @@ class IntrospectionTokenVerifier(TokenVerifier): introspection_url: URL of the OAuth 2.0 token introspection endpoint client_id: OAuth client ID for authenticating to the introspection endpoint client_secret: OAuth client secret for authenticating to the introspection endpoint + client_auth_method: Client authentication method. "client_secret_basic" (default) + uses HTTP Basic Auth header, "client_secret_post" sends credentials in POST body timeout_seconds: HTTP request timeout in seconds (default: 10) required_scopes: Required scopes for all tokens (optional) base_url: Base URL for TokenVerifier protocol @@ -100,6 +108,17 @@ class IntrospectionTokenVerifier(TokenVerifier): if isinstance(client_secret, SecretStr) else client_secret ) + + # Validate client_auth_method to catch typos/invalid values early + valid_methods = get_args(ClientAuthMethod) + if client_auth_method not in valid_methods: + options = " or ".join(f"'{m}'" for m in valid_methods) + raise ValueError( + f"Invalid client_auth_method: {client_auth_method!r}. " + f"Must be {options}." + ) + self.client_auth_method: ClientAuthMethod = client_auth_method + self.timeout_seconds = timeout_seconds self.logger = get_logger(__name__) @@ -137,7 +156,8 @@ class IntrospectionTokenVerifier(TokenVerifier): Verify a bearer token using OAuth 2.0 Token Introspection (RFC 7662). This method makes a POST request to the introspection endpoint with the token, - authenticated using HTTP Basic Auth with the client credentials. + authenticated using the configured client authentication method (client_secret_basic + or client_secret_post). Args: token: The opaque token string to validate @@ -148,19 +168,29 @@ class IntrospectionTokenVerifier(TokenVerifier): try: async with httpx.AsyncClient(timeout=self.timeout_seconds) as client: # Prepare introspection request per RFC 7662 - auth_header = self._create_basic_auth_header() + # Build request data with token and token_type_hint + data = { + "token": token, + "token_type_hint": "access_token", + } + + # Build headers + headers = { + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + } + + # Add client authentication based on method + if self.client_auth_method == "client_secret_basic": + headers["Authorization"] = self._create_basic_auth_header() + elif self.client_auth_method == "client_secret_post": + data["client_id"] = self.client_id + data["client_secret"] = self.client_secret response = await client.post( self.introspection_url, - data={ - "token": token, - "token_type_hint": "access_token", - }, - headers={ - "Authorization": auth_header, - "Content-Type": "application/x-www-form-urlencoded", - "Accept": "application/json", - }, + data=data, + headers=headers, ) # Check for HTTP errors diff --git a/tests/server/auth/providers/test_introspection.py b/tests/server/auth/providers/test_introspection.py index c0c2c9493..c589213ce 100644 --- a/tests/server/auth/providers/test_introspection.py +++ b/tests/server/auth/providers/test_introspection.py @@ -47,6 +47,7 @@ class TestIntrospectionTokenVerifier: assert verifier.client_id == "test-client" assert verifier.client_secret == "test-secret" assert verifier.timeout_seconds == 10 + assert verifier.client_auth_method == "client_secret_basic" def test_initialization_requires_introspection_url(self): """Test that introspection_url is required.""" @@ -390,6 +391,137 @@ class TestIntrospectionTokenVerifier: assert access_token is not None assert access_token.client_id == "unknown" + def test_initialization_with_client_secret_post(self): + """Test verifier initialization with client_secret_post method.""" + verifier = IntrospectionTokenVerifier( + introspection_url="https://auth.example.com/oauth/introspect", + client_id="test-client", + client_secret="test-secret", + client_auth_method="client_secret_post", + ) + + assert verifier.client_auth_method == "client_secret_post" + assert verifier.introspection_url == "https://auth.example.com/oauth/introspect" + assert verifier.client_id == "test-client" + assert verifier.client_secret == "test-secret" + + def test_initialization_defaults_to_client_secret_basic(self): + """Test that client_secret_basic is the default auth method.""" + verifier = IntrospectionTokenVerifier( + introspection_url="https://auth.example.com/oauth/introspect", + client_id="test-client", + client_secret="test-secret", + ) + + assert verifier.client_auth_method == "client_secret_basic" + + def test_initialization_rejects_invalid_client_auth_method(self): + """Test that invalid client_auth_method values are rejected.""" + # Test typo with trailing space + with pytest.raises(ValueError) as exc_info: + IntrospectionTokenVerifier( + introspection_url="https://auth.example.com/oauth/introspect", + client_id="test-client", + client_secret="test-secret", + client_auth_method="client_secret_basic ", # ty: ignore[invalid-argument-type] + ) + assert "Invalid client_auth_method" in str(exc_info.value) + assert "client_secret_basic " in str(exc_info.value) + + # Test completely invalid value + with pytest.raises(ValueError) as exc_info: + IntrospectionTokenVerifier( + introspection_url="https://auth.example.com/oauth/introspect", + client_id="test-client", + client_secret="test-secret", + client_auth_method="basic", # ty: ignore[invalid-argument-type] + ) + assert "Invalid client_auth_method" in str(exc_info.value) + assert "basic" in str(exc_info.value) + + async def test_client_secret_post_includes_credentials_in_body( + self, httpx_mock: HTTPXMock + ): + """Test that client_secret_post includes credentials in POST body.""" + verifier = IntrospectionTokenVerifier( + introspection_url="https://auth.example.com/oauth/introspect", + client_id="test-client", + client_secret="test-secret", + client_auth_method="client_secret_post", + ) + + httpx_mock.add_response( + url="https://auth.example.com/oauth/introspect", + method="POST", + json={"active": True, "client_id": "user-123"}, + ) + + await verifier.verify_token("test-token") + + # Verify request was made with credentials in body, not header + request = httpx_mock.get_request() + assert request is not None + assert request.method == "POST" + assert "Authorization" not in request.headers + assert request.headers["Content-Type"] == "application/x-www-form-urlencoded" + assert request.headers["Accept"] == "application/json" + + # Parse form data + body = request.content.decode("utf-8") + assert "token=test-token" in body + assert "token_type_hint=access_token" in body + assert "client_id=test-client" in body + assert "client_secret=test-secret" in body + + async def test_client_secret_post_verification_success(self, httpx_mock: HTTPXMock): + """Test successful token verification with client_secret_post.""" + verifier = IntrospectionTokenVerifier( + introspection_url="https://auth.example.com/oauth/introspect", + client_id="test-client", + client_secret="test-secret", + client_auth_method="client_secret_post", + ) + + httpx_mock.add_response( + url="https://auth.example.com/oauth/introspect", + method="POST", + json={ + "active": True, + "client_id": "user-123", + "scope": "read write", + "exp": int(time.time()) + 3600, + }, + ) + + access_token = await verifier.verify_token("test-token") + + assert access_token is not None + assert access_token.client_id == "user-123" + assert access_token.scopes == ["read", "write"] + + async def test_client_secret_basic_still_works( + self, verifier: IntrospectionTokenVerifier, httpx_mock: HTTPXMock + ): + """Test that client_secret_basic continues to work unchanged.""" + httpx_mock.add_response( + url="https://auth.example.com/oauth/introspect", + method="POST", + json={"active": True, "client_id": "user-123"}, + ) + + await verifier.verify_token("test-token") + + # Verify request was made with Basic Auth header + request = httpx_mock.get_request() + assert request is not None + assert "Authorization" in request.headers + assert request.headers["Authorization"].startswith("Basic ") + + # Verify credentials are NOT in body + body = request.content.decode("utf-8") + assert "client_id=" not in body + assert "client_secret=" not in body + class TestIntrospectionTokenVerifierIntegration: """Integration tests with FastMCP server."""