Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 50 additions & 1 deletion src/mcp/server/streamable_http_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
EventStore,
StreamableHTTPServerTransport,
)
from mcp.server.transport_security import TransportSecuritySettings
from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings
from mcp.types import INVALID_REQUEST, ErrorData, JSONRPCError

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -85,6 +85,15 @@ def __init__(
self.retry_interval = retry_interval
self.session_idle_timeout = session_idle_timeout

# Pre-check middleware: the manager runs security validation *before*
# deciding whether to allocate a session. Without this, a request that
# fails security (DNS rebinding, bad Host) would either be rejected too
# late (after a session has already been created and would leak) or with
# the wrong status code (400 "Missing session ID" instead of 421
# "Invalid Host header"). Using the same middleware the transport uses
# keeps the two validation paths consistent.
self._security = TransportSecurityMiddleware(security_settings)

# Session tracking (only used if not stateless)
self._session_creation_lock = anyio.Lock()
self._server_instances: dict[str, StreamableHTTPServerTransport] = {}
Expand Down Expand Up @@ -229,6 +238,16 @@ async def _handle_stateful_request(
send: ASGI send function
"""
request = Request(scope, receive)

# Run security validation FIRST so DNS-rebinding / bad-Host requests
# are rejected with 421 (or the transport-level Content-Type 400)
# regardless of whether a session existed or not — and so bad-Host
# requests can never trigger a session allocation.
security_error = await self._security.validate_request(request, is_post=(request.method == "POST"))
if security_error is not None:
await security_error(scope, receive, send)
return

request_mcp_session_id = request.headers.get(MCP_SESSION_ID_HEADER)

user = scope.get("user")
Expand Down Expand Up @@ -262,6 +281,36 @@ async def _handle_stateful_request(
return

if request_mcp_session_id is None:
# Only POST may initialize a new session (per MCP spec: a session is
# opened by the response to the ``initialize`` JSON-RPC request,
# which is always a POST). GET and DELETE without a session-id are
# protocol errors and must be rejected here — the transport-layer
# rejection happens too late to matter: by the time
# `StreamableHTTPServerTransport.handle_request()` sees the request,
# the session has already been allocated in `_server_instances`
# AND a background `run_server` task has been spawned, waiting on a
# stream that will never receive anything. Both leak forever unless
# `session_idle_timeout` is set (which is opt-in and off by default).
# Other unsupported methods (PUT/PATCH/OPTIONS/...) are intentionally
# let through to the existing ``_handle_unsupported_request`` path,
# which returns 405 — preserved for API stability.
if request.method in ("GET", "DELETE"):
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
Outdated
error_response = JSONRPCError(
jsonrpc="2.0",
id="server-error",
error=ErrorData(
code=INVALID_REQUEST,
message="Bad Request: Missing session ID",
),
)
response = Response(
content=error_response.model_dump_json(by_alias=True, exclude_none=True),
status_code=400,
media_type="application/json",
)
await response(scope, receive, send)
return

# New session case
logger.debug("Creating new transport")
async with self._session_creation_lock:
Expand Down
71 changes: 71 additions & 0 deletions tests/server/test_streamable_http_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -382,6 +382,77 @@ async def capture_send(message: Message):
assert response_start["status"] == 404


@pytest.mark.anyio
@pytest.mark.parametrize("method", ["GET", "DELETE"])
async def test_non_post_without_session_id_does_not_allocate_session(method: str):
"""Regression test: GET/DELETE without a session-id must return 400 without
allocating a session or spawning a background task.

Before this fix, any request without a session id — including GET/DELETE —
entered the "new session" branch of the manager, created a transport,
registered it in ``_server_instances``, and launched a ``run_server`` task
that would wait forever for messages that never come. The transport-level
validation then rejected the request (with 400 or 406), but the allocated
session + task were leaked. Under a Docker healthcheck polling /mcp every
30 seconds this accumulated ~2 sessions/min indefinitely (~1 GiB/week).
"""
app = Server("test-non-post-no-session")
manager = StreamableHTTPSessionManager(app=app)

async with manager.run():
sent_messages: list[Message] = []
response_body = b""

async def mock_send(message: Message):
nonlocal response_body
sent_messages.append(message)
if message["type"] == "http.response.body":
response_body += message.get("body", b"")

scope: Scope = {
"type": "http",
"method": method,
"path": "/mcp",
"headers": [
(b"accept", b"application/json, text/event-stream"),
],
}

async def mock_receive(): # pragma: no cover
return {"type": "http.request", "body": b"", "more_body": False}

# Snapshot before
assert len(manager._server_instances) == 0

await manager.handle_request(scope, mock_receive, mock_send)

# Give any accidentally-spawned background task a chance to register
await anyio.sleep(0.05)

# No session, no task
assert len(manager._server_instances) == 0, (
f"{method} without session-id must not allocate a session — leaked {len(manager._server_instances)}"
)
assert manager._task_group is not None
# anyio TaskGroup internals: no live tasks belonging to run_server
assert len(manager._task_group._tasks) == 0, (
f"{method} without session-id must not spawn a background task — leaked {len(manager._task_group._tasks)}"
)

# Response should be a well-formed JSON-RPC error at status 400
response_start = next(
(msg for msg in sent_messages if msg["type"] == "http.response.start"),
None,
)
assert response_start is not None, "Should have sent a response"
assert response_start["status"] == 400

error_data = json.loads(response_body)
assert error_data["jsonrpc"] == "2.0"
assert error_data["error"]["code"] == INVALID_REQUEST
assert "Missing session ID" in error_data["error"]["message"]


def test_session_idle_timeout_rejects_non_positive():
with pytest.raises(ValueError, match="positive number"):
StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=-1)
Expand Down
Loading