fix(mcp): cap transport response bodies to bound a hostile tools/call payload

apps/sim/lib/mcp/pinned-fetch.test.ts

@@ -43,17 +43,50 @@ describe('createGuardedMcpFetch', () => {
     })
   })
 
-  it('builds the transport on the guarded connector with no response cap (streaming)', () => {
+  it('builds the transport on the guarded connector with no dispatcher-level response cap', () => {
     const { close } = createGuardedMcpFetch()
 
-    // No options: no `allowH2` opt-in (h1.1 default) and no maxResponseSize —
-    // the long-lived transport must stream unbounded SSE.
+    // No dispatcher options: no `allowH2` opt-in (h1.1 default) and no Agent-level
+    // maxResponseSize — the standalone GET SSE stream must stream unbounded (the body cap
+    // is applied per-response to non-GET exchanges instead).
     expect(mockCreateGuardedFetchWithDispatcher).toHaveBeenCalledWith()
 
-    // close() tears down the pooled sockets (incl. the long-lived SSE) on disconnect.
     void close()
     expect(mockDestroy).toHaveBeenCalledTimes(1)
   })
+
+  it('caps an oversized non-GET response body but leaves the GET SSE stream unbounded', async () => {
+    const big = new Uint8Array(20 * 1024 * 1024) // 20 MiB > 16 MiB cap
+    const makeBody = () =>
+      new ReadableStream<Uint8Array>({
+        start(c) {
+          c.enqueue(big)
+          c.close()
+        },
+      })
+    sentinelFetch.mockImplementation(async () => new Response(makeBody()))
+    const { fetch: guarded } = createGuardedMcpFetch()
+
+    // A POST (tools/call) body over the cap errors when read.
+    const post = await guarded('https://mcp.example/mcp', { method: 'POST' })
+    await expect(new Response(post.body).arrayBuffer()).rejects.toThrow(/exceeded \d+ bytes/)
+
+    // The standalone GET SSE stream is not capped — its body streams through.
+    const get = await guarded('https://mcp.example/mcp', { method: 'GET' })
+    await expect(new Response(get.body).arrayBuffer()).resolves.toBeInstanceOf(ArrayBuffer)
+  })
+
+  it('preserves url and redirected on a capped response (SDK auth-metadata resolution)', async () => {
+    const small = new Response('{"ok":true}', { status: 200 })
+    Object.defineProperty(small, 'url', { value: 'https://mcp.example/mcp' })
+    Object.defineProperty(small, 'redirected', { value: true })
+    sentinelFetch.mockImplementation(async () => small)
+    const { fetch: guarded } = createGuardedMcpFetch()
+
+    const res = await guarded('https://mcp.example/mcp', { method: 'POST' })
+    expect(res.url).toBe('https://mcp.example/mcp')
+    expect(res.redirected).toBe(true)
+  })
 })
 
 describe('createSsrfGuardedMcpFetch', () => {

apps/sim/lib/mcp/pinned-fetch.ts

@@ -36,6 +36,57 @@ export interface GuardedMcpFetch {
  * the transport has no concurrency to gain from h2. `close` tears down pooled sockets
  * (incl. the SSE connection) on disconnect.
  */
+/**
+ * Byte ceiling for a single request/response exchange on the transport (JSON-RPC
+ * results, `initialize`). A hostile server could otherwise stream an unbounded
+ * `tools/call` body and OOM the process. Applied ONLY to non-GET responses — the
+ * standalone GET SSE notification stream is deliberately long-lived and would be
+ * broken by a cumulative cap. Mirrors LibreChat's transport response-size cap.
+ */
+const MAX_TRANSPORT_RESPONSE_BYTES = 16 * 1024 * 1024
+
+/** True for the standalone server→client SSE stream (GET), which must stay uncapped. */
+function isStandaloneStream(method: string): boolean {
+  return method.toUpperCase() === 'GET'
+}
+
+/**
+ * Wraps a response so its body errors once it exceeds `maxBytes`, without buffering —
+ * bytes are counted as they stream, so an oversized body aborts the SDK's read instead
+ * of accumulating in memory. Passthrough for normal-sized responses.
+ */
+function capResponseBody(response: Response, maxBytes: number): Response {
+  if (!response.body) return response
+  let seen = 0
+  const limited = response.body.pipeThrough(
+    new TransformStream<Uint8Array, Uint8Array>({
+      transform(chunk, controller) {
+        seen += chunk.byteLength
+        if (seen > maxBytes) {
+          controller.error(new McpError(`MCP response body exceeded ${maxBytes} bytes`))
+          return
+        }
+        controller.enqueue(chunk)
+      },
+    })
+  )
+  const headers = new Headers(response.headers)
+  // The wrapped body is the already-decoded stream; drop framing headers that would
+  // misdescribe it (consistent with `bufferUnderDeadline`).
+  headers.delete('content-encoding')
+  headers.delete('content-length')
+  const wrapped = new Response(limited, {
+    status: response.status,
+    statusText: response.statusText,
+    headers,
+  })
+  // `new Response()` resets `url`/`redirected` (empty/false); the SDK resolves relative
+  // auth-metadata URLs (e.g. `resource_metadata`) against `response.url`, so carry them over.
+  Object.defineProperty(wrapped, 'url', { value: response.url, configurable: true })
+  Object.defineProperty(wrapped, 'redirected', { value: response.redirected, configurable: true })
+  return wrapped
+}
+
 /**
  * Legacy single-IP pin, kept ONLY for self-hosted private/loopback resolutions
  * (a DNS alias the policy explicitly permits): the guarded lookup would filter
@@ -45,7 +96,14 @@ export interface GuardedMcpFetch {
  */
 export function createPinnedPrivateMcpFetch(resolvedIP: string): GuardedMcpFetch {
   const { fetch: pinnedFetch, dispatcher } = createPinnedFetchWithDispatcher(resolvedIP)
-  return { fetch: pinnedFetch, close: () => dispatcher.destroy() }
+  const capped: typeof fetch = async (input, init) => {
+    const method = init?.method ?? (input instanceof Request ? input.method : 'GET')
+    const response = await pinnedFetch(input, init)
+    return isStandaloneStream(method)
+      ? response
+      : capResponseBody(response, MAX_TRANSPORT_RESPONSE_BYTES)
+  }
+  return { fetch: capped, close: () => dispatcher.destroy() }
 }
 
 export function createGuardedMcpFetch(): GuardedMcpFetch {
@@ -75,7 +133,9 @@ export function createGuardedMcpFetch(): GuardedMcpFetch {
         status: response.status,
         ttfbMs: Date.now() - startedAt,
       })
-      return response
+      return isStandaloneStream(method)
+        ? response
+        : capResponseBody(response, MAX_TRANSPORT_RESPONSE_BYTES)
     } catch (error) {
       const e = error as { name?: string; code?: string; cause?: { name?: string; code?: string } }
       transportLogger.warn('MCP transport request failed', {