diff --git a/src/mcp/client/auth/utils.py b/src/mcp/client/auth/utils.py index fc87c9e46..d32bdf8a4 100644 --- a/src/mcp/client/auth/utils.py +++ b/src/mcp/client/auth/utils.py @@ -1,4 +1,4 @@ -import re +from collections.abc import Iterator from urllib.parse import urljoin, urlparse from httpx2 import Request, Response @@ -16,6 +16,76 @@ from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER +def _read_quoted_www_auth_value(header: str, value_start: int) -> tuple[str | None, int]: + value_chars: list[str] = [] + escaped = False + index = value_start + 1 + + while index < len(header): + char = header[index] + if escaped: + value_chars.append(char) + escaped = False + elif char == "\\": + escaped = True + elif char == '"': + value = "".join(value_chars) + return (value or None), index + 1 + else: + value_chars.append(char) + index += 1 + + return None, len(header) + + +def _read_www_auth_value(header: str, value_start: int) -> tuple[str | None, int]: + while value_start < len(header) and header[value_start] in " \t": + value_start += 1 + + if value_start >= len(header): + return None, value_start + + if header[value_start] == '"': + return _read_quoted_www_auth_value(header, value_start) + + value_end = value_start + while value_end < len(header) and header[value_end] not in " \t,": + value_end += 1 + + if value_end == value_start: + return None, value_end + + return header[value_start:value_end], value_end + + +def _iter_www_auth_params(header: str) -> Iterator[tuple[str, str]]: + index = 0 + while index < len(header): + while index < len(header) and header[index] in " \t,": + index += 1 + + name_start = index + while index < len(header) and header[index] not in " \t=,": + index += 1 + + if index == name_start: + index += 1 + continue + + name = header[name_start:index] + value_start = index + while value_start < len(header) and header[value_start] in " \t": + value_start += 1 + + if value_start >= len(header) or header[value_start] != "=": + index = value_start + continue + + value, index = _read_www_auth_value(header, value_start + 1) + if value is not None: + yield name, value + + def extract_field_from_www_auth(response: Response, field_name: str) -> str | None: """Extract field from WWW-Authenticate header. @@ -26,13 +96,9 @@ 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) - - if match: - # Return quoted value if present, otherwise unquoted value - return match.group(1) or match.group(2) + for name, value in _iter_www_auth_params(www_auth_header): + if name == field_name: + return value return None diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py index 9e9599f86..649f04c4f 100644 --- a/tests/client/test_auth.py +++ b/tests/client/test_auth.py @@ -2005,12 +2005,30 @@ class TestWWWAuthenticate: ), # Multiple parameters with unquoted value ('Bearer realm="api", scope=basic', "scope", "basic"), + ("Bearer scope= read", "scope", "read"), + ('Bearer =bad, scope="read"', "scope", "read"), + # Decoy parameter name before the real field + ('Bearer error_scope="decoy", scope="read write"', "scope", "read write"), + ('Bearer error_description="missing scope=wrong", scope="read write"', "scope", "read write"), + ( + 'Bearer x_resource_metadata="https://decoy.example.com", ' + 'resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"', + "resource_metadata", + "https://api.example.com/.well-known/oauth-protected-resource", + ), + ( + 'Bearer error_description="missing resource_metadata=https://decoy.example.com", ' + 'resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"', + "resource_metadata", + "https://api.example.com/.well-known/oauth-protected-resource", + ), # Values with special characters ( 'Bearer scope="resource:read resource:write user_profile"', "scope", "resource:read resource:write user_profile", ), + ('Bearer scope="say \\"hi\\""', "scope", 'say "hi"'), ( 'Bearer resource_metadata="https://api.example.com/auth/metadata?version=1"', "resource_metadata", @@ -2047,8 +2065,23 @@ 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", "substring param name should not match scope"), + ('Bearer error_description="missing scope=wrong"', "scope", "field-like text in quoted value"), + ( + 'Bearer x_resource_metadata="https://decoy.example.com"', + "resource_metadata", + "substring param name should not match resource_metadata", + ), + ( + 'Bearer error_description="missing resource_metadata=https://decoy.example.com"', + "resource_metadata", + "field-like text in quoted value", + ), # Malformed field (empty value) ("Bearer scope=", "scope", "malformed scope parameter"), + ("Bearer scope= ", "scope", "malformed scope parameter with only whitespace"), + ("Bearer scope=,", "scope", "malformed scope parameter with delimiter"), + ('Bearer scope="unterminated', "scope", "unterminated quoted scope parameter"), ("Bearer resource_metadata=", "resource_metadata", "malformed resource_metadata parameter"), ], ) @@ -2070,6 +2103,18 @@ def test_extract_field_from_www_auth_invalid_cases( result = extract_field_from_www_auth(init_response, field_name) assert result is None, f"Should return None for {description}" + def test_extract_resource_metadata_from_www_auth_ignores_substring_param_name(self): + """Test resource_metadata extraction ignores auth-params with longer names.""" + init_response = httpx2.Response( + status_code=401, + headers={"WWW-Authenticate": 'Bearer x_resource_metadata="https://decoy.example.com"'}, + request=httpx2.Request("GET", "https://api.example.com/test"), + ) + + result = extract_resource_metadata_from_www_auth(init_response) + + assert result is None + class TestCIMD: """Test Client ID Metadata Document (CIMD) support."""