Add support for DNS rebinding protections

src/mcp/server/fastmcp/server.py

@@ -49,6 +49,7 @@
 from mcp.server.stdio import stdio_server
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.context import LifespanContextT, RequestContext, RequestT
 from mcp.types import (
     AnyFunction,
@@ -119,6 +120,9 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
 
     auth: AuthSettings | None = None
 
+    # Transport security settings (DNS rebinding protection)
+    transport_security: TransportSecuritySettings | None = None
+
 
 def lifespan_wrapper(
     app: FastMCP,
@@ -645,6 +649,7 @@ def sse_app(self, mount_path: str | None = None) -> Starlette:
 
         sse = SseServerTransport(
             normalized_message_endpoint,
+            security_settings=self.settings.transport_security,
         )
 
         async def handle_sse(scope: Scope, receive: Receive, send: Send):
@@ -750,6 +755,7 @@ def streamable_http_app(self) -> Starlette:
                 event_store=self._event_store,
                 json_response=self.settings.json_response,
                 stateless=self.settings.stateless_http,  # Use the stateless setting
+                security_settings=self.settings.transport_security,
             )
 
         # Create the ASGI handler

src/mcp/server/sse.py

@@ -52,6 +52,10 @@ async def handle_sse(request):
 from starlette.types import Receive, Scope, Send
 
 import mcp.types as types
+from mcp.server.transport_security import (
+    TransportSecurityMiddleware,
+    TransportSecuritySettings,
+)
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
 
 logger = logging.getLogger(__name__)
@@ -71,16 +75,22 @@ class SseServerTransport:
 
     _endpoint: str
     _read_stream_writers: dict[UUID, MemoryObjectSendStream[SessionMessage | Exception]]
+    _security: TransportSecurityMiddleware
 
-    def __init__(self, endpoint: str) -> None:
+    def __init__(self, endpoint: str, security_settings: TransportSecuritySettings | None = None) -> None:
         """
         Creates a new SSE server transport, which will direct the client to POST
         messages to the relative or absolute URL given.
+
+        Args:
+            endpoint: The relative or absolute URL for POST messages.
+            security_settings: Optional security settings for DNS rebinding protection.
         """
 
         super().__init__()
         self._endpoint = endpoint
         self._read_stream_writers = {}
+        self._security = TransportSecurityMiddleware(security_settings)
         logger.debug(f"SseServerTransport initialized with endpoint: {endpoint}")
 
     @asynccontextmanager
@@ -89,6 +99,13 @@ async def connect_sse(self, scope: Scope, receive: Receive, send: Send):
             logger.error("connect_sse received non-HTTP request")
             raise ValueError("connect_sse can only handle HTTP requests")
 
+        # Validate request headers for DNS rebinding protection
+        request = Request(scope, receive)
+        error_response = await self._security.validate_request(request, is_post=False)
+        if error_response:
+            await error_response(scope, receive, send)
+            raise ValueError("Request validation failed")
+
         logger.debug("Setting up SSE connection")
         read_stream: MemoryObjectReceiveStream[SessionMessage | Exception]
         read_stream_writer: MemoryObjectSendStream[SessionMessage | Exception]
@@ -160,6 +177,11 @@ async def handle_post_message(self, scope: Scope, receive: Receive, send: Send)
         logger.debug("Handling POST message")
         request = Request(scope, receive)
 
+        # Validate request headers for DNS rebinding protection
+        error_response = await self._security.validate_request(request, is_post=True)
+        if error_response:
+            return await error_response(scope, receive, send)
+
         session_id_param = request.query_params.get("session_id")
         if session_id_param is None:
             logger.warning("Received request without session_id")

src/mcp/server/streamable_http.py

@@ -24,6 +24,10 @@
 from starlette.responses import Response
 from starlette.types import Receive, Scope, Send
 
+from mcp.server.transport_security import (
+    TransportSecurityMiddleware,
+    TransportSecuritySettings,
+)
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.types import (
     INTERNAL_ERROR,
@@ -127,12 +131,14 @@ class StreamableHTTPServerTransport:
     _read_stream: MemoryObjectReceiveStream[SessionMessage | Exception] | None = None
     _write_stream: MemoryObjectSendStream[SessionMessage] | None = None
     _write_stream_reader: MemoryObjectReceiveStream[SessionMessage] | None = None
+    _security: TransportSecurityMiddleware
 
     def __init__(
         self,
         mcp_session_id: str | None,
         is_json_response_enabled: bool = False,
         event_store: EventStore | None = None,
+        security_settings: TransportSecuritySettings | None = None,
     ) -> None:
         """
         Initialize a new StreamableHTTP server transport.
@@ -145,6 +151,7 @@ def __init__(
             event_store: Event store for resumability support. If provided,
                         resumability will be enabled, allowing clients to
                         reconnect and resume messages.
+            security_settings: Optional security settings for DNS rebinding protection.
 
         Raises:
             ValueError: If the session ID contains invalid characters.
@@ -155,6 +162,7 @@ def __init__(
         self.mcp_session_id = mcp_session_id
         self.is_json_response_enabled = is_json_response_enabled
         self._event_store = event_store
+        self._security = TransportSecurityMiddleware(security_settings)
         self._request_streams: dict[
             RequestId,
             tuple[
@@ -248,6 +256,14 @@ async def _clean_up_memory_streams(self, request_id: RequestId) -> None:
     async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> None:
         """Application entry point that handles all HTTP requests"""
         request = Request(scope, receive)
+
+        # Validate request headers for DNS rebinding protection
+        is_post = request.method == "POST"
+        error_response = await self._security.validate_request(request, is_post=is_post)
+        if error_response:
+            await error_response(scope, receive, send)
+            return
+
         if self._terminated:
             # If the session has been terminated, return 404 Not Found
             response = self._create_error_response(

src/mcp/server/streamable_http_manager.py

@@ -22,6 +22,7 @@
     EventStore,
     StreamableHTTPServerTransport,
 )
+from mcp.server.transport_security import TransportSecuritySettings
 
 logger = logging.getLogger(__name__)
 
@@ -60,11 +61,13 @@ def __init__(
         event_store: EventStore | None = None,
         json_response: bool = False,
         stateless: bool = False,
+        security_settings: TransportSecuritySettings | None = None,
     ):
         self.app = app
         self.event_store = event_store
         self.json_response = json_response
         self.stateless = stateless
+        self.security_settings = security_settings
 
         # Session tracking (only used if not stateless)
         self._session_creation_lock = anyio.Lock()
@@ -162,6 +165,7 @@ async def _handle_stateless_request(
             mcp_session_id=None,  # No session tracking in stateless mode
             is_json_response_enabled=self.json_response,
             event_store=None,  # No event store in stateless mode
+            security_settings=self.security_settings,
         )
 
         # Start server in a new task
@@ -217,6 +221,7 @@ async def _handle_stateful_request(
                     mcp_session_id=new_session_id,
                     is_json_response_enabled=self.json_response,
                     event_store=self.event_store,  # May be None (no resumability)
+                    security_settings=self.security_settings,
                 )
 
                 assert http_transport.mcp_session_id is not None

src/mcp/server/transport_security.py

@@ -0,0 +1,127 @@
+"""DNS rebinding protection for MCP server transports."""
+
+import logging
+
+from pydantic import BaseModel, Field
+from starlette.requests import Request
+from starlette.responses import Response
+
+logger = logging.getLogger(__name__)
+
+
+class TransportSecuritySettings(BaseModel):
+    """Settings for MCP transport security features.
+
+    These settings help protect against DNS rebinding attacks by validating
+    incoming request headers.
+    """
+
+    enable_dns_rebinding_protection: bool = Field(
+        default=True,
+        description="Enable DNS rebinding protection (recommended for production)",
+    )
+
+    allowed_hosts: list[str] = Field(
+        default=[],
+        description="List of allowed Host header values. Only applies when "
+        + "enable_dns_rebinding_protection is True.",
+    )
+
+    allowed_origins: list[str] = Field(
+        default=[],
+        description="List of allowed Origin header values. Only applies when "
+        + "enable_dns_rebinding_protection is True.",
+    )
+
+
+class TransportSecurityMiddleware:
+    """Middleware to enforce DNS rebinding protection for MCP transport endpoints."""
+
+    def __init__(self, settings: TransportSecuritySettings | None = None):
+        # If not specified, disable DNS rebinding protection by default
+        # for backwards compatibility
+        self.settings = settings or TransportSecuritySettings(enable_dns_rebinding_protection=False)
+
+    def _validate_host(self, host: str | None) -> bool:
+        """Validate the Host header against allowed values."""
+        if not host:
+            logger.warning("Missing Host header in request")
+            return False
+
+        # Check exact match first
+        if host in self.settings.allowed_hosts:
+            return True
+
+        # Check wildcard port patterns
+        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
+
+        logger.warning(f"Invalid Host header: {host}")
+        return False
+
+    def _validate_origin(self, origin: str | None) -> bool:
+        """Validate the Origin header against allowed values."""
+        # Origin can be absent for same-origin requests
+        if not origin:
+            return True
+
+        # Check exact match first
+        if origin in self.settings.allowed_origins:
+            return True
+
+        # Check wildcard port patterns
+        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
+
+        logger.warning(f"Invalid Origin header: {origin}")
+        return False
+
+    def _validate_content_type(self, content_type: str | None) -> bool:
+        """Validate the Content-Type header for POST requests."""
+        if not content_type:
+            logger.warning("Missing Content-Type header in POST request")
+            return False
+
+        # Content-Type must start with application/json
+        if not content_type.lower().startswith("application/json"):
+            logger.warning(f"Invalid Content-Type header: {content_type}")
+            return False
+
+        return True
+
+    async def validate_request(self, request: Request, is_post: bool = False) -> Response | None:
+        """Validate request headers for DNS rebinding protection.
+
+        Returns None if validation passes, or an error Response if validation fails.
+        """
+        # Always validate Content-Type for POST requests
+        if is_post:
+            content_type = request.headers.get("content-type")
+            if not self._validate_content_type(content_type):
+                return Response("Invalid Content-Type header", status_code=400)
+
+        # Skip remaining validation if DNS rebinding protection is disabled
+        if not self.settings.enable_dns_rebinding_protection:
+            return None
+
+        # Validate Host header
+        host = request.headers.get("host")
+        if not self._validate_host(host):
+            return Response("Invalid Host header", status_code=421)
+
+        # Validate Origin header
+        origin = request.headers.get("origin")
+        if not self._validate_origin(origin):
+            return Response("Invalid Origin header", status_code=400)
+
+        return None

tests/server/fastmcp/test_integration.py

@@ -22,8 +22,9 @@
 from mcp.client.session import ClientSession
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamablehttp_client
-from mcp.server.fastmcp import FastMCP
+from mcp.server.fastmcp import Context, FastMCP
 from mcp.server.fastmcp.resources import FunctionResource
+from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.context import RequestContext
 from mcp.types import (
     CreateMessageRequestParams,
@@ -82,7 +83,10 @@ def stateless_http_server_url(stateless_http_server_port: int) -> str:
 # Create a function to make the FastMCP server app
 def make_fastmcp_app():
     """Create a FastMCP server without auth settings."""
-    mcp = FastMCP(name="NoAuthServer")
+    transport_security = TransportSecuritySettings(
+        allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"]
+    )
+    mcp = FastMCP(name="NoAuthServer", transport_security=transport_security)
 
     # Add a simple tool
     @mcp.tool(description="A simple echo tool")
@@ -97,9 +101,10 @@ def echo(message: str) -> str:
 
 def make_everything_fastmcp() -> FastMCP:
     """Create a FastMCP server with all features enabled for testing."""
-    from mcp.server.fastmcp import Context
-
-    mcp = FastMCP(name="EverythingServer")
+    transport_security = TransportSecuritySettings(
+        allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"]
+    )
+    mcp = FastMCP(name="EverythingServer", transport_security=transport_security)
 
     # Tool with context for logging and progress
     @mcp.tool(description="A tool that demonstrates logging and progress")
@@ -231,8 +236,10 @@ def make_everything_fastmcp_app():
 
 def make_fastmcp_streamable_http_app():
     """Create a FastMCP server with StreamableHTTP transport."""
-
-    mcp = FastMCP(name="NoAuthServer")
+    transport_security = TransportSecuritySettings(
+        allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"]
+    )
+    mcp = FastMCP(name="NoAuthServer", transport_security=transport_security)
 
     # Add a simple tool
     @mcp.tool(description="A simple echo tool")
@@ -257,8 +264,10 @@ def make_everything_fastmcp_streamable_http_app():
 
 def make_fastmcp_stateless_http_app():
     """Create a FastMCP server with stateless StreamableHTTP transport."""
-
-    mcp = FastMCP(name="StatelessServer", stateless_http=True)
+    transport_security = TransportSecuritySettings(
+        allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"]
+    )
+    mcp = FastMCP(name="StatelessServer", stateless_http=True, transport_security=transport_security)
 
     # Add a simple tool
     @mcp.tool(description="A simple echo tool")