fix(server): require a numeric port for Host/Origin :* allowlist entries

src/mcp/server/transport_security.py

@@ -16,6 +16,21 @@
 """Default maximum HTTP request body size in bytes (4 MiB)."""
 
 
+def _matches_wildcard_port(value: str, allowed: str) -> bool:
+    """Return True when ``allowed`` is ``base:*`` and ``value`` is ``base:<digits>``.
+
+    A prefix check alone accepts ``127.0.0.1:8080.evil`` for ``127.0.0.1:*``.
+    The port suffix must be digits so the wildcard cannot match a longer host
+    or origin.
+    """
+    if not allowed.endswith(":*"):
+        return False
+    prefix = allowed[:-1]  # "base:"
+    if not value.startswith(prefix):
+        return False
+    return value[len(prefix) :].isdigit()
+
+
 # TODO(Marcelo): We should flatten these settings. To be fair, I don't think we should even have this middleware.
 class TransportSecuritySettings(BaseModel):
     """Settings for MCP transport security features.
@@ -57,14 +72,10 @@ def _validate_host(self, host: str | None) -> bool:
         if host in self.settings.allowed_hosts:
             return True
 
-        # Check wildcard port patterns
+        # Check wildcard port patterns (base:* matches only base:<digits>)
         for allowed in self.settings.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 + ":"):
-                    return True
+            if _matches_wildcard_port(host, allowed):
+                return True
 
         logger.warning(f"Invalid Host header: {host}")
         return False
@@ -79,14 +90,10 @@ def _validate_origin(self, origin: str | None) -> bool:
         if origin in self.settings.allowed_origins:
             return True
 
-        # Check wildcard port patterns
+        # Check wildcard port patterns (base:* matches only base:<digits>)
         for allowed in self.settings.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 + ":"):
-                    return True
+            if _matches_wildcard_port(origin, allowed):
+                return True
 
         logger.warning(f"Invalid Origin header: {origin}")
         return False

tests/server/test_transport_security.py

@@ -41,10 +41,13 @@ def _request(host: str | None, origin: str | None, content_type: str | None = "a
         pytest.param("evil.example:9000", None, 421, id="host-wildcard-base-mismatch"),
         pytest.param("good.example", None, None, id="host-exact-no-origin"),
         pytest.param("wild.example:9000", None, None, id="host-wildcard-match"),
+        pytest.param("wild.example:9000.evil", None, 421, id="host-wildcard-suffix-rejected"),
+        pytest.param("wild.example:", None, 421, id="host-wildcard-empty-port"),
         pytest.param("good.example", "http://evil.example", 403, id="origin-no-match"),
         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", "http://wild.example:9000.evil", 403, id="origin-wildcard-suffix-rejected"),
     ],
 )
 async def test_validate_request_checks_host_then_origin(