ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 07:09:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

Relax host origin guard defaults (#4439)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-07-06 20:59:28 -04:00 • committed by GitHub
commit 400db61b8b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 259 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

8
docs/deployment/http.mdx
View file

@ -103,11 +103,11 @@ If you're mounting an authenticated server under a path prefix, see [Mounting Au
### 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.
FastMCP validates `Host` and browser `Origin` headers for Streamable HTTP requests automatically where it can infer a safe request boundary. This protects localhost-bound servers from DNS rebinding attacks while letting ASGI, serverless, and reverse-proxy deployments provide their public hostnames explicitly.
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:
When you deploy behind a public hostname, add the hostname clients use to reach your MCP endpoint. This enables strict Host validation for that deployment. If a browser-based MCP client runs on a separate origin, add that origin as well:
```python
from fastmcp import FastMCP
@ -144,7 +144,7 @@ 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.
Use `host_origin_protection=True` to require strict Host and Origin validation for every request. Use `host_origin_protection=False` only for trusted internal deployments that provide equivalent validation at another layer, such as an ingress proxy.
### Health Checks
@ -201,7 +201,7 @@ 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.
Host and Origin protection runs before CORS when it is active for a request. 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:

2
docs/more/settings.mdx
View file

@ -42,7 +42,7 @@ 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_HOST_ORIGIN_PROTECTION` | `bool \| "auto"` | `auto` | Validate `Host` and browser `Origin` headers for Streamable HTTP requests. `auto` protects localhost-bound servers and explicit host/origin allowlists. |
| `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. |

79
fastmcp_slim/fastmcp/server/http.py
View file

@ -5,7 +5,7 @@ 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 typing import TYPE_CHECKING, Any, Literal
from urllib.parse import urlsplit
from uuid import uuid4
@ -36,6 +36,8 @@ if TYPE_CHECKING:
logger = get_logger(__name__)
DEFAULT_HOSTS = ("127.0.0.1", "localhost", "::1")
HostOriginProtection = bool | Literal["auto"]
HostOriginProtectionMode = Literal["auto", "strict"]
class FastMCPStreamableHTTPSessionManager(StreamableHTTPSessionManager):
@ -228,34 +230,81 @@ class HostOriginGuardMiddleware:
app: ASGIApp,
allowed_hosts: Sequence[str] | None = None,
allowed_origins: Sequence[str] | None = None,
mode: HostOriginProtectionMode = "auto",
) -> None:
self.app = app
self.allowed_hosts = tuple(allowed_hosts or ())
self.allowed_origins = tuple(allowed_origins or ())
self.mode = mode
self.has_explicit_allowed_hosts = allowed_hosts is not None
self.has_explicit_allowed_origins = allowed_origins is not None
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):
if self._should_validate_host(scope) and not _host_matches(
host,
self._allowed_hosts_for_scope(scope),
):
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):
if (
origin
and self._should_validate_origin(scope, host)
and not self._origin_allowed(
origin,
request_origin,
host,
allow_same_origin_fallback=self._allow_same_origin_fallback(
scope,
host,
),
)
):
response = Response("Forbidden Origin", status_code=403)
await response(scope, receive, send)
return
await self.app(scope, receive, send)
def _should_validate_host(self, scope: Scope) -> bool:
if self.mode == "strict" or self.has_explicit_allowed_hosts:
return True
server = scope.get("server")
return bool(server and _is_loopback_host(server[0]))
def _should_validate_origin(self, scope: Scope, host: str) -> bool:
if (
self.mode == "strict"
or self.has_explicit_allowed_hosts
or self.has_explicit_allowed_origins
or _is_loopback_host(host)
):
return True
server = scope.get("server")
return bool(server and _is_loopback_host(server[0]))
def _allow_same_origin_fallback(self, scope: Scope, host: str) -> bool:
if not self.has_explicit_allowed_origins:
return True
if self.mode == "strict" or self.has_explicit_allowed_hosts:
return True
server = scope.get("server")
return _is_loopback_host(host) or bool(server and _is_loopback_host(server[0]))
def _allowed_hosts_for_scope(self, scope: Scope) -> tuple[str, ...]:
allowed_hosts = list(DEFAULT_HOSTS)
allowed_hosts.extend(self.allowed_hosts)
@ -268,10 +317,19 @@ class HostOriginGuardMiddleware:
return tuple(allowed_hosts)
def _origin_allowed(self, origin: str, request_origin: str, host: str) -> bool:
def _origin_allowed(
self,
origin: str,
request_origin: str,
host: str,
allow_same_origin_fallback: bool,
) -> bool:
if _origin_matches(origin, self.allowed_origins):
return True
if not allow_same_origin_fallback:
return False
origin_host = _origin_host(origin)
if _is_loopback_host(origin_host) and _is_loopback_host(host):
return True
@ -491,7 +549,7 @@ def create_streamable_http_app(
debug: bool = False,
routes: list[BaseRoute] | None = None,
middleware: list[Middleware] | None = None,
host_origin_protection: bool = True,
host_origin_protection: HostOriginProtection = "auto",
allowed_hosts: Sequence[str] | None = None,
allowed_origins: Sequence[str] | None = None,
) -> StarletteWithLifespan:
@ -511,7 +569,8 @@ def create_streamable_http_app(
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.
before requests reach the MCP endpoint. "auto" protects
localhost-bound servers and explicit host/origin allowlists.
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
@ -576,13 +635,17 @@ def create_streamable_http_app(
server_routes.extend(server._get_additional_http_routes())
# Add middleware
if host_origin_protection:
if host_origin_protection not in (True, False, "auto"):
raise ValueError("host_origin_protection must be True, False, or 'auto'.")
if host_origin_protection is not False:
server_middleware.insert(
0,
Middleware(
HostOriginGuardMiddleware,
allowed_hosts=allowed_hosts,
allowed_origins=allowed_origins,
mode="strict" if host_origin_protection is True else "auto",
),
)
if middleware:

43
fastmcp_slim/fastmcp/server/mixins/transport.py
View file

@ -19,7 +19,9 @@ from starlette.routing import BaseRoute, Route
import fastmcp
from fastmcp.server.event_store import EventStore
from fastmcp.server.http import (
HostOriginProtection,
StarletteWithLifespan,
_is_loopback_host,
create_sse_app,
create_streamable_http_app,
)
@ -48,6 +50,22 @@ def _format_host_for_url(host: str) -> str:
return host
def _resolve_allowed_hosts_for_run(
*,
host: str,
host_origin_protection: HostOriginProtection,
allowed_hosts: list[str] | None,
configured_allowed_hosts: list[str] | None,
) -> list[str] | None:
if allowed_hosts is not None:
return allowed_hosts
if host_origin_protection == "auto" and _is_loopback_host(host):
return [*(configured_allowed_hosts or []), host]
return configured_allowed_hosts
class TransportMixin:
"""Mixin providing transport-related methods for FastMCP.
@ -250,7 +268,7 @@ class TransportMixin:
json_response: bool | None = None,
stateless_http: bool | None = None,
stateless: bool | None = None,
host_origin_protection: bool | None = None,
host_origin_protection: HostOriginProtection | None = None,
allowed_hosts: list[str] | None = None,
allowed_origins: list[str] | None = None,
sockets: list[socket.socket] | None = None,
@ -269,7 +287,8 @@ class TransportMixin:
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.
before requests reach the MCP endpoint. "auto" protects
localhost-bound servers and explicit host/origin allowlists.
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
@ -290,6 +309,17 @@ class TransportMixin:
host = host if host is not None else fastmcp.settings.host
port = port if port is not None else fastmcp.settings.port
resolved_host_origin_protection = (
host_origin_protection
if host_origin_protection is not None
else fastmcp.settings.http_host_origin_protection
)
resolved_allowed_hosts = _resolve_allowed_hosts_for_run(
host=host,
host_origin_protection=resolved_host_origin_protection,
allowed_hosts=allowed_hosts,
configured_allowed_hosts=fastmcp.settings.http_allowed_hosts,
)
default_log_level_to_use = (
log_level if log_level is not None else fastmcp.settings.log_level
).lower()
@ -300,8 +330,8 @@ class TransportMixin:
middleware=middleware,
json_response=json_response,
stateless_http=stateless_http,
host_origin_protection=host_origin_protection,
allowed_hosts=allowed_hosts,
host_origin_protection=resolved_host_origin_protection,
allowed_hosts=resolved_allowed_hosts,
allowed_origins=allowed_origins,
)
@ -345,7 +375,7 @@ 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,
host_origin_protection: HostOriginProtection | None = None,
allowed_hosts: list[str] | None = None,
allowed_origins: list[str] | None = None,
) -> StarletteWithLifespan:
@ -365,7 +395,8 @@ class TransportMixin:
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.
before requests reach the MCP endpoint. "auto" protects
localhost-bound servers and explicit host/origin allowlists.
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

2
fastmcp_slim/fastmcp/settings.py
View file

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

110
tests/server/http/test_http_auth_middleware.py
View file

@ -1,11 +1,16 @@
from collections.abc import MutableMapping
from typing import Any, Literal
import pytest
from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware
from starlette.responses import Response
from starlette.routing import Route
from starlette.testclient import TestClient
from starlette.types import Receive, Scope, Send
from fastmcp.server import FastMCP
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
from fastmcp.server.http import create_streamable_http_app
from fastmcp.server.http import HostOriginGuardMiddleware, create_streamable_http_app
INITIALIZE_REQUEST = {
"jsonrpc": "2.0",
@ -19,6 +24,59 @@ INITIALIZE_REQUEST = {
}
async def _ok_app(scope: Scope, receive: Receive, send: Send) -> None:
response = Response("OK")
await response(scope, receive, send)
async def _empty_receive() -> dict[str, Any]:
return {"type": "http.request", "body": b"", "more_body": False}
async def _guard_status(
*,
host: str,
origin: str | None = None,
server: tuple[str, int] | None = None,
mode: Literal["auto", "strict"] = "auto",
allowed_hosts: list[str] | None = None,
allowed_origins: list[str] | None = None,
) -> int:
app = HostOriginGuardMiddleware(
_ok_app,
allowed_hosts=allowed_hosts,
allowed_origins=allowed_origins,
mode=mode,
)
headers = [(b"host", host.encode())]
if origin is not None:
headers.append((b"origin", origin.encode()))
scope: Scope = {
"type": "http",
"asgi": {"version": "3.0"},
"http_version": "1.1",
"method": "POST",
"scheme": "https",
"path": "/mcp",
"raw_path": b"/mcp",
"query_string": b"",
"headers": headers,
"client": ("127.0.0.1", 12345),
"server": server,
}
sent_messages: list[MutableMapping[str, Any]] = []
async def send(message: MutableMapping[str, Any]) -> None:
sent_messages.append(message)
await app(scope, _empty_receive, send)
response_start = next(
message for message in sent_messages if message["type"] == "http.response.start"
)
return response_start["status"]
class TestStreamableHTTPAppResourceMetadataURL:
"""Test resource_metadata_url logic in create_streamable_http_app."""
@ -108,6 +166,56 @@ class TestStreamableHTTPAppResourceMetadataURL:
class TestStreamableHTTPHostOriginProtection:
"""Test host and origin validation for streamable HTTP apps."""
async def test_auto_allows_public_host_when_server_scope_is_ambiguous(self):
status = await _guard_status(
host="mcp.example.com",
origin="https://app.example.com",
server=None,
)
assert status == 200
async def test_auto_rejects_untrusted_host_when_server_scope_is_loopback(self):
status = await _guard_status(
host="attacker.example",
origin="https://attacker.example",
server=("127.0.0.1", 8000),
)
assert status == 421
async def test_strict_rejects_public_host_when_server_scope_is_ambiguous(self):
status = await _guard_status(
host="mcp.example.com",
origin="https://app.example.com",
server=None,
mode="strict",
)
assert status == 421
async def test_auto_rejects_same_origin_fallback_without_trusted_host_boundary(
self,
):
status = await _guard_status(
host="attacker.example",
origin="https://attacker.example",
server=None,
allowed_origins=["https://app.example.com"],
)
assert status == 403
async def test_auto_allows_configured_origin_without_trusted_host_boundary(self):
status = await _guard_status(
host="mcp.example.com",
origin="https://app.example.com",
server=None,
allowed_origins=["https://app.example.com"],
)
assert status == 200
def test_rejects_untrusted_host_before_session_initialization(self):
server = FastMCP(name="TestServer")
app = create_streamable_http_app(

23
tests/server/test_transport.py
View file

@ -1,6 +1,9 @@
import pytest
from fastmcp.server.mixins.transport import _format_host_for_url
from fastmcp.server.mixins.transport import (
_format_host_for_url,
_resolve_allowed_hosts_for_run,
)
@pytest.mark.parametrize(
@ -18,3 +21,21 @@ from fastmcp.server.mixins.transport import _format_host_for_url
def test_format_host_for_url(host: str, expected: str):
"""IPv6 hosts are bracketed for use in a URL; everything else is unchanged."""
assert _format_host_for_url(host) == expected
def test_resolve_allowed_hosts_for_run_merges_configured_hosts_with_loopback_host():
assert _resolve_allowed_hosts_for_run(
host="127.0.0.1",
host_origin_protection="auto",
allowed_hosts=None,
configured_allowed_hosts=["mcp.example.com"],
) == ["mcp.example.com", "127.0.0.1"]
def test_resolve_allowed_hosts_for_run_preserves_explicit_hosts():
assert _resolve_allowed_hosts_for_run(
host="127.0.0.1",
host_origin_protection="auto",
allowed_hosts=["mcp.example.com"],
configured_allowed_hosts=["settings.example.com"],
) == ["mcp.example.com"]

14
tests/test_settings.py
View file

@ -37,3 +37,17 @@ def test_get_setting_raises_for_missing_nested_parent():
test_settings.get_setting("docket__missing__value")
assert str(exc_info.value) == "Setting missing does not exist."
@pytest.mark.parametrize(
("value", "expected"),
[
("auto", "auto"),
("true", True),
("false", False),
],
)
def test_http_host_origin_protection_env_var(value, expected, monkeypatch):
monkeypatch.setenv("FASTMCP_HTTP_HOST_ORIGIN_PROTECTION", value)
assert Settings().http_host_origin_protection == expected