From 572e35bbbe9a93520116522e64a226cfe3c62744 Mon Sep 17 00:00:00 2001 From: Adam Kunicki Date: Tue, 3 Nov 2026 13:56:22 -0800 Subject: [PATCH] feat(sync): complete encrypted relay sync and readiness --- .../src/relay-peer-client-test-fixtures.ts | 300 ++++++++ packages/core/src/relay-peer-client.test.ts | 710 ++++++++++++++++++ 2 files changed, 1010 insertions(+) create mode 100644 packages/core/src/relay-peer-client-test-fixtures.ts create mode 100644 packages/core/src/relay-peer-client.test.ts diff --git a/packages/core/src/relay-peer-client-test-fixtures.ts b/packages/core/src/relay-peer-client-test-fixtures.ts new file mode 100644 index 000000000..257d2ebdd --- /dev/null +++ b/packages/core/src/relay-peer-client-test-fixtures.ts @@ -0,0 +1,300 @@ +import { createPrivateKey } from "node:crypto"; +import { createServer, type Server } from "node:http"; +import { performance } from "node:perf_hooks"; +import Database from "better-sqlite3"; +import { Hono } from "hono"; +import { vi } from "vitest"; +import WebSocket from "ws"; +import { ADMISSION_BUDGET_SCHEMA_STATEMENTS } from "./admission-budget-schema.js"; +import { registerCoordinatorRelayAdmissionRoute } from "./coordinator-relay-admission-route.js"; +import { decisionFixture, decisionSeed } from "./relay-admission-decision-test-fixtures.js"; +import { RELAY_ADMISSION_METADATA_SCHEMA_STATEMENTS } from "./relay-admission-metadata-schema.js"; +import { PARTICIPANT_TEST_SCHEMA_SQL } from "./relay-admission-participant-nonce-test-fixtures.js"; +import { createRelayCoordinatorClient } from "./relay-coordinator-client.js"; +import { + createRelayPeerClient, + type RelayPeerClient, + type RelayPeerClientOptions, +} from "./relay-peer-client.js"; +import { + createSqliteRelayQuota, + initializeRelayQuotaSchema, + provisionLocalRelayQuota, + setRelayQuotaEnabled, +} from "./relay-quota-sqlite.js"; +import type { RelayServerOptions } from "./relay-server.js"; +import { finiteLimits, testSigner } from "./relay-server-test-fixtures.js"; +import { relayAudienceId, relayRealmId } from "./relay-service-proof.js"; +import { createRelayWebSocketRuntime } from "./relay-websocket-runtime.js"; +import { runtimeLimits } from "./relay-websocket-runtime-test-fixtures.js"; + +const bind = (values: readonly (string | number | null)[]) => + Object.fromEntries(values.map((v, i) => [i + 1, v])); +export const admissionPath = "/trusted/v1/relay/admission-check"; +export const relayDeadline = () => performance.now() + 8000; +export function observeRelayWrites() { + const spy = vi.spyOn(WebSocket.prototype, "send"); + return () => spy.mock.calls.length; +} +export function disposableIdentity(seed: number) { + return { + key: createPrivateKey({ + key: Buffer.concat([ + Buffer.from("302e020100300506032b657004220420", "hex"), + Buffer.alloc(32, seed), + ]), + format: "der", + type: "pkcs8", + }), + expectedPin: testSigner(seed).publicKey, + }; +} +async function listen(server: Server) { + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing loopback listener"); + return `127.0.0.1:${address.port}`; +} +async function closeServer(server: Server) { + await new Promise((resolve) => { + server.close(() => resolve()); + server.closeAllConnections(); + }); +} + +type Identities = Record<"source" | "target", ReturnType>; +type DeviceIds = Record<"source" | "target", string>; +type Namespace = RelayServerOptions["namespace"]; +export interface RelayClientTestOptions { + readonly monotonicMs?: () => number; + readonly beforeCheck?: ( + options: Parameters[0], + ) => Promise; +} + +async function admissionDatabase( + identities: Identities, + deviceIds: DeviceIds, + namespace: Namespace, +): Promise { + const template = await decisionFixture({ createdAt: Math.floor(Date.now() / 1000) }); + template.request.namespace = namespace; + template.rows = template.rows.map((row) => ({ + ...row, + device_id: deviceIds[row.device_id === "source" ? "source" : "target"], + public_key: identities[row.device_id === "source" ? "source" : "target"].expectedPin, + })); + const db = new Database(":memory:"); + db.exec(PARTICIPANT_TEST_SCHEMA_SQL); + for (const sql of [ + ...RELAY_ADMISSION_METADATA_SCHEMA_STATEMENTS, + ...ADMISSION_BUDGET_SCHEMA_STATEMENTS, + ]) + db.exec(sql); + for (const { sql, values } of decisionSeed(template, Date.now(), { + audience: namespace.relayAudienceId, + })) + db.prepare(sql).run(bind(values)); + const allocate = (pool: string, audience = "", group = "", rate = 1, burst = 16) => + db + .prepare( + "INSERT INTO admission_budget_allocations(authority_id,pool,audience_id,group_id,rate_per_min,burst) VALUES (?1,?2,?3,?4,?5,?6)", + ) + .run(bind([namespace.authorityId, pool, audience, group, rate, burst])); + allocate("authority", "", "", 4, 64); + allocate("device"); + allocate("device-registration", "", "", 1, 2); + for (const pool of ["relay-route", "relay-registration"]) { + allocate(pool, namespace.relayAudienceId); + allocate(pool, namespace.relayAudienceId, namespace.groupId); + } + return db; +} + +function admissionHttp(db: Database.Database, namespace: Namespace) { + const child = new Hono(); + registerCoordinatorRelayAdmissionRoute(child, { + namespace, + externalPathname: admissionPath, + ingressKey: "disposable-test", + open: () => ({ + database: { + query: async (sql, values) => db.prepare(sql).all(bind(values)), + insertReturning: async (sql, values) => db.prepare(sql).get(bind(values)) ?? null, + }, + close() {}, + }), + }); + const app = new Hono().route("/trusted", child); + const http = createServer(async (incoming, outgoing) => { + const chunks: Buffer[] = []; + for await (const chunk of incoming) chunks.push(Buffer.from(chunk)); + const headers = new Headers(); + for (const [name, value] of Object.entries(incoming.headers)) + if (value !== undefined) headers.set(name, Array.isArray(value) ? value.join(",") : value); + const response = await app.fetch( + new Request(`http://127.0.0.1${incoming.url}`, { + method: incoming.method, + headers, + body: new Uint8Array(Buffer.concat(chunks)), + }), + ); + outgoing.writeHead(response.status, Object.fromEntries(response.headers)); + outgoing.end(Buffer.from(await response.arrayBuffer())); + }); + return http; +} + +async function coordinatorFixture(http: Server, options: RelayClientTestOptions) { + const coordinatorBase = createRelayCoordinatorClient({ + url: `http://${await listen(http)}${admissionPath}`, + allowLoopbackHttp: true, + }); + const checks: Parameters[0][] = []; + const checkTimes: number[] = []; + const coordinator = { + check: async (request: Parameters[0]) => { + checks.push(request); + checkTimes.push(performance.now()); + await options.beforeCheck?.(request); + return coordinatorBase.check(request); + }, + }; + return { coordinator, checks, checkTimes }; +} + +function quotaFixture(namespace: Namespace) { + const quotaDb = new Database(":memory:"); + initializeRelayQuotaSchema(quotaDb); + const realmId = relayRealmId(namespace.authorityId, namespace.groupId); + const control = { ratePerMinute: 120, burst: 16 }, + traffic = { ratePerMinute: 16 * 1024 * 1024, burst: 16 * 1024 * 1024 }; + provisionLocalRelayQuota(quotaDb, { + realmId, + profile: { + upgrade: { service: control, realm: control }, + control: { service: control, realm: control, device: control }, + open: { + service: { ratePerMinute: 30, burst: 4 }, + realm: { ratePerMinute: 30, burst: 4 }, + device: { ratePerMinute: 30, burst: 4 }, + pair: { ratePerMinute: 20, burst: 2 }, + }, + traffic: { service: traffic, realm: traffic, device: traffic, pair: traffic }, + }, + }); + setRelayQuotaEnabled(quotaDb, { realmId, enabled: true }); + const quota = createSqliteRelayQuota({ database: quotaDb, realmId, enabled: true }); + return { + quotaDb: quotaDb as Database.Database, + quota, + }; +} + +function clientFactory( + base: Omit, + identities: Identities, + deviceIds: DeviceIds, +) { + const clients: RelayPeerClient[] = []; + function client(deviceId: "source" | "target", overrides: Partial = {}) { + const options: RelayPeerClientOptions = { + ...base, + deviceId: deviceIds[deviceId], + localIdentity: identities[deviceId], + resolvePeerPin: (id) => { + if (id === deviceIds.source) return identities.source.expectedPin; + if (id === deviceIds.target) return identities.target.expectedPin; + return null; + }, + ...overrides, + }; + const result = createRelayPeerClient(options); + clients.push(result); + return result; + } + return { clients, client }; +} + +function runtimeFixture( + server: Server, + namespace: Namespace, + coordinator: RelayServerOptions["coordinator"], + quota: ReturnType, + options: RelayClientTestOptions, +) { + return createRelayWebSocketRuntime({ + server, + enabled: true, + relay: { + namespace, + signer: testSigner(3), + coordinator, + quota, + limits: finiteLimits, + clock: { + monotonicMs: options.monotonicMs ?? (() => performance.now()), + utcMs: () => Date.now(), + }, + }, + limits: runtimeLimits, + reserveUpgrade: (signal) => quota.reserveUpgrade(signal), + }); +} + +/** Real local protocol participants with disposable identities and separate quota/admission databases. */ +export async function relayClientFixture( + identities: Identities = { source: disposableIdentity(1), target: disposableIdentity(2) }, + deviceIds: DeviceIds = { source: "source", target: "target" }, + options: RelayClientTestOptions = {}, +) { + const service = testSigner(3); + const socketServer = createServer(); + const host = await listen(socketServer); + const namespace = { + authorityId: "a".repeat(64), + groupId: "group-A", + relayAudienceId: relayAudienceId(`wss://${host}`, service.keyId), + }; + const db = await admissionDatabase(identities, deviceIds, namespace); + const http = admissionHttp(db, namespace); + const { coordinator, checks, checkTimes } = await coordinatorFixture(http, options); + const { quotaDb, quota } = quotaFixture(namespace); + const runtime = runtimeFixture(socketServer, namespace, coordinator, quota, options); + const { clients, client } = clientFactory( + { + enabled: true, + url: `ws://${host}/v1/relay`, + allowLoopbackWs: true, + servicePublicKey: service.publicKey, + ...namespace, + externalPathname: admissionPath, + coordinator, + resolvePeerPin: () => null, + validateLocal: () => true, + onInbound: async (connection) => { + await connection.dispose(new Error("No inbound handler")); + }, + }, + identities, + deviceIds, + ); + return { + client, + identities, + db: db as Database.Database, + quotaDb: quotaDb as Database.Database, + runtime, + checks, + checkTimes, + namespace, + async dispose() { + await Promise.all(clients.map((client) => client.close())); + await runtime.close(); + await closeServer(socketServer); + await closeServer(http); + quotaDb.close(); + db.close(); + }, + }; +} diff --git a/packages/core/src/relay-peer-client.test.ts b/packages/core/src/relay-peer-client.test.ts new file mode 100644 index 000000000..fea7cd722 --- /dev/null +++ b/packages/core/src/relay-peer-client.test.ts @@ -0,0 +1,710 @@ +import { once } from "node:events"; +import { performance } from "node:perf_hooks"; +import { setImmediate as nextTurn } from "node:timers/promises"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import WebSocket from "ws"; +import { createPeerHttpExecutor } from "./peer-http-client.js"; +import type { PeerHttpServerRequest } from "./peer-http-server.js"; +import { decodeRelayCheckPacket } from "./relay-admission-codec.js"; +import { decodeRelayDataFrame, encodeRelayDataFrame } from "./relay-data-frame.js"; +import type { RelayPeerClientOptions } from "./relay-peer-client.js"; +import { + disposableIdentity, + relayClientFixture, + relayDeadline, +} from "./relay-peer-client-test-fixtures.js"; +import { relayBodyDigest } from "./relay-service-proof.js"; + +const cleanups: (() => Promise)[] = []; +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); + vi.restoreAllMocks(); +}); +async function fixture(options?: Parameters[2]) { + const f = await relayClientFixture(undefined, undefined, options); + cleanups.push(() => f.dispose()); + return f; +} + +describe("actual relay peer HTTP exchange", () => { + it("authenticates both clients and completes one HTTP exchange with three independent admission roles", async () => { + // Arrange: the dispatcher sees application bytes only after real Noise authentication. + const f = await fixture(); + const wire = vi.spyOn(WebSocket.prototype, "send"); + const dispatch = vi.fn(async (request: PeerHttpServerRequest) => ({ + status: 200, + bodyBytes: Buffer.from(JSON.stringify({ target: request.target, pin: request.peerPin })), + })); + const completions: Promise[] = []; + const target = f.client("target", { + onInbound: async (connection, context) => { + completions.push(connection.completion); + await context.serve({ origin: "https://peer.invalid", dispatch }); + }, + }); + const source = f.client("source"); + await target.connect(new AbortController().signal, relayDeadline()); + const execute = createPeerHttpExecutor({ + factory: source.factory("target", f.identities.target.expectedPin), + }); + // Act. + const result = await execute("GET", "https://peer.invalid/private-marker?x=1"); + await Promise.all(completions); + // Assert: success includes physical route completion, not merely HTTP parsing. + expect(result).toEqual([ + 200, + { target: "/private-marker?x=1", pin: f.identities.source.expectedPin }, + ]); + expect(dispatch).toHaveBeenCalledTimes(1); + expect( + f.checks + .map(({ packet }) => decodeRelayCheckPacket(packet)) + .filter((p) => p.kind === "admission") + .map((p) => p.value.role) + .sort(), + ).toEqual(["relay", "responder", "source"]); + expect(f.runtime.counters().routes).toBe(0); + const frames = wire.mock.calls + .map(([data]) => data) + .filter((data): data is Uint8Array => data instanceof Uint8Array) + .map(decodeRelayDataFrame); + const opening = frames[0]?.payload; + expect(opening?.byteLength).toBe(34); + expect(frames.length).toBeGreaterThan(3); + for (const frame of frames) + expect(Buffer.from(frame.payload).toString()).not.toContain("private-marker"); + const packets = f.checks.filter( + ({ packet }) => decodeRelayCheckPacket(packet).kind === "admission", + ); + expect(new Set(packets.map(({ packet }) => packet.nonce)).size).toBe(3); + if (!opening) throw new Error("Missing Noise opening"); + for (const { packet } of packets) { + const decoded = decodeRelayCheckPacket(packet); + if (decoded.kind !== "admission") throw new Error("Expected admission packet"); + expect(decoded.value.messageADigest).toBe(relayBodyDigest(opening)); + } + }, 10000); + + it("fails an offline target once without dispatch or retry", async () => { + // Arrange. + const f = await fixture(); + const source = f.client("source"); + const execute = createPeerHttpExecutor({ + factory: source.factory("target", f.identities.target.expectedPin), + }); + // Act. + const result = execute("POST", "https://peer.invalid/private-marker", { + bodyBytes: Buffer.from("{}"), + }); + // Assert. + await expect(result).rejects.toThrow(); + expect( + f.checks + .map(({ packet }) => decodeRelayCheckPacket(packet)) + .filter((p) => p.kind === "admission") + .map((p) => p.value.role) + .sort(), + ).toEqual(["relay", "source"]); + expect(f.runtime.counters().routes).toBe(0); + }, 10000); + + it.each(["disabled", "stale-pin", "cancelled"])( + "rejects %s before any network admission", + async (condition) => { + // Arrange. + const f = await fixture(); + const source = f.client("source", { + enabled: condition === "disabled" ? undefined : true, + resolvePeerPin: + condition === "stale-pin" + ? () => null + : (id) => (id === "target" ? f.identities.target.expectedPin : null), + }); + const abort = new AbortController(); + if (condition === "cancelled") abort.abort(); + // Act. + const result = source + .factory("target", f.identities.target.expectedPin) + .open({ signal: abort.signal, deadlineMs: relayDeadline() }); + // Assert. + await expect(result).rejects.toThrow(); + expect(f.checks).toEqual([]); + expect(f.runtime.counters().sockets).toBe(0); + }, + ); +}); + +describe("relay peer trust and independent admission", () => { + it.each(["source", "responder"])( + "requires the %s independent coordinator check before application dispatch", + async (role) => { + // Arrange: registration and relay checks still use the real HTTP coordinator. + const f = await fixture(); + const deny = vi.fn(async () => null); + const dispatch = vi.fn(async () => ({ status: 200, bodyBytes: Buffer.from("{}") })); + const target = f.client("target", { + ...(role === "responder" ? { coordinator: { check: deny } } : {}), + onInbound: async (_connection, context) => + context.serve({ origin: "https://peer.invalid", dispatch }), + }); + const source = f.client("source", role === "source" ? { coordinator: { check: deny } } : {}); + await target.connect(new AbortController().signal, relayDeadline()); + const execute = createPeerHttpExecutor({ + factory: source.factory("target", f.identities.target.expectedPin), + }); + // Act. + const result = execute("POST", "https://peer.invalid/private-marker", { + bodyBytes: Buffer.from("{}"), + }); + // Assert. + await expect(result).rejects.toThrow(); + expect(deny).toHaveBeenCalledTimes(1); + expect(dispatch).not.toHaveBeenCalled(); + expect(f.runtime.counters().routes).toBe(0); + }, + 10000, + ); + + it("revalidates the current source pin immediately before shared HTTP dispatch", async () => { + // Arrange: invalidate trust only after Noise authenticated the inbound connection. + const f = await fixture(); + let current = true; + const dispatch = vi.fn(async () => ({ status: 200, bodyBytes: Buffer.from("{}") })); + const inbound = vi.fn(async (_connection, context) => { + current = false; + await context.serve({ origin: "https://peer.invalid", dispatch }); + }); + const target = f.client("target", { + resolvePeerPin: () => (current ? f.identities.source.expectedPin : null), + onInbound: inbound, + }); + const source = f.client("source"); + await target.connect(new AbortController().signal, relayDeadline()); + // Act. + const result = createPeerHttpExecutor({ + factory: source.factory("target", f.identities.target.expectedPin), + })("POST", "https://peer.invalid/private-marker", { bodyBytes: Buffer.from("{}") }); + // Assert. + await expect(result).rejects.toThrow(); + expect(inbound).toHaveBeenCalledTimes(1); + expect(dispatch).not.toHaveBeenCalled(); + }, 10000); + + it.each(["rotated-key", "stale-readiness"])( + "fails registration against actual coordinator %s state", + async (condition) => { + // Arrange. + const f = await fixture(); + if (condition === "rotated-key") + f.db + .prepare("UPDATE enrolled_devices SET public_key = ? WHERE device_id = 'source'") + .run(f.identities.target.expectedPin); + else { + for (const scope of ["authority", "realm"]) + f.db + .prepare( + "UPDATE relay_admission_readiness SET revision = revision + 1, observed_at_ms = ? WHERE scope = ?", + ) + .run(Date.now() - 120000, scope); + } + const source = f.client("source"); + // Act. + const result = source.connect(new AbortController().signal, relayDeadline()); + // Assert. + await expect(result).rejects.toThrow(); + expect(f.checks).toHaveLength(1); + expect(f.runtime.counters().routes).toBe(0); + }, + 10000, + ); +}); + +describe("relay peer initial registration cancellation", () => { + it("cancels one peer's initial opener without closing shared registration for another peer", async () => { + // Arrange: different peers create independent routes, rather than a same-peer capacity waiter. + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const f = await fixture({ + beforeCheck: async ({ packet }) => { + const decoded = decodeRelayCheckPacket(packet); + if (decoded.kind === "registration" && decoded.value.deviceId === "source") { + entered.resolve(); + await release.promise; + } + }, + }); + cleanups.push(async () => release.resolve()); + const dispatch = vi.fn(async () => ({ + status: 200, + bodyBytes: Buffer.from('{"survived":true}'), + })); + const target = f.client("target", { + onInbound: async (_connection, context) => + context.serve({ origin: "https://peer.invalid", dispatch }), + }); + const otherPin = disposableIdentity(4).expectedPin; + const source = f.client("source", { + resolvePeerPin: (id) => { + if (id === "cancelled-peer") return otherPin; + if (id === "target") return f.identities.target.expectedPin; + return null; + }, + }); + await target.connect(new AbortController().signal, relayDeadline()); + const wire = vi.spyOn(WebSocket.prototype, "send"); + const registration = vi.spyOn(source, "maintainRegistration"); + const abort = new AbortController(); + const first = source.factory("cancelled-peer", otherPin).open({ + signal: abort.signal, + deadlineMs: relayDeadline(), + }); + const cancelled = expect(first).rejects.toThrow(/cancelled|expired|removed/); + await entered.promise; + const second = source.factory("target", f.identities.target.expectedPin).open({ + signal: new AbortController().signal, + deadlineMs: relayDeadline(), + }); + void second.catch(() => {}); + await nextTurn(); + const state = source as unknown as { + socket: WebSocket; + routes: Map; + waiters: Map; + }; + const socket = state.socket; + expect(registration).toHaveBeenCalledTimes(2); + expect([...state.routes.values()].map(({ peer, opened }) => ({ peer, opened }))).toEqual([ + { peer: "cancelled-peer", opened: false }, + { peer: "target", opened: false }, + ]); + expect(state.waiters.size).toBe(0); + // Act: cancellation belongs to the first route, while both await the same registration. + abort.abort(); + await cancelled; + await nextTurn(); + // Assert: an unopened route must not send CLOSE during the server's pending REGISTER check. + const prematureCloses = wire.mock.calls.filter( + ([message], index) => + wire.mock.contexts[index] === socket && + typeof message === "string" && + JSON.parse(message).type === "CLOSE", + ); + expect(prematureCloses).toEqual([]); + release.resolve(); + const owner = await second; + expect(state.socket).toBe(socket); + await expect( + createPeerHttpExecutor({ factory: { open: async () => owner } })( + "GET", + "https://peer.invalid/after-cancellation", + ), + ).resolves.toEqual([200, { survived: true }]); + await expect(owner.completion).resolves.toBeUndefined(); + expect(dispatch).toHaveBeenCalledTimes(1); + expect( + f.checks.filter(({ packet }) => { + const decoded = decodeRelayCheckPacket(packet); + return decoded.kind === "registration" && decoded.value.deviceId === "source"; + }), + ).toHaveLength(1); + }, 10000); +}); + +describe("relay peer crossed cancellation", () => { + it("keeps a reverse route usable when forwarded DATA crosses local cancellation", async () => { + // Arrange: both directions authenticate before holding the forward response on the wire. + const f = await fixture(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const forwarded = Promise.withResolvers(); + cleanups.push(async () => release.resolve()); + const target = f.client("target", { + onInbound: async (_connection, context) => + context.serve({ + origin: "https://peer.invalid", + dispatch: async () => { + entered.resolve(); + await release.promise; + return { status: 200, bodyBytes: Buffer.from('{"cancelled":"marker"}') }; + }, + }), + }); + const source = f.client("source", { + onInbound: async (_connection, context) => + context.serve({ + origin: "https://peer.invalid", + dispatch: async () => ({ status: 200, bodyBytes: Buffer.from('{"reverse":true}') }), + }), + }); + const send = WebSocket.prototype.send; + const wire = vi.spyOn(WebSocket.prototype, "send"); + await target.connect(new AbortController().signal, relayDeadline()); + const abort = new AbortController(); + const owner = await source.factory("target", f.identities.target.expectedPin).open({ + signal: abort.signal, + deadlineMs: relayDeadline(), + }); + const reverse = await target.factory("source", f.identities.source.expectedPin).open({ + signal: new AbortController().signal, + deadlineMs: relayDeadline(), + }); + const state = source as unknown as { + socket: WebSocket; + routes: Map; + }; + const session = [...state.routes].find(([, route]) => route.side === "source")?.[0]; + const index = wire.mock.calls.findIndex( + ([data]) => + typeof data === "string" && data.includes('"REGISTERED"') && data.includes('"source"'), + ); + const relaySocket = wire.mock.contexts[index]; + if (!session || !relaySocket) throw new Error("Missing forward route or relay socket"); + wire.mockImplementation(function (this: WebSocket, ...args: Parameters) { + const result = send.apply(this, args); + if ( + this === relaySocket && + args[0] instanceof Uint8Array && + decodeRelayDataFrame(args[0]).sessionId === session + ) + forwarded.resolve(); + return result; + }); + const delivered = vi.fn(); + const response = createPeerHttpExecutor({ factory: { open: async () => owner } })( + "GET", + "https://peer.invalid/cancelled", + ); + const rejected = expect(response).rejects.toThrow(); + await entered.promise; + owner.duplex.on("data", delivered); + const socket = state.socket; + socket.pause(); + cleanups.push(async () => socket.resume()); + // Act: DATA is already forwarded, but local cancellation removes its route before receive. + release.resolve(); + await forwarded.promise; + abort.abort(); + await rejected; + await expect(owner.completion).rejects.toThrow(); + expect(state.routes.has(session)).toBe(false); + socket.resume(); + // Assert: cancelled plaintext never reaches the app, and the unrelated route still works. + const result = createPeerHttpExecutor({ factory: { open: async () => reverse } })( + "GET", + "https://peer.invalid/reverse", + ); + expect(delivered).not.toHaveBeenCalled(); + await expect(result).resolves.toEqual([200, { reverse: true }]); + await expect(reverse.completion).resolves.toBeUndefined(); + }, 10000); +}); + +describe("relay peer retired session capacity", () => { + it.each(["DATA", "ACCEPTED"])( + "ignores crossed %s without evicting old IDs or recycling a busy socket", + async (type) => { + // Arrange: keep a reverse exchange alive while retiring the forward route at the cap. + const f = await fixture(); + const wire = vi.spyOn(WebSocket.prototype, "send"); + const source = f.client("source", { + onInbound: async (_connection, context) => + context.serve({ + origin: "https://peer.invalid", + dispatch: async () => ({ status: 200, bodyBytes: Buffer.from("{}") }), + }), + }); + const target = f.client("target", { onInbound: async (connection) => connection.completion }); + await target.connect(new AbortController().signal, relayDeadline()); + const forward = await source.factory("target", f.identities.target.expectedPin).open({ + signal: new AbortController().signal, + deadlineMs: relayDeadline(), + }); + const reverse = await target.factory("source", f.identities.source.expectedPin).open({ + signal: new AbortController().signal, + deadlineMs: relayDeadline(), + }); + const state = source as unknown as { + socket: WebSocket; + routes: Map; + retiredRoutes: Set; + }; + const session = [...state.routes].find(([, route]) => route.side === "source")?.[0]; + const index = wire.mock.calls.findIndex( + ([data]) => + typeof data === "string" && data.includes('"REGISTERED"') && data.includes('"source"'), + ); + const relaySocket = wire.mock.contexts[index]; + if (!session || !relaySocket) throw new Error("Missing forward route or relay socket"); + const socket = state.socket; + const oldest = "0".repeat(32); + // Model earlier sessions without thousands of Noise handshakes. + for (let id = 0; id < 4094; id++) state.retiredRoutes.add(id.toString(16).padStart(32, "0")); + // Act: retire one route, then deliver frames that crossed its cancellation. + await forward.dispose(new Error("Local cancellation")); + expect(state.retiredRoutes.has(session)).toBe(true); + expect(state.retiredRoutes.size).toBe(4095); + await expect( + source.factory("target", f.identities.target.expectedPin).open({ + signal: new AbortController().signal, + deadlineMs: relayDeadline(), + }), + ).rejects.toThrow("capacity exhausted"); + const received = once(socket, "message"); + relaySocket.send( + type === "ACCEPTED" + ? JSON.stringify({ v: 1, type: "ACCEPTED", session, body: {} }) + : encodeRelayDataFrame({ sessionId: oldest, payload: Buffer.from("cancelled bytes") }), + ); + await received; + // Assert: history and the unrelated route survive until that last exchange completes. + expect(state.retiredRoutes.has(oldest)).toBe(true); + expect(state.retiredRoutes.has(session)).toBe(true); + expect(socket.readyState).toBe(WebSocket.OPEN); + expect(reverse.duplex.destroyed).toBe(false); + const closed = once(socket, "close"); + const relayClosed = once(relaySocket, "close"); + await expect( + createPeerHttpExecutor({ factory: { open: async () => reverse } })( + "GET", + "https://peer.invalid/at-cap", + ), + ).resolves.toEqual([200, {}]); + await expect(reverse.completion).resolves.toBeUndefined(); + await closed; + await relayClosed; + expect(state.routes.size).toBe(0); + expect(state.retiredRoutes.size).toBe(0); + // Replenish the fixture's two-upgrade address burst without a real-time sleep. + const runtime = f.runtime as unknown as { addresses: Map }; + for (const allowance of runtime.addresses.values()) allowance.tokens = 1; + await source.connect(new AbortController().signal, relayDeadline()); + expect(state.socket).not.toBe(socket); + expect(state.socket.readyState).toBe(WebSocket.OPEN); + expect(state.retiredRoutes.size).toBe(0); + }, + 10000, + ); +}); + +describe("relay peer active route cleanup", () => { + it("rejects unknown-session DATA instead of delivering it to an authenticated application", async () => { + // Arrange: retain one real authenticated connection without sending an HTTP request. + const f = await fixture(); + const wire = vi.spyOn(WebSocket.prototype, "send"); + const target = f.client("target", { onInbound: async (connection) => connection.completion }); + const source = f.client("source"); + await target.connect(new AbortController().signal, relayDeadline()); + const owner = await source + .factory("target", f.identities.target.expectedPin) + .open({ signal: new AbortController().signal, deadlineMs: relayDeadline() }); + const index = wire.mock.calls.findIndex( + ([data]) => + typeof data === "string" && data.includes('"REGISTERED"') && data.includes('"source"'), + ); + const socket = wire.mock.contexts[index]; + if (!socket) throw new Error("Missing source relay socket"); + const closed = once(socket, "close"); + // Act. + socket.send( + encodeRelayDataFrame({ + sessionId: "0".repeat(32), + payload: Buffer.from("unauthenticated marker"), + }), + ); + // Assert. + await expect(owner.completion).rejects.toThrow(); + await closed; + expect(owner.duplex.readableLength).toBe(0); + }, 10000); + + it("allows only one active source route and one waiting attempt per peer, then cancels the waiter", async () => { + // Arrange. + const f = await fixture(); + const target = f.client("target", { onInbound: async (connection) => connection.completion }); + const source = f.client("source"); + await target.connect(new AbortController().signal, relayDeadline()); + const factory = source.factory("target", f.identities.target.expectedPin); + const owner = await factory.open({ + signal: new AbortController().signal, + deadlineMs: relayDeadline(), + }); + const abort = new AbortController(); + const waiting = factory.open({ signal: abort.signal, deadlineMs: relayDeadline() }); + // Act. + const excess = factory.open({ + signal: new AbortController().signal, + deadlineMs: relayDeadline(), + }); + // Assert. + await expect(excess).rejects.toThrow("capacity exhausted"); + abort.abort(); + await expect(waiting).rejects.toThrow("cancelled or expired"); + expect( + f.checks.filter(({ packet }) => decodeRelayCheckPacket(packet).kind === "admission"), + ).toHaveLength(3); + await owner.dispose(new Error("Acceptance test complete")); + }, 10000); +}); + +describe("relay peer pair pacing", () => { + it.each(["cancel", "expire"])( + "settles a %s during pacing without new admission proofs, OPEN or DATA", + async (mode) => { + // Arrange: a completed real exchange starts the client's six-second pair interval. + const f = await fixture(); + const wire = vi.spyOn(WebSocket.prototype, "send"); + const completions: Promise[] = []; + const target = f.client("target", { + onInbound: async (connection, context) => { + completions.push(connection.completion); + await context.serve({ + origin: "https://peer.invalid", + dispatch: async () => ({ status: 200, bodyBytes: Buffer.from("{}") }), + }); + }, + }); + const source = f.client("source"); + await target.connect(new AbortController().signal, relayDeadline()); + const factory = source.factory("target", f.identities.target.expectedPin); + await createPeerHttpExecutor({ factory })("GET", "https://peer.invalid/first"); + await Promise.all(completions); + const previousChecks = f.checks.length, + previousWrites = wire.mock.calls.length; + const abort = new AbortController(); + // Act: retain the original short deadline and cancel only after registration has yielded to pacing. + const opening = factory.open({ signal: abort.signal, deadlineMs: performance.now() + 500 }); + const rejected = expect(opening).rejects.toThrow(/cancelled|expired/); + await nextTurn(); + if (mode === "cancel") abort.abort(); + await rejected; + // Assert: cleanup may send CLOSE, but the second route never starts authentication or application I/O. + expect(f.checks).toHaveLength(previousChecks); + for (const [message] of wire.mock.calls.slice(previousWrites)) { + expect(message).not.toBeInstanceOf(Uint8Array); + if (typeof message === "string") expect(JSON.parse(message).type).not.toBe("OPEN"); + } + expect(f.runtime.counters().routes).toBe(0); + }, + 10000, + ); + + it("paces two honest sources within the shared real-clock pair quota", async () => { + // Arrange: both directions use actual clients, Noise, admission and native SQLite quota time. + const f = await fixture(); + const onInbound: RelayPeerClientOptions["onInbound"] = async (_connection, context) => + context.serve({ + origin: "https://peer.invalid", + dispatch: async () => ({ status: 200, bodyBytes: Buffer.from("{}") }), + }); + const source = f.client("source", { onInbound }), + target = f.client("target", { onInbound }); + await Promise.all([ + source.connect(new AbortController().signal, relayDeadline()), + target.connect(new AbortController().signal, relayDeadline()), + ]); + const forward = createPeerHttpExecutor({ + factory: source.factory("target", f.identities.target.expectedPin), + }); + const reverse = createPeerHttpExecutor({ + factory: target.factory("source", f.identities.source.expectedPin), + }); + // Act: the second round must wait rather than exhaust the pair's two-token burst. + const first = await Promise.all([ + forward("GET", "https://peer.invalid/first"), + reverse("GET", "https://peer.invalid/first"), + ]); + const second = await Promise.all([ + forward("GET", "https://peer.invalid/second"), + reverse("GET", "https://peer.invalid/second"), + ]); + // Assert: each direction checks admission at least six real seconds apart and all exchanges complete. + expect([...first, ...second]).toEqual([ + [200, {}], + [200, {}], + [200, {}], + [200, {}], + ]); + for (const device of ["source", "target"]) { + const times = f.checks.flatMap(({ packet }, index) => { + const decoded = decodeRelayCheckPacket(packet); + return decoded.kind === "admission" && + decoded.value.role === "source" && + decoded.value.sourceDeviceId === device + ? [f.checkTimes[index] ?? 0] + : []; + }); + expect(times).toHaveLength(2); + expect((times[1] ?? 0) - (times[0] ?? 0)).toBeGreaterThanOrEqual(6000); + } + expect(f.runtime.counters().routes).toBe(0); + }, 20000); +}); + +it("keeps an unrelated live Noise route and shared renewal alive when the first maintenance caller cancels", async () => { + // Arrange: age only the relay's registration clock before opening the fresh live route. + // The private client expiry tweak models its renewal window without changing UTC, SQLite or Noise clocks. + let elapsed = 0, + holdRenewal = false; + const entered = Promise.withResolvers(), + release = Promise.withResolvers(); + const f = await fixture({ + monotonicMs: () => performance.now() + elapsed, + beforeCheck: async ({ packet }) => { + const decoded = decodeRelayCheckPacket(packet); + if (holdRenewal && decoded.kind === "registration" && decoded.value.deviceId === "source") { + entered.resolve(); + await release.promise; + } + }, + }); + cleanups.push(async () => release.resolve()); + const target = f.client("target", { + onInbound: async (_connection, context) => + context.serve({ + origin: "https://peer.invalid", + dispatch: async () => ({ status: 200, bodyBytes: Buffer.from("{}") }), + }), + }); + const source = f.client("source"); + await Promise.all([ + source.connect(new AbortController().signal, relayDeadline()), + target.connect(new AbortController().signal, relayDeadline()), + ]); + elapsed = 750200; + const owner = await source + .factory("target", f.identities.target.expectedPin) + .open({ signal: new AbortController().signal, deadlineMs: relayDeadline() }); + (source as unknown as { registrationExpires: number }).registrationExpires = + performance.now() + 149000; + holdRenewal = true; + const abort = new AbortController(); + const first = source.maintainRegistration(abort.signal, performance.now() + 3000); + const cancelled = expect(first).rejects.toThrow(/cancelled|expired/); + await entered.promise; + const shared = source.maintainRegistration(new AbortController().signal, relayDeadline()); + void shared.catch(() => {}); + let routeEnded = false; + void owner.completion.then( + () => { + routeEnded = true; + }, + () => { + routeEnded = true; + }, + ); + // Act: the first caller owns only its wait, not the shared renewal or the existing route. + abort.abort(); + await cancelled; + await nextTurn(); + // Assert: another waiter finishes renewal, then the original authenticated route completes HTTP normally. + expect(routeEnded).toBe(false); + expect(owner.duplex.destroyed).toBe(false); + release.resolve(); + await shared; + const result = await createPeerHttpExecutor({ factory: { open: async () => owner } })( + "GET", + "https://peer.invalid/after-renewal", + ); + expect(result).toEqual([200, {}]); + await expect(owner.completion).resolves.toBeUndefined(); +}, 10000);