diff --git a/docs/docs.json b/docs/docs.json index c52daada0..8200d61c9 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -286,6 +286,7 @@ "pages": [ "integrations/auth0", "integrations/authkit", + "integrations/authplane", "integrations/aws-cognito", "integrations/azure", "integrations/descope", diff --git a/docs/integrations/authplane.mdx b/docs/integrations/authplane.mdx new file mode 100644 index 000000000..cbb28fac7 --- /dev/null +++ b/docs/integrations/authplane.mdx @@ -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" + + + +[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 + + +`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. + + +```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 + + + Base URL of your Authplane authorization server. This is the `iss` claim value + and the root of the RFC 8414 discovery document. + + + + Public URL of this FastMCP server. + + + + Scopes that must be present on every incoming token. Defaults to none, leaving + enforcement to individual tools. + + + + Scopes advertised in Protected Resource Metadata so clients know what to + request. Defaults to `required_scopes`. + + + + Expected `aud` claim. Defaults to this server's resource URL — see + [Audience binding](#audience-binding). + + + + 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. + + + + Public base URL of the protected resource when it differs from `base_url`, + for example behind a reverse proxy. + + + + Custom token verifier. Defaults to a `JWTVerifier` pointed at + `{issuer}/.well-known/jwks.json`. + + +## 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). diff --git a/examples/auth/authplane_oauth/README.md b/examples/auth/authplane_oauth/README.md new file mode 100644 index 000000000..e33964809 --- /dev/null +++ b/examples/auth/authplane_oauth/README.md @@ -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. diff --git a/examples/auth/authplane_oauth/client.py b/examples/auth/authplane_oauth/client.py new file mode 100644 index 000000000..099b9cc5d --- /dev/null +++ b/examples/auth/authplane_oauth/client.py @@ -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()) diff --git a/examples/auth/authplane_oauth/server.py b/examples/auth/authplane_oauth/server.py new file mode 100644 index 000000000..e24e6c2fd --- /dev/null +++ b/examples/auth/authplane_oauth/server.py @@ -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) diff --git a/fastmcp_slim/fastmcp/server/auth/providers/authplane.py b/fastmcp_slim/fastmcp/server/auth/providers/authplane.py new file mode 100644 index 000000000..cf84de4d3 --- /dev/null +++ b/fastmcp_slim/fastmcp/server/auth/providers/authplane.py @@ -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 `_ 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, + ) diff --git a/tests/server/auth/providers/test_authplane.py b/tests/server/auth/providers/test_authplane.py new file mode 100644 index 000000000..8b70f3619 --- /dev/null +++ b/tests/server/auth/providers/test_authplane.py @@ -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)