T-HUB/packages/egress/src/client.test.ts

import { EventEmitter } from "node:events";
import http, { type ClientRequest, type IncomingMessage, type RequestOptions } from "node:http";
import https from "node:https";
import { PassThrough } from "node:stream";
import { afterEach, describe, expect, it, vi } from "vitest";
import { createEgressClient } from "./client.js";
import { testDatabase } from "./test-db.js";

interface Fixture { status?: number; headers?: Record<string, string>; body?: string; hang?: boolean }
interface Call { url: URL; options: RequestOptions; pin?: string; body?: Buffer }
function mockWire(fixtures: Fixture[]) {
  const calls: Call[] = [];
  const implementation = ((url: URL, options: RequestOptions, callback: (response: IncomingMessage) => void) => {
    const fixture = fixtures.shift();
    if (!fixture) throw new Error("Unexpected connection");
    const call: Call = { url, options };
    calls.push(call);
    const request = new EventEmitter() as ClientRequest;
    request.destroy = () => { request.emit("close"); return request; };
    request.end = ((body?: Buffer) => {
      call.body = body;
      const lookup = options.lookup as (host: string, options: object, cb: (error: Error | null, address: string) => void) => void;
      lookup(url.hostname, {}, (_error, address) => { call.pin = address; });
      if (!fixture.hang) queueMicrotask(() => {
        const stream = new PassThrough();
        const incoming = stream as unknown as IncomingMessage;
        incoming.statusCode = fixture.status ?? 200;
        incoming.statusMessage = "OK";
        incoming.rawHeaders = Object.entries(fixture.headers ?? {}).flat();
        callback(incoming);
        stream.end(fixture.body ?? "ok");
      });
      return request;
    }) as ClientRequest["end"];
    return request;
  }) as typeof http.request;
  vi.spyOn(http, "request").mockImplementation(implementation);
  vi.spyOn(https, "request").mockImplementation(implementation as typeof https.request);
  return calls;
}

describe("validated, pinned, budgeted fetch", () => {
  const pool = testDatabase();
  afterEach(() => { vi.restoreAllMocks(); });
  const publicResolver = async () => [{ address: "8.8.8.8", family: 4 }];
  it("connects to the pinned answer once, retaining TLS hostname and overriding Host spoofing", async () => {
    const calls = mockWire([{ body: "pinned" }]);
    const resolver = vi.fn().mockResolvedValueOnce([{ address: "8.8.8.8", family: 4 }]).mockResolvedValueOnce([{ address: "127.0.0.1", family: 4 }]);
    const client = createEgressClient({ pool, resolver });
    const response = await client.egressFetch("https://example.com/asset", { headers: { Host: "internal.local" } }, { hostClass: "monitor" });
    expect(await response.text()).toBe("pinned");
    expect(response.url).toBe("https://example.com/asset");
    expect(resolver).toHaveBeenCalledTimes(1);
    expect(calls[0]!.pin).toBe("8.8.8.8");
    expect(calls[0]!.url.hostname).toBe("example.com");
    expect(calls[0]!.options.agent).toBe(false);
    expect(calls[0]!.options.headers).toMatchObject({ host: "example.com" });
  });
  it.each(["http://127.0.0.1/secret", "https://[::1]/secret", "https://internal.example/secret"])("denies redirect to %s before making the second connection", async (location) => {
    const calls = mockWire([{ status: 302, headers: { location } }]);
    const resolver = vi.fn(async (host: string) => [{ address: host === "internal.example" ? "10.0.0.1" : "8.8.8.8", family: 4 }]);
    const client = createEgressClient({ pool, resolver });
    await expect(client.egressFetch("https://example.com/start", {}, { hostClass: "monitor" })).rejects.toMatchObject({ code: "DENIED_PRIVATE_ADDRESS" });
    expect(calls).toHaveLength(1);
  });
  it("resolves and pins again at each redirect hop, detecting same-host DNS rebinding", async () => {
    const calls = mockWire([{ status: 302, headers: { location: "/next" } }]);
    const resolver = vi.fn().mockResolvedValueOnce([{ address: "8.8.8.8", family: 4 }]).mockResolvedValueOnce([{ address: "127.0.0.1", family: 4 }]);
    await expect(createEgressClient({ pool, resolver }).egressFetch("https://example.com/start", {}, { hostClass: "monitor" })).rejects.toMatchObject({ code: "DENIED_PRIVATE_ADDRESS" });
    expect(resolver).toHaveBeenCalledTimes(2);
    expect(calls).toHaveLength(1);
  });
  it.each([302, 307, 308])("pinned-origin rejects an off-origin %s without forwarding auth or body", async (status) => {
    const calls = mockWire([{ status, headers: { location: "https://sink.example/token" } }]);
    const resolver = vi.fn(publicResolver);
    await expect(createEgressClient({ pool, resolver }).egressFetch("https://example.com/token",
      { method: "POST", body: "secret", headers: { authorization: "Bearer secret" } }, { hostClass: "monitor", pinnedOrigin: true })).rejects.toMatchObject({ code: "ORIGIN_DENIED" });
    expect(calls).toHaveLength(1);
    expect(resolver).toHaveBeenCalledTimes(1);
  });
  it("strips sensitive headers on public cross-origin redirects and normalizes POST to GET", async () => {
    const calls = mockWire([{ status: 302, headers: { location: "//cdn.example/next" } }, { body: "done" }]);
    const response = await createEgressClient({ pool, resolver: publicResolver }).egressFetch("https://example.com/start",
      { method: "POST", body: "form", headers: { authorization: "secret", cookie: "session=secret", "content-type": "text/plain" } }, { hostClass: "monitor" });
    expect(await response.text()).toBe("done");
    expect(response.redirected).toBe(true);
    expect(calls[1]!.options.method).toBe("GET");
    expect(calls[1]!.body).toBeUndefined();
    expect(calls[1]!.options.headers).not.toHaveProperty("authorization");
    expect(calls[1]!.options.headers).not.toHaveProperty("cookie");
    expect(calls[1]!.options.headers).not.toHaveProperty("content-type");
  });
  it("follows same-origin 307 without changing method/body", async () => {
    const calls = mockWire([{ status: 307, headers: { location: "/next" } }, {}]);
    await createEgressClient({ pool, resolver: publicResolver }).egressFetch("https://example.com/start", { method: "POST", body: "payload" }, { hostClass: "monitor", pinnedOrigin: true });
    expect(calls[1]!.options.method).toBe("POST");
    expect(calls[1]!.body?.toString()).toBe("payload");
  });
  it("bounds cycles and redirect depth; manual returns without following", async () => {
    let calls = mockWire([{ status: 302, headers: { location: "/start" } }]);
    await expect(createEgressClient({ pool, resolver: publicResolver }).egressFetch("https://example.com/start", {}, { hostClass: "monitor" })).rejects.toMatchObject({ code: "REDIRECT_LOOP" });
    expect(calls).toHaveLength(1);
    calls = mockWire([{ status: 302, headers: { location: "/next" } }]);
    await expect(createEgressClient({ pool, resolver: publicResolver, maxRedirects: 0 }).egressFetch("https://other.example/start", {}, { hostClass: "monitor" })).rejects.toMatchObject({ code: "REDIRECT_LOOP" });
    calls = mockWire([{ status: 302, headers: { location: "http://127.0.0.1/" } }]);
    const response = await createEgressClient({ pool, resolver: publicResolver }).egressFetch("https://manual.example/start", { redirect: "manual" }, { hostClass: "monitor" });
    expect(response.status).toBe(302);
    expect(calls).toHaveLength(1);
  });
  it("parks on 429 before another caller can connect", async () => {
    const calls = mockWire([{ status: 429, headers: { "retry-after": "10" } }]);
    const client = createEgressClient({ pool, resolver: publicResolver });
    expect((await client.egressFetch("https://example.com/start", {}, { hostClass: "monitor" })).status).toBe(429);
    await expect(client.egressFetch("https://example.com/next", {}, { hostClass: "monitor" })).rejects.toMatchObject({ code: "BUDGET_EXHAUSTED" });
    expect(calls).toHaveLength(1);
  });
  it.each<Record<string, string>>([{}, { "content-length": "100" }])("enforces response caps for streamed and declared lengths: %j", async (headers) => {
    mockWire([{ headers, body: "12345" }]);
    await expect(createEgressClient({ pool, resolver: publicResolver, maxResponseBytes: 4 }).egressFetch("https://example.com", {}, { hostClass: "monitor" })).rejects.toMatchObject({ code: "RESPONSE_TOO_LARGE" });
  });
  it("caps request bodies before acquiring a token or connecting", async () => {
    const calls = mockWire([]);
    await expect(createEgressClient({ pool, resolver: publicResolver, maxRequestBytes: 4 }).egressFetch("https://example.com", { method: "POST", body: "12345" }, { hostClass: "monitor" })).rejects.toMatchObject({ code: "REQUEST_TOO_LARGE" });
    expect(calls).toHaveLength(0);
    expect((await pool.query("SELECT * FROM host_budgets")).rows).toHaveLength(0);
  });
  it("returns HEAD metadata without applying a body cap to Content-Length", async () => {
    mockWire([{ headers: { "content-length": "100000", "content-encoding": "gzip" }, body: "" }]);
    const response = await createEgressClient({ pool, resolver: publicResolver, maxResponseBytes: 4 }).egressFetch("https://example.com", { method: "HEAD" }, { hostClass: "monitor" });
    expect(response.headers.get("content-length")).toBe("100000");
    expect(await response.text()).toBe("");
  });
  it("rejects whitespace in redirect targets instead of normalizing it away", async () => {
    const calls = mockWire([{ status: 302, headers: { location: "https://next.example/a b" } }]);
    await expect(createEgressClient({ pool, resolver: publicResolver }).egressFetch("https://example.com", {}, { hostClass: "monitor" })).rejects.toMatchObject({ code: "INVALID_URL" });
    expect(calls).toHaveLength(1);
  });
  it("times out a stalled DNS resolution and a stalled connection", async () => {
    const stalled = () => new Promise<never>(() => undefined);
    await expect(createEgressClient({ pool, resolver: stalled, timeoutMs: 30 }).egressFetch("https://example.com", {}, { hostClass: "monitor" })).rejects.toMatchObject({ code: "TIMEOUT" });
    mockWire([{ hang: true }]);
    await expect(createEgressClient({ pool, resolver: publicResolver, timeoutMs: 100 }).egressFetch("https://example.com", {}, { hostClass: "monitor" })).rejects.toMatchObject({ code: "TIMEOUT" });
  });
  it("honors an already-aborted signal without DNS, debit, or network", async () => {
    const controller = new AbortController(); controller.abort();
    const resolver = vi.fn(publicResolver);
    await expect(createEgressClient({ pool, resolver }).egressFetch("https://example.com", { signal: controller.signal }, { hostClass: "monitor" })).rejects.toMatchObject({ code: "ABORTED" });
    expect(resolver).not.toHaveBeenCalled();
  });
});