From e8b2999aa2abf30ee3a96f239ed3fd5ce206c7f8 Mon Sep 17 00:00:00 2001 From: teddiesloco Date: Sat, 15 Aug 2026 13:05:34 +0700 Subject: [PATCH 1/2] fix(auth): canonicalize resource URLs by stripping default ports (RFC 3986) --- src/mcp/shared/auth_utils.py | 36 +++++++++++++++++++++++++++------ tests/shared/test_auth_utils.py | 13 ++++++++++-- 2 files changed, 41 insertions(+), 8 deletions(-) diff --git a/src/mcp/shared/auth_utils.py b/src/mcp/shared/auth_utils.py index 3ba880f40d..ecab6feec2 100644 --- a/src/mcp/shared/auth_utils.py +++ b/src/mcp/shared/auth_utils.py @@ -1,16 +1,35 @@ """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() + port = parsed.port + + 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 +42,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 +63,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..863b361107 100644 --- a/tests/shared/test_auth_utils.py +++ b/tests/shared/test_auth_utils.py @@ -29,9 +29,13 @@ 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_lowercase_scheme_and_host(): @@ -69,9 +73,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(): From 4729ea78cd730d83ec111c486eab1d7560a8dac8 Mon Sep 17 00:00:00 2001 From: teddiesloco Date: Sat, 15 Aug 2026 18:01:33 +0700 Subject: [PATCH 2/2] fix: catch malformed port ValueError and reach 100% coverage - _canonical_netloc now catches ValueError from parsed.port for malformed explicit ports (e.g. out-of-range), falling back to the original netloc instead of letting check_resource_allowed() crash - add test coverage for the userinfo-in-netloc and IPv6-literal branches that were previously untested (CI requires 100% coverage) - add regression test for the malformed-port fallback --- src/mcp/shared/auth_utils.py | 6 +++++- tests/shared/test_auth_utils.py | 22 ++++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/src/mcp/shared/auth_utils.py b/src/mcp/shared/auth_utils.py index ecab6feec2..78255c292e 100644 --- a/src/mcp/shared/auth_utils.py +++ b/src/mcp/shared/auth_utils.py @@ -10,7 +10,11 @@ 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() - port = parsed.port + 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 diff --git a/tests/shared/test_auth_utils.py b/tests/shared/test_auth_utils.py index 863b361107..d2b9ef6b69 100644 --- a/tests/shared/test_auth_utils.py +++ b/tests/shared/test_auth_utils.py @@ -38,6 +38,23 @@ def test_resource_url_from_server_url_preserves_port(): 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(): """Scheme and host should be lowercase for canonical form.""" assert resource_url_from_server_url("HTTPS://EXAMPLE.COM/path") == "https://example.com/path" @@ -125,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