Fix substring matches in WWW-Authenticate parsing

src/mcp/client/auth/utils.py

@@ -16,6 +16,54 @@
 from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER
 
 
+def _split_www_authenticate_segments(header_value: str) -> list[str]:
+    """Split a WWW-Authenticate header on top-level commas."""
+    segments: list[str] = []
+    current: list[str] = []
+    in_quotes = False
+
+    for char in header_value:
+        if char == '"':
+            in_quotes = not in_quotes
+        if char == "," and not in_quotes:
+            segment = "".join(current).strip()
+            if segment:
+                segments.append(segment)
+            current = []
+            continue
+        current.append(char)
+
+    tail = "".join(current).strip()
+    if tail:
+        segments.append(tail)
+    return segments
+
+
+def _extract_bearer_auth_params(www_auth_header: str) -> str | None:
+    """Return the auth-param portion of the first Bearer challenge."""
+    segments = _split_www_authenticate_segments(www_auth_header)
+    collecting = False
+    auth_params: list[str] = []
+
+    for segment in segments:
+        scheme, separator, remainder = segment.partition(" ")
+        if scheme.lower() == "bearer" and separator:
+            if collecting:
+                break
+            collecting = True
+            auth_params = [remainder.strip()]
+            continue
+
+        if collecting:
+            if separator and "=" not in scheme and not remainder.lstrip().startswith("="):
+                break
+            auth_params.append(segment)
+
+    if not auth_params:
+        return None
+    return ", ".join(part for part in auth_params if part)
+
+
 def extract_field_from_www_auth(response: Response, field_name: str) -> str | None:
     """Extract field from WWW-Authenticate header.
 
@@ -26,13 +74,18 @@ def extract_field_from_www_auth(response: Response, field_name: str) -> str | No
     if not www_auth_header:
         return None
 
-    # Pattern matches: field_name="value" or field_name=value (unquoted)
-    pattern = rf'{field_name}=(?:"([^"]+)"|([^\s,]+))'
-    match = re.search(pattern, www_auth_header)
+    auth_params = _extract_bearer_auth_params(www_auth_header)
+    if auth_params is None:
+        return None
 
-    if match:
-        # Return quoted value if present, otherwise unquoted value
-        return match.group(1) or match.group(2)
+    # Match comma-delimited auth-params while respecting quoted values.
+    pattern = re.compile(
+        r'(?:^|,\s*)(?P<name>[A-Za-z][A-Za-z0-9_-]*)\s*=\s*(?:"(?P<quoted>[^"]*)"|(?P<unquoted>[^,\s]+))'
+    )
+    for match in pattern.finditer(auth_params):
+        if match.group("name") == field_name:
+            # Return quoted value if present, otherwise unquoted value
+            return match.group("quoted") or match.group("unquoted")
 
     return None
 

tests/client/test_auth.py

@@ -181,6 +181,39 @@ def test_pkce_uniqueness(self):
         assert pkce1.code_challenge != pkce2.code_challenge
 
 
+class TestExtractFieldFromWwwAuth:
+    def test_uses_first_bearer_challenge_when_multiple_are_present(self):
+        response = httpx.Response(
+            401,
+            headers={
+                "WWW-Authenticate": (
+                    'Basic realm="legacy", Bearer scope="read", error="insufficient_scope", Bearer scope="write"'
+                )
+            },
+            request=httpx.Request("GET", "https://example.com"),
+        )
+
+        assert extract_scope_from_www_auth(response) == "read"
+
+    def test_supports_optional_whitespace_around_equals(self):
+        response = httpx.Response(
+            401,
+            headers={
+                "WWW-Authenticate": (
+                    'Bearer error="insufficient_scope", scope = "read write", '
+                    'resource_metadata = "https://example.com/.well-known/oauth-protected-resource"'
+                )
+            },
+            request=httpx.Request("GET", "https://example.com"),
+        )
+
+        assert extract_scope_from_www_auth(response) == "read write"
+        assert (
+            extract_resource_metadata_from_www_auth(response)
+            == "https://example.com/.well-known/oauth-protected-resource"
+        )
+
+
 class TestOAuthContext:
     """Test OAuth context functionality."""
 
@@ -2016,6 +2049,17 @@ class TestWWWAuthenticate:
                 "resource_metadata",
                 "https://api.example.com/auth/metadata?version=1",
             ),
+            ('Bearer error_scope="decoy", scope="read write"', "scope", "read write"),
+            (
+                'Bearer error_description="missing scope=write permission", scope="read write"',
+                "scope",
+                "read write",
+            ),
+            (
+                'Basic realm="legacy", Bearer scope="read write", error="insufficient_scope"',
+                "scope",
+                "read write",
+            ),
         ],
     )
     def test_extract_field_from_www_auth_valid_cases(
@@ -2047,6 +2091,12 @@ def test_extract_field_from_www_auth_valid_cases(
             # Header without requested field
             ('Bearer realm="api", error="insufficient_scope"', "scope", "no scope parameter"),
             ('Bearer realm="api", scope="read write"', "resource_metadata", "no resource_metadata parameter"),
+            ('Bearer custom_scope="leaked"', "scope", "field name appears only as a substring"),
+            (
+                'Bearer x_resource_metadata="https://decoy.example.com"',
+                "resource_metadata",
+                "field name appears only as a substring",
+            ),
             # Malformed field (empty value)
             ("Bearer scope=", "scope", "malformed scope parameter"),
             ("Bearer resource_metadata=", "resource_metadata", "malformed resource_metadata parameter"),