diff --git a/src/mcp/shared/auth_utils.py b/src/mcp/shared/auth_utils.py index 3ba880f40d..b640a6d01e 100644 --- a/src/mcp/shared/auth_utils.py +++ b/src/mcp/shared/auth_utils.py @@ -1,16 +1,41 @@ """Utilities for OAuth 2.0 Resource Indicators (RFC 8707) and PKCE (RFC 7636).""" import time -from urllib.parse import urlparse, urlsplit, urlunsplit +from urllib.parse import urlsplit, urlunsplit from pydantic import AnyUrl, HttpUrl +# WHATWG URL treats these percent-encoded spellings as dot-segments too. +_SINGLE_DOT_SEGMENTS = {".", "%2e"} +_DOUBLE_DOT_SEGMENTS = {"..", ".%2e", "%2e.", "%2e%2e"} + + +def _remove_dot_segments(path: str) -> str: + """Resolve "." and ".." segments in a URL path (RFC 3986 section 5.2.4).""" + segments = path.split("/") + output: list[str] = [] + for index, segment in enumerate(segments): + is_last = index == len(segments) - 1 + kind = segment.lower() + if kind in _DOUBLE_DOT_SEGMENTS: + if len(output) > 1: + output.pop() + if is_last: + output.append("") + elif kind in _SINGLE_DOT_SEGMENTS: + if is_last: + output.append("") + else: + output.append(segment) + return "/".join(output) + def resource_url_from_server_url(url: str | HttpUrl | AnyUrl) -> str: """Convert server URL to canonical resource URL per RFC 8707. RFC 8707 section 2 states that resource URIs "MUST NOT include a fragment component". - Returns absolute URI with lowercase scheme/host for canonical form. + Returns absolute URI with lowercase scheme/host and dot-segments resolved, so the + resource identifies the same location an HTTP client would actually request. Args: url: Server URL to convert @@ -23,7 +48,14 @@ def resource_url_from_server_url(url: str | HttpUrl | AnyUrl) -> str: # Parse the URL and remove fragment, create canonical form parsed = urlsplit(url_str) - canonical = urlunsplit(parsed._replace(scheme=parsed.scheme.lower(), netloc=parsed.netloc.lower(), fragment="")) + canonical = urlunsplit( + parsed._replace( + scheme=parsed.scheme.lower(), + netloc=parsed.netloc.lower(), + path=_remove_dot_segments(parsed.path), + fragment="", + ) + ) return canonical @@ -34,7 +66,8 @@ def check_resource_allowed(requested_resource: str, configured_resource: str) -> A requested resource matches if it has the same scheme, domain, port, and its path starts with the configured resource's path. This allows hierarchical matching where a token for a parent resource can be used - for child resources. + for child resources. Dot-segments in either path are resolved before + comparing. Args: requested_resource: The resource URL being requested @@ -44,8 +77,8 @@ def check_resource_allowed(requested_resource: str, configured_resource: str) -> True if the requested resource matches the configured resource """ # Parse both URLs - requested = urlparse(requested_resource) - configured = urlparse(configured_resource) + requested = urlsplit(requested_resource) + configured = urlsplit(configured_resource) # Compare scheme, host, and port (origin) if requested.scheme.lower() != configured.scheme.lower() or requested.netloc.lower() != configured.netloc.lower(): @@ -53,8 +86,8 @@ def check_resource_allowed(requested_resource: str, configured_resource: str) -> # Normalize trailing slashes before comparison so that # "/foo" and "/foo/" are treated as equivalent. - requested_path = requested.path - configured_path = configured.path + requested_path = _remove_dot_segments(requested.path) + configured_path = _remove_dot_segments(configured.path) if not requested_path.endswith("/"): requested_path += "/" if not configured_path.endswith("/"): diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py index be96cc8eec..24795b421c 100644 --- a/tests/client/test_auth.py +++ b/tests/client/test_auth.py @@ -862,6 +862,36 @@ async def test_validate_resource_rejects_mismatched_resource( await provider._validate_resource_match(prm) +@pytest.mark.anyio +async def test_validate_resource_rejects_sibling_path_reached_via_dot_segments( + client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage +) -> None: + """A `server_url` whose dot-segments resolve to `/m/mcp` rejects a PRM `resource` of `/victim/mcp`. + + SDK-defined: the resource identifier is derived from the location the HTTP client actually + requests (RFC 3986 section 5.2.4), so a same-origin sibling path is neither accepted during + discovery nor adopted as the RFC 8707 `resource` parameter. + """ + provider = OAuthClientProvider( + server_url="https://shared.example.com/victim/mcp/../../m/mcp", + client_metadata=client_metadata, + storage=mock_storage, + ) + provider._initialized = True + + prm = ProtectedResourceMetadata( + resource=AnyHttpUrl("https://shared.example.com/victim/mcp"), + authorization_servers=[AnyHttpUrl("https://auth.example.com")], + ) + with pytest.raises(OAuthFlowError) as exc_info: + await provider._validate_resource_match(prm) + assert str(exc_info.value) == snapshot( + "Protected resource https://shared.example.com/victim/mcp does not match expected https://shared.example.com/m/mcp" + ) + provider.context.protected_resource_metadata = prm + assert provider.context.get_resource_url() == snapshot("https://shared.example.com/m/mcp") + + @pytest.mark.anyio async def test_validate_resource_accepts_matching_resource( client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage diff --git a/tests/shared/test_auth_utils.py b/tests/shared/test_auth_utils.py index 5ae0e22b0c..2e78807eba 100644 --- a/tests/shared/test_auth_utils.py +++ b/tests/shared/test_auth_utils.py @@ -1,6 +1,9 @@ """Tests for OAuth 2.0 Resource Indicators utilities.""" -from pydantic import HttpUrl +import itertools + +import pytest +from pydantic import AnyHttpUrl, HttpUrl from mcp.shared.auth_utils import check_resource_allowed, resource_url_from_server_url @@ -46,6 +49,42 @@ def test_resource_url_from_server_url_handles_pydantic_urls(): assert resource_url_from_server_url(url) == "https://example.com/path" +@pytest.mark.parametrize( + ("server_url", "expected"), + [ + ("https://example.com/api/../admin", "https://example.com/admin"), + ("https://example.com/api/%2E%2e/admin", "https://example.com/admin"), + ("https://example.com/api/.%2e/admin", "https://example.com/admin"), + ("https://example.com/api/./v1", "https://example.com/api/v1"), + ("https://example.com/api/v1/..", "https://example.com/api/"), + ("https://example.com/api/v1/.", "https://example.com/api/v1/"), + ("https://example.com/../admin", "https://example.com/admin"), + ("https://example.com/a//b", "https://example.com/a//b"), + ("https://example.com/a%2Fb/c", "https://example.com/a%2Fb/c"), + ], +) +def test_resource_url_from_server_url_resolves_dot_segments(server_url: str, expected: str): + """Dot-segments (including `%2E` spellings) are resolved per RFC 3986 section 5.2.4. + + Empty segments and encoded slashes are not path separators and stay as written. + """ + assert resource_url_from_server_url(server_url) == expected + + +def test_resource_url_from_server_url_path_matches_whatwg_resolution_for_literal_dot_segments(): + """Every combination of literal `.`, `..`, empty and plain segments resolves as pydantic's WHATWG parser does. + + The PRM `resource` side is parsed by `AnyHttpUrl`, so both operands of `check_resource_allowed` + must agree on dot-segment resolution for the comparison to be meaningful. + """ + atoms = ["", ".", "..", "a", "b.", "..."] + for count in range(1, 5): + for segments in itertools.product(atoms, repeat=count): + path = "/" + "/".join(segments) + expected = AnyHttpUrl(f"https://example.com{path}").path + assert resource_url_from_server_url(f"https://example.com{path}") == f"https://example.com{expected}" + + # Tests for check_resource_allowed function @@ -121,3 +160,30 @@ def test_check_resource_allowed_empty_paths(): assert check_resource_allowed("https://example.com", "https://example.com") is True assert check_resource_allowed("https://example.com/", "https://example.com") is True assert check_resource_allowed("https://example.com/api", "https://example.com") is True + + +@pytest.mark.parametrize( + "requested", + [ + "https://example.com/api/../admin", + "https://example.com/api/%2e%2e/admin", + "https://example.com/api/v1/../../admin", + "https://example.com/api/..", + ], +) +def test_check_resource_allowed_rejects_dot_segments_escaping_configured_path(requested: str): + """A requested path that resolves outside the configured path is not a hierarchical match.""" + assert check_resource_allowed(requested, "https://example.com/api") is False + + +def test_check_resource_allowed_resolves_dot_segments_on_both_sides(): + """Both URLs are compared in resolved form, so equivalent spellings agree (SDK-defined matching).""" + assert check_resource_allowed("https://example.com/api/./v1", "https://example.com/api") is True + assert check_resource_allowed("https://example.com/api/v1", "https://example.com/other/../api") is True + assert check_resource_allowed("https://example.com/api/v1", "https://example.com/api/v1/../v2") is False + + +def test_check_resource_allowed_keeps_encoded_slash_and_params_in_segment(): + """`%2F` and `;params` are part of a segment (RFC 3986 sections 2.2, 3.3), not a boundary.""" + assert check_resource_allowed("https://example.com/api%2Fv1", "https://example.com/api") is False + assert check_resource_allowed("https://example.com/api/v1", "https://example.com/api;x") is False