diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index caecb360c..e22a61545 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed the Amazon Bedrock and Cursor transports ignoring `StreamOptions.headers`; both built their request headers from scratch, so caller-supplied tracing or attribution headers were silently dropped while working on every other provider ([#8107](https://github.com/can1357/oh-my-pi/pull/8107) by [@svperfecta](https://github.com/svperfecta)). + ## [17.2.12] - 2026-08-08 ### Fixed diff --git a/packages/ai/src/providers/amazon-bedrock.ts b/packages/ai/src/providers/amazon-bedrock.ts index 453d90e15..4473f46d2 100644 --- a/packages/ai/src/providers/amazon-bedrock.ts +++ b/packages/ai/src/providers/amazon-bedrock.ts @@ -47,6 +47,19 @@ import { decodeEventStream } from "./aws-eventstream"; import { signRequest } from "./aws-sigv4"; import { transformMessages } from "./transform-messages"; +/** + * Headers SigV4 generates for itself. A caller cannot be allowed to supply these: + * `signRequest` would sign the caller's value but return its own, so the signature + * would not match what goes on the wire. + */ +const SIGNER_OWNED_HEADERS = new Set(["host", "x-amz-date", "x-amz-content-sha256", "x-amz-security-token"]); + +/** Headers the Bedrock request sets itself; a caller copy in any casing duplicates them. */ +// `content-length` included: the fetch layer recomputes it from the serialized +// body, so a caller value would be signed but not sent, and AWS rejects the +// mismatch. +const BEDROCK_RESERVED_HEADERS = new Set(["content-type", "accept", "authorization", "content-length"]); + export type BedrockThinkingDisplay = "summarized" | "omitted"; export interface BedrockOptions extends StreamOptions { @@ -356,7 +369,32 @@ export const streamBedrock: StreamFunction<"bedrock-converse-stream"> = ( const bodyText = JSON.stringify(commandInput); const body = new TextEncoder().encode(bodyText); + // Caller headers are merged BEFORE signing, so SigV4 covers them and they + // reach the wire. Bedrock built its header map from scratch and ignored + // `options.headers` entirely, so tracing/attribution headers set by a + // caller (or by a `before_provider_headers` extension) were silently + // dropped here while working on every other provider. Content-type and + // accept stay last: the eventstream framing is not the caller's to change. + // + // The signer's OWN headers are dropped first, and that is load-bearing: + // `signRequest` lets a caller value overwrite `host`/`x-amz-*` in the map + // it signs, but always RETURNS the generated ones, which `requestHeaders` + // below then puts on the wire. A caller supplying any of them would sign + // one set of values and send another, and Bedrock would reject every + // request with a signature mismatch. + // Lower-cased, and names the request sets itself are dropped. Keeping a + // caller `Content-Type` beside the fixed `content-type` leaves TWO object + // keys: SigV4 signs one value while fetch canonicalizes both into a single + // comma-joined wire header, so AWS validates different bytes than were + // signed and rejects the request. + const callerHeaders: Record = {}; + for (const [name, value] of Object.entries(options?.headers ?? {})) { + const field = name.toLowerCase(); + if (SIGNER_OWNED_HEADERS.has(field) || BEDROCK_RESERVED_HEADERS.has(field)) continue; + callerHeaders[field] = value; + } const baseHeaders: Record = { + ...callerHeaders, "content-type": "application/json", accept: "application/vnd.amazon.eventstream", }; diff --git a/packages/ai/src/providers/cursor.ts b/packages/ai/src/providers/cursor.ts index d276a27b0..23bcae9af 100644 --- a/packages/ai/src/providers/cursor.ts +++ b/packages/ai/src/providers/cursor.ts @@ -223,6 +223,67 @@ import { export const CURSOR_API_URL = "https://api2.cursor.sh"; export const CURSOR_CLIENT_VERSION = "cli-2026.07.23-e383d2b"; +/** + * HTTP/1 connection-specific headers that HTTP/2 forbids. Node's `http2.request()` + * throws `ERR_HTTP2_INVALID_CONNECTION_HEADERS` on these rather than dropping + * them, so a caller sending one would kill the request outright. + */ +const HTTP2_FORBIDDEN_HEADERS = new Set([ + "connection", + "keep-alive", + "proxy-connection", + "transfer-encoding", + "upgrade", + "http2-settings", +]); + +/** + * Header names the Cursor request sets for itself. A caller copy in ANY casing + * has to go: the spread below adds the fixed lower-case name regardless, and two + * spellings of one field are a duplicate rather than an override. + */ +const CURSOR_RESERVED_HEADERS = new Set([ + "content-type", + "connect-protocol-version", + "te", + "authorization", + "x-ghost-mode", + "x-cursor-client-version", + "x-cursor-client-type", + "x-request-id", + // Transport-owned even though this request never sets it: node's http2 client + // suppresses the `:authority` it derives from the URL when a plain `host` + // header is present, so a caller value here silently retargets the request at + // a different virtual host. + "host", + // The Connect body is streamed after the headers (initial frame, heartbeats, + // tool responses), so no caller-supplied length can describe it and an HTTP/2 + // peer resets the stream once the body diverges. + "content-length", +]); + +/** + * Reduce caller-supplied headers to what this HTTP/2 request can legally carry. + * + * Everything is lower-cased, because HTTP/2 field names are lower-case and node + * compares them that way. A caller `Authorization` next to the fixed + * `authorization` does not lose to it, it DUPLICATES it, and node throws + * `ERR_HTTP2_HEADER_SINGLE_VALUE` before the request goes out. Same for a `TE` + * that is not `trailers`. Node throws on all three classes here rather than + * ignoring them, so a miss turns a harmless header into a dead request. + */ +function sanitizeCursorCallerHeaders(headers: Record | undefined): Record { + const sanitized: Record = {}; + for (const [name, value] of Object.entries(headers ?? {})) { + const field = name.toLowerCase(); + if (field.startsWith(":")) continue; + if (HTTP2_FORBIDDEN_HEADERS.has(field)) continue; + if (CURSOR_RESERVED_HEADERS.has(field)) continue; + sanitized[field] = value; + } + return sanitized; +} + const CURSOR_PROXY_TUNNEL_TIMEOUT_MS = 30_000; /** @@ -545,7 +606,22 @@ export const streamCursor: StreamFunction<"cursor-agent"> = ( const baseUrl = model.baseUrl || CURSOR_API_URL; const requestPath = "/agent.v1.AgentService/Run"; + // Caller headers are additive, and are spread FIRST so the protocol + // framing, auth, and request id below always win. Cursor built this map + // from scratch and never read `options.headers`, so tracing/attribution + // headers set by a caller (or a `before_provider_headers` extension) were + // silently dropped here while working on other providers. + // + // Two classes are stripped because node's http2 client THROWS on them + // rather than ignoring them, which would turn a harmless header into a + // dead request: pseudo-headers, which belong to the transport, and the + // HTTP/1 connection-specific headers HTTP/2 forbids outright + // (ERR_HTTP2_INVALID_CONNECTION_HEADERS). `te` needs no filtering here — + // HTTP/2 allows it only as `trailers`, which is exactly what the fixed + // set below re-applies over anything a caller sent. + const callerHeaders = sanitizeCursorCallerHeaders(options?.headers); const requestHeaders = { + ...callerHeaders, ":method": "POST", ":path": requestPath, "content-type": "application/connect+proto", diff --git a/packages/ai/test/bedrock-caller-headers.test.ts b/packages/ai/test/bedrock-caller-headers.test.ts new file mode 100644 index 000000000..f9b251650 --- /dev/null +++ b/packages/ai/test/bedrock-caller-headers.test.ts @@ -0,0 +1,178 @@ +import { describe, expect, it } from "bun:test"; +import { streamBedrock } from "@oh-my-pi/pi-ai/providers/amazon-bedrock"; +import { crc32 } from "@oh-my-pi/pi-ai/providers/aws-eventstream"; +import type { Context, FetchImpl, Model } from "@oh-my-pi/pi-ai/types"; +import { buildModel } from "@oh-my-pi/pi-catalog/build"; + +// Caller headers (including `before_provider_headers` extension edits) reach the +// Bedrock request, but SigV4's own headers must never come from the caller: +// `signRequest` signs the caller's value and then RETURNS its own, so the wire +// would carry different bytes than the signature covers and Bedrock would reject +// every request. Exercised through the real signing path, not a unit stub. + +/** + * Run `body` with dummy AWS credentials, restoring the environment immediately. + * + * Scoped to the one test rather than the file: a `beforeAll` override leaves + * every later Bedrock file in the same Bun process on the dummy-credential path + * until `afterAll` runs, which is the full-suite hazard `AGENTS.md` rules out. + */ +async function withSkippedAuth(body: () => Promise): Promise { + const original = process.env.AWS_BEDROCK_SKIP_AUTH; + process.env.AWS_BEDROCK_SKIP_AUTH = "1"; + try { + return await body(); + } finally { + if (original === undefined) delete process.env.AWS_BEDROCK_SKIP_AUTH; + else process.env.AWS_BEDROCK_SKIP_AUTH = original; + } +} + +function encodeFrame(headers: Record, payload: Uint8Array): Uint8Array { + const headerParts: Uint8Array[] = []; + for (const [name, value] of Object.entries(headers)) { + const nameBytes = new TextEncoder().encode(name); + const valueBytes = new TextEncoder().encode(value); + const part = new Uint8Array(1 + nameBytes.length + 1 + 2 + valueBytes.length); + const partView = new DataView(part.buffer); + let cursor = 0; + partView.setUint8(cursor, nameBytes.length); + cursor += 1; + part.set(nameBytes, cursor); + cursor += nameBytes.length; + partView.setUint8(cursor, 7); + cursor += 1; + partView.setUint16(cursor, valueBytes.length, false); + cursor += 2; + part.set(valueBytes, cursor); + headerParts.push(part); + } + const headerLength = headerParts.reduce((total, part) => total + part.length, 0); + const headerBytes = new Uint8Array(headerLength); + let offset = 0; + for (const part of headerParts) { + headerBytes.set(part, offset); + offset += part.length; + } + const totalLength = 12 + headerLength + payload.length + 4; + const frame = new Uint8Array(totalLength); + const view = new DataView(frame.buffer); + view.setUint32(0, totalLength, false); + view.setUint32(4, headerLength, false); + view.setUint32(8, crc32(frame.subarray(0, 8)), false); + frame.set(headerBytes, 12); + frame.set(payload, 12 + headerLength); + view.setUint32(totalLength - 4, crc32(frame.subarray(0, totalLength - 4)), false); + return frame; +} + +function bedrockEvent(eventType: string, payload: string): Uint8Array { + return encodeFrame({ ":message-type": "event", ":event-type": eventType }, new TextEncoder().encode(payload)); +} + +/** Captures the headers actually sent, and replies with a minimal valid stream. */ +function capturingFetch(seen: { headers?: Record }): FetchImpl { + const frames = [ + bedrockEvent("messageStart", '{"role":"assistant"}'), + bedrockEvent("contentBlockDelta", '{"contentBlockIndex":0,"delta":{"text":"hi"}}'), + bedrockEvent("contentBlockStop", '{"contentBlockIndex":0}'), + bedrockEvent("messageStop", '{"stopReason":"end_turn"}'), + bedrockEvent("metadata", '{"usage":{"inputTokens":1,"outputTokens":1,"totalTokens":2}}'), + ]; + return Object.assign( + async (_input: string | URL | Request, init?: RequestInit) => { + seen.headers = (init?.headers ?? {}) as Record; + let index = 0; + const body = new ReadableStream({ + pull(controller) { + if (index < frames.length) controller.enqueue(frames[index++]!); + else controller.close(); + }, + }); + return new Response(body, { status: 200, headers: { "content-type": "application/vnd.amazon.eventstream" } }); + }, + { preconnect: fetch.preconnect }, + ); +} + +function model(): Model<"bedrock-converse-stream"> { + return buildModel({ + id: "anthropic.claude-3-5-sonnet-20241022-v2:0", + name: "Claude 3.5 Sonnet", + api: "bedrock-converse-stream", + provider: "amazon-bedrock", + baseUrl: "https://bedrock-runtime.us-east-1.amazonaws.com", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 200_000, + maxTokens: 8_192, + }); +} + +const context: Context = { messages: [{ role: "user", content: "hi", timestamp: 0 }] }; + +describe("Bedrock caller headers", () => { + it("forwards caller headers but never lets them supply SigV4's own", async () => { + const seen: { headers?: Record } = {}; + await withSkippedAuth(async () => { + const stream = streamBedrock(model(), context, { + region: "us-east-1", + fetch: capturingFetch(seen), + headers: { + "x-trace": "kept", + // Every header SigV4 generates for itself. Signed as the caller's value + // but sent as the signer's, these would break the signature. + host: "evil.example.com", + "x-amz-date": "19700101T000000Z", + "x-amz-content-sha256": "deadbeef", + "x-amz-security-token": "forged", + }, + }); + await stream.result(); + }); + + const headers = seen.headers ?? {}; + // The benign caller header still reaches the request: that is the feature. + expect(headers["x-trace"]).toBe("kept"); + // None of the signer-owned values are the caller's. + expect(headers.host).not.toBe("evil.example.com"); + expect(headers["x-amz-date"]).not.toBe("19700101T000000Z"); + expect(headers["x-amz-content-sha256"]).not.toBe("deadbeef"); + expect(headers["x-amz-security-token"]).not.toBe("forged"); + // And the request was actually signed, so this is the real path. + expect(headers.authorization ?? headers.Authorization).toContain("AWS4-HMAC-SHA256"); + }); + + // A caller spelling differing only in case leaves two object keys: SigV4 signs + // one, fetch comma-joins both onto the wire, and AWS rejects the mismatch. + it("does not leave a differently cased duplicate of a header it sets itself", async () => { + const seen: { headers?: Record } = {}; + await withSkippedAuth(async () => { + const stream = streamBedrock(model(), context, { + region: "us-east-1", + fetch: capturingFetch(seen), + headers: { + "Content-Type": "text/plain", + Accept: "text/plain", + Host: "evil.example.com", + // Recomputed by the fetch layer from the serialized body, so a caller + // value would be signed but never sent. + "Content-Length": "999", + "X-Trace": "kept", + }, + }); + await stream.result(); + }); + + const headers = seen.headers ?? {}; + const names = Object.keys(headers).map(name => name.toLowerCase()); + // Each field appears exactly once, whatever casing the caller used. + for (const field of ["content-type", "accept", "host", "content-length"]) { + expect(names.filter(name => name === field).length).toBeLessThanOrEqual(1); + } + expect(headers["content-type"]).toBe("application/json"); + // Ordinary caller headers still land, lower-cased. + expect(headers["x-trace"]).toBe("kept"); + }); +}); diff --git a/packages/ai/test/cursor-caller-headers.test.ts b/packages/ai/test/cursor-caller-headers.test.ts new file mode 100644 index 000000000..14b21613a --- /dev/null +++ b/packages/ai/test/cursor-caller-headers.test.ts @@ -0,0 +1,189 @@ +import { afterEach, describe, expect, it } from "bun:test"; +import * as http2 from "node:http2"; +import { create, toBinary } from "@bufbuild/protobuf"; +import { streamCursor } from "@oh-my-pi/pi-ai/providers/cursor"; +import type { Context, Model } from "@oh-my-pi/pi-ai/types"; +import { buildModel } from "@oh-my-pi/pi-catalog/build"; +import { + AgentServerMessageSchema, + InteractionUpdateSchema, + TextDeltaUpdateSchema, + TurnEndedUpdateSchema, +} from "@oh-my-pi/pi-catalog/discovery/cursor-gen/agent_pb"; + +// Cursor forwards caller headers (including `before_provider_headers` extension +// edits), and it speaks HTTP/2. These assert the TRANSPORT contract against a +// real local HTTP/2 server rather than the sanitizer in isolation: if +// `streamCursor` stopped merging caller headers, or merged the wrong ones, a +// helper-level test would still pass while the wire lost them. +// +// Two classes must never reach `http2.request()`, because node THROWS on them +// rather than ignoring them, turning a harmless header into a dead request: +// pseudo-headers and HTTP/1 connection-specific headers. A third class — +// headers the request sets for itself — must not arrive duplicated, since names +// are matched case-insensitively on the wire. + +let server: http2.Http2Server | undefined; +const sessions = new Set(); +let received: http2.IncomingHttpHeaders = {}; + +function frameConnectMessage(data: Uint8Array, flags = 0): Buffer { + const frame = Buffer.alloc(5 + data.length); + frame[0] = flags; + frame.writeUInt32BE(data.length, 1); + frame.set(data, 5); + return frame; +} + +function textDeltaFrame(text: string): Buffer { + const message = create(AgentServerMessageSchema, { + message: { + case: "interactionUpdate", + value: create(InteractionUpdateSchema, { + message: { case: "textDelta", value: create(TextDeltaUpdateSchema, { text }) }, + }), + }, + }); + return frameConnectMessage(toBinary(AgentServerMessageSchema, message)); +} + +function turnEndedFrame(): Buffer { + const message = create(AgentServerMessageSchema, { + message: { + case: "interactionUpdate", + value: create(InteractionUpdateSchema, { + message: { case: "turnEnded", value: create(TurnEndedUpdateSchema, {}) }, + }), + }, + }); + return frameConnectMessage(toBinary(AgentServerMessageSchema, message)); +} + +/** Records the headers the client actually sent, then replies with a clean turn. */ +async function startServer(): Promise { + server = http2.createServer(); + server.on("session", session => { + sessions.add(session); + session.on("close", () => sessions.delete(session)); + }); + server.on("stream", (stream: http2.ServerHttp2Stream, headers: http2.IncomingHttpHeaders) => { + stream.on("data", () => {}); + received = headers; + stream.respond({ ":status": 200, "content-type": "application/connect+proto" }); + stream.write(textDeltaFrame("ok")); + stream.write(turnEndedFrame()); + stream.end(); + }); + + const listening = Promise.withResolvers(); + server.once("error", listening.reject); + server.listen(0, "127.0.0.1", listening.resolve); + await listening.promise; + const address = server.address(); + if (!address || typeof address === "string") throw new Error("expected the fixture server to bind a tcp port"); + return `http://127.0.0.1:${address.port}`; +} + +async function stopServer(): Promise { + for (const session of sessions) session.destroy(); + sessions.clear(); + if (!server) return; + const closing = server; + server = undefined; + const closed = Promise.withResolvers(); + closing.close(error => (error ? closed.reject(error) : closed.resolve())); + await closed.promise; +} + +function makeModel(baseUrl: string): Model<"cursor-agent"> { + return buildModel({ + id: "cursor-caller-headers-fixture", + name: "Cursor caller headers fixture", + api: "cursor-agent", + provider: "cursor", + baseUrl, + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 1, + maxTokens: 1, + }); +} + +const context: Context = { messages: [{ role: "user", content: "headers", timestamp: 1 }] }; + +/** Drive one request to completion and hand back the headers the server saw. */ +async function send(headers: Record): Promise { + const baseUrl = await startServer(); + const stream = streamCursor(makeModel(baseUrl), context, { apiKey: "test-token", headers }); + for await (const _event of stream) { + // drain + } + await stream.result(); + return received; +} + +afterEach(async () => { + received = {}; + await stopServer(); +}); + +describe("Cursor caller headers reach the wire", () => { + it("delivers an ordinary caller header to the server", async () => { + const sent = await send({ "x-trace": "abc", "x-waygate-activity": "mode=plan" }); + expect(sent["x-trace"]).toBe("abc"); + expect(sent["x-waygate-activity"]).toBe("mode=plan"); + }); + + it("normalizes a caller header name to lower case", async () => { + const sent = await send({ "X-Trace": "abc" }); + expect(sent["x-trace"]).toBe("abc"); + }); + + // The request still has to go out. Node throws on these rather than dropping + // them, so a leak here is a dead request, not a missing header. + it("survives HTTP/1 connection-specific headers and pseudo-headers", async () => { + const sent = await send({ + connection: "keep-alive", + "keep-alive": "timeout=5", + "transfer-encoding": "chunked", + upgrade: "h2c", + ":path": "/evil", + "x-trace": "kept", + }); + // The request completed, and the benign header still landed. + expect(sent["x-trace"]).toBe("kept"); + expect(sent[":path"]).toBe("/agent.v1.AgentService/Run"); + expect(sent.connection).toBeUndefined(); + expect(sent["transfer-encoding"]).toBeUndefined(); + }); + + it("does not let a caller override the headers the request sets itself", async () => { + const sent = await send({ + Authorization: "Bearer stolen", + "Content-Type": "text/plain", + TE: "gzip", + "X-Request-Id": "forged", + // The Connect body is streamed after the headers, so no caller length can + // describe it; an HTTP/2 peer resets the stream once the body diverges. + "Content-Length": "999", + "x-trace": "kept", + }); + expect(sent.authorization).toBe("Bearer test-token"); + expect(sent["content-type"]).toBe("application/connect+proto"); + expect(sent.te).toBe("trailers"); + expect(sent["x-request-id"]).not.toBe("forged"); + expect(sent["content-length"]).toBeUndefined(); + expect(sent["x-trace"]).toBe("kept"); + }); + + // A plain `host` header suppresses the `:authority` node derives from the URL, + // so a caller value would silently retarget the request at another vhost. + it("does not let a caller header retarget the request authority", async () => { + const sent = await send({ Host: "evil.example.com", "x-trace": "kept" }); + expect(sent[":authority"]).not.toBe("evil.example.com"); + expect(sent[":authority"]).toContain("127.0.0.1"); + expect(sent.host).toBeUndefined(); + expect(sent["x-trace"]).toBe("kept"); + }); +});