diff --git a/src/mcp/client/auth/utils.py b/src/mcp/client/auth/utils.py index fc87c9e46..beaf14e06 100644 --- a/src/mcp/client/auth/utils.py +++ b/src/mcp/client/auth/utils.py @@ -16,6 +16,88 @@ 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 + escaped = False + + for char in header_value: + 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() + 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) + + +_AUTH_PARAM_PATTERN = re.compile( + r"(?:^|,\s*)(?P[A-Za-z][A-Za-z0-9_-]*)\s*=\s*" + r'(?:"(?P(?:\\.|[^"\\])*)"|(?P[^,\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. @@ -26,13 +108,11 @@ 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 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 @@ -43,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: @@ -55,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]: diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py index 9e9599f86..954004db3 100644 --- a/tests/client/test_auth.py +++ b/tests/client/test_auth.py @@ -181,6 +181,91 @@ def test_pkce_uniqueness(self): assert pkce1.code_challenge != pkce2.code_challenge +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 = httpx2.Response( + 401, + headers={ + "WWW-Authenticate": ( + 'Basic scope="legacy", Bearer scope="read", error="insufficient_scope", Bearer scope="write"' + ) + }, + request=httpx2.Request("GET", "https://example.com"), + ) + + assert extract_scope_from_www_auth(response) == "read" + + +def test_bearer_fields_support_optional_whitespace_around_equals(): + """Bearer auth-params allow optional whitespace around the equals sign.""" + response = httpx2.Response( + 401, + headers={ + "WWW-Authenticate": ( + 'Bearer error="insufficient_scope", scope = "read write", ' + 'resource_metadata = "https://example.com/.well-known/oauth-protected-resource"' + ) + }, + request=httpx2.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 = httpx2.Response( + 401, + headers={"WWW-Authenticate": 'Newauth realm="apps"'}, + request=httpx2.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 = httpx2.Response( + 401, + headers={ + "WWW-Authenticate": ( + 'Newauth realm="apps \\"beta\\", east", ' + 'Bearer error_description="try \\"again\\", later", scope="read write"' + ) + }, + request=httpx2.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 = httpx2.Response( + 401, + headers={"WWW-Authenticate": ', Bearer scope="read", Basic realm="legacy",'}, + request=httpx2.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 = httpx2.Response( + 401, + headers={"WWW-Authenticate": 'Bearer scope=""'}, + request=httpx2.Request("GET", "https://example.com"), + ) + + assert extract_scope_from_www_auth(response) is None + + class TestOAuthContext: """Test OAuth context functionality.""" @@ -2016,6 +2101,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 +2143,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"),