Add AuthSettings.validate_token_resource to check a bearer token's resource (#3447)

tests/docs_src/test_identity_assertion.py

@@ -151,6 +151,7 @@ async def test_the_whole_grant_is_one_token_request() -> None:
             issuer_url=AnyHttpUrl(tutorial002.ISSUER),
             resource_server_url=AnyHttpUrl(MCP_SERVER_URL),
             required_scopes=["notes:read"],
+            validate_token_resource=True,
         ),
     )
 

tests/interaction/_requirements.py

@@ -2848,11 +2848,11 @@ REQUIREMENTS: dict[str, Requirement] = {
         source=f"{SPEC_BASE_URL}/basic/authorization#access-token-usage",
         behavior="The resource server validates that the token audience matches its resource identifier.",
         transports=("streamable-http",),
-        note="Auth is enforced at the HTTP layer.",
+        note="Auth is enforced at the HTTP layer; the conformant tests enable AuthSettings.validate_token_resource.",
         divergence=Divergence(
             note=(
-                "BearerAuthBackend never inspects AccessToken.resource; a token issued for a different "
-                "resource is accepted. Spec MUST."
+                "Off by default: without AuthSettings.validate_token_resource the bearer gate does not compare "
+                "AccessToken.resource with resource_server_url and the check is the token verifier's. Spec MUST."
             ),
         ),
     ),

tests/interaction/auth/_harness.py

@@ -182,6 +182,7 @@ def auth_settings(
     required_scopes: Sequence[str] = ("mcp",),
     valid_scopes: Sequence[str] | None = None,
     identity_assertion_enabled: bool = False,
+    validate_token_resource: bool = False,
 ) -> AuthSettings:
     """Build `AuthSettings` for the co-hosted authorization + resource server.
 
@@ -194,7 +195,8 @@ def auth_settings(
 
     `identity_assertion_enabled` advertises and accepts the SEP-990 ID-JAG grant (RFC 7523
     jwt-bearer); the provider must implement `exchange_identity_assertion` for the endpoint to
-    issue tokens.
+    issue tokens. `validate_token_resource` makes the bearer gate refuse tokens whose verifier
+    does not report them as issued for the resource URL.
     """
     required = list(required_scopes)
     valid = list(valid_scopes) if valid_scopes is not None else required
@@ -202,6 +204,7 @@ def auth_settings(
         issuer_url=AnyHttpUrl(BASE_URL),
         resource_server_url=AnyHttpUrl(f"{BASE_URL}/mcp"),
         required_scopes=required,
+        validate_token_resource=validate_token_resource,
         client_registration_options=ClientRegistrationOptions(
             enabled=True, valid_scopes=valid, default_scopes=required
         ),

tests/interaction/auth/_provider.py

@@ -157,6 +157,7 @@ class InMemoryAuthorizationServerProvider(
             token=refresh,
             client_id=client.client_id,
             scopes=authorization_code.scopes,
+            resource=authorization_code.resource,
         )
         del self.codes[authorization_code.code]
         return OAuthToken(
@@ -183,9 +184,11 @@ class InMemoryAuthorizationServerProvider(
         if self._fail_next_refresh:
             self._fail_next_refresh = False
             raise TokenError(error="invalid_grant", error_description="refresh denied by harness")
-        access = self.mint_access_token(client_id=client.client_id, scopes=scopes)
+        access = self.mint_access_token(client_id=client.client_id, scopes=scopes, resource=refresh_token.resource)
         new_refresh = f"refresh_{secrets.token_hex(16)}"
-        self.refresh_tokens[new_refresh] = RefreshToken(token=new_refresh, client_id=client.client_id, scopes=scopes)
+        self.refresh_tokens[new_refresh] = RefreshToken(
+            token=new_refresh, client_id=client.client_id, scopes=scopes, resource=refresh_token.resource
+        )
         del self.refresh_tokens[refresh_token.token]
         return OAuthToken(
             access_token=access,

tests/interaction/auth/test_bearer.py

@@ -2,9 +2,9 @@
 
 These tests mount only the resource-server side of the auth wiring (a `StaticTokenVerifier`
 seeded with hand-built tokens, no authorization-server provider) and speak raw HTTP, since
-every assertion is about HTTP semantics the SDK `Client` cannot observe: the 401/403 status,
-the `WWW-Authenticate` header structure, and that a wrong-audience token reaches the MCP
-endpoint behind the gate. The flow side of the same 401 is `test_flow.py`'s flagship test.
+every assertion is about HTTP semantics the SDK `Client` cannot observe: the 401/403 status
+and the `WWW-Authenticate` header structure. The flow side of the same 401 is `test_flow.py`'s
+flagship test.
 """
 
 import time
@@ -24,15 +24,23 @@ from tests.interaction.auth._harness import StaticTokenVerifier, auth_settings
 pytestmark = pytest.mark.anyio
 
 REQUIRED_SCOPE = "mcp:read"
+RESOURCE = "http://127.0.0.1:8000/mcp"
 RESOURCE_METADATA_URL = "http://127.0.0.1:8000/.well-known/oauth-protected-resource/mcp"
 
 _FUTURE = int(time.time()) + 3600
 _PAST = int(time.time()) - 3600
 
+
 TOKENS = {
-    "tok-valid": AccessToken(token="tok-valid", client_id="c", scopes=[REQUIRED_SCOPE], expires_at=_FUTURE),
-    "tok-expired": AccessToken(token="tok-expired", client_id="c", scopes=[REQUIRED_SCOPE], expires_at=_PAST),
-    "tok-noscope": AccessToken(token="tok-noscope", client_id="c", scopes=["other:thing"], expires_at=_FUTURE),
+    "tok-valid": AccessToken(
+        token="tok-valid", client_id="c", scopes=[REQUIRED_SCOPE], expires_at=_FUTURE, resource=RESOURCE
+    ),
+    "tok-expired": AccessToken(
+        token="tok-expired", client_id="c", scopes=[REQUIRED_SCOPE], expires_at=_PAST, resource=RESOURCE
+    ),
+    "tok-noscope": AccessToken(
+        token="tok-noscope", client_id="c", scopes=["other:thing"], expires_at=_FUTURE, resource=RESOURCE
+    ),
     "tok-wrong-aud": AccessToken(
         token="tok-wrong-aud",
         client_id="c",
@@ -40,6 +48,7 @@ TOKENS = {
         expires_at=_FUTURE,
         resource="https://other.example/mcp",
     ),
+    "tok-no-aud": AccessToken(token="tok-no-aud", client_id="c", scopes=[REQUIRED_SCOPE], expires_at=_FUTURE),
 }
 
 
@@ -47,7 +56,7 @@ TOKENS = {
 async def protected() -> AsyncIterator[httpx2.AsyncClient]:
     """A bearer-gated streamable-HTTP app (resource server only) on the in-process bridge."""
     server = Server("rs")
-    settings = auth_settings(required_scopes=[REQUIRED_SCOPE])
+    settings = auth_settings(required_scopes=[REQUIRED_SCOPE], validate_token_resource=True)
     async with mounted_app(server, auth=settings, token_verifier=StaticTokenVerifier(TOKENS)) as (http, _):
         yield http
 
@@ -157,19 +166,38 @@ async def test_a_token_missing_a_required_scope_is_answered_403_insufficient_sco
 
 
 @requirement("hosting:auth:aud-validation")
-async def test_a_token_with_a_mismatched_audience_is_accepted(protected: httpx2.AsyncClient) -> None:
-    """A token whose `resource` does not match the server's resource identifier is accepted.
-
-    The spec mandates the resource server validate the token's audience; the bearer backend
-    never inspects `AccessToken.resource`, so the request passes the gate and the MCP endpoint
-    serves it. This pins current behaviour with the divergence recorded on the requirement.
+@pytest.mark.parametrize("bearer", ["tok-wrong-aud", "tok-no-aud"])
+async def test_a_token_not_issued_for_this_resource_is_answered_401(protected: httpx2.AsyncClient, bearer: str) -> None:
+    """Spec-mandated audience check, which the SDK performs when `AuthSettings.validate_token_resource`
+    is set (off by default, the recorded divergence): a token whose verifier-reported `resource`
+    (RFC 8707) is another URL, or absent, is answered 401 `invalid_token` like an unrecognized token.
     """
-    response = await post_mcp(protected, bearer="tok-wrong-aud")
+    response = await post_mcp(protected, bearer=bearer)
+
+    assert response.status_code == 401
+    assert parse_www_authenticate(response.headers["www-authenticate"])["error"] == "invalid_token"
+
+
+@requirement("hosting:auth:aud-validation")
+async def test_a_token_for_another_resource_is_served_when_validate_token_resource_is_off() -> None:
+    """The recorded divergence: with `AuthSettings` at their defaults the gate does not compare
+    `AccessToken.resource` with `resource_server_url`, so a token the verifier reports as issued
+    for another resource still reaches the MCP endpoint (SDK default; the check is the verifier's)."""
+    settings = auth_settings(required_scopes=[REQUIRED_SCOPE])
+    async with mounted_app(Server("rs"), auth=settings, token_verifier=StaticTokenVerifier(TOKENS)) as (http, _):
+        response = await post_mcp(http, bearer="tok-wrong-aud")
 
     assert response.status_code == 200
-    assert response.headers["content-type"].startswith("text/event-stream")
-    # The body is finite SSE: a result event followed by stream close. Pull the JSON-RPC response
-    # out of the buffered text to prove the MCP endpoint actually answered the initialize request.
+
+
+@requirement("hosting:auth:aud-validation")
+async def test_a_token_issued_for_this_resource_is_served(protected: httpx2.AsyncClient) -> None:
+    """The other half: a token the verifier reports as issued for `resource_server_url` passes the
+    gate and the MCP endpoint answers the initialize request."""
+    response = await post_mcp(protected, bearer="tok-valid")
+
+    assert response.status_code == 200
+    # Finite SSE body: pull out the JSON-RPC result to prove the endpoint actually answered.
     [data] = [line.removeprefix("data: ") for line in response.text.splitlines() if line.startswith("data: ")]
     assert "protocolVersion" in JSONRPCResponse.model_validate_json(data).result
 

tests/interaction/auth/test_lifecycle.py

@@ -135,6 +135,31 @@ async def test_an_expired_access_token_is_transparently_refreshed_before_the_nex
     assert storage.tokens.expires_in == 3600
 
 
+@requirement("client-auth:refresh:transparent")
+async def test_a_refreshed_access_token_is_still_accepted_when_the_server_validates_token_resource() -> None:
+    """With `AuthSettings.validate_token_resource` on, the access token minted by the refresh grant
+    still passes the gate: the provider carries the original grant's resource on `RefreshToken` and
+    onto the refreshed `AccessToken` (SDK-defined propagation), so the request after the refresh is
+    served rather than answered 401."""
+    recorded, on_request = record_requests()
+    provider = InMemoryAuthorizationServerProvider(issue_expired_first=True)
+    server = Server("guarded", on_list_tools=list_tools)
+    settings = auth_settings(validate_token_resource=True)
+
+    with anyio.fail_after(5):
+        async with connect_with_oauth(server, provider=provider, settings=settings, on_request=on_request) as (
+            client,
+            _,
+        ):
+            result = await client.list_tools()
+
+    assert result.tools[0].name == "echo"
+    assert [form_body(r)["grant_type"] for r in find(recorded, "POST", "/token")] == snapshot(
+        ["authorization_code", "refresh_token"]
+    )
+    assert {t.resource for t in provider.access_tokens.values()} == {f"{BASE_URL}/mcp"}
+
+
 @requirement("client-auth:403-scope-upgrade")
 async def test_a_403_insufficient_scope_triggers_one_reauthorize_with_the_challenged_scope() -> None:
     """A 403 `insufficient_scope` challenge is answered by one re-authorize with the challenge's scope.

tests/server/auth/middleware/test_bearer_auth.py

@@ -4,6 +4,7 @@ import time
 from typing import Any, cast
 
 import pytest
+from pydantic import AnyHttpUrl
 from starlette.authentication import AuthCredentials
 from starlette.datastructures import Headers
 from starlette.requests import Request
@@ -273,6 +274,56 @@ class TestBearerAuthBackend:
         assert user.access_token == valid_access_token
 
 
+class SingleTokenVerifier:
+    """A `TokenVerifier` that knows exactly one token."""
+
+    def __init__(self, access_token: AccessToken) -> None:
+        self.access_token = access_token
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        return self.access_token if token == self.access_token.token else None
+
+
+RS = "https://api.example.com/mcp"
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("resource_server_url", "token_resource", "accepted"),
+    [
+        (None, "https://other.example.com/mcp", True),  # nothing configured to compare against
+        (None, None, True),
+        (RS, None, False),  # the verifier did not report what the token was issued for
+        (RS, RS, True),
+        (RS, RS + "/", True),
+        (RS, "https://API.EXAMPLE.COM:443/mcp", True),  # same URL, different spelling
+        (RS, "https://api.example.com", False),
+        (RS, RS + "/child", False),
+        (RS, "https://api.example.com/other", False),
+        (RS, "https://other.example.com/mcp", False),
+        (RS, "api.example.com", False),  # not a URL
+    ],
+)
+async def test_backend_accepts_only_tokens_issued_for_its_resource(
+    resource_server_url: str | None, token_resource: str | None, accepted: bool
+):
+    """With `resource_server_url` set, only a token whose `resource` (RFC 8707) is that URL is
+    accepted and anything else is treated like an unrecognized token (spec-mandated audience
+    check); without it the verifier's answer stands (SDK-defined, the default wiring)."""
+    token = AccessToken(token="t", client_id="c", scopes=["read"], resource=token_resource)
+    backend = BearerAuthBackend(
+        SingleTokenVerifier(token),
+        resource_server_url=AnyHttpUrl(resource_server_url) if resource_server_url else None,
+    )
+
+    result = await backend.authenticate(Request({"type": "http", "headers": [(b"authorization", b"Bearer t")]}))
+
+    if accepted:
+        assert result is not None and result[1].access_token == token
+    else:
+        assert result is None
+
+
 @pytest.mark.anyio
 class TestRequireAuthMiddleware:
     """Tests for the RequireAuthMiddleware class."""

tests/server/auth/test_routes.py

@@ -53,6 +53,7 @@ def test_auth_settings_preserves_path_less_issuer():
     settings = AuthSettings(
         issuer_url="https://as.example.com",  # type: ignore[arg-type]
         resource_server_url="https://rs.example.com",  # type: ignore[arg-type]
+        validate_token_resource=True,
     )
     assert str(settings.issuer_url) == "https://as.example.com"
     assert str(settings.resource_server_url) == "https://rs.example.com"
@@ -63,6 +64,7 @@ def test_build_metadata_serves_issuer_without_trailing_slash():
     settings = AuthSettings(
         issuer_url="https://as.example.com",  # type: ignore[arg-type]
         resource_server_url="https://rs.example.com",  # type: ignore[arg-type]
+        validate_token_resource=True,
     )
     metadata = build_metadata(settings.issuer_url, None, ClientRegistrationOptions(), RevocationOptions())
 

tests/server/auth/test_settings.py

@@ -0,0 +1,34 @@
+import warnings
+
+import pytest
+from pydantic import AnyHttpUrl, ValidationError
+
+from mcp.server.auth.settings import AuthSettings
+from mcp.shared.exceptions import MCPDeprecationWarning
+
+ISSUER = AnyHttpUrl("https://auth.example.com")
+RESOURCE = AnyHttpUrl("https://mcp.example.com/mcp")
+
+
+def test_validate_token_resource_requires_a_resource_server_url():
+    """SDK-defined: asking the bearer gate to compare tokens against `resource_server_url` without
+    configuring one is refused at construction time rather than silently comparing nothing."""
+    AuthSettings(issuer_url=ISSUER, resource_server_url=RESOURCE, validate_token_resource=True)
+    with pytest.raises(ValidationError, match="validate_token_resource requires resource_server_url"):
+        AuthSettings(issuer_url=ISSUER, resource_server_url=None, validate_token_resource=True)
+
+
+def test_leaving_validate_token_resource_unset_warns_when_a_resource_server_url_is_configured():
+    """Unset behaves as False but says so: a resource server that has not chosen gets an
+    `MCPDeprecationWarning` pointing at its own `AuthSettings(...)` call (3.0 flips the default)."""
+    with pytest.warns(MCPDeprecationWarning, match="validate_token_resource") as record:
+        settings = AuthSettings(issuer_url=ISSUER, resource_server_url=RESOURCE)
+    assert settings.validate_token_resource is None
+    assert record[0].filename == __file__
+
+
+@pytest.mark.parametrize("kwargs", [{"validate_token_resource": False}, {"resource_server_url": None}])
+def test_an_explicit_choice_or_no_resource_server_url_does_not_warn(kwargs: dict[str, object]):
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        AuthSettings.model_validate({"issuer_url": ISSUER, "resource_server_url": RESOURCE, **kwargs})