fix: add RFC 8707 resource validation to OAuth client

.github/actions/conformance/expected-failures.yml

@@ -1,9 +1,4 @@
 # Known conformance test failures for v1.x
 # These are tracked and should be removed as they're fixed.
-#
-# auth/resource-mismatch: Client must validate that the Protected Resource
-# Metadata (PRM) resource field matches the server URL before proceeding
-# with authorization (RFC 8707). Implemented on main (PR #2010), needs
-# backport to v1.x.
-client:
-  - auth/resource-mismatch
+server: []
+client: []

src/mcp/client/auth/oauth2.py

@@ -267,6 +267,15 @@ def __init__(
         )
         self._initialized = False
 
+    async def _validate_resource_match(self, prm: ProtectedResourceMetadata) -> None:
+        """Validate that PRM resource matches the server URL per RFC 8707."""
+        prm_resource = str(prm.resource) if prm.resource else None
+        if not prm_resource:
+            return  # pragma: no cover
+        default_resource = resource_url_from_server_url(self.context.server_url)
+        if not check_resource_allowed(requested_resource=default_resource, configured_resource=prm_resource):
+            raise OAuthFlowError(f"Protected resource {prm_resource} does not match expected {default_resource}")
+
     async def _handle_protected_resource_response(self, response: httpx.Response) -> bool:
         """
         Handle protected resource metadata discovery response.
@@ -520,6 +529,8 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
 
                         prm = await handle_protected_resource_response(discovery_response)
                         if prm:
+                            # Validate PRM resource matches server URL (RFC 8707)
+                            await self._validate_resource_match(prm)
                             self.context.protected_resource_metadata = prm
 
                             # todo: try all authorization_servers to find the OASM

src/mcp/shared/auth_utils.py

@@ -51,22 +51,17 @@ def check_resource_allowed(requested_resource: str, configured_resource: str) ->
     if requested.scheme.lower() != configured.scheme.lower() or requested.netloc.lower() != configured.netloc.lower():
         return False
 
-    # Handle cases like requested=/foo and configured=/foo/
+    # Normalize trailing slashes before comparison so that
+    # "/foo" and "/foo/" are treated as equivalent.
     requested_path = requested.path
     configured_path = configured.path
-
-    # If requested path is shorter, it cannot be a child
-    if len(requested_path) < len(configured_path):
-        return False
-
-    # Check if the requested path starts with the configured path
-    # Ensure both paths end with / for proper comparison
-    # This ensures that paths like "/api123" don't incorrectly match "/api"
     if not requested_path.endswith("/"):
         requested_path += "/"
     if not configured_path.endswith("/"):
         configured_path += "/"
 
+    # Check hierarchical match: requested must start with configured path.
+    # The trailing-slash normalization ensures "/api123/" won't match "/api/".
     return requested_path.startswith(configured_path)
 
 

tests/client/test_auth.py

@@ -13,6 +13,7 @@
 from pydantic import AnyHttpUrl, AnyUrl
 
 from mcp.client.auth import OAuthClientProvider, PKCEParameters
+from mcp.client.auth.exceptions import OAuthFlowError
 from mcp.client.auth.utils import (
     build_oauth_authorization_server_metadata_discovery_urls,
     build_protected_resource_metadata_discovery_urls,
@@ -965,7 +966,7 @@ async def test_auth_flow_with_no_tokens(self, oauth_provider: OAuthClientProvide
         # Send a successful discovery response with minimal protected resource metadata
         discovery_response = httpx.Response(
             200,
-            content=b'{"resource": "https://api.example.com/mcp", "authorization_servers": ["https://auth.example.com"]}',
+            content=b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}',
             request=discovery_request,
         )
 
@@ -2030,3 +2031,85 @@ async def callback_handler() -> tuple[str, str | None]:
             await auth_flow.asend(final_response)
         except StopAsyncIteration:
             pass
+
+
+@pytest.mark.anyio
+async def test_validate_resource_rejects_mismatched_resource(
+    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
+) -> None:
+    """Client must reject PRM resource that doesn't match server URL."""
+    provider = OAuthClientProvider(
+        server_url="https://api.example.com/v1/mcp",
+        client_metadata=client_metadata,
+        storage=mock_storage,
+    )
+    provider._initialized = True
+
+    prm = ProtectedResourceMetadata(
+        resource=AnyHttpUrl("https://evil.example.com/mcp"),
+        authorization_servers=[AnyHttpUrl("https://auth.example.com")],
+    )
+    with pytest.raises(OAuthFlowError, match="does not match expected"):
+        await provider._validate_resource_match(prm)
+
+
+@pytest.mark.anyio
+async def test_validate_resource_accepts_matching_resource(
+    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
+) -> None:
+    """Client must accept PRM resource that matches server URL."""
+    provider = OAuthClientProvider(
+        server_url="https://api.example.com/v1/mcp",
+        client_metadata=client_metadata,
+        storage=mock_storage,
+    )
+    provider._initialized = True
+
+    prm = ProtectedResourceMetadata(
+        resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
+        authorization_servers=[AnyHttpUrl("https://auth.example.com")],
+    )
+    # Should not raise
+    await provider._validate_resource_match(prm)
+
+
+@pytest.mark.anyio
+async def test_validate_resource_accepts_root_url_with_trailing_slash(
+    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
+) -> None:
+    """Root URLs with trailing slash normalization should match."""
+    provider = OAuthClientProvider(
+        server_url="https://api.example.com/",
+        client_metadata=client_metadata,
+        storage=mock_storage,
+    )
+    provider._initialized = True
+
+    prm = ProtectedResourceMetadata(
+        resource=AnyHttpUrl("https://api.example.com/"),
+        authorization_servers=[AnyHttpUrl("https://auth.example.com")],
+    )
+    # Should not raise - both already have trailing slashes
+    await provider._validate_resource_match(prm)
+
+
+@pytest.mark.anyio
+async def test_get_resource_url_falls_back_when_prm_mismatches(
+    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
+) -> None:
+    """get_resource_url returns canonical URL when PRM resource doesn't match."""
+    provider = OAuthClientProvider(
+        server_url="https://api.example.com/v1/mcp",
+        client_metadata=client_metadata,
+        storage=mock_storage,
+    )
+    provider._initialized = True
+
+    # Set PRM with a resource that is NOT a parent of the server URL
+    provider.context.protected_resource_metadata = ProtectedResourceMetadata(
+        resource=AnyHttpUrl("https://other.example.com/mcp"),
+        authorization_servers=[AnyHttpUrl("https://auth.example.com")],
+    )
+
+    # get_resource_url should return the canonical server URL, not the PRM resource
+    assert provider.context.get_resource_url() == "https://api.example.com/v1/mcp"

tests/shared/test_auth_utils.py

@@ -95,7 +95,7 @@ def test_trailing_slash_handling(self):
         """Trailing slashes should be handled correctly."""
         # With and without trailing slashes
         assert check_resource_allowed("https://example.com/api/", "https://example.com/api") is True
-        assert check_resource_allowed("https://example.com/api", "https://example.com/api/") is False
+        assert check_resource_allowed("https://example.com/api", "https://example.com/api/") is True
         assert check_resource_allowed("https://example.com/api/v1", "https://example.com/api") is True
         assert check_resource_allowed("https://example.com/api/v1", "https://example.com/api/") is True