Skip to content
Closed
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
25 changes: 19 additions & 6 deletions src/mcp/server/transport_security.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,17 +53,24 @@ def _validate_host(self, host: str | None) -> bool:
logger.warning("Missing Host header in request")
return False

# Host names are case-insensitive (RFC 9110 Section 4.2.3), and WHATWG-URL
# clients (fetch, undici, browsers) lowercase the URL host before sending,
# so compare case-insensitively regardless of how allowed_hosts was written.
normalized_host = host.lower()

# Check exact match first
if host in self.settings.allowed_hosts:
return True
for allowed in self.settings.allowed_hosts:
if allowed.lower() == normalized_host:
return True

# Check wildcard port patterns
for allowed in self.settings.allowed_hosts:
allowed = allowed.lower()
if allowed.endswith(":*"):
# Extract base host from pattern
base_host = allowed[:-2]
# Check if the actual host starts with base host and has a port
if host.startswith(base_host + ":"):
if normalized_host.startswith(base_host + ":"):
return True

logger.warning(f"Invalid Host header: {host}")
Expand All @@ -75,17 +82,23 @@ def _validate_origin(self, origin: str | None) -> bool:
if not origin:
return True

# An origin is scheme://host[:port]; scheme and host are case-insensitive
# (RFC 6454 Section 4), and clients send them lowercased. Compare case-insensitively.
normalized_origin = origin.lower()

# Check exact match first
if origin in self.settings.allowed_origins:
return True
for allowed in self.settings.allowed_origins:
if allowed.lower() == normalized_origin:
return True

# Check wildcard port patterns
for allowed in self.settings.allowed_origins:
allowed = allowed.lower()
if allowed.endswith(":*"):
# Extract base origin from pattern
base_origin = allowed[:-2]
# Check if the actual origin starts with base origin and has a port
if origin.startswith(base_origin + ":"):
if normalized_origin.startswith(base_origin + ":"):
return True

logger.warning(f"Invalid Origin header: {origin}")
Expand Down
24 changes: 24 additions & 0 deletions tests/server/test_transport_security.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,11 @@ def _request(host: str | None, origin: str | None, content_type: str | None = "a
pytest.param("good.example", "http://evil.example:9000", 403, id="origin-wildcard-base-mismatch"),
pytest.param("good.example", "http://good.example", None, id="origin-exact"),
pytest.param("good.example", "http://wild.example:9000", None, id="origin-wildcard-match"),
# Host / Origin are case-insensitive (RFC 9110 / RFC 6454); clients lowercase them.
pytest.param("GOOD.EXAMPLE", None, None, id="host-exact-uppercase-request"),
pytest.param("WILD.EXAMPLE:9000", None, None, id="host-wildcard-uppercase-request"),
pytest.param("good.example", "HTTP://GOOD.EXAMPLE", None, id="origin-exact-uppercase-request"),
pytest.param("good.example", "http://WILD.EXAMPLE:9000", None, id="origin-wildcard-uppercase-request"),
],
)
async def test_validate_request_checks_host_then_origin(
Expand All @@ -56,6 +61,25 @@ async def test_validate_request_checks_host_then_origin(
assert (None if response is None else response.status_code) == expected


@pytest.mark.anyio
@pytest.mark.parametrize(
("host_header", "expected"),
[
# The Windows default path into the bug: allowed_hosts derived from
# %COMPUTERNAME% (always uppercase), client sends the lowercased host.
pytest.param("myhost:8000", None, id="uppercase-config-lowercase-request"),
pytest.param("MYHOST:8000", None, id="uppercase-config-uppercase-request"),
pytest.param("other:8000", 421, id="uppercase-config-still-rejects-non-match"),
],
)
async def test_validate_host_case_insensitive_with_uppercase_allowlist(host_header: str, expected: int | None) -> None:
"""An uppercase allowed_hosts entry still matches a lowercased Host header (RFC 9110 Section 4.2.3)."""
settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["MYHOST:*"])
middleware = TransportSecurityMiddleware(settings)
response = await middleware.validate_request(_request(host_header, None))
assert (None if response is None else response.status_code) == expected


@pytest.mark.anyio
async def test_validate_request_skips_host_and_origin_when_protection_is_disabled() -> None:
"""With DNS-rebinding protection off, any Host/Origin is accepted."""
Expand Down
Loading