From 196b9598e63acdfd3254863e02c81d3a58fb61ac Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Tue, 11 Aug 2026 15:39:56 +0000 Subject: [PATCH] Let a token verifier gate the server without AuthSettings `MCPServer(token_verifier=...)` no longer needs `auth=AuthSettings(...)`. On its own a verifier is now a plain bearer gate: requests without a token it accepts get a 401 whose `WWW-Authenticate` carries no `resource_metadata`, no protected-resource metadata route is published, and `get_access_token()` works as before. `AuthSettings` keeps its job of describing that gate to OAuth clients (required scopes, RFC 9728 metadata, the discovery pointer in the 401), so it is what you add when a real authorization server issues the tokens. Previously the constructor refused a verifier without settings, which forced anyone with a pre-shared token to invent an issuer URL, and the low-level `Server.streamable_http_app(token_verifier=...)` accepted the same shape but answered every request 401, valid token included, because the authentication backend was only installed when settings were given. Both wiring sites (and `MCPServer.sse_app`) now install the backend whenever a verifier is present. The authorization docs gain a "Just a pre-shared token" section with a runnable example, and the constructor still refuses the two shapes that cannot work: settings with nothing to gate with, and an embedded authorization-server provider without settings for its issuer. --- docs/run/authorization.md | 33 +++++++- docs_src/authorization/tutorial003.py | 27 +++++++ src/mcp/server/lowlevel/server.py | 75 +++++++++---------- src/mcp/server/mcpserver/server.py | 72 +++++++----------- tests/docs_src/test_authorization.py | 44 +++++++++-- tests/interaction/_requirements.py | 13 ++++ tests/interaction/auth/test_bearer.py | 26 +++++++ .../mcpserver/auth/test_auth_integration.py | 20 ++++- tests/server/mcpserver/test_server.py | 60 ++++++++++++++- 9 files changed, 276 insertions(+), 94 deletions(-) create mode 100644 docs_src/authorization/tutorial003.py diff --git a/docs/run/authorization.md b/docs/run/authorization.md index b7d731b1e2..ed9f644901 100644 --- a/docs/run/authorization.md +++ b/docs/run/authorization.md @@ -1,6 +1,6 @@ # Authorization -Over Streamable HTTP your MCP server is an ordinary web service, and you protect it the way you protect any web service: with OAuth 2.1 bearer tokens. +Over Streamable HTTP your MCP server is an ordinary web service, and you protect it the way you protect any web service: with bearer tokens. Most of this page is the OAuth 2.1 shape, where an authorization server issues them; **[Just a pre-shared token](#just-a-pre-shared-token)** at the end is the smaller case where you hand one out yourself. In OAuth terms, your server is a **resource server**. It never signs anyone in and it never issues a token. It does one thing: look at the `Authorization` header on each request and decide whether the token in it is good. @@ -24,7 +24,7 @@ The SDK has no opinion about what a valid token looks like. You tell it, by impl * `TokenVerifier` is a protocol with one async method. `verify_token` gets the raw token from the `Authorization` header and returns an **`AccessToken`** if it's valid, `None` if it isn't. There is nothing else to implement. * This one looks the token up in a table. A real one verifies a JWT signature or calls the authorization server's token-introspection endpoint. That code is yours; the SDK only calls it. -* `token_verifier=` and `auth=` always travel together. Pass one without the other and `MCPServer(...)` raises a `ValueError` before it ever serves a request. +* `token_verifier=` is the gate. `auth=` is what the server *publishes* about that gate, plus the scopes it insists on, so it is meaningless alone: pass `auth=` without a verifier and `MCPServer(...)` raises a `ValueError` before it ever serves a request. The reverse, a verifier with no `auth=`, is legitimate and smaller: **[Just a pre-shared token](#just-a-pre-shared-token)**. `AuthSettings` is the public face of your resource server: @@ -113,12 +113,37 @@ To watch all three parties move, run `examples/servers/simple-auth/` from the SD An authorization server can also accept an enterprise identity provider's signed assertion in place of a user clicking through a consent screen, and the SDK supports both sides of that exchange. The grant, and the client that presents it, is **[Identity assertion](../client/identity-assertion.md)**. +## Just a pre-shared token + +Sometimes there is no authorization server anywhere: you minted a token yourself, handed it to the one client that needs it, and all the server has to do is check it. Keep the verifier and drop `auth=`: + +```python title="server.py" hl_lines="8 13-15 18" +--8<-- "docs_src/authorization/tutorial003.py" +``` + +* No `AuthSettings` means nothing is advertised. The app has the one `/mcp` route and no `/.well-known/oauth-protected-resource/mcp`, and the 401 loses its `resource_metadata` pointer. The gate itself is the same, and so is `get_access_token()`. +* With nothing to discover, the client must arrive already holding the token. For the python `Client` that is an `Authorization` header on the `httpx2.AsyncClient` you hand to `streamable_http_client` (**[Client transports](../client/transports.md#bring-your-own-httpx2asyncclient)** has it); for a host, it is wherever that host's server entry takes request headers, usually a `headers` block. An OAuth-capable client that turns up without the token gets the 401 and has nowhere to go from there. +* A pre-shared token is a password. Compare it with `secrets.compare_digest`, keep it in the environment and out of the source (unset, this server mints a random one at startup, so a missing variable locks the door rather than opening it), and put TLS in front of anything that is not localhost. + +!!! check + Call `/mcp` with no token and the door is exactly as shut: + + ```text + HTTP/1.1 401 Unauthorized + WWW-Authenticate: Bearer error="invalid_token", error_description="Authentication required" + + {"error": "invalid_token", "error_description": "Authentication required"} + ``` + + The same refusal as before, minus the `resource_metadata` that would have sent a client looking + for an authorization server you don't have. + ## Recap * Over Streamable HTTP your server is an OAuth 2.1 **resource server**: it verifies tokens, it never issues them. * `TokenVerifier` is the whole integration surface: one async method, token in, `AccessToken | None` out. -* `token_verifier=` and `auth=AuthSettings(issuer_url=..., resource_server_url=..., required_scopes=[...])` always travel together. -* The SDK publishes [RFC 9728](https://datatracker.ietf.org/doc/html/rfc9728) Protected Resource Metadata at `/.well-known/oauth-protected-resource/...` and answers unauthenticated requests with a 401 whose `WWW-Authenticate` header points at it. That is the entire discovery story. +* `token_verifier=` alone is a complete gate, and the right one for a token you hand out yourself. Add `auth=AuthSettings(issuer_url=..., resource_server_url=..., required_scopes=[...])` when a real authorization server issues the tokens. +* With `AuthSettings`, the SDK publishes [RFC 9728](https://datatracker.ietf.org/doc/html/rfc9728) Protected Resource Metadata at `/.well-known/oauth-protected-resource/...` and answers unauthenticated requests with a 401 whose `WWW-Authenticate` header points at it. That is the entire discovery story. * `get_access_token()` in any handler is who's calling. * Authorization is an HTTP concern. `stdio` and the in-memory client never see it. diff --git a/docs_src/authorization/tutorial003.py b/docs_src/authorization/tutorial003.py new file mode 100644 index 0000000000..641cb61d5e --- /dev/null +++ b/docs_src/authorization/tutorial003.py @@ -0,0 +1,27 @@ +import os +import secrets + +from mcp.server import MCPServer +from mcp.server.auth.middleware.auth_context import get_access_token +from mcp.server.auth.provider import AccessToken, TokenVerifier + +API_TOKEN = os.environ.get("NOTES_API_TOKEN") or secrets.token_urlsafe(32) + + +class PresharedTokenVerifier(TokenVerifier): + async def verify_token(self, token: str) -> AccessToken | None: + if secrets.compare_digest(token.encode(), API_TOKEN.encode()): + return AccessToken(token=token, client_id="notes-client", scopes=[]) + return None + + +mcp = MCPServer("Notes", token_verifier=PresharedTokenVerifier()) + + +@mcp.tool() +def whoami() -> str: + """Report which client is calling.""" + token = get_access_token() + if token is None: + return "anonymous" + return token.client_id diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py index efdf4b216e..23d0bb51d5 100644 --- a/src/mcp/server/lowlevel/server.py +++ b/src/mcp/server/lowlevel/server.py @@ -734,7 +734,18 @@ def streamable_http_app( custom_starlette_routes: list[Route] | None = None, debug: bool = False, ) -> Starlette: - """Return an instance of the StreamableHTTP server app.""" + """Return an instance of the StreamableHTTP server app. + + `token_verifier` is the bearer gate: with one, every request to the MCP + endpoint must carry an `Authorization: Bearer` token the verifier + accepts, and anything else is answered 401. `auth` describes that gate + to clients: its `required_scopes` are enforced, and when + `resource_server_url` is set the app serves RFC 9728 protected-resource + metadata and points the 401 challenge at it. Without a verifier nothing + is gated. `auth_server_provider` (with `auth`) additionally mounts the + SDK's authorization-server routes, advertised with `auth.issuer_url` + as the issuer. + """ # Auto-enable DNS rebinding protection for localhost (IPv4 and IPv6) if transport_security is None and host in ("127.0.0.1", "localhost", "::1"): transport_security = TransportSecuritySettings( @@ -760,43 +771,33 @@ def streamable_http_app( # Create routes routes: list[Route | Mount] = [] middleware: list[Middleware] = [] - required_scopes: list[str] = [] - - # Set up auth if configured - if auth: - required_scopes = auth.required_scopes or [] - - # Add auth middleware if token verifier is available - if token_verifier: - middleware = [ - Middleware( - AuthenticationMiddleware, - backend=BearerAuthBackend(token_verifier), - ), - Middleware(AuthContextMiddleware), - ] - - # Add auth endpoints if auth server provider is configured - if auth_server_provider: - routes.extend( - create_auth_routes( - provider=auth_server_provider, - issuer_url=auth.issuer_url, - service_documentation_url=auth.service_documentation_url, - client_registration_options=auth.client_registration_options, - revocation_options=auth.revocation_options, - identity_assertion_enabled=auth.identity_assertion_enabled, - ) + + # Embedded authorization server (the legacy all-in-one shape) + if auth and auth_server_provider: + routes.extend( + create_auth_routes( + provider=auth_server_provider, + issuer_url=auth.issuer_url, + service_documentation_url=auth.service_documentation_url, + client_registration_options=auth.client_registration_options, + revocation_options=auth.revocation_options, + identity_assertion_enabled=auth.identity_assertion_enabled, ) + ) - # Set up routes with or without auth + # A token verifier is the bearer gate: authenticate every request and + # refuse the MCP endpoint to anything the verifier does not accept. + # `auth` only adds to that: required scopes, and the RFC 9728 metadata + # URL the 401 challenge points at. if token_verifier: - # Determine resource metadata URL + middleware = [ + Middleware(AuthenticationMiddleware, backend=BearerAuthBackend(token_verifier)), + Middleware(AuthContextMiddleware), + ] + required_scopes = (auth.required_scopes if auth else None) or [] resource_metadata_url = None - if auth and auth.resource_server_url: # pragma: no branch - # Build compliant metadata URL for WWW-Authenticate header + if auth and auth.resource_server_url: resource_metadata_url = build_resource_metadata_url(auth.resource_server_url) - routes.append( Route( streamable_http_path, @@ -804,13 +805,7 @@ def streamable_http_app( ) ) else: - # Auth is disabled, no wrapper needed - routes.append( - Route( - streamable_http_path, - endpoint=streamable_http_app, - ) - ) + routes.append(Route(streamable_http_path, endpoint=streamable_http_app)) # Add protected resource metadata endpoint if configured as RS if auth and auth.resource_server_url: diff --git a/src/mcp/server/mcpserver/server.py b/src/mcp/server/mcpserver/server.py index bc79c44a36..ae2f1e15f9 100644 --- a/src/mcp/server/mcpserver/server.py +++ b/src/mcp/server/mcpserver/server.py @@ -57,6 +57,7 @@ from mcp.server.auth.middleware.auth_context import AuthContextMiddleware from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware from mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier, TokenVerifier +from mcp.server.auth.routes import build_resource_metadata_url, create_auth_routes, create_protected_resource_routes from mcp.server.auth.settings import AuthSettings from mcp.server.caching import CacheableMethod, CacheHint from mcp.server.context import HandlerResult, ServerMiddleware, ServerRequestContext @@ -232,14 +233,16 @@ def __init__( # User middleware runs inside the SDK's built-ins (OpenTelemetry, then the # request-state boundary), outermost-first in the order given. self._lowlevel_server.middleware.extend(middleware or ()) - # Validate auth configuration + # Validate auth configuration. A token_verifier on its own is a plain + # bearer gate; `auth` is what publishes metadata about it, so it needs + # something to gate with, and an embedded AS needs `auth` for its issuer. if self.settings.auth is not None: - if auth_server_provider and token_verifier: # pragma: no cover + if auth_server_provider and token_verifier: raise ValueError("Cannot specify both auth_server_provider and token_verifier") - if not auth_server_provider and not token_verifier: # pragma: no cover - raise ValueError("Must specify either auth_server_provider or token_verifier when auth is enabled") - elif auth_server_provider or token_verifier: - raise ValueError("Cannot specify auth_server_provider or token_verifier without auth settings") + if not auth_server_provider and not token_verifier: + raise ValueError("Must specify either auth_server_provider or token_verifier with auth settings") + elif auth_server_provider: + raise ValueError("Cannot specify auth_server_provider without auth settings") self._auth_server_provider = auth_server_provider self._token_verifier = token_verifier @@ -1121,45 +1124,30 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send): # pragma: no middleware: list[Middleware] = [] required_scopes: list[str] = [] - # Set up auth if configured - if self.settings.auth: # pragma: no cover - required_scopes = self.settings.auth.required_scopes or [] - - # Add auth middleware if token verifier is available - if self._token_verifier: - middleware = [ - # extract auth info from request (but do not require it) - Middleware( - AuthenticationMiddleware, - backend=BearerAuthBackend(self._token_verifier), - ), - # Add the auth context middleware to store - # authenticated user in a contextvar - Middleware(AuthContextMiddleware), - ] - - # Add auth endpoints if auth server provider is configured - if self._auth_server_provider: - from mcp.server.auth.routes import create_auth_routes - - routes.extend( - create_auth_routes( - provider=self._auth_server_provider, - issuer_url=self.settings.auth.issuer_url, - service_documentation_url=self.settings.auth.service_documentation_url, - client_registration_options=self.settings.auth.client_registration_options, - revocation_options=self.settings.auth.revocation_options, - identity_assertion_enabled=self.settings.auth.identity_assertion_enabled, - ) + # Add auth endpoints if auth server provider is configured + if self.settings.auth and self._auth_server_provider: # pragma: no cover + routes.extend( + create_auth_routes( + provider=self._auth_server_provider, + issuer_url=self.settings.auth.issuer_url, + service_documentation_url=self.settings.auth.service_documentation_url, + client_registration_options=self.settings.auth.client_registration_options, + revocation_options=self.settings.auth.revocation_options, + identity_assertion_enabled=self.settings.auth.identity_assertion_enabled, ) + ) - # When auth is configured, require authentication - if self._token_verifier: # pragma: no cover + # A token verifier is the bearer gate (see Server.streamable_http_app) + if self._token_verifier: + middleware = [ + Middleware(AuthenticationMiddleware, backend=BearerAuthBackend(self._token_verifier)), + Middleware(AuthContextMiddleware), + ] + if self.settings.auth: + required_scopes = self.settings.auth.required_scopes or [] # Determine resource metadata URL resource_metadata_url = None if self.settings.auth and self.settings.auth.resource_server_url: - from mcp.server.auth.routes import build_resource_metadata_url - # Build compliant metadata URL for WWW-Authenticate header resource_metadata_url = build_resource_metadata_url(self.settings.auth.resource_server_url) @@ -1198,9 +1186,7 @@ async def sse_endpoint(request: Request) -> Response: # pragma: no cover ) ) # Add protected resource metadata endpoint if configured as RS - if self.settings.auth and self.settings.auth.resource_server_url: # pragma: no cover - from mcp.server.auth.routes import create_protected_resource_routes - + if self.settings.auth and self.settings.auth.resource_server_url: routes.extend( create_protected_resource_routes( resource_url=self.settings.auth.resource_server_url, diff --git a/tests/docs_src/test_authorization.py b/tests/docs_src/test_authorization.py index 00c9adc81c..9ebf8b2d3b 100644 --- a/tests/docs_src/test_authorization.py +++ b/tests/docs_src/test_authorization.py @@ -6,7 +6,7 @@ from mcp_types import TextContent from starlette.routing import Route -from docs_src.authorization import tutorial001, tutorial002 +from docs_src.authorization import tutorial001, tutorial002, tutorial003 from mcp import Client from mcp.client.streamable_http import streamable_http_client from mcp.server import MCPServer @@ -23,10 +23,11 @@ async def test_the_in_memory_client_never_authenticates() -> None: assert result.structured_content == {"result": ["Buy milk", "Ship the release"]} -async def test_token_verifier_and_auth_settings_must_travel_together() -> None: - """tutorial001: passing `token_verifier=` without `auth=` is refused at construction time.""" - with pytest.raises(ValueError, match="Cannot specify auth_server_provider or token_verifier without auth settings"): - MCPServer("Notes", token_verifier=tutorial001.StaticTokenVerifier()) +async def test_auth_settings_without_a_verifier_are_refused_at_construction() -> None: + """tutorial001: `auth=` publishes metadata about a gate, so passing it with nothing to gate with is refused.""" + with pytest.raises(ValueError) as exc_info: + MCPServer("Notes", auth=tutorial001.mcp.settings.auth) + assert str(exc_info.value) == "Must specify either auth_server_provider or token_verifier with auth settings" async def test_the_app_grows_a_protected_resource_metadata_route() -> None: @@ -96,3 +97,36 @@ async def test_get_access_token_is_the_callers_access_token() -> None: result = await client.call_tool("whoami", {}) assert result.content == [TextContent(type="text", text="alice (scopes: notes:read)")] assert result.structured_content == {"result": "alice (scopes: notes:read)"} + + +async def test_a_verifier_alone_gates_the_endpoint_and_publishes_nothing() -> None: + """tutorial003: no `auth=` means one `/mcp` route, no well-known route, and a 401 without `resource_metadata`.""" + app = tutorial003.mcp.streamable_http_app() + [mcp_route] = app.routes + assert isinstance(mcp_route, Route) + assert mcp_route.path == "/mcp" + + transport = httpx2.ASGITransport(app=app) + async with httpx2.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as http_client: + unauthenticated = await http_client.post("/mcp", json={}) + metadata = await http_client.get("/.well-known/oauth-protected-resource/mcp") + assert unauthenticated.status_code == 401 + assert unauthenticated.json() == {"error": "invalid_token", "error_description": "Authentication required"} + assert unauthenticated.headers["www-authenticate"] == ( + 'Bearer error="invalid_token", error_description="Authentication required"' + ) + assert metadata.status_code == 404 + + +async def test_a_pre_shared_token_reaches_the_tool() -> None: + """tutorial003: the client that arrives holding the token gets through the gate, and `get_access_token()` is it.""" + url = "http://127.0.0.1:8000/mcp" + transport = httpx2.ASGITransport(app=tutorial003.mcp.streamable_http_app()) + headers = {"Authorization": f"Bearer {tutorial003.API_TOKEN}"} + async with tutorial003.mcp.session_manager.run(): + async with ( + httpx2.AsyncClient(transport=transport, base_url=url, headers=headers) as http_client, + Client(streamable_http_client(url, http_client=http_client)) as client, + ): + result = await client.call_tool("whoami", {}) + assert result.structured_content == {"result": "notes-client"} diff --git a/tests/interaction/_requirements.py b/tests/interaction/_requirements.py index 964a1829d2..d5d617a27a 100644 --- a/tests/interaction/_requirements.py +++ b/tests/interaction/_requirements.py @@ -2933,6 +2933,19 @@ def __post_init__(self) -> None: ), ), ), + "hosting:auth:verifier-only": Requirement( + source="sdk", + behavior=( + "A token verifier with no AuthSettings is a plain bearer gate: a request without a token the " + "verifier accepts is answered 401 with a WWW-Authenticate challenge carrying no resource_metadata, " + "a valid token reaches the MCP endpoint, and no protected-resource metadata route is published." + ), + transports=("streamable-http",), + note=( + "Auth is enforced at the HTTP layer; the gate and its challenge are HTTP. Deliberately outside the " + "spec's OAuth profile (no RFC 9728 metadata to publish), matching the other SDKs' verifier-only helpers." + ), + ), "hosting:auth:as:authorize-requires-pkce": Requirement( source=f"{SPEC_BASE_URL}/basic/authorization#authorization-code-protection", behavior=( diff --git a/tests/interaction/auth/test_bearer.py b/tests/interaction/auth/test_bearer.py index c70a27c52e..783952880b 100644 --- a/tests/interaction/auth/test_bearer.py +++ b/tests/interaction/auth/test_bearer.py @@ -187,3 +187,29 @@ async def test_an_access_token_in_the_query_string_is_not_accepted(protected: ht assert response.status_code == 401 assert parse_www_authenticate(response.headers["www-authenticate"])["error"] == "invalid_token" + + +@requirement("hosting:auth:verifier-only") +async def test_a_verifier_without_auth_settings_is_a_plain_bearer_gate() -> None: + """`token_verifier=` with no `auth=` still gates `/mcp`, but advertises nothing. + + The challenge carries no `resource_metadata` (there is no metadata document to point at, and no + route answers where one would live), and a token the verifier accepts reaches the MCP endpoint. + This is the pre-shared-token shape: the same gate as `protected`, minus everything + `AuthSettings` would publish about it. + """ + server = Server("rs") + async with mounted_app(server, token_verifier=StaticTokenVerifier(TOKENS)) as (http, _): + unauthenticated = await post_mcp(http) + metadata = await http.get("/.well-known/oauth-protected-resource/mcp") + authenticated = await post_mcp(http, bearer="tok-valid") + + assert unauthenticated.status_code == 401 + assert parse_www_authenticate(unauthenticated.headers["www-authenticate"]) == { + "error": "invalid_token", + "error_description": "Authentication required", + } + assert metadata.status_code == 404 + assert authenticated.status_code == 200 + [data] = [line.removeprefix("data: ") for line in authenticated.text.splitlines() if line.startswith("data: ")] + assert "protocolVersion" in JSONRPCResponse.model_validate_json(data).result diff --git a/tests/server/mcpserver/auth/test_auth_integration.py b/tests/server/mcpserver/auth/test_auth_integration.py index e9c1df8465..84f7cb9d50 100644 --- a/tests/server/mcpserver/auth/test_auth_integration.py +++ b/tests/server/mcpserver/auth/test_auth_integration.py @@ -18,11 +18,13 @@ AuthorizationCode, AuthorizationParams, OAuthAuthorizationServerProvider, + ProviderTokenVerifier, RefreshToken, construct_redirect_uri, ) from mcp.server.auth.routes import create_auth_routes -from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions +from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions, RevocationOptions +from mcp.server.mcpserver import MCPServer from mcp.shared.auth import OAuthClientInformationFull, OAuthToken @@ -310,6 +312,22 @@ async def auth_code( } +def test_auth_server_provider_without_auth_settings_is_refused_at_construction() -> None: + """An embedded authorization server takes its issuer from `auth=`, so the provider alone is refused.""" + with pytest.raises(ValueError) as exc_info: + MCPServer("as", auth_server_provider=MockOAuthProvider()) + assert str(exc_info.value) == "Cannot specify auth_server_provider without auth settings" + + +def test_auth_server_provider_and_token_verifier_together_are_refused_at_construction() -> None: + """An embedded authorization server verifies its own tokens, so a second verifier alongside it is refused.""" + provider = MockOAuthProvider() + settings = AuthSettings(issuer_url=AnyHttpUrl("https://auth.example.com"), resource_server_url=None) + with pytest.raises(ValueError) as exc_info: + MCPServer("as", auth=settings, auth_server_provider=provider, token_verifier=ProviderTokenVerifier(provider)) + assert str(exc_info.value) == "Cannot specify both auth_server_provider and token_verifier" + + class TestAuthEndpoints: @pytest.mark.anyio async def test_metadata_endpoint(self, test_client: httpx2.AsyncClient): diff --git a/tests/server/mcpserver/test_server.py b/tests/server/mcpserver/test_server.py index 48e900dcab..78492f72e7 100644 --- a/tests/server/mcpserver/test_server.py +++ b/tests/server/mcpserver/test_server.py @@ -5,6 +5,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import anyio +import httpx2 import pytest from inline_snapshot import snapshot from mcp_types import ( @@ -41,11 +42,13 @@ TextContent, TextResourceContents, ) -from pydantic import BaseModel +from pydantic import AnyHttpUrl, BaseModel from starlette.applications import Starlette from starlette.routing import Mount, Route from mcp.client import Client +from mcp.server.auth.provider import AccessToken, TokenVerifier +from mcp.server.auth.settings import AuthSettings from mcp.server.context import ServerRequestContext from mcp.server.mcpserver import Context, MCPServer, ResourceSecurity from mcp.server.mcpserver.exceptions import ResourceNotFoundError, ToolError @@ -1799,6 +1802,61 @@ def test_streamable_http_no_redirect() -> None: assert streamable_routes[0].path == "/mcp", "Streamable route path should be /mcp" +class _PresharedTokenVerifier(TokenVerifier): + async def verify_token(self, token: str) -> AccessToken | None: + return AccessToken(token=token, client_id="notes-client", scopes=["notes:read"]) if token == "good" else None + + +_NO_DNS_REBINDING_PROTECTION = TransportSecuritySettings(enable_dns_rebinding_protection=False) + + +async def test_sse_app_with_a_verifier_alone_gates_both_routes_and_publishes_nothing() -> None: + """`token_verifier=` without `auth=` puts `/sse` and `/messages/` behind the bearer gate, with no metadata route.""" + app = MCPServer("test", token_verifier=_PresharedTokenVerifier()).sse_app( + transport_security=_NO_DNS_REBINDING_PROTECTION + ) + async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), base_url="http://127.0.0.1:8000") as http: + sse = await http.get("/sse") + message = await http.post("/messages/", json={}) + authenticated = await http.post("/messages/", json={}, headers={"Authorization": "Bearer good"}) + metadata = await http.get("/.well-known/oauth-protected-resource/sse") + + challenge = 'Bearer error="invalid_token", error_description="Authentication required"' + assert (sse.status_code, sse.headers["www-authenticate"]) == (401, challenge) + assert (message.status_code, message.headers["www-authenticate"]) == (401, challenge) + # Past the gate, the transport itself answers: a POST with no session_id is a 400 from SseServerTransport. + assert (authenticated.status_code, authenticated.text) == (400, "session_id is required") + assert metadata.status_code == 404 + + +async def test_sse_app_with_auth_settings_points_the_challenge_at_its_metadata() -> None: + """With `auth=`, the SSE gate's 401 carries `resource_metadata` and the RFC 9728 document is served.""" + settings = AuthSettings( + issuer_url=AnyHttpUrl("https://auth.example.com"), + resource_server_url=AnyHttpUrl("http://127.0.0.1:8000/sse"), + required_scopes=["notes:read"], + ) + app = MCPServer("test", token_verifier=_PresharedTokenVerifier(), auth=settings).sse_app( + transport_security=_NO_DNS_REBINDING_PROTECTION + ) + async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), base_url="http://127.0.0.1:8000") as http: + sse = await http.get("/sse") + metadata = await http.get("/.well-known/oauth-protected-resource/sse") + + assert sse.status_code == 401 + assert sse.headers["www-authenticate"] == ( + 'Bearer error="invalid_token", error_description="Authentication required", ' + 'resource_metadata="http://127.0.0.1:8000/.well-known/oauth-protected-resource/sse"' + ) + assert metadata.status_code == 200 + assert metadata.json() == { + "resource": "http://127.0.0.1:8000/sse", + "authorization_servers": ["https://auth.example.com/"], + "scopes_supported": ["notes:read"], + "bearer_methods_supported": ["header"], + } + + async def test_report_progress_delegates_to_session_report_progress(): """Context.report_progress delegates to ServerSession.report_progress unconditionally.