diff --git a/CHANGELOG.md b/CHANGELOG.md index 14cf6a2..2c87714 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,7 @@ image and creates the GitHub release from this file. - Request shapes (`SHAPE_LOG`): the shape of each request that names its subjects, with none of its text (keys, size classes, line counts and listed enum values; tool names as digests; caller-defined keys anonymised; no caller-chosen value kept), recorded once per distinct shape. `npm run replay:shapes` replays them with invented people and fails on a leak or a refusal of ordinary text, so real traffic shapes are tested without keeping any of it. The test battery records its own traffic's shapes and replays them (`_test_shapes.mjs`, `_test_replay_recorded.mjs`). ### Fixed +- Test battery: the runner picks free loopback ports for each run and hands them to the stub and the suites (`CORDON_TEST_PORTS`), so two batteries on one host, such as two CI jobs of this repository on one self-hosted machine, no longer share a stub or kill each other's instances; the old fixed ports made one battery count the other's requests and fail cases that had passed. - Tenant identity under a bearer token: the credential tenant (`TENANT_FROM_AUTH`) now hashes the key with any `Bearer ` scheme removed from `Authorization`, so one key is one tenant whether a client sends it as `x-api-key` or as a bearer token. Before, the same key became a second tenant the day a client switched headers (the Anthropic SDK did at 1.0), and anything keyed on the tenant lost it: per-tenant policy, the audit trail, a canary counting the tenant's requests. - Replay: a thinking block's text and signature, tool-call ids and other opaque protocol values are filled with neutral words, not identifiers. Redaction leaves them as sent and the restore never writes identifiers into them, so the replay counted a leak no real request can carry. diff --git a/CLAUDE.md b/CLAUDE.md index cd0a0da..d166ff0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -20,7 +20,7 @@ npm run audit # verify the hash-chained audit log → tamper report (src/ docker compose up -d --build # single self-contained container, 127.0.0.1:8080 ``` -**Running one suite** (the runner `_run_tests.mjs` must already have the stub on :8900 and a cordon instance on :8810 up — easiest is to keep `npm test` as the entry point, but to iterate on a single suite, launch them by hand): +**Running one suite** (the runner `_run_tests.mjs` picks free ports for each run and passes them in `CORDON_TEST_PORTS`; run by hand, a suite falls back to the stub on :8900 and a cordon instance on :8810, which must already be up; the easiest path is to keep `npm test` as the entry point, but to iterate on a single suite, launch them by hand): ```bash node _stub-upstream.mjs & # :8900 diff --git a/_run_tests.mjs b/_run_tests.mjs index 144f429..680b849 100644 --- a/_run_tests.mjs +++ b/_run_tests.mjs @@ -4,29 +4,45 @@ // npm test import { spawn, execSync } from "node:child_process"; import { rmSync } from "node:fs"; +import { parsePorts, portsEnv, reservePorts, url } from "./_test_ports.mjs"; -const STUB_PORT = 8900; -const STUB = `http://localhost:${STUB_PORT}`; +/** + * Each run reserves its six ports host-wide (see reservePorts), so concurrent batteries on one + * host get distinct stubs and instances even when the kernel hands both the same free port. + * The reservation is held until this process exits. CORDON_TEST_PORTS set by hand pins the + * ports instead. + */ +async function pickPorts() { + if (process.env.CORDON_TEST_PORTS) return parsePorts(process.env.CORDON_TEST_PORTS); + const { ports, release } = await reservePorts(); + process.on("exit", release); + for (const sig of ["SIGINT", "SIGTERM", "SIGHUP"]) process.on(sig, () => process.exit(1)); + return ports; +} + +const PORTS = await pickPorts(); +process.env.CORDON_TEST_PORTS = portsEnv(PORTS); +const STUB = url(PORTS.stub); const base = { OPENAI_BASE: STUB, ANTHROPIC_BASE: STUB }; // port → instance env const instances = { // Primary instance: has a strong TENANT_SECRET, so consistent-pseudonym requests work, and // opts in to X-Tenant (TRUST_TENANT_HEADER) so the tenant-selection rule is exercised. - 8810: { ADMIN_TOKEN: "secret", AUDIT_LOG: "./_audit_test.jsonl", SHAPE_LOG: "./_shapes_test.jsonl", CORDON_RESTORE_TOOLS: "*", CORDON_TEST_HOOKS: "1", TENANT_SECRET: "test-secret-0123456789abcdef", TRUST_TENANT_HEADER: "true" }, + [PORTS.main]: { ADMIN_TOKEN: "secret", AUDIT_LOG: "./_audit_test.jsonl", SHAPE_LOG: "./_shapes_test.jsonl", CORDON_RESTORE_TOOLS: "*", CORDON_TEST_HOOKS: "1", TENANT_SECRET: "test-secret-0123456789abcdef", TRUST_TENANT_HEADER: "true" }, // Secret-less instance: consistent pseudonyms enabled per-tenant here must FAIL CLOSED // (no ALLOW_WEAK_PSEUDONYM_SECRET) — exercises the per-request pseudonym-secret guard. // TRUST_TENANT_HEADER is left at its default, so X-Tenant is ignored here. - 8811: { ADMIN_TOKEN: "secret", AUDIT_LOG: "./_audit_test_nosecret.jsonl", CORDON_RESTORE_TOOLS: "*" }, + [PORTS.nosecret]: { ADMIN_TOKEN: "secret", AUDIT_LOG: "./_audit_test_nosecret.jsonl", CORDON_RESTORE_TOOLS: "*" }, // No ADMIN_TOKEN; an Anthropic upstream accepts connections but never answers. - // The proxy suite opens it on :8901 with a short timeout for the 504 path. - 8812: { + // The proxy suite opens it on the dead port with a short timeout for the 504 path. + [PORTS.noadmin]: { AUDIT_LOG: "./_audit_test_noadmin.jsonl", - ANTHROPIC_BASE: "http://127.0.0.1:8901", + ANTHROPIC_BASE: `http://127.0.0.1:${PORTS.dead}`, UPSTREAM_TIMEOUT_MS: "500", }, // A restore allowlist: only the `text` argument of the `echo` tool gets real values back. - 8813: { AUDIT_LOG: "./_audit_test_restore.jsonl", CORDON_RESTORE_TOOLS: "echo.text,send" }, + [PORTS.restore]: { AUDIT_LOG: "./_audit_test_restore.jsonl", CORDON_RESTORE_TOOLS: "echo.text,send" }, }; // Suites importing TS modules directly run under the tsx loader. @@ -37,33 +53,6 @@ const suites = ["_test_unit.mjs", "_test_subject.mjs", "_test_fuzz.js", "_test_f for (const f of ["./_audit_test.jsonl", "./_shapes_test.jsonl", "./_audit_test_nosecret.jsonl", "./_audit_test_noadmin.jsonl", "./_audit_test_restore.jsonl", "./_policy_unit.json", "./_policy_bad.json"]) try { rmSync(f, { force: true }); } catch {} -/** - * Kill whatever is already LISTENING on `port`. Prior runs can leave an orphaned stub - * or cordon instance behind — most often on Windows, where a parent's .kill() doesn't - * always reap the spawned node process — which would then EADDRINUSE this run. - */ -const freePort = (port) => { - try { - if (process.platform === "win32") { - const out = execSync("netstat -ano -p tcp", { encoding: "utf8" }); - const pids = new Set(); - for (const line of out.split("\n")) { - const t = line.trim().split(/\s+/); // [proto, local, foreign, state, pid] - if (t.length >= 5 && t[3] === "LISTENING" && t[1].endsWith(":" + port)) pids.add(t[4]); - } - for (const pid of pids) try { execSync(`taskkill /PID ${pid} /F /T`, { stdio: "ignore" }); } catch {} - } else { - try { execSync(`fuser -k ${port}/tcp`, { stdio: "ignore" }); } - catch { - try { - for (const pid of execSync(`lsof -ti tcp:${port}`, { encoding: "utf8" }).trim().split(/\s+/).filter(Boolean)) - try { process.kill(Number(pid), "SIGKILL"); } catch {} - } catch {} - } - } - } catch {} -}; - const children = []; const launch = (args, env = {}) => { const c = spawn(process.execPath, args, { env: { ...process.env, ...env }, stdio: ["ignore", "pipe", "pipe"] }); @@ -95,10 +84,8 @@ async function waitUp(url, label, tries = 50) { } (async () => { - // Reclaim any port a prior run left occupied before we try to bind it. - for (const port of [STUB_PORT, 8901, ...Object.keys(instances).map(Number)]) freePort(port); - - launch(["_stub-upstream.mjs"]); + console.log(`ports: ${process.env.CORDON_TEST_PORTS} (stub,dead,main,nosecret,noadmin,restore)`); + launch(["_stub-upstream.mjs"], { STUB_PORT: String(PORTS.stub) }); await waitUp(`${STUB}/__calls`, "stub"); for (const [port, env] of Object.entries(instances)) { diff --git a/_stub-upstream.mjs b/_stub-upstream.mjs index 2af625f..500dd30 100644 --- a/_stub-upstream.mjs +++ b/_stub-upstream.mjs @@ -418,4 +418,4 @@ http res.end("nope"); }); }) - .listen(8900, () => console.log("stub upstream (echo) on :8900")); + .listen(Number(process.env.STUB_PORT || 8900), () => console.log(`stub upstream (echo) on :${process.env.STUB_PORT || 8900}`)); diff --git a/_test_ports.mjs b/_test_ports.mjs new file mode 100644 index 0000000..f7e3a0c --- /dev/null +++ b/_test_ports.mjs @@ -0,0 +1,92 @@ +// The loopback ports the test battery uses, one table for the runner, the stub and every suite. +// The runner picks free ports for each run and hands them down in CORDON_TEST_PORTS, so two +// batteries on one host (two CI jobs of this repository on the same self-hosted machine, say) +// never share a stub or kill each other's instances. Without the variable the historical fixed +// ports apply, so a suite can still be run by hand against a stub and an instance started by hand. +// +// CORDON_TEST_PORTS=stub,dead,main,nosecret,noadmin,restore +// +// stub: the echo upstream; dead: an upstream that accepts and never answers (the proxy suite +// listens on it); main, nosecret, noadmin, restore: the four cordon instances the runner starts. +// The six ports must be distinct: the runner keys its instances by port. +import { closeSync, mkdirSync, openSync, rmSync, writeSync } from "node:fs"; +import { createServer } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const DEFAULTS = { stub: 8900, dead: 8901, main: 8810, nosecret: 8811, noadmin: 8812, restore: 8813 }; +const NAMES = Object.keys(DEFAULTS); + +export function parsePorts(value) { + const ports = { ...DEFAULTS }; + if (!value) return ports; + const given = String(value).split(",").map((x) => Number(x.trim())); + if (given.length !== NAMES.length || given.some((p) => !Number.isInteger(p) || p < 1 || p > 65535) || new Set(given).size !== NAMES.length) + throw new Error(`CORDON_TEST_PORTS must be ${NAMES.length} distinct ports (${NAMES.join(",")}), got "${value}"`); + NAMES.forEach((n, i) => { ports[n] = given[i]; }); + return ports; +} + +export const PORTS = parsePorts(process.env.CORDON_TEST_PORTS); +export const url = (port) => `http://localhost:${port}`; +export const portsEnv = (ports) => NAMES.map((n) => ports[n]).join(","); + +/** + * Host-wide directory where batteries record the ports they own, one `.lock` file each. + * It has to be the same for every battery on the host, so it does not follow a per-job TMPDIR + * on POSIX; CORDON_TEST_PORT_LOCKS overrides it. + */ +export const LOCK_DIR = + process.env.CORDON_TEST_PORT_LOCKS || (process.platform === "win32" ? join(tmpdir(), "cordon-test-ports") : "/tmp/cordon-test-ports"); + +/** A loopback port the kernel reports free right now. Nothing holds it once this resolves. */ +export const probePort = () => + new Promise((resolve, reject) => { + const srv = createServer(); + srv.once("error", reject); + srv.listen(0, "127.0.0.1", () => { + const { port } = srv.address(); + srv.close(() => resolve(port)); + }); + }); + +/** + * Reserve one port per name. A probed port is only taken once its lock file is created with + * O_EXCL, so of two batteries that probe the same port in the window before either binds it, + * exactly one owns it and the other probes again. The locks are held until `release()`, which + * covers the whole run, including the dead port that is only bound when its suite runs. + * A lock left by a battery that was killed is skipped, never reclaimed: deleting another + * process's lock cannot be made atomic with the check that its owner is gone. + */ +export async function reservePorts({ dir = LOCK_DIR, probe = probePort, tries = 200 } = {}) { + mkdirSync(dir, { recursive: true }); + const held = []; + const release = () => { + for (const f of held.splice(0)) try { rmSync(f, { force: true }); } catch {} + }; + const ports = {}; + try { + for (const n of NAMES) { + for (let i = 0; ; i++) { + if (i >= tries) throw new Error(`no free test port for ${n} after ${tries} tries (locks in ${dir})`); + const p = await probe(); + const file = join(dir, `${p}.lock`); + let fd; + try { + fd = openSync(file, "wx"); + } catch (e) { + if (e.code === "EEXIST") continue; + throw e; + } + held.push(file); + try { writeSync(fd, String(process.pid)); } finally { closeSync(fd); } + ports[n] = p; + break; + } + } + } catch (e) { + release(); + throw e; + } + return { ports, release }; +} diff --git a/_test_proxy.mjs b/_test_proxy.mjs index 4eaa448..7c95f03 100644 --- a/_test_proxy.mjs +++ b/_test_proxy.mjs @@ -1,4 +1,4 @@ -// Integration: cordon (:8810) in front of the echo stub (:8900). Proves the model +// Integration: the main cordon instance in front of the echo stub (ports from _test_ports.mjs). Proves the model // never sees raw PII, reversible restores it, strip/off behave, fail-closed blocks, // and the audit log verifies and holds no values. Run with: node _test_proxy.mjs import { createHash } from "node:crypto"; @@ -6,10 +6,11 @@ import { readFileSync } from "node:fs"; import http from "node:http"; import { createServer } from "node:net"; -const BASE = "http://localhost:8810"; -const BASE2 = "http://localhost:8811"; // secret-less instance (pseudonym-without-secret must fail closed) -const BASE3 = "http://localhost:8812"; // no ADMIN_TOKEN, Anthropic upstream that never answers, 500ms timeout -const STUB = "http://localhost:8900"; +import { PORTS, url } from "./_test_ports.mjs"; +const BASE = url(PORTS.main); +const BASE2 = url(PORTS.nosecret); // secret-less instance (pseudonym-without-secret must fail closed) +const BASE3 = url(PORTS.noadmin); // no ADMIN_TOKEN, Anthropic upstream that never answers, 500ms timeout +const STUB = url(PORTS.stub); const ADMIN = "secret"; const PII = "email john@acme.com about card 4012888888881881"; @@ -42,7 +43,7 @@ const asKey = (key) => ({ "x-api-key": key }); const rawPost = (path, body) => new Promise((resolve, reject) => { const req = http.request( - { host: "localhost", port: 8810, path, method: "POST", headers: { "content-type": "application/json", "x-api-key": "test-key" } }, + { host: "localhost", port: PORTS.main, path, method: "POST", headers: { "content-type": "application/json", "x-api-key": "test-key" } }, (res) => { let b = ""; res.on("data", (c) => (b += c)); @@ -364,7 +365,7 @@ const rawPost = (path, body) => // ---- upstream that accepts but never answers: 504 after UPSTREAM_TIMEOUT_MS ---- { const sockets = []; - const hang = createServer((sock) => sockets.push(sock)).listen(8901); + const hang = createServer((sock) => sockets.push(sock)).listen(PORTS.dead); await new Promise((r) => hang.once("listening", r)); const t0 = Date.now(); res = await postTo(BASE3)("/v1/messages", aBody("hi")); @@ -532,7 +533,7 @@ const rawPost = (path, body) => ok("tool/unresolved: a foreign placeholder in prose does not fail the reply", res.status === 200); // CORDON_RESTORE_TOOLS=echo.text on :8813: only echo's `text` comes back real. - const post4 = postTo("http://localhost:8813"); + const post4 = postTo(url(PORTS.restore)); await reset(); res = await post4("/v1/responses", rBody("TOOLCALL " + PII, { tools: TOOL })); { diff --git a/_test_stream.mjs b/_test_stream.mjs index c214c3e..a24b99b 100644 --- a/_test_stream.mjs +++ b/_test_stream.mjs @@ -1,8 +1,9 @@ // Integration: streaming re-identification. The stub chunks its echoed reply into // 3-char SSE frames, so placeholders like are SPLIT across frame // boundaries — the critical hold-back-buffer case. Run with: node _test_stream.mjs -const BASE = "http://localhost:8810"; -const STUB = "http://localhost:8900"; +import { PORTS, url } from "./_test_ports.mjs"; +const BASE = url(PORTS.main); +const STUB = url(PORTS.stub); const PII = "email john@acme.com about card 4012888888881881"; let pass = 0, fail = 0; @@ -476,7 +477,7 @@ function validateSSE(sse) { // CORDON_RESTORE_TOOLS=echo.text on :8813, streamed: same per-field decision as whole. { const post4 = (path, body) => - fetch("http://localhost:8813" + path, { method: "POST", headers: { "content-type": "application/json", "x-api-key": "test-key", "anthropic-version": "2023-06-01" }, body: JSON.stringify(body) }); + fetch(url(PORTS.restore) + path, { method: "POST", headers: { "content-type": "application/json", "x-api-key": "test-key", "anthropic-version": "2023-06-01" }, body: JSON.stringify(body) }); await reset(); txt = await (await post4("/v1/responses", respTool())).text(); const f = frames(txt); diff --git a/_test_subject_proxy.mjs b/_test_subject_proxy.mjs index 8462285..127780c 100644 --- a/_test_subject_proxy.mjs +++ b/_test_subject_proxy.mjs @@ -1,4 +1,4 @@ -// Integration for known subjects: cordon (:8810) in front of the echo stub (:8900). Proves the +// Integration for known subjects: the main cordon instance in front of the echo stub (ports from _test_ports.mjs). Proves the // model never sees a subject's identifiers, the reply comes back real, a request the egress // check or the media rule refuses never reaches the stub, report-only forwards unchanged and // says so, and receipts are served per ref with no values. Every person here is invented. @@ -6,8 +6,9 @@ import { createHash } from "node:crypto"; import { readFileSync } from "node:fs"; -const BASE = "http://localhost:8810"; -const STUB = "http://localhost:8900"; +import { PORTS, url } from "./_test_ports.mjs"; +const BASE = url(PORTS.main); +const STUB = url(PORTS.stub); const ADMIN = "secret"; let pass = 0, fail = 0; const ok = (n, c, e = "") => { c ? pass++ : fail++; console.log(`${c ? "PASS" : "FAIL"} ${n}${e ? " (" + e + ")" : ""}`); }; @@ -426,9 +427,9 @@ const leaked = (s) => RAW.filter((r) => s.includes(r)); // :8811 has no TENANT_SECRET, so a consistent-pseudonym tenant fails at the secret guard; with // failMode open that would forward the raw body, but never for a request that names subjects. - await fetch("http://localhost:8811/admin/tenant", { method: "POST", headers: { "content-type": "application/json", "x-admin-token": ADMIN }, body: JSON.stringify({ tenant: keyTenant("open-key"), consistentPseudonyms: true, failMode: "open" }) }); + await fetch(url(PORTS.nosecret) + "/admin/tenant", { method: "POST", headers: { "content-type": "application/json", "x-admin-token": ADMIN }, body: JSON.stringify({ tenant: keyTenant("open-key"), consistentPseudonyms: true, failMode: "open" }) }); await reset(); - res = await fetch("http://localhost:8811/v1/messages", { + res = await fetch(url(PORTS.nosecret) + "/v1/messages", { method: "POST", headers: { "content-type": "application/json", "x-api-key": "open-key", "anthropic-version": "2023-06-01", "x-cordon-subject": SUBJECT() }, body: JSON.stringify(aBody(TEXT)), diff --git a/_test_unit.mjs b/_test_unit.mjs index 924cf0a..fd2915e 100644 --- a/_test_unit.mjs +++ b/_test_unit.mjs @@ -950,6 +950,41 @@ ok("timeout: empty, zero, negative and Infinity -> default", ok("resolveTenant: no credential is the public tenant", resolveTenant({}) === "public"); } +// ---- test battery ports: CORDON_TEST_PORTS ---- +{ + const { parsePorts } = await import("./_test_ports.mjs"); + const throws = (v) => { try { parsePorts(v); return false; } catch { return true; } }; + ok("parsePorts: six distinct ports are taken in order", JSON.stringify(parsePorts("10000,10001,10002,10003,10004,10005")) === JSON.stringify({ stub: 10000, dead: 10001, main: 10002, nosecret: 10003, noadmin: 10004, restore: 10005 })); + ok("parsePorts: unset falls back to the fixed ports", parsePorts("").main === 8810 && parsePorts(undefined).stub === 8900); + ok("parsePorts: a port named twice is refused", throws("10000,10001,10002,10002,10004,10005")); + ok("parsePorts: the wrong count is refused", throws("10000,10001,10002")); + ok("parsePorts: an out-of-range port is refused", throws("10000,10001,10002,10003,10004,70000")); +} + +// ---- test battery ports: host-wide reservation ---- +{ + const { reservePorts } = await import("./_test_ports.mjs"); + const { mkdtempSync, readdirSync } = await import("node:fs"); + const { tmpdir } = await import("node:os"); + const { join } = await import("node:path"); + const dir = mkdtempSync(join(tmpdir(), "cordon-ports-")); + // Two batteries can be handed the same free port before either binds it: this probe returns + // every port twice, to whichever reservation asks next. + let n = 0; + const probe = async () => { await new Promise((r) => setImmediate(r)); return 20000 + Math.floor(n++ / 2); }; + const [a, b] = await Promise.all([reservePorts({ dir, probe }), reservePorts({ dir, probe })]); + const pa = Object.values(a.ports), pb = Object.values(b.ports); + ok("reservePorts: two simultaneous reservations never share a port", pa.length === 6 && pb.length === 6 && new Set([...pa, ...pb]).size === 12); + ok("reservePorts: every reserved port stays locked until released", readdirSync(dir).length === 12); + a.release(); + const seq = (from) => { let i = from; return async () => i++; }; + const c = await reservePorts({ dir, probe: seq(Math.min(...pb)) }); + ok("reservePorts: a port another run still holds is skipped", Object.values(c.ports).every((p) => !pb.includes(p))); + b.release(); c.release(); + ok("reservePorts: release frees the locks", readdirSync(dir).length === 0); + rmSync(dir, { recursive: true, force: true }); +} + console.log(`\n${pass} passed, ${fail} failed`); process.exit(fail ? 1 : 0);