apply suggested changes

Author: ihrpr

examples/servers/simple-auth/mcp_simple_auth/token_verifier.py

@@ -73,7 +73,7 @@ async def verify_token(self, token: str) -> AccessToken | None:
                     client_id=data.get("client_id", "unknown"),
                     scopes=data.get("scope", "").split() if data.get("scope") else [],
                     expires_at=data.get("exp"),
-                    resource=data.get("aud") or data.get("resource"),  # Include resource in token
+                    resource=data.get("aud"),  # Include resource in token
                 )
             except Exception as e:
                 logger.warning(f"Token introspection failed: {e}")
@@ -82,7 +82,7 @@ async def verify_token(self, token: str) -> AccessToken | None:
     def _validate_resource(self, token_data: dict) -> bool:
         """Validate token was issued for this resource server."""
         if not self.server_url or not self.resource_url:
-            return True  # No validation if server URL not configured
+            return False  # Fail if strict validation requested but URLs missing
 
         # Check 'aud' claim first (standard JWT audience)
         aud = token_data.get("aud")
@@ -94,11 +94,6 @@ def _validate_resource(self, token_data: dict) -> bool:
         elif aud:
             return self._is_valid_resource(aud)
 
-        # Check custom 'resource' claim if no 'aud'
-        resource = token_data.get("resource")
-        if resource:
-            return self._is_valid_resource(resource)
-
         # No resource binding - invalid per RFC 8707
         return False
 

src/mcp/shared/auth_utils.py

@@ -1,6 +1,6 @@
 """Utilities for OAuth 2.0 Resource Indicators (RFC 8707)."""
 
-from urllib.parse import urlparse, urlunparse
+from urllib.parse import urlparse, urlsplit, urlunsplit
 
 from pydantic import AnyUrl, HttpUrl
 
@@ -20,20 +20,9 @@ def resource_url_from_server_url(url: str | HttpUrl | AnyUrl) -> str:
     # Convert to string if needed
     url_str = str(url)
 
-    # Parse the URL
-    parsed = urlparse(url_str)
-
-    # Create canonical form: lowercase scheme and host, no fragment
-    canonical = urlunparse(
-        (
-            parsed.scheme.lower(),  # Lowercase scheme
-            parsed.netloc.lower(),  # Lowercase host (includes port)
-            parsed.path,
-            parsed.params,
-            parsed.query,
-            "",  # Remove fragment
-        )
-    )
+    # Parse the URL and remove fragment, create canonical form
+    parsed = urlsplit(url_str)
+    canonical = urlunsplit(parsed._replace(scheme=parsed.scheme.lower(), netloc=parsed.netloc.lower(), fragment=""))
 
     return canonical
 

tests/client/test_auth.py

@@ -259,6 +259,56 @@ async def test_token_exchange_request(self, oauth_provider):
         assert "code_verifier=test_verifier" in content
         assert "client_id=test_client" in content
         assert "client_secret=test_secret" in content
+        # Resource parameter should be included per RFC 8707
+        assert "resource=https%3A%2F%2Fapi.example.com%2Fv1%2Fmcp" in content
+
+    @pytest.mark.anyio
+    async def test_authorization_url_request(self, oauth_provider):
+        """Test authorization URL construction with resource parameter."""
+        from unittest.mock import patch
+
+        # Set up required context
+        oauth_provider.context.client_info = OAuthClientInformationFull(
+            client_id="test_client",
+            client_secret="test_secret",
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        )
+
+        # Mock the redirect handler to capture the URL
+        captured_url = None
+
+        async def mock_redirect_handler(url: str):
+            nonlocal captured_url
+            captured_url = url
+
+        oauth_provider.context.redirect_handler = mock_redirect_handler
+
+        # Mock callback handler
+        async def mock_callback_handler():
+            return "test_auth_code", "test_state"
+
+        oauth_provider.context.callback_handler = mock_callback_handler
+
+        # Mock pkce and state generation for predictable testing
+        with (
+            patch("mcp.client.auth.PKCEParameters.generate") as mock_pkce,
+            patch("mcp.client.auth.secrets.token_urlsafe") as mock_state,
+        ):
+            mock_pkce.return_value.code_verifier = "test_verifier"
+            mock_pkce.return_value.code_challenge = "test_challenge"
+            mock_state.return_value = "test_state"
+
+            # Mock compare_digest to return True
+            with patch("mcp.client.auth.secrets.compare_digest", return_value=True):
+                await oauth_provider._perform_authorization()
+
+        # Verify the captured URL contains resource parameter
+        assert captured_url is not None
+        assert "resource=https%3A%2F%2Fapi.example.com%2Fv1%2Fmcp" in captured_url
+        assert "client_id=test_client" in captured_url
+        assert "response_type=code" in captured_url
+        assert "code_challenge=test_challenge" in captured_url
+        assert "code_challenge_method=S256" in captured_url
 
     @pytest.mark.anyio
     async def test_refresh_token_request(self, oauth_provider, valid_tokens):
@@ -283,6 +333,8 @@ async def test_refresh_token_request(self, oauth_provider, valid_tokens):
         assert "refresh_token=test_refresh_token" in content
         assert "client_id=test_client" in content
         assert "client_secret=test_secret" in content
+        # Resource parameter should be included per RFC 8707
+        assert "resource=https%3A%2F%2Fapi.example.com%2Fv1%2Fmcp" in content
 
 
 class TestAuthFlow: