diff --git a/clients/web/src/test/core/mcp/fetchTracking.test.ts b/clients/web/src/test/core/mcp/fetchTracking.test.ts index 76abcb0f..4640b694 100644 --- a/clients/web/src/test/core/mcp/fetchTracking.test.ts +++ b/clients/web/src/test/core/mcp/fetchTracking.test.ts @@ -1,5 +1,8 @@ import { describe, it, expect, vi } from "vitest"; import { + CHALLENGE_BODY_MAX_BYTES, + CHALLENGE_BODY_READ_MS, + TRUNCATED_BODY_SUFFIX, createFetchTracker, redactSensitiveHeaders, redactBody, @@ -539,6 +542,123 @@ describe("createFetchTracker", () => { }); }); +describe("createFetchTracker — 401/403 challenge bodies (#2297)", () => { + function track(response: Response) { + const tracked: FetchRequestEntryBase[] = []; + const updateResponseBody = vi.fn(); + const fetcher = createFetchTracker((async () => response) as typeof fetch, { + trackRequest: (entry) => tracked.push(entry), + updateResponseBody, + }); + return { fetcher, tracked, updateResponseBody }; + } + + it("records the body on the entry itself, before returning", async () => { + const body = JSON.stringify({ error: "invalid_token" }); + const { fetcher, tracked, updateResponseBody } = track( + new Response(body, { + status: 401, + headers: { "content-type": "application/json" }, + }), + ); + + const res = await fetcher("https://mcp.example/mcp", { method: "POST" }); + + expect(tracked).toHaveLength(1); + expect(tracked[0]?.responseStatus).toBe(401); + expect(tracked[0]?.responseBody).toBe(body); + expect(updateResponseBody).not.toHaveBeenCalled(); + // The caller still gets an unread body. + expect(await res.text()).toBe(body); + }); + + it("does the same for a 403", async () => { + const { fetcher, tracked } = track( + new Response("insufficient_scope", { status: 403 }), + ); + await fetcher("https://mcp.example/mcp", { method: "POST" }); + expect(tracked[0]?.responseBody).toBe("insufficient_scope"); + }); + + it("stops at the byte cap and marks the body truncated", async () => { + const chunk = new Uint8Array(CHALLENGE_BODY_MAX_BYTES / 2).fill(97); + let pulls = 0; + const cancel = vi.fn(); + const stream = new ReadableStream({ + pull(controller) { + pulls += 1; + controller.enqueue(chunk); + }, + cancel, + }); + const { fetcher, tracked } = track(new Response(stream, { status: 401 })); + + const res = await fetcher("https://mcp.example/mcp", { method: "POST" }); + + const recorded = tracked[0]?.responseBody ?? ""; + expect(recorded.endsWith(TRUNCATED_BODY_SUFFIX)).toBe(true); + expect(recorded.length - TRUNCATED_BODY_SUFFIX.length).toBe( + CHALLENGE_BODY_MAX_BYTES, + ); + // With the clone released, cancelling the caller's branch reaches the source. + await res.body?.cancel(); + expect(cancel).toHaveBeenCalled(); + expect(pulls).toBeLessThan(10); + }); + + it("gives up on a stalled body at the deadline and releases the source", async () => { + vi.useFakeTimers(); + try { + const cancel = vi.fn(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("partial")); + }, + cancel, + }); + const { fetcher, tracked } = track(new Response(stream, { status: 401 })); + + const pending = fetcher("https://mcp.example/mcp", { method: "POST" }); + await vi.advanceTimersByTimeAsync(CHALLENGE_BODY_READ_MS); + const res = await pending; + + expect(tracked[0]?.responseBody).toBe(`partial${TRUNCATED_BODY_SUFFIX}`); + await res.body?.cancel(); + expect(cancel).toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it("records the entry with no body when the clone cannot be made", async () => { + const response = new Response("nope", { status: 401 }); + vi.spyOn(response, "clone").mockImplementation(() => { + throw new Error("clone failed"); + }); + const { fetcher, tracked } = track(response); + + await fetcher("https://mcp.example/mcp", { method: "POST" }); + + expect(tracked).toHaveLength(1); + expect(tracked[0]?.responseStatus).toBe(401); + expect(tracked[0]?.responseBody).toBeUndefined(); + }); + + it("records the entry with no body when the read errors", async () => { + const stream = new ReadableStream({ + pull(controller) { + controller.error(new Error("socket reset")); + }, + }); + const { fetcher, tracked } = track(new Response(stream, { status: 401 })); + + await fetcher("https://mcp.example/mcp", { method: "POST" }); + + expect(tracked[0]?.responseStatus).toBe(401); + expect(tracked[0]?.responseBody).toBeUndefined(); + }); +}); + describe("redactUrlQuery", () => { it("redacts sensitive params and keeps the path + other params", () => { expect( diff --git a/clients/web/src/test/integration/mcp/node/transport.test.ts b/clients/web/src/test/integration/mcp/node/transport.test.ts index 95481a30..eba85969 100644 --- a/clients/web/src/test/integration/mcp/node/transport.test.ts +++ b/clients/web/src/test/integration/mcp/node/transport.test.ts @@ -260,14 +260,14 @@ describe("Transport", () => { }); const fetchRequests: FetchRequestEntryBase[] = []; - const bodies = new Map(); + const onFetchResponseBody = vi.fn(); const result = createTransportNode( { type: "streamable-http", url: "https://mcp.example/mcp" }, { fetchFn, interceptAuthChallenges: true, onFetchRequest: (entry) => fetchRequests.push(entry), - onFetchResponseBody: (id, body) => bodies.set(id, body), + onFetchResponseBody, }, ); @@ -279,12 +279,15 @@ describe("Transport", () => { "MCP auth challenge (401)", ); + // The body is on the entry by the time the challenge is thrown — not a + // later update — so a caller that stops listening at the throw has it. expect(fetchRequests).toHaveLength(1); const [entry] = fetchRequests; expect(entry.error).toBeUndefined(); expect(entry.responseStatus).toBe(401); expect(entry.responseHeaders?.["www-authenticate"]).toBe(wwwAuthenticate); - await vi.waitFor(() => expect(bodies.get(entry.id)).toBe(challengeBody)); + expect(entry.responseBody).toBe(challengeBody); + expect(onFetchResponseBody).not.toHaveBeenCalled(); }); it("still throws an intercepted 401 whose body never ends (#2297)", async () => { @@ -319,6 +322,8 @@ describe("Transport", () => { "MCP auth challenge (401)", ); expect(fetchRequests[0]?.responseStatus).toBe(401); + // Cut off at the tracker's read deadline rather than buffered forever. + expect(fetchRequests[0]?.responseBody).toBe("partial\n[truncated]"); }); it("applies settings.headers to the outgoing streamable-http request", async () => { diff --git a/clients/web/src/test/integration/mcp/remote/remote-auth-branches.test.ts b/clients/web/src/test/integration/mcp/remote/remote-auth-branches.test.ts index 355f5bb8..ef1a7967 100644 --- a/clients/web/src/test/integration/mcp/remote/remote-auth-branches.test.ts +++ b/clients/web/src/test/integration/mcp/remote/remote-auth-branches.test.ts @@ -193,6 +193,69 @@ describe("server.ts /api/mcp/* branch coverage", () => { await new Promise((r) => upstream.server.close(() => r())); } }); + + it("relays a streamable-http 401's status and body over the event stream (#2297)", async () => { + const upstream = await startUnauthorizedUpstream(); + const events = new AbortController(); + try { + const connected = (await ( + await connect(h, { type: "streamable-http", url: upstream.url }) + ).json()) as { sessionId?: string }; + expect(connected.sessionId).toBeDefined(); + + const stream = await fetch( + `${h.baseUrl}/api/mcp/events?sessionId=${connected.sessionId}`, + { signal: events.signal }, + ); + const sent = await fetch(`${h.baseUrl}/api/mcp/send`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + sessionId: connected.sessionId, + message: { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "test", version: "1.0.0" }, + }, + }, + }), + }); + expect(await sent.json()).toMatchObject({ + ok: false, + kind: "auth_challenge", + }); + + const reader = stream.body!.getReader(); + const decoder = new TextDecoder(); + const pattern = /data: (\{"type":"fetch_request",.*?\})\n/; + let text = ""; + let match: RegExpMatchArray | null = null; + while (!(match = text.match(pattern))) { + const { done, value } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + } + const event = JSON.parse(match![1]!) as { + data: { + responseStatus?: number; + responseBody?: string; + error?: string; + }; + }; + expect(event.data.error).toBeUndefined(); + expect(event.data.responseStatus).toBe(401); + expect(event.data.responseBody).toBe( + JSON.stringify({ error: "unauthorized" }), + ); + } finally { + events.abort(); + await new Promise((r) => upstream.server.close(() => r())); + } + }); }); describe("POST /api/mcp/send", () => { diff --git a/core/mcp/fetchTracking.ts b/core/mcp/fetchTracking.ts index 7e1e5864..f756b1d4 100644 --- a/core/mcp/fetchTracking.ts +++ b/core/mcp/fetchTracking.ts @@ -223,6 +223,65 @@ export function isLongLivedStreamResponse( ); } +/** Most of a 401/403 body the tracker will buffer before it stops reading. */ +export const CHALLENGE_BODY_MAX_BYTES = 64 * 1024; +/** How long the tracker waits for a 401/403 body before giving up on the rest. */ +export const CHALLENGE_BODY_READ_MS = 2_000; +/** Appended to a 401/403 body the tracker stopped reading early. */ +export const TRUNCATED_BODY_SUFFIX = "\n[truncated]"; + +/** + * Read a 401/403 response body from a clone, bounded by + * {@link CHALLENGE_BODY_MAX_BYTES} and {@link CHALLENGE_BODY_READ_MS}. + * + * Unlike every other status, a challenge body is read *before* the entry is + * recorded (#2297). An auth-challenge interceptor above the tracker cancels the + * response and throws as soon as the tracker returns, and on the web backend + * that ends the connect request — so a body delivered later, as a separate + * update, arrives after anything could still carry it to the log. The bound is + * what makes waiting safe: a body that never ends is cut off, and cancelling the + * clone's reader lets the interceptor's cancel of the original release the + * connection instead of leaving the clone draining it. + */ +async function readChallengeBody( + response: Response, +): Promise { + let reader: ReadableStreamDefaultReader; + try { + const body = response.clone().body; + /* v8 ignore next -- the caller checked response.body, and a clone of a response with a body has one */ + if (!body) return undefined; + reader = body.getReader(); + } catch { + return undefined; + } + + const decoder = new TextDecoder(); + let text = ""; + let bytes = 0; + let timer: ReturnType | undefined; + const deadline = new Promise<"timeout">((resolve) => { + timer = setTimeout(() => resolve("timeout"), CHALLENGE_BODY_READ_MS); + }); + try { + while (bytes < CHALLENGE_BODY_MAX_BYTES) { + const next = await Promise.race([reader.read(), deadline]); + if (next === "timeout") break; + if (next.done) return text + decoder.decode(); + bytes += next.value.byteLength; + text += decoder.decode(next.value, { stream: true }); + } + // Stopped early. Not awaited: cancel() adopts the source's promise, which a + // stalled body may never settle. + reader.cancel().catch(() => {}); + return text + decoder.decode() + TRUNCATED_BODY_SUFFIX; + } catch { + return undefined; + } finally { + clearTimeout(timer); + } +} + export interface FetchTrackingCallbacks { trackRequest?: (entry: FetchRequestEntryBase) => void; /** @@ -376,6 +435,21 @@ export function createFetchTracker( duration, }; + if ( + (responseStatus === 401 || responseStatus === 403) && + !isLongLivedStream && + response.body && + !response.bodyUsed + ) { + const body = await readChallengeBody(response); + entry.responseBody = + body === undefined + ? undefined + : (redactBody(body, response.headers.get("content-type")) ?? body); + callbacks.trackRequest?.(entry); + return response; + } + callbacks.trackRequest?.(entry); // Kick off a fire-and-forget read of the cloned body. The clone is an diff --git a/core/mcp/node/authChallengeFetch.ts b/core/mcp/node/authChallengeFetch.ts index 6bfbf31f..1ed8e34a 100644 --- a/core/mcp/node/authChallengeFetch.ts +++ b/core/mcp/node/authChallengeFetch.ts @@ -26,8 +26,8 @@ export function createAuthChallengeInterceptFetch( // Release the connection before throwing so the SDK transport is not left // with a half-read 401/403 body on streamable HTTP. Not awaited: when the // network tracker sits below this wrapper (#2297) the body is one branch of - // a tee, and cancelling a branch settles only once the tracker's clone has - // read to the end — so a 401 whose body never ends would hang the throw. + // a tee, and cancel() adopts the source's promise, which a stalled body may + // never settle — so awaiting it could hang the throw. void response.body?.cancel().catch(() => {}); throw new AuthChallengeError( diff --git a/core/mcp/node/transport.ts b/core/mcp/node/transport.ts index 8ed872f4..45fc2c89 100644 --- a/core/mcp/node/transport.ts +++ b/core/mcp/node/transport.ts @@ -83,7 +83,8 @@ export function createTransportNode( // throws, so the entry was recorded from the thrown error alone — no status, // no `WWW-Authenticate`, and no response body, which is exactly the detail // someone debugging an OAuth challenge needs. Below it, the tracker records - // the real response and clones its body before the interceptor cancels it. + // the real response and reads its body (bounded) before the interceptor + // cancels it. const trackedBase = withTracking(withChallengeObserver(baseFetch)); const fetchWithOptionalAuthIntercept = interceptAuthChallenges ? createAuthChallengeInterceptFetch(trackedBase)