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

host_lower = host.lower()
allowed_hosts = [allowed.lower() for allowed in self.settings.allowed_hosts]

# Check exact match first
if host in self.settings.allowed_hosts:
if host_lower in allowed_hosts:
return True

# Check wildcard port patterns
for allowed in self.settings.allowed_hosts:
for allowed in allowed_hosts:
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 host_lower.startswith(base_host + ":"):
return True

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

origin_lower = origin.lower()
allowed_origins = [allowed.lower() for allowed in self.settings.allowed_origins]

# Check exact match first
if origin in self.settings.allowed_origins:
if origin_lower in allowed_origins:
return True

# Check wildcard port patterns
for allowed in self.settings.allowed_origins:
for allowed in allowed_origins:
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 origin_lower.startswith(base_origin + ":"):
return True

logger.warning(f"Invalid Origin header: {origin}")
Expand Down
30 changes: 30 additions & 0 deletions tests/server/test_transport_security.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,9 @@ 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"),
pytest.param("GOOD.EXAMPLE", None, None, id="host-exact-uppercase"),
pytest.param("Good.Example", "HTTP://GOOD.EXAMPLE", None, id="host-origin-mixedcase"),
pytest.param("WILD.EXAMPLE:9000", "http://WILD.EXAMPLE:9000", None, id="wildcard-uppercase"),
],
)
async def test_validate_request_checks_host_then_origin(
Expand All @@ -56,6 +59,33 @@ async def test_validate_request_checks_host_then_origin(
assert (None if response is None else response.status_code) == expected


@pytest.mark.anyio
async def test_validate_request_case_insensitive_with_uppercase_allowed_settings() -> None:
"""Uppercase entries in allowed_hosts and allowed_origins match lowercase and mixed-case requests."""
settings = TransportSecuritySettings(
enable_dns_rebinding_protection=True,
allowed_hosts=["MYHOST", "MYWILD:*"],
allowed_origins=["HTTP://MYORIGIN", "HTTP://MYWILDORIGIN:*"],
)
middleware = TransportSecurityMiddleware(settings)

# Exact host lowercase & mixed-case
assert await middleware.validate_request(_request("myhost", None)) is None
assert await middleware.validate_request(_request("MyHost", None)) is None

# Wildcard host lowercase & mixed-case
assert await middleware.validate_request(_request("mywild:8000", None)) is None
assert await middleware.validate_request(_request("MyWild:8000", None)) is None

# Exact origin lowercase & mixed-case
assert await middleware.validate_request(_request("myhost", "http://myorigin")) is None
assert await middleware.validate_request(_request("myhost", "http://MyOrigin")) is None

# Wildcard origin lowercase & mixed-case
assert await middleware.validate_request(_request("myhost", "http://mywildorigin:8000")) is None
assert await middleware.validate_request(_request("myhost", "http://MyWildOrigin:8000")) is None


@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