fix(mcp): stream guarded transport via undici.request so SSE responses work under Bun (#5897)

* fix(mcp): stream guarded transport via undici.request so SSE responses work under Bun

undici's fetch exposes its response body as a WHATWG ReadableStream whose
bridge is broken under the Bun runtime the standalone server runs on: headers
arrive but response.body never yields data, hanging every MCP streamable-HTTP
(text/event-stream) tools/list read to its 30s timeout. undici's lower-level
request() returns a Node Readable, which Bun streams natively.

createSsrfGuardedFetchWithDispatcher (the MCP transport builder) now routes
through undiciRequestAsResponse: same guarded Agent + connect.lookup (SSRF
unchanged), request() instead of fetch(), and a hand-rolled Node->Web body
bridge (not Readable.toWeb, which throws ERR_INVALID_STATE on the redirect
body.cancel()). Buffered reads, followRedirectsGuarded, maxResponseSize, and
abort all preserved and verified on both Node and Bun.

* fix(mcp): use a single cast for header-iterable detection to satisfy boundary audit

* fix(mcp): handle URLSearchParams OAuth bodies and copy stream chunks

- toUndiciRequestBody serializes a URLSearchParams body (the MCP SDK's OAuth
  token/refresh exchange sends one); undici.request rejects it otherwise.
- Default content-type application/x-www-form-urlencoded when the caller didn't
  set one (fetch parity).
- nodeReadableToWebStream enqueues a copy (new Uint8Array(chunk)) not a view, so
  undici recycling the pooled source buffer can't corrupt queued chunks.
- Tests for both.

* fix(mcp): decode Content-Encoding on the undici.request transport (fetch parity)

undici.request returns raw bytes; fetch auto-decompresses. Restore that: pipe a
gzip/deflate/br response body through the matching zlib decoder before the WHATWG
bridge and strip content-encoding/content-length. maxResponseSize still caps the
compressed wire bytes; errors forward into the decoder and the source is torn down
when the decoded stream ends or is cancelled. Adds a gzip decode test.

* fix(mcp): guard decoder errors and null-body drain against unhandled crashes

- Attach the stream bridge's error listener before piping into the zlib decoder, so
  a synchronous zlib error (server mislabeling a non-gzip body as gzip) rejects the
  reader instead of crashing the process.
- Attach an error listener before draining a null-body response, and wrap Response
  construction in try/catch that destroys the source (no socket leak) on an
  out-of-range status. Adds an invalid-gzip regression test.
This commit is contained in:
Waleed
2026-07-23 14:17:52 -07:00
committed by GitHub
parent e737901b0b
commit a5e1030e9e
2 changed files with 497 additions and 6 deletions
@@ -0,0 +1,232 @@
/**
* @vitest-environment node
*
* Covers the `undici.request()`-backed guarded fetch: `createSsrfGuardedFetchWithDispatcher`
* builds its `fetch` on `undici.request` (not `undici.fetch`) because undici's `fetch` never
* delivers a streaming `response.body` under the Bun runtime the server runs on. These tests
* drive the real builder with a mocked `undici.request` and assert the constructed `Response`
* preserves status/headers/url, streams its body, follows redirects via `followRedirectsGuarded`,
* serves buffered reads, and settles the reader when the source is destroyed without an error.
*/
import { Readable } from 'node:stream'
import { gzipSync } from 'node:zlib'
import { beforeEach, describe, expect, it, vi } from 'vitest'
const { mockAgent, mockUndiciRequest } = vi.hoisted(() => {
class MockAgent {
close() {
return Promise.resolve()
}
destroy() {
return Promise.resolve()
}
}
return { mockAgent: MockAgent, mockUndiciRequest: vi.fn() }
})
vi.mock('undici', () => ({ Agent: mockAgent, request: mockUndiciRequest, fetch: vi.fn() }))
declare module '@/lib/core/security/input-validation.server?guarded-request-test' {
// biome-ignore lint/suspicious/noExportsInTest: ambient re-declaration for the query-suffixed specifier
export * from '@/lib/core/security/input-validation.server'
}
import { createSsrfGuardedFetchWithDispatcher } from '@/lib/core/security/input-validation.server?guarded-request-test'
/** A byte stream that yields a single `Buffer` chunk then ends — mirrors undici's body. */
function byteStream(text: string): Readable {
const stream = new Readable({ read() {} })
stream.push(Buffer.from(text))
stream.push(null)
return stream
}
function undiciReply(
statusCode: number,
headers: Record<string, string | string[]>,
body: Readable
) {
return { statusCode, headers, body, trailers: {}, opaque: null, context: {} }
}
describe('createSsrfGuardedFetchWithDispatcher (undici.request backed)', () => {
beforeEach(() => {
vi.clearAllMocks()
})
it('constructs a Response with the reply status, headers, url, and a streaming body', async () => {
mockUndiciRequest.mockResolvedValueOnce(
undiciReply(
200,
{ 'content-type': 'text/event-stream', 'mcp-session-id': 's-1' },
byteStream('event: message\ndata: {"id":1}\n\n')
)
)
const { fetch } = createSsrfGuardedFetchWithDispatcher()
const response = await fetch('https://mcp.example.com/serve', {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: '{"jsonrpc":"2.0"}',
})
expect(response.status).toBe(200)
expect(response.headers.get('mcp-session-id')).toBe('s-1')
expect(response.url).toBe('https://mcp.example.com/serve')
const text = await response.text()
expect(text).toContain('"id":1')
// Does NOT auto-follow redirects — followRedirectsGuarded drives them instead.
expect(mockUndiciRequest).toHaveBeenCalledTimes(1)
const [, options] = mockUndiciRequest.mock.calls[0]
expect(options.method).toBe('POST')
expect(options.headers).toEqual({ 'content-type': 'application/json' })
expect(options.body).toBe('{"jsonrpc":"2.0"}')
expect(options.maxRedirections).toBeUndefined()
})
it('follows a redirect through followRedirectsGuarded and reports the final url', async () => {
mockUndiciRequest
.mockResolvedValueOnce(
undiciReply(302, { location: 'https://mcp.example.com/final' }, byteStream('redirect'))
)
.mockResolvedValueOnce(undiciReply(200, {}, byteStream('final-body')))
const { fetch } = createSsrfGuardedFetchWithDispatcher()
const response = await fetch('https://mcp.example.com/start', { method: 'GET' })
expect(mockUndiciRequest).toHaveBeenCalledTimes(2)
expect(response.status).toBe(200)
expect(response.url).toBe('https://mcp.example.com/final')
expect(await response.text()).toBe('final-body')
})
it('supports buffered reads (.json()) through the constructed body', async () => {
mockUndiciRequest.mockResolvedValueOnce(
undiciReply(
200,
{ 'content-type': 'application/json' },
byteStream(JSON.stringify({ ok: true }))
)
)
const { fetch } = createSsrfGuardedFetchWithDispatcher()
const response = await fetch('https://mcp.example.com/data', { method: 'GET' })
expect(await response.json()).toEqual({ ok: true })
})
it('normalizes a Headers instance and an ArrayBuffer body for undici.request', async () => {
mockUndiciRequest.mockResolvedValueOnce(undiciReply(200, {}, byteStream('x')))
const { fetch } = createSsrfGuardedFetchWithDispatcher()
await fetch('https://mcp.example.com/x', {
method: 'POST',
headers: new Headers({ authorization: 'Bearer t' }),
body: new TextEncoder().encode('payload').buffer,
})
const [, options] = mockUndiciRequest.mock.calls[0]
expect(options.headers).toEqual({ authorization: 'Bearer t' })
expect(Buffer.isBuffer(options.body)).toBe(true)
expect(Buffer.from(options.body).toString()).toBe('payload')
})
it('serializes a URLSearchParams body and defaults the form content-type (OAuth token exchange)', async () => {
mockUndiciRequest.mockResolvedValueOnce(undiciReply(200, {}, byteStream('{}')))
const { fetch } = createSsrfGuardedFetchWithDispatcher()
await fetch('https://auth.example.com/token', {
method: 'POST',
body: new URLSearchParams({ grant_type: 'refresh_token', refresh_token: 'r-1' }),
})
const [, options] = mockUndiciRequest.mock.calls[0]
expect(options.body).toBe('grant_type=refresh_token&refresh_token=r-1')
expect(options.headers['content-type']).toBe('application/x-www-form-urlencoded;charset=UTF-8')
})
it('does not override an explicit content-type on a URLSearchParams body', async () => {
mockUndiciRequest.mockResolvedValueOnce(undiciReply(200, {}, byteStream('{}')))
const { fetch } = createSsrfGuardedFetchWithDispatcher()
await fetch('https://auth.example.com/token', {
method: 'POST',
headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
body: new URLSearchParams({ grant_type: 'authorization_code' }),
})
const [, options] = mockUndiciRequest.mock.calls[0]
expect(options.body).toBe('grant_type=authorization_code')
expect(options.headers['Content-Type']).toBe('application/x-www-form-urlencoded')
expect(options.headers['content-type']).toBeUndefined()
})
it('copies each chunk so a recycled source buffer cannot corrupt queued data', async () => {
const source = new Readable({ read() {} })
mockUndiciRequest.mockResolvedValueOnce(undiciReply(200, {}, source))
const { fetch } = createSsrfGuardedFetchWithDispatcher()
const response = await fetch('https://mcp.example.com/stream', { method: 'GET' })
const reader = response.body!.getReader()
// Emit a chunk, then mutate the SAME backing buffer (as undici's pool reuse would).
const buf = Buffer.from('AB')
source.push(buf)
const first = await reader.read()
buf[0] = 0x00 // corrupt the source buffer after the chunk was enqueued
expect(Buffer.from(first.value!).toString()).toBe('AB') // copy is unaffected
source.push(null)
await reader.read()
})
it('decodes a gzip Content-Encoding body and strips the framing headers (fetch parity)', async () => {
const gzipped = gzipSync(Buffer.from(JSON.stringify({ ok: true, msg: 'compressed' })))
const source = new Readable({ read() {} })
source.push(gzipped)
source.push(null)
mockUndiciRequest.mockResolvedValueOnce(
undiciReply(
200,
{ 'content-type': 'application/json', 'content-encoding': 'gzip', 'content-length': '40' },
source
)
)
const { fetch } = createSsrfGuardedFetchWithDispatcher()
const response = await fetch('https://mcp.example.com/data', { method: 'GET' })
expect(await response.json()).toEqual({ ok: true, msg: 'compressed' })
expect(response.headers.get('content-encoding')).toBeNull()
expect(response.headers.get('content-length')).toBeNull()
})
it('rejects the reader (not crash) when Content-Encoding is gzip but the body is not', async () => {
const source = new Readable({ read() {} })
source.push(Buffer.from('this is definitely not gzip'))
source.push(null)
mockUndiciRequest.mockResolvedValueOnce(
undiciReply(200, { 'content-type': 'application/json', 'content-encoding': 'gzip' }, source)
)
const { fetch } = createSsrfGuardedFetchWithDispatcher()
const response = await fetch('https://mcp.example.com/bad', { method: 'GET' })
// The zlib error must surface as a rejected read, never an unhandled 'error' event.
await expect(response.text()).rejects.toThrow()
})
it('rejects the reader when the source is destroyed without an error (abort/reset)', async () => {
const source = new Readable({ read() {} }) // stays open, never pushes
mockUndiciRequest.mockResolvedValueOnce(undiciReply(200, {}, source))
const { fetch } = createSsrfGuardedFetchWithDispatcher()
const response = await fetch('https://mcp.example.com/hang', { method: 'GET' })
const reader = response.body!.getReader()
const read = reader.read()
source.destroy() // no error argument — mirrors an aborted/reset socket
await expect(read).rejects.toThrow(/closed before completing/)
})
})
@@ -1,3 +1,5 @@
import { Readable } from 'node:stream'
import zlib from 'node:zlib'
import dns from 'dns/promises'
import http from 'http'
import https from 'https'
@@ -8,7 +10,13 @@ import { omit } from '@sim/utils/object'
import { HttpProxyAgent } from 'http-proxy-agent'
import { HttpsProxyAgent } from 'https-proxy-agent'
import * as ipaddr from 'ipaddr.js'
import { Agent, type RequestInit as UndiciRequestInit, fetch as undiciFetch } from 'undici'
import {
Agent,
type Dispatcher,
type RequestInit as UndiciRequestInit,
fetch as undiciFetch,
request as undiciRequest,
} from 'undici'
import { isHosted, isPrivateDatabaseHostsAllowed } from '@/lib/core/config/env-flags'
import { type ValidationResult, validateExternalUrl } from '@/lib/core/security/input-validation'
import { PayloadSizeLimitError } from '@/lib/core/utils/stream-limits'
@@ -619,6 +627,259 @@ export async function followRedirectsGuarded(
}
}
/** Coerce a DOM/undici `HeadersInit` into the record shape undici `request` accepts. */
function toUndiciRequestHeaders(
headers: UndiciRequestInit['headers']
): Record<string, string> | undefined {
if (!headers) return undefined
const record: Record<string, string> = {}
if (Array.isArray(headers)) {
for (const [key, value] of headers as [string, string][]) {
if (value != null) record[key] = String(value)
}
return record
}
// Single cast (no `as unknown`): the optional `forEach` is satisfiable by both a plain
// record (absent) and a `Headers` instance (present), so it detects the iterable form.
const iterableHeaders = headers as {
forEach?: (cb: (value: string, key: string) => void) => void
}
if (typeof iterableHeaders.forEach === 'function') {
iterableHeaders.forEach((value, key) => {
record[key] = value
})
return record
}
for (const [key, value] of Object.entries(headers as Record<string, unknown>)) {
if (value != null) record[key] = Array.isArray(value) ? value.join(', ') : String(value)
}
return record
}
/** Coerce a DOM/undici body init into a value undici `request` accepts. */
function toUndiciRequestBody(
body: UndiciRequestInit['body']
): string | Buffer | Uint8Array | Readable | undefined {
if (body == null) return undefined
// fetch accepts URLSearchParams (form-encoded) and undici.request does not — the MCP SDK's
// OAuth token/refresh exchange sends one. Serialize it to its wire form.
if (body instanceof URLSearchParams) return body.toString()
if (body instanceof ArrayBuffer) return Buffer.from(body)
if (ArrayBuffer.isView(body) && !(body instanceof Uint8Array)) {
return Buffer.from(body.buffer, body.byteOffset, body.byteLength)
}
if (typeof (body as ReadableStream).getReader === 'function') {
// double-cast-allowed: DOM ReadableStream and the node:stream Web type differ but are structurally compatible at runtime
return Readable.fromWeb(body as unknown as Parameters<typeof Readable.fromWeb>[0])
}
// string, Uint8Array/Buffer, or Readable — passed through unchanged.
// double-cast-allowed: undici BodyInit is wider than what request() accepts; our guarded/pinned callers only send these
return body as unknown as string | Buffer | Uint8Array | Readable
}
/**
* Decompression transform for a `Content-Encoding`, or `null` to pass the body through.
* `undici.fetch` decodes the body automatically; `undici.request` does not, so this restores
* fetch parity for gzip/deflate/br responses (common behind CDNs). Unknown/absent encodings
* pass through untouched.
*/
function contentEncodingDecoder(
encoding: string
): zlib.Gunzip | zlib.Inflate | zlib.BrotliDecompress | null {
switch (encoding) {
case 'gzip':
case 'x-gzip':
return zlib.createGunzip()
case 'deflate':
return zlib.createInflate()
case 'br':
return zlib.createBrotliDecompress()
default:
return null
}
}
/**
* Bridges an undici `request()` Node `Readable` into a WHATWG `ReadableStream` for a
* `Response` body. Node's built-in `Readable.toWeb` is NOT used: its adapter throws an
* unhandled `ERR_INVALID_STATE` ("Controller is already closed") when the web stream is
* cancelled while the Node stream is still flowing — which `followRedirectsGuarded` does
* on every redirect hop (`response.body.cancel()`). This bridge instead swallows a late
* enqueue after close and destroys the source on cancel, so cancelling a live body frees
* its socket cleanly. `maxResponseSize` overruns surface as the source's `error` event and
* reject the read, preserving the DoS backstop.
*/
function nodeReadableToWebStream(nodeStream: Readable): ReadableStream<Uint8Array> {
let settled = false
return new ReadableStream<Uint8Array>({
start(controller) {
nodeStream.on('data', (chunk: Buffer) => {
try {
// Copy, not a view: undici may recycle the pooled buffer backing `chunk` after
// this handler returns, which would corrupt a chunk still queued for a slow
// consumer. `new Uint8Array(chunk)` allocates a fresh backing buffer.
controller.enqueue(new Uint8Array(chunk))
} catch {
// Controller already closed (consumer cancelled) — stop the source, drop the chunk.
nodeStream.destroy()
return
}
if ((controller.desiredSize ?? 1) <= 0) nodeStream.pause()
})
nodeStream.once('end', () => {
settled = true
try {
controller.close()
} catch {}
})
nodeStream.once('error', (err) => {
settled = true
try {
controller.error(err)
} catch {}
})
// An abort (signal) or upstream reset can `destroy()` the source with no `error`
// event; without this the reader would hang forever. `close` fires after every
// terminal path, so only act when `end`/`error` didn't already settle the stream.
nodeStream.once('close', () => {
if (settled) return
settled = true
try {
controller.error(new Error('MCP transport stream closed before completing'))
} catch {}
})
// Start paused so nothing buffers before the consumer pulls (backpressure).
nodeStream.pause()
},
pull() {
nodeStream.resume()
},
cancel(reason) {
nodeStream.destroy(reason instanceof Error ? reason : undefined)
},
})
}
/**
* Streaming-safe replacement for `undiciFetch(url, { ...init, dispatcher })`.
*
* undici's `fetch` exposes the response body as a WHATWG `ReadableStream` whose
* bridge is broken under the Bun runtime (which the standalone server runs on):
* response headers arrive but `response.body` never yields data, hanging every
* incremental read — MCP SSE `tools/list`, provider streaming — to its timeout.
* undici's lower-level `request()` returns a Node `Readable` instead, which Bun
* implements natively and streams correctly; `Readable.toWeb` bridges it back to
* a spec `Response`. Buffered reads (`.json()`/`.text()`/`.arrayBuffer()`) behave
* identically on both runtimes, so this is a drop-in substitute.
*
* SSRF is unchanged: the same `dispatcher` (Agent carrying the guarded/pinned
* `connect.lookup`) governs every connection, and `maxResponseSize` still caps the
* body. Redirects are NOT followed here (`maxRedirections: 0`); the caller drives
* them via {@link followRedirectsGuarded}, exactly as it did over `fetch`'s
* `redirect: 'manual'`.
*/
async function undiciRequestAsResponse(
input: RequestInfo | URL,
init: RequestInit,
dispatcher: Agent
): Promise<Response> {
let url: string
let effectiveInit = init as UndiciRequestInit
if (typeof Request !== 'undefined' && input instanceof Request) {
// A Request input carries its own method/headers/body/signal; lift them (explicit
// init fields win, per fetch semantics) so a guarded POST isn't downgraded to GET.
const bodyAllowed = input.method !== 'GET' && input.method !== 'HEAD'
effectiveInit = {
method: input.method,
headers: input.headers,
body: bodyAllowed ? await input.clone().arrayBuffer() : undefined,
signal: input.signal,
...(init as UndiciRequestInit),
// double-cast-allowed: DOM RequestInit and undici RequestInit differ in TS but match at runtime
} as unknown as UndiciRequestInit
url = input.url
} else {
url = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url
}
const method = (effectiveInit.method ?? 'GET').toUpperCase()
const canHaveBody = method !== 'GET' && method !== 'HEAD'
const requestHeaders = toUndiciRequestHeaders(effectiveInit.headers) ?? {}
const requestBody = canHaveBody ? toUndiciRequestBody(effectiveInit.body) : undefined
// fetch auto-adds a form content-type for a URLSearchParams body; preserve that parity
// when the caller didn't set one (the MCP SDK does set it explicitly, but not every caller).
if (
canHaveBody &&
effectiveInit.body instanceof URLSearchParams &&
!Object.keys(requestHeaders).some((key) => key.toLowerCase() === 'content-type')
) {
requestHeaders['content-type'] = 'application/x-www-form-urlencoded;charset=UTF-8'
}
const { statusCode, headers, body } = await undiciRequest(url, {
method: method as Dispatcher.HttpMethod,
headers: requestHeaders,
body: requestBody,
signal: effectiveInit.signal ?? undefined,
dispatcher,
// No `maxRedirections`: request() does not auto-follow by default, so the caller's
// `followRedirectsGuarded` drives every hop with per-hop SSRF validation.
})
const responseHeaders = new Headers()
for (const [key, value] of Object.entries(headers)) {
if (Array.isArray(value)) for (const v of value) responseHeaders.append(key, v)
else if (value != null) responseHeaders.append(key, value)
}
// Null-body statuses (204/205/304) can't carry a body; drain undici's (empty) stream so its
// socket returns to the pool. Attach an error listener first so a socket reset mid-drain
// surfaces as a handled event, not an unhandled 'error' that crashes the process.
const isNullBody = statusCode === 204 || statusCode === 205 || statusCode === 304
if (isNullBody) {
body.on('error', () => {})
body.resume()
const response = new Response(null, { status: statusCode, headers: responseHeaders })
Object.defineProperty(response, 'url', { value: url, configurable: true })
return response
}
// Decode Content-Encoding like `fetch` does (`request()` returns raw bytes). `maxResponseSize`
// still caps the compressed wire bytes on `body`.
const contentEncoding = String(headers['content-encoding'] ?? '')
.toLowerCase()
.trim()
const decoder = contentEncodingDecoder(contentEncoding)
if (decoder) {
// The bridged body is now decoded; drop framing headers that would misdescribe it.
responseHeaders.delete('content-encoding')
responseHeaders.delete('content-length')
}
// Build the bridge over the stream the consumer reads (the decoder when decoding).
// `nodeReadableToWebStream` attaches its `error` listener synchronously, so wiring the pipe
// AFTER it means a synchronous zlib error (e.g. a server mislabeling a non-gzip body as gzip)
// is caught and rejects the reader instead of taking down the process.
const webBody = nodeReadableToWebStream(decoder ?? body)
if (decoder) {
body.once('error', (err) => decoder.destroy(err)) // forward maxResponseSize / socket reset
decoder.once('close', () => body.destroy()) // tear the source down so the socket can't leak
body.pipe(decoder)
}
try {
const response = new Response(webBody, { status: statusCode, headers: responseHeaders })
// undici.request never sets `url`; `fetch` did, and consumers rely on it (the MCP
// transport's response-cap wrapper copies it; the SDK resolves relative
// auth-metadata URLs against it). Preserve parity.
Object.defineProperty(response, 'url', { value: url, configurable: true })
return response
} catch (err) {
// `new Response` rejects an out-of-range status (a 1xx undici shouldn't surface, but
// defensively): destroy the source so its socket can't leak, then rethrow.
body.destroy()
throw err
}
}
/**
* SSRF-guarded `fetch` + its `Agent` for outbound requests to user-controlled
* hosts: DNS resolves normally, and every socket connect validates the chosen
@@ -637,11 +898,9 @@ export function createSsrfGuardedFetchWithDispatcher(options?: { maxResponseSize
...(options?.maxResponseSize !== undefined ? { maxResponseSize: options.maxResponseSize } : {}),
})
const rawFetch = async (url: string, init: UndiciRequestInit): Promise<Response> => {
const response = await undiciFetch(url, { ...init, dispatcher })
// double-cast-allowed: undici Response and DOM Response are structurally compatible at runtime
return response as unknown as Response
}
const rawFetch = (url: string, init: UndiciRequestInit): Promise<Response> =>
// double-cast-allowed: DOM RequestInit and undici RequestInit differ in TS but match at runtime
undiciRequestAsResponse(url, init as unknown as RequestInit, dispatcher)
const guarded = async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
const target = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url