Add client_secret_post authentication to IntrospectionTokenVerifier (#2884)

This commit is contained in:
Marcus Shu 2026-01-17 02:07:33 +08:00 committed by GitHub
commit 083999ca14
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 195 additions and 15 deletions

View file

@ -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

View file

@ -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

View file

@ -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."""