fix(mcp): cap HTTP/SSE response bodies before SDK parse
Port from openclaw/openclaw#123194: a hostile or misbehaving remote MCP server could stream an unbounded HTTP catalog/tool-result body that the MCP SDK buffers and JSON-parses before any of Hermes' post-parse limits (resource cap, tool-result truncation) run. New _make_mcp_body_cap_transport wraps the owned httpx AsyncClient's transport on the Streamable HTTP (mcp >= 1.24) and SSE paths: - finite HTTP bodies capped at 10 MiB (Content-Length rejected up front, streamed bodies capped chunk-by-chunk); - each SSE event capped at 10 MiB, with accounting reset at completed event boundaries so long-lived streams/keepalives are unlimited; - violations raise httpx.ReadError naming the byte cap, handled by the existing transport teardown/reconnect path (#66092). verify/cert now live on the inner AsyncHTTPTransport (client-level TLS kwargs are inert once a custom transport is passed); the SSE httpx_client_factory is always injected so the cap applies with default TLS too. Legacy mcp < 1.24 path (SDK-internal client, no hook) stays uncapped — same degradation as strict_redirect_headers.
This commit is contained in:
@@ -3,6 +3,7 @@ headers, redirect header stripping, exception-group unwrapping, auth/session-exp
|
||||
method-not-found detection and connect-error formatting. Split from tools/mcp_tool.py."""
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import errno
|
||||
import importlib
|
||||
import logging
|
||||
@@ -231,6 +232,73 @@ def _make_redirect_header_stripper(original_url, *, strict: bool = False,
|
||||
return _strip_on_cross_origin_redirect
|
||||
|
||||
|
||||
# Wire-body cap, applied at the httpx transport before the SDK buffers/JSON-parses a response. A
|
||||
# hostile or misbehaving remote MCP server can stream an unbounded catalog/tool-result body and none
|
||||
# of the post-parse limits (resource cap, tool-result truncation) run before the parse blows up.
|
||||
# Finite HTTP bodies are capped at this many bytes (a larger Content-Length is rejected up front);
|
||||
# each SSE *event* is capped, with the counter reset at completed event boundaries so a long-lived
|
||||
# stream and its keepalives have no cumulative limit. Violations raise the SDK httpx's ReadError and
|
||||
# flow through the ordinary transport teardown/reconnect path (#66092).
|
||||
_MCP_HTTP_MAX_BODY_BYTES = 10 * 1024 * 1024
|
||||
_SSE_EVENT_BOUNDARIES = (b"\n\n", b"\r\n\r\n")
|
||||
|
||||
|
||||
def _make_mcp_body_cap_transport(httpx_mod, inner_transport, limit: int = _MCP_HTTP_MAX_BODY_BYTES):
|
||||
"""Wrap ``inner_transport`` so every response body is size-capped. ``httpx_mod`` must be the SDK's
|
||||
own httpx module (``sdk_httpx()``): the transport is handed to that SDK's ``AsyncClient``."""
|
||||
|
||||
class _CappedStream(httpx_mod.AsyncByteStream):
|
||||
def __init__(self, inner, is_sse: bool, url: str):
|
||||
self._inner, self._is_sse, self._url = inner, is_sse, url
|
||||
|
||||
def _reject(self, kind: str):
|
||||
return httpx_mod.ReadError(f"MCP {kind} exceeds {limit} bytes (from {self._url})")
|
||||
|
||||
async def __aiter__(self):
|
||||
counted = 0
|
||||
async for chunk in self._inner:
|
||||
if self._is_sse:
|
||||
# Bytes up to the last completed event boundary belong to finished events (they must
|
||||
# still fit the per-event cap together with the carried prefix); the remainder starts
|
||||
# the next event's budget.
|
||||
boundary_end = max(chunk.rfind(sep) + len(sep) if sep in chunk else -1 for sep in _SSE_EVENT_BOUNDARIES)
|
||||
if boundary_end != -1:
|
||||
if counted + boundary_end > limit:
|
||||
raise self._reject("SSE event")
|
||||
counted = len(chunk) - boundary_end
|
||||
else:
|
||||
counted += len(chunk)
|
||||
else:
|
||||
counted += len(chunk)
|
||||
if counted > limit:
|
||||
raise self._reject("SSE event" if self._is_sse else "HTTP response")
|
||||
yield chunk
|
||||
|
||||
async def aclose(self):
|
||||
await self._inner.aclose()
|
||||
|
||||
class _BodyCapTransport(httpx_mod.AsyncBaseTransport):
|
||||
def __init__(self, inner):
|
||||
self._inner = inner
|
||||
|
||||
async def handle_async_request(self, request):
|
||||
response = await self._inner.handle_async_request(request)
|
||||
declared = response.headers.get("content-length")
|
||||
with contextlib.suppress(ValueError): # malformed header: the streamed cap still applies
|
||||
if declared is not None and int(declared) > limit:
|
||||
await response.aclose()
|
||||
raise httpx_mod.ReadError(f"MCP HTTP response declares Content-Length {declared} > {limit} "
|
||||
f"bytes cap (from {request.url})")
|
||||
is_sse = "text/event-stream" in response.headers.get("content-type", "").lower()
|
||||
response.stream = _CappedStream(response.stream, is_sse, str(request.url))
|
||||
return response
|
||||
|
||||
async def aclose(self):
|
||||
await self._inner.aclose()
|
||||
|
||||
return _BodyCapTransport(inner_transport)
|
||||
|
||||
|
||||
def _exc_children(exc: BaseException) -> List[BaseException]:
|
||||
"""Sub-exceptions of a group, else ``__cause__``/``__context__`` when they are exceptions."""
|
||||
nested = getattr(exc, "exceptions", None)
|
||||
|
||||
Reference in New Issue
Block a user