Address WWW-Authenticate parser review feedback

Author: Whning0513

src/mcp/client/auth/utils.py

@@ -21,9 +21,14 @@ def _split_www_authenticate_segments(header_value: str) -> list[str]:
     segments: list[str] = []
     current: list[str] = []
     in_quotes = False
+    escaped = False
 
     for char in header_value:
-        if char == '"':
+        if escaped:
+            escaped = False
+        elif char == "\\" and in_quotes:
+            escaped = True
+        elif char == '"':
             in_quotes = not in_quotes
         if char == "," and not in_quotes:
             segment = "".join(current).strip()
@@ -64,6 +69,35 @@ def _extract_bearer_auth_params(www_auth_header: str) -> str | None:
     return ", ".join(part for part in auth_params if part)
 
 
+_AUTH_PARAM_PATTERN = re.compile(
+    r"(?:^|,\s*)(?P<name>[A-Za-z][A-Za-z0-9_-]*)\s*=\s*"
+    r'(?:"(?P<quoted>(?:\\.|[^"\\])*)"|(?P<unquoted>[^,\s]+))'
+)
+
+
+def _extract_auth_param(auth_params: str, field_name: str) -> str | None:
+    """Extract an auth-param from a comma-delimited parameter string."""
+    for match in _AUTH_PARAM_PATTERN.finditer(auth_params):
+        if match.group("name") != field_name:
+            continue
+        quoted = match.group("quoted")
+        value = quoted if quoted is not None else match.group("unquoted")
+        if not value:
+            return None
+        return re.sub(r"\\(.)", r"\1", value) if quoted is not None else value
+    return None
+
+
+def _extract_field_from_bearer_challenge(response: Response, field_name: str) -> str | None:
+    """Extract an auth-param from the first Bearer challenge."""
+    www_auth_header = response.headers.get("WWW-Authenticate")
+    if not www_auth_header:
+        return None
+
+    auth_params = _extract_bearer_auth_params(www_auth_header)
+    return _extract_auth_param(auth_params, field_name) if auth_params is not None else None
+
+
 def extract_field_from_www_auth(response: Response, field_name: str) -> str | None:
     """Extract field from WWW-Authenticate header.
 
@@ -74,18 +108,11 @@ def extract_field_from_www_auth(response: Response, field_name: str) -> str | No
     if not www_auth_header:
         return None
 
-    auth_params = _extract_bearer_auth_params(www_auth_header)
-    if auth_params is None:
-        return None
-
-    # 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")
+    for segment in _split_www_authenticate_segments(www_auth_header):
+        scheme, separator, remainder = segment.partition(" ")
+        auth_params = remainder.strip() if separator and "=" not in scheme else segment
+        if value := _extract_auth_param(auth_params, field_name):
+            return value
 
     return None
 
@@ -96,7 +123,7 @@ def extract_scope_from_www_auth(response: Response) -> str | None:
     Returns:
         Scope string if found in WWW-Authenticate header, None otherwise
     """
-    return extract_field_from_www_auth(response, "scope")
+    return _extract_field_from_bearer_challenge(response, "scope")
 
 
 def extract_resource_metadata_from_www_auth(response: Response) -> str | None:
@@ -108,7 +135,7 @@ def extract_resource_metadata_from_www_auth(response: Response) -> str | None:
     if not response or response.status_code != 401:
         return None  # pragma: no cover
 
-    return extract_field_from_www_auth(response, "resource_metadata")
+    return _extract_field_from_bearer_challenge(response, "resource_metadata")
 
 
 def build_protected_resource_metadata_discovery_urls(www_auth_url: str | None, server_url: str) -> list[str]:

tests/client/test_auth.py

@@ -181,37 +181,89 @@ 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"),
-        )
+def test_scope_uses_first_bearer_challenge_when_multiple_are_present():
+    """RFC 6750 fields come from the first Bearer challenge, not another scheme or later challenge."""
+    response = httpx.Response(
+        401,
+        headers={
+            "WWW-Authenticate": (
+                'Basic scope="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"
+    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"
-        )
+def test_bearer_fields_support_optional_whitespace_around_equals():
+    """Bearer auth-params allow optional whitespace around the equals sign."""
+    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"
+    )
+
+
+def test_generic_field_lookup_preserves_non_bearer_challenges():
+    """The generic helper keeps its pre-existing ability to read fields from other auth schemes."""
+    response = httpx.Response(
+        401,
+        headers={"WWW-Authenticate": 'Newauth realm="apps"'},
+        request=httpx.Request("GET", "https://example.com"),
+    )
+
+    assert extract_field_from_www_auth(response, "realm") == "apps"
+    assert extract_scope_from_www_auth(response) is None
+
+
+def test_quoted_auth_params_handle_escaped_quotes_and_commas():
+    """RFC quoted-pairs do not terminate a quoted value or split its embedded comma."""
+    response = httpx.Response(
+        401,
+        headers={
+            "WWW-Authenticate": (
+                'Newauth realm="apps \\"beta\\", east", '
+                'Bearer error_description="try \\"again\\", later", scope="read write"'
+            )
+        },
+        request=httpx.Request("GET", "https://example.com"),
+    )
+
+    assert extract_field_from_www_auth(response, "realm") == 'apps "beta", east'
+    assert extract_scope_from_www_auth(response) == "read write"
+
+
+def test_bearer_parser_ignores_empty_segments_and_stops_at_next_challenge():
+    """The first Bearer challenge ends when another authentication scheme begins."""
+    response = httpx.Response(
+        401,
+        headers={"WWW-Authenticate": ', Bearer scope="read", Basic realm="legacy",'},
+        request=httpx.Request("GET", "https://example.com"),
+    )
+
+    assert extract_scope_from_www_auth(response) == "read"
+
+
+def test_empty_quoted_auth_param_returns_none():
+    """An empty quoted auth-param has the same missing-value result as an unquoted empty parameter."""
+    response = httpx.Response(
+        401,
+        headers={"WWW-Authenticate": 'Bearer scope=""'},
+        request=httpx.Request("GET", "https://example.com"),
+    )
+
+    assert extract_scope_from_www_auth(response) is None
 
 
 class TestOAuthContext: