Skip to content
Open
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
40 changes: 34 additions & 6 deletions src/mcp/shared/auth_utils.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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

Expand All @@ -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():
Expand Down
35 changes: 33 additions & 2 deletions tests/shared/test_auth_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -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():
Expand Down Expand Up @@ -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
Expand Down
Loading