diff --git a/src/lib/inspector/harExport.ts b/src/lib/inspector/harExport.ts new file mode 100644 index 0000000000..441e5e1833 --- /dev/null +++ b/src/lib/inspector/harExport.ts @@ -0,0 +1,188 @@ +/** + * HAR (HTTP Archive) v1.2 export for the Traffic Inspector. + * + * HAR is the standard format consumed by Chrome DevTools, Charles, Fiddler, + * Postman, and most observability tools. Exporting lets users carry the + * trace out of OmniRoute into their existing workflow. + * + * Secrets are *always* masked on export, regardless of the UI state — see + * Hard Rule #1 (no credentials in artefacts). The OmniRoute capture source + * (agent-bridge / custom-host / http-proxy / system-proxy) is preserved as + * `_source`, a custom field allowed by the HAR spec's underscore convention. + */ + +import { maskSecret } from "@/mitm/maskSecrets"; +import type { InterceptedRequest } from "@/mitm/inspector/types"; + +const HAR_VERSION = "1.2"; +const CREATOR_NAME = "OmniRoute Traffic Inspector"; +const CREATOR_VERSION = "3.8.6"; + +interface HarNameValue { + name: string; + value: string; +} + +interface HarPostData { + mimeType: string; + text: string; +} + +interface HarRequest { + method: string; + url: string; + httpVersion: string; + headers: HarNameValue[]; + queryString: HarNameValue[]; + cookies: HarNameValue[]; + headersSize: number; + bodySize: number; + postData?: HarPostData; +} + +interface HarContent { + size: number; + mimeType: string; + text?: string; +} + +interface HarResponse { + status: number; + statusText: string; + httpVersion: string; + headers: HarNameValue[]; + cookies: HarNameValue[]; + content: HarContent; + redirectURL: string; + headersSize: number; + bodySize: number; +} + +interface HarTimings { + send: number; + wait: number; + receive: number; +} + +interface HarEntry { + startedDateTime: string; + time: number; + request: HarRequest; + response: HarResponse; + cache: Record; + timings: HarTimings; + serverIPAddress?: string; + _source?: string; + _agent?: string; + _detectedKind?: string; + _contextKey?: string; + _sessionId?: string; + _annotation?: string; + _note?: string; + _omniRouteId?: string; +} + +export interface HarFile { + log: { + version: string; + creator: { name: string; version: string }; + entries: HarEntry[]; + }; +} + +function headersToList(headers: Record): HarNameValue[] { + return Object.entries(headers).map(([name, value]) => ({ + name, + value: maskSecret(value), + })); +} + +function buildUrl(host: string, path: string): string { + // CONNECT entries carry path ":443" — treat them as opaque pseudo-URL. + if (path.startsWith(":")) return `https://${host}${path}`; + if (!host) return path; + return `https://${host}${path}`; +} + +function buildPostData(req: InterceptedRequest): HarPostData | undefined { + if (!req.requestBody) return undefined; + const ct = + req.requestHeaders["content-type"] ?? + req.requestHeaders["Content-Type"] ?? + "application/octet-stream"; + return { mimeType: ct, text: maskSecret(req.requestBody) }; +} + +function buildResponseContent(req: InterceptedRequest): HarContent { + const ct = + req.responseHeaders["content-type"] ?? + req.responseHeaders["Content-Type"] ?? + "application/octet-stream"; + if (req.responseBody == null) { + return { size: req.responseSize, mimeType: ct }; + } + return { size: req.responseSize, mimeType: ct, text: maskSecret(req.responseBody) }; +} + +function buildEntry(req: InterceptedRequest): HarEntry { + const numericStatus = typeof req.status === "number" ? req.status : 0; + const statusText = typeof req.status === "string" ? req.status : ""; + + const entry: HarEntry = { + startedDateTime: req.timestamp, + time: req.totalLatencyMs ?? 0, + request: { + method: req.method, + url: buildUrl(req.host, req.path), + httpVersion: "HTTP/1.1", + headers: headersToList(req.requestHeaders), + queryString: [], + cookies: [], + headersSize: -1, + bodySize: req.requestSize, + postData: buildPostData(req), + }, + response: { + status: numericStatus, + statusText, + httpVersion: "HTTP/1.1", + headers: headersToList(req.responseHeaders), + cookies: [], + content: buildResponseContent(req), + redirectURL: "", + headersSize: -1, + bodySize: req.responseSize, + }, + cache: {}, + timings: { + send: 0, + wait: req.upstreamLatencyMs ?? 0, + receive: (req.totalLatencyMs ?? 0) - (req.upstreamLatencyMs ?? 0), + }, + _source: req.source, + _omniRouteId: req.id, + }; + + if (req.agent) entry._agent = req.agent; + if (req.detectedKind) entry._detectedKind = req.detectedKind; + if (req.contextKey) entry._contextKey = req.contextKey; + if (req.sessionId) entry._sessionId = req.sessionId; + if (req.annotation) entry._annotation = req.annotation; + if (req.note) entry._note = req.note; + + return entry; +} + +/** + * Convert intercepted requests into a HAR v1.2 file. Always masks secrets in + * headers and bodies — callers do not need to pre-mask. + */ +export function toHar(requests: InterceptedRequest[]): HarFile { + return { + log: { + version: HAR_VERSION, + creator: { name: CREATOR_NAME, version: CREATOR_VERSION }, + entries: requests.map(buildEntry), + }, + }; +} diff --git a/src/mitm/inspector/agentBridgeHook.ts b/src/mitm/inspector/agentBridgeHook.ts new file mode 100644 index 0000000000..e247f38628 --- /dev/null +++ b/src/mitm/inspector/agentBridgeHook.ts @@ -0,0 +1,100 @@ +/** + * Inspector hook called from `MitmHandlerBase` (F3) on every intercepted + * AgentBridge request. Centralises buffer push/update so handlers do not need + * to know the inspector internals. + * + * Contract: see `_orchestration/master-plan-group-A.md` §3.11. + */ + +import { randomUUID } from "node:crypto"; +import type { IncomingMessage } from "node:http"; +import { sanitizeErrorMessage } from "@omniroute/open-sse/utils/error"; +import { maskSecret } from "../maskSecrets.ts"; +import { sanitizeHeaders } from "../sanitizeHeaders.ts"; +import type { AgentId } from "../types.ts"; +import { globalTrafficBuffer } from "./buffer.ts"; +import type { InterceptedRequest } from "./types.ts"; + +export interface RecordRequestStartOpts { + req: IncomingMessage; + body: Buffer; + agentId: AgentId; + mappedModel: string; + sourceModel?: string | null; + sessionId?: string; +} + +export interface RecordRequestCompleteOpts { + status: number; + responseHeaders: Record; + responseBody: string | null; + responseSize: number; + proxyLatencyMs: number; + upstreamLatencyMs: number; +} + +/** + * Build the initial buffer entry and push it. Returned object is mutable by + * design — handlers call `recordRequestComplete()` (or `recordRequestError`) + * with the same reference once the upstream call resolves. + */ +export async function recordRequestStart( + opts: RecordRequestStartOpts +): Promise { + const requestBody = opts.body.length > 0 ? maskSecret(opts.body.toString("utf8")) : null; + const intercepted: InterceptedRequest = { + id: randomUUID(), + source: "agent-bridge", + agent: opts.agentId, + timestamp: new Date().toISOString(), + method: opts.req.method ?? "GET", + host: opts.req.headers.host ?? "", + path: opts.req.url ?? "/", + requestHeaders: sanitizeHeaders(opts.req.headers), + requestBody, + requestSize: opts.body.length, + responseHeaders: {}, + responseBody: null, + responseSize: 0, + status: "in-flight", + sourceModel: opts.sourceModel ?? null, + mappedModel: opts.mappedModel, + }; + if (opts.sessionId) intercepted.sessionId = opts.sessionId; + + globalTrafficBuffer.push(intercepted); + return intercepted; +} + +/** + * Finalise the buffer entry with the upstream response data. Latencies are + * stored as-given and combined into `totalLatencyMs`. + */ +export function recordRequestComplete( + intercepted: InterceptedRequest, + opts: RecordRequestCompleteOpts +): void { + intercepted.status = opts.status; + intercepted.responseHeaders = opts.responseHeaders; + intercepted.responseBody = + opts.responseBody != null ? maskSecret(opts.responseBody) : null; + intercepted.responseSize = opts.responseSize; + intercepted.proxyLatencyMs = opts.proxyLatencyMs; + intercepted.upstreamLatencyMs = opts.upstreamLatencyMs; + intercepted.totalLatencyMs = opts.proxyLatencyMs + opts.upstreamLatencyMs; + + globalTrafficBuffer.update(intercepted.id, intercepted); +} + +/** + * Mark the buffer entry as failed. Error messages are sanitized so stack + * traces or absolute paths cannot leak to dashboards/exports (Hard Rule #12). + */ +export function recordRequestError( + intercepted: InterceptedRequest, + err: unknown +): void { + intercepted.status = "error"; + intercepted.error = sanitizeErrorMessage(err); + globalTrafficBuffer.update(intercepted.id, intercepted); +} diff --git a/src/mitm/inspector/buffer.ts b/src/mitm/inspector/buffer.ts new file mode 100644 index 0000000000..955d8a8fe3 --- /dev/null +++ b/src/mitm/inspector/buffer.ts @@ -0,0 +1,201 @@ +/** + * In-memory ring buffer for intercepted traffic. + * + * Stores up to `INSPECTOR_BUFFER_SIZE` (default 1000) entries; rotates + * oldest-first when capacity is reached. Auto-applies kind detection and + * context-key fingerprinting on push, and broadcasts mutations to all + * subscribers (WebSocket consumers). + * + * Body sizes are clamped to `INSPECTOR_MAX_BODY_KB` (default 1024 KiB) and + * marked with a truncation suffix so the UI does not have to guess. + * + * See `_orchestration/master-plan-group-A.md` §3.6 and + * `12-traffic-inspector.plan.md` §4.1. + */ + +import { computeContextKey } from "./contextKey.ts"; +import { detectKind } from "./kindDetector.ts"; +import type { InterceptedRequest, ListFilters, WsEvent } from "./types.ts"; + +const TRUNCATION_MARKER = "\n…(truncated for performance)"; + +function parseEnvNumber(value: string | undefined, fallback: number): number { + if (!value) return fallback; + const parsed = Number(value); + if (!Number.isFinite(parsed) || parsed <= 0) return fallback; + return parsed; +} + +function getMaxBodyBytes(): number { + const kb = parseEnvNumber(process.env.INSPECTOR_MAX_BODY_KB, 1024); + return Math.max(1, Math.floor(kb)) * 1024; +} + +function capBody(body: string | null, maxBytes: number): string | null { + if (body == null) return body; + if (body.length <= maxBytes) return body; + return body.slice(0, maxBytes) + TRUNCATION_MARKER; +} + +function statusBucket(status: InterceptedRequest["status"]): string { + if (status === "error") return "error"; + if (status === "in-flight") return "in-flight"; + if (typeof status !== "number") return "unknown"; + if (status >= 200 && status < 300) return "2xx"; + if (status >= 300 && status < 400) return "3xx"; + if (status >= 400 && status < 500) return "4xx"; + if (status >= 500 && status < 600) return "5xx"; + return "unknown"; +} + +function matchesFilters(req: InterceptedRequest, filters?: ListFilters): boolean { + if (!filters) return true; + + if (filters.profile && filters.profile !== "all") { + if (filters.profile === "llm" && req.detectedKind !== "llm") return false; + if (filters.profile === "custom" && req.source !== "custom-host") return false; + } + + if (filters.host && req.host !== filters.host) return false; + if (filters.agent && req.agent !== filters.agent) return false; + if (filters.source && req.source !== filters.source) return false; + if (filters.sessionId && req.sessionId !== filters.sessionId) return false; + + if (filters.status) { + const bucket = statusBucket(req.status); + if (bucket !== filters.status) return false; + } + + return true; +} + +/** + * Ring buffer with broadcast support. + * + * Designed to be process-singleton (`globalTrafficBuffer`); tests can + * instantiate isolated buffers when needed. + */ +export class TrafficBuffer { + private buffer: InterceptedRequest[] = []; + private subscribers = new Set<(ev: WsEvent) => void>(); + private maxSize: number; + private maxBodyBytes: number; + + constructor( + maxSize: number = parseEnvNumber(process.env.INSPECTOR_BUFFER_SIZE, 1000), + maxBodyBytes: number = getMaxBodyBytes() + ) { + this.maxSize = Math.max(1, Math.floor(maxSize)); + this.maxBodyBytes = Math.max(1, Math.floor(maxBodyBytes)); + } + + /** + * Append a new intercepted request. Applies kind detection and + * context-key fingerprinting if missing, and clamps body sizes. + * Broadcasts a `new` event to all subscribers. + */ + push(req: InterceptedRequest): void { + if (!req.detectedKind) { + req.detectedKind = detectKind(req); + } + if (!req.contextKey && req.detectedKind === "llm") { + const key = computeContextKey(req); + if (key) req.contextKey = key; + } + + req.requestBody = capBody(req.requestBody, this.maxBodyBytes); + req.responseBody = capBody(req.responseBody, this.maxBodyBytes); + + this.buffer.push(req); + while (this.buffer.length > this.maxSize) { + this.buffer.shift(); + } + + this.broadcast({ type: "new", data: req }); + } + + /** + * Update an existing entry in place by id. No-op if the id is unknown + * (e.g. already rotated out). Broadcasts an `update` event on success. + */ + update(id: string, req: InterceptedRequest): void { + const idx = this.buffer.findIndex((r) => r.id === id); + if (idx < 0) return; + + req.requestBody = capBody(req.requestBody, this.maxBodyBytes); + req.responseBody = capBody(req.responseBody, this.maxBodyBytes); + + this.buffer[idx] = req; + this.broadcast({ type: "update", data: req }); + } + + /** + * Lookup by id (linear scan — buffer is bounded to ~1000 entries). + */ + get(id: string): InterceptedRequest | null { + return this.buffer.find((r) => r.id === id) ?? null; + } + + /** + * Return a filtered snapshot of the buffer. Filtering is in-memory and + * cheap (~O(maxSize)). A new array is returned each call. + */ + list(filters?: ListFilters): InterceptedRequest[] { + if (!filters) return [...this.buffer]; + return this.buffer.filter((r) => matchesFilters(r, filters)); + } + + /** + * Empty the buffer and notify subscribers. Subscriber count is preserved. + */ + clear(): void { + this.buffer = []; + this.broadcast({ type: "clear" }); + } + + /** + * Register a listener. Immediately receives a `snapshot` event with the + * current buffer state. Returns an `unsubscribe` function. + */ + subscribe(fn: (ev: WsEvent) => void): () => void { + this.subscribers.add(fn); + try { + fn({ type: "snapshot", data: [...this.buffer] }); + } catch { + // a subscriber's snapshot handler failure must not break subscription + } + return () => { + this.subscribers.delete(fn); + }; + } + + /** + * Current subscriber count — exposed for tests / diagnostics. + */ + subscriberCount(): number { + return this.subscribers.size; + } + + /** + * Current entry count — exposed for tests / diagnostics. + */ + size(): number { + return this.buffer.length; + } + + private broadcast(ev: WsEvent): void { + for (const fn of this.subscribers) { + try { + fn(ev); + } catch { + // one subscriber's failure must not block others + } + } + } +} + +/** + * Process-wide singleton consumed by `agentBridgeHook`, `httpProxyServer`, + * REST/WS routes, and tests. + */ +export const globalTrafficBuffer = new TrafficBuffer(); diff --git a/src/mitm/inspector/conversationNormalizer.ts b/src/mitm/inspector/conversationNormalizer.ts new file mode 100644 index 0000000000..b38e9acd2b --- /dev/null +++ b/src/mitm/inspector/conversationNormalizer.ts @@ -0,0 +1,393 @@ +/** + * Conversation normalizer — converts OpenAI / Anthropic / Gemini request + + * response payloads into a single provider-agnostic shape. + * + * MIT — port from https://github.com/chouzz/llm-interceptor (ui/utils.ts) + * + * Returns `null` for non-LLM requests or payloads we cannot understand — + * never throws — so the renderer can fall back to the raw view. + */ + +import { mergeStream, parseSseStream } from "./sseMerger.ts"; +import type { + InterceptedRequest, + NormalizedBlock, + NormalizedConversation, + NormalizedTurn, +} from "./types.ts"; + +type NormalizedRole = NormalizedTurn["role"]; + +function asRecord(value: unknown): Record | null { + if (value && typeof value === "object" && !Array.isArray(value)) { + return value as Record; + } + return null; +} + +function tryParseJson(value: string | null | undefined): unknown { + if (!value) return null; + try { + return JSON.parse(value); + } catch { + return null; + } +} + +function normalizeRole(raw: unknown): NormalizedRole { + if (raw === "system" || raw === "user" || raw === "assistant" || raw === "tool") { + return raw; + } + if (raw === "model") return "assistant"; + if (raw === "function") return "tool"; + return "user"; +} + +/** + * OpenAI / Anthropic message content can be a string, or an array of blocks. + * Returns a list of normalized blocks. + */ +function blocksFromOpenAiContent(content: unknown): NormalizedBlock[] { + if (content == null) return []; + if (typeof content === "string") { + if (content.length === 0) return []; + return [{ type: "text", text: content }]; + } + if (!Array.isArray(content)) return []; + const out: NormalizedBlock[] = []; + for (const raw of content) { + if (typeof raw === "string") { + out.push({ type: "text", text: raw }); + continue; + } + const block = asRecord(raw); + if (!block) continue; + const type = block.type; + if (type === "text" || type === "output_text") { + const text = typeof block.text === "string" ? block.text : ""; + out.push({ type: "text", text }); + } else if (type === "input_text") { + const text = typeof block.text === "string" ? block.text : ""; + out.push({ type: "text", text }); + } else if (type === "tool_use") { + out.push({ + type: "tool_use", + id: typeof block.id === "string" ? block.id : "", + name: typeof block.name === "string" ? block.name : "", + input: block.input ?? {}, + }); + } else if (type === "tool_result") { + out.push({ + type: "tool_result", + tool_use_id: + typeof block.tool_use_id === "string" ? block.tool_use_id : "", + content: block.content ?? null, + }); + } else if (typeof block.text === "string") { + out.push({ type: "text", text: block.text }); + } + } + return out; +} + +/** + * OpenAI assistant messages may declare `tool_calls`. Each becomes a + * `tool_use` block alongside any text content. + */ +function appendOpenAiToolCalls( + blocks: NormalizedBlock[], + toolCalls: unknown +): NormalizedBlock[] { + if (!Array.isArray(toolCalls)) return blocks; + for (const raw of toolCalls) { + const tc = asRecord(raw); + if (!tc) continue; + const fn = asRecord(tc.function) ?? {}; + let parsedInput: unknown = {}; + if (typeof fn.arguments === "string") { + try { + parsedInput = JSON.parse(fn.arguments); + } catch { + parsedInput = fn.arguments; + } + } else if (fn.arguments != null) { + parsedInput = fn.arguments; + } + blocks.push({ + type: "tool_use", + id: typeof tc.id === "string" ? tc.id : "", + name: typeof fn.name === "string" ? fn.name : "", + input: parsedInput, + }); + } + return blocks; +} + +/** + * Build NormalizedTurn[] from OpenAI / Anthropic chat messages. + */ +function turnsFromOpenAiMessages(messages: unknown[]): NormalizedTurn[] { + const out: NormalizedTurn[] = []; + for (const raw of messages) { + const msg = asRecord(raw); + if (!msg) continue; + const role = normalizeRole(msg.role); + + if (msg.role === "tool" || msg.role === "function") { + const content = msg.content; + out.push({ + role: "tool", + blocks: [ + { + type: "tool_result", + tool_use_id: + typeof msg.tool_call_id === "string" + ? msg.tool_call_id + : typeof msg.name === "string" + ? msg.name + : "", + content, + }, + ], + }); + continue; + } + + const blocks = blocksFromOpenAiContent(msg.content); + if ("tool_calls" in msg) { + appendOpenAiToolCalls(blocks, msg.tool_calls); + } + if (blocks.length === 0 && msg.content == null && !("tool_calls" in msg)) { + continue; + } + out.push({ role, blocks }); + } + return out; +} + +/** + * Gemini contents have a different shape: `[{role, parts: [{text|...}]}]`. + */ +function turnsFromGeminiContents(contents: unknown[]): NormalizedTurn[] { + const out: NormalizedTurn[] = []; + for (const raw of contents) { + const turn = asRecord(raw); + if (!turn) continue; + const role = normalizeRole(turn.role); + const blocks: NormalizedBlock[] = []; + if (Array.isArray(turn.parts)) { + for (const partRaw of turn.parts) { + const part = asRecord(partRaw); + if (!part) continue; + if (typeof part.text === "string") { + blocks.push({ type: "text", text: part.text }); + } else if (part.functionCall) { + const fc = asRecord(part.functionCall) ?? {}; + blocks.push({ + type: "tool_use", + id: typeof fc.name === "string" ? fc.name : "", + name: typeof fc.name === "string" ? fc.name : "", + input: fc.args ?? {}, + }); + } else if (part.functionResponse) { + const fr = asRecord(part.functionResponse) ?? {}; + blocks.push({ + type: "tool_result", + tool_use_id: typeof fr.name === "string" ? fr.name : "", + content: fr.response ?? null, + }); + } + } + } + if (blocks.length > 0) out.push({ role, blocks }); + } + return out; +} + +/** + * Anthropic Messages API requests carry a top-level `system` field (string + * or array of `{type:"text"|text}` blocks). Convert to a `system` turn. + */ +function systemTurnFromAnthropic(system: unknown): NormalizedTurn | null { + if (!system) return null; + if (typeof system === "string") { + return system.length === 0 + ? null + : { role: "system", blocks: [{ type: "text", text: system }] }; + } + if (!Array.isArray(system)) return null; + const blocks: NormalizedBlock[] = []; + for (const raw of system) { + const item = asRecord(raw); + if (item && typeof item.text === "string") { + blocks.push({ type: "text", text: item.text }); + } else if (typeof raw === "string") { + blocks.push({ type: "text", text: raw }); + } + } + if (blocks.length === 0) return null; + return { role: "system", blocks }; +} + +function buildRequestTurns(body: unknown): NormalizedTurn[] | null { + const obj = asRecord(body); + if (!obj) return null; + + if (Array.isArray(obj.messages)) { + const turns: NormalizedTurn[] = []; + const systemTurn = systemTurnFromAnthropic(obj.system); + if (systemTurn) turns.push(systemTurn); + turns.push(...turnsFromOpenAiMessages(obj.messages)); + return turns; + } + + if (Array.isArray(obj.contents)) { + const turns: NormalizedTurn[] = []; + const sysObj = asRecord(obj.systemInstruction); + if (sysObj && Array.isArray(sysObj.parts)) { + const parts: NormalizedBlock[] = []; + for (const partRaw of sysObj.parts) { + const p = asRecord(partRaw); + if (p && typeof p.text === "string") parts.push({ type: "text", text: p.text }); + } + if (parts.length > 0) turns.push({ role: "system", blocks: parts }); + } + turns.push(...turnsFromGeminiContents(obj.contents)); + return turns; + } + + if (typeof obj.prompt === "string") { + return [{ role: "user", blocks: [{ type: "text", text: obj.prompt }] }]; + } + if (typeof obj.input === "string") { + return [{ role: "user", blocks: [{ type: "text", text: obj.input }] }]; + } + if (Array.isArray(obj.input)) { + return turnsFromOpenAiMessages(obj.input); + } + + return null; +} + +function isSseResponse(req: InterceptedRequest): boolean { + const accept = req.requestHeaders["accept"] ?? req.requestHeaders["Accept"] ?? ""; + const ct = req.responseHeaders["content-type"] ?? req.responseHeaders["Content-Type"] ?? ""; + return ( + accept.includes("event-stream") || + ct.includes("event-stream") || + /^\s*event:|^\s*data:/m.test(req.responseBody ?? "") + ); +} + +function extractAnthropicResponseTurn(message: unknown): NormalizedTurn | null { + const obj = asRecord(message); + if (!obj) return null; + const content = obj.content; + if (!Array.isArray(content)) return null; + const blocks: NormalizedBlock[] = []; + for (const raw of content) { + const block = asRecord(raw); + if (!block) continue; + if (block.type === "text" && typeof block.text === "string") { + blocks.push({ type: "text", text: block.text }); + } else if (block.type === "tool_use") { + blocks.push({ + type: "tool_use", + id: typeof block.id === "string" ? block.id : "", + name: typeof block.name === "string" ? block.name : "", + input: block.input ?? {}, + }); + } else if (block.type === "thinking" && typeof block.thinking === "string") { + blocks.push({ type: "text", text: block.thinking }); + } + } + if (blocks.length === 0) return null; + return { role: "assistant", blocks }; +} + +function extractOpenAiResponseTurn(message: unknown): NormalizedTurn | null { + const obj = asRecord(message); + if (!obj || !Array.isArray(obj.choices)) return null; + const first = asRecord(obj.choices[0]); + if (!first) return null; + const msg = asRecord(first.message) ?? asRecord(first.delta); + if (!msg) return null; + const blocks = blocksFromOpenAiContent(msg.content); + if ("tool_calls" in msg) appendOpenAiToolCalls(blocks, msg.tool_calls); + if (blocks.length === 0) return null; + return { role: "assistant", blocks }; +} + +function extractGeminiResponseTurn(message: unknown): NormalizedTurn | null { + const obj = asRecord(message); + if (!obj || !Array.isArray(obj.candidates)) return null; + const first = asRecord(obj.candidates[0]); + if (!first) return null; + const content = asRecord(first.content); + if (!content || !Array.isArray(content.parts)) return null; + const blocks: NormalizedBlock[] = []; + for (const partRaw of content.parts) { + const part = asRecord(partRaw); + if (!part) continue; + if (typeof part.text === "string") { + blocks.push({ type: "text", text: part.text }); + } else if (part.functionCall) { + const fc = asRecord(part.functionCall) ?? {}; + blocks.push({ + type: "tool_use", + id: typeof fc.name === "string" ? fc.name : "", + name: typeof fc.name === "string" ? fc.name : "", + input: fc.args ?? {}, + }); + } + } + if (blocks.length === 0) return null; + return { role: "assistant", blocks }; +} + +function buildResponseTurns(req: InterceptedRequest): NormalizedTurn[] { + const raw = req.responseBody ?? ""; + if (!raw) return []; + + let payload: unknown = null; + + if (isSseResponse(req)) { + const merged = mergeStream(parseSseStream(raw)); + payload = merged.message ?? null; + } else { + payload = tryParseJson(raw); + } + + if (!payload) return []; + + const anth = extractAnthropicResponseTurn(payload); + if (anth) return [anth]; + const oai = extractOpenAiResponseTurn(payload); + if (oai) return [oai]; + const gem = extractGeminiResponseTurn(payload); + if (gem) return [gem]; + + return []; +} + +/** + * Normalize an intercepted LLM request + response into a provider-agnostic + * conversation. Returns `null` for non-LLM requests or unparseable payloads. + */ +export function normalizeConversation( + req: InterceptedRequest +): NormalizedConversation | null { + if (req.detectedKind !== "llm") return null; + + const requestBody = tryParseJson(req.requestBody); + const requestTurns = buildRequestTurns(requestBody); + if (!requestTurns) return null; + + const responseTurns = buildResponseTurns(req); + + return { + request: requestTurns, + response: responseTurns, + contextKey: req.contextKey ?? null, + }; +} diff --git a/src/mitm/inspector/httpProxyServer.ts b/src/mitm/inspector/httpProxyServer.ts new file mode 100644 index 0000000000..5fdb7dc80b --- /dev/null +++ b/src/mitm/inspector/httpProxyServer.ts @@ -0,0 +1,264 @@ +/** + * HTTP_PROXY listener for the Traffic Inspector. + * + * Accepts HTTP_PROXY=http://127.0.0.1:8080 style upstream traffic. Two paths: + * + * 1. HTTP direct (non-CONNECT): the proxy reads the request body, forwards + * it via `fetch()`, captures the response, and records the full exchange. + * 2. CONNECT (TLS tunnel): the proxy opens a raw TCP bridge so HTTPS still + * works, but only metadata (host:port) is captured — bodies stay opaque. + * The buffer entry carries a `note` field explaining why. + * + * `EADDRINUSE` during `listen()` rejects the returned promise so callers can + * surface a clean error to the user. See master-plan §3.12 + plan 12 §4.2.6. + */ + +import http from "node:http"; +import net from "node:net"; +import { randomUUID } from "node:crypto"; +import { sanitizeErrorMessage } from "@omniroute/open-sse/utils/error"; +import { sanitizeHeaders } from "../sanitizeHeaders.ts"; +import { maskSecret } from "../maskSecrets.ts"; +import { globalTrafficBuffer } from "./buffer.ts"; +import type { InterceptedRequest } from "./types.ts"; + +const DEFAULT_PORT = parseEnvNumber(process.env.INSPECTOR_HTTP_PROXY_PORT, 8080); + +function parseEnvNumber(value: string | undefined, fallback: number): number { + if (!value) return fallback; + const parsed = Number(value); + if (!Number.isFinite(parsed) || parsed <= 0) return fallback; + return parsed; +} + +export interface HttpProxyServerHandle { + port: number; + server: http.Server; + stop(): Promise; +} + +/** + * Build a sanitized Headers object suitable for an upstream `fetch()`. + * Drops hop-by-hop fields via the existing denylist and coerces array values. + */ +function buildFetchHeaders(raw: http.IncomingHttpHeaders): Record { + // sanitizeHeaders applies upstream denylist + masks Authorization for buffer + // logging; here we want denylist only (so the upstream still sees the auth + // header). Reuse sanitizeHeaders with a separate masking pass for the buffer. + const out: Record = {}; + for (const [name, value] of Object.entries(raw)) { + if (value === undefined || value === null) continue; + const lower = name.toLowerCase(); + // Skip hop-by-hop / framing — same names sanitizeHeaders also drops. + if ( + lower === "host" || + lower === "connection" || + lower === "keep-alive" || + lower === "proxy-authenticate" || + lower === "proxy-authorization" || + lower === "te" || + lower === "trailer" || + lower === "transfer-encoding" || + lower === "upgrade" || + lower === "content-length" + ) { + continue; + } + out[lower] = Array.isArray(value) ? value.join(", ") : String(value); + } + return out; +} + +function safeUrl(rawUrl: string | undefined, hostHeader: string | undefined): URL | null { + if (!rawUrl) return null; + try { + if (/^https?:\/\//i.test(rawUrl)) return new URL(rawUrl); + if (hostHeader) return new URL(`http://${hostHeader}${rawUrl}`); + } catch { + return null; + } + return null; +} + +async function readBody(req: http.IncomingMessage): Promise { + const chunks: Buffer[] = []; + for await (const chunk of req) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + return Buffer.concat(chunks); +} + +function handleHttp(req: http.IncomingMessage, res: http.ServerResponse): void { + const startedAt = performance.now(); + const intercepted: InterceptedRequest = { + id: randomUUID(), + source: "http-proxy", + timestamp: new Date().toISOString(), + method: req.method ?? "GET", + host: req.headers.host ?? "", + path: req.url ?? "/", + requestHeaders: sanitizeHeaders(req.headers), + requestBody: null, + requestSize: 0, + responseHeaders: {}, + responseBody: null, + responseSize: 0, + status: "in-flight", + }; + + globalTrafficBuffer.push(intercepted); + + void (async () => { + try { + const target = safeUrl(req.url, req.headers.host); + if (!target) { + throw new Error("Invalid request URL"); + } + intercepted.host = target.host; + intercepted.path = target.pathname + target.search; + + const body = await readBody(req); + intercepted.requestSize = body.length; + intercepted.requestBody = body.length > 0 ? maskSecret(body.toString("utf8")) : null; + + const upstreamHeaders = buildFetchHeaders(req.headers); + const upstream = await fetch(target.toString(), { + method: req.method ?? "GET", + headers: upstreamHeaders, + body: body.length > 0 ? body : undefined, + redirect: "manual", + }); + + const respBuf = Buffer.from(await upstream.arrayBuffer()); + const totalLatencyMs = performance.now() - startedAt; + + intercepted.responseHeaders = sanitizeHeaders( + Object.fromEntries(upstream.headers) as Record + ); + intercepted.responseBody = maskSecret(respBuf.toString("utf8")); + intercepted.responseSize = respBuf.length; + intercepted.status = upstream.status; + intercepted.totalLatencyMs = totalLatencyMs; + intercepted.upstreamLatencyMs = totalLatencyMs; + intercepted.proxyLatencyMs = 0; + + const safeRespHeaders: Record = {}; + upstream.headers.forEach((value, key) => { + if (key.toLowerCase() === "content-length") return; + if (key.toLowerCase() === "transfer-encoding") return; + safeRespHeaders[key] = value; + }); + res.writeHead(upstream.status, safeRespHeaders); + res.end(respBuf); + + globalTrafficBuffer.update(intercepted.id, intercepted); + } catch (err) { + intercepted.status = "error"; + intercepted.error = sanitizeErrorMessage(err); + intercepted.totalLatencyMs = performance.now() - startedAt; + globalTrafficBuffer.update(intercepted.id, intercepted); + if (!res.headersSent) { + res.writeHead(502, { "content-type": "text/plain" }); + res.end("Bad Gateway"); + } else { + res.end(); + } + } + })(); +} + +function handleConnect( + req: http.IncomingMessage, + clientSocket: net.Socket, + head: Buffer +): void { + const target = req.url ?? ""; + const [host, rawPort] = target.split(":"); + const port = Number(rawPort) || 443; + + const intercepted: InterceptedRequest = { + id: randomUUID(), + source: "http-proxy", + timestamp: new Date().toISOString(), + method: "CONNECT", + host, + path: `:${port}`, + requestHeaders: sanitizeHeaders(req.headers), + requestBody: null, + requestSize: 0, + responseHeaders: {}, + responseBody: null, + responseSize: 0, + status: "in-flight", + note: "TLS tunnel — for body capture, redirect host via Custom Hosts mode", + }; + + globalTrafficBuffer.push(intercepted); + + const targetSocket = net.connect(port, host); + + const finalize = (status: number | "error", err?: unknown): void => { + intercepted.status = status; + if (err !== undefined) intercepted.error = sanitizeErrorMessage(err); + globalTrafficBuffer.update(intercepted.id, intercepted); + }; + + targetSocket.once("connect", () => { + clientSocket.write("HTTP/1.1 200 Connection Established\r\n\r\n"); + if (head && head.length > 0) targetSocket.write(head); + targetSocket.pipe(clientSocket); + clientSocket.pipe(targetSocket); + finalize(200); + }); + + const onError = (err: unknown): void => { + finalize("error", err); + try { + clientSocket.end(); + } catch { + // socket already closed + } + try { + targetSocket.destroy(); + } catch { + // already destroyed + } + }; + + targetSocket.once("error", onError); + clientSocket.once("error", onError); +} + +/** + * Start the HTTP_PROXY listener. Resolves with a handle once `listening` has + * fired; rejects with `Error.code === "EADDRINUSE"` (and similar) when the + * bind fails so callers can surface a clean error. + */ +export function startHttpProxyServer(port: number = DEFAULT_PORT): Promise { + return new Promise((resolve, reject) => { + const server = http.createServer(); + + server.on("request", (req, res) => handleHttp(req, res)); + server.on("connect", (req, socket, head) => handleConnect(req, socket as net.Socket, head)); + + server.once("error", (err: NodeJS.ErrnoException) => { + // Decorate with a code so callers can pattern-match without parsing strings. + reject(Object.assign(err, { code: err.code ?? "ELISTEN" })); + }); + + server.once("listening", () => { + const addr = server.address(); + const boundPort = typeof addr === "object" && addr ? addr.port : port; + resolve({ + port: boundPort, + server, + stop: () => + new Promise((res) => { + server.close(() => res()); + }), + }); + }); + + server.listen(port, "127.0.0.1"); + }); +} diff --git a/src/mitm/inspector/llmMetadataExtractor.ts b/src/mitm/inspector/llmMetadataExtractor.ts new file mode 100644 index 0000000000..97209513db --- /dev/null +++ b/src/mitm/inspector/llmMetadataExtractor.ts @@ -0,0 +1,182 @@ +/** + * Extract LLM-specific metadata from intercepted requests so the UI can + * render summary chips (provider, model, tokens, cost). Provider/api inference + * is host- and path-based; token counts come from the upstream `usage` block. + * + * Replaces the stub `extractLlmMetadata` left in `kindDetector.ts` by F1. + * Cost estimation is deferred to a future audit pass; we return `null`. + */ + +import { detectKind } from "./kindDetector.ts"; +import { mergeStream, parseSseStream } from "./sseMerger.ts"; +import type { InterceptedRequest, LlmMetadata } from "./types.ts"; + +interface ProviderMatch { + pattern: RegExp; + provider: string; +} + +const PROVIDER_MATCHERS: ProviderMatch[] = [ + { pattern: /(^|\.)openai\.com$/i, provider: "openai" }, + { pattern: /(^|\.)openai\.azure\.com$/i, provider: "azure-openai" }, + { pattern: /(^|\.)anthropic\.com$/i, provider: "anthropic" }, + { pattern: /generativelanguage\.googleapis\.com$/i, provider: "gemini" }, + { pattern: /(^|\.)aiplatform\.googleapis\.com$/i, provider: "vertex" }, + { pattern: /(^|\.)mistral\.ai$/i, provider: "mistral" }, + { pattern: /(^|\.)deepseek\.com$/i, provider: "deepseek" }, + { pattern: /(^|\.)groq\.com$/i, provider: "groq" }, + { pattern: /(^|\.)together\.xyz$/i, provider: "together" }, + { pattern: /(^|\.)fireworks\.ai$/i, provider: "fireworks" }, + { pattern: /(^|\.)cohere\.com$/i, provider: "cohere" }, + { pattern: /(^|\.)perplexity\.ai$/i, provider: "perplexity" }, + { pattern: /(^|\.)huggingface\.co$/i, provider: "huggingface" }, + { pattern: /(^|\.)openrouter\.ai$/i, provider: "openrouter" }, + { pattern: /(^|\.)x\.ai$/i, provider: "xai" }, + { pattern: /(^|\.)moonshot\.ai$/i, provider: "moonshot" }, + { pattern: /bigmodel\.cn$/i, provider: "bigmodel" }, + { pattern: /(^|\.)githubcopilot\.com$/i, provider: "github-copilot" }, + { pattern: /(^|\.)cursor\.sh$/i, provider: "cursor" }, + { pattern: /(^|\.)zed\.dev$/i, provider: "zed" }, +]; + +interface ApiKindMatch { + pattern: RegExp; + apiKind: string; +} + +const API_KIND_MATCHERS: ApiKindMatch[] = [ + { pattern: /\/(v1|v1beta)?\/?chat\/completions/i, apiKind: "chat.completions" }, + { pattern: /\/(v1|v1beta)\/messages/i, apiKind: "messages" }, + { pattern: /\/(v1|v1beta)?\/?embeddings/i, apiKind: "embeddings" }, + { pattern: /\/(v1|v1beta)?\/?responses/i, apiKind: "responses" }, + { pattern: /\/streamGenerateContent/i, apiKind: "streamGenerateContent" }, + { pattern: /\/generateContent/i, apiKind: "generateContent" }, + { pattern: /\/(v1|v1beta)\/completions/i, apiKind: "completions" }, +]; + +function asRecord(value: unknown): Record | null { + if (value && typeof value === "object" && !Array.isArray(value)) { + return value as Record; + } + return null; +} + +function safeParseJson(value: string | null | undefined): unknown { + if (!value) return null; + try { + return JSON.parse(value); + } catch { + return null; + } +} + +function inferProvider(host: string): string | null { + for (const m of PROVIDER_MATCHERS) { + if (m.pattern.test(host)) return m.provider; + } + return null; +} + +function inferApiKind(path: string): string | null { + for (const m of API_KIND_MATCHERS) { + if (m.pattern.test(path)) return m.apiKind; + } + return null; +} + +function countMessages(body: Record | null): number { + if (!body) return 0; + if (Array.isArray(body.messages)) return body.messages.length; + if (Array.isArray(body.contents)) return body.contents.length; + if (Array.isArray(body.input)) return body.input.length; + return 0; +} + +function isSseRequest(req: InterceptedRequest): boolean { + const accept = req.requestHeaders["accept"] ?? req.requestHeaders["Accept"] ?? ""; + const ct = req.responseHeaders["content-type"] ?? req.responseHeaders["Content-Type"] ?? ""; + return ( + accept.includes("event-stream") || + ct.includes("event-stream") || + /^\s*event:|^\s*data:/m.test(req.responseBody ?? "") + ); +} + +function maybeNumber(v: unknown): number | null { + return typeof v === "number" && Number.isFinite(v) ? v : null; +} + +function extractUsage(resp: unknown): { tokensIn: number | null; tokensOut: number | null } { + // Direct JSON + const respObj = asRecord(resp); + const usage = asRecord(respObj?.usage); + if (usage) { + const inTok = + maybeNumber(usage.prompt_tokens) ?? + maybeNumber(usage.input_tokens) ?? + maybeNumber((asRecord(usage.promptTokensDetails) ?? {}).total) ?? + null; + const outTok = + maybeNumber(usage.completion_tokens) ?? + maybeNumber(usage.output_tokens) ?? + maybeNumber((asRecord(usage.completionTokensDetails) ?? {}).total) ?? + null; + return { tokensIn: inTok, tokensOut: outTok }; + } + // Gemini-style usageMetadata + const um = asRecord(respObj?.usageMetadata); + if (um) { + const inTok = maybeNumber(um.promptTokenCount); + const outTok = maybeNumber(um.candidatesTokenCount); + return { tokensIn: inTok, tokensOut: outTok }; + } + return { tokensIn: null, tokensOut: null }; +} + +/** + * Extract LLM metadata. Returns `null` for non-LLM requests; otherwise + * returns best-effort fields (any unknown field is `null`). + */ +export function extractLlmMetadata(req: InterceptedRequest): LlmMetadata | null { + const kind = req.detectedKind ?? detectKind(req); + if (kind !== "llm") return null; + + const body = asRecord(safeParseJson(req.requestBody)); + let resp: unknown = safeParseJson(req.responseBody); + + // If response was SSE, try to merge it for usage metadata. + if (!resp && req.responseBody && isSseRequest(req)) { + const merged = mergeStream(parseSseStream(req.responseBody)); + resp = merged.message ?? null; + } + + const respObj = asRecord(resp); + + const provider = inferProvider(req.host); + const apiKind = inferApiKind(req.path); + const model = + (body && typeof body.model === "string" ? body.model : null) ?? + (respObj && typeof respObj.model === "string" ? respObj.model : null) ?? + (respObj && typeof respObj.modelVersion === "string" ? respObj.modelVersion : null) ?? + null; + const messages = countMessages(body); + const { tokensIn, tokensOut } = extractUsage(resp); + const streamed = isSseRequest(req); + const mappedTo = + req.mappedModel ?? + req.requestHeaders["x-omniroute-mapped"] ?? + req.requestHeaders["X-Omniroute-Mapped"] ?? + null; + + return { + provider, + apiKind, + model, + messages, + tokensIn, + tokensOut, + streamed, + mappedTo, + costEstimateUsd: null, // cost table out of scope for F4 + }; +} diff --git a/src/mitm/inspector/sseMerger.ts b/src/mitm/inspector/sseMerger.ts new file mode 100644 index 0000000000..8a1d245603 --- /dev/null +++ b/src/mitm/inspector/sseMerger.ts @@ -0,0 +1,316 @@ +/** + * SSE merger — reconstructs complete LLM response from streaming SSE chunks. + * + * MIT — port from https://github.com/chouzz/llm-interceptor (merger.py) + * + * Detects API format by chunk shape (not URL — robust to URL rewrite) and + * rebuilds Anthropic / OpenAI / Gemini responses. Falls back to a raw event + * list when the format is unrecognised so the caller never crashes. + */ + +export type ApiFormat = "anthropic" | "openai" | "gemini" | "unknown"; + +export interface SseEvent { + event?: string; + data?: string; + // Parsed JSON payload when `data` was valid JSON. + json?: unknown; +} + +export interface MergedResponse { + format: ApiFormat; + message?: unknown; + raw?: SseEvent[]; +} + +function asRecord(value: unknown): Record | null { + if (value && typeof value === "object" && !Array.isArray(value)) { + return value as Record; + } + return null; +} + +/** + * Inspect chunk shapes to determine the upstream API. Matches on the first + * recognisable hint; returns `"unknown"` if none match. + */ +export function detectApiFormat(chunks: SseEvent[]): ApiFormat { + for (const c of chunks) { + const j = asRecord(c.json); + if (!j) continue; + if (j.type === "message_start" || j.type === "content_block_delta") return "anthropic"; + if (Array.isArray(j.choices)) { + const first = j.choices[0]; + if (first && typeof first === "object" && "delta" in first) return "openai"; + } + if (Array.isArray(j.candidates)) return "gemini"; + } + return "unknown"; +} + +/** + * Parse a raw SSE stream (the response body string captured by the proxy) + * into discrete events. Empty blocks and `[DONE]` terminators are skipped + * silently; malformed JSON payloads are kept as raw `data` (no `json`). + */ +export function parseSseStream(raw: string): SseEvent[] { + const events: SseEvent[] = []; + if (!raw) return events; + // SSE blocks separated by blank lines — accept both LF and CRLF. + for (const block of raw.split(/\r?\n\r?\n/)) { + if (!block.trim()) continue; + const ev: SseEvent = {}; + for (const line of block.split(/\r?\n/)) { + if (line.startsWith("event:")) { + ev.event = line.slice(6).trim(); + } else if (line.startsWith("data:")) { + ev.data = (ev.data ?? "") + line.slice(5).trim(); + } + } + if (ev.data === undefined) continue; + if (ev.data === "[DONE]") { + events.push(ev); + continue; + } + try { + ev.json = JSON.parse(ev.data); + } catch { + // keep raw data only + } + events.push(ev); + } + return events; +} + +interface AnthropicBlock { + type: string; + text?: string; + thinking?: string; + id?: string; + name?: string; + input?: unknown; +} + +/** + * Rebuild an Anthropic Messages API response from streaming events. + * Handles `text_delta`, `thinking_delta`, and `input_json_delta` deltas; + * applies `JSON.parse` (best-effort) on accumulated tool-use input. + */ +export function rebuildAnthropic(chunks: SseEvent[]): MergedResponse { + const blocks: AnthropicBlock[] = []; + let message: Record | null = null; + const inputJsonByIndex: Record = {}; + + for (const c of chunks) { + const j = asRecord(c.json); + if (!j) continue; + const t = j.type; + + if (t === "message_start") { + const m = asRecord(j.message); + message = m ? { ...m } : {}; + } else if (t === "content_block_start") { + const idx = typeof j.index === "number" ? j.index : blocks.length; + const cb = asRecord(j.content_block); + const block: AnthropicBlock = { type: "text" }; + if (cb) { + for (const [k, v] of Object.entries(cb)) (block as Record)[k] = v; + } + if (block.type === "text" && block.text === undefined) block.text = ""; + if (block.type === "thinking" && block.thinking === undefined) block.thinking = ""; + if (block.type === "tool_use" && block.input === undefined) block.input = {}; + blocks[idx] = block; + } else if (t === "content_block_delta") { + const idx = typeof j.index === "number" ? j.index : 0; + const d = asRecord(j.delta); + if (!d) continue; + // Ensure a block slot exists (some streams skip content_block_start). + const slot = blocks[idx] ?? (blocks[idx] = { type: "text", text: "" }); + const dType = d.type; + if (dType === "text_delta" && typeof d.text === "string") { + slot.text = (slot.text ?? "") + d.text; + } else if (dType === "thinking_delta" && typeof d.thinking === "string") { + slot.thinking = (slot.thinking ?? "") + d.thinking; + } else if (dType === "input_json_delta" && typeof d.partial_json === "string") { + inputJsonByIndex[idx] = (inputJsonByIndex[idx] ?? "") + d.partial_json; + } + } else if (t === "content_block_stop") { + const idx = typeof j.index === "number" ? j.index : 0; + const slot = blocks[idx]; + if (slot && slot.type === "tool_use" && inputJsonByIndex[idx]) { + try { + slot.input = JSON.parse(inputJsonByIndex[idx]); + } catch { + // keep accumulated string for forensic visibility + slot.input = inputJsonByIndex[idx]; + } + } + } else if (t === "message_delta") { + if (!message) message = {}; + const d = asRecord(j.delta); + if (d && typeof d.stop_reason === "string") { + message.stop_reason = d.stop_reason; + } + const usage = asRecord(j.usage); + if (usage) { + const prev = asRecord(message.usage) ?? {}; + message.usage = { ...prev, ...usage }; + } + } + } + + const filledBlocks = blocks.filter((b) => b !== undefined); + return { + format: "anthropic", + message: { ...(message ?? {}), content: filledBlocks }, + }; +} + +interface OpenAiToolCall { + index: number; + id?: string; + type?: string; + function: { name: string; arguments: string }; +} + +interface OpenAiChoice { + index: number; + message: { + role: string; + content: string; + tool_calls?: OpenAiToolCall[]; + refusal?: string; + }; + finish_reason: string | null; +} + +/** + * Rebuild an OpenAI Chat Completions response from streaming events. + * Accumulates content text and tool-call fragments per choice/index. + */ +export function rebuildOpenAI(chunks: SseEvent[]): MergedResponse { + const choicesByIdx: Record = {}; + let model: string | null = null; + let usage: unknown = null; + let id: string | null = null; + + for (const c of chunks) { + const j = asRecord(c.json); + if (!j) continue; + if (typeof j.model === "string") model = j.model; + if (typeof j.id === "string") id = j.id; + if (j.usage != null) usage = j.usage; + if (!Array.isArray(j.choices)) continue; + + for (const raw of j.choices) { + const ch = asRecord(raw); + if (!ch) continue; + const idx = typeof ch.index === "number" ? ch.index : 0; + const slot = (choicesByIdx[idx] ??= { + index: idx, + message: { role: "assistant", content: "" }, + finish_reason: null, + }); + const delta = asRecord(ch.delta) ?? {}; + if (typeof delta.role === "string") slot.message.role = delta.role; + if (typeof delta.content === "string") slot.message.content += delta.content; + if (typeof delta.refusal === "string") { + slot.message.refusal = (slot.message.refusal ?? "") + delta.refusal; + } + if (Array.isArray(delta.tool_calls)) { + slot.message.tool_calls ??= []; + for (const tcRaw of delta.tool_calls) { + const tc = asRecord(tcRaw); + if (!tc) continue; + const ti = typeof tc.index === "number" ? tc.index : 0; + const tcSlot = + slot.message.tool_calls[ti] ?? + (slot.message.tool_calls[ti] = { + index: ti, + function: { name: "", arguments: "" }, + }); + if (typeof tc.id === "string") tcSlot.id = tc.id; + if (typeof tc.type === "string") tcSlot.type = tc.type; + const fn = asRecord(tc.function); + if (fn) { + if (typeof fn.name === "string") tcSlot.function.name += fn.name; + if (typeof fn.arguments === "string") tcSlot.function.arguments += fn.arguments; + } + } + } + if (typeof ch.finish_reason === "string") slot.finish_reason = ch.finish_reason; + } + } + + return { + format: "openai", + message: { + id, + model, + choices: Object.values(choicesByIdx).sort((a, b) => a.index - b.index), + usage, + }, + }; +} + +/** + * Rebuild a Gemini `generateContent`-style response from streaming events. + * Concatenates all parts across emitted candidates into a single candidate. + */ +export function rebuildGemini(chunks: SseEvent[]): MergedResponse { + const parts: unknown[] = []; + let usageMetadata: unknown = null; + let finishReason: unknown = null; + let modelVersion: string | null = null; + + for (const c of chunks) { + const j = asRecord(c.json); + if (!j) continue; + if (j.usageMetadata != null) usageMetadata = j.usageMetadata; + if (typeof j.modelVersion === "string") modelVersion = j.modelVersion; + if (!Array.isArray(j.candidates)) continue; + for (const candRaw of j.candidates) { + const cand = asRecord(candRaw); + if (!cand) continue; + if (cand.finishReason != null) finishReason = cand.finishReason; + const content = asRecord(cand.content); + if (!content) continue; + const ps = content.parts; + if (Array.isArray(ps)) { + for (const p of ps) parts.push(p); + } + } + } + + return { + format: "gemini", + message: { + candidates: [ + { + content: { parts, role: "model" }, + ...(finishReason != null ? { finishReason } : {}), + }, + ], + ...(modelVersion ? { modelVersion } : {}), + ...(usageMetadata != null ? { usageMetadata } : {}), + }, + }; +} + +/** + * Merge an array of SSE events into a single rebuilt response. Returns + * `{ format: "unknown", raw }` (no throw) for unrecognised shapes. + */ +export function mergeStream(chunks: SseEvent[]): MergedResponse { + const format = detectApiFormat(chunks); + switch (format) { + case "anthropic": + return rebuildAnthropic(chunks); + case "openai": + return rebuildOpenAI(chunks); + case "gemini": + return rebuildGemini(chunks); + default: + return { format: "unknown", raw: chunks }; + } +} diff --git a/src/mitm/inspector/systemProxyConfig.ts b/src/mitm/inspector/systemProxyConfig.ts new file mode 100644 index 0000000000..2fbd3319e4 --- /dev/null +++ b/src/mitm/inspector/systemProxyConfig.ts @@ -0,0 +1,314 @@ +/** + * System-wide proxy configuration toggles. + * + * macOS: `networksetup -setwebproxy / -setsecurewebproxy` + * Linux: `gsettings set org.gnome.system.proxy. host/port` + mode + * Windows: `netsh winhttp set proxy ` + * + * Hard Rule #13: every shell invocation here uses `execFile` with an array of + * arguments (never a shell string), so runtime values cannot be interpreted + * as shell syntax. + * + * The returned `previousState` is JSON-serialisable so callers can persist it + * (DB row) and pass it back to `revert()` later — including across process + * restarts (the operator-facing "Restore system proxy" button). + */ + +import { execFile, type ExecFileOptions } from "node:child_process"; +import os from "node:os"; +import { sanitizeErrorMessage } from "@omniroute/open-sse/utils/error"; + +export type Platform = "linux" | "macos" | "windows"; + +export interface MacOsPreviousState { + platform: "macos"; + service: string; + http: { enabled: boolean; host: string; port: string }; + https: { enabled: boolean; host: string; port: string }; +} + +export interface LinuxPreviousState { + platform: "linux"; + gnomeMode: string; + httpHost: string; + httpPort: string; + httpsHost: string; + httpsPort: string; +} + +export interface WindowsPreviousState { + platform: "windows"; + netshOutput: string; +} + +export type PreviousState = + | MacOsPreviousState + | LinuxPreviousState + | WindowsPreviousState; + +export interface ApplyResult { + platform: Platform; + previousState: PreviousState; +} + +// Injection seam for tests. Default implementation wraps node:child_process +// `execFile` so call-sites use array args (Hard Rule #13). +export type ExecFileFn = ( + file: string, + args: string[], + options?: ExecFileOptions +) => Promise<{ stdout: string; stderr: string }>; + +let execImpl: ExecFileFn = defaultExec; + +function defaultExec( + file: string, + args: string[], + options: ExecFileOptions = {} +): Promise<{ stdout: string; stderr: string }> { + return new Promise((resolve, reject) => { + execFile(file, args, options, (err, stdout, stderr) => { + if (err) { + reject(err); + return; + } + resolve({ + stdout: stdout?.toString() ?? "", + stderr: stderr?.toString() ?? "", + }); + }); + }); +} + +/** + * Replace the underlying `execFile` runner (for tests). + * Returns a `restore()` function that puts the default back. + */ +export function __setExec(fn: ExecFileFn): () => void { + const prev = execImpl; + execImpl = fn; + return () => { + execImpl = prev; + }; +} + +function detectPlatform(): Platform { + const p = os.platform(); + if (p === "darwin") return "macos"; + if (p === "win32") return "windows"; + return "linux"; +} + +// ──────────────────────────────────────────────────────────────────────────── +// macOS — networksetup +// ──────────────────────────────────────────────────────────────────────────── + +const MAC_DEFAULT_SERVICE = "Wi-Fi"; + +interface NetworksetupRead { + enabled: boolean; + host: string; + port: string; +} + +function parseNetworksetupGet(output: string): NetworksetupRead { + // Example output: + // Enabled: Yes + // Server: 192.168.1.1 + // Port: 3128 + // Authenticated Proxy Enabled: 0 + const lines = output.split(/\r?\n/); + let enabled = false; + let host = ""; + let port = ""; + for (const line of lines) { + const m = line.match(/^(\S[^:]*):\s*(.*)$/); + if (!m) continue; + const key = m[1].trim().toLowerCase(); + const val = m[2].trim(); + if (key === "enabled") enabled = /yes/i.test(val); + else if (key === "server") host = val; + else if (key === "port") port = val; + } + return { enabled, host, port }; +} + +async function macosApply(port: number): Promise { + const service = MAC_DEFAULT_SERVICE; + const httpGet = await execImpl("networksetup", ["-getwebproxy", service]); + const httpsGet = await execImpl("networksetup", ["-getsecurewebproxy", service]); + const previousState: MacOsPreviousState = { + platform: "macos", + service, + http: parseNetworksetupGet(httpGet.stdout), + https: parseNetworksetupGet(httpsGet.stdout), + }; + + await execImpl("networksetup", ["-setwebproxy", service, "127.0.0.1", String(port)]); + await execImpl("networksetup", ["-setsecurewebproxy", service, "127.0.0.1", String(port)]); + return previousState; +} + +async function macosRevert(state: MacOsPreviousState): Promise { + const service = state.service; + if (state.http.enabled && state.http.host && state.http.port) { + await execImpl("networksetup", [ + "-setwebproxy", + service, + state.http.host, + state.http.port, + ]); + } else { + await execImpl("networksetup", ["-setwebproxystate", service, "off"]); + } + if (state.https.enabled && state.https.host && state.https.port) { + await execImpl("networksetup", [ + "-setsecurewebproxy", + service, + state.https.host, + state.https.port, + ]); + } else { + await execImpl("networksetup", ["-setsecurewebproxystate", service, "off"]); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// Linux — gsettings (GNOME); systems without gsettings are unsupported here. +// ──────────────────────────────────────────────────────────────────────────── + +async function readGsetting(key: string): Promise { + try { + const { stdout } = await execImpl("gsettings", ["get", "org.gnome.system.proxy", key]); + return stdout.trim(); + } catch { + return ""; + } +} + +async function readGsubsetting( + scheme: string, + key: string +): Promise { + try { + const { stdout } = await execImpl("gsettings", [ + "get", + `org.gnome.system.proxy.${scheme}`, + key, + ]); + return stdout.trim(); + } catch { + return ""; + } +} + +async function linuxApply(port: number): Promise { + const previousState: LinuxPreviousState = { + platform: "linux", + gnomeMode: await readGsetting("mode"), + httpHost: await readGsubsetting("http", "host"), + httpPort: await readGsubsetting("http", "port"), + httpsHost: await readGsubsetting("https", "host"), + httpsPort: await readGsubsetting("https", "port"), + }; + + const portStr = String(port); + await execImpl("gsettings", ["set", "org.gnome.system.proxy", "mode", "manual"]); + await execImpl("gsettings", ["set", "org.gnome.system.proxy.http", "host", "127.0.0.1"]); + await execImpl("gsettings", ["set", "org.gnome.system.proxy.http", "port", portStr]); + await execImpl("gsettings", ["set", "org.gnome.system.proxy.https", "host", "127.0.0.1"]); + await execImpl("gsettings", ["set", "org.gnome.system.proxy.https", "port", portStr]); + return previousState; +} + +async function linuxRevert(state: LinuxPreviousState): Promise { + const mode = state.gnomeMode || "'none'"; + await execImpl("gsettings", ["set", "org.gnome.system.proxy", "mode", mode]); + if (state.httpHost) { + await execImpl("gsettings", ["set", "org.gnome.system.proxy.http", "host", state.httpHost]); + } + if (state.httpPort) { + await execImpl("gsettings", ["set", "org.gnome.system.proxy.http", "port", state.httpPort]); + } + if (state.httpsHost) { + await execImpl("gsettings", [ + "set", + "org.gnome.system.proxy.https", + "host", + state.httpsHost, + ]); + } + if (state.httpsPort) { + await execImpl("gsettings", [ + "set", + "org.gnome.system.proxy.https", + "port", + state.httpsPort, + ]); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// Windows — netsh winhttp +// ──────────────────────────────────────────────────────────────────────────── + +async function windowsApply(port: number): Promise { + const showRes = await execImpl("netsh", ["winhttp", "show", "proxy"]); + const previousState: WindowsPreviousState = { + platform: "windows", + netshOutput: showRes.stdout, + }; + const proxyArg = `127.0.0.1:${String(port)}`; + await execImpl("netsh", ["winhttp", "set", "proxy", proxyArg]); + return previousState; +} + +async function windowsRevert(_state: WindowsPreviousState): Promise { + // netsh has no idempotent restore; the safe default is "reset". + // The previousState is preserved so the UI can show the operator what was + // configured before, but actual reapply of obscure netsh state is out of + // scope. + await execImpl("netsh", ["winhttp", "reset", "proxy"]); +} + +// ──────────────────────────────────────────────────────────────────────────── +// Public API +// ──────────────────────────────────────────────────────────────────────────── + +/** + * Apply OmniRoute as the system-wide HTTP/HTTPS proxy at `127.0.0.1:`. + * Captures and returns the prior configuration so callers can revert later. + * + * Throws a sanitized `Error` if the underlying command fails (no stack/path + * leakage — see Hard Rule #12). + */ +export async function apply(port: number): Promise { + const platform = detectPlatform(); + try { + let previousState: PreviousState; + if (platform === "macos") previousState = await macosApply(port); + else if (platform === "windows") previousState = await windowsApply(port); + else previousState = await linuxApply(port); + return { platform, previousState }; + } catch (err) { + throw new Error(sanitizeErrorMessage(err) || "system proxy apply failed"); + } +} + +/** + * Restore the prior configuration captured by `apply()`. No-op if the + * `previousState` payload does not match a known platform. + */ +export async function revert(previousState: PreviousState | unknown): Promise { + if (!previousState || typeof previousState !== "object") return; + const state = previousState as Record; + const platform = state.platform; + try { + if (platform === "macos") await macosRevert(state as unknown as MacOsPreviousState); + else if (platform === "linux") await linuxRevert(state as unknown as LinuxPreviousState); + else if (platform === "windows") + await windowsRevert(state as unknown as WindowsPreviousState); + } catch (err) { + throw new Error(sanitizeErrorMessage(err) || "system proxy revert failed"); + } +} diff --git a/tests/unit/inspector-buffer.test.ts b/tests/unit/inspector-buffer.test.ts new file mode 100644 index 0000000000..455715a95c --- /dev/null +++ b/tests/unit/inspector-buffer.test.ts @@ -0,0 +1,204 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { TrafficBuffer } from "../../src/mitm/inspector/buffer.ts"; +import type { + InterceptedRequest, + WsEvent, +} from "../../src/mitm/inspector/types.ts"; + +function makeReq(overrides: Partial = {}): InterceptedRequest { + return { + id: overrides.id ?? `id-${Math.random().toString(36).slice(2, 10)}`, + source: "agent-bridge", + timestamp: new Date().toISOString(), + method: "POST", + host: "api.openai.com", + path: "/v1/chat/completions", + requestHeaders: {}, + requestBody: JSON.stringify({ + messages: [ + { role: "system", content: "You are an assistant." }, + { role: "user", content: "Hi" }, + ], + }), + requestSize: 0, + responseHeaders: {}, + responseBody: null, + responseSize: 0, + status: 200, + ...overrides, + }; +} + +test("push appends entries and auto-applies detectedKind=llm", () => { + const buf = new TrafficBuffer(10); + const r = makeReq({ id: "r1" }); + buf.push(r); + const got = buf.get("r1"); + assert.ok(got); + assert.equal(got.detectedKind, "llm"); +}); + +test("push auto-computes contextKey from system prompt", () => { + const buf = new TrafficBuffer(10); + const r = makeReq({ id: "r1" }); + buf.push(r); + const got = buf.get("r1"); + assert.ok(got); + assert.ok(got.contextKey); + assert.match(got.contextKey!, /^[0-9a-f]{12}$/); +}); + +test("push does not override an existing contextKey", () => { + const buf = new TrafficBuffer(10); + const r = makeReq({ id: "r1", contextKey: "preexisting1" }); + buf.push(r); + const got = buf.get("r1"); + assert.equal(got!.contextKey, "preexisting1"); +}); + +test("push truncates large requestBody with marker", () => { + // 1 KiB so cap is hit reliably; override env via fresh buffer w/ explicit byte cap + const big = "a".repeat(3000); + const buf = new TrafficBuffer(10, 1024); // 1 KiB max body + buf.push(makeReq({ id: "r1", requestBody: big })); + const got = buf.get("r1"); + assert.ok(got); + assert.ok(got.requestBody!.length > 1024); // marker increases length slightly + assert.match(got.requestBody!, /truncated for performance/); +}); + +test("push rotates oldest when over maxSize", () => { + const buf = new TrafficBuffer(3, 1024); + buf.push(makeReq({ id: "a" })); + buf.push(makeReq({ id: "b" })); + buf.push(makeReq({ id: "c" })); + buf.push(makeReq({ id: "d" })); + assert.equal(buf.size(), 3); + assert.equal(buf.get("a"), null); + assert.ok(buf.get("d")); +}); + +test("update replaces existing entry by id and broadcasts update", () => { + const buf = new TrafficBuffer(5); + buf.push(makeReq({ id: "r1" })); + const events: WsEvent[] = []; + const off = buf.subscribe((e) => events.push(e)); + // initial snapshot received + buf.update("r1", makeReq({ id: "r1", status: 500, responseBody: "err" })); + off(); + const got = buf.get("r1"); + assert.equal(got!.status, 500); + const updates = events.filter((e) => e.type === "update"); + assert.equal(updates.length, 1); +}); + +test("update is a no-op when id is unknown", () => { + const buf = new TrafficBuffer(5); + buf.update("missing", makeReq({ id: "missing" })); + assert.equal(buf.size(), 0); +}); + +test("list applies filters by source, host, status, profile, agent, sessionId", () => { + const buf = new TrafficBuffer(20); + buf.push(makeReq({ id: "a", host: "api.openai.com", source: "agent-bridge", agent: "codex" })); + buf.push( + makeReq({ + id: "b", + host: "random.example.com", + source: "http-proxy", + requestBody: JSON.stringify({ name: "not-llm" }), + detectedKind: "app", + }) + ); + buf.push( + makeReq({ + id: "c", + host: "api.anthropic.com", + source: "custom-host", + status: 500, + sessionId: "00000000-0000-0000-0000-000000000000", + }) + ); + + assert.equal(buf.list({ profile: "llm" }).length, 2); + assert.equal(buf.list({ source: "http-proxy" }).length, 1); + assert.equal(buf.list({ host: "api.openai.com" }).length, 1); + assert.equal(buf.list({ status: "5xx" }).length, 1); + assert.equal(buf.list({ agent: "codex" }).length, 1); + assert.equal( + buf.list({ sessionId: "00000000-0000-0000-0000-000000000000" }).length, + 1 + ); + assert.equal(buf.list({ profile: "custom" }).length, 1); + assert.equal(buf.list({ profile: "all" }).length, 3); +}); + +test("clear empties the buffer and broadcasts a clear event", () => { + const buf = new TrafficBuffer(5); + buf.push(makeReq({ id: "a" })); + buf.push(makeReq({ id: "b" })); + const events: WsEvent[] = []; + const off = buf.subscribe((e) => events.push(e)); + buf.clear(); + off(); + assert.equal(buf.size(), 0); + assert.ok(events.some((e) => e.type === "clear")); +}); + +test("subscribe immediately delivers a snapshot of the current buffer", () => { + const buf = new TrafficBuffer(5); + buf.push(makeReq({ id: "a" })); + buf.push(makeReq({ id: "b" })); + const events: WsEvent[] = []; + const off = buf.subscribe((e) => events.push(e)); + off(); + assert.equal(events.length, 1); + assert.equal(events[0].type, "snapshot"); + if (events[0].type === "snapshot") { + assert.equal(events[0].data.length, 2); + } +}); + +test("subscribe returns an unsubscribe function", () => { + const buf = new TrafficBuffer(5); + const fn = (_e: WsEvent): void => {}; + const off = buf.subscribe(fn); + assert.equal(buf.subscriberCount(), 1); + off(); + assert.equal(buf.subscriberCount(), 0); +}); + +test("broadcast survives a throwing subscriber", () => { + const buf = new TrafficBuffer(5); + buf.subscribe(() => { + throw new Error("subscriber crash"); + }); + let okCount = 0; + buf.subscribe(() => { + okCount += 1; + }); + buf.push(makeReq({ id: "a" })); + // 1 snapshot from second subscribe + 1 new event from push + assert.ok(okCount >= 1); +}); + +test("broadcasts new event with the pushed request", () => { + const buf = new TrafficBuffer(5); + const events: WsEvent[] = []; + buf.subscribe((e) => events.push(e)); + buf.push(makeReq({ id: "evt" })); + const news = events.filter((e) => e.type === "new"); + assert.equal(news.length, 1); + if (news[0].type === "new") { + assert.equal(news[0].data.id, "evt"); + } +}); + +test("body cap applies to responseBody as well", () => { + const buf = new TrafficBuffer(5, 100); + const r = makeReq({ id: "r", responseBody: "x".repeat(500) }); + buf.push(r); + const got = buf.get("r"); + assert.match(got!.responseBody!, /truncated for performance/); +}); diff --git a/tests/unit/inspector-conversation-normalizer.test.ts b/tests/unit/inspector-conversation-normalizer.test.ts new file mode 100644 index 0000000000..51ed564a52 --- /dev/null +++ b/tests/unit/inspector-conversation-normalizer.test.ts @@ -0,0 +1,189 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { normalizeConversation } from "../../src/mitm/inspector/conversationNormalizer.ts"; +import type { InterceptedRequest } from "../../src/mitm/inspector/types.ts"; + +function makeReq(overrides: Partial = {}): InterceptedRequest { + return { + id: "test-id", + source: "agent-bridge", + timestamp: new Date().toISOString(), + method: "POST", + host: "api.openai.com", + path: "/v1/chat/completions", + requestHeaders: {}, + requestBody: null, + requestSize: 0, + responseHeaders: {}, + responseBody: null, + responseSize: 0, + status: 200, + detectedKind: "llm", + ...overrides, + }; +} + +test("returns null for non-llm requests", () => { + const req = makeReq({ detectedKind: "app" }); + assert.equal(normalizeConversation(req), null); +}); + +test("returns null when request body cannot yield turns", () => { + const req = makeReq({ requestBody: JSON.stringify({ foo: "bar" }) }); + assert.equal(normalizeConversation(req), null); +}); + +test("normalizes OpenAI request with system + user messages", () => { + const req = makeReq({ + requestBody: JSON.stringify({ + messages: [ + { role: "system", content: "You are helpful." }, + { role: "user", content: "Hello!" }, + ], + }), + }); + const conv = normalizeConversation(req); + assert.ok(conv); + assert.equal(conv.request.length, 2); + assert.equal(conv.request[0].role, "system"); + assert.equal(conv.request[0].blocks[0].type, "text"); + assert.equal(conv.request[1].role, "user"); +}); + +test("normalizes OpenAI assistant tool_calls into tool_use blocks", () => { + const req = makeReq({ + requestBody: JSON.stringify({ + messages: [ + { + role: "assistant", + content: null, + tool_calls: [ + { + id: "call-1", + function: { name: "get_weather", arguments: '{"city":"SP"}' }, + }, + ], + }, + ], + }), + }); + const conv = normalizeConversation(req); + assert.ok(conv); + const blocks = conv.request[0].blocks; + assert.equal(blocks.length, 1); + assert.equal(blocks[0].type, "tool_use"); + const tu = blocks[0] as { type: "tool_use"; id: string; name: string; input: unknown }; + assert.equal(tu.id, "call-1"); + assert.equal(tu.name, "get_weather"); + assert.deepEqual(tu.input, { city: "SP" }); +}); + +test("normalizes OpenAI tool role into tool_result", () => { + const req = makeReq({ + requestBody: JSON.stringify({ + messages: [ + { role: "tool", tool_call_id: "call-1", content: "sunny" }, + ], + }), + }); + const conv = normalizeConversation(req); + assert.ok(conv); + assert.equal(conv.request[0].role, "tool"); + const blk = conv.request[0].blocks[0] as { type: "tool_result"; tool_use_id: string }; + assert.equal(blk.type, "tool_result"); + assert.equal(blk.tool_use_id, "call-1"); +}); + +test("normalizes Anthropic request with top-level system + tool_use response", () => { + const req = makeReq({ + host: "api.anthropic.com", + path: "/v1/messages", + requestBody: JSON.stringify({ + system: "Be terse.", + messages: [{ role: "user", content: "hi" }], + }), + responseBody: JSON.stringify({ + content: [ + { type: "text", text: "Hello." }, + { type: "tool_use", id: "tu1", name: "lookup", input: { q: "x" } }, + ], + }), + }); + const conv = normalizeConversation(req); + assert.ok(conv); + assert.equal(conv.request[0].role, "system"); + assert.equal(conv.response.length, 1); + assert.equal(conv.response[0].role, "assistant"); + assert.equal(conv.response[0].blocks.length, 2); + assert.equal(conv.response[0].blocks[0].type, "text"); + assert.equal(conv.response[0].blocks[1].type, "tool_use"); +}); + +test("normalizes Gemini request contents + functionCall response", () => { + const req = makeReq({ + host: "generativelanguage.googleapis.com", + path: "/v1beta/models/gemini-pro:generateContent", + requestBody: JSON.stringify({ + systemInstruction: { parts: [{ text: "sys" }] }, + contents: [ + { role: "user", parts: [{ text: "hi" }] }, + { + role: "model", + parts: [{ functionCall: { name: "fn", args: { a: 1 } } }], + }, + ], + }), + responseBody: JSON.stringify({ + candidates: [ + { + content: { + parts: [{ text: "Hello from gemini" }], + }, + }, + ], + }), + }); + const conv = normalizeConversation(req); + assert.ok(conv); + assert.equal(conv.request[0].role, "system"); + // user + assistant (model -> assistant) + assert.equal(conv.request[1].role, "user"); + assert.equal(conv.request[2].role, "assistant"); + const tu = conv.request[2].blocks[0] as { type: string; name: string }; + assert.equal(tu.type, "tool_use"); + assert.equal(tu.name, "fn"); + assert.equal(conv.response[0].role, "assistant"); + assert.equal((conv.response[0].blocks[0] as { text: string }).text, "Hello from gemini"); +}); + +test("propagates contextKey from request", () => { + const req = makeReq({ + contextKey: "abc123def456", + requestBody: JSON.stringify({ + messages: [{ role: "user", content: "hi" }], + }), + }); + const conv = normalizeConversation(req); + assert.ok(conv); + assert.equal(conv.contextKey, "abc123def456"); +}); + +test("parses SSE response to extract OpenAI delta", () => { + const sse = [ + `data: {"choices":[{"delta":{"role":"assistant","content":"Hello"}}]}`, + "", + `data: {"choices":[{"delta":{"content":" world"}}]}`, + "", + "data: [DONE]", + "", + ].join("\n"); + const req = makeReq({ + requestBody: JSON.stringify({ messages: [{ role: "user", content: "hi" }] }), + responseHeaders: { "content-type": "text/event-stream" }, + responseBody: sse, + }); + const conv = normalizeConversation(req); + assert.ok(conv); + assert.ok(conv.response.length >= 1); + assert.equal(conv.response[0].role, "assistant"); +}); diff --git a/tests/unit/inspector-har-export.test.ts b/tests/unit/inspector-har-export.test.ts new file mode 100644 index 0000000000..50c6c5ec82 --- /dev/null +++ b/tests/unit/inspector-har-export.test.ts @@ -0,0 +1,128 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { toHar } from "../../src/lib/inspector/harExport.ts"; +import type { InterceptedRequest } from "../../src/mitm/inspector/types.ts"; + +function makeReq(overrides: Partial = {}): InterceptedRequest { + return { + id: "00000000-0000-4000-8000-000000000001", + source: "agent-bridge", + timestamp: "2026-05-27T12:00:00.000Z", + method: "POST", + host: "api.openai.com", + path: "/v1/chat/completions", + requestHeaders: { "content-type": "application/json" }, + requestBody: '{"model":"gpt-4"}', + requestSize: 17, + responseHeaders: { "content-type": "application/json" }, + responseBody: '{"ok":true}', + responseSize: 11, + status: 200, + totalLatencyMs: 200, + upstreamLatencyMs: 150, + ...overrides, + }; +} + +test("produces HAR v1.2 with creator + entries", () => { + const har = toHar([makeReq()]); + assert.equal(har.log.version, "1.2"); + assert.ok(har.log.creator.name.includes("OmniRoute")); + assert.equal(har.log.entries.length, 1); +}); + +test("entry fields match the spec", () => { + const har = toHar([makeReq()]); + const e = har.log.entries[0]; + assert.equal(e.startedDateTime, "2026-05-27T12:00:00.000Z"); + assert.equal(e.time, 200); + assert.equal(e.request.method, "POST"); + assert.equal(e.request.url, "https://api.openai.com/v1/chat/completions"); + assert.equal(e.request.httpVersion, "HTTP/1.1"); + assert.equal(e.request.bodySize, 17); + assert.ok(e.request.postData); + assert.equal(e.request.postData?.mimeType, "application/json"); + assert.equal(e.response.status, 200); + assert.equal(e.response.content.size, 11); + assert.equal(e.response.content.text, '{"ok":true}'); + assert.equal(e.timings.send, 0); + assert.equal(e.timings.wait, 150); + assert.equal(e.timings.receive, 50); +}); + +test("Bearer tokens in headers are masked", () => { + const req = makeReq({ + requestHeaders: { + "content-type": "application/json", + authorization: "Bearer sk-supersecretvalueabc1234567890XYZ", + }, + }); + const har = toHar([req]); + const authHeader = har.log.entries[0].request.headers.find( + (h) => h.name === "authorization" + ); + assert.ok(authHeader); + // Either Bearer regex (authorization:\sBearer prefix) or sk-/long-token regex must mask the value + assert.ok(!authHeader.value.includes("supersecretvalueabc1234567890XYZ")); + assert.ok(authHeader.value.includes("…") || authHeader.value.includes("***")); +}); + +test("sk- keys in bodies are masked", () => { + const req = makeReq({ + requestBody: '{"key":"sk-abcdef1234567890ABCDEF"}', + }); + const har = toHar([req]); + const body = har.log.entries[0].request.postData?.text ?? ""; + assert.ok(!body.includes("sk-abcdef1234567890ABCDEF")); + assert.match(body, /sk-abc/); +}); + +test("preserves _source custom property", () => { + const har = toHar([ + makeReq({ source: "http-proxy" }), + makeReq({ id: "00000000-0000-4000-8000-000000000002", source: "system-proxy" }), + ]); + assert.equal(har.log.entries[0]._source, "http-proxy"); + assert.equal(har.log.entries[1]._source, "system-proxy"); +}); + +test("preserves _detectedKind / _contextKey / _agent / _sessionId / _note", () => { + const har = toHar([ + makeReq({ + agent: "claude", + detectedKind: "llm", + contextKey: "abc123", + sessionId: "00000000-0000-4000-8000-000000000099", + note: "TLS tunnel", + }), + ]); + const e = har.log.entries[0]; + assert.equal(e._agent, "claude"); + assert.equal(e._detectedKind, "llm"); + assert.equal(e._contextKey, "abc123"); + assert.equal(e._sessionId, "00000000-0000-4000-8000-000000000099"); + assert.equal(e._note, "TLS tunnel"); + assert.equal(e._omniRouteId, "00000000-0000-4000-8000-000000000001"); +}); + +test("handles in-flight / error status without throwing", () => { + const har = toHar([ + makeReq({ status: "in-flight", responseBody: null }), + makeReq({ id: "x", status: "error", responseBody: null, error: "boom" }), + ]); + assert.equal(har.log.entries[0].response.status, 0); + assert.equal(har.log.entries[0].response.statusText, "in-flight"); + assert.equal(har.log.entries[1].response.statusText, "error"); +}); + +test("CONNECT-style path renders pseudo-URL", () => { + const har = toHar([ + makeReq({ + method: "CONNECT", + host: "api.example.com", + path: ":443", + responseBody: null, + }), + ]); + assert.equal(har.log.entries[0].request.url, "https://api.example.com:443"); +}); diff --git a/tests/unit/inspector-http-proxy.test.ts b/tests/unit/inspector-http-proxy.test.ts new file mode 100644 index 0000000000..bc39b3697b --- /dev/null +++ b/tests/unit/inspector-http-proxy.test.ts @@ -0,0 +1,142 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; +import net from "node:net"; +import { startHttpProxyServer } from "../../src/mitm/inspector/httpProxyServer.ts"; +import { globalTrafficBuffer } from "../../src/mitm/inspector/buffer.ts"; + +async function withUpstream( + handler: (req: http.IncomingMessage, res: http.ServerResponse) => void +): Promise<{ port: number; close: () => Promise }> { + const server = http.createServer(handler); + await new Promise((resolve) => server.listen(0, "127.0.0.1", () => resolve())); + const addr = server.address(); + const port = typeof addr === "object" && addr ? addr.port : 0; + return { + port, + close: () => new Promise((res) => server.close(() => res())), + }; +} + +async function withTcpServer(): Promise<{ port: number; close: () => Promise }> { + const server = net.createServer((socket) => { + socket.on("data", () => { + socket.end("ok"); + }); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", () => resolve())); + const addr = server.address(); + const port = typeof addr === "object" && addr ? addr.port : 0; + return { + port, + close: () => new Promise((res) => server.close(() => res())), + }; +} + +function sendThroughProxy( + proxyPort: number, + upstreamPort: number, + method = "GET" +): Promise<{ status: number; body: string }> { + return new Promise((resolve, reject) => { + const req = http.request( + { + host: "127.0.0.1", + port: proxyPort, + method, + path: `http://127.0.0.1:${upstreamPort}/test`, + headers: { host: `127.0.0.1:${upstreamPort}` }, + }, + (res) => { + const chunks: Buffer[] = []; + res.on("data", (c) => chunks.push(c)); + res.on("end", () => + resolve({ status: res.statusCode ?? 0, body: Buffer.concat(chunks).toString("utf8") }) + ); + } + ); + req.once("error", reject); + req.end(); + }); +} + +function sendConnect(proxyPort: number, target: string): Promise { + return new Promise((resolve, reject) => { + const socket = net.connect(proxyPort, "127.0.0.1"); + socket.once("error", reject); + socket.once("connect", () => { + socket.write(`CONNECT ${target} HTTP/1.1\r\nHost: ${target}\r\n\r\n`); + }); + socket.once("data", (chunk) => { + const line = chunk.toString("utf8").split("\r\n")[0]; + const m = line.match(/HTTP\/1\.1\s+(\d+)/); + socket.end(); + resolve(m ? Number(m[1]) : 0); + }); + }); +} + +test("HTTP direct passes through and records buffer entry", async () => { + globalTrafficBuffer.clear(); + const upstream = await withUpstream((_req, res) => { + res.writeHead(200, { "content-type": "text/plain" }); + res.end("hello"); + }); + const proxy = await startHttpProxyServer(0); + try { + const sizeBefore = globalTrafficBuffer.size(); + const { status, body } = await sendThroughProxy(proxy.port, upstream.port); + assert.equal(status, 200); + assert.equal(body, "hello"); + // give buffer.update a tick (it runs inside async path) + await new Promise((r) => setTimeout(r, 30)); + assert.ok(globalTrafficBuffer.size() > sizeBefore); + const entry = globalTrafficBuffer.list().at(-1); + assert.ok(entry); + assert.equal(entry.source, "http-proxy"); + assert.equal(entry.method, "GET"); + assert.equal(entry.status, 200); + assert.match(entry.responseBody ?? "", /hello/); + } finally { + await proxy.stop(); + await upstream.close(); + } +}); + +test("CONNECT tunnel returns 200 and records metadata-only entry", async () => { + globalTrafficBuffer.clear(); + const tcp = await withTcpServer(); + const proxy = await startHttpProxyServer(0); + try { + const sizeBefore = globalTrafficBuffer.size(); + const status = await sendConnect(proxy.port, `127.0.0.1:${tcp.port}`); + assert.equal(status, 200); + await new Promise((r) => setTimeout(r, 30)); + assert.ok(globalTrafficBuffer.size() > sizeBefore); + const entry = globalTrafficBuffer.list().at(-1); + assert.ok(entry); + assert.equal(entry.method, "CONNECT"); + assert.equal(entry.source, "http-proxy"); + assert.equal(entry.responseBody, null); + assert.match(entry.note ?? "", /TLS tunnel/); + } finally { + await proxy.stop(); + await tcp.close(); + } +}); + +test("EADDRINUSE rejects with code", async () => { + const first = await startHttpProxyServer(0); + try { + await assert.rejects( + () => startHttpProxyServer(first.port), + (err: NodeJS.ErrnoException) => { + assert.ok(err); + assert.equal(err.code, "EADDRINUSE"); + return true; + } + ); + } finally { + await first.stop(); + } +}); diff --git a/tests/unit/inspector-llm-metadata.test.ts b/tests/unit/inspector-llm-metadata.test.ts new file mode 100644 index 0000000000..babb1f14c5 --- /dev/null +++ b/tests/unit/inspector-llm-metadata.test.ts @@ -0,0 +1,153 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { extractLlmMetadata } from "../../src/mitm/inspector/llmMetadataExtractor.ts"; +import type { InterceptedRequest } from "../../src/mitm/inspector/types.ts"; + +function makeReq(overrides: Partial = {}): InterceptedRequest { + return { + id: "test", + source: "agent-bridge", + timestamp: new Date().toISOString(), + method: "POST", + host: "api.openai.com", + path: "/v1/chat/completions", + requestHeaders: {}, + requestBody: null, + requestSize: 0, + responseHeaders: {}, + responseBody: null, + responseSize: 0, + status: 200, + detectedKind: "llm", + ...overrides, + }; +} + +test("returns null for non-llm requests", () => { + const req = makeReq({ detectedKind: "app" }); + assert.equal(extractLlmMetadata(req), null); +}); + +test("infers provider=openai from host", () => { + const req = makeReq({ + requestBody: JSON.stringify({ model: "gpt-4", messages: [] }), + }); + const meta = extractLlmMetadata(req); + assert.ok(meta); + assert.equal(meta.provider, "openai"); + assert.equal(meta.apiKind, "chat.completions"); + assert.equal(meta.model, "gpt-4"); +}); + +test("infers provider=anthropic + apiKind=messages", () => { + const req = makeReq({ + host: "api.anthropic.com", + path: "/v1/messages", + requestBody: JSON.stringify({ model: "claude-3", messages: [{}, {}] }), + }); + const meta = extractLlmMetadata(req); + assert.ok(meta); + assert.equal(meta.provider, "anthropic"); + assert.equal(meta.apiKind, "messages"); + assert.equal(meta.messages, 2); +}); + +test("infers provider=gemini", () => { + const req = makeReq({ + host: "generativelanguage.googleapis.com", + path: "/v1beta/models/gemini-pro:generateContent", + requestBody: JSON.stringify({ contents: [{}, {}, {}] }), + }); + const meta = extractLlmMetadata(req); + assert.ok(meta); + assert.equal(meta.provider, "gemini"); + assert.equal(meta.messages, 3); +}); + +test("extracts model from response body when missing in request", () => { + const req = makeReq({ + requestBody: JSON.stringify({ messages: [] }), + responseBody: JSON.stringify({ model: "gpt-4-turbo", choices: [] }), + }); + const meta = extractLlmMetadata(req); + assert.ok(meta); + assert.equal(meta.model, "gpt-4-turbo"); +}); + +test("extracts tokensIn/tokensOut from prompt_tokens/completion_tokens", () => { + const req = makeReq({ + requestBody: JSON.stringify({ model: "gpt-4", messages: [] }), + responseBody: JSON.stringify({ + usage: { prompt_tokens: 10, completion_tokens: 25 }, + }), + }); + const meta = extractLlmMetadata(req); + assert.ok(meta); + assert.equal(meta.tokensIn, 10); + assert.equal(meta.tokensOut, 25); +}); + +test("extracts tokensIn/tokensOut from input_tokens/output_tokens (Anthropic)", () => { + const req = makeReq({ + host: "api.anthropic.com", + path: "/v1/messages", + requestBody: JSON.stringify({ model: "claude-3", messages: [] }), + responseBody: JSON.stringify({ + usage: { input_tokens: 50, output_tokens: 100 }, + }), + }); + const meta = extractLlmMetadata(req); + assert.ok(meta); + assert.equal(meta.tokensIn, 50); + assert.equal(meta.tokensOut, 100); +}); + +test("extracts tokens from Gemini usageMetadata", () => { + const req = makeReq({ + host: "generativelanguage.googleapis.com", + path: "/v1beta/models/gemini-pro:generateContent", + requestBody: JSON.stringify({ contents: [] }), + responseBody: JSON.stringify({ + usageMetadata: { promptTokenCount: 7, candidatesTokenCount: 14 }, + }), + }); + const meta = extractLlmMetadata(req); + assert.ok(meta); + assert.equal(meta.tokensIn, 7); + assert.equal(meta.tokensOut, 14); +}); + +test("flags streamed=true on SSE content-type", () => { + const req = makeReq({ + requestBody: JSON.stringify({ model: "gpt-4", messages: [] }), + responseHeaders: { "content-type": "text/event-stream" }, + responseBody: "data: {}\n", + }); + const meta = extractLlmMetadata(req); + assert.ok(meta); + assert.equal(meta.streamed, true); +}); + +test("returns null fields when no info available", () => { + const req = makeReq({ + host: "unknown.example.com", + path: "/v1/messages", + requestBody: JSON.stringify({ messages: [] }), + }); + const meta = extractLlmMetadata(req); + assert.ok(meta); + assert.equal(meta.provider, null); + assert.equal(meta.tokensIn, null); + assert.equal(meta.tokensOut, null); + assert.equal(meta.costEstimateUsd, null); +}); + +test("captures mappedTo from request override", () => { + const req = makeReq({ + mappedModel: "gpt-4o", + requestBody: JSON.stringify({ model: "gpt-3.5", messages: [] }), + }); + const meta = extractLlmMetadata(req); + assert.ok(meta); + assert.equal(meta.mappedTo, "gpt-4o"); +}); diff --git a/tests/unit/inspector-sse-merger.test.ts b/tests/unit/inspector-sse-merger.test.ts new file mode 100644 index 0000000000..10f8b996fb --- /dev/null +++ b/tests/unit/inspector-sse-merger.test.ts @@ -0,0 +1,210 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { + detectApiFormat, + mergeStream, + parseSseStream, + rebuildAnthropic, + rebuildGemini, + rebuildOpenAI, + type SseEvent, +} from "../../src/mitm/inspector/sseMerger.ts"; + +function jsonChunks(payloads: unknown[]): SseEvent[] { + return payloads.map((p) => ({ data: JSON.stringify(p), json: p })); +} + +test("detectApiFormat — message_start → anthropic", () => { + const chunks = jsonChunks([ + { type: "message_start", message: { id: "msg_1" } }, + ]); + assert.equal(detectApiFormat(chunks), "anthropic"); +}); + +test("detectApiFormat — content_block_delta → anthropic", () => { + const chunks = jsonChunks([ + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Hi" } }, + ]); + assert.equal(detectApiFormat(chunks), "anthropic"); +}); + +test("detectApiFormat — choices[].delta → openai", () => { + const chunks = jsonChunks([ + { choices: [{ index: 0, delta: { content: "Hi" } }] }, + ]); + assert.equal(detectApiFormat(chunks), "openai"); +}); + +test("detectApiFormat — candidates → gemini", () => { + const chunks = jsonChunks([ + { candidates: [{ content: { parts: [{ text: "Hello" }] } }] }, + ]); + assert.equal(detectApiFormat(chunks), "gemini"); +}); + +test("detectApiFormat — no JSON chunks → unknown", () => { + assert.equal(detectApiFormat([{ data: "weird non-json" }]), "unknown"); +}); + +test("rebuildAnthropic — concat text_delta by index", () => { + const chunks = jsonChunks([ + { type: "message_start", message: { id: "msg_1" } }, + { type: "content_block_start", index: 0, content_block: { type: "text" } }, + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Hello" } }, + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: " world" } }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 7 } }, + ]); + const merged = rebuildAnthropic(chunks); + assert.equal(merged.format, "anthropic"); + const msg = merged.message as { content: Array<{ text: string }>; stop_reason: string; usage: { output_tokens: number } }; + assert.equal(msg.content[0].text, "Hello world"); + assert.equal(msg.stop_reason, "end_turn"); + assert.equal(msg.usage.output_tokens, 7); +}); + +test("rebuildAnthropic — input_json_delta merges and JSON.parses", () => { + const chunks = jsonChunks([ + { type: "message_start", message: {} }, + { + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "tu_1", name: "search" }, + }, + { type: "content_block_delta", index: 0, delta: { type: "input_json_delta", partial_json: '{"q":' } }, + { type: "content_block_delta", index: 0, delta: { type: "input_json_delta", partial_json: '"hi"}' } }, + { type: "content_block_stop", index: 0 }, + ]); + const merged = rebuildAnthropic(chunks); + const msg = merged.message as { content: Array<{ type: string; input: { q: string }; name: string }> }; + assert.equal(msg.content[0].type, "tool_use"); + assert.equal(msg.content[0].name, "search"); + assert.deepEqual(msg.content[0].input, { q: "hi" }); +}); + +test("rebuildAnthropic — thinking_delta accumulates", () => { + const chunks = jsonChunks([ + { type: "content_block_start", index: 0, content_block: { type: "thinking" } }, + { type: "content_block_delta", index: 0, delta: { type: "thinking_delta", thinking: "I am " } }, + { type: "content_block_delta", index: 0, delta: { type: "thinking_delta", thinking: "thinking..." } }, + ]); + const merged = rebuildAnthropic(chunks); + const msg = merged.message as { content: Array<{ thinking: string }> }; + assert.equal(msg.content[0].thinking, "I am thinking..."); +}); + +test("rebuildOpenAI — concat delta.content per choice index", () => { + const chunks = jsonChunks([ + { id: "c1", model: "gpt-4", choices: [{ index: 0, delta: { role: "assistant", content: "Hel" } }] }, + { choices: [{ index: 0, delta: { content: "lo" }, finish_reason: null }] }, + { choices: [{ index: 0, delta: {}, finish_reason: "stop" }] }, + { usage: { prompt_tokens: 3, completion_tokens: 1 } }, + ]); + const merged = rebuildOpenAI(chunks); + assert.equal(merged.format, "openai"); + const msg = merged.message as { + id: string; + model: string; + choices: Array<{ message: { content: string; role: string }; finish_reason: string }>; + usage: { prompt_tokens: number }; + }; + assert.equal(msg.id, "c1"); + assert.equal(msg.model, "gpt-4"); + assert.equal(msg.choices[0].message.content, "Hello"); + assert.equal(msg.choices[0].message.role, "assistant"); + assert.equal(msg.choices[0].finish_reason, "stop"); + assert.equal(msg.usage.prompt_tokens, 3); +}); + +test("rebuildOpenAI — merges tool_calls per (choice, tool) index", () => { + const chunks = jsonChunks([ + { + choices: [ + { + index: 0, + delta: { + tool_calls: [{ index: 0, id: "tc_1", type: "function", function: { name: "search", arguments: '{"q":' } }], + }, + }, + ], + }, + { + choices: [ + { + index: 0, + delta: { + tool_calls: [{ index: 0, function: { arguments: '"hi"}' } }], + }, + }, + ], + }, + ]); + const merged = rebuildOpenAI(chunks); + const msg = merged.message as { + choices: Array<{ + message: { + tool_calls: Array<{ id: string; function: { name: string; arguments: string } }>; + }; + }>; + }; + assert.equal(msg.choices[0].message.tool_calls[0].id, "tc_1"); + assert.equal(msg.choices[0].message.tool_calls[0].function.name, "search"); + assert.equal(msg.choices[0].message.tool_calls[0].function.arguments, '{"q":"hi"}'); +}); + +test("rebuildGemini — merges parts across candidates", () => { + const chunks = jsonChunks([ + { candidates: [{ content: { parts: [{ text: "Hello " }] } }] }, + { candidates: [{ content: { parts: [{ text: "world" }] } }] }, + { usageMetadata: { promptTokenCount: 2, candidatesTokenCount: 2 } }, + ]); + const merged = rebuildGemini(chunks); + assert.equal(merged.format, "gemini"); + const msg = merged.message as { + candidates: Array<{ content: { parts: Array<{ text: string }> } }>; + usageMetadata: { promptTokenCount: number }; + }; + assert.equal(msg.candidates[0].content.parts.length, 2); + assert.equal(msg.candidates[0].content.parts[0].text, "Hello "); + assert.equal(msg.candidates[0].content.parts[1].text, "world"); + assert.equal(msg.usageMetadata.promptTokenCount, 2); +}); + +test("mergeStream — unknown format returns raw fallback (no crash)", () => { + const chunks: SseEvent[] = [ + { data: "garbage" }, + { data: "{}", json: { foo: 1 } }, + ]; + const merged = mergeStream(chunks); + assert.equal(merged.format, "unknown"); + assert.ok(Array.isArray(merged.raw)); + assert.equal(merged.raw!.length, 2); +}); + +test("parseSseStream — parses event/data blocks separated by blank lines", () => { + const raw = + "event: foo\ndata: 1\n\n" + + 'data: {"x":1}\n\n' + + "data: [DONE]\n\n"; + const events = parseSseStream(raw); + assert.equal(events.length, 3); + assert.equal(events[0].event, "foo"); + assert.deepEqual(events[1].json, { x: 1 }); + assert.equal(events[2].data, "[DONE]"); +}); + +test("parseSseStream — keeps raw data when JSON parse fails", () => { + const events = parseSseStream("data: {not-json\n\n"); + assert.equal(events.length, 1); + assert.equal(events[0].data, "{not-json"); + assert.equal(events[0].json, undefined); +}); + +test("mergeStream — dispatches by detected format", () => { + const anth = mergeStream(jsonChunks([{ type: "message_start", message: {} }])); + assert.equal(anth.format, "anthropic"); + const oai = mergeStream(jsonChunks([{ choices: [{ delta: { content: "x" } }] }])); + assert.equal(oai.format, "openai"); + const gem = mergeStream(jsonChunks([{ candidates: [{ content: { parts: [{ text: "x" }] } }] }])); + assert.equal(gem.format, "gemini"); +}); diff --git a/tests/unit/inspector-system-proxy.test.ts b/tests/unit/inspector-system-proxy.test.ts new file mode 100644 index 0000000000..cc033f38d2 --- /dev/null +++ b/tests/unit/inspector-system-proxy.test.ts @@ -0,0 +1,224 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import os from "node:os"; +import { + __setExec, + apply, + revert, + type ExecFileFn, +} from "../../src/mitm/inspector/systemProxyConfig.ts"; + +interface Call { + file: string; + args: string[]; +} + +function makeRecorder(stdoutByCmd: Record = {}): { + calls: Call[]; + exec: ExecFileFn; +} { + const calls: Call[] = []; + const exec: ExecFileFn = async (file, args) => { + calls.push({ file, args }); + const key = `${file} ${args.join(" ")}`; + for (const [pattern, out] of Object.entries(stdoutByCmd)) { + if (key.includes(pattern)) return { stdout: out, stderr: "" }; + } + return { stdout: "", stderr: "" }; + }; + return { calls, exec }; +} + +test("macOS apply uses execFile array-form and captures previous state", async (t) => { + const orig = os.platform; + (os as { platform: () => NodeJS.Platform }).platform = () => "darwin" as NodeJS.Platform; + t.after(() => { + (os as { platform: () => NodeJS.Platform }).platform = orig; + }); + + const { calls, exec } = makeRecorder({ + "-getwebproxy Wi-Fi": "Enabled: Yes\nServer: 10.0.0.1\nPort: 8888\n", + "-getsecurewebproxy Wi-Fi": "Enabled: No\nServer:\nPort: 0\n", + }); + const restore = __setExec(exec); + t.after(restore); + + const result = await apply(8080); + assert.equal(result.platform, "macos"); + // All calls must use array args, never a single shell string + for (const c of calls) { + assert.ok(Array.isArray(c.args)); + // file is bare command name, no spaces / pipes / redirects + assert.ok(!c.file.includes(" ")); + assert.ok(!c.file.includes(";")); + assert.ok(!c.file.includes("|")); + } + const setCall = calls.find((c) => c.args.includes("-setwebproxy")); + assert.ok(setCall); + assert.deepEqual(setCall.args, ["-setwebproxy", "Wi-Fi", "127.0.0.1", "8080"]); + const prev = result.previousState as { platform: string; http: { enabled: boolean } }; + assert.equal(prev.platform, "macos"); + assert.equal(prev.http.enabled, true); +}); + +test("macOS revert restores prior server when http was enabled", async (t) => { + const orig = os.platform; + (os as { platform: () => NodeJS.Platform }).platform = () => "darwin" as NodeJS.Platform; + t.after(() => { + (os as { platform: () => NodeJS.Platform }).platform = orig; + }); + + const { calls, exec } = makeRecorder(); + const restore = __setExec(exec); + t.after(restore); + + await revert({ + platform: "macos", + service: "Wi-Fi", + http: { enabled: true, host: "10.0.0.1", port: "8888" }, + https: { enabled: false, host: "", port: "" }, + }); + const restoreCall = calls.find((c) => c.args.includes("-setwebproxy")); + assert.ok(restoreCall); + assert.deepEqual(restoreCall.args, ["-setwebproxy", "Wi-Fi", "10.0.0.1", "8888"]); + // https disabled previously → revert should turn it off + const offCall = calls.find((c) => c.args.includes("-setsecurewebproxystate")); + assert.ok(offCall); + assert.deepEqual(offCall.args, ["-setsecurewebproxystate", "Wi-Fi", "off"]); +}); + +test("Linux apply uses gsettings with array args", async (t) => { + const orig = os.platform; + (os as { platform: () => NodeJS.Platform }).platform = () => "linux" as NodeJS.Platform; + t.after(() => { + (os as { platform: () => NodeJS.Platform }).platform = orig; + }); + + const { calls, exec } = makeRecorder({ + "get org.gnome.system.proxy mode": "'none'\n", + "get org.gnome.system.proxy.http host": "''\n", + }); + const restore = __setExec(exec); + t.after(restore); + + const result = await apply(9090); + assert.equal(result.platform, "linux"); + const setMode = calls.find( + (c) => c.args[0] === "set" && c.args[1] === "org.gnome.system.proxy" && c.args[2] === "mode" + ); + assert.ok(setMode); + assert.deepEqual(setMode.args, ["set", "org.gnome.system.proxy", "mode", "manual"]); + const setHost = calls.find( + (c) => + c.args[0] === "set" && + c.args[1] === "org.gnome.system.proxy.http" && + c.args[2] === "host" + ); + assert.ok(setHost); + assert.deepEqual(setHost.args, ["set", "org.gnome.system.proxy.http", "host", "127.0.0.1"]); + // port string is passed as own arg (no shell interpolation) + const setPort = calls.find( + (c) => + c.args[0] === "set" && + c.args[1] === "org.gnome.system.proxy.http" && + c.args[2] === "port" + ); + assert.ok(setPort); + assert.equal(setPort.args[3], "9090"); +}); + +test("Linux revert restores recorded gnomeMode", async (t) => { + const orig = os.platform; + (os as { platform: () => NodeJS.Platform }).platform = () => "linux" as NodeJS.Platform; + t.after(() => { + (os as { platform: () => NodeJS.Platform }).platform = orig; + }); + + const { calls, exec } = makeRecorder(); + const restore = __setExec(exec); + t.after(restore); + + await revert({ + platform: "linux", + gnomeMode: "'auto'", + httpHost: "old.host", + httpPort: "1234", + httpsHost: "", + httpsPort: "", + }); + const restoreMode = calls.find( + (c) => c.args[0] === "set" && c.args[1] === "org.gnome.system.proxy" && c.args[2] === "mode" + ); + assert.ok(restoreMode); + assert.equal(restoreMode.args[3], "'auto'"); +}); + +test("Windows apply passes proxyArg as single arg, no shell interpolation", async (t) => { + const orig = os.platform; + (os as { platform: () => NodeJS.Platform }).platform = () => "win32" as NodeJS.Platform; + t.after(() => { + (os as { platform: () => NodeJS.Platform }).platform = orig; + }); + + const { calls, exec } = makeRecorder({ + "winhttp show proxy": "Direct access (no proxy server).", + }); + const restore = __setExec(exec); + t.after(restore); + + const result = await apply(7777); + assert.equal(result.platform, "windows"); + const setCall = calls.find((c) => c.args.join(" ") === "winhttp set proxy 127.0.0.1:7777"); + assert.ok(setCall); + assert.equal(setCall.file, "netsh"); + // Argument is one literal token — no embedded spaces, semicolons, pipes + const proxyArg = setCall.args[setCall.args.length - 1]; + assert.equal(proxyArg, "127.0.0.1:7777"); +}); + +test("Windows revert calls netsh winhttp reset proxy", async (t) => { + const orig = os.platform; + (os as { platform: () => NodeJS.Platform }).platform = () => "win32" as NodeJS.Platform; + t.after(() => { + (os as { platform: () => NodeJS.Platform }).platform = orig; + }); + + const { calls, exec } = makeRecorder(); + const restore = __setExec(exec); + t.after(restore); + + await revert({ platform: "windows", netshOutput: "" }); + const resetCall = calls.find((c) => c.args.join(" ") === "winhttp reset proxy"); + assert.ok(resetCall); +}); + +test("apply throws sanitized error when exec fails", async (t) => { + const orig = os.platform; + (os as { platform: () => NodeJS.Platform }).platform = () => "darwin" as NodeJS.Platform; + t.after(() => { + (os as { platform: () => NodeJS.Platform }).platform = orig; + }); + + const exec: ExecFileFn = async () => { + throw new Error("ENOENT: /usr/bin/networksetup"); + }; + const restore = __setExec(exec); + t.after(restore); + + await assert.rejects(() => apply(8080), (err: Error) => { + // sanitizeErrorMessage strips paths; assert we still get an Error + assert.ok(err instanceof Error); + assert.ok(err.message.length > 0); + return true; + }); +}); + +test("revert no-ops for unknown platform payload", async (t) => { + const { calls, exec } = makeRecorder(); + const restore = __setExec(exec); + t.after(restore); + + await revert(null); + await revert({ platform: "freebsd" }); + assert.equal(calls.length, 0); +});