diff --git a/apps/sim/lib/mcp/pinned-fetch.test.ts b/apps/sim/lib/mcp/pinned-fetch.test.ts index 608499f17b1..32f15a96351 100644 --- a/apps/sim/lib/mcp/pinned-fetch.test.ts +++ b/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({ + 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', () => { diff --git a/apps/sim/lib/mcp/pinned-fetch.ts b/apps/sim/lib/mcp/pinned-fetch.ts index 3a83b783cc1..e739b2055f6 100644 --- a/apps/sim/lib/mcp/pinned-fetch.ts +++ b/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({ + 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', {