Include scopes in auth challenges (#4527)

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",

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

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": []}})

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."""