From aa3d6c9319456e87b7f876d91472322613e19572 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Wed, 14 Oct 2026 10:32:56 +0200 Subject: [PATCH 1/2] feat(relay): the watch exchange MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ADR 0009 step 4. watchOS gives an app HTTPS and nothing lower, so a watch cannot hold a stream. The box now may open a second socket, /r///exchange, and a watch posts one sealed message to POST /x//. The relay draws a random 16-byte id, sends id ‖ body down the exchange socket and returns the answer that comes back behind the same id. It reads that id and nothing else. - One exchange socket per room; the first holds it. It never takes one of the four stream places, no stream sees an exchange, and the exchange socket gets no stream frame and no control word. - The limits in docs/watch-exchange.md, none keeping a list: 30 requests a minute per address, 2 open and 30 a minute per room, 1 024 open in the process, bodies of exactly 608 or 536 bytes in and at most 4 KiB back, 10 s per exchange. - 409 carries the right epoch. During a rotation's spread /x/ also takes the previous epoch. - Exchanges stay out of the log line and the routed totals, so a small fleet's counters cannot time a command. - Caddy writes no access log line for /x/. - relay/README.md: the new claim and what the relay still knows. Tests 1-5 from the paired doc: the blindness test gains exchanges, and tests/relay-exchange.test.ts covers routing, shape, limits and the epoch against a stand-in box and watch. Co-Authored-By: Claude Opus 5.5 --- .changeset/relay-watch-exchange.md | 5 + relay/README.md | 88 +++- relay/deploy/Caddyfile | 8 + relay/deploy/README.md | 3 +- relay/deploy/caddyfile.test.ts | 9 + relay/src/limits.ts | 20 + relay/src/protocol.ts | 13 +- relay/src/server.ts | 292 +++++++++++++- tests/relay-blindness.test.ts | 134 +++++- tests/relay-exchange.test.ts | 627 +++++++++++++++++++++++++++++ tests/support/exchange-harness.ts | 280 +++++++++++++ tests/support/relay-harness.ts | 2 +- 12 files changed, 1444 insertions(+), 37 deletions(-) create mode 100644 .changeset/relay-watch-exchange.md create mode 100644 tests/relay-exchange.test.ts create mode 100644 tests/support/exchange-harness.ts diff --git a/.changeset/relay-watch-exchange.md b/.changeset/relay-watch-exchange.md new file mode 100644 index 0000000..0234d82 --- /dev/null +++ b/.changeset/relay-watch-exchange.md @@ -0,0 +1,5 @@ +--- +"ftw-webapp": patch +--- + +The relay can carry a watch's commands. A box opens a second socket, the exchange socket, and a watch posts one sealed message at a time to `/x//`. The relay reads only the 16-byte id it puts in front of each message, an exchange never takes a phone's stream place, and Caddy writes no access log line for these requests. A box that opens no exchange socket works as before. diff --git a/relay/README.md b/relay/README.md index 044517a..6886314 100644 --- a/relay/README.md +++ b/relay/README.md @@ -1,18 +1,19 @@ # The blind relay -Carries encrypted frames between an FTW box and the browsers allowed to see it. -It is the only remote path that carries commands, and it cannot read a watt. +Carries encrypted frames between an FTW box and the browsers and watches +allowed to reach it. It is the only remote path that carries commands, and it +cannot read a watt. FTW Cloud, which a box owner may turn on for the iPhone app, is a separate service that can read what the owner enabled. It never touches the relay ([ADR 0008](https://github.com/srcfl/ftw/blob/master/docs/adr/0008-ftw-cloud.md)). ``` -box ──ws──▶ ┌────────────┐ ◀──ws── phone - │ relay │ - │ no keys │ ◀──ws── laptop - │ no traffic │ - │ on disk │ - └────────────┘ +box ──ws: uplink────▶ ┌────────────┐ ◀──ws──── phone + ──ws: exchange──▶ │ relay │ + │ no keys │ ◀──ws──── laptop + │ no traffic │ + │ on disk │ ◀──https── watch + └────────────┘ ``` ## Running it @@ -22,7 +23,8 @@ npm run relay # PORT=8787 HOST=0.0.0.0 ``` Network settings and two state-file paths come from the environment. There is -no configuration file; every limit stays in `src/server.ts` beside the code. +no configuration file; every limit stays in `src/server.ts` or `src/limits.ts` +beside the code. ## What it does @@ -35,6 +37,14 @@ stream a ciphertext belongs to without reading it, and reading it is the one thing it must not do. At 512 bytes a second on the telemetry lane, four streams cost two kilobytes a second. That is the right way to spend the bandwidth. +A watch cannot hold a stream: watchOS gives an app HTTPS and nothing lower. So +the box opens a second socket, the **exchange socket**, and a watch posts one +sealed message at a time. The relay draws a random 16-byte id, sends the id and +the body down that socket, and returns the answer that comes back behind the +same id. An exchange never takes one of the four stream places, no stream sees +one, and the exchange socket sees no stream frame. +[docs/watch-exchange.md](../docs/watch-exchange.md) has the wire. + ## What it cannot do, and how you can check Read `src/`. The small files are meant to be read: routing and @@ -46,12 +56,13 @@ man's switch in `deadman.ts`. |---|---| | No keys, no decryption | Nothing in this directory imports a cipher | | No parsing of payloads | The only property read off a routed message is `.length` | +| On the exchange socket, only its own id | `#onAnswer()` reads the first 16 bytes, the id the relay drew, and returns the rest unchanged | | No compression | `perMessageDeflate: false` — a compressed frame's size depends on its content | | No padding, no trimming | `send()` forwards the received buffer unchanged | | No handle derivation | Needs a secret that never comes near here; see `src/lib/carrier/rendezvous.ts` | -| No storage of routed traffic | Rooms are a `Map` and are deleted with their last socket | +| No storage of routed traffic | Rooms are a `Map` and are deleted with their last socket. An open exchange leaves its room's map when its answer goes or its timer fires | | Bounded stored state | `deadman.json` holds sealed dead-man rows; `fleet-stats.json` holds daily counters for 90 days. Neither holds routed traffic | -| No record of who was here | `log` is given counts and statuses; `inspect()` is everything held, and holds counts | +| No record of who was here | `log` is given counts and statuses, and nothing about exchanges; `inspect()` is everything held, and holds counts. Caddy writes no access log line for `/x/` | **The dead man's switch is the one thing the relay holds**, because it is the one notification the box cannot push about itself: that it is gone. A box @@ -90,8 +101,9 @@ from a real box. The daily count is an operating estimate, not a billing or security measure. `tests/relay-blindness.test.ts` runs a real box against a real app across this -server, collects every routed byte, every logged line and everything in memory, -and fails if a message type, a device name or a watt reading is anywhere in it. +server, and a stand-in box and watch through the exchange. It collects every +routed byte, every logged line and everything in memory, and fails if a message +type, an op name, a device name or a watt reading is anywhere in it. It first checks that the same detector catches the unsealed frames, because a detector that catches nothing proves nothing. The dead man tests hold the new claims the same way: the persisted file is byte-audited against the five @@ -147,7 +159,8 @@ keeping the list, and it is paid on purpose. ## Wire ``` -ws://host/r/// +ws://host/r/// +POST https://host/x// ``` Binary messages are routed. Text from a peer is a protocol error and closes the @@ -161,7 +174,7 @@ interpret. Text from the relay is one of two words: | Close code | Meaning | |---|---| -| 4400 | Bad join, text on a binary channel, or an oversize message | +| 4400 | Bad join, text on a binary channel, an oversize message, or an exchange answer of the wrong size | | 4409 | Wrong epoch; the reason carries the right one | | 4410 | The epoch turned over; the reason carries the new one | | 4429 | Rate limited, room occupied, or the process is at its ceiling | @@ -170,6 +183,33 @@ A single timer drives the heartbeat, the reaping of dead sockets and the rotation check, so nothing the relay emits has a cadence that traffic can influence. +### The exchange + +The box joins with role `exchange`, one per room; the first holds it, as with +the uplink. Each message from the relay is `id(16) ‖ body`, and each answer +from the box is `id(16) ‖ reply`. The relay drops an answer whose id it is not +waiting for, and says no word on this socket: no `ready`, no `gone`. + +A watch posts `content-type: application/octet-stream` and gets a status back. +No CORS header is ever sent, so a web page cannot read an answer. + +| Status | Meaning | +|---|---| +| 200 | The box replied; the body is its reply, unchanged | +| 400 | The path, the content type or the body size is not one the route takes | +| 404 | No box on this handle now | +| 405 | Not a POST | +| 409 | Wrong epoch; the body carries the right one | +| 429 | Rate limited | +| 501 | A box is here but opened no exchange socket: it needs an update | +| 504 | The box did not reply within 10 s, or its exchange socket closed | + +The limits, in `src/limits.ts`, keep no list: 30 requests a minute per address; +2 open and 30 a minute per room; 1 024 open in the process; a body of exactly +608 or 536 bytes in and at most 4 KiB back. During the five minutes a rotation +is spread over, `/x/` also takes the previous epoch, so a watch whose box has +not moved yet can find it. + ## What it still knows Honesty about the residue, in the spirit of `docs/architecture.md`: @@ -182,6 +222,24 @@ Honesty about the residue, in the spirit of `docs/architecture.md`: - The six coarse fleet labels during one daily HTTP request. It stores their daily totals, not the request or the address that sent it. +While a watch's request is open, it also knows what +[docs/watch-exchange.md](../docs/watch-exchange.md#what-the-relay-must-never-learn) +lists: + +- That someone with this hour's handle sent a request at a given time from a + given address. [architecture.md](../docs/architecture.md#what-is-not-yet-true) + names the same channel for app-to-box frames, which leave only on a person's + action. +- Whether a request opens a session, from its size. The order already told it + that. +- How long the box took. A refusal answers in milliseconds; an applied command + waits for the charger. ADR 0009 measures whether that gap shows through + network jitter, and adds a fixed floor only if it does. +- That this handle had an HTTPS client this hour. + +It writes none of these down. No log line names a handle, an address, a size +or a time, and Caddy writes no line for `/x/`. + ## Not here yet - TLS. Terminate it in front; this speaks plain WebSocket. diff --git a/relay/deploy/Caddyfile b/relay/deploy/Caddyfile index 72f4ed1..cc5d43e 100644 --- a/relay/deploy/Caddyfile +++ b/relay/deploy/Caddyfile @@ -30,6 +30,14 @@ relay.ftw.energy { @fleet_stats path /fleet/stats /fleet/stats/ respond @fleet_stats 404 + # A watch's exchange is one request per command. With the URI and the + # addresses cut, the access log would still keep one line per request + # with its time, sizes and duration: a record of each command, which is + # what the escrow had to remove from its own proxy. So these requests + # write no access log line at all. + @exchange path /x/* + log_skip @exchange + # The relay reads this to rate-limit. Behind a terminator every client is # one address, so without it the whole fleet shares a single limit and # ordinary reconnection load is an outage. Caddy sets it; the relay only diff --git a/relay/deploy/README.md b/relay/deploy/README.md index c5d03d5..565a4b3 100644 --- a/relay/deploy/README.md +++ b/relay/deploy/README.md @@ -75,7 +75,8 @@ The relay logs one line every fifteen seconds with aggregate counts — epoch, rooms, sockets, frames, bytes and today's fleet reports. Deliberately no handles: a log line naming one is the household identifier this design exists to withhold, and logs are the least guarded thing in any deployment. The Caddy config strips the request URI -for the same reason, since the join path carries the handle. +for the same reason, since the join path carries the handle. It writes no line +at all for `/x/`, where a line per request would be a line per watch command. Read the detailed fleet totals from the host or an SSM session: diff --git a/relay/deploy/caddyfile.test.ts b/relay/deploy/caddyfile.test.ts index 62bb3a5..5a1b2e3 100644 --- a/relay/deploy/caddyfile.test.ts +++ b/relay/deploy/caddyfile.test.ts @@ -59,6 +59,15 @@ describe('the proxy in front of the relay', () => { } }) + it('writes no access log line for a watch exchange', () => { + // A line per request keeps a time, sizes and a duration even with the + // URI and addresses cut. On /x/ that is a line per command. + const source = caddyfile.replace(/^\s*#.*$/gm, '') + const matcher = /@(\w+)\s+path\s+\/x\/\*\s*$/m.exec(source) + expect(matcher, 'no matcher for /x/*').not.toBeNull() + expect(source).toMatch(new RegExp(`^\\s*log_skip\\s+@${matcher![1]}\\s*$`, 'm')) + }) + it('never writes a log to stdout or stderr, where Docker keeps it', () => { expect(caddyfile).not.toContain('output stdout') expect(caddyfile).not.toContain('output stderr') diff --git a/relay/src/limits.ts b/relay/src/limits.ts index 8a4d030..b4eca12 100644 --- a/relay/src/limits.ts +++ b/relay/src/limits.ts @@ -98,3 +98,23 @@ export class AttemptCounter { return digest.readUInt32BE(0) % this.#counts.length } } + +/** + * The exchange's bounds, from docs/watch-exchange.md. + * + * None of them keeps a list either. The per-address count is an + * AttemptCounter of its own. The per-room allowance is a TokenBucket on the + * room, gone with the room. Open exchanges sit in a map keyed by the relay's + * own random id, and each leaves it when its answer goes or its timer fires. + */ +export const EXCHANGE = { + /** IK message 1 (608), or one transport message (536). Nothing else. */ + requestBytes: [608, 536] as readonly number[], + /** The box's answer, after the id. */ + maxReplyBytes: 4096, + attemptsPerMinute: 30, + openPerRoom: 2, + perRoomPerMinute: 30, + openPerProcess: 1024, + timeoutMs: 10_000, +} as const diff --git a/relay/src/protocol.ts b/relay/src/protocol.ts index e184012..5e97728 100644 --- a/relay/src/protocol.ts +++ b/relay/src/protocol.ts @@ -3,7 +3,9 @@ * Everything the relay ever says to a peer is in this file. Close codes carry * the reason a socket ended, and the two control words say whether the other * side of the room is present. There is no relay message with a body, which is - * why the routing path never has to interpret anything. + * why the routing path never has to interpret anything. The one thing the + * relay adds to a message is the exchange id, on the box's exchange socket, + * and it is the relay's own random number. * * The carrier holds its own copy of these rather than importing them, so the * relay can be deployed and audited without the app beside it. @@ -30,3 +32,12 @@ export const CTRL_GONE = 'gone' /** A rendezvous handle is 128 bits in hex. See src/lib/carrier/rendezvous.ts. */ export const HANDLE_CHARS = 32 + +/** + * The relay's own id in front of every message on a box's exchange socket. + * + * Random, drawn per request, and copied back by the box in front of its + * answer. It names nothing outside the relay's memory. See + * docs/watch-exchange.md. + */ +export const EXCHANGE_ID_BYTES = 16 diff --git a/relay/src/server.ts b/relay/src/server.ts index d7f3144..3c93595 100644 --- a/relay/src/server.ts +++ b/relay/src/server.ts @@ -1,15 +1,19 @@ /* The blind relay. * * It joins one box uplink to a handful of browser streams under a rendezvous - * handle and moves bytes between them. That is the whole job, and the file is - * short on purpose: an operator's promise that a service cannot read your data - * is worth what an outsider can check in an afternoon. + * handle and moves bytes between them. For a watch, which can only speak + * HTTPS, it also joins one request to one answer on the box's exchange + * socket. That is the whole job, and the file is short on purpose: an + * operator's promise that a service cannot read your data is worth what an + * outsider can check in an afternoon. * * What it never does, and where you can see that it never does: * * It does not decrypt. It has no keys and no code that would take one. * It does not parse payloads. Binary messages are forwarded as received; - * the only property read off them is `.length`, for a memory bound. + * the only property read off them is `.length`, for a memory bound. On + * the exchange socket it also reads the first 16 bytes of an answer: the + * id it wrote there itself. * It does not derive handles. That needs the secret the box and the app * swapped optically, which never comes near here. See relay/README.md. * It does not pad or trim. Bytes out are byte-identical to bytes in, so @@ -18,8 +22,9 @@ * compressed frame's size depends on its content, which would undo the * padding that keeps the household's load pattern off the wire. * It does not store routed traffic. Rooms live in memory and are deleted - * when the last socket leaves. A separate /fleet HTTP door reduces the - * fixed anonymous report to daily totals and drops the request body. + * when the last socket leaves. An open exchange lives in its room until + * its answer goes or its timer fires. A separate /fleet HTTP door reduces + * the fixed anonymous report to daily totals and drops the request body. * It does not write down who was here. The log carries counts only. * * Its clock is the one thing peers do trust it for: it announces the epoch, @@ -27,6 +32,7 @@ */ import { WebSocketServer, type WebSocket, type RawData } from 'ws' +import { randomBytes } from 'node:crypto' import { createReadStream, existsSync } from 'node:fs' import { createServer, @@ -37,7 +43,7 @@ import { import { currentEpoch } from './epoch.ts' import { Deadman, rowError } from './deadman.ts' import { FleetStats, fleetReportError, type FleetReport } from './fleet.ts' -import { AttemptCounter, TokenBucket } from './limits.ts' +import { AttemptCounter, EXCHANGE, TokenBucket } from './limits.ts' import { CLOSE_BAD_JOIN, CLOSE_EPOCH, @@ -45,6 +51,7 @@ import { CLOSE_BUSY, CTRL_READY, CTRL_GONE, + EXCHANGE_ID_BYTES, HANDLE_CHARS, } from './protocol.ts' @@ -102,12 +109,22 @@ export interface RelayOptions { /** Fleet-report requests allowed per address per window. */ fleetAttemptLimit?: number fleetAttemptWindowMs?: number + /** Exchange requests allowed per address per window. */ + exchangeAttemptLimit?: number + exchangeAttemptWindowMs?: number + /** Open exchanges in the whole process. */ + maxOpenExchanges?: number + /** How long an exchange waits for the box. */ + exchangeTimeoutMs?: number } export interface RoomInspection { handle: string hasUplink: boolean streams: number + hasExchange: boolean + /** Exchanges waiting for the box's answer now. */ + exchanges: number } export interface RelayInspection { @@ -117,13 +134,16 @@ export interface RelayInspection { framesRouted: number bytesRouted: number heartbeats: number + /** Exchanges waiting for an answer now, in every room. */ + exchanges: number /** Counts only. The rows themselves never appear on any surface. */ deadman: { rows: number; claimed: number; armed: number } /** Counts only. Detailed daily totals stay on the local operator route. */ fleet: { days: number; reportsToday: number } } -type Role = 'box' | 'app' +/** The box's uplink, a browser stream, or the box's exchange socket. */ +type Role = 'box' | 'app' | 'exchange' interface Peer { socket: WebSocket @@ -139,9 +159,25 @@ interface Peer { claims?: Set } +/** One watch request waiting for the box. */ +interface OpenExchange { + res: ServerResponse + timer: ReturnType +} + interface Room { uplink: Peer | null streams: Set + /** + * The box's second socket, for exchanges. Never a stream: nothing routed + * between the uplink and the streams reaches it, and nothing it sends + * reaches them. + */ + exchange: Peer | null + /** Keyed by the relay's own id, in hex. Nothing else names an exchange. */ + open: Map + /** Exchanges allowed in this room. Lives on the room and dies with it. */ + exchangeRate: TokenBucket /** * The epoch this room's handle belongs to. * @@ -176,6 +212,10 @@ const DEFAULTS = { rotateSpreadMs: 300_000, fleetAttemptLimit: 120, fleetAttemptWindowMs: 60 * 60_000, + exchangeAttemptLimit: EXCHANGE.attemptsPerMinute, + exchangeAttemptWindowMs: 60_000, + maxOpenExchanges: EXCHANGE.openPerProcess, + exchangeTimeoutMs: EXCHANGE.timeoutMs, } as const /** @@ -225,6 +265,7 @@ export class RelayServer { #rooms = new Map() #attempts: AttemptCounter #fleetAttempts: AttemptCounter + #exchangeAttempts: AttemptCounter #timer: ReturnType #opts: Required< Omit @@ -237,6 +278,8 @@ export class RelayServer { #framesRouted = 0 #bytesRouted = 0 #heartbeats = 0 + /** Open exchanges in every room, for the process cap. */ + #openExchanges = 0 #deadman: Deadman #fleet: FleetStats @@ -258,6 +301,10 @@ export class RelayServer { maxSocketsPerAddress: opts.maxSocketsPerAddress ?? DEFAULTS.maxSocketsPerAddress, fleetAttemptLimit: opts.fleetAttemptLimit ?? DEFAULTS.fleetAttemptLimit, fleetAttemptWindowMs: opts.fleetAttemptWindowMs ?? DEFAULTS.fleetAttemptWindowMs, + exchangeAttemptLimit: opts.exchangeAttemptLimit ?? DEFAULTS.exchangeAttemptLimit, + exchangeAttemptWindowMs: opts.exchangeAttemptWindowMs ?? DEFAULTS.exchangeAttemptWindowMs, + maxOpenExchanges: opts.maxOpenExchanges ?? DEFAULTS.maxOpenExchanges, + exchangeTimeoutMs: opts.exchangeTimeoutMs ?? DEFAULTS.exchangeTimeoutMs, } const now = this.#opts.now() @@ -270,6 +317,10 @@ export class RelayServer { { limit: this.#opts.fleetAttemptLimit, windowMs: this.#opts.fleetAttemptWindowMs }, now ) + this.#exchangeAttempts = new AttemptCounter( + { limit: this.#opts.exchangeAttemptLimit, windowMs: this.#opts.exchangeAttemptWindowMs }, + now + ) this.#deadman = new Deadman({ path: opts.deadmanPath ?? '', now: this.#opts.now, @@ -353,7 +404,10 @@ export class RelayServer { http.on('error', (err) => log(`relay: http error after start: ${String(err)}`)) const relay = new RelayServer(wss, opts, http) - relayRoutes = (req, res) => relay.#serveFleet(req, res) || relay.#serveDeadman(req, res) + relayRoutes = (req, res) => + relay.#serveFleet(req, res) || + relay.#serveDeadman(req, res) || + relay.#serveExchange(req, res) resolve(relay) }) }) @@ -388,10 +442,13 @@ export class RelayServer { handle, hasUplink: room.uplink !== null, streams: room.streams.size, + hasExchange: room.exchange !== null, + exchanges: room.open.size, })), framesRouted: this.#framesRouted, bytesRouted: this.#bytesRouted, heartbeats: this.#heartbeats, + exchanges: this.#openExchanges, deadman: this.#deadman.inspect(), fleet: this.#fleet.inspect(), } @@ -399,6 +456,7 @@ export class RelayServer { stop(): Promise { clearInterval(this.#timer) + for (const room of this.#rooms.values()) this.#failOpen(room) for (const socket of this.#wss.clients) socket.terminate() this.#rooms.clear() // Both, and the HTTP server last: it owns the listening socket now, so @@ -458,8 +516,14 @@ export class RelayServer { return } - const room = - this.#rooms.get(join.handle) ?? { uplink: null, streams: new Set(), epoch: join.epoch } + const room = this.#rooms.get(join.handle) ?? { + uplink: null, + streams: new Set(), + exchange: null, + open: new Map(), + exchangeRate: new TokenBucket(EXCHANGE.perRoomPerMinute, EXCHANGE.perRoomPerMinute / 60, now), + epoch: join.epoch, + } if (join.role === 'box' && room.uplink) { // The incumbent uplink keeps the room. A box whose socket died silently @@ -469,6 +533,11 @@ export class RelayServer { socket.close(CLOSE_BUSY, 'uplink') return } + if (join.role === 'exchange' && room.exchange) { + // The same rule for the exchange socket, for the same reason. + socket.close(CLOSE_BUSY, 'exchange') + return + } if (join.role === 'app' && room.streams.size >= this.#opts.maxStreamsPerHandle) { socket.close(CLOSE_BUSY, 'streams') return @@ -484,6 +553,7 @@ export class RelayServer { } if (join.role === 'box') room.uplink = peer + else if (join.role === 'exchange') room.exchange = peer else room.streams.add(peer) this.#rooms.set(join.handle, room) this.#sockets += 1 @@ -534,6 +604,13 @@ export class RelayServer { return } + // An exchange answer is not room traffic. It goes to one waiting request + // and is counted nowhere, so the log's totals cannot time a command. + if (peer.role === 'exchange') { + this.#onAnswer(peer, room, frame) + return + } + this.#framesRouted += 1 this.#bytesRouted += frame.length @@ -562,9 +639,14 @@ export class RelayServer { else this.#perAddress.delete(peer.address) } if (room.uplink === peer) room.uplink = null - else room.streams.delete(peer) - - if (!room.uplink && room.streams.size === 0) { + else if (room.exchange === peer) { + room.exchange = null + // Nothing can answer them now. Saying so at once is kinder to a watch + // than letting it wait out the timer. + this.#failOpen(room) + } else room.streams.delete(peer) + + if (!room.uplink && !room.exchange && room.streams.size === 0) { // The room and its handle go with the last socket. This is what makes // "which boxes exist" unanswerable rather than merely unanswered. this.#rooms.delete(handle) @@ -767,6 +849,160 @@ export class RelayServer { return false } + + /** + * POST /x//: one sealed message from a watch to its box. + * + * watchOS gives an app HTTPS and nothing lower, so a watch cannot hold a + * stream. The box keeps a second socket for this. The relay puts an id of + * its own in front of the body, sends both down that socket, and returns + * the answer that comes back behind the same id. It reads the id and + * nothing else. See docs/watch-exchange.md. + * + * The answers are statuses, never words, and no CORS header is ever sent, + * so a web page cannot read one. + */ + #serveExchange(req: IncomingMessage, res: ServerResponse): boolean { + const url = req.url ?? '' + if (!url.startsWith('/x/')) return false + + if (req.method !== 'POST') { + res.writeHead(405, { allow: 'POST', 'cache-control': 'no-store' }) + res.end() + return true + } + if (!this.#exchangeAttempts.allow(clientAddress(req, this.#opts.trustProxy), this.#opts.now())) { + answer(res, 429) + return true + } + + const target = parseExchange(url) + const type = (req.headers['content-type'] ?? '').split(';')[0]!.trim().toLowerCase() + if (!target || type !== 'application/octet-stream') { + answer(res, 400) + return true + } + if (!this.#takesEpoch(target.epoch)) { + // As on a join: the right number, so the watch can derive this epoch's + // handle and ask again at once. + answer(res, 409, String(this.#epoch)) + return true + } + + const declared = req.headers['content-length'] + if (declared !== undefined && !EXCHANGE.requestBytes.includes(Number(declared))) { + answer(res, 400) + return true + } + + collectBody(req, Math.max(...EXCHANGE.requestBytes)) + .then((body) => { + if (!EXCHANGE.requestBytes.includes(body.length)) { + answer(res, 400) + return + } + this.#forward(target, body, res) + }) + .catch(() => answer(res, 400)) + return true + } + + /** + * The current epoch, and during a rotation the one before it. + * + * A box moves to the new handle only when its turn in the spread comes, so + * for those minutes a watch on the new handle may find no box. It tries the + * previous handle once; this is what lets that try land. + */ + #takesEpoch(epoch: number): boolean { + if (epoch === this.#epoch) return true + return this.#rotateStartedAtMs !== null && epoch === this.#epoch - 1 + } + + #forward(target: ExchangeTarget, body: Buffer, res: ServerResponse): void { + // Gone while its body was read: nobody to answer, so nothing to hold. + if (!res.socket || res.socket.destroyed) return + + const room = this.#rooms.get(target.handle) + if (!room || room.epoch !== target.epoch || (!room.uplink && !room.exchange)) { + answer(res, 404) + return + } + if (!room.exchange) { + // A box is here, but one too old to open the exchange socket. + answer(res, 501) + return + } + if ( + this.#openExchanges >= this.#opts.maxOpenExchanges || + room.open.size >= EXCHANGE.openPerRoom || + !room.exchangeRate.take(1, this.#opts.now()) + ) { + answer(res, 429) + return + } + + const id = randomBytes(EXCHANGE_ID_BYTES) + const key = id.toString('hex') + const timer = setTimeout(() => this.#settle(room, key, 504), this.#opts.exchangeTimeoutMs) + timer.unref?.() + room.open.set(key, { res, timer }) + this.#openExchanges += 1 + + // A watch that hangs up first is forgotten. A late answer for it finds no + // id and goes nowhere. + res.on('close', () => this.#settle(room, key)) + + send(room.exchange, Buffer.concat([id, body]), this.#opts.maxBufferedBytes) + } + + /** An answer on the exchange socket: the relay's id, then the box's bytes. */ + #onAnswer(peer: Peer, room: Room, frame: Buffer): void { + if (frame.length <= EXCHANGE_ID_BYTES || frame.length > EXCHANGE_ID_BYTES + EXCHANGE.maxReplyBytes) { + peer.socket.close(CLOSE_BAD_JOIN, 'size') + return + } + // An id this room is not waiting for is dropped. That covers a late + // answer, a repeated one, and an id from another room: a reply reaches + // only its own request. + const key = frame.subarray(0, EXCHANGE_ID_BYTES).toString('hex') + this.#settle(room, key, 200, frame.subarray(EXCHANGE_ID_BYTES)) + } + + /** + * End one open exchange. No status means the watch hung up and nobody is + * left to answer. + */ + #settle(room: Room, key: string, status?: number, reply?: Buffer): void { + const open = room.open.get(key) + if (!open) return + room.open.delete(key) + this.#openExchanges -= 1 + clearTimeout(open.timer) + + if (status === undefined) return + if (reply) { + open.res.writeHead(200, { + 'content-type': 'application/octet-stream', + 'cache-control': 'no-store', + }) + open.res.end(reply) + return + } + answer(open.res, status) + } + + /** Every open exchange in a room, answered 504: no box will answer now. */ + #failOpen(room: Room): void { + for (const key of [...room.open.keys()]) this.#settle(room, key, 504) + } +} + +/** A status and, for 409, the epoch. The relay's only words over HTTP. */ +function answer(res: ServerResponse, status: number, body = ''): void { + if (res.headersSent || res.destroyed) return + res.writeHead(status, { 'content-type': 'text/plain', 'cache-control': 'no-store' }) + res.end(body) } /** Read a request body, bounded. Rejects past the cap instead of buffering. */ @@ -789,7 +1025,10 @@ function collectBody(req: IncomingMessage, maxBytes: number): Promise { } function peers(room: Room): Peer[] { - return room.uplink ? [room.uplink, ...room.streams] : [...room.streams] + const out = [...room.streams] + if (room.uplink) out.unshift(room.uplink) + if (room.exchange) out.push(room.exchange) + return out } function send(peer: Peer, frame: Buffer, maxBufferedBytes: number): void { @@ -810,7 +1049,7 @@ interface Join { role: Role } -/** `/r///` and nothing else. */ +/** `/r///` and nothing else. */ export function parseJoin(url: string): Join | null { const path = url.split('?')[0] ?? '' const parts = path.split('/').filter((p) => p.length > 0) @@ -823,7 +1062,26 @@ export function parseJoin(url: string): Join | null { if (handle.length !== HANDLE_CHARS || !/^[0-9a-f]+$/.test(handle)) return null const role = parts[3] - if (role !== 'box' && role !== 'app') return null + if (role !== 'box' && role !== 'app' && role !== 'exchange') return null return { epoch, handle, role } } + +export interface ExchangeTarget { + epoch: number + handle: string +} + +/** `/x//` and nothing else. */ +export function parseExchange(url: string): ExchangeTarget | null { + const parts = url.split('/').filter((p) => p.length > 0) + if (parts.length !== 3 || parts[0] !== 'x') return null + + const epoch = Number(parts[1]) + if (!Number.isSafeInteger(epoch) || epoch < 0) return null + + const handle = parts[2]! + if (handle.length !== HANDLE_CHARS || !/^[0-9a-f]+$/.test(handle)) return null + + return { epoch, handle } +} diff --git a/tests/relay-blindness.test.ts b/tests/relay-blindness.test.ts index 66bbed1..fca836b 100644 --- a/tests/relay-blindness.test.ts +++ b/tests/relay-blindness.test.ts @@ -21,6 +21,14 @@ import { rendezvousHandle, currentEpoch } from '$lib/carrier/rendezvous' import { encode as cborEncode } from 'cbor2' import { encodeFrame } from '$lib/protocol/frame' import { sealedPair, waitFor } from './support/relay-harness.ts' +import { + CHARGER, + ExchangeBox, + HOUSE_NOW, + HOUSE_READINGS, + Watch, + command, +} from './support/exchange-harness.ts' /** * A secret the box and the app share optically. The relay never sees it. @@ -164,9 +172,9 @@ function concat(parts: Uint8Array[]): Uint8Array { } /** Which of the known strings and readings the dump gives away. */ -function leaks(dump: Uint8Array, readings: number[]): string[] { +function leaks(dump: Uint8Array, readings: number[], strings: string[] = KNOWN_STRINGS): string[] { const found: string[] = [] - for (const s of KNOWN_STRINGS) { + for (const s of strings) { const bytes = new TextEncoder().encode(s) const occurrences = countBytes(dump, bytes) if (occurrences >= (bytes.length >= MIN_NEEDLE_BYTES ? 1 : 2)) found.push(`string:${s}`) @@ -342,3 +350,125 @@ describe('the relay cannot read what it carries', () => { expect(rendezvousHandle(other, now)).not.toBe(rendezvousHandle(SECRET, now)) }) }) + +/** + * What a watch exchange would give away if the relay could read it: the + * message types, every op a command may name, how it turned out, the + * charger, and the pairing code that rides in message 1. + */ +const EXCHANGE_STRINGS = [ + 'hello', + 'hello_ok', + 'watch-code-7f3a9c', + 'ftw.dispatch.write', + 'cmd', + 'cmd.ack', + 'cmd.result', + 'now', + 'tick', + 'site.mode.set', + 'planner.prefs.set', + 'battery.hold', + 'loadpoint.hold', + 'loadpoint.boost', + 'loadpoint.soc.set', + 'loadpoint.surplus_only.set', + 'applied', + 'rejected', + 'unconfirmed', + 'E_SCOPE_DENIED', + 'observed', + CHARGER.src, + CHARGER.name, + 'meter.p1', + 'inverter.sungrow', +] + +describe('the relay cannot read a watch exchange', () => { + it('dumps everything it saw and gives nothing away', async () => { + const logs: string[] = [] + const relay = await RelayServer.start({ + heartbeatMs: 40, + exchangeAttemptLimit: 1000, + log: (line) => logs.push(line), + }) + + try { + const epoch = relay.epoch + const boxes: ExchangeBox[] = [] + const watches: Watch[] = [] + const handles: string[] = [] + + /** A house of its own: a room holds 30 exchanges a minute. */ + async function house(): Promise<{ box: ExchangeBox; watch: Watch }> { + const handle = rendezvousHandle(secretFor(30 + boxes.length), epoch) + const box = new ExchangeBox(relay.url, epoch, handle) + await waitFor(() => box.peer.open, 'the exchange socket') + const watch = new Watch(relay.url, box.keys.publicKey, epoch, handle) + boxes.push(box) + watches.push(watch) + handles.push(handle) + return { box, watch } + } + + // Every op the registry names, each way it can turn out, in one session + // of four exchanges per op. Then a read of Now. + const ops: [string, Record][] = [ + ['site.mode.set', { mode: 'planner_arbitrage' }], + ['planner.prefs.set', { style: 'balanced' }], + ['battery.hold', { w: -3000 }], + ['loadpoint.hold', { lp: CHARGER.src, amps: 16 }], + ['loadpoint.boost', { lp: CHARGER.src }], + ['loadpoint.soc.set', { lp: CHARGER.src, soc: 640 }], + ['loadpoint.surplus_only.set', { lp: CHARGER.src, on: true }], + ] + let n = 0 + for (const [op, args] of ops) { + const { box, watch } = await house() + expect((await watch.open()).status).toBe(200) + for (const outcome of ['applied', 'rejected', 'unconfirmed', 'tick'] as const) { + box.outcome = outcome + expect((await watch.ask(command(op, args, n++))).status).toBe(200) + } + } + const { watch } = await house() + expect((await watch.open()).status).toBe(200) + const read = await watch.ask({ t: 'now' }) + expect(read.envelopes[0]).toEqual({ t: 'now', b: HOUSE_NOW }) + + await new Promise((r) => setTimeout(r, 120)) + const inspection = relay.inspect() + const dump = concat([ + ...watches.flatMap((w) => w.wire), + ...boxes.flatMap((b) => [...b.seen, ...b.answered]), + new TextEncoder().encode(logs.join('\n')), + new TextEncoder().encode(JSON.stringify(inspection)), + ]) + + // The control: the same exchanges unsealed trip every word, and two + // readings packed raw trip the numbers. + const plain = concat(watches.flatMap((w) => w.plain)) + expect(leaks(plain, [], EXCHANGE_STRINGS)).toEqual(EXCHANGE_STRINGS.map((s) => `string:${s}`)) + const packed = new DataView(new ArrayBuffer(8)) + packed.setInt32(0, HOUSE_READINGS[0]!, true) + packed.setInt32(4, HOUSE_READINGS[1]!, true) + expect(leaks(new Uint8Array(packed.buffer), HOUSE_READINGS).some((f) => f.startsWith('reading:'))).toBe( + true + ) + + expect(leaks(dump, HOUSE_READINGS, [...KNOWN_STRINGS, ...EXCHANGE_STRINGS])).toEqual([]) + + // No log line names the handle, and none counts the exchanges, which + // in a small fleet would time each command. + expect(logs.length).toBeGreaterThan(0) + for (const handle of handles) expect(logs.join('\n')).not.toContain(handle) + expect(logs.join('\n')).not.toMatch(/exchange/i) + expect(inspection.framesRouted).toBe(0) + expect(inspection.bytesRouted).toBe(0) + + for (const box of boxes) box.peer.close() + } finally { + await relay.stop() + } + }) +}) diff --git a/tests/relay-exchange.test.ts b/tests/relay-exchange.test.ts new file mode 100644 index 0000000..72f19ac --- /dev/null +++ b/tests/relay-exchange.test.ts @@ -0,0 +1,627 @@ +// @vitest-environment node + +/* The watch exchange, held to docs/watch-exchange.md. + * + * Routing, shape, limits and the epoch: tests 2 to 5 in that document's list. + * Test 1, blindness, sits in tests/relay-blindness.test.ts beside the + * stream's. Every case here is something the relay does without reading what + * it carries: it reads the 16-byte id it wrote, and lengths. + */ + +import net from 'node:net' +import { describe, it, expect, beforeAll, afterAll } from 'vitest' +import { RelayServer, parseExchange, parseJoin, rotationOffsetMs } from '../relay/src/server.ts' +import { CLOSE_BAD_JOIN, CLOSE_BUSY, CLOSE_ROTATED, EXCHANGE_ID_BYTES } from '../relay/src/protocol.ts' +import { EXCHANGE } from '../relay/src/limits.ts' +import { EPOCH_MS } from '../relay/src/epoch.ts' +import { rendezvousHandle } from '$lib/carrier/rendezvous' +import { + OP_BATTERY_HOLD, + OP_LOADPOINT_BOOST, + OP_LOADPOINT_HOLD, + OP_LOADPOINT_SOC_SET, + OP_LOADPOINT_SURPLUS_ONLY_SET, + OP_PLANNER_PREFS_SET, + OP_SET_MODE, +} from '$lib/protocol/messages' +import { TestPeer, waitFor } from './support/relay-harness.ts' +import { + CHARGER, + ExchangeBox, + TRANSPORT_BYTES, + Watch, + command, + concat, + postExchange, + type Outcome, +} from './support/exchange-harness.ts' + +/** A handle is a room, so no two cases share a secret. */ +function secretFor(n: number): Uint8Array { + return new Uint8Array(32).fill(n) +} + +function filled(length: number, value: number): Uint8Array { + return new Uint8Array(length).fill(value) +} + +/** + * A box's exchange socket that answers each message with `answer(body)` + * behind the relay's id. No Noise: these cases are about where bytes go. + */ +function rawBox( + url: string, + epoch: number, + handle: string, + answer: ((body: Uint8Array) => Uint8Array) | null = (body) => body.slice().reverse() +): TestPeer { + const peer = new TestPeer(url, epoch, handle, 'exchange') + if (answer) { + peer.onBinary((message) => { + peer.send(concat([message.subarray(0, EXCHANGE_ID_BYTES), answer(message.subarray(EXCHANGE_ID_BYTES))])) + }) + } + return peer +} + +const ARGS: Record> = { + [OP_SET_MODE]: { mode: 'planner_arbitrage' }, + [OP_PLANNER_PREFS_SET]: { style: 'balanced' }, + [OP_BATTERY_HOLD]: { w: -3000 }, + [OP_LOADPOINT_HOLD]: { lp: CHARGER.src, amps: 16 }, + [OP_LOADPOINT_BOOST]: { lp: CHARGER.src }, + [OP_LOADPOINT_SOC_SET]: { lp: CHARGER.src, soc: 640 }, + [OP_LOADPOINT_SURPLUS_ONLY_SET]: { lp: CHARGER.src, on: true }, +} + +describe('the exchange paths', () => { + it('accept exactly one shape each', () => { + const handle = 'a'.repeat(32) + expect(parseJoin(`/r/12/${handle}/exchange`)).toEqual({ epoch: 12, handle, role: 'exchange' }) + expect(parseExchange(`/x/12/${handle}`)).toEqual({ epoch: 12, handle }) + + for (const path of [ + '/x/', + `/x/12`, + `/x/12/${handle}/box`, + `/x/-1/${handle}`, + `/x/1.5/${handle}`, + '/x/12/short', + `/x/12/${'A'.repeat(32)}`, + `/r/12/${handle}`, + `/x/12/${handle}?id=1`, + ]) { + expect(parseExchange(path), path).toBeNull() + } + }) +}) + +describe('routing', () => { + let relay: RelayServer + + beforeAll(async () => { + relay = await RelayServer.start({ heartbeatMs: 1000, exchangeAttemptLimit: 1000 }) + }) + + afterAll(async () => { + await relay.stop() + }) + + it('carries a request to the box and its answer back, byte for byte', async () => { + const handle = rendezvousHandle(secretFor(1), relay.epoch) + const answer = filled(1072, 0xab) + const box = rawBox(relay.url, relay.epoch, handle, () => answer) + await waitFor(() => box.open, 'the exchange socket') + + const request = filled(TRANSPORT_BYTES, 0x5c) + const res = await postExchange(relay.url, relay.epoch, handle, request) + + expect(res.status).toBe(200) + expect(res.headers.get('content-type')).toBe('application/octet-stream') + expect([...res.body]).toEqual([...answer]) + + // What the box got: a fresh id of the relay's own, then the body as sent. + expect(box.received).toHaveLength(1) + expect(box.received[0]!.length).toBe(EXCHANGE_ID_BYTES + TRANSPORT_BYTES) + expect([...box.received[0]!.subarray(EXCHANGE_ID_BYTES)]).toEqual([...request]) + box.close() + }) + + it('sends each answer to its own request only', async () => { + const handle = rendezvousHandle(secretFor(2), relay.epoch) + const box = new TestPeer(relay.url, relay.epoch, handle, 'exchange') + await waitFor(() => box.open, 'the exchange socket') + + const first = postExchange(relay.url, relay.epoch, handle, filled(TRANSPORT_BYTES, 1)) + const second = postExchange(relay.url, relay.epoch, handle, filled(TRANSPORT_BYTES, 2)) + await waitFor(() => box.received.length === 2, 'both requests at the box') + + // Answered in the other order, each with the value its request carried. + for (const message of [...box.received].reverse()) { + const id = message.subarray(0, EXCHANGE_ID_BYTES) + box.send(concat([id, filled(64, 100 + message[EXCHANGE_ID_BYTES]!)])) + } + + const [a, b] = await Promise.all([first, second]) + expect(new Set(a.body)).toEqual(new Set([101])) + expect(new Set(b.body)).toEqual(new Set([102])) + + // Two requests, two ids, never the same one. + const ids = box.received.map((m) => Buffer.from(m.subarray(0, EXCHANGE_ID_BYTES)).toString('hex')) + expect(new Set(ids).size).toBe(2) + box.close() + }) + + it('drops an answer whose id it is not waiting for', async () => { + const mine = rendezvousHandle(secretFor(3), relay.epoch) + const theirs = rendezvousHandle(secretFor(4), relay.epoch) + const box = new TestPeer(relay.url, relay.epoch, mine, 'exchange') + const uplink = new TestPeer(relay.url, relay.epoch, mine, 'box') + const stream = new TestPeer(relay.url, relay.epoch, mine, 'app') + const other = new TestPeer(relay.url, relay.epoch, theirs, 'exchange') + await waitFor(() => box.open && other.open && uplink.linked && stream.linked, 'two rooms') + + const pending = postExchange(relay.url, relay.epoch, mine, filled(TRANSPORT_BYTES, 9)) + await waitFor(() => box.received.length === 1, 'the request at the box') + const id = box.received[0]!.subarray(0, EXCHANGE_ID_BYTES) + + // Another room's box answers with this room's id, and this box answers + // with an id nobody drew. Neither reaches the watch or anyone else. + other.send(concat([id, filled(64, 0xee)])) + box.send(concat([filled(EXCHANGE_ID_BYTES, 0), filled(64, 0xdd)])) + await new Promise((r) => setTimeout(r, 60)) + expect(relay.inspect().exchanges).toBe(1) + + box.send(concat([id, filled(64, 0x11)])) + const res = await pending + expect(new Set(res.body)).toEqual(new Set([0x11])) + + // The same id again, now settled, goes nowhere too. + box.send(concat([id, filled(64, 0x22)])) + await new Promise((r) => setTimeout(r, 60)) + + expect(uplink.received).toEqual([]) + expect(stream.received).toEqual([]) + expect(box.open && other.open).toBe(true) + for (const peer of [box, uplink, stream, other]) peer.close() + }) + + it('keeps exchanges and the room’s streams apart', async () => { + const handle = rendezvousHandle(secretFor(5), relay.epoch) + const uplink = new TestPeer(relay.url, relay.epoch, handle, 'box') + const streams = [1, 2, 3, 4].map(() => new TestPeer(relay.url, relay.epoch, handle, 'app')) + const box = rawBox(relay.url, relay.epoch, handle) + await waitFor(() => uplink.linked && streams.every((s) => s.linked) && box.open, 'a full room') + + // Four streams and the exchange socket: the exchange took no stream's + // place, so a watch cannot push a phone out, and a fifth phone is still + // refused for the streams' own reason. + expect(relay.inspect().rooms.find((r) => r.handle === handle)).toMatchObject({ + hasUplink: true, + streams: 4, + hasExchange: true, + }) + const fifth = new TestPeer(relay.url, relay.epoch, handle, 'app') + await waitFor(() => fifth.closes.length > 0, 'the fifth stream refused') + expect(fifth.closes[0]!.code).toBe(CLOSE_BUSY) + + const res = await postExchange(relay.url, relay.epoch, handle, filled(TRANSPORT_BYTES, 7)) + expect(res.status).toBe(200) + + uplink.send(filled(512, 1)) + streams[0]!.send(filled(512, 2)) + await waitFor( + () => streams.every((s) => s.received.length === 1) && uplink.received.length === 1, + 'room traffic' + ) + await new Promise((r) => setTimeout(r, 60)) + + // No stream saw the exchange, and the exchange socket saw no stream + // frame and no control word. + for (const s of streams) expect([...s.received[0]!]).toEqual([...filled(512, 1)]) + expect([...uplink.received[0]!]).toEqual([...filled(512, 2)]) + expect(box.received).toHaveLength(1) + expect(box.control).toEqual([]) + + for (const peer of [uplink, ...streams, box]) peer.close() + }) + + it('keeps the first exchange socket when a second arrives', async () => { + const handle = rendezvousHandle(secretFor(6), relay.epoch) + const first = rawBox(relay.url, relay.epoch, handle) + await waitFor(() => first.open, 'the first') + const second = rawBox(relay.url, relay.epoch, handle) + await waitFor(() => second.closes.length > 0, 'the second refused') + + expect(second.closes[0]!.code).toBe(CLOSE_BUSY) + expect(first.open).toBe(true) + first.close() + }) + + it('says 404 with no box and 501 for a box without the exchange socket', async () => { + const empty = rendezvousHandle(secretFor(7), relay.epoch) + expect((await postExchange(relay.url, relay.epoch, empty, filled(TRANSPORT_BYTES, 0))).status).toBe(404) + + // Phones waiting for a box that is off: still no box. + const waiting = new TestPeer(relay.url, relay.epoch, empty, 'app') + await waitFor(() => waiting.open, 'the stream') + expect((await postExchange(relay.url, relay.epoch, empty, filled(TRANSPORT_BYTES, 0))).status).toBe(404) + + const old = rendezvousHandle(secretFor(8), relay.epoch) + const uplink = new TestPeer(relay.url, relay.epoch, old, 'box') + await waitFor(() => uplink.open, 'an old box') + expect((await postExchange(relay.url, relay.epoch, old, filled(TRANSPORT_BYTES, 0))).status).toBe(501) + + waiting.close() + uplink.close() + }) + + it('answers 504 at once when the exchange socket goes', async () => { + const handle = rendezvousHandle(secretFor(9), relay.epoch) + const box = new TestPeer(relay.url, relay.epoch, handle, 'exchange') + await waitFor(() => box.open, 'the exchange socket') + + const started = Date.now() + const pending = postExchange(relay.url, relay.epoch, handle, filled(TRANSPORT_BYTES, 0)) + await waitFor(() => box.received.length === 1, 'the request at the box') + box.close() + + expect((await pending).status).toBe(504) + expect(Date.now() - started).toBeLessThan(EXCHANGE.timeoutMs / 2) + expect(relay.inspect().exchanges).toBe(0) + }) + + it('closes an exchange socket that sends text or an answer of the wrong size', async () => { + const text = new TestPeer(relay.url, relay.epoch, rendezvousHandle(secretFor(10), relay.epoch), 'exchange') + const short = new TestPeer(relay.url, relay.epoch, rendezvousHandle(secretFor(11), relay.epoch), 'exchange') + const long = new TestPeer(relay.url, relay.epoch, rendezvousHandle(secretFor(12), relay.epoch), 'exchange') + await waitFor(() => text.open && short.open && long.open, 'three exchange sockets') + + // The uplink's dead man word is the uplink's alone. + text.sendText(`deadman ${'a'.repeat(32)}`) + short.send(filled(EXCHANGE_ID_BYTES, 1)) + long.send(filled(EXCHANGE_ID_BYTES + EXCHANGE.maxReplyBytes + 1, 1)) + await waitFor(() => [text, short, long].every((p) => p.closes.length > 0), 'all three closed') + + for (const peer of [text, short, long]) expect(peer.closes[0]!.code).toBe(CLOSE_BAD_JOIN) + expect(relay.inspect().deadman.claimed).toBe(0) + }) + + it('takes only a POST of octet-stream, and never says CORS', async () => { + const handle = rendezvousHandle(secretFor(13), relay.epoch) + const box = rawBox(relay.url, relay.epoch, handle) + await waitFor(() => box.open, 'the exchange socket') + const url = `${relay.url.replace(/^ws/, 'http')}/x/${relay.epoch}/${handle}` + + const get = await fetch(url) + expect(get.status).toBe(405) + + // A web page's preflight. No CORS header comes back, so a browser never + // lets a page send the real request, let alone read its answer. + const preflight = await fetch(url, { + method: 'OPTIONS', + headers: { origin: 'https://example.com', 'access-control-request-method': 'POST' }, + }) + expect(preflight.status).toBe(405) + + // The one type a page may POST without a preflight is refused too. + const text = await postExchange(relay.url, relay.epoch, handle, filled(TRANSPORT_BYTES, 0), { + 'content-type': 'text/plain', + }) + expect(text.status).toBe(400) + + const ok = await postExchange(relay.url, relay.epoch, handle, filled(TRANSPORT_BYTES, 0)) + expect(ok.status).toBe(200) + + for (const res of [get.headers, preflight.headers, text.headers, ok.headers]) { + expect([...res.keys()].filter((k) => k.startsWith('access-control-'))).toEqual([]) + } + expect(box.received).toHaveLength(1) + box.close() + }) +}) + +describe('shape', () => { + let relay: RelayServer + + beforeAll(async () => { + relay = await RelayServer.start({ heartbeatMs: 1000, exchangeAttemptLimit: 1000 }) + }) + + afterAll(async () => { + await relay.stop() + }) + + it('gives a read and every command the same lengths, whatever the outcome', async () => { + const outcomes: Outcome[] = ['applied', 'rejected', 'unconfirmed', 'tick'] + const opens = { up: new Set(), down: new Set() } + const asks = { up: new Set(), down: new Set() } + const said = new Set() + + async function room(n: number): Promise<{ box: ExchangeBox; watch: Watch }> { + const handle = rendezvousHandle(secretFor(40 + n), relay.epoch) + const box = new ExchangeBox(relay.url, relay.epoch, handle) + await waitFor(() => box.peer.open, 'the exchange socket') + const watch = new Watch(relay.url, box.keys.publicKey, relay.epoch, handle) + const opened = await watch.open() + expect(opened.status).toBe(200) + expect(opened.helloOk?.t).toBe('hello_ok') + opens.up.add(watch.wire.at(-2)!.length) + opens.down.add(opened.body.length) + return { box, watch } + } + + let n = 0 + for (const [i, op] of Object.keys(ARGS).entries()) { + // One room and one session per op: four exchanges, as a session allows. + const { box, watch } = await room(i) + for (const outcome of outcomes) { + box.outcome = outcome + const res = await watch.ask(command(op, ARGS[op]!, n++)) + expect(res.status, `${op} ${outcome}`).toBe(200) + asks.up.add(watch.wire.at(-2)!.length) + asks.down.add(res.body.length) + said.add(res.envelopes.map((e) => e.t).join(' ')) + } + box.peer.close() + } + + const { box, watch } = await room(Object.keys(ARGS).length) + const read = await watch.ask({ t: 'now' }) + expect(read.status).toBe(200) + asks.up.add(watch.wire.at(-2)!.length) + asks.down.add(read.body.length) + said.add(read.envelopes.map((e) => e.t).join(' ')) + + // The runs really did differ inside the seal. + expect(said).toEqual( + new Set(['cmd.ack cmd.result', 'cmd.result tick', 'cmd.ack tick', 'now tick']) + ) + + // And outside it they did not. + expect(opens).toEqual({ up: new Set([608]), down: new Set([1072]) }) + expect(asks).toEqual({ up: new Set([536]), down: new Set([1072]) }) + + // The box saw each request behind the relay's id, and nothing else. + expect(new Set(box.seen.map((m) => m.length))).toEqual( + new Set([EXCHANGE_ID_BYTES + 608, EXCHANGE_ID_BYTES + 536]) + ) + box.peer.close() + }) + + it('refuses any other size with 400, before the box hears of it', async () => { + const handle = rendezvousHandle(secretFor(60), relay.epoch) + const box = rawBox(relay.url, relay.epoch, handle) + await waitFor(() => box.open, 'the exchange socket') + + for (const size of [0, 1, 16, 535, 537, 607, 609, 1072, 4096, 65_536]) { + const res = await postExchange(relay.url, relay.epoch, handle, filled(size, 3)) + expect(res.status, `${size} bytes`).toBe(400) + } + expect(box.received).toEqual([]) + box.close() + }) +}) + +describe('limits', () => { + it('lets one address make 30 requests a minute', async () => { + const relay = await RelayServer.start({ heartbeatMs: 1000 }) + try { + const handle = rendezvousHandle(secretFor(70), relay.epoch) + const statuses: number[] = [] + for (let i = 0; i <= EXCHANGE.attemptsPerMinute; i++) { + statuses.push((await postExchange(relay.url, relay.epoch, handle, filled(TRANSPORT_BYTES, 0))).status) + } + expect(EXCHANGE.attemptsPerMinute).toBe(30) + expect(statuses.slice(0, 30)).toEqual(new Array(30).fill(404)) + expect(statuses[30]).toBe(429) + + // The counter is a keyed hash, so nothing names the address. + expect(JSON.stringify(relay.inspect())).not.toContain('127.0.0.1') + } finally { + await relay.stop() + } + }) + + it('holds two open exchanges per room, and refuses a third', async () => { + const relay = await RelayServer.start({ heartbeatMs: 1000, exchangeAttemptLimit: 1000 }) + try { + const handle = rendezvousHandle(secretFor(71), relay.epoch) + const box = new TestPeer(relay.url, relay.epoch, handle, 'exchange') + await waitFor(() => box.open, 'the exchange socket') + + const held = [1, 2].map(() => postExchange(relay.url, relay.epoch, handle, filled(TRANSPORT_BYTES, 0))) + await waitFor(() => box.received.length === 2, 'two requests at the box') + + const third = await postExchange(relay.url, relay.epoch, handle, filled(TRANSPORT_BYTES, 0)) + expect(EXCHANGE.openPerRoom).toBe(2) + expect(third.status).toBe(429) + expect(box.received).toHaveLength(2) + + box.close() + expect((await Promise.all(held)).map((r) => r.status)).toEqual([504, 504]) + } finally { + await relay.stop() + } + }) + + it('lets a room have 30 exchanges a minute', async () => { + // A still clock, so the allowance cannot refill while the test runs. + const clock = Date.now() + const relay = await RelayServer.start({ + heartbeatMs: 1000, + exchangeAttemptLimit: 1000, + now: () => clock, + }) + try { + const handle = rendezvousHandle(secretFor(72), relay.epoch) + const box = rawBox(relay.url, relay.epoch, handle) + await waitFor(() => box.open, 'the exchange socket') + + const statuses: number[] = [] + for (let i = 0; i <= EXCHANGE.perRoomPerMinute; i++) { + statuses.push((await postExchange(relay.url, relay.epoch, handle, filled(TRANSPORT_BYTES, 0))).status) + } + expect(EXCHANGE.perRoomPerMinute).toBe(30) + expect(statuses.slice(0, 30)).toEqual(new Array(30).fill(200)) + expect(statuses[30]).toBe(429) + box.close() + } finally { + await relay.stop() + } + }) + + it('caps open exchanges across the process', async () => { + const relay = await RelayServer.start({ + heartbeatMs: 1000, + exchangeAttemptLimit: 1000, + maxOpenExchanges: 2, + }) + try { + const handles = [73, 74, 75].map((n) => rendezvousHandle(secretFor(n), relay.epoch)) + const boxes = handles.map((h) => new TestPeer(relay.url, relay.epoch, h, 'exchange')) + await waitFor(() => boxes.every((b) => b.open), 'three exchange sockets') + + const held = handles + .slice(0, 2) + .map((h) => postExchange(relay.url, relay.epoch, h, filled(TRANSPORT_BYTES, 0))) + await waitFor(() => relay.inspect().exchanges === 2, 'two open exchanges') + + const third = await postExchange(relay.url, relay.epoch, handles[2]!, filled(TRANSPORT_BYTES, 0)) + expect(EXCHANGE.openPerProcess).toBe(1024) + expect(third.status).toBe(429) + expect(boxes[2]!.received).toEqual([]) + + for (const b of boxes) b.close() + expect((await Promise.all(held)).map((r) => r.status)).toEqual([504, 504]) + } finally { + await relay.stop() + } + }) + + it('waits for the box no longer than the timeout, and forgets the request', async () => { + const relay = await RelayServer.start({ + heartbeatMs: 1000, + exchangeAttemptLimit: 1000, + exchangeTimeoutMs: 200, + }) + try { + const handle = rendezvousHandle(secretFor(76), relay.epoch) + const box = new TestPeer(relay.url, relay.epoch, handle, 'exchange') + await waitFor(() => box.open, 'the exchange socket') + + const started = Date.now() + const res = await postExchange(relay.url, relay.epoch, handle, filled(TRANSPORT_BYTES, 0)) + expect(EXCHANGE.timeoutMs).toBe(10_000) + expect(res.status).toBe(504) + expect(Date.now() - started).toBeGreaterThanOrEqual(190) + expect(relay.inspect().exchanges).toBe(0) + + // The answer that comes too late finds nobody waiting, and the socket + // carries on. + box.send(concat([box.received[0]!.subarray(0, EXCHANGE_ID_BYTES), filled(64, 1)])) + await new Promise((r) => setTimeout(r, 60)) + expect(box.open).toBe(true) + expect(relay.inspect().exchanges).toBe(0) + box.close() + } finally { + await relay.stop() + } + }) +}) + +describe('a watch that hangs up', () => { + it('is forgotten at once, and its late answer goes nowhere', async () => { + const relay = await RelayServer.start({ heartbeatMs: 1000 }) + try { + const handle = rendezvousHandle(secretFor(77), relay.epoch) + const box = new TestPeer(relay.url, relay.epoch, handle, 'exchange') + await waitFor(() => box.open, 'the exchange socket') + + // A raw request, so the test can hang up without a client library + // retrying or waiting on its behalf. + const { port } = new URL(relay.url) + const sock = net.connect(Number(port), '127.0.0.1', () => { + sock.write( + `POST /x/${relay.epoch}/${handle} HTTP/1.1\r\nHost: relay\r\n` + + `Content-Type: application/octet-stream\r\nContent-Length: ${TRANSPORT_BYTES}\r\n\r\n` + ) + sock.write(Buffer.alloc(TRANSPORT_BYTES)) + }) + sock.on('error', () => {}) + await waitFor(() => box.received.length === 1, 'the request at the box') + expect(relay.inspect().exchanges).toBe(1) + + sock.destroy() + await waitFor(() => relay.inspect().exchanges === 0, 'the exchange to be forgotten') + + box.send(concat([box.received[0]!.subarray(0, EXCHANGE_ID_BYTES), filled(64, 1)])) + await new Promise((r) => setTimeout(r, 60)) + expect(box.open).toBe(true) + box.close() + } finally { + await relay.stop() + } + }) +}) + +describe('the epoch', () => { + it('answers a wrong one with 409 and the right one', async () => { + const relay = await RelayServer.start({ heartbeatMs: 1000 }) + try { + const res = await postExchange(relay.url, 0, rendezvousHandle(secretFor(80), 0), filled(TRANSPORT_BYTES, 0)) + expect(res.status).toBe(409) + expect(Number(new TextDecoder().decode(res.body))).toBe(relay.epoch) + } finally { + await relay.stop() + } + }) + + it('takes the previous epoch during the spread, closes the socket with its room, then refuses it', async () => { + const spreadMs = 60_000 + let clock = EPOCH_MS * 480_000 + 60_000 + const relay = await RelayServer.start({ + heartbeatMs: 50, + now: () => clock, + rotateSpreadMs: spreadMs, + exchangeAttemptLimit: 1000, + }) + try { + const epoch = relay.epoch + // A room whose turn comes after the window opens, so the window can be + // seen while it is still there. + let n = 81 + while (rotationOffsetMs(rendezvousHandle(secretFor(n), epoch), spreadMs) === 0) n++ + const handle = rendezvousHandle(secretFor(n), epoch) + + const uplink = new TestPeer(relay.url, epoch, handle, 'box') + const box = rawBox(relay.url, epoch, handle) + await waitFor(() => uplink.open && box.open, 'the box') + + const post = (e: number) => postExchange(relay.url, e, handle, filled(TRANSPORT_BYTES, 0)) + expect((await post(epoch)).status).toBe(200) + // No rotation yet, so the epoch before is just wrong. + expect((await post(epoch - 1)).status).toBe(409) + + clock = (epoch + 1) * EPOCH_MS + 1 + await waitFor(() => relay.epoch === epoch + 1, 'the epoch to turn') + + // The box has not moved yet. On the new epoch there is no box under this + // handle; on the previous one there is, for the length of the spread. + expect((await post(epoch + 1)).status).toBe(404) + expect((await post(epoch)).status).toBe(200) + expect((await post(epoch - 1)).status).toBe(409) + + clock += spreadMs + await waitFor(() => box.closes.length > 0 && uplink.closes.length > 0, 'the room to rotate') + expect(box.closes[0]).toEqual({ code: CLOSE_ROTATED, reason: String(epoch + 1) }) + expect(uplink.closes[0]!.code).toBe(CLOSE_ROTATED) + + const after = await post(epoch) + expect(after.status).toBe(409) + expect(Number(new TextDecoder().decode(after.body))).toBe(epoch + 1) + } finally { + await relay.stop() + } + }) +}) diff --git a/tests/support/exchange-harness.ts b/tests/support/exchange-harness.ts new file mode 100644 index 0000000..8508260 --- /dev/null +++ b/tests/support/exchange-harness.ts @@ -0,0 +1,280 @@ +/* Test rig for the watch exchange: a box's exchange socket and a watch, both + * played by the test, across a real relay. + * + * Core's side is ADR 0009 step 5 in srcfl/ftw, and the watch is Swift. This + * stand-in follows docs/watch-exchange.md: IK with the prologue + * `ftw.exchange.v1:` and the box's static key, `hello_ok` in a 1 KiB bulk + * bucket, a lane 0 frame in every other message, and two transport messages + * in every answer after the handshake. What the relay may learn does not + * depend on which side wrote the bytes, only on their being sealed and on + * their sizes. + */ + +import { HandshakeState, generateKeyPair, type KeyPair } from '$lib/crypto/noise' +import { NoiseTransport } from '$lib/crypto/transport' +import { decodeFrame, encodeFrame, LANE_BULK, LANE_CONTROL, type Envelope } from '$lib/protocol/frame' +import { EXCHANGE_ID_BYTES } from '../../relay/src/protocol.ts' +import { TestPeer } from './relay-harness.ts' + +/** Every frame but `hello_ok` is one lane 0 bucket. */ +export const LANE0 = 512 +export const HELLO_OK_BUCKET = 1024 +/** A transport message: 8 bytes of sequence, one lane 0 frame, a 16-byte tag. */ +export const TRANSPORT_BYTES = 536 + +/** The exchange's prologue, so a stream handshake can never stand in for one. */ +export function exchangePrologue(boxStatic: Uint8Array): Uint8Array { + const label = new TextEncoder().encode('ftw.exchange.v1:') + const out = new Uint8Array(label.length + boxStatic.length) + out.set(label) + out.set(boxStatic, label.length) + return out +} + +export function lane0(envelope: Envelope): Uint8Array { + return encodeFrame({ lane: LANE_CONTROL, flags: 0, envelope }, LANE0) +} + +export function concat(parts: Uint8Array[]): Uint8Array { + const out = new Uint8Array(parts.reduce((n, p) => n + p.length, 0)) + let at = 0 + for (const p of parts) { + out.set(p, at) + at += p.length + } + return out +} + +/** How the box answers a command. `tick` means it has not settled in 3 s. */ +export type Outcome = 'applied' | 'rejected' | 'unconfirmed' | 'tick' + +/** The charger the house has, as the box names it. */ +export const CHARGER = { src: 'charger.easee', name: 'Easee carport' } + +/** + * Now, as the box would send it: frozen fields, source states and the + * charger. Every reading is well clear of zero, so the blindness test has + * something worth hunting for. + */ +export const HOUSE_NOW = { + fields: { 1: 3, 2: 3456, 3: -4210, 4: -1234, 5: 640, 6: 2187, 10: 7360 }, + src: { 'meter.p1': 'live', 'inverter.sungrow': 'live', [CHARGER.src]: 'live' }, + lp: [{ src: CHARGER.src, name: CHARGER.name, plug: 'connected', soc: 640, w: 7360, controlRev: 41 }], +} + +/** The readings in HOUSE_NOW, for a detector to look for. */ +export const HOUSE_READINGS = [3456, -4210, -1234, 640, 2187, 7360] + +/** + * A box's exchange socket. It answers each message the way ADR 0009 says + * Core will, without Core's checks: what it answers is up to `outcome`. + */ +export class ExchangeBox { + readonly keys: KeyPair = generateKeyPair() + readonly peer: TestPeer + /** Each message as it reached the box: the relay's id, then the body. */ + readonly seen: Uint8Array[] = [] + /** Each answer as it left: the same id, then the sealed reply. */ + readonly answered: Uint8Array[] = [] + outcome: Outcome = 'applied' + /** When false the box reads and answers nothing, as a box that hung would. */ + answering = true + + #sessions: NoiseTransport[] = [] + #uptimeMs = 3_600_000 + + constructor(relayUrl: string, epoch: number, handle: string) { + this.peer = new TestPeer(relayUrl, epoch, handle, 'exchange') + this.peer.onBinary((message) => void this.#onMessage(message)) + } + + async #onMessage(message: Uint8Array): Promise { + this.seen.push(message) + if (!this.answering) return + + const id = message.subarray(0, EXCHANGE_ID_BYTES) + const body = message.subarray(EXCHANGE_ID_BYTES) + const reply = body.length === TRANSPORT_BYTES ? this.#exchange(body) : await this.#open(body) + if (!reply) return + + const out = concat([id, reply]) + this.answered.push(out) + this.peer.send(out) + } + + /** X1: IK message 1 in, message 2 out with `hello_ok`. */ + async #open(message1: Uint8Array): Promise { + const responder = HandshakeState.responder({ + staticKey: this.keys, + prologue: exchangePrologue(this.keys.publicKey), + }) + try { + const hello = decodeFrame(await responder.readMessage(message1)).envelope + if (hello.t !== 'hello') return null + } catch { + return null // a stranger gets no reply + } + + const helloOk: Envelope = { + t: 'hello_ok', + b: { + proto: 1, + role: 'watch', + scopes: ['ftw.live.read', 'ftw.plan.read', 'ftw.events.read', 'ftw.dispatch.write'], + clock: { uptimeMs: this.#uptimeMs, boot: 'boot-7c1d' }, + rev: 41, + }, + } + const message2 = await responder.writeMessage( + encodeFrame({ lane: LANE_BULK, flags: 0, envelope: helloOk }, HELLO_OK_BUCKET) + ) + this.#sessions.push(new NoiseTransport(responder.split())) + return message2 + } + + /** `cmd` or `now` in, two transport messages out. */ + #exchange(message: Uint8Array): Uint8Array | null { + for (const session of this.#sessions) { + let envelope: Envelope + try { + envelope = decodeFrame(session.decrypt(message)).envelope + } catch { + continue // another session's + } + const [first, second] = this.#answerTo(envelope) + return concat([session.encrypt(lane0(first)), session.encrypt(lane0(second))]) + } + return null + } + + #answerTo(envelope: Envelope): [Envelope, Envelope] { + this.#uptimeMs += 1_000 + const tick: Envelope = { t: 'tick', b: { seq: 1, uptimeMs: this.#uptimeMs } } + if (envelope.t === 'now') return [{ t: 'now', b: HOUSE_NOW }, tick] + + const cmdId = (envelope.b as { cmdId: string }).cmdId + const ack: Envelope = { + t: 'cmd.ack', + b: { cmdId, leaseId: 'lease-0c9e', expiresAtMs: this.#uptimeMs + 30_000 }, + } + const result = (state: string, extra: object = {}): Envelope => ({ + t: 'cmd.result', + b: { cmdId, state, ...extra }, + }) + + switch (this.outcome) { + case 'applied': + return [ack, result('applied', { observed: { value: 7360, src: CHARGER.src, uptimeMs: this.#uptimeMs } })] + case 'unconfirmed': + return [ack, result('unconfirmed')] + case 'rejected': + return [result('rejected', { error: { code: 'E_SCOPE_DENIED' } }), tick] + case 'tick': + return [ack, tick] + } + } +} + +export interface Posted { + status: number + body: Uint8Array + headers: Headers +} + +/** POST one body to `/x//`, as a watch's URLSession would. */ +export async function postExchange( + relayUrl: string, + epoch: number, + handle: string, + body: Uint8Array, + headers: Record = { 'content-type': 'application/octet-stream' } +): Promise { + const res = await fetch(`${relayUrl.replace(/^ws/, 'http')}/x/${epoch}/${handle}`, { + method: 'POST', + headers, + // A copy on its own ArrayBuffer, which is the type fetch takes. + body: new Uint8Array(body), + }) + return { status: res.status, body: new Uint8Array(await res.arrayBuffer()), headers: res.headers } +} + +/** A watch: its own key, the box's static key, and one session at a time. */ +export class Watch { + readonly keys: KeyPair = generateKeyPair() + /** Every body it posted and every answer it got, as the relay carried them. */ + readonly wire: Uint8Array[] = [] + /** The same exchanges unsealed, for a detector's control. */ + readonly plain: Uint8Array[] = [] + epoch: number + + #relayUrl: string + #boxStatic: Uint8Array + #handle: string + #transport: NoiseTransport | null = null + + constructor(relayUrl: string, boxStatic: Uint8Array, epoch: number, handle: string) { + this.#relayUrl = relayUrl + this.#boxStatic = boxStatic + this.epoch = epoch + this.#handle = handle + } + + async post(body: Uint8Array): Promise { + this.wire.push(body) + const answer = await postExchange(this.#relayUrl, this.epoch, this.#handle, body) + this.wire.push(answer.body) + return answer + } + + /** X1. The pairing code rides in message 1, sealed. */ + async open(code = 'watch-code-7f3a9c'): Promise { + const initiator = HandshakeState.initiator({ + staticKey: this.keys, + remoteStatic: this.#boxStatic, + prologue: exchangePrologue(this.#boxStatic), + }) + const hello: Envelope = { + t: 'hello', + b: { proto: { min: 1, max: 1 }, app: { build: 'watch-test', ua: 'watch' }, code }, + } + this.plain.push(lane0(hello)) + const answer = await this.post(await initiator.writeMessage(lane0(hello))) + if (answer.status !== 200) return answer + + const opened = await initiator.readMessage(answer.body) + this.plain.push(opened) + const helloOk = decodeFrame(opened).envelope + this.#transport = new NoiseTransport(initiator.split()) + return { ...answer, helloOk } + } + + /** One `cmd` or `now`, and the two envelopes that come back. */ + async ask(envelope: Envelope): Promise { + if (!this.#transport) throw new Error('open the exchange first') + this.plain.push(lane0(envelope)) + const answer = await this.post(this.#transport.encrypt(lane0(envelope))) + if (answer.status !== 200) return { ...answer, envelopes: [] } + + const envelopes: Envelope[] = [] + for (let at = 0; at < answer.body.length; at += TRANSPORT_BYTES) { + const frame = this.#transport.decrypt(answer.body.subarray(at, at + TRANSPORT_BYTES)) + this.plain.push(frame) + envelopes.push(decodeFrame(frame).envelope) + } + return { ...answer, envelopes } + } +} + +/** A `cmd` envelope as the watch builds it, with an id the watch minted. */ +export function command(op: string, args: Record, n: number): Envelope { + return { + t: 'cmd', + b: { + cmdId: `0192f0c4-7e1a-7000-8000-${String(n).padStart(12, '0')}`, + op, + args, + notValidAfterMs: 3_630_000, + expect: { rev: 41, guards: [] }, + }, + } +} diff --git a/tests/support/relay-harness.ts b/tests/support/relay-harness.ts index 55355e7..8cacba2 100644 --- a/tests/support/relay-harness.ts +++ b/tests/support/relay-harness.ts @@ -39,7 +39,7 @@ export class TestPeer { #ws: WebSocket #binary: ((bytes: Uint8Array) => void) | null = null - constructor(url: string, epoch: number, handle: string, role: 'box' | 'app') { + constructor(url: string, epoch: number, handle: string, role: 'box' | 'app' | 'exchange') { this.#ws = new WebSocket(`${url}/r/${epoch}/${handle}/${role}`) this.#ws.binaryType = 'arraybuffer' this.#ws.onmessage = (ev) => { From 01d43fbd5fa999eae396b70275624bdb9d031f2a Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Wed, 14 Oct 2026 11:14:18 +0200 Subject: [PATCH 2/2] test(relay): count each sealed exchange message once CI found "now" twice in the exchange blindness dump. The dump held each ciphertext twice, once as the watch sent or got it and once as the box got or sent it behind the relay's id, so one chance match of a three-byte word in random bytes counted as the two a short word needs. The dump now holds each message once and the ids, and the test checks that the box saw no bytes the watch did not. The relay and its peers share the test process, so a stall longer than one heartbeat leaves a pong unread and the relay reaps the socket, and the exchange on it. The exchange blindness test beats once a second and the rotation test every 500 ms, which a busy machine rides out. Co-Authored-By: Claude Opus 5.5 --- tests/relay-blindness.test.ts | 22 +++++++++++++++++++--- tests/relay-exchange.test.ts | 4 +++- 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/tests/relay-blindness.test.ts b/tests/relay-blindness.test.ts index fca836b..9667d69 100644 --- a/tests/relay-blindness.test.ts +++ b/tests/relay-blindness.test.ts @@ -17,6 +17,7 @@ import { describe, it, expect, beforeAll, afterAll } from 'vitest' import { RelayServer } from '../relay/src/server.ts' +import { EXCHANGE_ID_BYTES } from '../relay/src/protocol.ts' import { rendezvousHandle, currentEpoch } from '$lib/carrier/rendezvous' import { encode as cborEncode } from 'cbor2' import { encodeFrame } from '$lib/protocol/frame' @@ -387,8 +388,11 @@ const EXCHANGE_STRINGS = [ describe('the relay cannot read a watch exchange', () => { it('dumps everything it saw and gives nothing away', async () => { const logs: string[] = [] + // The relay and both peers share this process, so a stall longer than + // one beat leaves a pong unread and the relay reaps the socket, and the + // exchange on it. A second's beat rides out a busy machine. const relay = await RelayServer.start({ - heartbeatMs: 40, + heartbeatMs: 1000, exchangeAttemptLimit: 1000, log: (line) => logs.push(line), }) @@ -436,11 +440,23 @@ describe('the relay cannot read a watch exchange', () => { const read = await watch.ask({ t: 'now' }) expect(read.envelopes[0]).toEqual({ t: 'now', b: HOUSE_NOW }) - await new Promise((r) => setTimeout(r, 120)) + await waitFor(() => logs.length > 0, 'a log line') const inspection = relay.inspect() + + // Each sealed message once. The box got the watch's bytes behind the + // relay's id, and the watch got the box's answer as it was sent, so + // the two sides hold the same ciphertext. Counting both copies would + // count every chance match in random bytes twice, and a short word + // needs only two to count as a leak. + const hex = (b: Uint8Array) => Buffer.from(b).toString('hex') + const crossed = new Set(watches.flatMap((w) => w.wire.map(hex))) + const atTheBox = boxes.flatMap((b) => [...b.seen, ...b.answered]) + for (const message of atTheBox) { + expect(crossed.has(hex(message.subarray(EXCHANGE_ID_BYTES))), 'bytes only the box saw').toBe(true) + } const dump = concat([ ...watches.flatMap((w) => w.wire), - ...boxes.flatMap((b) => [...b.seen, ...b.answered]), + ...boxes.flatMap((b) => b.seen.map((m) => m.subarray(0, EXCHANGE_ID_BYTES))), new TextEncoder().encode(logs.join('\n')), new TextEncoder().encode(JSON.stringify(inspection)), ]) diff --git a/tests/relay-exchange.test.ts b/tests/relay-exchange.test.ts index 72f19ac..2c5f23d 100644 --- a/tests/relay-exchange.test.ts +++ b/tests/relay-exchange.test.ts @@ -580,8 +580,10 @@ describe('the epoch', () => { it('takes the previous epoch during the spread, closes the socket with its room, then refuses it', async () => { const spreadMs = 60_000 let clock = EPOCH_MS * 480_000 + 60_000 + // Each beat checks the clock. Not much faster: the relay and the peers + // share this process, and a stall longer than a beat gets a socket reaped. const relay = await RelayServer.start({ - heartbeatMs: 50, + heartbeatMs: 500, now: () => clock, rotateSpreadMs: spreadMs, exchangeAttemptLimit: 1000,