diff --git a/src/mcp/server/auth/middleware/bearer_auth.py b/src/mcp/server/auth/middleware/bearer_auth.py index 300b29892..26ac21192 100644 --- a/src/mcp/server/auth/middleware/bearer_auth.py +++ b/src/mcp/server/auth/middleware/bearer_auth.py @@ -113,16 +113,26 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: # auth_credentials should always be provided; this is just paranoia if auth_credentials is None or required_scope not in auth_credentials.scopes: await self._send_auth_error( - send, status_code=403, error="insufficient_scope", description=f"Required scope: {required_scope}" + send, + status_code=403, + error="insufficient_scope", + description=f"Required scope: {required_scope}", + # RFC 6750 3.1: on insufficient_scope, the challenge MAY carry the + # scope necessary to access the resource. + scope=" ".join(self.required_scopes), ) return await self.app(scope, receive, send) - async def _send_auth_error(self, send: Send, status_code: int, error: str, description: str) -> None: + async def _send_auth_error( + self, send: Send, status_code: int, error: str, description: str, scope: str | None = None + ) -> None: """Send an authentication error response with WWW-Authenticate header.""" # Build WWW-Authenticate header value www_auth_parts = [f'error="{error}"', f'error_description="{description}"'] + if scope: + www_auth_parts.append(f'scope="{scope}"') if self.resource_metadata_url: # pragma: no cover www_auth_parts.append(f'resource_metadata="{self.resource_metadata_url}"') diff --git a/tests/server/auth/middleware/test_bearer_auth.py b/tests/server/auth/middleware/test_bearer_auth.py index e13ab9639..83c2a5b0b 100644 --- a/tests/server/auth/middleware/test_bearer_auth.py +++ b/tests/server/auth/middleware/test_bearer_auth.py @@ -346,6 +346,54 @@ async def send(message: Message) -> None: assert any(h[0] == b"www-authenticate" for h in sent_messages[0]["headers"]) assert not app.called + async def test_missing_required_scope_includes_scope_in_challenge(self, valid_access_token: AccessToken): + """RFC 6750 3.1: an insufficient_scope challenge MAY include the required scope.""" + app = MockApp() + middleware = RequireAuthMiddleware(app, required_scopes=["admin", "superuser"]) + + # Create a user with read/write scopes but neither required scope + user = AuthenticatedUser(valid_access_token) + auth = AuthCredentials(["read", "write"]) + + scope: Scope = {"type": "http", "user": user, "auth": auth} + + async def receive() -> Message: # pragma: no cover + return {"type": "http.request"} + + sent_messages: list[Message] = [] + + async def send(message: Message) -> None: + sent_messages.append(message) + + await middleware(scope, receive, send) + + assert sent_messages[0]["status"] == 403 + www_authenticate = next(h[1] for h in sent_messages[0]["headers"] if h[0] == b"www-authenticate") + assert www_authenticate == ( + b'Bearer error="insufficient_scope", error_description="Required scope: admin", scope="admin superuser"' + ) + assert not app.called + + async def test_no_user_challenge_omits_scope(self): + """RFC 6750 3.1: scope is specific to insufficient_scope; a bare 401 challenge should not carry it.""" + app = MockApp() + middleware = RequireAuthMiddleware(app, required_scopes=["read"]) + scope: Scope = {"type": "http"} + + async def receive() -> Message: # pragma: no cover + return {"type": "http.request"} + + sent_messages: list[Message] = [] + + async def send(message: Message) -> None: + sent_messages.append(message) + + await middleware(scope, receive, send) + + assert sent_messages[0]["status"] == 401 + www_authenticate = next(h[1] for h in sent_messages[0]["headers"] if h[0] == b"www-authenticate") + assert b"scope=" not in www_authenticate + async def test_no_auth_credentials(self, valid_access_token: AccessToken): """Test middleware with no auth credentials in scope.""" app = MockApp()