diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index ddcd4df33..a6edffe46 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/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 diff --git a/src/mcp/server/sse.py b/src/mcp/server/sse.py index 52f273968..41145e49f 100644 --- a/src/mcp/server/sse.py +++ b/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") diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 9356a9948..dc5f7a986 100644 --- a/src/mcp/server/streamable_http.py +++ b/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( diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py index 3af9829d1..41b807388 100644 --- a/src/mcp/server/streamable_http_manager.py +++ b/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 diff --git a/src/mcp/server/transport_security.py b/src/mcp/server/transport_security.py new file mode 100644 index 000000000..b7eec474d --- /dev/null +++ b/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=400) + + # Validate Origin header + origin = request.headers.get("origin") + if not self._validate_origin(origin): + return Response("Invalid Origin header", status_code=400) + + return None diff --git a/tests/server/fastmcp/test_integration.py b/tests/server/fastmcp/test_integration.py index 84c59cbda..7128eccf1 100644 --- a/tests/server/fastmcp/test_integration.py +++ b/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") diff --git a/tests/server/test_sse_security.py b/tests/server/test_sse_security.py new file mode 100644 index 000000000..7db729764 --- /dev/null +++ b/tests/server/test_sse_security.py @@ -0,0 +1,293 @@ +"""Tests for SSE server DNS rebinding protection.""" + +import logging +import multiprocessing +import socket +import time + +import httpx +import pytest +import uvicorn +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import Response +from starlette.routing import Mount, Route + +from mcp.server import Server +from mcp.server.sse import SseServerTransport +from mcp.server.transport_security import TransportSecuritySettings +from mcp.types import Tool + +logger = logging.getLogger(__name__) +SERVER_NAME = "test_sse_security_server" + + +@pytest.fixture +def server_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@pytest.fixture +def server_url(server_port: int) -> str: + return f"http://127.0.0.1:{server_port}" + + +class SecurityTestServer(Server): + def __init__(self): + super().__init__(SERVER_NAME) + + async def on_list_tools(self) -> list[Tool]: + return [] + + +def run_server_with_settings(port: int, security_settings: TransportSecuritySettings | None = None): + """Run the SSE server with specified security settings.""" + app = SecurityTestServer() + sse_transport = SseServerTransport("/messages/", security_settings) + + async def handle_sse(request: Request): + try: + async with sse_transport.connect_sse(request.scope, request.receive, request._send) as streams: + if streams: + await app.run(streams[0], streams[1], app.create_initialization_options()) + except ValueError as e: + # Validation error was already handled inside connect_sse + logger.debug(f"SSE connection failed validation: {e}") + return Response() + + routes = [ + Route("/sse", endpoint=handle_sse), + Mount("/messages/", app=sse_transport.handle_post_message), + ] + + starlette_app = Starlette(routes=routes) + uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="error") + + +def start_server_process(port: int, security_settings: TransportSecuritySettings | None = None): + """Start server in a separate process.""" + process = multiprocessing.Process(target=run_server_with_settings, args=(port, security_settings)) + process.start() + # Give server time to start + time.sleep(1) + return process + + +@pytest.mark.anyio +async def test_sse_security_default_settings(server_port: int): + """Test SSE with default security settings (protection disabled).""" + process = start_server_process(server_port) + + try: + headers = {"Host": "evil.com", "Origin": "http://evil.com"} + + async with httpx.AsyncClient(timeout=5.0) as client: + async with client.stream("GET", f"http://127.0.0.1:{server_port}/sse", headers=headers) as response: + assert response.status_code == 200 + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_sse_security_invalid_host_header(server_port: int): + """Test SSE with invalid Host header.""" + # Enable security by providing settings with an empty allowed_hosts list + security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["example.com"]) + process = start_server_process(server_port, security_settings) + + try: + # Test with invalid host header + headers = {"Host": "evil.com"} + + async with httpx.AsyncClient() as client: + response = await client.get(f"http://127.0.0.1:{server_port}/sse", headers=headers) + assert response.status_code == 400 + assert response.text == "Invalid Host header" + + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_sse_security_invalid_origin_header(server_port: int): + """Test SSE with invalid Origin header.""" + # Configure security to allow the host but restrict origins + security_settings = TransportSecuritySettings( + enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"], allowed_origins=["http://localhost:*"] + ) + process = start_server_process(server_port, security_settings) + + try: + # Test with invalid origin header + headers = {"Origin": "http://evil.com"} + + async with httpx.AsyncClient() as client: + response = await client.get(f"http://127.0.0.1:{server_port}/sse", headers=headers) + assert response.status_code == 400 + assert response.text == "Invalid Origin header" + + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_sse_security_post_invalid_content_type(server_port: int): + """Test POST endpoint with invalid Content-Type header.""" + # Configure security to allow the host + security_settings = TransportSecuritySettings( + enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"], allowed_origins=["http://127.0.0.1:*"] + ) + process = start_server_process(server_port, security_settings) + + try: + async with httpx.AsyncClient(timeout=5.0) as client: + # Test POST with invalid content type + fake_session_id = "12345678123456781234567812345678" + response = await client.post( + f"http://127.0.0.1:{server_port}/messages/?session_id={fake_session_id}", + headers={"Content-Type": "text/plain"}, + content="test", + ) + assert response.status_code == 400 + assert response.text == "Invalid Content-Type header" + + # Test POST with missing content type + response = await client.post( + f"http://127.0.0.1:{server_port}/messages/?session_id={fake_session_id}", content="test" + ) + assert response.status_code == 400 + assert response.text == "Invalid Content-Type header" + + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_sse_security_disabled(server_port: int): + """Test SSE with security disabled.""" + settings = TransportSecuritySettings(enable_dns_rebinding_protection=False) + process = start_server_process(server_port, settings) + + try: + # Test with invalid host header - should still work + headers = {"Host": "evil.com"} + + async with httpx.AsyncClient(timeout=5.0) as client: + # For SSE endpoints, we need to use stream to avoid timeout + async with client.stream("GET", f"http://127.0.0.1:{server_port}/sse", headers=headers) as response: + # Should connect successfully even with invalid host + assert response.status_code == 200 + + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_sse_security_custom_allowed_hosts(server_port: int): + """Test SSE with custom allowed hosts.""" + settings = TransportSecuritySettings( + enable_dns_rebinding_protection=True, + allowed_hosts=["localhost", "127.0.0.1", "custom.host"], + allowed_origins=["http://localhost", "http://127.0.0.1", "http://custom.host"], + ) + process = start_server_process(server_port, settings) + + try: + # Test with custom allowed host + headers = {"Host": "custom.host"} + + async with httpx.AsyncClient(timeout=5.0) as client: + # For SSE endpoints, we need to use stream to avoid timeout + async with client.stream("GET", f"http://127.0.0.1:{server_port}/sse", headers=headers) as response: + # Should connect successfully with custom host + assert response.status_code == 200 + + # Test with non-allowed host + headers = {"Host": "evil.com"} + + async with httpx.AsyncClient() as client: + response = await client.get(f"http://127.0.0.1:{server_port}/sse", headers=headers) + assert response.status_code == 400 + assert response.text == "Invalid Host header" + + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_sse_security_wildcard_ports(server_port: int): + """Test SSE with wildcard port patterns.""" + settings = TransportSecuritySettings( + enable_dns_rebinding_protection=True, + allowed_hosts=["localhost:*", "127.0.0.1:*"], + allowed_origins=["http://localhost:*", "http://127.0.0.1:*"], + ) + process = start_server_process(server_port, settings) + + try: + # Test with various port numbers + for test_port in [8080, 3000, 9999]: + headers = {"Host": f"localhost:{test_port}"} + + async with httpx.AsyncClient(timeout=5.0) as client: + # For SSE endpoints, we need to use stream to avoid timeout + async with client.stream("GET", f"http://127.0.0.1:{server_port}/sse", headers=headers) as response: + # Should connect successfully with any port + assert response.status_code == 200 + + headers = {"Origin": f"http://localhost:{test_port}"} + + async with httpx.AsyncClient(timeout=5.0) as client: + # For SSE endpoints, we need to use stream to avoid timeout + async with client.stream("GET", f"http://127.0.0.1:{server_port}/sse", headers=headers) as response: + # Should connect successfully with any port + assert response.status_code == 200 + + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_sse_security_post_valid_content_type(server_port: int): + """Test POST endpoint with valid Content-Type headers.""" + # Configure security to allow the host + security_settings = TransportSecuritySettings( + enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"], allowed_origins=["http://127.0.0.1:*"] + ) + process = start_server_process(server_port, security_settings) + + try: + async with httpx.AsyncClient() as client: + # Test with various valid content types + valid_content_types = [ + "application/json", + "application/json; charset=utf-8", + "application/json;charset=utf-8", + "APPLICATION/JSON", # Case insensitive + ] + + for content_type in valid_content_types: + # Use a valid UUID format (even though session won't exist) + fake_session_id = "12345678123456781234567812345678" + response = await client.post( + f"http://127.0.0.1:{server_port}/messages/?session_id={fake_session_id}", + headers={"Content-Type": content_type}, + json={"test": "data"}, + ) + # Will get 404 because session doesn't exist, but that's OK + # We're testing that it passes the content-type check + assert response.status_code == 404 + assert response.text == "Could not find session" + + finally: + process.terminate() + process.join() diff --git a/tests/server/test_streamable_http_security.py b/tests/server/test_streamable_http_security.py new file mode 100644 index 000000000..b5dc81c4a --- /dev/null +++ b/tests/server/test_streamable_http_security.py @@ -0,0 +1,293 @@ +"""Tests for StreamableHTTP server DNS rebinding protection.""" + +import logging +import multiprocessing +import socket +import time +from collections.abc import AsyncGenerator +from contextlib import asynccontextmanager + +import httpx +import pytest +import uvicorn +from starlette.applications import Starlette +from starlette.routing import Mount +from starlette.types import Receive, Scope, Send + +from mcp.server import Server +from mcp.server.streamable_http_manager import StreamableHTTPSessionManager +from mcp.server.transport_security import TransportSecuritySettings +from mcp.types import Tool + +logger = logging.getLogger(__name__) +SERVER_NAME = "test_streamable_http_security_server" + + +@pytest.fixture +def server_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@pytest.fixture +def server_url(server_port: int) -> str: + return f"http://127.0.0.1:{server_port}" + + +class SecurityTestServer(Server): + def __init__(self): + super().__init__(SERVER_NAME) + + async def on_list_tools(self) -> list[Tool]: + return [] + + +def run_server_with_settings(port: int, security_settings: TransportSecuritySettings | None = None): + """Run the StreamableHTTP server with specified security settings.""" + app = SecurityTestServer() + + # Create session manager with security settings + session_manager = StreamableHTTPSessionManager( + app=app, + json_response=False, + stateless=False, + security_settings=security_settings, + ) + + # Create the ASGI handler + async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None: + await session_manager.handle_request(scope, receive, send) + + # Create Starlette app with lifespan + @asynccontextmanager + async def lifespan(app: Starlette) -> AsyncGenerator[None, None]: + async with session_manager.run(): + yield + + routes = [ + Mount("/", app=handle_streamable_http), + ] + + starlette_app = Starlette(routes=routes, lifespan=lifespan) + uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="error") + + +def start_server_process(port: int, security_settings: TransportSecuritySettings | None = None): + """Start server in a separate process.""" + process = multiprocessing.Process(target=run_server_with_settings, args=(port, security_settings)) + process.start() + # Give server time to start + time.sleep(1) + return process + + +@pytest.mark.anyio +async def test_streamable_http_security_default_settings(server_port: int): + """Test StreamableHTTP with default security settings (protection enabled).""" + process = start_server_process(server_port) + + try: + # Test with valid localhost headers + async with httpx.AsyncClient(timeout=5.0) as client: + # POST request to initialize session + response = await client.post( + f"http://127.0.0.1:{server_port}/", + json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, + headers={ + "Accept": "application/json, text/event-stream", + "Content-Type": "application/json", + }, + ) + assert response.status_code == 200 + assert "mcp-session-id" in response.headers + + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_streamable_http_security_invalid_host_header(server_port: int): + """Test StreamableHTTP with invalid Host header.""" + security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True) + process = start_server_process(server_port, security_settings) + + try: + # Test with invalid host header + headers = { + "Host": "evil.com", + "Accept": "application/json, text/event-stream", + "Content-Type": "application/json", + } + + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.post( + f"http://127.0.0.1:{server_port}/", + json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, + headers=headers, + ) + assert response.status_code == 400 + assert response.text == "Invalid Host header" + + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_streamable_http_security_invalid_origin_header(server_port: int): + """Test StreamableHTTP with invalid Origin header.""" + security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"]) + process = start_server_process(server_port, security_settings) + + try: + # Test with invalid origin header + headers = { + "Origin": "http://evil.com", + "Accept": "application/json, text/event-stream", + "Content-Type": "application/json", + } + + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.post( + f"http://127.0.0.1:{server_port}/", + json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, + headers=headers, + ) + assert response.status_code == 400 + assert response.text == "Invalid Origin header" + + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_streamable_http_security_invalid_content_type(server_port: int): + """Test StreamableHTTP POST with invalid Content-Type header.""" + process = start_server_process(server_port) + + try: + async with httpx.AsyncClient(timeout=5.0) as client: + # Test POST with invalid content type + response = await client.post( + f"http://127.0.0.1:{server_port}/", + headers={ + "Content-Type": "text/plain", + "Accept": "application/json, text/event-stream", + }, + content="test", + ) + assert response.status_code == 400 + assert response.text == "Invalid Content-Type header" + + # Test POST with missing content type + response = await client.post( + f"http://127.0.0.1:{server_port}/", + headers={"Accept": "application/json, text/event-stream"}, + content="test", + ) + assert response.status_code == 400 + assert response.text == "Invalid Content-Type header" + + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_streamable_http_security_disabled(server_port: int): + """Test StreamableHTTP with security disabled.""" + settings = TransportSecuritySettings(enable_dns_rebinding_protection=False) + process = start_server_process(server_port, settings) + + try: + # Test with invalid host header - should still work + headers = { + "Host": "evil.com", + "Accept": "application/json, text/event-stream", + "Content-Type": "application/json", + } + + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.post( + f"http://127.0.0.1:{server_port}/", + json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, + headers=headers, + ) + # Should connect successfully even with invalid host + assert response.status_code == 200 + + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_streamable_http_security_custom_allowed_hosts(server_port: int): + """Test StreamableHTTP with custom allowed hosts.""" + settings = TransportSecuritySettings( + enable_dns_rebinding_protection=True, + allowed_hosts=["localhost", "127.0.0.1", "custom.host"], + allowed_origins=["http://localhost", "http://127.0.0.1", "http://custom.host"], + ) + process = start_server_process(server_port, settings) + + try: + # Test with custom allowed host + headers = { + "Host": "custom.host", + "Accept": "application/json, text/event-stream", + "Content-Type": "application/json", + } + + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.post( + f"http://127.0.0.1:{server_port}/", + json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, + headers=headers, + ) + # Should connect successfully with custom host + assert response.status_code == 200 + finally: + process.terminate() + process.join() + + +@pytest.mark.anyio +async def test_streamable_http_security_get_request(server_port: int): + """Test StreamableHTTP GET request with security.""" + security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1"]) + process = start_server_process(server_port, security_settings) + + try: + # Test GET request with invalid host header + headers = { + "Host": "evil.com", + "Accept": "text/event-stream", + } + + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.get(f"http://127.0.0.1:{server_port}/", headers=headers) + assert response.status_code == 400 + assert response.text == "Invalid Host header" + + # Test GET request with valid host header + headers = { + "Host": "127.0.0.1", + "Accept": "text/event-stream", + } + + async with httpx.AsyncClient(timeout=5.0) as client: + # GET requests need a session ID in StreamableHTTP + # So it will fail with "Missing session ID" not security error + response = await client.get(f"http://127.0.0.1:{server_port}/", headers=headers) + # This should pass security but fail on session validation + assert response.status_code == 400 + body = response.json() + assert "Missing session ID" in body["error"]["message"] + + finally: + process.terminate() + process.join() diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py index 4d8f7717e..8e1912e9b 100644 --- a/tests/shared/test_sse.py +++ b/tests/shared/test_sse.py @@ -20,6 +20,7 @@ from mcp.client.sse import sse_client from mcp.server import Server from mcp.server.sse import SseServerTransport +from mcp.server.transport_security import TransportSecuritySettings from mcp.shared.exceptions import McpError from mcp.types import ( EmptyResult, @@ -80,7 +81,11 @@ async def handle_call_tool(name: str, args: dict) -> list[TextContent]: # Test fixtures def make_server_app() -> Starlette: """Create test Starlette app with SSE transport""" - sse = SseServerTransport("/messages/") + # Configure security with allowed hosts/origins for testing + security_settings = TransportSecuritySettings( + allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"] + ) + sse = SseServerTransport("/messages/", security_settings=security_settings) server = ServerTest() async def handle_sse(request: Request) -> Response: @@ -339,7 +344,11 @@ async def handle_list_tools() -> list[Tool]: def run_context_server(server_port: int) -> None: """Run a server that captures request context""" - sse = SseServerTransport("/messages/") + # Configure security with allowed hosts/origins for testing + security_settings = TransportSecuritySettings( + allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"] + ) + sse = SseServerTransport("/messages/", security_settings=security_settings) context_server = RequestContextServer() async def handle_sse(request: Request) -> Response: diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py index 615e68efc..0f2296fab 100644 --- a/tests/shared/test_streamable_http.py +++ b/tests/shared/test_streamable_http.py @@ -36,6 +36,7 @@ StreamId, ) from mcp.server.streamable_http_manager import StreamableHTTPSessionManager +from mcp.server.transport_security import TransportSecuritySettings from mcp.shared.context import RequestContext from mcp.shared.exceptions import McpError from mcp.shared.message import ( @@ -215,10 +216,14 @@ def create_app(is_json_response_enabled=False, event_store: EventStore | None = server = ServerTest() # Create the session manager + security_settings = TransportSecuritySettings( + allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"] + ) session_manager = StreamableHTTPSessionManager( app=server, event_store=event_store, json_response=is_json_response_enabled, + security_settings=security_settings, ) # Create an ASGI application that uses the session manager @@ -424,8 +429,9 @@ def test_content_type_validation(basic_server, basic_server_url): }, data="This is not JSON", ) - assert response.status_code == 415 - assert "Unsupported Media Type" in response.text + + assert response.status_code == 400 + assert "Invalid Content-Type" in response.text def test_json_validation(basic_server, basic_server_url):