Protect streamable HTTP from DNS rebinding (#4405)

docs/deployment/http.mdx

@@ -101,6 +101,51 @@ FastMCP supports multiple authentication methods to secure your remote server. S
 
 If you're mounting an authenticated server under a path prefix, see [Mounting Authenticated Servers](#mounting-authenticated-servers) below for important routing considerations.
 
+### Host and Origin Protection
+
+FastMCP validates `Host` and browser `Origin` headers for Streamable HTTP requests by default. This protects localhost-bound servers from DNS rebinding attacks and rejects browser requests from origins you have not trusted.
+
+Think of this as a request guard rather than CORS middleware. It decides whether a request can reach MCP session handling. CORS remains a separate browser response-header policy; configure CORS middleware separately when browser JavaScript must read cross-origin responses.
+
+When you deploy behind a public hostname, add the hostname clients use to reach your MCP endpoint. If a browser-based MCP client runs on a separate origin, add that origin as well:
+
+```python
+from fastmcp import FastMCP
+
+mcp = FastMCP("My Server")
+
+app = mcp.http_app(
+    allowed_hosts=["mcp.example.com"],
+    allowed_origins=["https://app.example.com"],
+)
+```
+
+For the direct server approach, pass the same values to `run()`:
+
+```python
+from fastmcp import FastMCP
+
+mcp = FastMCP("My Server")
+
+if __name__ == "__main__":
+    mcp.run(
+        transport="http",
+        host="0.0.0.0",
+        port=8000,
+        allowed_hosts=["mcp.example.com"],
+        allowed_origins=["https://app.example.com"],
+    )
+```
+
+You can also configure these values with environment variables:
+
+```bash
+export FASTMCP_HTTP_ALLOWED_HOSTS='["mcp.example.com"]'
+export FASTMCP_HTTP_ALLOWED_ORIGINS='["https://app.example.com"]'
+```
+
+Use `host_origin_protection=False` only for trusted internal deployments that provide equivalent validation at another layer, such as an ingress proxy.
+
 ### Health Checks
 
 Health check endpoints are essential for monitoring your deployed server and ensuring it's responding correctly. FastMCP allows you to add custom routes alongside your MCP endpoints, making it easy to implement health checks that work with both deployment approaches.
@@ -156,6 +201,8 @@ Most MCP clients, including those that you access through a browser like ChatGPT
 
 CORS (Cross-Origin Resource Sharing) is needed when JavaScript running in a web browser connects directly to your MCP server. This is different from using an LLM through a browser—in that case, the browser connects to the LLM service, and the LLM service connects to your MCP server (no CORS needed).
 
+Host and Origin protection runs before CORS. Add browser client origins to `allowed_origins` so trusted browser requests reach the CORS middleware, then configure CORS to let browser JavaScript read the MCP response headers it needs. Setting `allowed_origins` trusts the request; it does not emit `Access-Control-Allow-Origin` or other CORS response headers.
+
 Browser-based MCP clients that need CORS include:
 
 - **MCP Inspector** - Browser-based debugging tool for testing MCP servers

docs/more/settings.mdx

@@ -42,6 +42,9 @@ These control how the server listens when running with an HTTP transport.
 | `FASTMCP_STREAMABLE_HTTP_PATH` | `str` | `/mcp` | Path for Streamable HTTP endpoint. |
 | `FASTMCP_STATELESS_HTTP` | `bool` | `false` | Enable stateless HTTP mode (new transport per request). Useful for multi-worker deployments. |
 | `FASTMCP_JSON_RESPONSE` | `bool` | `false` | Use JSON responses instead of SSE for Streamable HTTP. |
+| `FASTMCP_HTTP_HOST_ORIGIN_PROTECTION` | `bool` | `true` | Validate `Host` and browser `Origin` headers for Streamable HTTP requests. |
+| `FASTMCP_HTTP_ALLOWED_HOSTS` | `list[str] \| null` | `null` | Additional trusted hostnames for Streamable HTTP requests. Use a JSON array, such as `["mcp.example.com"]`. |
+| `FASTMCP_HTTP_ALLOWED_ORIGINS` | `list[str] \| null` | `null` | Browser origins trusted by the Streamable HTTP request guard. Configure CORS separately for cross-origin browser reads. Use a JSON array, such as `["https://app.example.com"]`. |
 | `FASTMCP_DEBUG` | `bool` | `false` | Enable debug mode. |
 
 ## Error Handling

fastmcp_slim/fastmcp/server/http.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
-from collections.abc import AsyncGenerator, Callable, Generator
+from collections.abc import AsyncGenerator, Callable, Generator, Sequence
 from contextlib import asynccontextmanager, contextmanager
 from contextvars import ContextVar
+from fnmatch import fnmatchcase
+from ipaddress import ip_address
 from typing import TYPE_CHECKING, Any
+from urllib.parse import urlsplit
 from uuid import uuid4
 
 from mcp.server.auth.routes import build_resource_metadata_url
@@ -15,11 +18,12 @@ from mcp.server.streamable_http import (
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
 from starlette.applications import Starlette
+from starlette.datastructures import Headers
 from starlette.middleware import Middleware
 from starlette.requests import Request
 from starlette.responses import Response
 from starlette.routing import BaseRoute, Mount, Route
-from starlette.types import Lifespan, Receive, Scope, Send
+from starlette.types import ASGIApp, Lifespan, Receive, Scope, Send
 
 from fastmcp.server.auth import AuthProvider
 from fastmcp.server.auth.middleware import RequireAuthMiddleware
@@ -31,6 +35,8 @@ if TYPE_CHECKING:
 
 logger = get_logger(__name__)
 
+DEFAULT_HOSTS = ("127.0.0.1", "localhost", "::1")
+
 
 class FastMCPStreamableHTTPSessionManager(StreamableHTTPSessionManager):
     """Session manager that scopes resumability storage per transport session."""
@@ -103,6 +109,176 @@ class StreamableHTTPASGIApp:
                 raise
 
 
+def _normalize_host(host: str) -> str:
+    host = host.strip().lower()
+    if not host:
+        return ""
+
+    if host.startswith("["):
+        end = host.find("]")
+        if end == -1:
+            return host
+        return host[1:end]
+
+    if host.count(":") == 1:
+        return host.rsplit(":", 1)[0]
+
+    return host
+
+
+def _is_loopback_host(host: str) -> bool:
+    host = _normalize_host(host)
+    if host == "localhost":
+        return True
+
+    try:
+        return ip_address(host).is_loopback
+    except ValueError:
+        return False
+
+
+def _is_unspecified_host(host: str) -> bool:
+    host = _normalize_host(host)
+    if not host:
+        return True
+
+    try:
+        return ip_address(host).is_unspecified
+    except ValueError:
+        return False
+
+
+def _host_matches(host: str, allowed_hosts: Sequence[str]) -> bool:
+    host = _normalize_host(host)
+    for allowed_host in allowed_hosts:
+        pattern = _normalize_host(allowed_host)
+        if pattern == "*" or fnmatchcase(host, pattern):
+            return True
+
+    return False
+
+
+def _origin_host(origin: str) -> str:
+    try:
+        parsed = urlsplit(origin)
+    except ValueError:
+        return ""
+
+    return parsed.hostname or ""
+
+
+def _origin_port(scheme: str, port: int | None) -> int | None:
+    if port is not None:
+        return port
+    if scheme == "http":
+        return 80
+    if scheme == "https":
+        return 443
+    return None
+
+
+def _format_origin_host(host: str) -> str:
+    if ":" in host and not host.startswith("["):
+        return f"[{host}]"
+    return host
+
+
+def _normalize_origin(origin: str) -> str:
+    origin = origin.strip().rstrip("/")
+    try:
+        parsed = urlsplit(origin)
+        port = parsed.port
+    except ValueError:
+        return origin.lower()
+
+    if not parsed.scheme or not parsed.hostname:
+        return origin.lower()
+
+    if parsed.path or parsed.query or parsed.fragment:
+        return origin.lower()
+
+    scheme = parsed.scheme.lower()
+    host = _format_origin_host(_normalize_host(parsed.hostname))
+    normalized_port = _origin_port(scheme, port)
+    if normalized_port is None:
+        return f"{scheme}://{host}"
+
+    return f"{scheme}://{host}:{normalized_port}"
+
+
+def _request_origin(scope: Scope, host: str) -> str:
+    return _normalize_origin(f"{scope.get('scheme', 'http')}://{host}")
+
+
+def _origin_matches(origin: str, allowed_origins: Sequence[str]) -> bool:
+    origin = _normalize_origin(origin)
+    for allowed_origin in allowed_origins:
+        pattern = _normalize_origin(allowed_origin)
+        if pattern == "*" or fnmatchcase(origin, pattern):
+            return True
+
+    return False
+
+
+class HostOriginGuardMiddleware:
+    """Validate Host and Origin headers before requests reach MCP sessions."""
+
+    def __init__(
+        self,
+        app: ASGIApp,
+        allowed_hosts: Sequence[str] | None = None,
+        allowed_origins: Sequence[str] | None = None,
+    ) -> None:
+        self.app = app
+        self.allowed_hosts = tuple(allowed_hosts or ())
+        self.allowed_origins = tuple(allowed_origins or ())
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        allowed_hosts = self._allowed_hosts_for_scope(scope)
+        headers = Headers(scope=scope)
+        host = headers.get("host", "")
+
+        if not _host_matches(host, allowed_hosts):
+            response = Response("Misdirected Request", status_code=421)
+            await response(scope, receive, send)
+            return
+
+        origin = headers.get("origin")
+        request_origin = _request_origin(scope, host)
+        if origin and not self._origin_allowed(origin, request_origin, host):
+            response = Response("Forbidden Origin", status_code=403)
+            await response(scope, receive, send)
+            return
+
+        await self.app(scope, receive, send)
+
+    def _allowed_hosts_for_scope(self, scope: Scope) -> tuple[str, ...]:
+        allowed_hosts = list(DEFAULT_HOSTS)
+        allowed_hosts.extend(self.allowed_hosts)
+
+        server = scope.get("server")
+        if server:
+            server_host = server[0]
+            if not _is_unspecified_host(server_host):
+                allowed_hosts.append(server_host)
+
+        return tuple(allowed_hosts)
+
+    def _origin_allowed(self, origin: str, request_origin: str, host: str) -> bool:
+        if _origin_matches(origin, self.allowed_origins):
+            return True
+
+        origin_host = _origin_host(origin)
+        if _is_loopback_host(origin_host) and _is_loopback_host(host):
+            return True
+
+        return _normalize_origin(origin) == request_origin
+
+
 _current_http_request: ContextVar[Request | None] = ContextVar(
     "http_request",
     default=None,
@@ -315,6 +491,9 @@ def create_streamable_http_app(
     debug: bool = False,
     routes: list[BaseRoute] | None = None,
     middleware: list[Middleware] | None = None,
+    host_origin_protection: bool = True,
+    allowed_hosts: Sequence[str] | None = None,
+    allowed_origins: Sequence[str] | None = None,
 ) -> StarletteWithLifespan:
     """Return an instance of the StreamableHTTP server app.
 
@@ -331,6 +510,12 @@ def create_streamable_http_app(
         debug: Whether to enable debug mode
         routes: Optional list of custom routes
         middleware: Optional list of middleware
+        host_origin_protection: Whether to validate Host and Origin headers
+            before requests reach the MCP endpoint.
+        allowed_hosts: Additional hostnames that may appear in the Host header.
+        allowed_origins: Additional browser origins trusted by the request guard.
+            Configure CORS separately when browser JavaScript must read
+            cross-origin responses.
 
     Returns:
         A Starlette application with StreamableHTTP support
@@ -391,6 +576,15 @@ def create_streamable_http_app(
     server_routes.extend(server._get_additional_http_routes())
 
     # Add middleware
+    if host_origin_protection:
+        server_middleware.insert(
+            0,
+            Middleware(
+                HostOriginGuardMiddleware,
+                allowed_hosts=allowed_hosts,
+                allowed_origins=allowed_origins,
+            ),
+        )
     if middleware:
         server_middleware.extend(middleware)
 

fastmcp_slim/fastmcp/server/mixins/transport.py

@@ -250,6 +250,9 @@ class TransportMixin:
         json_response: bool | None = None,
         stateless_http: bool | None = None,
         stateless: bool | None = None,
+        host_origin_protection: bool | None = None,
+        allowed_hosts: list[str] | None = None,
+        allowed_origins: list[str] | None = None,
         sockets: list[socket.socket] | None = None,
     ) -> None:
         """Run the server using HTTP transport.
@@ -265,6 +268,12 @@ class TransportMixin:
             json_response: Whether to use JSON response format (defaults to settings.json_response)
             stateless_http: Whether to use stateless HTTP (defaults to settings.stateless_http)
             stateless: Alias for stateless_http for CLI consistency
+            host_origin_protection: Whether to validate Host and Origin headers
+                before requests reach the MCP endpoint.
+            allowed_hosts: Additional hostnames that may appear in the Host header.
+            allowed_origins: Additional browser origins trusted by the request guard.
+                Configure CORS separately when browser JavaScript must read
+                cross-origin responses.
             sockets: Pre-bound sockets to pass to Uvicorn
         """
         # Allow stateless as alias for stateless_http
@@ -291,6 +300,9 @@ class TransportMixin:
             middleware=middleware,
             json_response=json_response,
             stateless_http=stateless_http,
+            host_origin_protection=host_origin_protection,
+            allowed_hosts=allowed_hosts,
+            allowed_origins=allowed_origins,
         )
 
         # Display server banner
@@ -333,6 +345,9 @@ class TransportMixin:
         transport: Literal["http", "streamable-http", "sse"] = "http",
         event_store: EventStore | None = None,
         retry_interval: int | None = None,
+        host_origin_protection: bool | None = None,
+        allowed_hosts: list[str] | None = None,
+        allowed_origins: list[str] | None = None,
     ) -> StarletteWithLifespan:
         """Create a Starlette app using the specified HTTP transport.
 
@@ -349,6 +364,12 @@ class TransportMixin:
                 Controls how quickly clients should reconnect after server-initiated
                 disconnections. Requires event_store to be set. Only used with
                 streamable-http transport.
+            host_origin_protection: Whether to validate Host and Origin headers
+                before requests reach the MCP endpoint.
+            allowed_hosts: Additional hostnames that may appear in the Host header.
+            allowed_origins: Additional browser origins trusted by the request guard.
+                Configure CORS separately when browser JavaScript must read
+                cross-origin responses.
 
         Returns:
             A Starlette application configured with the specified transport
@@ -375,6 +396,21 @@ class TransportMixin:
                 ),
                 debug=fastmcp.settings.debug,
                 middleware=middleware,
+                host_origin_protection=(
+                    host_origin_protection
+                    if host_origin_protection is not None
+                    else fastmcp.settings.http_host_origin_protection
+                ),
+                allowed_hosts=(
+                    allowed_hosts
+                    if allowed_hosts is not None
+                    else fastmcp.settings.http_allowed_hosts
+                ),
+                allowed_origins=(
+                    allowed_origins
+                    if allowed_origins is not None
+                    else fastmcp.settings.http_allowed_origins
+                ),
             )
         elif transport == "sse":
             return create_sse_app(

fastmcp_slim/fastmcp/settings.py

@@ -320,6 +320,9 @@ class Settings(BaseSettings):
     stateless_http: bool = (
         False  # If True, uses true stateless mode (new transport per request)
     )
+    http_host_origin_protection: bool = True
+    http_allowed_hosts: list[str] | None = None
+    http_allowed_origins: list[str] | None = None
 
     mounted_components_raise_on_load_error: Annotated[
         bool,

tests/conformance/expected-failures.yml

@@ -3,4 +3,3 @@ server:
   - server-sse-polling
   - resources-subscribe
   - resources-unsubscribe
-  - dns-rebinding-protection

tests/server/http/test_http_auth_middleware.py

@@ -7,6 +7,17 @@ from fastmcp.server import FastMCP
 from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
 from fastmcp.server.http import create_streamable_http_app
 
+INITIALIZE_REQUEST = {
+    "jsonrpc": "2.0",
+    "id": 1,
+    "method": "initialize",
+    "params": {
+        "protocolVersion": "2024-11-05",
+        "capabilities": {},
+        "clientInfo": {"name": "attacker", "version": "0.1"},
+    },
+}
+
 
 class TestStreamableHTTPAppResourceMetadataURL:
     """Test resource_metadata_url logic in create_streamable_http_app."""
@@ -92,3 +103,204 @@ class TestStreamableHTTPAppResourceMetadataURL:
             response = client.post("/mcp")
             assert response.status_code == 401
             assert "www-authenticate" in response.headers
+
+
+class TestStreamableHTTPHostOriginProtection:
+    """Test host and origin validation for streamable HTTP apps."""
+
+    def test_rejects_untrusted_host_before_session_initialization(self):
+        server = FastMCP(name="TestServer")
+        app = create_streamable_http_app(
+            server=server,
+            streamable_http_path="/mcp",
+        )
+
+        with TestClient(app, base_url="http://127.0.0.1") as client:
+            response = client.post(
+                "/mcp",
+                headers={
+                    "accept": "application/json, text/event-stream",
+                    "host": "attacker.example",
+                },
+                json=INITIALIZE_REQUEST,
+            )
+
+        assert response.status_code == 421
+        assert "mcp-session-id" not in response.headers
+
+    def test_rejects_untrusted_origin_before_session_initialization(self):
+        server = FastMCP(name="TestServer")
+        app = create_streamable_http_app(
+            server=server,
+            streamable_http_path="/mcp",
+        )
+
+        with TestClient(app, base_url="http://127.0.0.1") as client:
+            response = client.post(
+                "/mcp",
+                headers={
+                    "accept": "application/json, text/event-stream",
+                    "origin": "https://attacker.example",
+                },
+                json=INITIALIZE_REQUEST,
+            )
+
+        assert response.status_code == 403
+        assert "mcp-session-id" not in response.headers
+
+    def test_allows_configured_host_and_origin(self):
+        server = FastMCP(name="TestServer")
+        app = create_streamable_http_app(
+            server=server,
+            streamable_http_path="/mcp",
+            allowed_hosts=["mcp.example.com"],
+            allowed_origins=["https://app.example.com"],
+        )
+
+        with TestClient(app, base_url="http://127.0.0.1") as client:
+            response = client.post(
+                "/mcp",
+                headers={
+                    "accept": "application/json, text/event-stream",
+                    "host": "mcp.example.com",
+                    "origin": "https://app.example.com",
+                },
+                json=INITIALIZE_REQUEST,
+            )
+
+        assert response.status_code == 200
+        assert "mcp-session-id" in response.headers
+
+    def test_allows_same_request_origin(self):
+        server = FastMCP(name="TestServer")
+        app = create_streamable_http_app(
+            server=server,
+            streamable_http_path="/mcp",
+            allowed_hosts=["mcp.example.com"],
+        )
+
+        with TestClient(app, base_url="https://mcp.example.com") as client:
+            response = client.post(
+                "/mcp",
+                headers={
+                    "accept": "application/json, text/event-stream",
+                    "origin": "https://mcp.example.com",
+                },
+                json=INITIALIZE_REQUEST,
+            )
+
+        assert response.status_code == 200
+        assert "mcp-session-id" in response.headers
+
+    def test_allows_loopback_origin_for_loopback_host(self):
+        server = FastMCP(name="TestServer")
+        app = create_streamable_http_app(
+            server=server,
+            streamable_http_path="/mcp",
+        )
+
+        with TestClient(app, base_url="http://127.0.0.1") as client:
+            response = client.post(
+                "/mcp",
+                headers={
+                    "accept": "application/json, text/event-stream",
+                    "origin": "http://localhost:3000",
+                },
+                json=INITIALIZE_REQUEST,
+            )
+
+        assert response.status_code == 200
+        assert "mcp-session-id" in response.headers
+
+    def test_rejects_loopback_origin_for_public_host(self):
+        server = FastMCP(name="TestServer")
+        app = create_streamable_http_app(
+            server=server,
+            streamable_http_path="/mcp",
+            allowed_hosts=["mcp.example.com"],
+        )
+
+        with TestClient(app, base_url="https://mcp.example.com") as client:
+            response = client.post(
+                "/mcp",
+                headers={
+                    "accept": "application/json, text/event-stream",
+                    "origin": "http://localhost:3000",
+                },
+                json=INITIALIZE_REQUEST,
+            )
+
+        assert response.status_code == 403
+        assert "mcp-session-id" not in response.headers
+
+    def test_allows_configured_loopback_origin_for_public_host(self):
+        server = FastMCP(name="TestServer")
+        app = create_streamable_http_app(
+            server=server,
+            streamable_http_path="/mcp",
+            allowed_hosts=["mcp.example.com"],
+            allowed_origins=["http://localhost:3000"],
+        )
+
+        with TestClient(app, base_url="https://mcp.example.com") as client:
+            response = client.post(
+                "/mcp",
+                headers={
+                    "accept": "application/json, text/event-stream",
+                    "origin": "http://localhost:3000",
+                },
+                json=INITIALIZE_REQUEST,
+            )
+
+        assert response.status_code == 200
+        assert "mcp-session-id" in response.headers
+
+    @pytest.mark.parametrize(
+        "origin",
+        [
+            "http://mcp.example.com",
+            "https://mcp.example.com:3000",
+        ],
+    )
+    def test_rejects_same_host_different_origin(self, origin: str):
+        server = FastMCP(name="TestServer")
+        app = create_streamable_http_app(
+            server=server,
+            streamable_http_path="/mcp",
+            allowed_hosts=["mcp.example.com"],
+        )
+
+        with TestClient(app, base_url="https://mcp.example.com") as client:
+            response = client.post(
+                "/mcp",
+                headers={
+                    "accept": "application/json, text/event-stream",
+                    "origin": origin,
+                },
+                json=INITIALIZE_REQUEST,
+            )
+
+        assert response.status_code == 403
+        assert "mcp-session-id" not in response.headers
+
+    def test_can_disable_host_origin_protection(self):
+        server = FastMCP(name="TestServer")
+        app = create_streamable_http_app(
+            server=server,
+            streamable_http_path="/mcp",
+            host_origin_protection=False,
+        )
+
+        with TestClient(app, base_url="http://127.0.0.1") as client:
+            response = client.post(
+                "/mcp",
+                headers={
+                    "accept": "application/json, text/event-stream",
+                    "host": "attacker.example",
+                    "origin": "https://attacker.example",
+                },
+                json=INITIALIZE_REQUEST,
+            )
+
+        assert response.status_code == 200
+        assert "mcp-session-id" in response.headers