Skip to content
82 changes: 74 additions & 8 deletions src/mcp/client/auth/utils.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import re
from collections.abc import Iterator
from urllib.parse import urljoin, urlparse

from httpx2 import Request, Response
Expand All @@ -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.

Expand All @@ -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

Expand Down
45 changes: 45 additions & 0 deletions tests/client/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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"),
],
)
Expand All @@ -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."""
Expand Down
Loading