From 486b8d0786d0594d12e77ce1f18cd4b89d49ab9a Mon Sep 17 00:00:00 2001 From: Karn Date: Mon, 10 Aug 2026 13:34:44 +0530 Subject: [PATCH 01/29] =?UTF-8?q?feat(relay):=20the=20fleet=20directory=20?= =?UTF-8?q?=E2=80=94=20blobs=20the=20Worker=20keeps=20and=20cannot=20read?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Auto-pair needs one piece of distribution: a device paired on machine A has to reach machine B, and the machine list has to reach the browser. Daemons do not talk to each other and should not start to, and everything that needs distributing is a public signed artifact — so the relay hosts a directory and verifies none of it. One Durable Object for the whole relay (idFromName("directory"), because one relay is one fleet) with three shapes on one URL: PUT and a push socket behind the daemon secret, GET behind nothing but the rate rule that already meters /client and POST /api/pair. The invariant the leg exists to preserve, and the one every future change to it has to keep: the relay stores and serves, and never verifies. Readers check every signature under the fleet public key; a hostile relay can serve a stale, truncated or empty directory, exactly as it could always refuse to route, and cannot mint anything, because minting needs the key the Worker deliberately does not hold. Entries are content-addressed — the storage key is SHA-256 of the exact bytes and there is no other name for one. That is the only key a Worker which cannot read a blob is entitled to compute: a caller-supplied name would put the relay in charge of a namespace it cannot check, and one buggy or hostile secret-holder could then PUT a machine cert over a revocation with the relay's help. A PUT can only ever add. Idempotence and byte-exactness fall straight out of that, and both are what the suite leans on hardest: bytes the relay had altered would not hash to the name it filed them under. Bounded at both ends, because this is a credential-less-readable store fed by secret-holders: 4 KiB a blob, 512 entries, and at the cap a refusal (507, its own status) rather than an eviction. Every eviction policy can drop a revocation, and a directory that silently forgets one re-admits the device it revoked to every machine that had not yet heard. The Go deploy owes the directory a binding, a v2 migration and two run_worker_first entries; it gets them with the daemon side of this, and the Worker answers 503 rather than throwing until it does. Co-Authored-By: Claude Fable 5 --- relay/src/directory.ts | 288 ++++++++++++++++++++++++++ relay/src/http.ts | 47 +++++ relay/src/hub.ts | 45 +---- relay/src/index.ts | 68 ++++++- relay/test/directory.test.ts | 381 +++++++++++++++++++++++++++++++++++ relay/wrangler.jsonc | 36 +++- spec/relay-protocol.md | 167 ++++++++++++++- 7 files changed, 973 insertions(+), 59 deletions(-) create mode 100644 relay/src/directory.ts create mode 100644 relay/src/http.ts create mode 100644 relay/test/directory.test.ts diff --git a/relay/src/directory.ts b/relay/src/directory.ts new file mode 100644 index 0000000..e4712be --- /dev/null +++ b/relay/src/directory.ts @@ -0,0 +1,288 @@ +import { DurableObject } from 'cloudflare:workers' +import { JSON_NO_STORE, readCapped } from './http' +import type { Env } from './index' + +/** + * FleetDirectory is the relay's store of signed blobs: machine certs, device + * certs and revocations, minted by the daemons under the fleet key + * (spec/fleet-trust.md, "The fleet directory"). One object per relay — + * `idFromName("directory")`, because one relay is one fleet — and it is the + * only Durable Object here that is not per machine. + * + * **The relay stores and serves; it never verifies.** The fleet key never + * touches the Worker, by design: not as a secret, not as a binding, not in a + * log. So this class cannot tell a machine cert from a revocation from 200 + * bytes of noise, and must not try — every reader (daemon and browser both) + * verifies every signature under the fleet public key and drops what fails. + * What a hostile relay can do to this store is serve it stale, truncated or + * empty; what it cannot do is mint an entry, because minting needs the key it + * does not hold. Availability stays the relay's only power, which is the spine + * of spec/relay-protocol.md and survives this leg intact. + * + * Entries are **content-addressed**: the storage key is the SHA-256 of the + * exact bytes PUT, and nothing else. That is the only key a Worker which + * cannot read a blob is entitled to compute — a caller-supplied name would + * make the relay arbitrate a namespace it cannot check, and one buggy or + * hostile secret-holder could then PUT a machine cert over a revocation and + * the relay would help. Content addressing makes that structurally + * impossible: a PUT can only ever *add*. Two consequences fall straight out of + * it — a duplicate PUT is idempotent (same bytes, same key, same value, no + * second entry), and a blob round-trips byte for byte, because bytes the + * relay mangled would not hash to the key it filed them under. Readers dedupe + * on their own terms; a revocation outranks a device cert for the same key + * whatever their timestamps, which is a rule about *meaning* and therefore + * theirs, not ours. + * + * Hibernation rules, the same ones src/hub.ts lives by: sockets are accepted + * with `ctx.acceptWebSocket` — never `ws.accept()`, which pins the object in + * memory — no timers outside a request already holding the object awake, and + * nothing about the store in memory, because a wake finds every field back at + * its initializer. The entry count is a storage counter for exactly that + * reason (`nextChannel` in the hub is the same idea). The daemon sockets carry + * no attachment at all, and that is not an oversight: a directory socket is a + * fan-out target and nothing more — no channel, no counters anything reads, no + * per-socket state a wake could lose — so `getWebSockets('daemon')` is the + * whole of what a handler needs to know about it. + */ +export class FleetDirectory extends DurableObject { + constructor(ctx: DurableObjectState, env: Env) { + super(ctx, env) + // The edge answers keepalives itself, without waking a hibernated object + // (spec/relay-protocol.md, Keepalive). The directory socket speaks the + // same two strings the hub's legs do. + this.ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair('flue-ping', 'flue-pong')) + } + + /** + * The three shapes of `/directory`, told apart the way the router told them + * apart before forwarding: an upgrade, a PUT, or a GET. Auth happened there + * — the daemon legs of this object are secret-gated by `authorizeDaemon` in + * src/index.ts, exactly as `/daemon` is, and this object trusts that the way + * DaemonHub does. + */ + async fetch(req: Request): Promise { + if (req.headers.get('Upgrade')?.toLowerCase() === 'websocket') return this.acceptDaemon() + if (req.method === 'PUT') return this.put(req) + if (req.method === 'GET') return this.snapshot() + return new Response('not found', { status: 404 }) + } + + /** + * A daemon's push socket. Unlike the hub's daemon leg there is no takeover + * here: every machine in the fleet holds one of these at once, which is the + * point — a device paired on machine A has to reach machine B without B + * polling. + * + * The socket is push-only. Nothing a daemon could say on it is part of this + * protocol — a write is `PUT /directory`, an HTTP request that answers with + * the key it filed — so any message from a daemon here is a protocol error, + * handled in `webSocketMessage`. + */ + private acceptDaemon(): Response { + // The fan-out bound. The leg is secret-gated, so this is not the DoS cap + // MAX_CLIENTS is on the hub's credential-less leg; it bounds the cost of + // one PUT, which is one send per socket, against a fleet that has left + // half-dead sockets behind or a secret-holder opening them in a loop. A + // one-operator fleet reaches double digits. + if (this.ctx.getWebSockets('daemon').length >= MAX_DAEMON_SOCKETS) { + return new Response('{"error":"too many directory sockets"}', { + status: 503, + headers: JSON_NO_STORE, + }) + } + const pair = new WebSocketPair() + this.ctx.acceptWebSocket(pair[1], ['daemon']) + return new Response(null, { status: 101, webSocket: pair[0] }) + } + + /** + * `PUT /directory`: one signed blob, stored under the hash of its own bytes. + * + * The body is read as bytes and never parsed. It reaches storage, the GET + * and every push socket unchanged — parts B and C verify Ed25519 signatures + * over *these* bytes, and a relay that re-encoded a blob into some canonical + * shape of its own would break every signature over it while believing it + * had been helpful. + */ + private async put(req: Request): Promise { + const blob = await readCapped(req, MAX_BLOB_BYTES) + if (blob === null) return tooLarge() + // An empty body is not a signed anything under any encoding, and hashing + // it would spend an entry — permanently, since nothing here is ever + // deleted — on the one blob that is certainly not a certificate. + if (blob.byteLength === 0) { + return new Response('{"error":"empty blob"}', { status: 400, headers: JSON_NO_STORE }) + } + const key = await digestKey(blob) + // Read both, decide, then write once. Nothing awaits between the read and + // the write except the write itself, so the input gate holds a second PUT + // at the door and the count cannot drift: two blobs, two entries, always. + const have = await this.ctx.storage.get([BLOB_PREFIX + key, COUNT_KEY]) + if (have.has(BLOB_PREFIX + key)) { + // Already here, byte for byte — that is what sharing a key means. No + // write, no push, no second entry: a daemon that re-PUTs on every + // reconnect (and one will) costs this object nothing, and a replayed + // PUT of a blob captured off the wire changes nothing either. + return stored(key, 200) + } + const count = (have.get(COUNT_KEY) as number | undefined) ?? 0 + if (count >= MAX_ENTRIES) return full() + // One multi-key put: the blob and the count land together or not at all. + await this.ctx.storage.put({ [BLOB_PREFIX + key]: blob, [COUNT_KEY]: count + 1 }) + this.push(blob) + return stored(key, 201) + } + + /** + * `GET /directory`: the whole set, credential-less. + * + * No order is promised. Storage hands these back sorted by key, which is to + * say by digest, which is to say by nothing — the directory is a set, and a + * reader that inferred "newest last" from this array would be reading a + * property of SHA-256. `iat` is inside the blobs, where a reader that has + * verified a signature can trust it. + */ + private async snapshot(): Promise { + const rows = await this.ctx.storage.list({ prefix: BLOB_PREFIX }) + const entries: { key: string; blob: string }[] = [] + for (const [k, blob] of rows) { + entries.push({ key: k.slice(BLOB_PREFIX.length), blob: base64(blob) }) + } + return new Response(JSON.stringify({ v: 1, entries }), { headers: JSON_NO_STORE }) + } + + /** + * The write, to every daemon socket, as one binary message of exactly the + * blob's bytes. + * + * Raw bytes and no envelope, because an envelope is a chance to reshape + * something the relay cannot read: what the daemon verifies is the signature + * over these bytes, and the shortest path to them is no framing at all. The + * key is not sent — it is SHA-256 of the message, derivable by anyone who + * wants it, and a daemon that verifies signatures has no use for it. + * + * The fan-out includes the socket belonging to the machine whose PUT this + * was: an HTTP request carries no socket identity, so there is nobody to + * exclude, and there is no need to — the receiver already holds the blob and + * a push it already has is a no-op. Pushes are best-effort by nature. A + * daemon that was offline, or whose send failed here, converges on its next + * `GET /directory`; to close the window it opens this socket *first* and + * GETs second, so a write in between arrives on one path or the other. + */ + private push(blob: Uint8Array): void { + for (const ws of this.ctx.getWebSockets('daemon')) { + try { + ws.send(blob) + } catch { + // Closing under us — its close event finishes the teardown, and the + // daemon's reconnect GET is what it converges on. + } + } + } + + async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): Promise { + // flue-ping never reaches here (the auto-response answers it at the edge) + // and a stray flue-pong is dropped silently, exactly as on the hub's legs. + if (message === 'flue-pong') return + // Everything else is a protocol error: this socket carries relay → daemon + // pushes and nothing in the other direction. A daemon with something to + // say says it with `PUT /directory`. + try { + ws.close(1002, 'the directory socket is push-only') + } catch { + // Already closing. + } + } +} + +/** The storage prefix for blobs; the rest of the key is the digest. */ +const BLOB_PREFIX = 'blob:' + +/** How many entries are stored. A counter rather than a `list()` per PUT + * because listing reads every blob back to answer a question about their + * number, and it survives hibernation for the reason the hub's `nextChannel` + * does: a wake finds memory empty and storage intact. */ +const COUNT_KEY = 'count' + +/** + * The largest single blob. A certificate is a version, a kind, a 32-byte key, + * a machine id of at most 63 characters, a display name, a timestamp and a + * 64-byte signature — a couple of hundred bytes, encoded generously. 4 KiB is + * the same bound the pairing body carries (src/hub.ts, MAX_PAIR_BYTES) and two + * orders of magnitude of headroom over any honest cert; it is here because the + * Worker cannot tell a cert from four megabytes of anything, and this store is + * readable without a credential by whoever asks. + */ +const MAX_BLOB_BYTES = 4096 + +/** + * How many entries the directory will hold. + * + * The arithmetic that picks it: entries are machines, devices, and one + * revocation per device ever revoked, and nothing is ever removed. A + * one-operator fleet — the model this whole design serves — is a handful of + * machines and a handful of devices, so 512 is years of pair-and-revoke churn + * with room to spare. The other end of the number is what it bounds: 512 × + * 4 KiB is 2 MiB stored and about 2.8 MiB of base64 on a credential-less GET, + * which is the worst case a secret-holder can build and not one an honest + * fleet comes near (512 real certs are on the order of 100 KiB). + */ +const MAX_ENTRIES = 512 + +/** Push sockets one directory will hold: one per machine, plus slack for + * reconnects whose close events have not landed yet. */ +const MAX_DAEMON_SOCKETS = 256 + +/** A stored blob: 201 when this PUT created the entry, 200 when it was already + * there. The body is the same either way — a caller need not branch, and the + * key is what it came for — so the status is the whole of the difference, for + * logs and for a daemon that wants to know whether it was first. */ +function stored(key: string, status: 200 | 201): Response { + return new Response(JSON.stringify({ key }), { status, headers: JSON_NO_STORE }) +} + +function tooLarge(): Response { + return new Response('{"error":"blob too large"}', { status: 413, headers: JSON_NO_STORE }) +} + +/** + * The directory is full (`MAX_ENTRIES`). 507, and deliberately its own status: + * nothing else on this leg means "your blob is fine and I will not keep it", + * and a daemon has to be able to tell that from a 413 (this blob is wrong) or a + * 401 (this caller is wrong) without reading prose. + * + * Refusing rather than evicting is the security decision here, and it is not a + * close call. Every eviction policy — oldest first, largest first, random — + * can drop a revocation, and a directory that silently forgets a revocation + * re-admits the device it revoked to every machine that had not yet heard. + * That is a compromise; a refused PUT is an operator with a full directory, + * who is told so, loudly, at the moment it happens. Nothing in this object + * ever deletes an entry, and if pruning is ever wanted — a device cert whose + * key is revoked, say — it has to be a decision signed under the fleet key and + * carried out by something that can read what it is deleting. That is not the + * relay, and it must never become the relay. + */ +function full(): Response { + return new Response('{"error":"directory full"}', { status: 507, headers: JSON_NO_STORE }) +} + +/** The storage key for a blob: SHA-256 of its exact bytes, lowercase hex. */ +async function digestKey(blob: Uint8Array): Promise { + const digest = await crypto.subtle.digest('SHA-256', blob as BufferSource) + return [...new Uint8Array(digest)].map((b) => b.toString(16).padStart(2, '0')).join('') +} + +/** + * Standard base64, with padding — the alphabet Go's `encoding/json` reads a + * `[]byte` field in and writes one out in, so the daemon's own struct decodes + * `blob` with no help (internal/crypto keys travel the same way). Chunked + * because `String.fromCharCode(...bytes)` on a whole blob is an argument list, + * and argument lists have a length limit that a 4 KiB cert would find. + */ +function base64(bytes: Uint8Array): string { + let binary = '' + for (let i = 0; i < bytes.length; i += 0x8000) { + binary += String.fromCharCode(...bytes.subarray(i, i + 0x8000)) + } + return btoa(binary) +} diff --git a/relay/src/http.ts b/relay/src/http.ts new file mode 100644 index 0000000..0878871 --- /dev/null +++ b/relay/src/http.ts @@ -0,0 +1,47 @@ +// The two things both Durable Objects in this Worker need from HTTP: a bounded +// body read, and the headers every JSON answer either of them writes. + +/** What every JSON answer this Worker writes carries. `no-store` because all of + * them are about live state — a refusal that got cached by anything in the path + * would outlive the condition that caused it, and the fleet directory is a set + * a revocation is expected to change under a reader's feet. */ +export const JSON_NO_STORE = { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' } + +/** + * The request body, or null if it runs past `max`. + * + * Content-Length is consulted first so an honestly-labelled oversized POST is + * refused without being read, and the stream is then counted as it arrives: + * a chunked body declares no length at all, and buffering an undeclared one + * would hand an endpoint a memory DoS — the exposure the channel cap and the + * handshake deadline bound on the client leg (spec/relay-protocol.md, Auth), + * and the one the blob cap bounds on `PUT /directory`. + */ +export async function readCapped(req: Request, max: number): Promise { + const declared = Number(req.headers.get('Content-Length')) + if (Number.isFinite(declared) && declared > max) return null + if (!req.body) return new Uint8Array(0) + const reader = req.body.getReader() + const chunks: Uint8Array[] = [] + let total = 0 + try { + for (;;) { + const { done, value } = await reader.read() + if (done) break + total += value.byteLength + if (total > max) return null + chunks.push(value) + } + } finally { + // Releases the leg of an oversized body we stopped reading; a no-op once + // the stream has ended on its own. + await reader.cancel().catch(() => {}) + } + const out = new Uint8Array(total) + let at = 0 + for (const chunk of chunks) { + out.set(chunk, at) + at += chunk.byteLength + } + return out +} diff --git a/relay/src/hub.ts b/relay/src/hub.ts index 85fdb50..0565204 100644 --- a/relay/src/hub.ts +++ b/relay/src/hub.ts @@ -1,5 +1,6 @@ import { DurableObject } from 'cloudflare:workers' import { decodeFrame, encodeFrame } from './frame' +import { JSON_NO_STORE, readCapped } from './http' import type { Env } from './index' /** @@ -557,11 +558,6 @@ const MAX_PENDING_PAIRS = 8 /** Statuses a `Response` may not carry a body for. */ const NULL_BODY_STATUS = new Set([204, 205, 304]) -/** What every JSON answer this hub writes carries. `no-store` because all of - * them are about the state of one live ceremony: a refusal that got cached by - * anything in the path would outlive the condition that caused it. */ -const JSON_NO_STORE = { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' } - /** What a daemon socket remembers across hibernation. */ interface DaemonAttachment { /** False once replaced. `getWebSockets` can still list the dying socket, so @@ -635,45 +631,6 @@ function pairStatus(status: unknown): number { : 502 } -/** - * The request body, or null if it runs past `max`. - * - * Content-Length is consulted first so an honestly-labelled oversized POST is - * refused without being read, and the stream is then counted as it arrives: - * a chunked body declares no length at all, and buffering an undeclared one - * would hand the credential-less pairing endpoint a memory DoS — the same - * exposure the channel cap and the handshake deadline bound on the client leg - * (spec/relay-protocol.md, Auth). - */ -async function readCapped(req: Request, max: number): Promise { - const declared = Number(req.headers.get('Content-Length')) - if (Number.isFinite(declared) && declared > max) return null - if (!req.body) return new Uint8Array(0) - const reader = req.body.getReader() - const chunks: Uint8Array[] = [] - let total = 0 - try { - for (;;) { - const { done, value } = await reader.read() - if (done) break - total += value.byteLength - if (total > max) return null - chunks.push(value) - } - } finally { - // Releases the leg of an oversized body we stopped reading; a no-op once - // the stream has ended on its own. - await reader.cancel().catch(() => {}) - } - const out = new Uint8Array(total) - let at = 0 - for (const chunk of chunks) { - out.set(chunk, at) - at += chunk.byteLength - } - return out -} - /** WebSocket endpoints answer plain HTTP with 426, upgrade required. */ function refuseNonUpgrade(req: Request): Response | null { if (req.headers.get('Upgrade')?.toLowerCase() === 'websocket') return null diff --git a/relay/src/index.ts b/relay/src/index.ts index 28a1278..e9c6ba7 100644 --- a/relay/src/index.ts +++ b/relay/src/index.ts @@ -1,9 +1,13 @@ export interface Env { HUB: DurableObjectNamespace + /** The fleet directory: one object per relay, because one relay is one fleet + * (spec/fleet-trust.md, "The fleet directory"). Not per machine, which is + * why it is a namespace of its own rather than a name in HUB's. */ + DIRECTORY: DurableObjectNamespace ASSETS: Fetcher DAEMON_SECRET: string /** The per-IP rate limiter over the credential-less routes (`/client/*`, - * `POST /api/pair/*`) — a Cloudflare rate-limiting binding, declared in + * `POST /api/pair/*`, `GET /directory`) — a Cloudflare rate-limiting binding, declared in * wrangler.jsonc (`ratelimits`) and in the deploy `flue relay setup` builds * (internal/relaydeploy, RateLimitBinding); the two must agree. Optional * and fail-open: the rule bounds quota burn, it is not auth, and a Worker @@ -132,6 +136,30 @@ const rateLimited = () => headers: { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }, }) +/** A path under a prefix the Worker owns but does not serve. Not the machine + * 404: `/directory/anything` names no machine, and answering that it is not + * one would be a lie about what was asked. */ +const notFound = () => + new Response('{"error":"not found"}', { + status: 404, + headers: { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }, + }) + +const methodNotAllowed = (allow: string) => + new Response('{"error":"method not allowed"}', { + status: 405, + headers: { + 'Content-Type': 'application/json', + 'Cache-Control': 'no-store', + Allow: allow, + }, + }) + +/** The name of the one directory object. One relay is one fleet, so this is a + * constant rather than anything read off the request: there is no second + * directory to route to, and a name in the path would only invent one. */ +const DIRECTORY_NAME = 'directory' + /** * May this credential-less request proceed, under the per-IP rate rule? * @@ -210,6 +238,43 @@ export default { // token and earns no HMAC: asset requests are unmetered, and the tag // check exists to guard Durable Object wakes, not page loads. } + if (claims(url.pathname, '/directory')) { + // The fleet directory: one object for the whole relay, no id in the + // path, three shapes on one URL (spec/fleet-trust.md, "The fleet + // directory"). Nothing lives under it, and the Worker keeps refusal + // authority over the whole prefix — `/directory/anything` is the + // Worker's own 404, never the SPA. + if (url.pathname !== '/directory') return notFound() + // The relay holds no fleet key and verifies none of what this leg + // carries; what it can check is who may *write*, and that is the same + // bearer secret the daemon leg presents. Checked once, up front, + // because it decides both the auth answer and the metering below. + const daemon = authorizeDaemon(req, env) + // The rate rule meters everything on this prefix that does not hold the + // secret — which is the credential-less `GET /directory` the spec puts + // behind it, and equally an anonymous caller waving an `Upgrade` header + // to reach the 401 by a cheaper road. The secret-holding legs stay + // unmetered for the reason `/daemon` is: they are secret-gated, and a + // fleet is a handful of machines. + if (!daemon && !(await allowRate(req, env))) return rateLimited() + const upgrade = req.headers.get('Upgrade')?.toLowerCase() === 'websocket' + if ((upgrade || req.method === 'PUT') && !daemon) return unauthorized() + if (!upgrade && req.method !== 'GET' && req.method !== 'PUT') { + return methodNotAllowed('GET, PUT') + } + // Fail closed on a Worker deployed without the binding — an older + // `flue relay setup` than this script. Unlike CLIENT_RATE, which is + // fail-open because it bounds cost rather than access, there is no + // degraded directory to serve without the object: 503 says so, and the + // rest of the relay (which is every session on it) keeps running. + if (!env.DIRECTORY) { + return new Response('{"error":"directory unavailable"}', { + status: 503, + headers: { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' }, + }) + } + return env.DIRECTORY.get(env.DIRECTORY.idFromName(DIRECTORY_NAME)).fetch(req) + } if (url.pathname === '/api/health' && req.method === 'GET') { // Liveness of the Worker and nothing else — no id, no Durable Object // woken, nothing about any daemon. An uptime monitor pointed here costs @@ -230,4 +295,5 @@ export default { return env.ASSETS.fetch(req) }, } +export { FleetDirectory } from './directory' export { DaemonHub } from './hub' diff --git a/relay/test/directory.test.ts b/relay/test/directory.test.ts new file mode 100644 index 0000000..7b4a582 --- /dev/null +++ b/relay/test/directory.test.ts @@ -0,0 +1,381 @@ +import { env, SELF } from 'cloudflare:test' +import { describe, expect, it } from 'vitest' + +import worker, { type Env } from '../src/index' +import { BASE, Leg, machineId, TEST_SECRET, within } from './harness' + +/** + * The fleet directory: the relay's store of blobs it cannot read + * (spec/fleet-trust.md, "The fleet directory"). Everything below is written + * from the outside — through the real Worker, over the real Durable Object — + * because the contract parts B and C are written against is the HTTP and + * WebSocket surface, not the class. + * + * The invariant every one of these tests is really about: the relay stores and + * serves, and verifies nothing. A blob that came back changed, or an entry that + * quietly displaced another, would be a Worker that had formed an opinion about + * bytes it holds no key for. + */ + +/** Mirrors MAX_BLOB_BYTES in src/directory.ts. */ +const MAX_BLOB_BYTES = 4096 + +/** Mirrors MAX_ENTRIES in src/directory.ts. */ +const MAX_ENTRIES = 512 + +const DIRECTORY = `${BASE}/directory` + +const AUTH = { Authorization: `Bearer ${TEST_SECRET}` } + +/** + * Every test in this file shares the one directory object — `idFromName` + * ("directory") is a constant, which is the whole point of "one relay is one + * fleet" — so blobs are made unique per test rather than per run, and the + * assertions are about *this* blob rather than about the size of the set. + */ +function blob(tag: string, fill = 'x'): Uint8Array { + return new TextEncoder().encode(`${tag}:${fill}`) +} + +function put(body: BodyInit, headers: Record = AUTH): Promise { + return SELF.fetch(DIRECTORY, { method: 'PUT', body, headers }) +} + +async function get(): Promise<{ v: number; entries: { key: string; blob: string }[] }> { + const res = await SELF.fetch(DIRECTORY) + expect(res.status).toBe(200) + return (await res.json()) as { v: number; entries: { key: string; blob: string }[] } +} + +/** The key the directory files a blob under: SHA-256 of its exact bytes. */ +async function digest(bytes: Uint8Array): Promise { + const d = await crypto.subtle.digest('SHA-256', bytes as BufferSource) + return [...new Uint8Array(d)].map((b) => b.toString(16).padStart(2, '0')).join('') +} + +/** The bytes of one entry of a GET, base64 undone. */ +function decode(b64: string): Uint8Array { + return Uint8Array.from(atob(b64), (c) => c.charCodeAt(0)) +} + +/** A daemon's push socket, as a harness Leg. */ +async function socket(): Promise { + const res = await SELF.fetch(DIRECTORY, { headers: { Upgrade: 'websocket', ...AUTH } }) + expect(res.status).toBe(101) + const ws = res.webSocket! + // The pool's test-side socket defaults to delivering binary as Blob; ask for + // ArrayBuffer so a push can be compared synchronously. + ;(ws as unknown as { binaryType: string }).binaryType = 'arraybuffer' + ws.accept() + return new Leg(ws) +} + +/** A stub env for the router unit tests, shaped like machineid.test.ts's: the + * directory namespace answers with a status nothing else in the Worker uses, + * so "reached the object" is unmistakable. */ +function stubEnv(overrides: Partial): Env { + return { + DAEMON_SECRET: TEST_SECRET, + DIRECTORY: { + idFromName: (name: string) => name, + get: () => ({ fetch: async () => new Response('reached the directory', { status: 299 }) }), + }, + ASSETS: { fetch: async () => new Response('spa', { status: 200 }) }, + ...overrides, + } as unknown as Env +} + +const denyAll = { limit: async () => ({ success: false }) } + +describe('PUT /directory', () => { + it('refuses a write without the daemon secret: 401', async () => { + const res = await put(blob('unauthorized'), {}) + expect(res.status).toBe(401) + }) + + it('refuses a write with the wrong secret: 401', async () => { + const res = await put(blob('wrong-secret'), { Authorization: 'Bearer nope' }) + expect(res.status).toBe(401) + }) + + it('stores a blob and answers 201 with the key it filed it under', async () => { + const body = blob('stored') + const res = await put(body) + expect(res.status).toBe(201) + expect(res.headers.get('Content-Type')).toBe('application/json') + expect(res.headers.get('Cache-Control')).toBe('no-store') + // Content-addressed: the key is SHA-256 of the exact bytes and nothing + // else, which is what makes a PUT unable to displace another entry. + expect(await res.json()).toEqual({ key: await digest(body) }) + }) + + it('is idempotent: the same bytes again answer 200 and add no second entry', async () => { + const body = blob('idempotent') + expect((await put(body)).status).toBe(201) + const before = (await get()).entries.length + const again = await put(body) + // 200 rather than 201 — same body, so a caller need not branch; the status + // is the whole of "you were not the first". + expect(again.status).toBe(200) + expect(await again.json()).toEqual({ key: await digest(body) }) + expect((await get()).entries.length).toBe(before) + }) + + it('refuses an empty body: 400, and spends no entry on it', async () => { + const res = await put(new Uint8Array(0)) + expect(res.status).toBe(400) + expect(await res.json()).toEqual({ error: 'empty blob' }) + }) + + it('takes a blob of exactly the cap and refuses one byte more: 413', async () => { + const at = new Uint8Array(MAX_BLOB_BYTES).fill(0x41) + expect((await put(at)).status).toBe(201) + const over = new Uint8Array(MAX_BLOB_BYTES + 1).fill(0x42) + const res = await put(over) + expect(res.status).toBe(413) + expect(await res.json()).toEqual({ error: 'blob too large' }) + // And nothing of the oversized blob was kept. + const keys = (await get()).entries.map((e) => e.key) + expect(keys).not.toContain(await digest(over)) + }) + + it('refuses an oversized chunked body too, where Content-Length says nothing', async () => { + // A body streamed without a declared length is the case a Content-Length + // check alone misses, and buffering it is the memory DoS readCapped + // exists to stop. + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(MAX_BLOB_BYTES).fill(0x43)) + controller.enqueue(new Uint8Array(64).fill(0x44)) + controller.close() + }, + }) + const res = await SELF.fetch(DIRECTORY, { + method: 'PUT', + body: stream, + headers: AUTH, + // Required by the fetch spec for a streaming body. + duplex: 'half', + } as RequestInit) + expect(res.status).toBe(413) + }) +}) + +describe('GET /directory', () => { + it('is credential-less: no secret, still the full set', async () => { + const body = blob('credential-less') + await put(body) + const res = await SELF.fetch(DIRECTORY) + expect(res.status).toBe(200) + expect(res.headers.get('Content-Type')).toBe('application/json') + expect(res.headers.get('Cache-Control')).toBe('no-store') + const doc = (await res.json()) as { v: number; entries: { key: string; blob: string }[] } + expect(doc.v).toBe(1) + expect(doc.entries.map((e) => e.key)).toContain(await digest(body)) + }) + + it('round-trips a blob byte for byte, whatever is in it', async () => { + // The load-bearing test of the whole leg. Parts B and C verify Ed25519 + // signatures over *these* bytes: a relay that re-encoded, trimmed, or + // ran a blob through UTF-8 anywhere would break every signature over it + // while believing it had been helpful. So the payload is deliberately not + // text — a NUL, a lone 0xFF that is not valid UTF-8, and a byte pattern + // that base64 has to pad. + const body = Uint8Array.of(0x00, 0xff, 0xfe, 0x0a, 0x7b, 0x22, 0x80, 0x01, 0x02) + const res = await put(body) + expect(res.status).toBe(201) + const key = await digest(body) + const entry = (await get()).entries.find((e) => e.key === key) + expect(entry).toBeDefined() + expect([...decode(entry!.blob)]).toEqual([...body]) + // And the key is still the digest of what came back: bytes the relay had + // altered could not hash to the name it filed them under. + expect(await digest(decode(entry!.blob))).toBe(key) + }) + + it('keys every entry by the digest of its own blob', async () => { + await put(blob('keyed-by-digest')) + const entries = (await get()).entries + expect(entries.length).toBeGreaterThan(0) + for (const e of entries) expect(await digest(decode(e.blob))).toBe(e.key) + }) + + it('sits behind the rate rule, like /client and POST /api/pair', async () => { + const res = await worker.fetch( + new Request(DIRECTORY), + stubEnv({ CLIENT_RATE: denyAll }), + ) + expect(res.status).toBe(429) + expect(await res.json()).toEqual({ error: 'rate limited' }) + }) + + it('does not meter the secret-holding legs: they are gated, not throttled', async () => { + const upgrade = await worker.fetch( + new Request(DIRECTORY, { headers: { Upgrade: 'websocket', ...AUTH } }), + stubEnv({ CLIENT_RATE: denyAll }), + ) + expect(upgrade.status).toBe(299) + const write = await worker.fetch( + new Request(DIRECTORY, { method: 'PUT', body: 'blob', headers: AUTH }), + stubEnv({ CLIENT_RATE: denyAll }), + ) + expect(write.status).toBe(299) + }) + + it('meters an anonymous upgrade too: the 401 is not a cheaper road past the rule', async () => { + // Without this, `Upgrade: websocket` would be a way to spend the Worker's + // billed requests on /directory without ever presenting a credential. + const res = await worker.fetch( + new Request(DIRECTORY, { headers: { Upgrade: 'websocket' } }), + stubEnv({ CLIENT_RATE: denyAll }), + ) + expect(res.status).toBe(429) + }) + + it('fails open when the rate binding is absent, as the other routes do', async () => { + const res = await worker.fetch(new Request(DIRECTORY), stubEnv({ CLIENT_RATE: undefined })) + expect(res.status).toBe(299) + }) +}) + +describe('the directory socket', () => { + it('refuses an upgrade without the daemon secret: 401', async () => { + const res = await SELF.fetch(DIRECTORY, { headers: { Upgrade: 'websocket' } }) + expect(res.status).toBe(401) + }) + + it('pushes a stored blob to every connected daemon, byte for byte', async () => { + const a = await socket() + const b = await socket() + const body = Uint8Array.of(0x00, 0xff, 0x10, 0x20, 0x30) + expect((await put(body)).status).toBe(201) + // One binary message of exactly the blob's bytes: no envelope, because an + // envelope is a chance to reshape something the relay cannot read. + expect([...new Uint8Array((await a.next('a push')) as ArrayBuffer)]).toEqual([...body]) + expect([...new Uint8Array((await b.next('a push')) as ArrayBuffer)]).toEqual([...body]) + a.ws.close() + b.ws.close() + }) + + it('pushes nothing for a duplicate PUT: the set did not change', async () => { + const body = blob('pushed-once') + await put(body) + const daemon = await socket() + // The socket is opened after the first PUT, so the only push it could see + // is one the duplicate produced. + await put(body) + // A blob that *does* change the set, sent second, is the fence: if the + // duplicate had pushed, this assertion would read its bytes instead. + const fresh = blob('pushed-once-fence') + await put(fresh) + expect([...new Uint8Array((await daemon.next('the fence push')) as ArrayBuffer)]).toEqual([ + ...fresh, + ]) + daemon.ws.close() + }) + + it('closes a daemon that speaks on it: the socket is push-only', async () => { + const daemon = await socket() + daemon.ws.send('hello') + expect(await within(daemon.closed, 'the socket to close')).toEqual({ + code: 1002, + reason: 'the directory socket is push-only', + }) + }) + + it('survives a flue-pong without closing', async () => { + const daemon = await socket() + daemon.ws.send('flue-pong') + const body = blob('after-a-pong') + await put(body) + expect([...new Uint8Array((await daemon.next('a push')) as ArrayBuffer)]).toEqual([...body]) + daemon.ws.close() + }) + + it('answers flue-ping from the edge auto-response, without waking the object', async () => { + const daemon = await socket() + daemon.ws.send('flue-ping') + expect(await daemon.next('the pong')).toBe('flue-pong') + daemon.ws.close() + }) +}) + +describe('the directory routes', () => { + it('404s a path under the prefix: the Worker owns it, the SPA does not answer', async () => { + const res = await SELF.fetch(`${BASE}/directory/anything`) + expect(res.status).toBe(404) + // Not the machine 404: nothing here names a machine, and saying so would + // be a lie about what was asked. + expect(await res.json()).toEqual({ error: 'not found' }) + }) + + it('405s a method the leg does not have', async () => { + const res = await SELF.fetch(DIRECTORY, { method: 'DELETE', headers: AUTH }) + expect(res.status).toBe(405) + expect(res.headers.get('Allow')).toBe('GET, PUT') + expect(await res.json()).toEqual({ error: 'method not allowed' }) + }) + + it('never confuses the directory with a machine: /directory is not an id', async () => { + // The two prefixes cannot collide — "directory" carries no MAC tag and + // could not be a machine id — but the check is cheap and the failure mode + // (a fleet's certs served out of a machine hub) would be silent. + const res = await SELF.fetch(`${BASE}/client/${await machineId('notdir-0a0a')}`, { + headers: { Upgrade: 'websocket' }, + }) + expect(res.status).toBe(503) + expect(await res.json()).toEqual({ error: 'daemon offline' }) + }) + + it('503s when the directory binding is absent: an older deploy than this script', async () => { + const res = await worker.fetch(new Request(DIRECTORY), stubEnv({ DIRECTORY: undefined })) + expect(res.status).toBe(503) + expect(await res.json()).toEqual({ error: 'directory unavailable' }) + }) + + it('binds one directory for the whole relay', async () => { + // One relay is one fleet: the name is a constant in the router, so two + // reads of it are two reads of the same object. The binding has to exist + // in the pool for that to mean anything. + expect(env.DIRECTORY).toBeDefined() + expect(typeof env.DIRECTORY.idFromName).toBe('function') + }) +}) + +/** + * The cap, tested against its own object rather than the shared one: filling + * the fleet directory would leave every other test in this file reading 512 + * entries. This dials the Durable Object directly, which is the one place in + * this suite the router is not the thing under test. + */ +describe('the entry cap', () => { + it('refuses a PUT past MAX_ENTRIES with 507, and keeps taking duplicates', async () => { + const full = env.DIRECTORY.get(env.DIRECTORY.idFromName(`full-${crypto.randomUUID()}`)) + const write = (body: Uint8Array): Promise => + full.fetch(DIRECTORY, { method: 'PUT', body, headers: AUTH }) + // Fill it. Small blobs — the cap under test is the count, not the bytes. + const first = blob('cap', '0') + expect((await write(first)).status).toBe(201) + for (let i = 1; i < MAX_ENTRIES; i++) expect((await write(blob('cap', String(i)))).status).toBe(201) + // One more distinct blob is refused, with a status nothing else on this + // leg wears: a daemon has to tell "your blob is fine and I will not keep + // it" from 413 (this blob is wrong) and 401 (this caller is wrong). + const over = blob('cap', 'over') + const res = await write(over) + expect(res.status).toBe(507) + expect(await res.json()).toEqual({ error: 'directory full' }) + // Refused, not evicted — a directory that dropped an old entry to take a + // new one could drop a revocation, and a forgotten revocation re-admits + // the device it revoked. + const doc = (await (await full.fetch(DIRECTORY)).json()) as { + entries: { key: string; blob: string }[] + } + expect(doc.entries.length).toBe(MAX_ENTRIES) + expect(doc.entries.map((e) => e.key)).toContain(await digest(first)) + expect(doc.entries.map((e) => e.key)).not.toContain(await digest(over)) + // And a re-PUT of something already stored still succeeds at the cap: it + // asks for no room, so refusing it would strand a daemon that re-announces + // on every reconnect. + expect((await write(first)).status).toBe(200) + }) +}) diff --git a/relay/wrangler.jsonc b/relay/wrangler.jsonc index 1a449b8..efca82b 100644 --- a/relay/wrangler.jsonc +++ b/relay/wrangler.jsonc @@ -22,13 +22,41 @@ "not_found_handling": "single-page-application", // The bare entries matter: "/daemon/*" alone would let the asset router // answer a bare /daemon with the SPA before the Worker's 404 could — the - // Worker keeps refusal authority over its whole prefix, ids or not. + // Worker keeps refusal authority over its whole prefix, ids or not. The + // same goes for /directory, where the bare path *is* the route and the + // starred form is the Worker's own 404. // internal/relaydeploy (RunWorkerFirst) ships the same list in the // deploy request `flue relay setup` builds; edit both or neither. - "run_worker_first": ["/daemon", "/daemon/*", "/client", "/client/*", "/api/*"] + "run_worker_first": [ + "/daemon", + "/daemon/*", + "/client", + "/client/*", + "/api/*", + "/directory", + "/directory/*" + ] }, - "durable_objects": { "bindings": [{ "name": "HUB", "class_name": "DaemonHub" }] }, - "migrations": [{ "tag": "v1", "new_sqlite_classes": ["DaemonHub"] }], + // Two classes: DaemonHub is one object per machine, FleetDirectory is one + // object for the relay (spec/fleet-trust.md, "The fleet directory"). + // + // The deploy twin owes three things to the directory — this binding, the v2 + // migration below, and the two "/directory" entries above — and does not + // have them yet: internal/relaydeploy gains them with the daemon side of + // this feature, which is what will first need a deployed relay to answer + // /directory. Until then this file is the only place the object exists, so + // `pnpm dev` and the vitest pool serve it and `flue relay setup` does not + // (the Worker answers 503 there rather than throwing — src/index.ts). + "durable_objects": { + "bindings": [ + { "name": "HUB", "class_name": "DaemonHub" }, + { "name": "DIRECTORY", "class_name": "FleetDirectory" } + ] + }, + "migrations": [ + { "tag": "v1", "new_sqlite_classes": ["DaemonHub"] }, + { "tag": "v2", "new_sqlite_classes": ["FleetDirectory"] } + ], // The per-IP bound on the credential-less routes (/client/*, POST // /api/pair/*): generous enough that a fleet of tabs never sees it, tight // enough that burning the daily request allowance needs a botnet diff --git a/spec/relay-protocol.md b/spec/relay-protocol.md index dd04d8d..1a662c9 100644 --- a/spec/relay-protocol.md +++ b/spec/relay-protocol.md @@ -1,7 +1,8 @@ # flue relay protocol -The relay is a Cloudflare Worker (one Durable Object per machine) that bridges -each daemon's single outbound socket to any number of browser tabs. It +The relay is a Cloudflare Worker (one Durable Object per machine, plus one for +the fleet directory) that bridges each daemon's single outbound socket to any +number of browser tabs. It forwards bytes and nothing else: it holds no Noise keys, reads no terminal traffic, and cannot tell one keystroke from another. What it *does* see — the control channel is cleartext — is set out under "What the relay sees" below. @@ -10,6 +11,11 @@ This document defines the two sockets that meet at the relay and the framing on each. It does **not** redefine the wire protocol — `spec/protocol.md` is unchanged and travels inside, encrypted. The relay is a transport for it. +It also defines one thing the relay *stores* rather than forwards: the fleet +directory, a set of signed blobs it cannot verify (below). That leg holds no +key either, and the rule it lives under is the same one this whole document +turns on — the relay's only power is availability. + ``` daemon ---- wss /daemon/ ----> Worker + one DO per machine <---- wss /client/ ---- browser [4B channel][payload] [payload] @@ -26,6 +32,25 @@ Three layers stack, outermost first: | Noise IK | handshake messages, then transport ciphertexts | daemon and browser | | Kind framing | `[1 byte kind][wire protocol bytes]` | daemon and browser | +The directory leg stacks none of them. It is HTTP plus one push socket, and +what travels on it is a signed blob in whatever encoding the fleet key signs +(`spec/fleet-trust.md`, Certificates) — bytes the relay stores, serves and +forwards without a frame of its own around them, because it cannot read them +and any framing it added would be a chance to reshape what a reader is about +to check a signature over. + +The routes, in one place: + +| Route | Auth | Metered | Object | +|---|---|---|---| +| `WS /daemon/` | Bearer daemon secret | no | that machine's hub | +| `WS /client/` | none | yes | that machine's hub | +| `POST /api/pair/` | none | yes | that machine's hub | +| `PUT /directory` | Bearer daemon secret | no | the directory | +| `GET /directory` | none | yes | the directory | +| `WS /directory` | Bearer daemon secret | no | the directory | +| `GET /api/health` | none | no | none — the Worker alone | + ## The daemon leg The daemon dials `wss:///daemon/` **outbound** — nothing @@ -148,15 +173,118 @@ distinction would be lost. This byte carries it, and the layer above the relay reads the same `(text, data)` pair it reads locally. An empty payload, or any kind byte other than `0x00` or `0x01`, is a protocol error. +## The fleet directory + +One more Durable Object, and the only one that is not per machine: the fleet +directory, `idFromName("directory")`, because one relay is one fleet +(`spec/fleet-trust.md`, "The fleet directory"). It holds the signed artifacts +that have to reach every machine and every device — machine certs, device +certs, revocations — and it holds them as **blobs it cannot verify**, because +the fleet key never touches the Worker. + +That is the invariant this leg exists to preserve, and it must survive every +future change to it: **the relay stores and serves, and never verifies.** Every +reader — daemon and browser both — checks every signature under the fleet +public key and drops what fails. A hostile relay can therefore serve a stale, +truncated or empty directory, exactly as it could always refuse to route; what +it cannot do is mint a machine or a device, because minting needs a key it does +not hold. The cost of a hostile relay stays availability, and nothing else. + +``` +PUT /directory Bearer daemon secret; body is one signed blob, raw bytes +GET /directory credential-less, rate limited; the whole set +WS /directory Bearer daemon secret; relay → daemon pushes, on write +``` + +Nothing lives under the prefix: `/directory/` is the Worker's own +`404 {"error":"not found"}` — not the machine 404, because nothing here names a +machine — and a method other than `GET` or `PUT` is `405` with +`Allow: GET, PUT`. + +**Entries are content-addressed.** An entry's key is the lowercase hex SHA-256 +of the blob's exact bytes, and there is no other name for it. That is the only +key a Worker that cannot read a blob is entitled to compute: a caller-supplied +name would put the relay in charge of a namespace it cannot check, and one +buggy or hostile secret-holder could then PUT a machine cert over a revocation +with the relay's help. Content addressing makes that structurally impossible — +a PUT can only ever *add* — and two properties fall out of it: + +- **Idempotence.** The same bytes PUT again are the same key and the same + value: no second entry, no push. A daemon may re-announce everything it holds + on every reconnect, and a PUT replayed from the wire changes nothing. +- **Byte-exactness.** A blob comes back exactly as it went in. Bytes the relay + had altered would not hash to the name it filed them under, and every reader + verifies a signature over these bytes. + +`PUT` answers `201 {"key":""}` when it created the entry and +`200 {"key":""}` when the blob was already there; the body is the same +either way, so the status is the whole of the difference. An empty body is +`400 {"error":"empty blob"}` — not a signed anything under any encoding. + +`GET` answers, as `application/json`, no-store: + +```json +{ "v": 1, "entries": [ { "key": "", "blob": "" } ] } +``` + +`blob` is standard base64 with padding — the alphabet Go's `encoding/json` +reads a `[]byte` field in, so a daemon's struct decodes it with no help. **No +order is promised.** Storage hands entries back sorted by key, which is to say +by digest, which is to say by nothing; a reader that inferred "newest last" +would be reading a property of SHA-256. Ordering that means anything lives +inside the blobs, in `iat`, where a reader that has checked a signature can +trust it. Neither is any *ranking* the relay's: a revocation outranks a device +cert for the same key whatever their timestamps, and that is a rule about +meaning, which readers hold and the relay does not. + +**The push socket** carries one binary message per new entry, and that message +is exactly the blob's bytes — no envelope, no key, no framing, for the reason +the whole leg is unframed. The fan-out reaches every connected daemon including +the one whose machine made the PUT (an HTTP request carries no socket identity, +and a push a receiver already holds is a no-op). Duplicates push nothing: the +set did not change. + +Unlike `/daemon`, there is no takeover here: every machine in the fleet holds +one of these at once, which is the point. At most 256 are held, answered +`503 {"error":"too many directory sockets"}` over that — not a DoS bound, since +the leg is secret-gated, but a bound on what one PUT costs, which is one send +per socket. + +Pushes are best effort, and the socket is **push-only** — a write is `PUT +/directory`, an HTTP request that answers with the key it filed, so any message +from a daemon on this socket other than the keepalive below is a protocol error +and closes it with `1002`. A daemon converges the rest of the way by reading +`GET /directory`, and the order that closes the gap is: **open the socket +first, then GET.** A write that lands in between arrives by one path or the +other — or by both, which content addressing makes harmless. + +**Bounds.** The directory is written by secret-holders and readable without a +credential, so it is bounded at both ends: a blob is at most **4 KiB** +(`413 {"error":"blob too large"}`), and the directory holds at most **512** +entries. At the cap a PUT of a new blob is refused with +`507 {"error":"directory full"}` — its own status, so a daemon can tell "your +blob is fine and I will not keep it" from a 413 or a 401 without reading prose +— while a PUT of a blob already stored still answers 200, because it asks for +no room. + +Refusing rather than evicting is deliberate and is a security decision, not a +capacity one. Every eviction policy can drop a revocation, and a directory that +silently forgets a revocation re-admits the device it revoked to every machine +that had not yet heard. Nothing in this leg ever deletes an entry. If pruning is +ever wanted — a device cert whose key is revoked, say — it has to be a decision +signed under the fleet key and carried out by something that can read what it is +deleting. That is not the relay, and it must not become the relay. + ## Keepalive -Either leg may send the **text** frame `flue-ping`; the Cloudflare edge answers +Any socket the relay holds — either leg of a machine hub, and the directory +socket — may send the **text** frame `flue-ping`; the Cloudflare edge answers `flue-pong` from the Durable Object's auto-response, without waking it. No leg ever has to answer a ping itself — the edge does, so neither the daemon nor a browser sees one — and a received `flue-pong` is dropped silently. Neither string ever carries a channel header: they are text frames, and everything -channel-framed is binary. Any other text frame on either socket is a protocol -error. +channel-framed is binary. Any other text frame is a protocol error, on all +three. This is the one place the relay adds something `spec/protocol.md` says does not exist ("WebSocket ping/pong frames only. There is no application-level ping"). @@ -181,10 +309,18 @@ Object bounds it directly: a cap on concurrent channels, a deadline on completing the handshake, after which the channel is closed, and a cap on the size of one client message. +The directory's legs are gated by the same one secret: `PUT /directory` and the +`/directory` upgrade present the bearer, and `GET /directory` presents nothing, +for the same reason `/client` presents nothing — what the directory holds is +signed, public by design, and worthless to forge without the fleet key. The +secret is what says who may *write*, and it is all the Worker has to say it +with, since it cannot verify a single blob it stores. + Both legs, and `POST /api/pair`, carry a **machine id** in the path — `/daemon/`, `/client/`, `/api/pair/` — and the Worker routes on it: `idFromName(id)` selects that machine's Durable Object, and the hub receives -the bare prefix, never the id. The id is **self-certifying**: +the bare prefix, never the id. (`/directory` carries none: one relay is one +fleet, so its object's name is a constant.) The id is **self-certifying**: ``` machine-id = "-" @@ -221,11 +357,15 @@ The tag changes none of that — it authenticates the *mint*, not the caller. The real id is semi-public (it rides pairing links), and the Worker bills every request before any of this runs, so the credential-less routes also sit behind a **rate rule**: one Cloudflare rate-limiting binding, keyed by -connecting IP, over `/client/*` and `POST /api/pair/*` — 300 requests per +connecting IP, over `/client/*`, `POST /api/pair/*` and every request to +`/directory` that does not present the secret — 300 requests per 60 s per IP per Cloudflare location, answered `429 {"error":"rate limited"}` over it. Generous enough that a fleet of tabs never sees it; tight enough -that burning quota, or walking the tag space, needs a botnet. The daemon leg -is not rate limited: it is secret-gated and one socket per machine. The rule +that burning quota, or walking the tag space, needs a botnet. The daemon legs +are not rate limited: they are secret-gated, one socket per machine. The +metering on `/directory` is decided by the credential rather than the method +on purpose — otherwise an anonymous caller waving an `Upgrade` header would +have a cheaper road to the 401 than the metered `GET` beside it. The rule is fail-open by design — a Worker deployed without the binding routes rather than refuses — because it bounds cost, not access. @@ -256,7 +396,14 @@ observe is: - channel ids, message counts and message sizes — enough for traffic analysis of a session, not its content; - the whole pairing exchange: the single-use pairing token, the device's public - key, and the daemon's. + key, and the daemon's; +- everything in the fleet directory. The blobs are opaque to the *code*, not to + the operator of the machine running it: machine ids and display names, device + public keys and their names, and who revoked what and when are all in there, + signed rather than secret. The relay already routed by machine id; the delta + is names and device keys. One operator on their own Worker is why that is + acceptable, and `docs/RELAY.md` states it rather than leaving it to be + discovered. Public-key material is public by design. The **token is not**: a hostile relay could spend a live one with a key of its own and register itself as a paired From d382196478945370ce734c168213140009a0fd92 Mon Sep 17 00:00:00 2001 From: Karn Date: Mon, 10 Aug 2026 14:31:31 +0530 Subject: [PATCH 02/29] =?UTF-8?q?feat(fleet):=20the=20daemon=20side=20of?= =?UTF-8?q?=20the=20directory=20=E2=80=94=20publish,=20verify,=20and=20a?= =?UTF-8?q?=20kill=20switch=20that=20crosses=20machines?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The relay grew a fleet directory in the last commit and nothing spoke to it. This is the other end: the deploy that ships the object, the leg that keeps a machine converged with its fleet, and the rule that a revocation performed on one machine closes the device on all of them. **The deploy.** `flue relay setup` was still deploying a Worker whose /directory answered 503, because the Cloudflare client could only ever say "introduce these classes" — and that is not expressible for a relay already sitting at migration v1 that needs FleetDirectory at v2. So the client now carries the migration *history* and reads the tag the account's copy of the script already has (GET /accounts//workers/scripts, the same fact wrangler reads from the same place), sending only the steps behind it. A fresh account runs v1 then v2; an existing relay runs v2 alone, against the precondition that it is still at v1; an up-to-date one sends no migration at all, which is one fewer script upload than the blind send-and-recover this replaces. migrationAlreadyApplied stays as the belt for the paths the read cannot cover. The DIRECTORY binding and the two /directory run_worker_first entries land beside it, and wrangler.jsonc's comment about what the deploy still owed is now a note that it owes nothing. **The leg.** internal/transport/relay/directory.go: socket first, snapshot second (a write in between arrives by one path or both, and ingest is idempotent), then everything this machine holds, re-offered on every connect and every half hour. Nothing the relay says is trusted — every blob, by GET or by push, is verified under the fleet public key and dropped if it fails — and nothing about the relay's own bounds is taken on faith: 4 KiB a blob, 512 entries a snapshot, 4 MiB a body, enforced here as well as there. **Publishing.** The machine cert is minted where the facts it asserts are decided — `flue relay setup`, `flue relay join`, the Remote screen's deploy — and stored in relay.json, because the directory is content-addressed and a cert re-signed at every boot would spend an entry per daemon restart. Device certs go out as the ceremony mints them, revocations as the operator makes them, and the leg refuses to publish a machine cert that no longer names this daemon's static key: a stale one is a machine every device dials and none can handshake with. **Ingest: revocations only.** A verified device cert read from the directory is deliberately *not* written to the local registry. Possession of a public blob proves nothing; the IK handshake is what proves a browser holds the key, which is why rule 2 of the acceptance order writes the row at that moment. A registry filled from the directory would be honoured by rule 1, which never looks at a cert — quietly turning "the fleet vouches for this key today" into "this machine pairs with it forever", surviving the fleet-key rotation meant to withdraw it. Revocations are the opposite: they only subtract authority, so honouring one from an untrusted channel can never grant access. A verified one reaches AddRevocation, drops the registry row by key (not by the 48-bit id), and closes that device's channels with the existing revoked{reason} flow. **Status.** `flue relay status` now reads the directory itself and reports what it is holding, how much of it this fleet key signed, and whether this machine is in there at all — the one fault a new machine actually has. The Remote screen's /api/relay/info carries the same counts. Tested against a fake directory that lies in every way a real one could: a blob signed by another fleet key is dropped, a truncated or extended or bit-flipped one is dropped, an oversized one is dropped unread, a revocation published elsewhere closes the device here, ingest is idempotent across both paths, 507 and 413 are loud and do not wedge the leg, and a relay serving a truncated or empty set costs freshness and nothing else. Co-Authored-By: Claude Fable 5 --- cmd/flue/main.go | 39 +- cmd/flue/main_test.go | 53 +- cmd/flue/relay.go | 238 +++++- cmd/flue/relay_test.go | 347 +++++++- cmd/flue/relayui.go | 10 + internal/cloudflare/client.go | 194 ++++- internal/cloudflare/client_test.go | 268 +++++- .../cloudflare/testdata/deploy_metadata.json | 12 +- internal/config/relay.go | 21 + internal/config/relay_test.go | 10 +- internal/crypto/devices.go | 31 + internal/crypto/devices_test.go | 45 + internal/daemon/pairing.go | 8 + internal/daemon/relayui.go | 52 ++ internal/daemon/server.go | 135 ++- internal/daemon/server_test.go | 287 +++++++ internal/relaydeploy/deploy.go | 59 +- internal/transport/relay/directory.go | 797 ++++++++++++++++++ internal/transport/relay/directory_test.go | 755 +++++++++++++++++ .../transport/relay/fake_directory_test.go | 319 +++++++ internal/transport/relay/relay.go | 16 +- relay/wrangler.jsonc | 16 +- 22 files changed, 3581 insertions(+), 131 deletions(-) create mode 100644 internal/transport/relay/directory.go create mode 100644 internal/transport/relay/directory_test.go create mode 100644 internal/transport/relay/fake_directory_test.go diff --git a/cmd/flue/main.go b/cmd/flue/main.go index a07d806..bbde581 100644 --- a/cmd/flue/main.go +++ b/cmd/flue/main.go @@ -411,12 +411,49 @@ func startRelay(ctx context.Context, srv *daemon.Server, identity daemon.Identit return false } } - cfg := relay.Config{URL: rc.URL, Secret: rc.Secret, Origin: rc.Origin, MachineID: rc.MachineID, FleetPub: fleetKey.Public()} + cfg := relay.Config{ + URL: rc.URL, + Secret: rc.Secret, + Origin: rc.Origin, + MachineID: rc.MachineID, + MachineCert: rc.MachineCert, + FleetPub: fleetKey.Public(), + } t, err := relay.New(cfg, srv, identity.Key, identity.Devices, logger) if err != nil { logger.Warn("relay not started", "err", err) return false } + // The directory leg, beside the hub leg and independent of it: one keeps + // this machine's browsers connected, the other keeps this machine's idea + // of who the fleet trusts up to date (spec/fleet-trust.md, "The fleet + // directory"). It is started here rather than inside the transport + // because neither needs the other — a daemon whose hub socket is down + // still has to hear a revocation, and one whose directory is down still + // serves every device paired to it. + // + // A directory this daemon cannot build is a warning and nothing more, for + // the reason every fault in this function is: flue's promise is a terminal + // in a browser tab, and the fleet is what makes that tab openable from the + // next machine along. + dir, err := relay.NewDirectory(cfg, srv, identity.Key, identity.Devices, logger) + if err != nil { + logger.Warn("fleet directory not started", "err", err) + } else { + // Installed before the goroutine, so a pairing or a revoke that + // happens while the first dial is still in flight is queued rather + // than lost. + srv.SetFleetPublisher(dir) + srv.SetDirectoryCounts(func() daemon.DirectoryCounts { + c := dir.Counts() + return daemon.DirectoryCounts(c) + }) + go func() { + if err := dir.Run(ctx); err != nil { + logger.Warn("fleet directory stopped", "err", err) + } + }() + } // The machine's identity rides every welcome alongside the status, so the // UI can build /client/ URLs for this machine. It is configuration // rather than socket state, which is why it is set here — once, by the diff --git a/cmd/flue/main_test.go b/cmd/flue/main_test.go index 4d76f6f..9522986 100644 --- a/cmd/flue/main_test.go +++ b/cmd/flue/main_test.go @@ -1297,16 +1297,27 @@ func TestStatusReportsAConfiguredRelayWithoutItsSecret(t *testing.T) { func TestStartRelayDialsAConfiguredRelay(t *testing.T) { t.Setenv("XDG_CONFIG_HOME", t.TempDir()) - var attempts atomic.Int64 - var auth, path atomic.Value + // Both legs dial the same host, so what is recorded is per path: the hub + // leg at /daemon/ and the fleet directory at /directory. Recording a + // single "last dial" would be a race between two goroutines that both + // start here. + var mu sync.Mutex + dialed := map[string]string{} // path -> Authorization ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - auth.Store(r.Header.Get("Authorization")) - path.Store(r.URL.Path) - attempts.Add(1) + mu.Lock() + dialed[r.URL.Path] = r.Header.Get("Authorization") + mu.Unlock() http.Error(w, "unauthorized", http.StatusUnauthorized) })) defer ts.Close() + sawDial := func(path string) (string, bool) { + mu.Lock() + defer mu.Unlock() + auth, ok := dialed[path] + return auth, ok + } + const secret = "s3cr3t-daemon-secret" if err := config.SaveRelay(config.Relay{ URL: "ws" + strings.TrimPrefix(ts.URL, "http"), @@ -1336,20 +1347,26 @@ func TestStartRelayDialsAConfiguredRelay(t *testing.T) { defer cancel() startRelay(ctx, srv, id) - deadline := time.Now().Add(3 * time.Second) - for attempts.Load() == 0 { - if time.Now().After(deadline) { - t.Fatal("the daemon never dialled the configured relay") + // One relay.json, two legs: the hub the browsers arrive on, and the fleet + // directory the revocations arrive on. The machine id from relay.json + // rides the hub path — it is how the Worker knows which machine's hub this + // socket is — and the directory has no id in its path at all, because one + // relay is one fleet. + for _, want := range []string{"/daemon/karns-macbook-pro-a1b2-0f9a12cd", "/directory"} { + deadline := time.Now().Add(3 * time.Second) + for { + auth, ok := sawDial(want) + if ok { + if auth != "Bearer "+secret { + t.Errorf("Authorization on %s = %q, want %q", want, auth, "Bearer "+secret) + } + break + } + if time.Now().After(deadline) { + t.Fatalf("the daemon never dialled %s", want) + } + time.Sleep(5 * time.Millisecond) } - time.Sleep(5 * time.Millisecond) - } - if got, want := auth.Load().(string), "Bearer "+secret; got != want { - t.Errorf("Authorization = %q, want %q", got, want) - } - // The machine id from relay.json rides the dial path — it is how the - // Worker knows which machine's hub this socket is. - if got, want := path.Load().(string), "/daemon/karns-macbook-pro-a1b2-0f9a12cd"; got != want { - t.Errorf("dial path = %q, want %q", got, want) } } diff --git a/cmd/flue/relay.go b/cmd/flue/relay.go index 7a3afa2..525ca26 100644 --- a/cmd/flue/relay.go +++ b/cmd/flue/relay.go @@ -3,12 +3,15 @@ package main import ( "bufio" "context" + "crypto/ed25519" "crypto/rand" + "encoding/json" "errors" "flag" "fmt" "io" "io/fs" + "net/http" "net/url" "os" "strconv" @@ -18,6 +21,7 @@ import ( "github.com/karnstack/flue/internal/cloudflare" "github.com/karnstack/flue/internal/config" + "github.com/karnstack/flue/internal/crypto" "github.com/karnstack/flue/internal/daemon" "github.com/karnstack/flue/internal/fleet" "github.com/karnstack/flue/internal/relaydeploy" @@ -298,6 +302,17 @@ func runRelaySetup(w io.Writer, r io.Reader, api *cloudflare.Client, args []stri machineID := config.MintMachineID(hostname, secret, rand.Reader) machineName := truncateRunes(hostname, machineNameMaxRunes) + // The machine's own certificate, signed under the fleet key just minted: + // what the daemon publishes to the relay's fleet directory so every + // browser in the fleet can reach this machine without pairing to it + // (spec/fleet-trust.md, "The fleet directory"). Not fatal — a relay whose + // machines cannot be discovered still carries every device paired + // directly to them — and the line says which half is missing. + machineCert, err := mintMachineCert(fleetKey, machineID, machineName) + if err != nil { + fmt.Fprintf(w, " could not mint this machine's fleet certificate (%v); other devices will not discover this machine\n", err) + } + // Last, deliberately. relay.json is what makes the daemon dial, and every // step above can fail; writing it earlier would leave a daemon dialling a // relay that was never finished. Re-running setup is the fix for anything @@ -316,6 +331,7 @@ func runRelaySetup(w io.Writer, r io.Reader, api *cloudflare.Client, args []stri Origin: origin, MachineID: machineID, MachineName: machineName, + MachineCert: machineCert, Worker: worker, }); err != nil { return fmt.Errorf("save the relay configuration: %w", err) @@ -353,6 +369,47 @@ func runRelaySetup(w io.Writer, r io.Reader, api *cloudflare.Client, args []stri // a machine list rendering as a list. const machineNameMaxRunes = 64 +// mintMachineCert signs this machine's fleet machine certificate: the id it +// holds on the relay, its display name, and the Noise static key every device +// that reaches it must pin (spec/fleet-trust.md, Certificates). +// +// It is minted here — in the three commands that write relay.json — rather +// than by the daemon, and the reason is the directory's shape rather than +// convenience. The relay stores blobs by the hash of their own bytes, so a +// cert re-minted at each start would carry a fresh `iat`, land under a fresh +// key, and spend one of the directory's 512 entries every time the daemon +// restarted. Minting it exactly where the facts it asserts are decided means +// one blob, for the life of this machine's place on this relay. +// +// It reads (and, on a machine that has never served, creates) the daemon's +// static key, which is the same key `flue serve` would load a moment later: +// the cert has to name the key devices will actually meet, and a cert naming a +// key that did not exist yet would be a browser pinning nothing. +// +// A failure is returned, not swallowed. Every caller treats it as "this +// machine joins without a machine cert" and says so — the honest half of a +// half-configured relay, exactly as `flue relay setup` already treats a fleet +// key it cannot mint. +func mintMachineCert(key fleet.Key, machineID, machineName string) ([]byte, error) { + if !key.Valid() { + return nil, fleet.ErrNoKey + } + dir, err := config.Dir() + if err != nil { + return nil, fmt.Errorf("locate the config directory: %w", err) + } + static, err := crypto.LoadOrCreateStaticKey(dir) + if err != nil { + return nil, fmt.Errorf("load the daemon static key: %w", err) + } + return key.Sign(fleet.MachineCert{ + ID: machineID, + Name: machineName, + Noise: static.Public, + IAT: time.Now().Unix(), + }) +} + const relayJoinUsage = "usage: flue relay join --secret --fleet [--name