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..9667d69 100644 --- a/tests/relay-blindness.test.ts +++ b/tests/relay-blindness.test.ts @@ -17,10 +17,19 @@ 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' 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 +173,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 +351,140 @@ 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[] = [] + // 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: 1000, + 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 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.map((m) => m.subarray(0, EXCHANGE_ID_BYTES))), + 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..2c5f23d --- /dev/null +++ b/tests/relay-exchange.test.ts @@ -0,0 +1,629 @@ +// @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 + // 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: 500, + 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) => {