From 243f054f657548ae49a7efed5aef0b2a85b52c71 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sat, 18 Jul 2026 20:53:52 -0400 Subject: [PATCH] Include scopes in auth challenges (#4527) --- fastmcp_slim/fastmcp/server/auth/auth.py | 105 +++++++--- .../fastmcp/server/auth/middleware.py | 31 ++- .../fastmcp/server/auth/providers/azure.py | 13 +- fastmcp_slim/fastmcp/server/http.py | 3 + .../auth/providers/test_azure_scopes.py | 9 + .../auth/test_enhanced_error_responses.py | 151 +++++++++++++- tests/server/auth/test_multi_auth.py | 194 ++++++++++++++++++ .../server/auth/test_remote_auth_provider.py | 38 ++++ 8 files changed, 513 insertions(+), 31 deletions(-) diff --git a/fastmcp_slim/fastmcp/server/auth/auth.py b/fastmcp_slim/fastmcp/server/auth/auth.py index 0b3e6abfe..10864a58a 100644 --- a/fastmcp_slim/fastmcp/server/auth/auth.py +++ b/fastmcp_slim/fastmcp/server/auth/auth.py @@ -348,6 +348,26 @@ class AuthProvider(TokenVerifierProtocol): """ raise NotImplementedError("Subclasses must implement verify_token") + @property + def scopes_supported(self) -> list[str]: + """Scopes advertised in protected resource metadata.""" + return self.required_scopes + + @property + def challenge_scopes(self) -> list[str]: + """Scopes clients must request to access this resource.""" + return self.get_challenge_scopes() + + def get_challenge_scopes( + self, required_scopes: list[str] | None = None + ) -> list[str]: + """Translate validation scopes into scopes clients should request. + + Providers whose authorization server uses a different scope format can + override this method to translate any effective set of validation scopes. + """ + return self.required_scopes if required_scopes is None else required_scopes + def set_mcp_path(self, mcp_path: str | None) -> None: """Set the MCP endpoint path and compute resource URL. @@ -485,17 +505,6 @@ class TokenVerifier(AuthProvider): required_scopes=required_scopes, ) - @property - def scopes_supported(self) -> list[str]: - """Scopes to advertise in OAuth metadata. - - Defaults to required_scopes. Override in subclasses when the - advertised scopes differ from the validation scopes (e.g., Azure AD - where tokens contain short-form scopes but clients request full URI - scopes). - """ - return self.required_scopes or [] - async def verify_token(self, token: str) -> AccessToken | None: """Verify a bearer token and return access info if valid.""" raise NotImplementedError("Subclasses must implement verify_token") @@ -525,6 +534,7 @@ class RemoteAuthProvider(AuthProvider): resource_base_url: AnyHttpUrl | str | None = None, resource_name: str | None = None, resource_documentation: AnyHttpUrl | None = None, + challenge_scopes: list[str] | None = None, ): """Initialize the remote auth provider. @@ -542,6 +552,8 @@ class RemoteAuthProvider(AuthProvider): uses the token verifier's scopes_supported property. Use this when the scopes clients request differ from the scopes that appear in tokens (e.g., Azure AD full URI scopes vs short-form). + challenge_scopes: Request-facing form of the required validation scopes. + When omitted, scope translation delegates to the token verifier. resource_name: Optional name for the protected resource resource_documentation: Optional documentation URL for the protected resource """ @@ -553,9 +565,34 @@ class RemoteAuthProvider(AuthProvider): self.token_verifier = token_verifier self.authorization_servers = authorization_servers self._scopes_supported = scopes_supported + self._challenge_scopes = challenge_scopes self.resource_name = resource_name self.resource_documentation = resource_documentation + @property + def scopes_supported(self) -> list[str]: + """Scopes advertised in protected resource metadata.""" + if self._scopes_supported is not None: + return self._scopes_supported + return self.token_verifier.scopes_supported + + def get_challenge_scopes( + self, required_scopes: list[str] | None = None + ) -> list[str]: + """Translate effective validation scopes for the authorization server.""" + effective_scopes = ( + self.required_scopes if required_scopes is None else required_scopes + ) + if ( + effective_scopes == self.required_scopes + and self._challenge_scopes is not None + ): + return self._challenge_scopes + translator = getattr(self.token_verifier, "get_challenge_scopes", None) + if translator is None: + return effective_scopes + return translator(effective_scopes) + async def verify_token(self, token: str) -> AccessToken | None: """Verify token using the configured token verifier.""" return await self.token_verifier.verify_token(token) @@ -585,11 +622,7 @@ class RemoteAuthProvider(AuthProvider): create_protected_resource_routes( resource_url=resource_url, authorization_servers=self.authorization_servers, - scopes_supported=( - self._scopes_supported - if self._scopes_supported is not None - else self.token_verifier.scopes_supported - ), + scopes_supported=self.scopes_supported, resource_name=self.resource_name, resource_documentation=self.resource_documentation, ) @@ -679,6 +712,28 @@ class MultiAuth(AuthProvider): self._sources.append(self.server) self._sources.extend(self.verifiers) + @property + def scopes_supported(self) -> list[str]: + """Scopes advertised by the delegated auth server.""" + if self.server is not None: + return self.server.scopes_supported + return self.required_scopes + + def get_challenge_scopes( + self, required_scopes: list[str] | None = None + ) -> list[str]: + """Translate effective scopes through an unambiguous auth source.""" + effective_scopes = ( + self.required_scopes if required_scopes is None else required_scopes + ) + if self.server is not None: + return self.server.get_challenge_scopes(effective_scopes) + if len(self.verifiers) == 1: + translator = getattr(self.verifiers[0], "get_challenge_scopes", None) + if translator is not None: + return translator(effective_scopes) + return effective_scopes + async def verify_token(self, token: str) -> AccessToken | None: """Verify a token by trying the server, then each verifier in order. @@ -811,6 +866,16 @@ class OAuthProvider( """ return await self.load_access_token(token) + @property + def scopes_supported(self) -> list[str]: + """Scopes advertised by this authorization server.""" + if ( + self.client_registration_options + and self.client_registration_options.valid_scopes + ): + return self.client_registration_options.valid_scopes + return self.required_scopes + def get_routes( self, mcp_path: str | None = None, @@ -872,16 +937,10 @@ class OAuthProvider( # Add protected resource routes if this server is also acting as a resource server if self._resource_url: - supported_scopes = ( - self.client_registration_options.valid_scopes - if self.client_registration_options - and self.client_registration_options.valid_scopes - else self.required_scopes - ) protected_routes = create_protected_resource_routes( resource_url=self._resource_url, authorization_servers=[self.issuer_url], - scopes_supported=supported_scopes, + scopes_supported=self.scopes_supported, ) oauth_routes.extend(protected_routes) diff --git a/fastmcp_slim/fastmcp/server/auth/middleware.py b/fastmcp_slim/fastmcp/server/auth/middleware.py index f0eb82460..3540f9125 100644 --- a/fastmcp_slim/fastmcp/server/auth/middleware.py +++ b/fastmcp_slim/fastmcp/server/auth/middleware.py @@ -11,10 +11,12 @@ authentication (no error attribute) and invalid authentication (with error). from __future__ import annotations import json +from typing import Any from mcp.server.auth.middleware.bearer_auth import ( RequireAuthMiddleware as SDKRequireAuthMiddleware, ) +from pydantic import AnyHttpUrl from starlette.types import Receive, Scope, Send from fastmcp.utilities.logging import get_logger @@ -34,6 +36,18 @@ class RequireAuthMiddleware(SDKRequireAuthMiddleware): (token validation failure). """ + def __init__( + self, + app: Any, + required_scopes: list[str], + resource_metadata_url: AnyHttpUrl | None = None, + challenge_scopes: list[str] | None = None, + ) -> None: + super().__init__(app, required_scopes, resource_metadata_url) + self.challenge_scopes = ( + required_scopes if challenge_scopes is None else challenge_scopes + ) + async def __call__( self, scope: Scope, @@ -86,9 +100,7 @@ class RequireAuthMiddleware(SDKRequireAuthMiddleware): Args: send: ASGI send callable """ - www_auth_parts = [] - if self.resource_metadata_url: - www_auth_parts.append(f'resource_metadata="{self.resource_metadata_url}"') + www_auth_parts = self._challenge_context() www_authenticate = ( ("Bearer " + ", ".join(www_auth_parts)) if www_auth_parts else "Bearer" @@ -110,6 +122,16 @@ class RequireAuthMiddleware(SDKRequireAuthMiddleware): "Missing auth: sent 401 without error attribute (RFC 6750 ยง3.1 compliant)" ) + def _challenge_context(self) -> list[str]: + """Build shared scope and resource metadata challenge parameters.""" + parts = [] + if self.challenge_scopes: + scope_value = " ".join(self.challenge_scopes) + parts.append(f'scope="{scope_value}"') + if self.resource_metadata_url: + parts.append(f'resource_metadata="{self.resource_metadata_url}"') + return parts + async def _send_auth_error( self, send: Send, status_code: int, error: str, description: str ) -> None: @@ -144,8 +166,7 @@ class RequireAuthMiddleware(SDKRequireAuthMiddleware): f'error="{error}"', f'error_description="{enhanced_description}"', ] - if self.resource_metadata_url: - www_auth_parts.append(f'resource_metadata="{self.resource_metadata_url}"') + www_auth_parts.extend(self._challenge_context()) www_authenticate = f"Bearer {', '.join(www_auth_parts)}" diff --git a/fastmcp_slim/fastmcp/server/auth/providers/azure.py b/fastmcp_slim/fastmcp/server/auth/providers/azure.py index 3914573d2..b8d4332e2 100644 --- a/fastmcp_slim/fastmcp/server/auth/providers/azure.py +++ b/fastmcp_slim/fastmcp/server/auth/providers/azure.py @@ -782,10 +782,19 @@ class AzureJWTVerifier(JWTVerifier): property returns the full-URI form for OAuth metadata while ``required_scopes`` retains the short form for token validation. """ - if not self.required_scopes: + return self.get_challenge_scopes() + + def get_challenge_scopes( + self, required_scopes: list[str] | None = None + ) -> list[str]: + """Prefix any effective validation scopes for Azure authorization.""" + effective_scopes = ( + self.required_scopes if required_scopes is None else required_scopes + ) + if not effective_scopes: return [] prefixed = [] - for scope in self.required_scopes: + for scope in effective_scopes: if scope in OIDC_SCOPES or "://" in scope or "/" in scope: prefixed.append(scope) else: diff --git a/fastmcp_slim/fastmcp/server/http.py b/fastmcp_slim/fastmcp/server/http.py index 2421badb4..f196627b7 100644 --- a/fastmcp_slim/fastmcp/server/http.py +++ b/fastmcp_slim/fastmcp/server/http.py @@ -476,6 +476,7 @@ def create_sse_app( handle_sse, auth.required_scopes, resource_metadata_url, + auth.challenge_scopes, ), methods=["GET"], ) @@ -489,6 +490,7 @@ def create_sse_app( sse.handle_post_message, auth.required_scopes, resource_metadata_url, + auth.challenge_scopes, ), ) ) @@ -622,6 +624,7 @@ def create_streamable_http_app( streamable_http_app, auth.required_scopes, resource_metadata_url, + auth.challenge_scopes, ), methods=http_methods, ) diff --git a/tests/server/auth/providers/test_azure_scopes.py b/tests/server/auth/providers/test_azure_scopes.py index 5f1bb6b18..3078468df 100644 --- a/tests/server/auth/providers/test_azure_scopes.py +++ b/tests/server/auth/providers/test_azure_scopes.py @@ -520,6 +520,15 @@ class TestAzureJWTVerifier: "api://my-client-id/write", ] + def test_translates_arbitrary_challenge_scopes(self): + verifier = AzureJWTVerifier( + client_id="my-client-id", + tenant_id="my-tenant-id", + required_scopes=["read"], + ) + + assert verifier.get_challenge_scopes(["admin"]) == ["api://my-client-id/admin"] + def test_already_prefixed_scopes_pass_through(self): verifier = AzureJWTVerifier( client_id="my-client-id", diff --git a/tests/server/auth/test_enhanced_error_responses.py b/tests/server/auth/test_enhanced_error_responses.py index 4d7a2ac34..e75a3dde0 100644 --- a/tests/server/auth/test_enhanced_error_responses.py +++ b/tests/server/auth/test_enhanced_error_responses.py @@ -9,15 +9,30 @@ This test suite covers: import pytest from mcp.shared.auth import OAuthClientInformationFull -from pydantic import AnyUrl +from pydantic import AnyHttpUrl, AnyUrl from starlette.applications import Starlette from starlette.testclient import TestClient from fastmcp import FastMCP +from fastmcp.server.auth import RemoteAuthProvider, TokenVerifier +from fastmcp.server.auth.auth import AccessToken from fastmcp.server.auth.oauth_proxy import OAuthProxy from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair +class _UnderScopedTokenVerifier(TokenVerifier): + def __init__(self, required_scopes: list[str]): + super().__init__(required_scopes=required_scopes) + + async def verify_token(self, token: str) -> AccessToken: + return AccessToken(token=token, client_id="test-client", scopes=["other"]) + + +class _UnderScopedOAuthProxy(OAuthProxy): + async def verify_token(self, token: str) -> AccessToken: + return AccessToken(token=token, client_id="test-client", scopes=["other"]) + + class TestEnhancedAuthorizationHandler: """Tests for enhanced authorization handler error responses.""" @@ -187,6 +202,38 @@ class TestEnhancedAuthorizationHandler: class TestEnhancedRequireAuthMiddleware: """Tests for enhanced authentication middleware error messages.""" + @staticmethod + def create_scoped_app( + required_scopes: list[str], + scopes_supported: list[str], + challenge_scopes: list[str] | None = None, + ) -> Starlette: + auth = RemoteAuthProvider( + token_verifier=_UnderScopedTokenVerifier(required_scopes), + authorization_servers=[AnyHttpUrl("https://auth.example.com")], + base_url="http://localhost:8000", + scopes_supported=scopes_supported, + challenge_scopes=challenge_scopes, + ) + return FastMCP("Test Server", auth=auth).http_app() + + @staticmethod + def create_oauth_app() -> Starlette: + from key_value.aio.stores.memory import MemoryStore + + auth = _UnderScopedOAuthProxy( + upstream_authorization_endpoint="https://auth.example.com/authorize", + upstream_token_endpoint="https://auth.example.com/token", + upstream_client_id="test-client-id", + upstream_client_secret="test-client-secret", + token_verifier=_UnderScopedTokenVerifier(["openid"]), + base_url="http://localhost:8000", + valid_scopes=["openid", "email", "calendar"], + jwt_signing_key="test-secret", + client_storage=MemoryStore(), + ) + return FastMCP("Test Server", auth=auth).http_app() + @pytest.fixture def rsa_key_pair(self) -> RSAKeyPair: """Generate RSA key pair for testing.""" @@ -230,6 +277,108 @@ class TestEnhancedRequireAuthMiddleware: assert "error=" not in www_auth assert response.content == b"" + def test_missing_auth_challenge_includes_supported_scopes(self): + app = self.create_scoped_app( + required_scopes=["read"], + scopes_supported=["api://client-id/read"], + challenge_scopes=["api://client-id/read"], + ) + + with TestClient(app) as client: + response = client.post("/mcp") + + assert response.status_code == 401 + assert response.headers["www-authenticate"] == ( + 'Bearer scope="api://client-id/read", ' + 'resource_metadata="http://localhost:8000/' + '.well-known/oauth-protected-resource/mcp"' + ) + + def test_insufficient_scope_challenge_includes_supported_scopes(self): + app = self.create_scoped_app( + required_scopes=["read"], + scopes_supported=["api://client-id/read"], + challenge_scopes=["api://client-id/read"], + ) + + with TestClient(app) as client: + response = client.post("/mcp", headers={"Authorization": "Bearer narrow"}) + + assert response.status_code == 403 + assert response.headers["www-authenticate"] == ( + 'Bearer error="insufficient_scope", ' + 'error_description="Required scope: read", ' + 'scope="api://client-id/read", ' + 'resource_metadata="http://localhost:8000/' + '.well-known/oauth-protected-resource/mcp"' + ) + + def test_missing_auth_challenge_uses_required_scope_with_empty_catalog(self): + app = self.create_scoped_app(required_scopes=["read"], scopes_supported=[]) + + with TestClient(app) as client: + response = client.post("/mcp") + + assert response.status_code == 401 + assert response.headers["www-authenticate"] == ( + 'Bearer scope="read", resource_metadata="http://localhost:8000/' + '.well-known/oauth-protected-resource/mcp"' + ) + + def test_remote_missing_auth_challenge_excludes_optional_catalog_scopes(self): + app = self.create_scoped_app( + required_scopes=["read"], + scopes_supported=["read", "admin"], + ) + + with TestClient(app) as client: + response = client.post("/mcp") + metadata = client.get("/.well-known/oauth-protected-resource/mcp").json() + + assert response.status_code == 401 + assert 'scope="read"' in response.headers["www-authenticate"] + assert "admin" not in response.headers["www-authenticate"] + assert metadata["scopes_supported"] == ["read", "admin"] + + def test_remote_insufficient_scope_challenge_excludes_optional_catalog_scopes( + self, + ): + app = self.create_scoped_app( + required_scopes=["read"], + scopes_supported=["read", "admin"], + ) + + with TestClient(app) as client: + response = client.post("/mcp", headers={"Authorization": "Bearer narrow"}) + + assert response.status_code == 403 + assert 'scope="read"' in response.headers["www-authenticate"] + assert "admin" not in response.headers["www-authenticate"] + + def test_oauth_missing_auth_challenge_excludes_optional_scopes(self): + app = self.create_oauth_app() + + with TestClient(app) as client: + response = client.post("/mcp") + metadata = client.get("/.well-known/oauth-protected-resource/mcp").json() + + assert response.status_code == 401 + assert 'scope="openid"' in response.headers["www-authenticate"] + assert "email" not in response.headers["www-authenticate"] + assert "calendar" not in response.headers["www-authenticate"] + assert metadata["scopes_supported"] == ["openid", "email", "calendar"] + + def test_oauth_insufficient_scope_challenge_excludes_optional_scopes(self): + app = self.create_oauth_app() + + with TestClient(app) as client: + response = client.post("/mcp", headers={"Authorization": "Bearer narrow"}) + + assert response.status_code == 403 + assert 'scope="openid"' in response.headers["www-authenticate"] + assert "email" not in response.headers["www-authenticate"] + assert "calendar" not in response.headers["www-authenticate"] + def test_invalid_token_enhanced_error_message(self, jwt_verifier): """Test that invalid_token errors have enhanced error messages.""" from fastmcp.server.http import create_streamable_http_app diff --git a/tests/server/auth/test_multi_auth.py b/tests/server/auth/test_multi_auth.py index 08aaa380b..f0d3993cd 100644 --- a/tests/server/auth/test_multi_auth.py +++ b/tests/server/auth/test_multi_auth.py @@ -5,6 +5,7 @@ from pydantic import AnyHttpUrl from fastmcp import FastMCP from fastmcp.server.auth import MultiAuth, RemoteAuthProvider, TokenVerifier from fastmcp.server.auth.auth import AccessToken +from fastmcp.server.auth.providers.azure import AzureJWTVerifier from fastmcp.server.auth.providers.jwt import StaticTokenVerifier @@ -15,6 +16,20 @@ class RaisingVerifier(TokenVerifier): raise RuntimeError("simulated failure") +class UnderScopedVerifier(TokenVerifier): + """A verifier that returns a token missing its required scopes.""" + + async def verify_token(self, token: str) -> AccessToken: + return AccessToken(token=token, client_id="c", scopes=[]) + + +class UnderScopedAzureJWTVerifier(AzureJWTVerifier): + """An Azure verifier that returns a token missing its required scopes.""" + + async def verify_token(self, token: str) -> AccessToken: + return AccessToken(token=token, client_id="c", scopes=[]) + + class TestMultiAuthInit: """Test MultiAuth initialization and validation.""" @@ -107,6 +122,80 @@ class TestMultiAuthInit: auth = MultiAuth(server=provider) assert auth.required_scopes == ["read"] + def test_supported_scopes_from_server(self): + verifier = StaticTokenVerifier( + tokens={"t": {"client_id": "c", "scopes": ["read"]}}, + required_scopes=["read"], + ) + provider = RemoteAuthProvider( + token_verifier=verifier, + authorization_servers=[AnyHttpUrl("https://auth.example.com")], + base_url="https://api.example.com", + scopes_supported=["api://client-id/read"], + challenge_scopes=["api://client-id/read"], + ) + + auth = MultiAuth(server=provider) + + assert auth.required_scopes == ["read"] + assert auth.scopes_supported == ["api://client-id/read"] + assert auth.challenge_scopes == ["api://client-id/read"] + + def test_supported_scopes_from_verifier_only_configuration(self): + verifier = StaticTokenVerifier( + tokens={"t": {"client_id": "c", "scopes": ["read"]}}, + ) + + auth = MultiAuth(verifiers=[verifier], required_scopes=["read"]) + + assert auth.scopes_supported == ["read"] + assert auth.challenge_scopes == ["read"] + + def test_challenge_scopes_translated_by_single_verifier(self): + verifier = AzureJWTVerifier( + client_id="client-id", + tenant_id="test-tenant", + required_scopes=["read"], + ) + + auth = MultiAuth(verifiers=[verifier], required_scopes=["admin"]) + + assert auth.challenge_scopes == ["api://client-id/admin"] + + def test_challenge_scopes_not_translated_by_multiple_verifiers(self): + first = AzureJWTVerifier( + client_id="first-client", + tenant_id="test-tenant", + required_scopes=["read"], + ) + second = AzureJWTVerifier( + client_id="second-client", + tenant_id="test-tenant", + required_scopes=["read"], + ) + + auth = MultiAuth(verifiers=[first, second], required_scopes=["admin"]) + + assert auth.challenge_scopes == ["admin"] + + def test_challenge_scopes_respect_required_scopes_override(self): + verifier = StaticTokenVerifier( + tokens={"t": {"client_id": "c", "scopes": ["read"]}}, + required_scopes=["read"], + ) + provider = RemoteAuthProvider( + token_verifier=verifier, + authorization_servers=[AnyHttpUrl("https://auth.example.com")], + base_url="https://api.example.com", + scopes_supported=["api://client-id/read"], + challenge_scopes=["api://client-id/read"], + ) + + auth = MultiAuth(server=provider, required_scopes=["admin"]) + + assert auth.scopes_supported == ["api://client-id/read"] + assert auth.challenge_scopes == ["admin"] + class TestMultiAuthVerifyToken: """Test MultiAuth token verification chain.""" @@ -381,6 +470,111 @@ class TestMultiAuthIntegration: in response.headers["www-authenticate"] ) + async def test_multi_auth_uses_server_supported_scopes_in_auth_challenges(self): + """Challenges should match the request-facing scopes in delegated metadata.""" + verifier = UnderScopedVerifier(required_scopes=["read"]) + server = RemoteAuthProvider( + token_verifier=verifier, + authorization_servers=[AnyHttpUrl("https://auth.example.com")], + base_url="https://api.example.com", + scopes_supported=["api://client-id/read"], + challenge_scopes=["api://client-id/read"], + ) + + auth = MultiAuth(server=server) + app = FastMCP("test", auth=auth).http_app(path="/mcp") + + async with httpx2.AsyncClient( + transport=httpx2.ASGITransport(app=app), + base_url="https://api.example.com", + ) as client: + missing_response = await client.get("/mcp") + narrow_response = await client.get( + "/mcp", headers={"Authorization": "Bearer narrow"} + ) + + assert missing_response.status_code == 401 + assert ( + 'scope="api://client-id/read"' + in missing_response.headers["www-authenticate"] + ) + assert narrow_response.status_code == 403 + assert ( + 'scope="api://client-id/read"' + in narrow_response.headers["www-authenticate"] + ) + + async def test_multi_auth_scope_override_wins_in_auth_challenges(self): + """Outer overrides are translated for both 401 and 403 challenges.""" + verifier = UnderScopedAzureJWTVerifier( + client_id="client-id", + tenant_id="test-tenant", + required_scopes=["read"], + ) + server = RemoteAuthProvider( + token_verifier=verifier, + authorization_servers=[AnyHttpUrl("https://auth.example.com")], + base_url="https://api.example.com", + ) + + auth = MultiAuth(server=server, required_scopes=["admin"]) + app = FastMCP("test", auth=auth).http_app(path="/mcp") + + async with httpx2.AsyncClient( + transport=httpx2.ASGITransport(app=app), + base_url="https://api.example.com", + ) as client: + missing_response = await client.get("/mcp") + narrow_response = await client.get( + "/mcp", headers={"Authorization": "Bearer narrow"} + ) + + assert missing_response.status_code == 401 + assert ( + 'scope="api://client-id/admin"' + in missing_response.headers["www-authenticate"] + ) + assert ( + "api://client-id/read" not in missing_response.headers["www-authenticate"] + ) + assert narrow_response.status_code == 403 + assert ( + 'scope="api://client-id/admin"' + in narrow_response.headers["www-authenticate"] + ) + assert "api://client-id/read" not in narrow_response.headers["www-authenticate"] + + async def test_verifier_only_scope_translation_in_auth_challenges(self): + """A sole verifier translates challenge scopes for both 401 and 403.""" + verifier = UnderScopedAzureJWTVerifier( + client_id="client-id", + tenant_id="test-tenant", + required_scopes=["read"], + ) + + auth = MultiAuth(verifiers=[verifier], required_scopes=["read"]) + app = FastMCP("test", auth=auth).http_app(path="/mcp") + + async with httpx2.AsyncClient( + transport=httpx2.ASGITransport(app=app), + base_url="https://api.example.com", + ) as client: + missing_response = await client.get("/mcp") + narrow_response = await client.get( + "/mcp", headers={"Authorization": "Bearer narrow"} + ) + + assert missing_response.status_code == 401 + assert ( + 'scope="api://client-id/read"' + in missing_response.headers["www-authenticate"] + ) + assert narrow_response.status_code == 403 + assert ( + 'scope="api://client-id/read"' + in narrow_response.headers["www-authenticate"] + ) + async def test_multi_auth_override_propagates_to_served_metadata(self): """Override on MultiAuth must propagate so served metadata matches the challenge.""" verifier = StaticTokenVerifier(tokens={"t": {"client_id": "c", "scopes": []}}) diff --git a/tests/server/auth/test_remote_auth_provider.py b/tests/server/auth/test_remote_auth_provider.py index c56cd7b2b..c7ef28fab 100644 --- a/tests/server/auth/test_remote_auth_provider.py +++ b/tests/server/auth/test_remote_auth_provider.py @@ -180,6 +180,44 @@ class TestRemoteAuthProvider: "https://api.example.com/mcp" ) + def test_init_preserves_all_legacy_positional_slots(self, test_tokens): + token_verifier = StaticTokenVerifier( + tokens=test_tokens, required_scopes=["read"] + ) + documentation_url = AnyHttpUrl("https://docs.example.com/auth") + + provider = RemoteAuthProvider( + token_verifier, + [AnyHttpUrl("https://auth.example.com")], + "https://auth.example.com/proxy", + ["read"], + "https://api.example.com", + "Example API", + documentation_url, + ) + + assert provider._scopes_supported == ["read"] + assert provider.resource_base_url == AnyHttpUrl("https://api.example.com/") + assert provider.resource_name == "Example API" + assert provider.resource_documentation == documentation_url + assert provider._challenge_scopes is None + + def test_challenge_scope_translation_falls_back_for_protocol_verifier(self): + class ProtocolVerifier: + required_scopes = ["read"] + scopes_supported = ["read", "admin"] + + async def verify_token(self, token: str): + return None + + provider = RemoteAuthProvider( + token_verifier=ProtocolVerifier(), # ty: ignore[invalid-argument-type] + authorization_servers=[AnyHttpUrl("https://auth.example.com")], + base_url="https://api.example.com", + ) + + assert provider.get_challenge_scopes() == ["read"] + class TestRemoteAuthProviderIntegration: """Integration tests for RemoteAuthProvider with FastMCP server."""