fix(auth): include RFC 6750 scope attribute in WWW-Authenticate challenges

src/mcp/server/auth/middleware/bearer_auth.py

@@ -114,6 +114,12 @@
         """Send an authentication error response with WWW-Authenticate header."""
         # Build WWW-Authenticate header value
         www_auth_parts = [f'error="{error}"', f'error_description="{description}"']
+        # RFC 6750 section 3: the challenge's `scope` attribute advertises the scope
+        # needed to access the resource (section 3.1: an insufficient_scope response
+        # MAY carry it). Clients read it as the highest-priority scope source, both
+        # for initial authorization (401) and for step-up on 403 insufficient_scope.
+        if self.required_scopes:
+            www_auth_parts.append(f'scope="{" ".join(self.required_scopes)}"')
         if self.resource_metadata_url:
             www_auth_parts.append(f'resource_metadata="{self.resource_metadata_url}"')
 

tests/client/test_auth.py

@@ -6,6 +6,7 @@
 from unittest import mock
 from urllib.parse import parse_qs, quote, unquote, urlparse
 
+import anyio
 import httpx2
 import pytest
 from inline_snapshot import Is, snapshot
@@ -31,8 +32,10 @@
     validate_authorization_response_iss,
     validate_metadata_issuer,
 )
+from mcp.server.auth.provider import AccessToken
 from mcp.server.auth.routes import build_metadata
-from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
+from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions, RevocationOptions
+from mcp.server.lowlevel.server import Server
 from mcp.shared.auth import (
     AuthorizationCodeResult,
     OAuthClientInformationFull,
@@ -1593,6 +1596,101 @@ async def mock_callback() -> AuthorizationCodeResult:
         pass
 
 
+@pytest.mark.anyio
+async def test_403_step_up_consumes_scope_emitted_by_require_auth_middleware(oauth_provider: OAuthClientProvider):
+    """End-to-end #3103 regression: the `scope` attribute the SDK server emits in its
+    insufficient_scope challenge (RFC 6750 section 3.1) is what the client's step-up union
+    consumes, without falling back to protected-resource metadata.
+
+    Steps:
+    1. An SDK server app requiring "read admin" rejects a token granting only "read" with 403.
+    2. The server's real WWW-Authenticate challenge is replayed into the client's auth flow.
+    3. The client re-authorizes with the union of the granted and challenged scopes.
+    """
+
+    class ReadScopedVerifier:
+        """Accepts any token, granting only the "read" scope."""
+
+        async def verify_token(self, token: str) -> AccessToken:
+            return AccessToken(token=token, client_id="test_client_id", scopes=["read"])
+
+    server_app = Server("step-up-repro").streamable_http_app(
+        auth=AuthSettings(
+            issuer_url=AnyHttpUrl("https://auth.example.com"),
+            resource_server_url=AnyHttpUrl("http://127.0.0.1:8000/mcp"),
+            required_scopes=["read", "admin"],
+        ),
+        token_verifier=ReadScopedVerifier(),
+    )
+    transport = httpx2.ASGITransport(app=server_app)
+    async with httpx2.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as http_client:
+        with anyio.fail_after(5):
+            server_response = await http_client.post(
+                "/mcp",
+                json={"jsonrpc": "2.0", "id": 1, "method": "ping"},
+                headers={
+                    "accept": "application/json, text/event-stream",
+                    "authorization": "Bearer read-only-token",
+                },
+            )
+    assert server_response.status_code == 403
+    assert 'scope="read admin"' in server_response.headers["WWW-Authenticate"]
+
+    # Client state: a stored token granted "read"; client_metadata carries no scope, as after a
+    # restart, so the challenge is the only source for the missing "admin" scope.
+    client_info = OAuthClientInformationFull(
+        client_id="test_client_id",
+        client_secret="test_client_secret",
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+    )
+    oauth_provider.context.current_tokens = OAuthToken(access_token="read-only-token", scope="read")
+    oauth_provider.context.token_expiry_time = time.time() + 1800
+    oauth_provider.context.client_info = client_info
+    oauth_provider.context.client_metadata.scope = None
+    oauth_provider._initialized = True
+
+    captured_state: str | None = None
+    reauthorize_scope: str | None = None
+
+    async def capture_redirect(url: str) -> None:
+        nonlocal captured_state, reauthorize_scope
+        params = parse_qs(urlparse(url).query)
+        reauthorize_scope = params["scope"][0]
+        captured_state = params.get("state", [None])[0]
+
+    async def mock_callback() -> AuthorizationCodeResult:
+        return AuthorizationCodeResult(code="auth_code", state=captured_state)
+
+    oauth_provider.context.redirect_handler = capture_redirect
+    oauth_provider.context.callback_handler = mock_callback
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/mcp"))
+    with anyio.fail_after(5):
+        request = await auth_flow.__anext__()
+        response_403 = httpx2.Response(
+            403,
+            headers={"WWW-Authenticate": server_response.headers["WWW-Authenticate"]},
+            request=request,
+        )
+        token_exchange_request = await auth_flow.asend(response_403)
+
+    # SEP-2350: the union of the stored token's grant and the server-advertised requirement
+    assert reauthorize_scope == "read admin"
+
+    # Drive the flow to completion so the context lock is released cleanly
+    token_response = httpx2.Response(
+        200,
+        json={"access_token": "new", "token_type": "Bearer", "expires_in": 3600, "scope": "read admin"},
+        request=token_exchange_request,
+    )
+    with anyio.fail_after(5):
+        final_request = await auth_flow.asend(token_response)
+        try:
+            await auth_flow.asend(httpx2.Response(200, request=final_request))
+        except StopAsyncIteration:
+            pass
+
+
 @pytest.mark.parametrize(
     (
         "issuer_url",

tests/server/auth/middleware/test_bearer_auth.py

@@ -3,9 +3,12 @@
 import time
 from typing import Any, cast
 
+import anyio
+import httpx2
 import pytest
 from starlette.authentication import AuthCredentials
 from starlette.datastructures import Headers
+from starlette.middleware.authentication import AuthenticationMiddleware
 from starlette.requests import Request
 from starlette.types import Message, Receive, Scope, Send
 
@@ -458,6 +461,72 @@ async def send(message: Message) -> None:  # pragma: no cover
         assert app.send == send
 
 
+@pytest.mark.anyio
+async def test_insufficient_scope_challenge_advertises_required_scopes(
+    mock_oauth_provider: OAuthAuthorizationServerProvider[Any, Any, Any], valid_access_token: AccessToken
+):
+    """The 403 insufficient_scope challenge carries a `scope` attribute listing the configured
+    required scopes, per RFC 6750 section 3.1, so clients can step-up (#3103)."""
+    add_token_to_provider(mock_oauth_provider, "valid_token", valid_access_token)
+    inner_app = MockApp()
+    # Production wiring: the authentication middleware populates the connection's user/auth
+    # from the bearer token, then RequireAuthMiddleware enforces the required scopes.
+    app = AuthenticationMiddleware(
+        RequireAuthMiddleware(inner_app, required_scopes=["read", "admin"]),
+        backend=BearerAuthBackend(ProviderTokenVerifier(mock_oauth_provider)),
+    )
+
+    transport = httpx2.ASGITransport(app=app)
+    async with httpx2.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as client:
+        with anyio.fail_after(5):
+            # valid_access_token grants read/write, so the required "admin" scope is missing
+            response = await client.get("/", headers={"Authorization": "Bearer valid_token"})
+
+    assert response.status_code == 403
+    assert response.headers["WWW-Authenticate"] == (
+        'Bearer error="insufficient_scope", error_description="Required scope: admin", scope="read admin"'
+    )
+    assert not inner_app.called
+
+
+@pytest.mark.anyio
+async def test_unauthenticated_challenge_advertises_required_scopes():
+    """The 401 challenge carries a `scope` attribute (RFC 6750 section 3) when required scopes
+    are configured, so clients can request them on initial authorization (#3103)."""
+    inner_app = MockApp()
+    middleware = RequireAuthMiddleware(inner_app, required_scopes=["read", "admin"])
+
+    transport = httpx2.ASGITransport(app=middleware)
+    async with httpx2.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as client:
+        with anyio.fail_after(5):
+            response = await client.get("/")
+
+    assert response.status_code == 401
+    assert response.headers["WWW-Authenticate"] == (
+        'Bearer error="invalid_token", error_description="Authentication required", scope="read admin"'
+    )
+    assert not inner_app.called
+
+
+@pytest.mark.anyio
+async def test_challenge_omits_scope_when_no_scopes_configured():
+    """A challenge from a middleware with no required scopes carries no `scope` attribute —
+    there is nothing to advertise, and RFC 6750 section 3 makes the attribute optional."""
+    inner_app = MockApp()
+    middleware = RequireAuthMiddleware(inner_app, required_scopes=[])
+
+    transport = httpx2.ASGITransport(app=middleware)
+    async with httpx2.AsyncClient(transport=transport, base_url="http://127.0.0.1:8000") as client:
+        with anyio.fail_after(5):
+            response = await client.get("/")
+
+    assert response.status_code == 401
+    assert response.headers["WWW-Authenticate"] == (
+        'Bearer error="invalid_token", error_description="Authentication required"'
+    )
+    assert not inner_app.called
+
+
 def test_authorization_context_is_built_from_principal_components() -> None:
     """Session ownership identifies the principal via the shared principal_components triple."""
     token = AccessToken(