diff --git a/src/mcp/shared/auth_utils.py b/src/mcp/shared/auth_utils.py index 3ba880f40d..78255c292e 100644 --- a/src/mcp/shared/auth_utils.py +++ b/src/mcp/shared/auth_utils.py @@ -1,16 +1,39 @@ """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 SplitResult, urlparse, urlsplit, urlunsplit from pydantic import AnyUrl, HttpUrl +def _canonical_netloc(parsed: SplitResult) -> str: + """Normalize netloc by lowercasing and stripping explicit default ports (RFC 3986 §6.2.3).""" + scheme = parsed.scheme.lower() + netloc = parsed.netloc.lower() + try: + port = parsed.port + except ValueError: + # Malformed explicit port (e.g. non-numeric) - not canonicalizable, fall back as-is + return netloc + + if (scheme == "http" and port == 80) or (scheme == "https" and port == 443): + # Strip default port while preserving userinfo and IPv6 brackets + userinfo = "" + if "@" in netloc: + userinfo = netloc.split("@", 1)[0] + "@" + hostname = parsed.hostname.lower() if parsed.hostname else "" + if ":" in hostname: # IPv6 literal + hostname = f"[{hostname}]" + return f"{userinfo}{hostname}" + return netloc + + 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. + RFC 3986 section 6.2.3 specifies normalization of default ports (80 for http, 443 for https). + Returns absolute URI with lowercase scheme/host and stripped default ports for canonical form. Args: url: Server URL to convert @@ -23,7 +46,8 @@ 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_netloc = _canonical_netloc(parsed) + canonical = urlunsplit(parsed._replace(scheme=parsed.scheme.lower(), netloc=canonical_netloc, fragment="")) return canonical @@ -43,9 +67,13 @@ def check_resource_allowed(requested_resource: str, configured_resource: str) -> Returns: True if the requested resource matches the configured resource """ - # Parse both URLs - requested = urlparse(requested_resource) - configured = urlparse(configured_resource) + # Canonicalize both resource URLs (RFC 8707 & RFC 3986 default port normalization) + requested_canonical = resource_url_from_server_url(requested_resource) + configured_canonical = resource_url_from_server_url(configured_resource) + + # Parse both canonical URLs + requested = urlparse(requested_canonical) + configured = urlparse(configured_canonical) # Compare scheme, host, and port (origin) if requested.scheme.lower() != configured.scheme.lower() or requested.netloc.lower() != configured.netloc.lower(): diff --git a/tests/shared/test_auth_utils.py b/tests/shared/test_auth_utils.py index 5ae0e22b0c..d2b9ef6b69 100644 --- a/tests/shared/test_auth_utils.py +++ b/tests/shared/test_auth_utils.py @@ -29,9 +29,30 @@ def test_resource_url_from_server_url_preserves_query(): def test_resource_url_from_server_url_preserves_port(): - """Non-default ports should be preserved.""" + """Non-default ports should be preserved while default ports are stripped per RFC 3986 §6.2.3.""" assert resource_url_from_server_url("https://example.com:8443/path") == "https://example.com:8443/path" assert resource_url_from_server_url("http://example.com:8080/") == "http://example.com:8080/" + assert resource_url_from_server_url("https://example.com:443/path") == "https://example.com/path" + assert resource_url_from_server_url("http://example.com:80/path") == "http://example.com/path" + assert resource_url_from_server_url("http://example.com:80") == "http://example.com" + assert resource_url_from_server_url("https://example.com:443") == "https://example.com" + + +def test_resource_url_from_server_url_strips_default_port_with_userinfo(): + """Default port stripping must preserve userinfo in the netloc.""" + assert resource_url_from_server_url("https://user:pass@example.com:443/mcp") == "https://user:pass@example.com/mcp" + assert resource_url_from_server_url("http://user@example.com:80/api") == "http://user@example.com/api" + + +def test_resource_url_from_server_url_strips_default_port_with_ipv6(): + """Default port stripping must preserve IPv6 literal brackets.""" + assert resource_url_from_server_url("https://[::1]:443/path") == "https://[::1]/path" + assert resource_url_from_server_url("http://[2001:db8::1]:80/api") == "http://[2001:db8::1]/api" + + +def test_resource_url_from_server_url_malformed_port_falls_back(): + """A malformed explicit port must not raise; it should fall back to the original netloc.""" + assert resource_url_from_server_url("https://example.com:99999999/path") == "https://example.com:99999999/path" def test_resource_url_from_server_url_lowercase_scheme_and_host(): @@ -69,9 +90,14 @@ def test_check_resource_allowed_different_domains(): def test_check_resource_allowed_different_ports(): - """Different ports should not match.""" + """Different ports should not match, but explicit default ports are equivalent to omitted ports.""" assert check_resource_allowed("https://example.com:8443/path", "https://example.com/path") is False assert check_resource_allowed("https://example.com:8080/", "https://example.com:8443/") is False + # Explicit default ports per RFC 3986 §6.2.3 + assert check_resource_allowed("https://example.com:443/mcp", "https://example.com/mcp") is True + assert check_resource_allowed("https://example.com/mcp", "https://example.com:443/mcp") is True + assert check_resource_allowed("http://example.com:80/api", "http://example.com/api") is True + assert check_resource_allowed("http://example.com/api", "http://example.com:80/api") is True def test_check_resource_allowed_hierarchical_matching(): @@ -116,6 +142,11 @@ def test_check_resource_allowed_case_insensitive_origin(): assert check_resource_allowed("https://Example.Com:8080/api", "https://example.com:8080/api") is True +def test_check_resource_allowed_malformed_port_does_not_raise(): + """A malformed explicit port must not raise; check_resource_allowed should return a bool.""" + assert check_resource_allowed("https://example.com:99999999/path", "https://example.com/path") is False + + def test_check_resource_allowed_empty_paths(): """Empty paths should be handled correctly.""" assert check_resource_allowed("https://example.com", "https://example.com") is True