Merge PR #8107: fix(ai): honor StreamOptions.headers in Bedrock and Cursor (@svperfecta)

This commit is contained in:
can1357
2026-08-11 15:06:14 +02:00
5 changed files with 485 additions and 0 deletions
+4
View File
@@ -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
@@ -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<string, string> = {};
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<string, string> = {
...callerHeaders,
"content-type": "application/json",
accept: "application/vnd.amazon.eventstream",
};
+76
View File
@@ -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<string, string> | undefined): Record<string, string> {
const sanitized: Record<string, string> = {};
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",
@@ -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<T>(body: () => Promise<T>): Promise<T> {
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<string, string>, 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<string, string> }): 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<string, string>;
let index = 0;
const body = new ReadableStream<Uint8Array>({
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<string, string> } = {};
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<string, string> } = {};
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");
});
});
@@ -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<http2.Http2Session>();
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<string> {
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<void>();
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<void> {
for (const session of sessions) session.destroy();
sessions.clear();
if (!server) return;
const closing = server;
server = undefined;
const closed = Promise.withResolvers<void>();
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<string, string>): Promise<http2.IncomingHttpHeaders> {
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");
});
});