860b8d27ad
Signed-off-by: cliffhall <cliff@futurescale.com> # Conflicts: # core/mcp/node/transport.ts
741 lines
28 KiB
TypeScript
741 lines
28 KiB
TypeScript
import type { FetchRequestEntryBase, FetchStreamState } from "./types.js";
|
|
|
|
/**
|
|
* Header names whose values are replaced with `REDACTED_HEADER_VALUE` before a
|
|
* fetch entry is recorded. The recorded entry flows to the in-memory log, the
|
|
* pino logger, and (via session storage) to disk — none of those sinks should
|
|
* ever see a live bearer token or session cookie.
|
|
*/
|
|
const SENSITIVE_HEADERS: ReadonlySet<string> = new Set([
|
|
"authorization",
|
|
"cookie",
|
|
"set-cookie",
|
|
"proxy-authorization",
|
|
"x-api-key",
|
|
// The inspector backend's own bearer (createRemoteFetch stamps this on every
|
|
// proxied request); same exposure as Authorization.
|
|
"x-mcp-remote-auth",
|
|
]);
|
|
|
|
/** Placeholder substituted for sensitive header values in recorded entries. */
|
|
export const REDACTED_HEADER_VALUE = "[REDACTED]";
|
|
|
|
/**
|
|
* Field / query-parameter names whose values are masked in a recorded fetch
|
|
* entry's request body, response body, and URL query string. These are the
|
|
* credentials that ride in OAuth token exchanges (and similar flows): the
|
|
* header slice masks `Authorization`, but the same secrets show up verbatim in
|
|
* the form/JSON body (`client_secret`, `code`, `refresh_token`, …) and are
|
|
* sometimes carried as URL query params. Matching is case-insensitive.
|
|
*/
|
|
const SENSITIVE_BODY_FIELDS: ReadonlySet<string> = new Set([
|
|
"client_secret",
|
|
"code",
|
|
"refresh_token",
|
|
"access_token",
|
|
"id_token",
|
|
"code_verifier",
|
|
"client_assertion",
|
|
"assertion",
|
|
"password",
|
|
"token",
|
|
]);
|
|
|
|
/**
|
|
* Placeholder substituted for sensitive body / URL values in recorded entries.
|
|
* Deliberately kept separate from {@link REDACTED_HEADER_VALUE} (even though both
|
|
* are `"[REDACTED]"` today) so the header and body/URL redaction paths can evolve
|
|
* their sentinels independently.
|
|
*/
|
|
export const REDACTED_VALUE = "[REDACTED]";
|
|
|
|
/** Whether `name` (any casing) is a known-sensitive field / query-param name. */
|
|
function isSensitiveField(name: string): boolean {
|
|
return SENSITIVE_BODY_FIELDS.has(name.toLowerCase());
|
|
}
|
|
|
|
/**
|
|
* Returns a copy of `headers` with every {@link SENSITIVE_HEADERS} value
|
|
* replaced by {@link REDACTED_HEADER_VALUE}. Comparison is case-insensitive
|
|
* (HTTP header names are case-insensitive); the original casing of every key is
|
|
* preserved so the recorded entry still shows what the client actually sent.
|
|
*/
|
|
export function redactSensitiveHeaders(
|
|
headers: Record<string, string>,
|
|
): Record<string, string> {
|
|
const out: Record<string, string> = {};
|
|
for (const [key, value] of Object.entries(headers)) {
|
|
out[key] = SENSITIVE_HEADERS.has(key.toLowerCase())
|
|
? REDACTED_HEADER_VALUE
|
|
: value;
|
|
}
|
|
return out;
|
|
}
|
|
|
|
/**
|
|
* Returns `url` with every {@link SENSITIVE_BODY_FIELDS} query-parameter value
|
|
* replaced by {@link REDACTED_VALUE}. The path and non-sensitive params stay
|
|
* readable. Best-effort: if the URL (or its query string) can't be parsed the
|
|
* original string is returned unchanged. Only the recorded copy is redacted —
|
|
* the live request still uses the original `input`/`init`.
|
|
*/
|
|
export function redactUrlQuery(url: string): string {
|
|
const queryStart = url.indexOf("?");
|
|
if (queryStart === -1) return url;
|
|
|
|
const base = url.slice(0, queryStart);
|
|
const afterQuery = url.slice(queryStart + 1);
|
|
// Preserve a trailing fragment (#…) untouched — it never carries query params.
|
|
const hashStart = afterQuery.indexOf("#");
|
|
const query = hashStart === -1 ? afterQuery : afterQuery.slice(0, hashStart);
|
|
const fragment = hashStart === -1 ? "" : afterQuery.slice(hashStart);
|
|
|
|
try {
|
|
const params = new URLSearchParams(query);
|
|
let changed = false;
|
|
for (const key of new Set(params.keys())) {
|
|
if (isSensitiveField(key)) {
|
|
changed = true;
|
|
// Collapse repeated occurrences to a single redacted value.
|
|
params.set(key, REDACTED_VALUE);
|
|
}
|
|
}
|
|
if (!changed) return url;
|
|
return `${base}?${params.toString()}${fragment}`;
|
|
} catch {
|
|
return url;
|
|
}
|
|
}
|
|
|
|
/** Recursively redact sensitive keys in a parsed JSON value (in place). */
|
|
function redactJsonValue(value: unknown): unknown {
|
|
if (Array.isArray(value)) {
|
|
return value.map(redactJsonValue);
|
|
}
|
|
if (value !== null && typeof value === "object") {
|
|
const out: Record<string, unknown> = {};
|
|
for (const [key, val] of Object.entries(value)) {
|
|
// Only STRING values of a sensitive-named field are masked. The secrets
|
|
// this targets (OAuth `code`, `access_token`, `client_secret`, …) are
|
|
// always strings; a non-string is never one of them. This is important
|
|
// for JSON-RPC bodies, whose numeric `error.code` (e.g. -32020) collides
|
|
// with the OAuth authorization-`code` name — masking it would destroy the
|
|
// very field the Network tab classifies the modern spec errors on.
|
|
//
|
|
// Assumes a sensitive value is a scalar or an object (recursed, so an inner
|
|
// sensitive string key is still masked). A sensitive key whose value is an
|
|
// *array of scalars* (e.g. `{ password: ["a", "b"] }`) would recurse
|
|
// element-by-element with no key context and slip through — no OAuth/token
|
|
// payload has that shape, so it is not handled.
|
|
out[key] =
|
|
isSensitiveField(key) && typeof val === "string"
|
|
? REDACTED_VALUE
|
|
: redactJsonValue(val);
|
|
}
|
|
return out;
|
|
}
|
|
return value;
|
|
}
|
|
|
|
/**
|
|
* Returns `body` with every {@link SENSITIVE_BODY_FIELDS} value masked, for
|
|
* `application/x-www-form-urlencoded` and JSON payloads. The surrounding shape
|
|
* (field order, non-sensitive fields, JSON structure) is preserved — only the
|
|
* values change. Best-effort and never throws: an empty, non-string, or
|
|
* unparseable body is returned unchanged. Only the recorded copy is redacted;
|
|
* the live request body is never touched.
|
|
*
|
|
* Scope is deliberately limited to `application/x-www-form-urlencoded` and JSON:
|
|
* these cover the OAuth token flows this redaction targets. `multipart/form-data`
|
|
* (and other binary/opaque bodies) are passed through verbatim — OAuth never uses
|
|
* multipart, so the risk is low; revisit if a multipart secret path appears.
|
|
*/
|
|
export function redactBody(
|
|
body: string,
|
|
contentType: string | null | undefined,
|
|
): string;
|
|
export function redactBody(
|
|
body: string | undefined,
|
|
contentType: string | null | undefined,
|
|
): string | undefined;
|
|
export function redactBody(
|
|
body: string | undefined,
|
|
contentType: string | null | undefined,
|
|
): string | undefined {
|
|
if (!body) return body;
|
|
|
|
const type = (contentType ?? "").toLowerCase();
|
|
|
|
// Form-encoded bodies (the OAuth token endpoint's request format).
|
|
if (type.includes("application/x-www-form-urlencoded")) {
|
|
try {
|
|
const params = new URLSearchParams(body);
|
|
let changed = false;
|
|
for (const key of new Set(params.keys())) {
|
|
if (isSensitiveField(key)) {
|
|
changed = true;
|
|
params.set(key, REDACTED_VALUE);
|
|
}
|
|
}
|
|
return changed ? params.toString() : body;
|
|
} catch {
|
|
return body;
|
|
}
|
|
}
|
|
|
|
// JSON bodies — either explicitly typed, or (when the content-type is
|
|
// missing/other) any string that parses as a JSON object/array. A bare
|
|
// JSON scalar has no field names, so it can't carry a sensitive key.
|
|
try {
|
|
const parsed: unknown = JSON.parse(body);
|
|
if (parsed !== null && typeof parsed === "object") {
|
|
return JSON.stringify(redactJsonValue(parsed));
|
|
}
|
|
} catch {
|
|
// Not JSON — fall through and leave as-is.
|
|
}
|
|
|
|
return body;
|
|
}
|
|
|
|
/**
|
|
* Case-insensitive lookup of a header value from a plain header record.
|
|
* Exported because a recorded entry's header casing is host-dependent (Node
|
|
* lowercases, happy-dom preserves the wire casing), so any consumer reading
|
|
* one back — the Network UI, the connection diagnostics — needs the same
|
|
* tolerance the tracker itself has.
|
|
*/
|
|
export function findHeader(
|
|
headers: Record<string, string> | undefined,
|
|
name: string,
|
|
): string | undefined {
|
|
if (!headers) return undefined;
|
|
const target = name.toLowerCase();
|
|
for (const [key, value] of Object.entries(headers)) {
|
|
if (key.toLowerCase() === target) return value;
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
/**
|
|
* Whether a response represents an unbounded (long-lived) HTTP stream
|
|
* whose body cannot be cloned + read to completion. The streamable HTTP
|
|
* spec uses `GET` + `text/event-stream` for the long-lived server-push
|
|
* channel; `POST` SSE replies are bounded (server closes after the
|
|
* JSON-RPC response) and therefore safe to capture. Shared between the
|
|
* fetch tracker (where it decides whether to read the body) and the
|
|
* Network UI (where it decides which placeholder to show).
|
|
*/
|
|
export function isLongLivedStreamResponse(
|
|
method: string,
|
|
contentType: string | null | undefined,
|
|
): boolean {
|
|
if (method.toUpperCase() !== "GET") return false;
|
|
if (!contentType) return false;
|
|
// Media types are case-insensitive (RFC 9110 §8.3.1); a server may send
|
|
// `Text/Event-Stream` and still be sending an event stream.
|
|
const type = contentType.toLowerCase();
|
|
return (
|
|
type.includes("text/event-stream") || type.includes("application/x-ndjson")
|
|
);
|
|
}
|
|
|
|
/** 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]";
|
|
/**
|
|
* Recorded in place of a 401/403 body that was cut off and looks like JSON.
|
|
* {@link redactBody} can only mask fields in JSON that parses, so a cut-off
|
|
* prefix such as `{"access_token":"…",` would otherwise reach the log verbatim.
|
|
*/
|
|
export const WITHHELD_TRUNCATED_JSON_BODY =
|
|
"[truncated JSON body withheld: an incomplete body cannot be redacted]";
|
|
|
|
interface ChallengeBodyRead {
|
|
text: string;
|
|
truncated: boolean;
|
|
}
|
|
|
|
/**
|
|
* The recorded form of a 401/403 body: redacted, marked when cut off, and never
|
|
* a cut-off JSON prefix. Form-encoded bodies still redact when cut, since each
|
|
* surviving `key=value` pair is parsed on its own; plain text has nothing to
|
|
* redact either way.
|
|
*/
|
|
function recordableChallengeBody(
|
|
read: ChallengeBodyRead,
|
|
contentType: string | null,
|
|
): string {
|
|
if (!read.truncated) return redactBody(read.text, contentType);
|
|
const looksLikeJson =
|
|
(contentType ?? "").toLowerCase().includes("json") ||
|
|
/^\s*[{[]/.test(read.text);
|
|
if (looksLikeJson) return WITHHELD_TRUNCATED_JSON_BODY;
|
|
return redactBody(read.text, contentType) + TRUNCATED_BODY_SUFFIX;
|
|
}
|
|
|
|
/**
|
|
* 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<ChallengeBodyRead | undefined> {
|
|
let reader: ReadableStreamDefaultReader<Uint8Array>;
|
|
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<typeof setTimeout> | 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: text + decoder.decode(), truncated: false };
|
|
// A stream may hand back one chunk far larger than the budget, so the cap
|
|
// is applied to the bytes, not to the chunk count.
|
|
const chunk = next.value.subarray(0, CHALLENGE_BODY_MAX_BYTES - bytes);
|
|
bytes += chunk.byteLength;
|
|
text += decoder.decode(chunk, { stream: true });
|
|
}
|
|
// Stopped early. Not awaited: cancel() adopts the source's promise, which a
|
|
// stalled body may never settle.
|
|
reader.cancel().catch(() => {});
|
|
return { text: text + decoder.decode(), truncated: true };
|
|
} catch {
|
|
return undefined;
|
|
} finally {
|
|
clearTimeout(timer);
|
|
}
|
|
}
|
|
|
|
export interface FetchTrackingCallbacks {
|
|
trackRequest?: (entry: FetchRequestEntryBase) => void;
|
|
/**
|
|
* Called after the response body has been read asynchronously. Lets the
|
|
* consumer patch the already-dispatched entry with the body without
|
|
* blocking the transport on body reading. Fires only on success — if the
|
|
* body couldn't be read (long-lived stream, clone failure), this is
|
|
* never invoked and the entry's responseBody stays undefined.
|
|
*/
|
|
updateResponseBody?: (id: string, responseBody: string) => void;
|
|
/**
|
|
* Called for a long-lived stream response — the one case `updateResponseBody`
|
|
* never covers — each time an SSE event is delivered on it, and once more
|
|
* when it ends (with `closedAt` set). Lets the consumer show that the stream
|
|
* is open and what it has carried, which is the whole diagnostic when a
|
|
* server serializes requests behind it (#2318).
|
|
*/
|
|
updateStream?: (id: string, stream: FetchStreamState) => void;
|
|
/**
|
|
* How often, at most, `updateStream` reports a moving event count, in ms.
|
|
* The first report (open, zero events) and the close are immediate; counts
|
|
* in between are coalesced so a server pushing hundreds of notifications a
|
|
* second costs one update per window rather than one per event — each
|
|
* update re-publishes the whole fetch log and, on the web client, crosses
|
|
* the backend-to-browser event stream. Defaults to
|
|
* `DEFAULT_STREAM_UPDATE_INTERVAL_MS`; `0` reports every event, for tests.
|
|
*/
|
|
streamUpdateIntervalMs?: number;
|
|
}
|
|
|
|
/** See `FetchTrackingCallbacks.streamUpdateIntervalMs`. */
|
|
export const DEFAULT_STREAM_UPDATE_INTERVAL_MS = 250;
|
|
|
|
/**
|
|
* The framing a long-lived stream's events arrive in — the two content types
|
|
* `isLongLivedStreamResponse` accepts. SSE dispatches on a blank line;
|
|
* NDJSON dispatches on every line.
|
|
*/
|
|
export type LongLivedStreamFraming = "sse" | "ndjson";
|
|
|
|
/** The framing for a long-lived stream's content type. */
|
|
export function longLivedStreamFraming(
|
|
contentType: string | null | undefined,
|
|
): LongLivedStreamFraming {
|
|
return contentType?.toLowerCase().includes("application/x-ndjson")
|
|
? "ndjson"
|
|
: "sse";
|
|
}
|
|
|
|
/**
|
|
* Count the events on a long-lived stream without buffering it, and report
|
|
* when it ends.
|
|
*
|
|
* Reads a `clone()` of the response through a reader and discards the bytes,
|
|
* so the transport keeps consuming the original at its own pace (the same
|
|
* tee the bounded-body path relies on). Counting follows the framing the
|
|
* transport's own parser applies. For SSE a blank line dispatches the block
|
|
* before it, and only a block carrying a `data` field is an event —
|
|
* keepalive comments (`: ping`) and bare `event:`/`id:` lines are not; a
|
|
* partial block at end-of-stream is dropped, exactly as the SSE spec and the
|
|
* SDK's `EventSourceParserStream` (which has no flush) drop it. For NDJSON
|
|
* every non-blank line is one event.
|
|
*
|
|
* Reports once as soon as watching starts (zero events, open), once per event,
|
|
* and once more on every exit from the read loop — a clean end-of-stream, a
|
|
* network error, or the transport aborting the fetch on disconnect (which
|
|
* errors both tee branches) — with `closedAt` set. The reader is never cancelled from here:
|
|
* cancelling one branch of a tee does not release the source, and
|
|
* `ReadableStream.cancel()` adopts the source's promise, which on a teardown
|
|
* path may never settle (the same hazard `core/mcp/remote/node/server.ts`
|
|
* documents at its own `cancel()` call).
|
|
*/
|
|
function watchLongLivedStream(
|
|
response: Response,
|
|
id: string,
|
|
framing: LongLivedStreamFraming,
|
|
callbacks: FetchTrackingCallbacks,
|
|
): void {
|
|
const listener = callbacks.updateStream;
|
|
if (!listener) return;
|
|
// Every report goes through here: the listener is consumer code (a relay
|
|
// sink that may already be closed, a state store), and it runs from three
|
|
// places none of which may throw outward — synchronously inside the fetch
|
|
// wrapper (the open), from a timer (a coalesced count), and off the
|
|
// discarded watcher promise (the close). A throw is the listener's
|
|
// problem; the stream's bookkeeping carries on.
|
|
const report = (state: FetchStreamState): void => {
|
|
try {
|
|
listener(id, state);
|
|
} catch {
|
|
// See above.
|
|
}
|
|
};
|
|
const intervalMs =
|
|
callbacks.streamUpdateIntervalMs ?? DEFAULT_STREAM_UPDATE_INTERVAL_MS;
|
|
let body: ReadableStream<Uint8Array> | null;
|
|
try {
|
|
body = response.clone().body;
|
|
} catch {
|
|
// Clone failed (consumed body, transport quirks) — there is nothing to
|
|
// watch; the entry stays as tracked with no stream state.
|
|
return;
|
|
}
|
|
if (!body) return;
|
|
const reader = body.getReader();
|
|
const decoder = new TextDecoder();
|
|
let eventCount = 0;
|
|
let pending = "";
|
|
let blockHasData = false;
|
|
// A moving count is reported at most once per interval (see
|
|
// `streamUpdateIntervalMs`); the open and the close go out at once.
|
|
let reportTimer: ReturnType<typeof setTimeout> | null = null;
|
|
const reportCount = (): void => {
|
|
if (intervalMs <= 0) {
|
|
report({ eventCount });
|
|
return;
|
|
}
|
|
if (reportTimer !== null) return;
|
|
reportTimer = setTimeout(() => {
|
|
reportTimer = null;
|
|
report({ eventCount });
|
|
}, intervalMs);
|
|
};
|
|
// Announce the stream as open before a byte arrives. A stream that never
|
|
// delivers anything is the #2187 tell, and it would otherwise be the one
|
|
// stream with no state to show — the count only moves on an event.
|
|
report({ eventCount });
|
|
|
|
const consumeLine = (line: string): void => {
|
|
if (framing === "ndjson") {
|
|
if (line.trim() === "") return;
|
|
eventCount += 1;
|
|
reportCount();
|
|
return;
|
|
}
|
|
if (line === "") {
|
|
if (blockHasData) {
|
|
eventCount += 1;
|
|
blockHasData = false;
|
|
reportCount();
|
|
}
|
|
return;
|
|
}
|
|
// A field line is `name[:value]`; `data` with no colon is still a data
|
|
// field per the spec, so match the name rather than the `data:` prefix.
|
|
if (line === "data" || line.startsWith("data:")) blockHasData = true;
|
|
};
|
|
|
|
// Split off every complete line in `pending`. The SSE grammar (and the
|
|
// SDK's parser) accept LF, CRLF and a bare CR as a line ending, so all three
|
|
// end a line here; a CR that is the last byte so far is held back, since
|
|
// the LF of a CRLF pair may still be in flight in the next chunk.
|
|
const consumeCompleteLines = (): void => {
|
|
for (;;) {
|
|
const cr = pending.indexOf("\r");
|
|
const lf = pending.indexOf("\n");
|
|
const end = cr === -1 ? lf : lf === -1 ? cr : Math.min(cr, lf);
|
|
if (end === -1) return;
|
|
if (pending[end] === "\r") {
|
|
if (end === pending.length - 1) return;
|
|
const width = pending[end + 1] === "\n" ? 2 : 1;
|
|
consumeLine(pending.slice(0, end));
|
|
pending = pending.slice(end + width);
|
|
} else {
|
|
consumeLine(pending.slice(0, end));
|
|
pending = pending.slice(end + 1);
|
|
}
|
|
}
|
|
};
|
|
|
|
const pump = async (): Promise<void> => {
|
|
try {
|
|
for (;;) {
|
|
const { value, done } = await reader.read();
|
|
if (done) break;
|
|
pending += decoder.decode(value, { stream: true });
|
|
consumeCompleteLines();
|
|
}
|
|
// A CR held back as a possible half of a CRLF is, at end-of-stream, a
|
|
// whole line ending — `data: x\r\r` is a complete event. Anything after
|
|
// it is a partial block and is dropped, as above.
|
|
if (pending.endsWith("\r")) consumeLine(pending.slice(0, -1));
|
|
} catch {
|
|
// An errored or aborted stream has still ended — fall through to report
|
|
// the close; the entry's `error` field is for a fetch that never got a
|
|
// response, which this one did.
|
|
}
|
|
if (reportTimer !== null) {
|
|
clearTimeout(reportTimer);
|
|
reportTimer = null;
|
|
}
|
|
report({ eventCount, closedAt: new Date() });
|
|
};
|
|
// Deliberately not awaited: the fetch wrapper has to return the response
|
|
// now, and this watcher lives as long as the stream does. It owns its own
|
|
// failures — the read loop is wrapped in a `try`, and every report goes
|
|
// through the non-throwing `report` above — so the promise cannot reject.
|
|
void pump();
|
|
}
|
|
|
|
/**
|
|
* Creates a fetch wrapper that tracks HTTP requests and responses
|
|
*/
|
|
export function createFetchTracker(
|
|
baseFetch: typeof fetch,
|
|
callbacks: FetchTrackingCallbacks,
|
|
): typeof fetch {
|
|
return async (
|
|
input: RequestInfo | URL,
|
|
init?: RequestInit,
|
|
): Promise<Response> => {
|
|
const startTime = Date.now();
|
|
const timestamp = new Date();
|
|
const id = `${timestamp.getTime()}-${Math.random().toString(36).slice(2, 11)}`;
|
|
|
|
// Extract request information
|
|
const url =
|
|
typeof input === "string"
|
|
? input
|
|
: input instanceof URL
|
|
? input.toString()
|
|
: input.url;
|
|
// The method decides the long-lived-stream classification below, so it
|
|
// has to be the one the request actually carries: a `Request` input
|
|
// carries its own, and HTTP methods compare case-insensitively.
|
|
const method = (
|
|
init?.method ?? (input instanceof Request ? input.method : "GET")
|
|
).toUpperCase();
|
|
|
|
// Extract headers, redacting sensitive values BEFORE they reach any
|
|
// downstream sink (logger, in-memory list, persisted session storage).
|
|
const rawRequestHeaders: Record<string, string> = {};
|
|
if (input instanceof Request) {
|
|
input.headers.forEach((value, key) => {
|
|
rawRequestHeaders[key] = value;
|
|
});
|
|
}
|
|
if (init?.headers) {
|
|
const headers = new Headers(init.headers);
|
|
headers.forEach((value, key) => {
|
|
rawRequestHeaders[key] = value;
|
|
});
|
|
}
|
|
const requestHeaders = redactSensitiveHeaders(rawRequestHeaders);
|
|
const requestContentType = findHeader(rawRequestHeaders, "content-type");
|
|
|
|
// Redact sensitive query params in the recorded URL (live `input` is
|
|
// untouched — only this logged copy is masked).
|
|
const redactedUrl = redactUrlQuery(url);
|
|
|
|
// Extract body (if present and readable)
|
|
let requestBody: string | undefined;
|
|
if (init?.body) {
|
|
if (typeof init.body === "string") {
|
|
requestBody = init.body;
|
|
} else {
|
|
// Try to convert to string, but skip if it fails (e.g., ReadableStream)
|
|
try {
|
|
requestBody = String(init.body);
|
|
} catch {
|
|
requestBody = undefined;
|
|
}
|
|
}
|
|
} else if (input instanceof Request && input.body) {
|
|
// Try to clone and read the request body
|
|
// Clone protects the original body from being consumed
|
|
try {
|
|
const cloned = input.clone();
|
|
requestBody = await cloned.text();
|
|
} catch {
|
|
// Can't read body (might be consumed, not readable, or other issue)
|
|
requestBody = undefined;
|
|
}
|
|
}
|
|
|
|
// Redact sensitive fields in the recorded request body. The live request
|
|
// body (`init.body` / `input`) is never touched — only this logged string.
|
|
const redactedRequestBody = redactBody(requestBody, requestContentType);
|
|
|
|
// Make the actual fetch request
|
|
let response: Response;
|
|
let error: string | undefined;
|
|
try {
|
|
response = await baseFetch(input, init);
|
|
} catch (err) {
|
|
error = err instanceof Error ? err.message : String(err);
|
|
// Create a minimal error entry
|
|
const entry: FetchRequestEntryBase = {
|
|
id,
|
|
timestamp,
|
|
method,
|
|
url: redactedUrl,
|
|
requestHeaders,
|
|
requestBody: redactedRequestBody,
|
|
error,
|
|
duration: Date.now() - startTime,
|
|
};
|
|
callbacks.trackRequest?.(entry);
|
|
throw err;
|
|
}
|
|
|
|
// Extract response information
|
|
const responseStatus = response.status;
|
|
const responseStatusText = response.statusText;
|
|
|
|
// Extract response headers (redacted — Set-Cookie etc. are credentials too)
|
|
const rawResponseHeaders: Record<string, string> = {};
|
|
response.headers.forEach((value, key) => {
|
|
rawResponseHeaders[key] = value;
|
|
});
|
|
const responseHeaders = redactSensitiveHeaders(rawResponseHeaders);
|
|
|
|
// Skip body reading only for *long-lived* streams. On streamable HTTP,
|
|
// GET /mcp opens an unbounded SSE channel for server-to-client pushes
|
|
// — calling `.text()` on a clone of that would buffer forever. POST
|
|
// responses with the same content-type are bounded: the server emits
|
|
// the JSON-RPC reply (sometimes preceded by progress events) and
|
|
// closes the connection, so cloning + reading is safe and gives the
|
|
// user the raw SSE payload they were missing.
|
|
const isLongLivedStream = isLongLivedStreamResponse(
|
|
method,
|
|
response.headers.get("content-type"),
|
|
);
|
|
|
|
const duration = Date.now() - startTime;
|
|
|
|
// Create entry and track it immediately. The body is read asynchronously
|
|
// below to avoid blocking the transport — for streaming responses (POST
|
|
// + SSE), the server keeps the connection open until it has delivered
|
|
// every progress notification plus the final reply, so awaiting
|
|
// `.text()` here would force the transport to wait for all events
|
|
// before it could process any of them.
|
|
const entry: FetchRequestEntryBase = {
|
|
id,
|
|
timestamp,
|
|
method,
|
|
url: redactedUrl,
|
|
requestHeaders,
|
|
requestBody: redactedRequestBody,
|
|
responseStatus,
|
|
responseStatusText,
|
|
responseHeaders,
|
|
responseBody: undefined,
|
|
duration,
|
|
};
|
|
|
|
// Not gated on isLongLivedStream: that exclusion guards an unbounded read,
|
|
// and this one is bounded — so a challenge to an SSE GET is recorded too.
|
|
if (
|
|
(responseStatus === 401 || responseStatus === 403) &&
|
|
response.body &&
|
|
!response.bodyUsed
|
|
) {
|
|
const read = await readChallengeBody(response);
|
|
entry.responseBody =
|
|
read === undefined
|
|
? undefined
|
|
: recordableChallengeBody(read, response.headers.get("content-type"));
|
|
callbacks.trackRequest?.(entry);
|
|
return response;
|
|
}
|
|
|
|
callbacks.trackRequest?.(entry);
|
|
|
|
// A long-lived stream is watched instead of read: its body never ends,
|
|
// but its event count and its eventual close are exactly the state a
|
|
// stalled connection is diagnosed from (#2318).
|
|
if (isLongLivedStream && response.body && !response.bodyUsed) {
|
|
watchLongLivedStream(
|
|
response,
|
|
id,
|
|
longLivedStreamFraming(response.headers.get("content-type")),
|
|
callbacks,
|
|
);
|
|
}
|
|
|
|
// Kick off a fire-and-forget read of the cloned body. The clone is an
|
|
// independent tee'd stream so the transport keeps consuming the
|
|
// original at its own pace. When the read resolves we patch the entry
|
|
// via `updateResponseBody`. Skipped for long-lived streams (GET +
|
|
// SSE / ndjson) because `.text()` would never resolve on those.
|
|
if (!isLongLivedStream && response.body && !response.bodyUsed) {
|
|
const responseContentType = response.headers.get("content-type");
|
|
try {
|
|
const cloned = response.clone();
|
|
cloned
|
|
.text()
|
|
.then((body) => {
|
|
// Mask token-endpoint secrets (access_token, refresh_token, …)
|
|
// before the body reaches any sink.
|
|
callbacks.updateResponseBody?.(
|
|
id,
|
|
redactBody(body, responseContentType) ?? body,
|
|
);
|
|
})
|
|
.catch(() => {
|
|
// Stream errored after clone — leave the body undefined.
|
|
});
|
|
} catch {
|
|
// Clone failed (consumed body, transport quirks). Leave body
|
|
// undefined; the entry is already dispatched.
|
|
}
|
|
}
|
|
|
|
return response;
|
|
};
|
|
}
|