Check Host and Origin before answering HEAD, and send nothing after the SSE stream

Author: Kludex

src/mcp/server/mcpserver/server.py

@@ -51,7 +51,8 @@
 from starlette.middleware.authentication import AuthenticationMiddleware
 from starlette.requests import Request
 from starlette.responses import Response
-from starlette.routing import Mount, Route, request_response
+from starlette.routing import Mount, Route
+from starlette.types import ASGIApp, Receive, Scope, Send
 
 from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
 from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
@@ -98,7 +99,11 @@
     StreamableHTTPSessionManager,
 )
 from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler, SubscriptionBus
-from mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, TransportSecuritySettings
+from mcp.server.transport_security import (
+    DEFAULT_MAX_REQUEST_BODY_SIZE,
+    TransportSecurityMiddleware,
+    TransportSecuritySettings,
+)
 from mcp.shared.exceptions import MCPError
 from mcp.shared.uri_template import UriTemplate
 
@@ -1171,23 +1176,23 @@ def sse_app(
             message_path, security_settings=transport_security, max_request_body_size=max_request_body_size
         )
 
-        async def sse_endpoint(request: Request) -> Response:
-            if request.method == "HEAD":
-                # GET's headers without its never-ending stream; a Content-Length would claim an empty one.
-                response = Response(
-                    media_type="text/event-stream", headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}
-                )
-                del response.headers["content-length"]
-                return response
-            async with sse.connect_sse(  # pragma: no cover
-                request.scope,
-                request.receive,
-                request._send,  # type: ignore[reportPrivateUsage]
-            ) as streams:
+        security = TransportSecurityMiddleware(transport_security)
+
+        async def handle_sse(scope: Scope, receive: Receive, send: Send) -> None:
+            if scope["method"] == "HEAD":
+                response = await security.validate_request(Request(scope, receive))
+                if response is None:
+                    # GET's headers without its never-ending stream; a Content-Length would claim an empty one.
+                    response = Response(
+                        media_type="text/event-stream", headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}
+                    )
+                    del response.headers["content-length"]
+                await response(scope, receive, send)
+                return
+            async with sse.connect_sse(scope, receive, send) as streams:  # pragma: no cover
                 await self._lowlevel_server.run(
                     streams[0], streams[1], self._lowlevel_server.create_initialization_options()
                 )
-            return Response()  # pragma: no cover
 
         # Create routes
         routes: list[Route | Mount] = []
@@ -1245,9 +1250,7 @@ async def sse_endpoint(request: Request) -> Response:
             routes.append(
                 Route(
                     sse_path,
-                    endpoint=RequireAuthMiddleware(
-                        request_response(sse_endpoint), required_scopes, resource_metadata_url
-                    ),
+                    endpoint=RequireAuthMiddleware(handle_sse, required_scopes, resource_metadata_url),
                     methods=["GET"],
                 )
             )
@@ -1262,7 +1265,7 @@ async def sse_endpoint(request: Request) -> Response:
             routes.append(
                 Route(
                     sse_path,
-                    endpoint=sse_endpoint,
+                    endpoint=_ASGIEndpoint(handle_sse),
                     methods=["GET"],
                 )
             )
@@ -1426,3 +1429,14 @@ def require_client_extension(ctx: ServerRequestContext[Any, Any], identifier: st
             message=f"Client did not declare required extension {identifier!r}",
             data=data.model_dump(by_alias=True, mode="json", exclude_none=True),
         )
+
+
+class _ASGIEndpoint:
+    """Hands a `Route` a raw ASGI app; a plain function would be wrapped as a request/response endpoint,
+    which sends a second response after the SSE stream has already completed."""
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        await self.app(scope, receive, send)

tests/interaction/_requirements.py

@@ -2709,7 +2709,7 @@ def __post_init__(self) -> None:
         source="issue:#3736",
         behavior=(
             "A HEAD request to the SSE endpoint of `MCPServer.sse_app()` is answered at once with the "
-            "headers a GET would get and no body or Content-Length, without opening a session or a stream."
+            "headers a GET would get, no body and no Content-Length, after the same Host/Origin checks as a GET."
         ),
         transports=("sse",),
         note="Only observable over the legacy SSE transport.",

tests/interaction/transports/test_sse.py

@@ -94,20 +94,31 @@ async def test_post_for_an_unknown_session_is_rejected() -> None:
 
 @requirement("transport:sse:head")
 async def test_head_on_the_sse_endpoint_is_answered_with_headers_only() -> None:
-    """`MCPServer.sse_app()` answers HEAD at once instead of streaming forever."""
+    """`MCPServer.sse_app()` answers HEAD at once with GET's headers and no body instead of streaming forever."""
     app = MCPServer("legacy").sse_app(
         transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False)
     )
     async with httpx2.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as http:
         with anyio.fail_after(5):
             response = await http.head("/sse")
-    assert (response.status_code, dict(response.headers)) == snapshot(
+    assert (response.status_code, dict(response.headers), response.content) == snapshot(
         (
             200,
             {
                 "cache-control": "no-store",
                 "x-accel-buffering": "no",
                 "content-type": "text/event-stream; charset=utf-8",
             },
+            b"",
         )
     )
+
+
+@requirement("transport:sse:head")
+async def test_head_on_the_sse_endpoint_is_subject_to_dns_rebinding_protection() -> None:
+    """A HEAD from a Host the transport security settings reject gets the same refusal a GET would."""
+    app = MCPServer("legacy").sse_app(host="127.0.0.1")
+    async with httpx2.AsyncClient(transport=StreamingASGITransport(app), base_url="http://evil.example") as http:
+        with anyio.fail_after(5):
+            response = await http.head("/sse")
+    assert response.status_code == snapshot(421)