mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Add MultiAuth for composing multiple token verification sources (#3335)
* Add MultiAuth for composing multiple token verification sources 🤖 Generated with Claude Code https://claude.ai/code/session_01WwKYDCqjM2FqYwY5ZNVvjb * Fix ruff lint/format in MultiAuth tests 🤖 Generated with Claude Code https://claude.ai/code/session_01WwKYDCqjM2FqYwY5ZNVvjb * Fix MultiAuth well-known route delegation and empty scopes handling 🤖 Generated with Claude Code https://claude.ai/code/session_01WwKYDCqjM2FqYwY5ZNVvjb * Harden MultiAuth: exception resilience, mcp_path propagation, test coverage - verify_token now catches exceptions from individual sources and continues to the next, so one broken verifier can't take down the whole chain - set_mcp_path propagates to verifiers, not just the server - Fix jwks_url→jwks_uri typo in class docstring - Add tests for raising verifiers, valid-token HTTP acceptance, and set_mcp_path propagation * Clean up MultiAuth: precompute sources, deduplicate test helpers * Fix version badges to 3.1.0 --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Jeremiah Lowin
GitHub
parent
1e72f2457b
commit
33a69d7d0a
6 changed files with 653 additions and 1 deletions
|
|
@ -179,7 +179,8 @@
|
|||
"servers/auth/remote-oauth",
|
||||
"servers/auth/oauth-proxy",
|
||||
"servers/auth/oidc-proxy",
|
||||
"servers/auth/full-oauth-server"
|
||||
"servers/auth/full-oauth-server",
|
||||
"servers/auth/multi-auth"
|
||||
]
|
||||
},
|
||||
"servers/authorization",
|
||||
|
|
|
|||
|
|
@ -178,6 +178,40 @@ This example shows the basic structure of a custom OAuth provider. The actual im
|
|||
|
||||
→ **Complete guide**: [Full OAuth Server](/servers/auth/full-oauth-server)
|
||||
|
||||
### MultiAuth
|
||||
|
||||
<VersionBadge version="3.1.0" />
|
||||
|
||||
`MultiAuth` composes multiple authentication sources into a single `auth` provider. When a server needs to accept tokens from different issuers — for example, an OAuth proxy for interactive clients alongside JWT verification for machine-to-machine tokens — `MultiAuth` tries each source in order and accepts the first successful verification.
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth import MultiAuth, OAuthProxy
|
||||
from fastmcp.server.auth.providers.jwt import JWTVerifier
|
||||
|
||||
auth = MultiAuth(
|
||||
server=OAuthProxy(
|
||||
issuer_url="https://login.example.com/...",
|
||||
client_id="my-app",
|
||||
client_secret="secret",
|
||||
base_url="https://my-server.com",
|
||||
),
|
||||
verifiers=[
|
||||
JWTVerifier(
|
||||
jwks_uri="https://internal-issuer.example.com/.well-known/jwks.json",
|
||||
issuer="https://internal-issuer.example.com",
|
||||
audience="my-mcp-server",
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
mcp = FastMCP("My Server", auth=auth)
|
||||
```
|
||||
|
||||
The server (if provided) owns all OAuth routes and metadata. Verifiers contribute only token verification logic. This keeps the MCP discovery surface clean while supporting multiple token sources.
|
||||
|
||||
→ **Complete guide**: [Multiple Auth Sources](/servers/auth/multi-auth)
|
||||
|
||||
## Configuration
|
||||
|
||||
Authentication providers are configured programmatically by instantiating them directly in your code with their required parameters. This makes dependencies explicit and allows your IDE to provide helpful autocompletion and type checking.
|
||||
|
|
@ -211,6 +245,8 @@ The authentication approach you choose depends on your existing infrastructure,
|
|||
|
||||
**Token validation works well when you already have authentication infrastructure that issues structured tokens.** If your organization already uses JWT-based systems, API gateways, or enterprise SSO that can generate tokens, this approach integrates seamlessly while keeping your MCP server focused on its core functionality. The simplicity comes from leveraging existing investment in authentication infrastructure.
|
||||
|
||||
**When you need tokens from multiple sources, use MultiAuth.** This is common in hybrid architectures where interactive clients authenticate through an OAuth proxy while backend services send JWT tokens directly. `MultiAuth` composes an optional auth server with additional token verifiers, trying each source in order until one succeeds.
|
||||
|
||||
**Full OAuth implementation should be avoided unless you have compelling reasons that external providers cannot address.** Air-gapped environments, specialized compliance requirements, or unique organizational constraints might justify this approach, but it requires significant security expertise and ongoing maintenance commitment. The complexity extends far beyond initial implementation to include threat monitoring, security updates, and keeping pace with evolving attack vectors.
|
||||
|
||||
FastMCP's architecture supports migration between these approaches as your requirements evolve. You can integrate with existing token systems initially and migrate to external identity providers as your application scales, or implement custom solutions when your requirements outgrow standard patterns.
|
||||
95
docs/servers/auth/multi-auth.mdx
Normal file
95
docs/servers/auth/multi-auth.mdx
Normal file
|
|
@ -0,0 +1,95 @@
|
|||
---
|
||||
title: Multiple Auth Sources
|
||||
sidebarTitle: Multiple Auth Sources
|
||||
description: Accept tokens from multiple authentication sources with a single server.
|
||||
icon: layer-group
|
||||
---
|
||||
|
||||
import { VersionBadge } from "/snippets/version-badge.mdx"
|
||||
|
||||
<VersionBadge version="3.1.0" />
|
||||
|
||||
Production servers often need to accept tokens from multiple authentication sources. An interactive application might authenticate through an OAuth proxy, while a backend service sends machine-to-machine JWT tokens directly. `MultiAuth` composes these sources into a single `auth` provider so every valid token is accepted regardless of where it was issued.
|
||||
|
||||
## Understanding MultiAuth
|
||||
|
||||
`MultiAuth` wraps an optional auth server (like `OAuthProxy`) together with one or more token verifiers (like `JWTVerifier`). When a request arrives with a bearer token, `MultiAuth` tries each source in order and accepts the first successful verification.
|
||||
|
||||
The auth server, if provided, is tried first. It owns all OAuth routes and metadata — the verifiers contribute only token verification logic. This keeps the MCP discovery surface clean: one set of routes, one set of metadata, multiple verification paths.
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth import MultiAuth, OAuthProxy
|
||||
from fastmcp.server.auth.providers.jwt import JWTVerifier
|
||||
|
||||
auth = MultiAuth(
|
||||
server=OAuthProxy(
|
||||
issuer_url="https://login.example.com/...",
|
||||
client_id="my-app",
|
||||
client_secret="secret",
|
||||
base_url="https://my-server.com",
|
||||
),
|
||||
verifiers=[
|
||||
JWTVerifier(
|
||||
jwks_uri="https://internal-issuer.example.com/.well-known/jwks.json",
|
||||
issuer="https://internal-issuer.example.com",
|
||||
audience="my-mcp-server",
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
mcp = FastMCP("My Server", auth=auth)
|
||||
```
|
||||
|
||||
Interactive MCP clients authenticate through the OAuth proxy as usual. Backend services skip OAuth entirely and send a JWT signed by the internal issuer. Both paths are validated, and the first match wins.
|
||||
|
||||
## Verification Order
|
||||
|
||||
`MultiAuth` checks sources in a deterministic order:
|
||||
|
||||
1. **Server** (if provided) — the full auth provider's `verify_token` runs first
|
||||
2. **Verifiers** — each `TokenVerifier` is tried in list order
|
||||
|
||||
The first source that returns a valid `AccessToken` wins. If every source returns `None`, the request receives a 401 response.
|
||||
|
||||
This ordering means the server acts as the "primary" authentication path, with verifiers as fallbacks for tokens the server doesn't recognize.
|
||||
|
||||
## Verifiers Only
|
||||
|
||||
You don't always need a full OAuth server. If your server only needs to accept tokens from multiple issuers, pass verifiers without a server:
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth import MultiAuth
|
||||
from fastmcp.server.auth.providers.jwt import JWTVerifier, StaticTokenVerifier
|
||||
|
||||
auth = MultiAuth(
|
||||
verifiers=[
|
||||
JWTVerifier(
|
||||
jwks_uri="https://issuer-a.example.com/.well-known/jwks.json",
|
||||
issuer="https://issuer-a.example.com",
|
||||
audience="my-server",
|
||||
),
|
||||
JWTVerifier(
|
||||
jwks_uri="https://issuer-b.example.com/.well-known/jwks.json",
|
||||
issuer="https://issuer-b.example.com",
|
||||
audience="my-server",
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
mcp = FastMCP("Multi-Issuer Server", auth=auth)
|
||||
```
|
||||
|
||||
Without a server, no OAuth routes or metadata are served. This is appropriate for internal systems where clients already know how to obtain tokens.
|
||||
|
||||
## API Reference
|
||||
|
||||
### MultiAuth
|
||||
|
||||
| Parameter | Type | Description |
|
||||
| --- | --- | --- |
|
||||
| `server` | `AuthProvider \| None` | Optional auth provider that owns routes and OAuth metadata. Also tried first for token verification. |
|
||||
| `verifiers` | `list[TokenVerifier] \| TokenVerifier` | One or more token verifiers tried after the server. |
|
||||
| `base_url` | `str \| None` | Override the base URL. Defaults to the server's `base_url`. |
|
||||
| `required_scopes` | `list[str] \| None` | Override required scopes. Defaults to the server's scopes. |
|
||||
|
|
@ -4,6 +4,7 @@ from .auth import (
|
|||
OAuthProvider,
|
||||
TokenVerifier,
|
||||
RemoteAuthProvider,
|
||||
MultiAuth,
|
||||
AccessToken,
|
||||
AuthProvider,
|
||||
)
|
||||
|
|
@ -61,6 +62,7 @@ __all__ = [
|
|||
"AuthProvider",
|
||||
"DebugTokenVerifier",
|
||||
"JWTVerifier",
|
||||
"MultiAuth",
|
||||
"OAuthProvider",
|
||||
"OAuthProxy",
|
||||
"OIDCProxy",
|
||||
|
|
|
|||
|
|
@ -469,6 +469,117 @@ class RemoteAuthProvider(AuthProvider):
|
|||
return routes
|
||||
|
||||
|
||||
class MultiAuth(AuthProvider):
|
||||
"""Composes an optional auth server with additional token verifiers.
|
||||
|
||||
Use this when a single server needs to accept tokens from multiple sources.
|
||||
For example, an OAuth proxy for interactive clients combined with a JWT
|
||||
verifier for machine-to-machine tokens.
|
||||
|
||||
Token verification tries the server first (if present), then each verifier
|
||||
in order, returning the first successful result. Routes and OAuth metadata
|
||||
come from the server; verifiers contribute only token verification.
|
||||
|
||||
Example:
|
||||
```python
|
||||
from fastmcp.server.auth import MultiAuth, JWTVerifier, OAuthProxy
|
||||
|
||||
auth = MultiAuth(
|
||||
server=OAuthProxy(issuer_url="https://login.example.com/..."),
|
||||
verifiers=[JWTVerifier(jwks_uri="https://example.com/.well-known/jwks.json")],
|
||||
)
|
||||
mcp = FastMCP("my-server", auth=auth)
|
||||
```
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
server: AuthProvider | None = None,
|
||||
verifiers: list[TokenVerifier] | TokenVerifier | None = None,
|
||||
base_url: AnyHttpUrl | str | None = None,
|
||||
required_scopes: list[str] | None = None,
|
||||
):
|
||||
"""Initialize the multi-auth provider.
|
||||
|
||||
Args:
|
||||
server: Optional auth provider (e.g., OAuthProxy) that owns routes
|
||||
and OAuth metadata. Also participates in token verification as
|
||||
the first verifier tried.
|
||||
verifiers: One or more token verifiers to try after the server.
|
||||
base_url: Override the base URL. Defaults to the server's base_url.
|
||||
required_scopes: Override required scopes. Defaults to the server's.
|
||||
"""
|
||||
if verifiers is None:
|
||||
verifiers = []
|
||||
elif isinstance(verifiers, TokenVerifier):
|
||||
verifiers = [verifiers]
|
||||
|
||||
if server is None and not verifiers:
|
||||
raise ValueError("MultiAuth requires at least a server or one verifier")
|
||||
|
||||
effective_base_url = base_url or (server.base_url if server else None)
|
||||
effective_scopes = (
|
||||
required_scopes
|
||||
if required_scopes is not None
|
||||
else (server.required_scopes if server else None)
|
||||
)
|
||||
|
||||
super().__init__(base_url=effective_base_url, required_scopes=effective_scopes)
|
||||
self.server = server
|
||||
self.verifiers = list(verifiers)
|
||||
|
||||
self._sources: list[AuthProvider] = []
|
||||
if self.server is not None:
|
||||
self._sources.append(self.server)
|
||||
self._sources.extend(self.verifiers)
|
||||
|
||||
async def verify_token(self, token: str) -> AccessToken | None:
|
||||
"""Verify a token by trying the server, then each verifier in order.
|
||||
|
||||
Each source is tried independently. If a source raises an exception,
|
||||
it is logged and treated as a non-match so that remaining sources
|
||||
still get a chance to verify the token.
|
||||
"""
|
||||
for source in self._sources:
|
||||
try:
|
||||
result = await source.verify_token(token)
|
||||
if result is not None:
|
||||
return result
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"Token verification failed for %s, trying next source",
|
||||
type(source).__name__,
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
def set_mcp_path(self, mcp_path: str | None) -> None:
|
||||
"""Propagate MCP path to the server and all verifiers."""
|
||||
super().set_mcp_path(mcp_path)
|
||||
if self.server is not None:
|
||||
self.server.set_mcp_path(mcp_path)
|
||||
for verifier in self.verifiers:
|
||||
verifier.set_mcp_path(mcp_path)
|
||||
|
||||
def get_routes(self, mcp_path: str | None = None) -> list[Route]:
|
||||
"""Delegate route creation to the server."""
|
||||
if self.server is not None:
|
||||
return self.server.get_routes(mcp_path)
|
||||
return []
|
||||
|
||||
def get_well_known_routes(self, mcp_path: str | None = None) -> list[Route]:
|
||||
"""Delegate well-known route creation to the server.
|
||||
|
||||
This ensures that server-specific well-known route logic (e.g.,
|
||||
OAuthProvider's RFC 8414 path-aware discovery) is preserved.
|
||||
"""
|
||||
if self.server is not None:
|
||||
return self.server.get_well_known_routes(mcp_path)
|
||||
return []
|
||||
|
||||
|
||||
class OAuthProvider(
|
||||
AuthProvider,
|
||||
OAuthAuthorizationServerProvider[AuthorizationCode, RefreshToken, AccessToken],
|
||||
|
|
|
|||
407
tests/server/auth/test_multi_auth.py
Normal file
407
tests/server/auth/test_multi_auth.py
Normal file
|
|
@ -0,0 +1,407 @@
|
|||
import httpx
|
||||
import pytest
|
||||
from pydantic import AnyHttpUrl
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth import MultiAuth, RemoteAuthProvider, TokenVerifier
|
||||
from fastmcp.server.auth.auth import AccessToken
|
||||
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier
|
||||
|
||||
|
||||
class RaisingVerifier(TokenVerifier):
|
||||
"""A verifier that always raises, for testing exception resilience."""
|
||||
|
||||
async def verify_token(self, token: str) -> AccessToken | None:
|
||||
raise RuntimeError("simulated failure")
|
||||
|
||||
|
||||
class TestMultiAuthInit:
|
||||
"""Test MultiAuth initialization and validation."""
|
||||
|
||||
def test_requires_server_or_verifiers(self):
|
||||
"""MultiAuth with neither server nor verifiers raises ValueError."""
|
||||
with pytest.raises(ValueError, match="at least a server or one verifier"):
|
||||
MultiAuth()
|
||||
|
||||
def test_server_only(self):
|
||||
verifier = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
provider = RemoteAuthProvider(
|
||||
token_verifier=verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
auth = MultiAuth(server=provider)
|
||||
assert auth.server is provider
|
||||
assert auth.verifiers == []
|
||||
|
||||
def test_verifiers_only(self):
|
||||
v = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
auth = MultiAuth(verifiers=[v])
|
||||
assert auth.server is None
|
||||
assert auth.verifiers == [v]
|
||||
|
||||
def test_single_verifier_not_in_list(self):
|
||||
"""A single TokenVerifier (not in a list) is accepted."""
|
||||
v = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
auth = MultiAuth(verifiers=v)
|
||||
assert auth.verifiers == [v]
|
||||
|
||||
def test_base_url_from_server(self):
|
||||
verifier = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
provider = RemoteAuthProvider(
|
||||
token_verifier=verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
auth = MultiAuth(server=provider)
|
||||
assert auth.base_url == AnyHttpUrl("https://api.example.com/")
|
||||
|
||||
def test_base_url_override(self):
|
||||
verifier = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
provider = RemoteAuthProvider(
|
||||
token_verifier=verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
auth = MultiAuth(server=provider, base_url="https://override.example.com")
|
||||
assert auth.base_url == AnyHttpUrl("https://override.example.com/")
|
||||
|
||||
def test_required_scopes_from_server(self):
|
||||
verifier = StaticTokenVerifier(
|
||||
tokens={"t": {"client_id": "c", "scopes": ["read"]}},
|
||||
required_scopes=["read"],
|
||||
)
|
||||
provider = RemoteAuthProvider(
|
||||
token_verifier=verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
auth = MultiAuth(server=provider)
|
||||
assert auth.required_scopes == ["read"]
|
||||
|
||||
|
||||
class TestMultiAuthVerifyToken:
|
||||
"""Test MultiAuth token verification chain."""
|
||||
|
||||
async def test_server_verified_first(self):
|
||||
"""Server's verify_token is tried before verifiers."""
|
||||
server_verifier = StaticTokenVerifier(
|
||||
tokens={"server_token": {"client_id": "server-client", "scopes": []}}
|
||||
)
|
||||
server = RemoteAuthProvider(
|
||||
token_verifier=server_verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
extra = StaticTokenVerifier(
|
||||
tokens={"extra_token": {"client_id": "extra-client", "scopes": []}}
|
||||
)
|
||||
|
||||
auth = MultiAuth(server=server, verifiers=[extra])
|
||||
|
||||
result = await auth.verify_token("server_token")
|
||||
assert result is not None
|
||||
assert result.client_id == "server-client"
|
||||
|
||||
async def test_falls_back_to_verifiers(self):
|
||||
"""When server rejects a token, verifiers are tried."""
|
||||
server_verifier = StaticTokenVerifier(
|
||||
tokens={"server_token": {"client_id": "server-client", "scopes": []}}
|
||||
)
|
||||
server = RemoteAuthProvider(
|
||||
token_verifier=server_verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
extra = StaticTokenVerifier(
|
||||
tokens={"m2m_token": {"client_id": "m2m-service", "scopes": []}}
|
||||
)
|
||||
|
||||
auth = MultiAuth(server=server, verifiers=[extra])
|
||||
|
||||
result = await auth.verify_token("m2m_token")
|
||||
assert result is not None
|
||||
assert result.client_id == "m2m-service"
|
||||
|
||||
async def test_verifier_order_matters(self):
|
||||
"""Verifiers are tried in order; first match wins."""
|
||||
v1 = StaticTokenVerifier(
|
||||
tokens={"shared_token": {"client_id": "first", "scopes": []}}
|
||||
)
|
||||
v2 = StaticTokenVerifier(
|
||||
tokens={"shared_token": {"client_id": "second", "scopes": []}}
|
||||
)
|
||||
|
||||
auth = MultiAuth(verifiers=[v1, v2])
|
||||
result = await auth.verify_token("shared_token")
|
||||
assert result is not None
|
||||
assert result.client_id == "first"
|
||||
|
||||
async def test_no_match_returns_none(self):
|
||||
"""When no server or verifier accepts the token, returns None."""
|
||||
v = StaticTokenVerifier(tokens={"known": {"client_id": "c", "scopes": []}})
|
||||
auth = MultiAuth(verifiers=[v])
|
||||
result = await auth.verify_token("unknown")
|
||||
assert result is None
|
||||
|
||||
async def test_verifiers_only_no_server(self):
|
||||
"""MultiAuth with only verifiers (no server) works."""
|
||||
v1 = StaticTokenVerifier(tokens={"token_a": {"client_id": "a", "scopes": []}})
|
||||
v2 = StaticTokenVerifier(tokens={"token_b": {"client_id": "b", "scopes": []}})
|
||||
|
||||
auth = MultiAuth(verifiers=[v1, v2])
|
||||
|
||||
result_a = await auth.verify_token("token_a")
|
||||
assert result_a is not None
|
||||
assert result_a.client_id == "a"
|
||||
|
||||
result_b = await auth.verify_token("token_b")
|
||||
assert result_b is not None
|
||||
assert result_b.client_id == "b"
|
||||
|
||||
async def test_raising_verifier_does_not_break_chain(self):
|
||||
"""If a verifier raises, the chain continues to the next source."""
|
||||
good = StaticTokenVerifier(
|
||||
tokens={"valid": {"client_id": "good-client", "scopes": []}}
|
||||
)
|
||||
auth = MultiAuth(verifiers=[RaisingVerifier(), good])
|
||||
|
||||
result = await auth.verify_token("valid")
|
||||
assert result is not None
|
||||
assert result.client_id == "good-client"
|
||||
|
||||
async def test_raising_server_does_not_break_chain(self):
|
||||
"""If the server raises, verifiers are still tried."""
|
||||
good = StaticTokenVerifier(
|
||||
tokens={"valid": {"client_id": "fallback", "scopes": []}}
|
||||
)
|
||||
auth = MultiAuth(server=RaisingVerifier(), verifiers=[good])
|
||||
|
||||
result = await auth.verify_token("valid")
|
||||
assert result is not None
|
||||
assert result.client_id == "fallback"
|
||||
|
||||
async def test_all_raising_returns_none(self):
|
||||
"""If every source raises, verify_token returns None."""
|
||||
auth = MultiAuth(verifiers=[RaisingVerifier(), RaisingVerifier()])
|
||||
result = await auth.verify_token("anything")
|
||||
assert result is None
|
||||
|
||||
async def test_server_match_short_circuits(self):
|
||||
"""When the server matches, verifiers are not consulted."""
|
||||
# Both server and verifier know the same token with different client_ids
|
||||
server_verifier = StaticTokenVerifier(
|
||||
tokens={"token": {"client_id": "from-server", "scopes": []}}
|
||||
)
|
||||
server = RemoteAuthProvider(
|
||||
token_verifier=server_verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
extra = StaticTokenVerifier(
|
||||
tokens={"token": {"client_id": "from-verifier", "scopes": []}}
|
||||
)
|
||||
|
||||
auth = MultiAuth(server=server, verifiers=[extra])
|
||||
result = await auth.verify_token("token")
|
||||
assert result is not None
|
||||
assert result.client_id == "from-server"
|
||||
|
||||
|
||||
class TestMultiAuthRoutes:
|
||||
"""Test that routes delegate to the server."""
|
||||
|
||||
def test_routes_from_server(self):
|
||||
verifier = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
server = RemoteAuthProvider(
|
||||
token_verifier=verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
auth = MultiAuth(server=server)
|
||||
routes = auth.get_routes(mcp_path="/mcp")
|
||||
# RemoteAuthProvider creates a protected resource metadata route
|
||||
assert len(routes) >= 1
|
||||
|
||||
def test_no_routes_without_server(self):
|
||||
v = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
auth = MultiAuth(verifiers=[v])
|
||||
assert auth.get_routes() == []
|
||||
|
||||
def test_well_known_routes_delegate_to_server(self):
|
||||
"""get_well_known_routes delegates to the server's implementation."""
|
||||
verifier = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
server = RemoteAuthProvider(
|
||||
token_verifier=verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
auth = MultiAuth(server=server)
|
||||
well_known = auth.get_well_known_routes(mcp_path="/mcp")
|
||||
server_well_known = server.get_well_known_routes(mcp_path="/mcp")
|
||||
# MultiAuth should produce the same well-known routes as the server
|
||||
assert len(well_known) == len(server_well_known)
|
||||
assert [r.path for r in well_known] == [r.path for r in server_well_known]
|
||||
|
||||
def test_well_known_routes_empty_without_server(self):
|
||||
v = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
auth = MultiAuth(verifiers=[v])
|
||||
assert auth.get_well_known_routes() == []
|
||||
|
||||
def test_required_scopes_explicit_empty_list(self):
|
||||
"""Passing required_scopes=[] explicitly clears inherited scopes."""
|
||||
verifier = StaticTokenVerifier(
|
||||
tokens={"t": {"client_id": "c", "scopes": ["read"]}},
|
||||
required_scopes=["read"],
|
||||
)
|
||||
server = RemoteAuthProvider(
|
||||
token_verifier=verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
# Server has required_scopes=["read"], but we explicitly clear them
|
||||
auth = MultiAuth(server=server, required_scopes=[])
|
||||
assert auth.required_scopes == []
|
||||
|
||||
|
||||
class TestMultiAuthIntegration:
|
||||
"""Integration tests: MultiAuth with a real FastMCP HTTP app."""
|
||||
|
||||
async def test_multi_auth_rejects_bad_tokens(self):
|
||||
"""End-to-end: MultiAuth rejects unknown tokens at the HTTP layer."""
|
||||
oauth_tokens = StaticTokenVerifier(
|
||||
tokens={
|
||||
"oauth_token": {
|
||||
"client_id": "interactive-client",
|
||||
"scopes": ["read"],
|
||||
}
|
||||
}
|
||||
)
|
||||
m2m_tokens = StaticTokenVerifier(
|
||||
tokens={
|
||||
"m2m_token": {
|
||||
"client_id": "backend-service",
|
||||
"scopes": ["read"],
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
auth = MultiAuth(verifiers=[oauth_tokens, m2m_tokens])
|
||||
mcp = FastMCP("test", auth=auth)
|
||||
app = mcp.http_app(path="/mcp")
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=httpx.ASGITransport(app=app),
|
||||
base_url="http://localhost",
|
||||
) as client:
|
||||
# No token → 401
|
||||
response = await client.get("/mcp")
|
||||
assert response.status_code == 401
|
||||
|
||||
# Bad token → 401
|
||||
response = await client.get(
|
||||
"/mcp", headers={"Authorization": "Bearer bad_token"}
|
||||
)
|
||||
assert response.status_code == 401
|
||||
|
||||
async def test_multi_auth_with_server_provides_routes(self):
|
||||
"""MultiAuth with a server exposes the server's metadata routes."""
|
||||
verifier = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
server = RemoteAuthProvider(
|
||||
token_verifier=verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
extra = StaticTokenVerifier(tokens={"m2m": {"client_id": "svc", "scopes": []}})
|
||||
|
||||
auth = MultiAuth(server=server, verifiers=[extra])
|
||||
mcp = FastMCP("test", auth=auth)
|
||||
app = mcp.http_app(path="/mcp")
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=httpx.ASGITransport(app=app),
|
||||
base_url="https://api.example.com",
|
||||
) as client:
|
||||
# Protected resource metadata should be available
|
||||
response = await client.get("/.well-known/oauth-protected-resource/mcp")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["resource"] == "https://api.example.com/mcp"
|
||||
|
||||
async def test_multi_auth_accepts_valid_verifier_token(self):
|
||||
"""MultiAuth accepts tokens from verifiers (not just the server).
|
||||
|
||||
Verifies that both server and verifier tokens pass the HTTP auth
|
||||
middleware. We use GET /mcp to check: 401 means auth rejected,
|
||||
any other status means auth accepted and the request reached the
|
||||
MCP session layer.
|
||||
"""
|
||||
interactive_tokens = StaticTokenVerifier(
|
||||
tokens={
|
||||
"interactive_token": {
|
||||
"client_id": "interactive-client",
|
||||
"scopes": [],
|
||||
}
|
||||
}
|
||||
)
|
||||
m2m_tokens = StaticTokenVerifier(
|
||||
tokens={
|
||||
"m2m_token": {
|
||||
"client_id": "backend-service",
|
||||
"scopes": [],
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
auth = MultiAuth(verifiers=[interactive_tokens, m2m_tokens])
|
||||
mcp = FastMCP("test", auth=auth)
|
||||
app = mcp.http_app(path="/mcp")
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=httpx.ASGITransport(app=app, raise_app_exceptions=False),
|
||||
base_url="http://localhost",
|
||||
) as client:
|
||||
# No token → 401
|
||||
response = await client.get("/mcp")
|
||||
assert response.status_code == 401
|
||||
|
||||
# Interactive token passes auth (non-401 means auth accepted)
|
||||
response = await client.get(
|
||||
"/mcp", headers={"Authorization": "Bearer interactive_token"}
|
||||
)
|
||||
assert response.status_code != 401
|
||||
|
||||
# M2M token also passes auth
|
||||
response = await client.get(
|
||||
"/mcp", headers={"Authorization": "Bearer m2m_token"}
|
||||
)
|
||||
assert response.status_code != 401
|
||||
|
||||
# Bad token → 401
|
||||
response = await client.get(
|
||||
"/mcp", headers={"Authorization": "Bearer bad_token"}
|
||||
)
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestMultiAuthSetMcpPath:
|
||||
"""Test that set_mcp_path propagates to server and verifiers."""
|
||||
|
||||
def test_propagates_to_server(self):
|
||||
verifier = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
server = RemoteAuthProvider(
|
||||
token_verifier=verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
auth = MultiAuth(server=server)
|
||||
auth.set_mcp_path("/mcp")
|
||||
assert server._mcp_path == "/mcp"
|
||||
|
||||
def test_propagates_to_verifiers(self):
|
||||
v1 = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}})
|
||||
v2 = StaticTokenVerifier(tokens={"t2": {"client_id": "c2", "scopes": []}})
|
||||
auth = MultiAuth(verifiers=[v1, v2])
|
||||
auth.set_mcp_path("/mcp")
|
||||
assert v1._mcp_path == "/mcp"
|
||||
assert v2._mcp_path == "/mcp"
|
||||
Loading…
Add table
Add a link
Reference in a new issue