-
Notifications
You must be signed in to change notification settings - Fork 3.9k
Let a token verifier gate the server without AuthSettings #3292
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -737,7 +737,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( | ||
|
|
@@ -765,57 +776,41 @@ 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: | ||
|
Comment on lines
+780
to
+781
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟡 (optional) Extended reasoning...At src/mcp/server/lowlevel/server.py:781 the embedded-AS routes are added only under Verification: pre-existing — acknowledged in diff: the PR description states "At the low level, |
||
| 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, | ||
| endpoint=RequireAuthMiddleware(streamable_http_app, required_scopes, resource_metadata_url), | ||
| ) | ||
| ) | ||
| 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)) | ||
|
Comment on lines
812
to
+813
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟣 pre-existing, not blocking: Pre-existing, security-relevant: lowlevel Extended reasoning...Path: Server.streamable_http_app is called with auth=AuthSettings(issuer_url=..., resource_server_url=..., required_scopes=[...]) but no token_verifier (e.g. a lowlevel user who read the MCPServer docs where auth= implies gating, or who wired a provider-less RS and forgot the verifier). Line 797 Verification: pre-existing — acknowledged in diff: the new docstring (src/mcp/server/lowlevel/server.py:747-748) states "Without a verifier nothing is gated", accurately describing the hazard. Triggering condition: a lowlevel caller invokes Server.streamable_http_app(auth=AuthSettings(..., resource_server_url=...), token_verifier=None). Mechanism verified: line 797 |
||
|
|
||
| # Add protected resource metadata endpoint if configured as RS | ||
| if auth and auth.resource_server_url: | ||
|
|
||
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -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 | ||||||
|
|
@@ -242,14 +243,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 | ||||||
|
|
@@ -1177,45 +1180,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: | ||||||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. P1: When a valid Prompt for AI agents
Suggested change
|
||||||
| 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) | ||||||
|
|
||||||
|
|
@@ -1254,9 +1242,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, | ||||||
|
|
||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
P2:
auth=without an explicittoken_verifieris also valid whenauth_server_provider=is supplied;MCPServerderives the verifier from that provider. Qualify this statement to say thatauth=alone, without either a verifier or provider, raises.Prompt for AI agents