Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 41 additions & 8 deletions src/mcp/shared/auth_utils.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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

Expand All @@ -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
Expand All @@ -44,17 +77,17 @@ 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():
return False

# 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("/"):
Expand Down
30 changes: 30 additions & 0 deletions tests/client/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
68 changes: 67 additions & 1 deletion tests/shared/test_auth_utils.py
Original file line number Diff line number Diff line change
@@ -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

Expand Down Expand Up @@ -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


Expand Down Expand Up @@ -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
Loading