ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-24 06:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add Authplane auth provider

Browse source 
RemoteAuthProvider on JWTVerifier (ES256/RS256 only), RFC 8707 audience binding, and DPoP-bound (cnf) token rejection. Provider + 31 tests + docs page + runnable example. Closes #4667.
This commit is contained in:
muralx 2026-08-07 11:34:17 +01:00
commit 0c9ef9a716
7 changed files with 878 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docs/docs.json
View file

@ -286,6 +286,7 @@
"pages": [
"integrations/auth0",
"integrations/authkit",
"integrations/authplane",
"integrations/aws-cognito",
"integrations/azure",
"integrations/descope",

218
docs/integrations/authplane.mdx Normal file
View file

@ -0,0 +1,218 @@
---
title: Authplane OAuth 🤝 FastMCP
sidebarTitle: Authplane
description: Secure your FastMCP server with Authplane, a self-hosted OAuth 2.1 authorization server for MCP.
icon: shield-check
tag: NEW
---
import { VersionBadge } from "/snippets/version-badge.mdx"
<VersionBadge version="4.0.0" />
[Authplane](https://github.com/AuthPlane/authserver) is a self-hosted OAuth 2.1
authorization server for the Model Context Protocol, shipped as a single Go
binary. It implements the MCP Authorization specification (2025-11-25),
including Dynamic Client Registration (RFC 7591), Client ID Metadata Documents,
Resource Indicators (RFC 8707) and JWT access tokens (RFC 9068).
Because Authplane supports DCR, MCP clients register themselves at runtime — you
do not have to pre-provision a `client_id` for every client that wants to reach
your server.
## Configuration
<Note>
`AuthplaneAuthProvider` makes your FastMCP server a **resource server**. It
verifies incoming JWTs against Authplane's JWKS and serves Protected Resource
Metadata (RFC 9728) so clients can discover the authorization server. Authplane
itself runs separately and handles the authorization flow.
</Note>
```python
from fastmcp import FastMCP
from fastmcp.server.auth.providers.authplane import AuthplaneAuthProvider
auth = AuthplaneAuthProvider(
issuer="https://auth.example.com",
base_url="https://my-mcp-server.example.com",
required_scopes=["tools/read"],
)
mcp = FastMCP("My App", auth=auth)
@mcp.tool
def read_record(record_id: str) -> str:
return f"record {record_id}"
if __name__ == "__main__":
mcp.run(transport="http", port=8000)
```
### Parameters
<ParamField body="issuer" type="str" required>
Base URL of your Authplane authorization server. This is the `iss` claim value
and the root of the RFC 8414 discovery document.
</ParamField>
<ParamField body="base_url" type="str" required>
Public URL of this FastMCP server.
</ParamField>
<ParamField body="required_scopes" type="list[str] | str">
Scopes that must be present on every incoming token. Defaults to none, leaving
enforcement to individual tools.
</ParamField>
<ParamField body="scopes_supported" type="list[str] | str">
Scopes advertised in Protected Resource Metadata so clients know what to
request. Defaults to `required_scopes`.
</ParamField>
<ParamField body="audience" type="str | list[str]">
Expected `aud` claim. Defaults to this server's resource URL — see
[Audience binding](#audience-binding).
</ParamField>
<ParamField body="algorithm" type='"ES256" | "RS256"' default="ES256">
JWT signing algorithm to accept. Authplane signs access tokens with ES256 (its
default) or RS256; only these two are accepted. HS256 and `none` are rejected.
</ParamField>
<ParamField body="resource_base_url" type="str">
Public base URL of the protected resource when it differs from `base_url`,
for example behind a reverse proxy.
</ParamField>
<ParamField body="token_verifier" type="TokenVerifier">
Custom token verifier. Defaults to a `JWTVerifier` pointed at
`{issuer}/.well-known/jwks.json`.
</ParamField>
## Audience binding
Authplane audience-binds every access token to the resource URI the client asked
for (RFC 8707), so a token minted for one MCP server cannot be replayed against
another. The provider enforces that binding for you: once FastMCP reports the
path your MCP endpoint is mounted at, the verifier's expected `aud` is set to the
resulting resource URL — the same URL advertised in Protected Resource Metadata.
With the configuration above and the default `/mcp` path, tokens must carry
`aud: "https://my-mcp-server.example.com/mcp"`.
Pass `audience` explicitly only if your deployment overrides resource indicators.
## Running Authplane locally
```bash
export AUTHPLANE_ADMIN_API_KEY="$(openssl rand -hex 32)"
export AUTHPLANE_SESSION_SECRET="$(openssl rand -hex 32)"
docker run -p 9000:9000 -p 9001:9001 \
-e AUTHPLANE_ADMIN_API_KEY \
-e AUTHPLANE_SESSION_SECRET \
-v authserver-data:/data \
authplane/authserver:latest serve
```
The public OAuth endpoints are on port `9000`; point `issuer` at
`http://localhost:9000` and the Admin UI at `http://localhost:9001/admin/ui/`.
## Register your server as a resource
Register your MCP server's URL as a resource in Authplane before clients connect.
The registered `uri` must match the URL this provider advertises — `base_url`
plus the MCP path — **exactly**:
```bash
curl -X POST http://localhost:9001/admin/resources \
-H "Authorization: Bearer $AUTHPLANE_ADMIN_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"slug": "my-mcp-server",
"uri": "https://my-mcp-server.example.com/mcp",
"backend_kind": "mint",
"display_name": "My App",
"scopes": [
{"name": "tools/read", "description": "Read records"}
]
}'
```
Authplane audience-binds tokens to this `uri` (RFC 8707) and the provider
requires that audience, so the two must match. Every scope you list in
`required_scopes` must also be declared here.
## Verifying the wiring
```bash
# The authorization server's metadata (served by Authplane)
curl -s http://localhost:9000/.well-known/oauth-authorization-server | jq
# Your server's protected resource metadata (served by this provider)
curl -s http://localhost:8000/.well-known/oauth-protected-resource/mcp | jq
```
The `authorization_servers` array in the second document must contain the issuer
from the first. If it does not, `issuer` and the running Authplane instance
disagree.
## Going further with `authplane-fastmcp`
`AuthplaneAuthProvider` validates JWTs against Authplane's JWKS, enforces scopes,
binds token audience to your resource (RFC 8707), and serves Protected Resource
Metadata — with no dependency beyond FastMCP. That covers most MCP servers.
When your server needs to do more than validate tokens, Authplane's first-party
adapter [`authplane-fastmcp`](https://github.com/AuthPlane/python-sdk/tree/main/authplane-fastmcp)
is a drop-in `FastMCP(**...)` backed by the full Authplane Python SDK. It keeps
everything above and adds:
- **Inbound DPoP** proof-of-possession (RFC 9449) — full proof verification with
server nonces and `jti` replay protection (pluggable store), so a stolen token
can't be replayed from another machine.
- **Introspection** (RFC 7662) and **revocation checking** (RFC 7009), so a
revoked token stops working immediately instead of at expiry.
- **Token exchange** (RFC 8693) for calling a downstream API on the user's
behalf, including decoding upstream-provider consent (the Broker flow) into
MCP `-32042` errors.
- **Metadata discovery** (RFC 8414): the JWKS URI is read from the authorization
server's metadata, and both JWKS and metadata refresh in the background — with
stale-cache fallback if the server is briefly unreachable.
- **SSRF-hardened fetching**, a **circuit breaker**, and a **token cache** for
resilient production deployments.
```bash
pip install authplane-fastmcp
```
```python
import asyncio
from authplane_fastmcp import authplane_auth
from fastmcp import FastMCP
async def main():
result = await authplane_auth(
issuer="https://auth.example.com",
base_url="https://my-mcp-server.example.com",
scopes=["tools/read"],
)
mcp = FastMCP("My App", **result)
try:
await mcp.run_async(transport="http", port=8000)
finally:
await result.aclose()
asyncio.run(main())
```
`authplane_auth()` holds background JWKS and metadata refresh tasks, so call
`aclose()` on shutdown.
See the [authplane-fastmcp user guide](https://github.com/AuthPlane/python-sdk/blob/main/authplane-fastmcp/docs/user-guide.md).

82
examples/auth/authplane_oauth/README.md Normal file
View file

@ -0,0 +1,82 @@
# Authplane OAuth Example
Demonstrates FastMCP server protection with [Authplane](https://github.com/AuthPlane/authserver),
a self-hosted OAuth 2.1 authorization server for MCP.
Authplane supports Dynamic Client Registration, so MCP clients register
themselves at runtime — no pre-provisioned `client_id` is needed.
## Setup
1. Run Authplane (Docker):
```bash
export AUTHPLANE_ADMIN_API_KEY="$(openssl rand -hex 32)"
export AUTHPLANE_SESSION_SECRET="$(openssl rand -hex 32)"
docker run -p 9000:9000 -p 9001:9001 \
-e AUTHPLANE_ADMIN_API_KEY -e AUTHPLANE_SESSION_SECRET \
authplane/authserver:latest serve
```
Public OAuth endpoints are on `:9000`; the Admin UI is at
`http://localhost:9001/admin/ui/`.
2. Register this server's resource. The `uri` must match `base_url` + the MCP
path (`/mcp`) exactly, and must declare the scopes the server requires:
```bash
curl -X POST http://localhost:9001/admin/resources \
-H "Authorization: Bearer $AUTHPLANE_ADMIN_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"slug": "example",
"uri": "http://127.0.0.1:8000/mcp",
"backend_kind": "mint",
"display_name": "Authplane Example",
"scopes": [{"name": "tools/read", "description": "Read access"}]
}'
```
3. Create a user to sign in as (the client opens a browser to log in):
```bash
curl -X POST http://localhost:9001/admin/users \
-H "Authorization: Bearer $AUTHPLANE_ADMIN_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"email": "user@example.com",
"name": "Example User",
"password": "changeme",
"role": "user"
}'
```
4. Point the server at your Authplane instance:
```bash
export AUTHPLANE_ISSUER="http://localhost:9000"
```
5. Run the server:
```bash
python server.py
```
6. In another terminal, run the client:
```bash
python client.py
```
The client opens your browser for Authplane authentication — sign in with the
email and password from step 3. It then calls the protected
`get_access_token_claims` tool.
## Notes
- The server accepts tokens signed with **ES256** (Authplane's default) or
**RS256**.
- The token's audience is bound to `http://127.0.0.1:8000/mcp` (RFC 8707), so a
token minted for a different resource will not work here.

33
examples/auth/authplane_oauth/client.py Normal file
View file

@ -0,0 +1,33 @@
"""OAuth client example for connecting to an Authplane-protected FastMCP server.
To run:
python client.py
"""
import asyncio
from fastmcp import Client
SERVER_URL = "http://127.0.0.1:8000/mcp"
async def main():
async with Client(SERVER_URL, auth="oauth") as client:
assert await client.ping()
print("Successfully authenticated!")
tools = await client.list_tools()
print(f"Available tools ({len(tools)}):")
for tool in tools:
print(f" - {tool.name}: {tool.description}")
print("Calling protected tool: get_access_token_claims")
result = await client.call_tool("get_access_token_claims")
claims = result.data
print(f" sub: {claims.get('sub', 'N/A')}")
print(f" scope: {claims.get('scope', 'N/A')}")
print(f" aud: {claims.get('aud', 'N/A')}")
if __name__ == "__main__":
asyncio.run(main())

48
examples/auth/authplane_oauth/server.py Normal file
View file

@ -0,0 +1,48 @@
"""Authplane OAuth server example for FastMCP.
This example demonstrates how to protect a FastMCP server with Authplane,
a self-hosted OAuth 2.1 authorization server for MCP.
Before running, register this server's resource URL in Authplane (see README).
To run:
AUTHPLANE_ISSUER=https://your-authplane.com python server.py
"""
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.authplane import AuthplaneAuthProvider
from fastmcp.server.dependencies import get_access_token
auth = AuthplaneAuthProvider(
issuer=os.getenv("AUTHPLANE_ISSUER") or "http://localhost:9000",
base_url="http://127.0.0.1:8000",
required_scopes=["tools/read"],
)
mcp = FastMCP("Authplane Example Server", auth=auth)
@mcp.tool
def echo(message: str) -> str:
"""Echo the provided message."""
return message
@mcp.tool
async def get_access_token_claims() -> dict:
"""Get the authenticated user's access token claims."""
token = get_access_token()
if token is None:
return {"error": "Not authenticated"}
return {
"sub": token.claims.get("sub"),
"scope": token.claims.get("scope"),
"aud": token.claims.get("aud"),
"client_id": token.claims.get("client_id"),
}
if __name__ == "__main__":
mcp.run(transport="http", port=8000)

216
fastmcp_slim/fastmcp/server/auth/providers/authplane.py Normal file
View file

@ -0,0 +1,216 @@
"""Authplane authentication provider for FastMCP."""
from __future__ import annotations
from typing import Literal
from pydantic import AnyHttpUrl
from fastmcp.server.auth import AccessToken, RemoteAuthProvider, TokenVerifier
from fastmcp.server.auth.providers.jwt import JWTVerifier
from fastmcp.utilities.auth import parse_scopes
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
# The only algorithms Authplane signs access tokens with. The AS's
# `signing.algorithm` config accepts exactly these two — ES256 (default) or
# RS256 — and its validator rejects anything else. (PS256 appears elsewhere in
# Authplane, but only as an accepted *DPoP proof* algorithm — a different
# verification context, not access-token signing.) HS256 and `none` are never
# issued and are always rejected server-side, so we never accept them here
# either, even though FastMCP's JWTVerifier would allow HS256. Restricting the
# accepted set to what the AS actually signs with closes off algorithm confusion.
AuthplaneAlgorithm = Literal["ES256", "RS256"]
_SUPPORTED_ALGORITHMS: frozenset[str] = frozenset({"ES256", "RS256"})
class _BearerOnlyJWTVerifier(JWTVerifier):
"""A `JWTVerifier` that refuses DPoP-bound (sender-constrained) tokens.
Authplane can issue DPoP-bound access tokens (RFC 9449); those carry a `cnf`
claim and are only safe to accept alongside a verified DPoP proof. This
provider validates the bearer JWT only — it never sees the DPoP proof header
(FastMCP's `verify_token(token)` hook receives just the token string), so
accepting a `cnf`-bound token as a plain bearer would silently defeat the
sender-constraint: a stolen token would be replayable from any machine.
So any token carrying `cnf` is rejected here. Deployments that issue
DPoP-bound tokens should use Authplane's `authplane-fastmcp` package, which
verifies the proof. Tokens without `cnf` are unaffected.
"""
async def load_access_token(self, token: str) -> AccessToken | None:
access = await super().load_access_token(token)
if access is not None and "cnf" in access.claims:
logger.warning(
"Authplane: rejecting DPoP-bound token (cnf present). This "
"provider validates bearer tokens only; use authplane-fastmcp "
"for DPoP proof verification."
)
return None
return access
class AuthplaneAuthProvider(RemoteAuthProvider):
"""Authplane authentication provider.
`Authplane <https://github.com/AuthPlane/authserver>`_ is a self-hosted
OAuth 2.1 authorization server for the Model Context Protocol, shipped as a
single Go binary. It implements the MCP Authorization specification
(2025-11-25): Dynamic Client Registration (RFC 7591), Client ID Metadata
Documents, Resource Indicators (RFC 8707), and JWT access tokens (RFC 9068).
Because Authplane supports DCR, MCP clients can register themselves at
runtime — no pre-provisioned `client_id` is needed for the FastMCP server
operator to hand out.
This provider makes the FastMCP server a resource server: it verifies
incoming JWTs against the Authplane JWKS and serves Protected Resource
Metadata (RFC 9728) pointing clients at the Authplane instance.
Audience binding
Authplane audience-binds every access token to the resource URI the
client asked for (RFC 8707), so a token minted for one MCP server cannot
be replayed against another. This provider enforces that binding
automatically: once FastMCP reports the path the MCP endpoint is mounted
at, the verifier's expected audience is set to the resulting resource
URL. Pass ``audience`` explicitly to override.
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.auth.providers.authplane import AuthplaneAuthProvider
auth = AuthplaneAuthProvider(
issuer="https://auth.example.com",
base_url="https://my-mcp-server.example.com",
required_scopes=["tools/read"],
)
mcp = FastMCP("My App", auth=auth)
```
Note:
This provider validates JWTs against Authplane's JWKS, enforces scopes,
and binds token audience to the resource (RFC 8707) — the common case,
with no dependency beyond FastMCP. It validates *bearer* tokens only:
a DPoP-bound token (RFC 9449, carrying a `cnf` claim) is rejected rather
than accepted as a plain bearer, since this provider does not verify the
DPoP proof and accepting one would defeat the sender-constraint. For
inbound DPoP proof-of-possession, token introspection (RFC 7662),
revocation checking (RFC 7009), RFC 8693 token exchange, and RFC 8414
metadata discovery with background JWKS/metadata refresh, use Authplane's
first-party ``authplane-fastmcp`` package — a drop-in ``FastMCP(**...)``
backed by the full Authplane SDK.
"""
def __init__(
self,
*,
issuer: AnyHttpUrl | str,
base_url: AnyHttpUrl | str,
required_scopes: list[str] | str | None = None,
scopes_supported: list[str] | str | None = None,
audience: str | list[str] | None = None,
algorithm: AuthplaneAlgorithm = "ES256",
resource_base_url: AnyHttpUrl | str | None = None,
resource_name: str | None = None,
resource_documentation: AnyHttpUrl | None = None,
token_verifier: TokenVerifier | None = None,
):
"""Initialize the Authplane auth provider.
Args:
issuer: Base URL of the Authplane authorization server (e.g.
"https://auth.example.com"). This is the `iss` claim value and
the root of the RFC 8414 discovery document.
base_url: Public URL of this FastMCP server.
required_scopes: Scopes to require on incoming tokens. Defaults to
none, leaving per-tool enforcement to the server.
scopes_supported: Scopes to advertise in Protected Resource
Metadata so clients know what to request. Defaults to
`required_scopes`.
audience: Expected `aud` claim. Defaults to the resource URL, which
is what Authplane audience-binds tokens to. Set explicitly only
when the deployment overrides resource indicators.
algorithm: JWT signing algorithm to accept. Authplane signs access
tokens with ES256 (the AS's default) or RS256 — those are the
only two its `signing.algorithm` config permits. Only those two
are accepted here; HS256 and `none` are never issued by Authplane
and are rejected regardless of what the caller passes. Ignored
when `token_verifier` is supplied.
resource_base_url: Optional public base URL for the protected
resource when it differs from `base_url` (e.g. behind a proxy).
resource_name: Optional human-readable name for the resource.
resource_documentation: Optional documentation URL for the resource.
token_verifier: Optional custom token verifier. Defaults to a
`JWTVerifier` pointed at Authplane's JWKS endpoint.
"""
self.issuer = str(issuer).rstrip("/")
parsed_required_scopes = (
parse_scopes(required_scopes) if required_scopes is not None else []
)
parsed_scopes_supported = (
parse_scopes(scopes_supported)
if scopes_supported is not None
else parsed_required_scopes or None
)
# Only bind the audience automatically when we own the verifier and the
# caller did not pin one. A caller-supplied verifier is theirs to
# configure; silently rewriting its audience would be surprising.
self._bind_audience_to_resource = audience is None and token_verifier is None
if token_verifier is None:
# Runtime guard, not just the Literal hint: a dynamically supplied
# string (e.g. from config) must not widen the accepted set to an
# algorithm Authplane never signs with. Only enforced when we build
# the verifier — a caller-supplied verifier owns its own policy.
if algorithm not in _SUPPORTED_ALGORITHMS:
raise ValueError(
f"Unsupported signing algorithm {algorithm!r}. Authplane "
f"signs access tokens only with {sorted(_SUPPORTED_ALGORITHMS)}."
)
token_verifier = _BearerOnlyJWTVerifier(
jwks_uri=f"{self.issuer}/.well-known/jwks.json",
issuer=self.issuer,
algorithm=algorithm,
required_scopes=parsed_required_scopes,
audience=audience,
)
super().__init__(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl(self.issuer)],
base_url=AnyHttpUrl(str(base_url).rstrip("/")),
scopes_supported=parsed_scopes_supported,
resource_base_url=resource_base_url,
resource_name=resource_name,
resource_documentation=resource_documentation,
)
def set_mcp_path(self, mcp_path: str | None) -> None:
"""Bind the expected token audience to this server's resource URL.
Authplane issues tokens whose `aud` is the RFC 8707 resource indicator
the client requested — the full MCP endpoint URL. That URL is only known
once FastMCP reports where the endpoint is mounted, which is what this
hook is for.
"""
super().set_mcp_path(mcp_path)
if not self._bind_audience_to_resource:
return
if self._resource_url is None:
return
if not isinstance(self.token_verifier, JWTVerifier):
return
resource_url = str(self._resource_url)
self.token_verifier.audience = resource_url
logger.info(
"Authplane: bound expected token audience to resource URL %s",
resource_url,
)

280
tests/server/auth/providers/test_authplane.py Normal file
View file

@ -0,0 +1,280 @@
"""Tests for the Authplane auth provider."""
from __future__ import annotations
from typing import Any
import pytest
from pydantic import AnyHttpUrl
from fastmcp.server.auth import AccessToken
from fastmcp.server.auth.providers.authplane import (
AuthplaneAuthProvider,
_BearerOnlyJWTVerifier,
)
from fastmcp.server.auth.providers.jwt import JWTVerifier
ISSUER = "https://auth.example.com"
BASE_URL = "https://mcp.example.com"
def make_provider(**kwargs: Any) -> AuthplaneAuthProvider:
params: dict[str, Any] = {"issuer": ISSUER, "base_url": BASE_URL}
params.update(kwargs)
return AuthplaneAuthProvider(**params)
def jwt_verifier(provider: AuthplaneAuthProvider) -> JWTVerifier:
"""Narrow the provider's verifier to the concrete `JWTVerifier` it builds."""
verifier = provider.token_verifier
assert isinstance(verifier, JWTVerifier)
return verifier
def bearer_verifier(provider: AuthplaneAuthProvider) -> _BearerOnlyJWTVerifier:
"""Narrow the provider's verifier to the `_BearerOnlyJWTVerifier` it builds."""
verifier = provider.token_verifier
assert isinstance(verifier, _BearerOnlyJWTVerifier)
return verifier
class TestDefaults:
def test_builds_jwt_verifier_against_authplane_jwks(self):
provider = make_provider()
verifier = provider.token_verifier
assert isinstance(verifier, JWTVerifier)
assert verifier.jwks_uri == f"{ISSUER}/.well-known/jwks.json"
assert verifier.issuer == ISSUER
assert verifier.algorithm == "ES256"
def test_advertises_issuer_as_authorization_server(self):
provider = make_provider()
assert provider.authorization_servers == [AnyHttpUrl(ISSUER)]
def test_no_scopes_are_required_by_default(self):
provider = make_provider()
assert provider.required_scopes == []
def test_default_algorithm_matches_the_as_default(self):
# Authplane's own signing default is ES256; the provider must default to
# the same, or every token fails validation on a default install.
provider = make_provider()
assert jwt_verifier(provider).algorithm == "ES256"
@pytest.mark.parametrize("alg", ["ES256", "RS256"])
def test_supported_algorithms_are_accepted(self, alg):
provider = make_provider(algorithm=alg)
assert jwt_verifier(provider).algorithm == alg
class TestAlgorithmRestriction:
"""Only the two algorithms Authplane signs access tokens with are accepted."""
@pytest.mark.parametrize("alg", ["HS256", "HS384", "HS512", "none", "PS256"])
def test_unsupported_algorithms_are_rejected(self, alg):
# HS256 is accepted by FastMCP's own JWTVerifier but is never issued by
# Authplane; accepting it would open algorithm confusion. PS256 is a
# valid DPoP-proof algorithm but Authplane never *signs access tokens*
# with it, so it must not be accepted here either. Rejected before the
# verifier is built.
with pytest.raises(ValueError, match="Unsupported signing algorithm"):
make_provider(algorithm=alg)
def test_a_supplied_verifier_owns_its_algorithm_policy(self):
# A caller who brings their own verifier is not subject to the guard —
# the `algorithm` argument is ignored entirely, even if unsupported.
custom = JWTVerifier(
jwks_uri="https://elsewhere.example.com/keys",
issuer=ISSUER,
algorithm="ES256",
)
provider = make_provider(token_verifier=custom, algorithm="PS256")
assert provider.token_verifier is custom
assert custom.algorithm == "ES256"
def _fake_access_token(**claims: Any) -> AccessToken:
return AccessToken(
token="tok",
client_id="client",
scopes=list(claims.get("scope", "").split()),
expires_at=None,
claims=claims,
)
class TestDpopBoundTokenRejection:
"""The default verifier accepts bearer tokens only; DPoP-bound (cnf) tokens
are rejected rather than silently accepted as plain bearer."""
def test_default_verifier_is_bearer_only(self):
provider = make_provider()
assert isinstance(provider.token_verifier, _BearerOnlyJWTVerifier)
async def test_cnf_bound_token_is_rejected(self, monkeypatch):
provider = make_provider()
verifier = bearer_verifier(provider)
bound = _fake_access_token(sub="u", scope="tools/read", cnf={"jkt": "abc"})
async def fake_super(self, token): # noqa: ANN001
return bound
# The parent JWTVerifier does all the real validation and returns a
# valid token; our subclass must still reject it for carrying `cnf`.
monkeypatch.setattr(JWTVerifier, "load_access_token", fake_super)
assert await verifier.load_access_token("tok") is None
async def test_plain_bearer_token_passes_through(self, monkeypatch):
provider = make_provider()
verifier = bearer_verifier(provider)
plain = _fake_access_token(sub="u", scope="tools/read")
async def fake_super(self, token): # noqa: ANN001
return plain
monkeypatch.setattr(JWTVerifier, "load_access_token", fake_super)
assert await verifier.load_access_token("tok") is plain
async def test_invalid_token_stays_rejected(self, monkeypatch):
# A token the parent rejects (None) must remain rejected.
provider = make_provider()
verifier = bearer_verifier(provider)
async def fake_super(self, token): # noqa: ANN001
return None
monkeypatch.setattr(JWTVerifier, "load_access_token", fake_super)
assert await verifier.load_access_token("tok") is None
async def test_supplied_verifier_is_not_wrapped(self, monkeypatch):
# A caller-supplied verifier owns its own DPoP policy — we don't wrap it.
custom = JWTVerifier(
jwks_uri="https://elsewhere.example.com/keys",
issuer=ISSUER,
)
provider = make_provider(token_verifier=custom)
assert provider.token_verifier is custom
assert not isinstance(provider.token_verifier, _BearerOnlyJWTVerifier)
class TestUrlNormalization:
@pytest.mark.parametrize("issuer", [ISSUER, f"{ISSUER}/"])
def test_trailing_slash_is_stripped_from_issuer(self, issuer):
provider = make_provider(issuer=issuer)
assert provider.issuer == ISSUER
assert jwt_verifier(provider).jwks_uri == f"{ISSUER}/.well-known/jwks.json"
def test_trailing_slash_is_stripped_from_base_url(self):
provider = make_provider(base_url=f"{BASE_URL}/")
assert str(provider.base_url).rstrip("/") == BASE_URL
class TestScopes:
def test_space_delimited_required_scopes_are_parsed(self):
provider = make_provider(required_scopes="tools/read tools/write")
assert provider.required_scopes == ["tools/read", "tools/write"]
def test_scopes_supported_defaults_to_required_scopes(self):
provider = make_provider(required_scopes=["tools/read"])
assert jwt_verifier(provider).required_scopes == ["tools/read"]
assert provider._scopes_supported == ["tools/read"]
def test_scopes_supported_can_exceed_required_scopes(self):
provider = make_provider(
required_scopes=["tools/read"],
scopes_supported=["tools/read", "tools/write"],
)
assert provider.required_scopes == ["tools/read"]
assert provider._scopes_supported == ["tools/read", "tools/write"]
class TestAudienceBinding:
def test_audience_is_bound_to_the_resource_url(self):
provider = make_provider()
provider.get_routes(mcp_path="/mcp")
assert jwt_verifier(provider).audience == f"{BASE_URL}/mcp"
def test_explicit_audience_is_not_overwritten(self):
provider = make_provider(audience="https://pinned.example.com/mcp")
provider.get_routes(mcp_path="/mcp")
assert jwt_verifier(provider).audience == "https://pinned.example.com/mcp"
def test_resource_base_url_drives_the_bound_audience(self):
provider = make_provider(resource_base_url="https://public.example.com")
provider.get_routes(mcp_path="/mcp")
assert jwt_verifier(provider).audience == "https://public.example.com/mcp"
def test_audience_falls_back_to_base_url_when_no_mcp_path_is_known(self):
provider = make_provider()
provider.get_routes(mcp_path=None)
# The invariant is that the expected audience always equals the
# resource URL advertised in Protected Resource Metadata; with no
# mounted path, that resource URL is base_url itself.
assert jwt_verifier(provider).audience == str(provider._resource_url)
assert str(jwt_verifier(provider).audience).rstrip("/") == BASE_URL
class TestCustomVerifier:
def test_custom_verifier_is_used_as_is(self):
custom = JWTVerifier(
jwks_uri="https://elsewhere.example.com/keys",
issuer=ISSUER,
audience="https://pinned.example.com/mcp",
)
provider = make_provider(token_verifier=custom)
assert provider.token_verifier is custom
def test_custom_verifier_audience_is_never_rewritten(self):
custom = JWTVerifier(
jwks_uri="https://elsewhere.example.com/keys",
issuer=ISSUER,
audience="https://pinned.example.com/mcp",
)
provider = make_provider(token_verifier=custom)
provider.get_routes(mcp_path="/mcp")
assert custom.audience == "https://pinned.example.com/mcp"
class TestProtectedResourceMetadata:
def test_prm_route_is_registered_for_the_mcp_path(self):
provider = make_provider()
paths = [route.path for route in provider.get_routes(mcp_path="/mcp")]
assert "/.well-known/oauth-protected-resource/mcp" in paths
def test_well_known_routes_are_a_subset_of_all_routes(self):
provider = make_provider()
well_known = provider.get_well_known_routes(mcp_path="/mcp")
assert well_known
assert all(route.path.startswith("/.well-known/") for route in well_known)