From 4952dc3a240c5d23a3c0c5f534cc44d5862cbf0f Mon Sep 17 00:00:00 2001 From: nyblnet <306268465+nyblnet@users.noreply.github.com> Date: Sun, 13 Sep 2026 02:49:23 +0100 Subject: [PATCH 1/3] relay: broadcast as a special case of collaboration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The relay half of live broadcast (#293, the maintainer's unified design). An audience member is a collaborator holding an owner-signed `audience` invite and a per-show key; the relay routes by stream and holds the show for late joiners. Five rules, each mutation-tested: - Admission is the invite, not the token: the audience copy cannot derive the room token (it holds the show key), so the compare is skipped for the role and only the role. `w` rooms only; only while live (4002 not-live); never on a revoked invite; 4003 show-full past the cap. - Audience sockets are receive-only. Control verbs verify by ROLE against the socket's pinned writer key, never by chain membership. - Two streams. s:'aud' frames reach audience sockets only, are never persisted even when p:1 and validly signed (the shape the real client sends), and come only from writer sockets while live. The room stream — presence included — never reaches an audience socket. nav/black are signed with the stream in the text and retained; laser is unsigned and not retained. - The show lives in DO storage, not memory: hibernation evicts the object mid-connection. A late joiner is served snapshot + aud ops + nav/black and never the op log. The rig serves a joiner from a SECOND Room over the same storage. 256 KB of held ops asks the presenter to checkpoint once; twice that refuses joiners until it does. - `end` is any writer's; grace (60 s on presenter loss, cancelled by a writer's `live`) is the only unsigned path. When it fires the room SURVIVES. The DO has one alarm and it meant "wipe the room". Every timer now goes through schedule/rearm/alarm; a grace timer armed the naive way would have evaporated a live room. Asserted: grace expiry ends the show and leaves the room; the idle alarm still wipes. `ready` gains `v` (relay protocol version 2) — the one read-only way to tell a deployed relay from the last one, asked for by the lead after verifying the #452 deploy could only be confirmed by its deployment record — and `bc:1` for feature detection. The fake Durable Object the relay rigs drive is lifted into scripts/lib/relay-harness.ts and shared by test-relay-auth and the new test-relay-broadcast, so the two cannot drift on what "the relay" is. test-relay-broadcast 85 checks. Mutations: audience not receive-only 1 red; streams not isolated 4; grace running the wipe 3; stream absent from the signed text 6; aud frames falling through to the persist path 4 — that last one passed until the fixture was made realistic (p:1 + a valid signature, as the client's send() produces), which is the shape that would have put show-key ciphertext in the room's op log forever. test-relay-auth 49, test-relay-protocol 19, test-sync 45368, test-offline 33, test-sync-vouch 20, test-ci-registered 295. Deploys after #452's relay (62a12ffa), on its own. Additive to every shipped client. The client half (slides) lands separately and feature-detects. --- .github/workflows/ci.yml | 16 ++ docs/DECISIONS.md | 52 +++++ scripts/lib/relay-harness.ts | 202 +++++++++++++++++ scripts/test-relay-auth.ts | 183 ++------------- scripts/test-relay-broadcast.ts | 373 +++++++++++++++++++++++++++++++ server/sync-worker/src/worker.js | 355 +++++++++++++++++++++++++++-- 6 files changed, 1005 insertions(+), 176 deletions(-) create mode 100644 scripts/lib/relay-harness.ts create mode 100644 scripts/test-relay-broadcast.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fca46908..4eafc840 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -847,6 +847,22 @@ jobs: # in people's files. This diffs only what goes on the wire. run: node scripts/test-relay-protocol.ts + - name: relay broadcast rig + # A show is a special case of collaboration: an audience member holds + # an owner-signed `audience` invite and a per-show key, and the relay + # routes by stream. What breaks silently if this regresses: an audience + # socket admitted on the room token (it holds the show key, not the + # room key, so the compare must be skipped for the role and only the + # role); an audience socket that can send anything (the invite is on the + # same chain as a writer's, so the check is by role); an aud frame that + # reaches the room stream or the room's op log (show-key ciphertext + # persisted forever, replayed to every future collaborator as noise); a + # late joiner served the room's op log; show state kept in memory, + # which hibernation evicts mid-show; and the presenter-loss grace timer + # firing the DO's single alarm as the 30-day room WIPE. Each is a + # mutation this rig turns red on. + run: node scripts/test-relay-broadcast.ts + - name: relay authorization rig # Drives the real Room durable object with fake sockets and storage. # What it guards: the relay stamps a fanned-out frame with exactly what diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index 3c143b54..f53d070d 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -6390,3 +6390,55 @@ chance to run and it is cheap. Reconciliation for this cycle: 41 commits, 40 mapped, 1 correctly absent, run by bento-team-slides. Claude-Session: https://claude.ai/code/session_01Jcfdy8A69nonyATtm8vRy8 + +## 2026-09-13 — Broadcast is a special case of collaboration: the relay half + +**Decision.** A live show is not a second transport. An audience member is a +collaborator holding a TICKET — an owner-signed invite with role `audience`, +on the same chain as "Invite to edit" — whose `collab.key` is a per-show SHOW +KEY rather than the room key. The relay (`server/sync-worker/`) implements +the show as five rules, each guarded by `scripts/test-relay-broadcast.ts`: + +1. **Admission is the invite, not the token.** `?tok=` is a hash of the room + key and an audience copy cannot derive it, so the token compare is skipped + for the `audience` role and only for it. `w` rooms only; only while live + (else close `4002 not-live`); refused on a revoked invite; refused `4003 + show-full` when the held show has outgrown its cap. +2. **Audience sockets are receive-only.** Every frame from one is dropped. + Control verbs verify by ROLE — the socket's pinned writer key — never by + chain membership, because the audience invite is on the same chain. +3. **Two streams.** A frame tagged `s:'aud'` goes to audience sockets only, + is never persisted whatever else it carries, and comes only from writer + sockets while live. The room stream never reaches an audience socket — + including presence, in either direction. `nav`/`black` are signed with the + stream in the text and retained as latest state; `laser` is unsigned and + never retained. +4. **The relay holds the show, in DO storage.** One presenter `audsnap` on + `live` and at checkpoints; aud ops since; nav/black. A late joiner is served + that and never the op log or the room's persisted snapshot. Storage, not + memory: the Hibernation API evicts the object mid-connection, and an + in-memory show would vanish silently. Past 256 KB of held ops the presenter + is asked to checkpoint once; past twice that, joiners are refused until it + does. +5. **`end` is any writer's; grace is the only unsigned path.** Presenter + socket loss starts 60 s; a writer's `live` cancels it; audience activity + never extends it. When it fires the show ends and **the room survives**. + +**The Durable Object has one alarm, and it used to mean "wipe the room".** +Every timer now goes through one multiplexer (`schedule`/`rearm`/`alarm`): +each kind stores its due time, the DO alarm is armed to the earliest, and the +handler runs whichever are due. Arming the grace timer with a bare `setAlarm` +would have replaced the idle alarm and, on firing, run the wipe on a live +room. The rig asserts the room survives a grace expiry and that the idle +alarm still evaporates it. + +**`ready` carries `v` (relay protocol version, 2) and `bc:1`.** `v` is the +one read-only way to tell a deployed relay from the last one; every other +discriminator is a write or needs an owner key. Clients feature-detect +broadcast on `bc`. + +Deploy: after #452's relay (deployed 2026-09-13 as `62a12ffa`, from +`b1b4a67`), as its own deploy. Additive to every shipped client — none sends +`ivr=audience` or `s:'aud'`, and `v`/`bc` on `ready` are ignored by clients +that do not read them. The client half (slides) feature-detects and lands +separately. Design note: private until the client ships, then promoted. diff --git a/scripts/lib/relay-harness.ts b/scripts/lib/relay-harness.ts new file mode 100644 index 00000000..bc8a2e04 --- /dev/null +++ b/scripts/lib/relay-harness.ts @@ -0,0 +1,202 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 The Bento authors +// A fake Durable Object world for driving the REAL `Room` from +// server/sync-worker/src/worker.js in node: transactional storage as a Map, +// sockets that record what they were sent and how they were closed, the +// WebSocketPair the upgrade path constructs, and a Response shim (undici's +// refuses status 101). Shared by test-relay-auth.ts and test-relay-broadcast.ts +// so the two rigs cannot drift on what "the relay" is. +// +// Importing this module installs the globals the worker reaches for. Do it +// BEFORE importing the worker — `Room` below is imported here for that reason +// and re-exported. + +// The DO returns a 101 for the WebSocket upgrade; undici's Response refuses any +// status outside 200–599, so the rig supplies a shim before importing. +class FakeResponse { + status: number + body: unknown + webSocket: unknown + headers: Map + constructor(body: unknown, init: { status?: number; headers?: Record; webSocket?: unknown } = {}) { + this.body = body + this.status = init.status ?? 200 + this.webSocket = init.webSocket + this.headers = new Map(Object.entries(init.headers ?? {})) + } + async text() { + return typeof this.body === 'string' ? this.body : '' + } +} +;(globalThis as Record).Response = FakeResponse + +export type Sock = { + sent: string[] + closed: boolean + /** the code and reason the relay closed with, if it did */ + closeCode: number | null + closeReason: string | null + send(t: string): void + close(code?: number, reason?: string): void + serializeAttachment(a: unknown): void + deserializeAttachment(): Record | null +} + +export function mkSocket(att: Record | null = null): Sock { + let attachment = att + return { + sent: [], + closed: false, + closeCode: null, + closeReason: null, + send(t) { this.sent.push(t) }, + close(code, reason) { this.closed = true; this.closeCode = code ?? null; this.closeReason = reason ?? null }, + serializeAttachment(a) { attachment = JSON.parse(JSON.stringify(a)) }, + deserializeAttachment() { return attachment }, + } +} + +let lastPair: { client: Sock; server: Sock } | null = null +;(globalThis as Record).WebSocketPair = function () { + const client = mkSocket() + const server = mkSocket() + lastPair = { client, server } + return { 0: client, 1: server } +} +/** The server half of the most recent upgrade — what the relay talks to. */ +export const lastServer = (): Sock => lastPair!.server + +/** + * The DO's `state`: transactional storage over a Map, the hibernation socket + * list, and a `setAlarm` that RECORDS rather than fires. A rig that wants an + * alarm to go off calls `room.alarm()` itself after putting the relevant + * `al:` key in the past — the worker multiplexes every timer through + * that one handler, and that dispatch is exactly what is worth testing. + */ +export function mkState() { + const store = new Map() + const sockets: Sock[] = [] + const state = { + store, + sockets, + /** what the DO alarm was last armed to, or null once deleted */ + alarmAt: null as number | null, + storage: { + async get(k: string) { return store.get(k) }, + async put(k: string, v: unknown) { store.set(k, v) }, + async delete(k: string | string[]) { + for (const one of Array.isArray(k) ? k : [k]) store.delete(one) + }, + async list({ start, end, prefix }: { start?: string; end?: string; prefix?: string } = {}) { + const out = new Map() + for (const [k, v] of [...store.entries()].sort((a, b) => (a[0] < b[0] ? -1 : 1))) { + if (prefix && !k.startsWith(prefix)) continue + if (start && k < start) continue + if (end && k >= end) continue + out.set(k, v) + } + return out + }, + async deleteAll() { store.clear() }, + async setAlarm(at: number) { state.alarmAt = at }, + async deleteAlarm() { state.alarmAt = null }, + }, + acceptWebSocket(ws: Sock) { sockets.push(ws) }, + getWebSockets() { return sockets.filter((s) => !s.closed) }, + setWebSocketAutoResponse() { /* keepalive, not under test */ }, + } + return state +} + +export const req = (url: string, headers: Record = {}) => ({ + url, + method: 'GET', + headers: { get: (k: string) => headers[k.toLowerCase()] ?? null }, +}) + +export const { Room } = await import('../../server/sync-worker/src/worker.js') + +// --- key material ------------------------------------------------------------- +const EC = { name: 'ECDSA', namedCurve: 'P-256' } as const +const SIGN = { name: 'ECDSA', hash: 'SHA-256' } as const +export const b64u = { + enc(bytes: Uint8Array) { + let s = '' + for (const b of bytes) s += String.fromCharCode(b) + return btoa(s).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '') + }, +} + +export type Keys = { + pub: string + /** the `w` room name this key commits to */ + room: string + sign(text: string): Promise +} + +export async function mintKeys(): Promise { + const kp = (await crypto.subtle.generateKey(EC, true, ['sign', 'verify'])) as CryptoKeyPair + const raw = new Uint8Array(await crypto.subtle.exportKey('raw', kp.publicKey)) + const commit = new Uint8Array(await crypto.subtle.digest('SHA-256', raw as BufferSource)) + return { + pub: b64u.enc(raw), + room: 'w' + b64u.enc(commit), + async sign(text: string) { + return b64u.enc(new Uint8Array(await crypto.subtle.sign(SIGN, kp.privateKey, new TextEncoder().encode(text)))) + }, + } +} + +/** Build the query string a CHAIN member sends: owner-signed invite of `role`, + * invite-signed delegation of the member key. Exactly what a real client + * (kernel/src/sync/online.ts) puts on the wire. */ +export async function chainQuery(owner: Keys, invite: Keys, member: Keys, role: string, exp = 0): Promise { + const ivs = await owner.sign(`inv.${invite.pub}.${role}.${exp}`) + const dg = await invite.sign(`dlg.${member.pub}`) + return `&w=${member.pub}&o=${owner.pub}&ivp=${invite.pub}&ivr=${role}&ive=${exp}&ivs=${ivs}&dg=${dg}` +} + +export const TOK = 'tok0123456789abcd' +/** a frame body the relay will never read — it only ever sees ciphertext */ +export const IV = 'aXZpdmluaXY' +export const CT = 'Y2lwaGVydGV4dA' +export const parse = (s: string) => JSON.parse(s) as Record + +// --- a tiny check counter, shared so the summary line is uniform ------------- +export const tally = { failures: 0, checks: 0 } +export function ok(cond: boolean, msg: string) { + tally.checks++ + if (!cond) { + tally.failures++ + console.error(` ✗ ${msg}`) + } +} +export function finish(name: string): never { + console.log(tally.failures === 0 ? `\nALL PASS (${tally.checks} checks)` : `\n${tally.failures} FAILURES of ${tally.checks} checks`) + void name + process.exit(tally.failures ? 1 : 0) +} + +/** Open a socket on `room` with the given query (after `?tok=`). Returns the + * server socket and the `ready` frame it was sent, if any. */ +export const connect = async (room: InstanceType, roomName: string, query: string, tok = TOK) => { + const res = await room.fetch(req(`https://relay/d/${roomName}?tok=${tok}${query}`, { upgrade: 'websocket' })) + const server = lastServer() + const ready = server.sent.map(parse).find((f) => f.ctl === 'ready') + return { status: res.status as number, server, ready } +} + +/** Answer the relay's possession challenge and return the ticket, or null. */ +export const prove = async ( + room: InstanceType, + c: { server: Sock; ready?: Record }, + signer: { sign(text: string): Promise }, + roomName: string, +) => { + const nonce = c.ready?.c + if (typeof nonce !== 'string') return null + const before = c.server.sent.length + await room.onMessage(c.server, JSON.stringify({ ctl: 'prove', g: await signer.sign(`prove.${nonce}.${roomName}`) })) + const wtFrame = c.server.sent.slice(before).map(parse).find((f) => f.ctl === 'wt') + return (wtFrame?.wt as string | undefined) ?? null +} diff --git a/scripts/test-relay-auth.ts b/scripts/test-relay-auth.ts index e0c8d872..cd364e21 100644 --- a/scripts/test-relay-auth.ts +++ b/scripts/test-relay-auth.ts @@ -29,140 +29,35 @@ // anything else. The client half (online.ts `vouched`) refuses content-bearing // frames that carry no stamp. This rig pins the relay half. // -// It drives the real Durable Object class against fake storage/sockets, so the +// It drives the real Durable Object class against fake storage/sockets (the +// harness in scripts/lib/relay-harness.ts, shared with the broadcast rig), so the // assertions are about worker.js's actual control flow. Everything here fails // against the pre-fix relay except the cases marked as regression guards. -// The DO returns a 101 for the WebSocket upgrade; undici's Response refuses any -// status outside 200–599, so the rig supplies a shim before importing. -class FakeResponse { - status: number - body: unknown - webSocket: unknown - headers: Map - constructor(body: unknown, init: { status?: number; headers?: Record; webSocket?: unknown } = {}) { - this.body = body - this.status = init.status ?? 200 - this.webSocket = init.webSocket - this.headers = new Map(Object.entries(init.headers ?? {})) - } - async text() { - return typeof this.body === 'string' ? this.body : '' - } -} -;(globalThis as Record).Response = FakeResponse - -type Sock = { - sent: string[] - closed: boolean - send(t: string): void - close(): void - serializeAttachment(a: unknown): void - deserializeAttachment(): Record | null -} - -function mkSocket(att: Record | null = null): Sock { - let attachment = att - return { - sent: [], - closed: false, - send(t) { this.sent.push(t) }, - close() { this.closed = true }, - serializeAttachment(a) { attachment = JSON.parse(JSON.stringify(a)) }, - deserializeAttachment() { return attachment }, - } -} - -let lastPair: { client: Sock; server: Sock } | null = null -;(globalThis as Record).WebSocketPair = function () { - const client = mkSocket() - const server = mkSocket() - lastPair = { client, server } - return { 0: client, 1: server } -} -function mkState() { - const store = new Map() - const sockets: Sock[] = [] - return { - store, - sockets, - storage: { - async get(k: string) { return store.get(k) }, - async put(k: string, v: unknown) { store.set(k, v) }, - async delete(k: string | string[]) { - for (const one of Array.isArray(k) ? k : [k]) store.delete(one) - }, - async list({ start, end, prefix }: { start?: string; end?: string; prefix?: string } = {}) { - const out = new Map() - for (const [k, v] of [...store.entries()].sort((a, b) => (a[0] < b[0] ? -1 : 1))) { - if (prefix && !k.startsWith(prefix)) continue - if (start && k < start) continue - if (end && k >= end) continue - out.set(k, v) - } - return out - }, - async deleteAll() { store.clear() }, - async setAlarm() { /* expiry is not what this rig is about */ }, - }, - acceptWebSocket(ws: Sock) { sockets.push(ws) }, - getWebSockets() { return sockets }, - setWebSocketAutoResponse() { /* keepalive, not auth */ }, - } -} - -const req = (url: string, headers: Record = {}) => ({ - url, - method: 'GET', - headers: { get: (k: string) => headers[k.toLowerCase()] ?? null }, -}) - -const { Room } = await import('../server/sync-worker/src/worker.js') - -// --- key material ----------------------------------------------------------- -const EC = { name: 'ECDSA', namedCurve: 'P-256' } as const -const SIGN = { name: 'ECDSA', hash: 'SHA-256' } as const -const b64u = { - enc(bytes: Uint8Array) { - let s = '' - for (const b of bytes) s += String.fromCharCode(b) - return btoa(s).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '') - }, -} - -async function mintKeys() { - const kp = (await crypto.subtle.generateKey(EC, true, ['sign', 'verify'])) as CryptoKeyPair - const raw = new Uint8Array(await crypto.subtle.exportKey('raw', kp.publicKey)) - const commit = new Uint8Array(await crypto.subtle.digest('SHA-256', raw as BufferSource)) - return { - pub: b64u.enc(raw), - room: 'w' + b64u.enc(commit), - async sign(text: string) { - return b64u.enc(new Uint8Array(await crypto.subtle.sign(SIGN, kp.privateKey, new TextEncoder().encode(text)))) - }, - } -} +import { + mkState, mkSocket, Room, mintKeys, chainQuery, connect as hconnect, prove as hprove, + TOK, IV, CT, parse, ok, finish, b64u, type Sock, +} from './lib/relay-harness.ts' const owner = await mintKeys() const stranger = await mintKeys() // a key the room does NOT commit to -const TOK = 'tok0123456789abcd' - -let failures = 0 -let checks = 0 -function ok(cond: boolean, msg: string) { - checks++ - if (!cond) { - failures++ - console.error(` ✗ ${msg}`) - } -} - -/** a frame body the relay will never read — it only ever sees ciphertext */ -const IV = 'aXZpdmluaXY' -const CT = 'Y2lwaGVydGV4dA' -const parse = (s: string) => JSON.parse(s) as Record +void chainQuery +// The harness's connect/prove take the room name; every room here is the owner's. +const connect = (room: InstanceType, query: string) => hconnect(room, owner.room, query) +const prove = ( + room: InstanceType, + c: { server: Sock; ready?: Record }, + signer: { sign(text: string): Promise }, + opts: { wrongRoom?: boolean } = {}, +) => hprove(room, c, signer, opts.wrongRoom ? 'wSOMEOTHERROOM' : owner.room) +const authz = async (room: InstanceType, tok: string, put?: { size: number; bkey: string }) => + (await room.fetch({ + url: `https://do/authz?tok=${tok}` + (put ? `&size=${put.size}&bkey=${put.bkey}` : ''), + method: 'GET', + headers: { get: () => null }, + })).status // --------------------------------------------------------------------------- // Fan-out authentication (hole 1) // --------------------------------------------------------------------------- @@ -233,38 +128,6 @@ const parse = (s: string) => JSON.parse(s) as Record // --------------------------------------------------------------------------- /** the call the blob route makes into the DO: reads omit size/bkey */ -const authz = async (room: InstanceType, tok: string, put?: { size: number; bkey: string }) => - (await room.fetch(req( - `https://do/authz?tok=${tok}` + (put ? `&size=${put.size}&bkey=${put.bkey}` : ''), - ))).status - -const connect = async (room: InstanceType, query: string) => { - const res = await room.fetch(req(`https://relay/d/${owner.room}?tok=${TOK}${query}`, { upgrade: 'websocket' })) - const server = lastPair!.server - const ready = server.sent.map(parse).find((f) => f.ctl === 'ready') - return { status: res.status, server, ready } -} - - -/** Answer the relay's possession challenge the way a real writer does: sign - * `prove..` with the private key, and read back the `wt` frame. - * Returns null when the relay hands out nothing — which is the assertion in - * half the checks below. */ -const prove = async ( - room: InstanceType, - c: { server: Sock; ready?: Record }, - signer: { sign(text: string): Promise }, - opts: { wrongRoom?: boolean } = {}, -) => { - const nonce = c.ready?.c - if (typeof nonce !== 'string') return null - const text = `prove.${nonce}.${opts.wrongRoom ? 'wSOMEOTHERROOM' : owner.room}` - const before = c.server.sent.length - await room.onMessage(c.server, JSON.stringify({ ctl: 'prove', g: await signer.sign(text) })) - const wtFrame = c.server.sent.slice(before).map(parse).find((f) => f.ctl === 'wt') - return (wtFrame?.wt as string | undefined) ?? null -} - // --------------------------------------------------------------------------- // The write ticket: hash-match certifies, only PROOF issues (review §2.1) // --------------------------------------------------------------------------- @@ -406,5 +269,5 @@ const prove = async ( ok((await state.storage.list({ start: 'op:', end: 'op;' })).size === 0, 'and the ops it covers are pruned') } -console.log(failures === 0 ? `\nALL PASS (${checks} checks)` : `\n${failures} FAILURES of ${checks} checks`) -process.exit(failures ? 1 : 0) + +finish('test-relay-auth') diff --git a/scripts/test-relay-broadcast.ts b/scripts/test-relay-broadcast.ts new file mode 100644 index 00000000..b566377e --- /dev/null +++ b/scripts/test-relay-broadcast.ts @@ -0,0 +1,373 @@ +#!/usr/bin/env node +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 The Bento authors +// The relay half of BROADCAST — a show as a special case of collaboration. +// +// node scripts/test-relay-broadcast.ts (Node ≥ 23.6 strips types natively) +// +// Drives the real `Room` from server/sync-worker/src/worker.js through the +// harness in scripts/lib/relay-harness.ts. What it proves, and why each one +// would be silent if it regressed: +// +// 1. ADMISSION is the owner-signed `audience` invite, not the room token. An +// audience copy holds the SHOW key, derives a different `?tok=`, and would +// be 403'd by the token compare — so that compare is skipped for the +// role, and ONLY for the role. Admitted only while live (else close 4002), +// only in `w` rooms, and never on a revoked invite. +// 2. AUDIENCE SOCKETS ARE RECEIVE-ONLY. Every frame from one is dropped — +// ops, presence, control. The invite is on the same chain as a writer's, +// so any membership check would let a ticket drive. +// 3. THE `aud` STREAM is writer-only, live-only, NEVER persisted (p:1 is +// ignored on it), and routed to audience sockets only; the room stream +// never reaches an audience socket. That routing is the whole isolation. +// 4. CONTROL VERBS verify by ROLE against the socket's pinned writer key. +// nav/black are signed with the STREAM in the text and retained as latest +// state; laser is unsigned and never retained. +// 5. A LATE JOINER is served the relay-held show — snapshot, aud ops since, +// nav/black — and NOT the op log or the persisted room snapshot, which are +// room-key ciphertext and collaborator metadata. +// 6. THE SHOW STATE IS IN DO STORAGE, not memory. Hibernation evicts the +// object mid-connection; the rig constructs a SECOND Room over the same +// storage and serves a joiner from it. +// 7. THE BYTE CAP asks the presenter to checkpoint once, refuses new joiners +// at twice the cap, and a checkpoint resets both. +// 8. GRACE AND THE SINGLE ALARM. Presenter loss schedules a 60 s grace; a +// writer's `live` cancels it; when it fires the show ends — and the ROOM +// SURVIVES. The DO has one alarm, it used to mean "wipe the room", and a +// grace timer armed the naive way would have evaporated a live room. +// 9. THE COUNT reaches writer sockets only, on a tick, only when changed. +// 10. `ready` advertises `bc:1` so a client can feature-detect the channel. + +import { + mkState, mkSocket, Room, mintKeys, chainQuery, connect, prove, + TOK, IV, CT, parse, ok, finish, type Sock, +} from './lib/relay-harness.ts' + +const owner = await mintKeys() +const NAME = owner.room +const audInvite = await mintKeys() // the owner-signed `audience` invite keypair +const writerInvite = await mintKeys() // an ordinary `writer` invite, for contrast +const viewerA = await mintKeys() +const viewerB = await mintKeys() +const member = await mintKeys() + +/** A fresh room with the owner connected and proven as presenter. */ +async function stage() { + const state = mkState() + const room = new Room(state, {}) + await state.storage.put('name', NAME) + await state.storage.put('tok', TOK) + const p = await connect(room, NAME, `&bt=1&w=${owner.pub}`) + ok(p.ready?.v === 2 && p.ready?.bc === 1, 'a writer’s ready carries the version and bc:1') + await prove(room, p, owner, NAME) + return { state, room, presenter: p.server } +} +const audQuery = (viewer: { pub: string; sign(t: string): Promise }) => chainQuery(owner, audInvite, viewer, 'audience') +const goLive = (room: InstanceType, sock: Sock) => + owner.sign('live').then((g) => room.onMessage(sock, JSON.stringify({ ctl: 'live', g }))) +const endIt = (room: InstanceType, sock: Sock) => + owner.sign('end').then((g) => room.onMessage(sock, JSON.stringify({ ctl: 'end', g }))) +const frames = (s: Sock) => s.sent.map(parse) + +// --------------------------------------------------------------------------- +{ + console.log('admission: the audience invite, not the token; only while live…') + const { state, room, presenter } = await stage() + + const early = await connect(room, NAME, await audQuery(viewerA), 'notTheRoomToken0') + ok(early.server.closeCode === 4002 && early.server.closeReason === 'not-live', 'before `live`, an audience socket is closed 4002 not-live') + + await goLive(room, presenter) + ok((await state.storage.get('show:live')) === 1, 'a writer-signed `live` starts the show') + + const a = await connect(room, NAME, await audQuery(viewerA), 'notTheRoomToken0') + ok(a.status === 101 && !a.server.closed, 'while live, an audience socket is admitted WITHOUT the room token') + ok(a.ready?.bc === 1, 'its ready advertises bc:1') + ok(a.ready?.v === 2, 'and the relay protocol version — the read-only way to tell a deploy apart') + ok(a.ready?.q === undefined, 'and carries no room sequence — the op log is not its business') + ok(!!(a.server.deserializeAttachment() as { audience?: boolean }).audience, 'pinned as audience') + ok((a.server.deserializeAttachment() as { w?: unknown }).w === null, 'with no writer key') + + const r = await connect(room, NAME, '', 'notTheRoomToken0') + ok(r.status === 403, 'a NON-audience socket with the wrong token is still 403 — the skip is for the role only') + + // a writer invite presented as audience: the owner signed `inv..writer.0`, + // so the audience-text verification fails + const forged = await chainQuery(owner, writerInvite, viewerB, 'writer') + const f = await connect(room, NAME, forged.replace('&ivr=writer', '&ivr=audience')) + ok(f.status === 403, 'a writer invite relabelled `audience` is refused (role is in the signed text)') + + const bad = await connect(room, NAME, forged.replace(`&w=${viewerB.pub}`, `&w=${viewerA.pub}`).replace('&ivr=writer', '&ivr=audience')) + ok(bad.status === 403, 'a delegation for another key is refused') + + // r-room: no chain, no audience + const rstate = mkState() + const rroom = new Room(rstate, {}) + await rstate.storage.put('name', 'rLEGACY') + await rstate.storage.put('tok', TOK) + const rr = await connect(rroom, 'rLEGACY', await audQuery(viewerA)) + ok(rr.status === 403, 'a legacy r-room admits no audience') +} + +// --------------------------------------------------------------------------- +{ + console.log('audience sockets are receive-only…') + const { state, room, presenter } = await stage() + await goLive(room, presenter) + const a = await connect(room, NAME, await audQuery(viewerA)) + const b = await connect(room, NAME, await audQuery(viewerB)) + presenter.sent.length = 0 + b.server.sent.length = 0 + const seqBefore = await state.storage.get('seq') + + await room.onMessage(a.server, JSON.stringify({ i: IV, d: CT })) // presence-shaped + await room.onMessage(a.server, JSON.stringify({ p: 1, i: IV, d: CT, g: 'x' })) // op-shaped + await room.onMessage(a.server, JSON.stringify({ s: 'aud', i: IV, d: CT })) // aud-shaped + await room.onMessage(a.server, JSON.stringify({ ctl: 'nav', s: 'aud', i: IV, d: CT, g: await viewerA.sign(`nav.aud.${IV}.${CT}`) })) + await room.onMessage(a.server, JSON.stringify({ ctl: 'end', g: await viewerA.sign('end') })) + + ok(presenter.sent.length === 0, 'nothing an audience socket sends reaches the presenter') + ok(b.server.sent.length === 0, 'nothing reaches another audience socket') + ok((await state.storage.get('seq')) === seqBefore, 'nothing was persisted') + ok((await state.storage.get('show:nav:aud')) === undefined, 'no control state was retained') + ok((await state.storage.get('show:live')) === 1, 'and an audience `end` did not end the show') +} + +// --------------------------------------------------------------------------- +{ + console.log('the aud stream: writer-only, never persisted, audience-only…') + const { state, room, presenter } = await stage() + await goLive(room, presenter) + const reader = await connect(room, NAME, '') // a collaborator reader on the room stream + const a = await connect(room, NAME, await audQuery(viewerA)) + reader.server.sent.length = 0; a.server.sent.length = 0; presenter.sent.length = 0 + const seqBefore = (await state.storage.get('seq')) || 0 + + // presenter → aud stream, in the shape the REAL client sends it: the + // transport's send() marks every ops frame p:1 and signs it, and the aud + // copy inherits both. So this frame would pass the persist path's signature + // gate if it ever reached it — which is exactly the regression to catch: + // show-key ciphertext landing in the room's op log, where the audience's + // edits would be replayed to every future collaborator as undecryptable + // noise and the room's byte cap would be spent on a stream that is not + // supposed to exist after `end`. + const audG = await owner.sign(`${IV}.${CT}`) + await room.onMessage(presenter, JSON.stringify({ s: 'aud', p: 1, k: 'k1', i: IV, d: CT, g: audG })) + ok(((await state.storage.get('seq')) || 0) === seqBefore, 'an aud frame is NOT persisted even when it is p:1 AND validly signed') + ok((await state.storage.list({ prefix: 'op:' })).size === 0, 'no op: key was written for it') + ok(!frames(presenter).some((f) => f.ctl === 'ack'), 'and the presenter was not acked for it as if it had been') + ok(frames(a.server).some((f) => f.s === 'aud' && f.i === IV), 'it reaches the audience') + ok(reader.server.sent.length === 0, 'it does NOT reach a room-stream reader') + ok((await state.storage.list({ prefix: 'show:op:' })).size === 1, 'it is held in the show state') + + // presenter → room stream (a signed persisted op): audience must not see it + a.server.sent.length = 0 + const g = await owner.sign(`${IV}.${CT}`) + await room.onMessage(presenter, JSON.stringify({ p: 1, i: IV, d: CT, g })) + ok(frames(reader.server).some((f) => f.q === 1), 'a room-stream op reaches the reader, stamped') + ok(a.server.sent.length === 0, 'and never reaches the audience') + + // a reader on the room stream cannot feed the audience + a.server.sent.length = 0 + await room.onMessage(reader.server, JSON.stringify({ s: 'aud', i: IV, d: CT })) + ok(a.server.sent.length === 0, 'a READER’s aud-tagged frame is dropped') + ok((await state.storage.list({ prefix: 'show:op:' })).size === 1, 'and not held') + + // not live: aud frames are dropped + await endIt(room, presenter) + const a2 = await connect(room, NAME, await audQuery(viewerA)) + ok(a2.server.closeCode === 4002, 'after `end` the audience is not admitted') +} + +// --------------------------------------------------------------------------- +{ + console.log('control verbs verify by role; nav/black retained, laser not…') + const { state, room, presenter } = await stage() + const reader = await connect(room, NAME, '') + // a reader (no pinned key) cannot start a show, however it signs + await room.onMessage(reader.server, JSON.stringify({ ctl: 'live', g: await owner.sign('live') })) + ok((await state.storage.get('show:live')) === undefined, 'a socket with no writer key cannot `live`, even with the owner’s signature') + // a writer with a bad signature cannot either + await room.onMessage(presenter, JSON.stringify({ ctl: 'live', g: await viewerA.sign('live') })) + ok((await state.storage.get('show:live')) === undefined, 'a writer socket with a signature from another key cannot `live`') + await goLive(room, presenter) + + const a = await connect(room, NAME, await audQuery(viewerA)) + a.server.sent.length = 0; reader.server.sent.length = 0 + + const navG = await owner.sign(`nav.aud.${IV}.${CT}`) + await room.onMessage(presenter, JSON.stringify({ ctl: 'nav', s: 'aud', i: IV, d: CT, g: navG })) + ok(frames(a.server).some((f) => f.ctl === 'nav' && f.s === 'aud' && f.g === navG), 'a signed aud nav reaches the audience with its signature') + ok(reader.server.sent.length === 0, 'and not the room stream') + ok((await state.storage.get('show:nav:aud'))?.i === IV, 'it is retained as the latest aud nav') + + // the same ciphertext signed for the ROOM stream, replayed as aud: refused + a.server.sent.length = 0 + const roomG = await owner.sign(`nav.room.${IV}.${CT}`) + await room.onMessage(presenter, JSON.stringify({ ctl: 'nav', s: 'aud', i: IV, d: CT, g: roomG })) + ok(a.server.sent.length === 0, 'a room-stream nav signature cannot be replayed onto the aud stream') + + // room-stream nav goes to collaborators, not the audience + await room.onMessage(presenter, JSON.stringify({ ctl: 'nav', s: 'room', i: IV, d: CT, g: roomG })) + ok(frames(reader.server).some((f) => f.ctl === 'nav' && f.s === 'room'), 'a room-stream nav reaches collaborators') + ok(a.server.sent.length === 0, 'and not the audience') + + // laser: unsigned, fanned, not retained + a.server.sent.length = 0 + await room.onMessage(presenter, JSON.stringify({ ctl: 'laser', s: 'aud', i: IV, d: CT })) + ok(frames(a.server).some((f) => f.ctl === 'laser'), 'an unsigned laser from a writer reaches the audience') + ok((await state.storage.get('show:laser:aud')) === undefined, 'and is not retained') + a.server.sent.length = 0 + await room.onMessage(reader.server, JSON.stringify({ ctl: 'laser', s: 'aud', i: IV, d: CT })) + ok(a.server.sent.length === 0, 'a laser from a reader is dropped') + + // black, unsigned: dropped; signed: retained + await room.onMessage(presenter, JSON.stringify({ ctl: 'black', s: 'aud', i: IV, d: CT })) + ok((await state.storage.get('show:black:aud')) === undefined, 'an unsigned black is dropped') + await room.onMessage(presenter, JSON.stringify({ ctl: 'black', s: 'aud', i: IV, d: CT, g: await owner.sign(`black.aud.${IV}.${CT}`) })) + ok((await state.storage.get('show:black:aud'))?.i === IV, 'a signed black is retained') +} + +// --------------------------------------------------------------------------- +{ + console.log('a late joiner is served the show, not the room…') + const { state, room, presenter } = await stage() + // some persisted room history the audience must never see + for (let n = 0; n < 3; n++) { + const i = IV + n, d = CT + n + await room.onMessage(presenter, JSON.stringify({ p: 1, i, d, g: await owner.sign(`${i}.${d}`) })) + } + const si = IV + 'S', sd = CT + 'S' + await room.onMessage(presenter, JSON.stringify({ snap: 1, q: 3, i: si, d: sd, g: await owner.sign(`${si}.${sd}`) })) + ok((await state.storage.get('snap'))?.q === 3, 'the room holds a persisted snapshot') + + await goLive(room, presenter) + await room.onMessage(presenter, JSON.stringify({ ctl: 'audsnap', i: 'AUDI', d: 'AUDD' })) + await room.onMessage(presenter, JSON.stringify({ s: 'aud', i: 'OP1I', d: 'OP1D' })) + await room.onMessage(presenter, JSON.stringify({ s: 'aud', i: 'OP2I', d: 'OP2D' })) + await room.onMessage(presenter, JSON.stringify({ ctl: 'nav', s: 'aud', i: 'NAVI', d: 'NAVD', g: await owner.sign(`nav.aud.NAVI.NAVD`) })) + + const late = await connect(room, NAME, await audQuery(viewerB)) + const got = frames(late.server) + ok(got[0]?.ctl === 'audsnap' && got[0].i === 'AUDI', 'first: the held aud snapshot') + ok(got[1]?.s === 'aud' && got[1].i === 'OP1I' && got[2]?.i === 'OP2I', 'then the aud ops since, in order') + ok(got.some((f) => f.ctl === 'nav' && f.i === 'NAVI'), 'then the retained nav') + ok(got[got.length - 1]?.ctl === 'ready' && got[got.length - 1].bc === 1, 'then ready, with bc:1') + ok(!got.some((f) => f.snap === 1), 'NEVER the room’s persisted snapshot') + ok(!got.some((f) => typeof f.q === 'number' && f.ctl !== 'ready'), 'NEVER the op log') + + // 6. the show state survives eviction: a fresh Room over the same storage + const again = new Room(state, {}) + const late2 = await connect(again, NAME, await audQuery(viewerA)) + const got2 = frames(late2.server) + ok(got2[0]?.ctl === 'audsnap' && got2.some((f) => f.i === 'OP2I') && got2.some((f) => f.ctl === 'nav'), 'a NEW Room over the same storage serves the same show — it lives in storage, not memory') + + // a checkpoint supersedes the ops + await room.onMessage(presenter, JSON.stringify({ ctl: 'audsnap', i: 'AUD2', d: 'AUD2' })) + ok((await state.storage.list({ prefix: 'show:op:' })).size === 0, 'a checkpoint drops the aud ops it supersedes') + ok(frames(late.server).some((f) => f.ctl === 'audsnap' && f.i === 'AUD2'), 'and reaches the audience already present') +} + +// --------------------------------------------------------------------------- +{ + console.log('the byte cap: checkpoint asked once, joiners refused at twice…') + const { state, room, presenter } = await stage() + await goLive(room, presenter) + presenter.sent.length = 0 + const big = 'x'.repeat(4096) + let asked = 0 + let n = 0 + // push until the checkpoint request appears, then keep going + while (n < 200) { + await room.onMessage(presenter, JSON.stringify({ s: 'aud', i: big, d: big })) + n++ + asked = frames(presenter).filter((f) => f.ctl === 'audckpt').length + if (asked && (await state.storage.get('show:full'))) break + } + ok(asked === 1, `the presenter was asked to checkpoint exactly once (after ${n} frames)`) + ok((await state.storage.get('show:full')) === 1, 'past twice the cap the show is marked full') + const refused = await connect(room, NAME, await audQuery(viewerA)) + ok(refused.server.closeCode === 4003 && refused.server.closeReason === 'show-full', 'a joiner is refused 4003 show-full') + await room.onMessage(presenter, JSON.stringify({ ctl: 'audsnap', i: 'AUDI', d: 'AUDD' })) + ok((await state.storage.get('show:full')) === undefined && (await state.storage.get('show:bytes')) === 0, 'a checkpoint resets the cap') + const admitted = await connect(room, NAME, await audQuery(viewerA)) + ok(admitted.status === 101 && !admitted.server.closed, 'and joiners are admitted again') +} + +// --------------------------------------------------------------------------- +{ + console.log('grace, and the one alarm the DO has…') + const { state, room, presenter } = await stage() + await goLive(room, presenter) + const a = await connect(room, NAME, await audQuery(viewerA)) + ok((await state.storage.get('al:idle')) !== undefined, 'the idle wipe is scheduled') + + // the presenter's socket dies without `end` + await room.webSocketClose(presenter) + const graceAt = await state.storage.get('al:grace') as number | undefined + ok(typeof graceAt === 'number' && graceAt > Date.now(), 'presenter loss schedules a grace period') + ok((await state.storage.get('al:idle')) !== undefined, 'and the idle wipe is STILL scheduled beside it') + const dueTimes = [...(await state.storage.list({ prefix: 'al:' })).values()] as number[] + ok(state.alarmAt === Math.min(...dueTimes) && dueTimes.length === 3, 'the DO alarm is armed to the EARLIEST of idle, grace and the count tick') + + // a writer's `live` cancels it — here a co-presenter on a member chain + const co = await connect(room, NAME, await chainQuery(owner, writerInvite, member, 'writer')) + await room.onMessage(co.server, JSON.stringify({ ctl: 'live', g: await member.sign('live') })) + ok((await state.storage.get('al:grace')) === undefined, 'a writer’s `live` cancels the grace') + ok((await state.storage.get('show:live')) === 1, 'and the show continues') + + // grace runs out: the show ends and THE ROOM SURVIVES + await room.webSocketClose(co.server) + await state.storage.put('al:grace', Date.now() - 1) + await room.alarm() + ok(a.server.closeCode === 4001 && a.server.closeReason === 'grace', 'when grace fires, audience sockets are closed 4001') + ok((await state.storage.get('show:live')) === undefined, 'the show is over') + ok((await state.storage.get('name')) === NAME && (await state.storage.get('tok')) === TOK, 'and the room was NOT wiped — the grace alarm did not run the idle wipe') + ok((await state.storage.get('al:idle')) !== undefined && state.alarmAt !== null, 'the idle wipe is still armed') + + // and the idle wipe, when IT fires, still wipes + await state.storage.put('al:idle', Date.now() - 1) + await room.alarm() + ok((await state.storage.get('name')) === undefined, 'the idle alarm still evaporates the room') +} + +// --------------------------------------------------------------------------- +{ + console.log('the audience count: writers only, on a tick, when changed…') + const { state, room, presenter } = await stage() + const reader = await connect(room, NAME, '') + await goLive(room, presenter) + const a = await connect(room, NAME, await audQuery(viewerA)) + const b = await connect(room, NAME, await audQuery(viewerB)) + ok((await state.storage.get('al:count')) !== undefined, 'joins schedule a count tick') + ok(!frames(presenter).some((f) => f.ctl === 'audcount'), 'but nothing is sent per join') + presenter.sent.length = 0; reader.server.sent.length = 0; a.server.sent.length = 0 + + await room.pushCount() + ok(frames(presenter).some((f) => f.ctl === 'audcount' && f.n === 2), 'the tick tells the presenter how many are watching') + ok(reader.server.sent.length === 0, 'a room reader is not told') + ok(a.server.sent.length === 0, 'the audience is not told') + presenter.sent.length = 0 + await room.pushCount() + ok(presenter.sent.length === 0, 'an unchanged count sends nothing') + await room.webSocketClose(b.server) + await room.pushCount() + ok(frames(presenter).some((f) => f.ctl === 'audcount' && f.n === 1), 'a leave shows on the next tick') +} + +// --------------------------------------------------------------------------- +{ + console.log('"Issue new tickets": revoking the audience invite…') + const { state, room, presenter } = await stage() + await goLive(room, presenter) + const a = await connect(room, NAME, await audQuery(viewerA)) + await room.onMessage(presenter, JSON.stringify({ + ctl: 'revoke', p: audInvite.pub, o: owner.pub, g: await owner.sign(`rev.${audInvite.pub}`), + })) + ok(a.server.closeCode === 1008 && a.server.closeReason === 'revoked', 'a socket admitted on the old invite is closed') + const again = await connect(room, NAME, await audQuery(viewerB)) + ok(again.status === 403, 'and the old invite admits nobody hereafter') + void state +} + +finish('test-relay-broadcast') diff --git a/server/sync-worker/src/worker.js b/server/sync-worker/src/worker.js index 8261bd50..158d2cd3 100644 --- a/server/sync-worker/src/worker.js +++ b/server/sync-worker/src/worker.js @@ -63,6 +63,21 @@ const MAX_BLOB = 8 * 1024 * 1024 // but a trust-on-first-use token — the same unbounded-bill shape we closed for // frames. Counted in the DO, which is the only component that knows the room. const ROOM_BLOB_CAP = 256 * 1024 * 1024 +// Broadcast. GRACE: a presenter socket lost without `end` keeps the show +// alive for one reconnect window. COUNT_TICK: the audience count reaches +// writer sockets at most this often. SHOW_OPS_CAP: bytes of aud ops the relay +// holds past the last checkpoint before asking the presenter for a new one; +// at twice this, joiners are refused until a checkpoint arrives. +// The relay's protocol version, on every `ready`. A READ-ONLY discriminator: +// every other way to tell a deployed relay from the last one is a write (a +// snapshot ahead of seq, to see `snap-ahead`) or needs an owner key (to be +// challenged). A reader socket on any room can read this. Bump it when the +// wire changes; 1 was the relay before it said so. +// 2 — relay-auth stamps + possession proof (#452), broadcast verbs +const RELAY_V = 2 +const GRACE_MS = 60_000 +const COUNT_TICK_MS = 5_000 +const SHOW_OPS_CAP = 256 * 1024 const BKEY = (k) => `b:${k}` const OP_KEY = (seq) => `op:${String(seq).padStart(10, '0')}` @@ -106,6 +121,7 @@ const rawFrameId = (raw) => // echoed `g` on a signed frame it fanned out. A client refuses content-bearing // frames (op batches, whole-document fork snapshots) that carry neither, so // read-only holds for live peers too and not merely for the stored log. +const ipvOk = (s) => /^[A-Za-z0-9_-]{80,200}$/.test(s) const b64uDec = (s) => { const b = atob(s.replace(/-/g, '+').replace(/_/g, '/')) const out = new Uint8Array(b.length) @@ -360,7 +376,7 @@ export class Room { await this.state.storage.put(BKEY(bkey), size) await this.state.storage.put('blobBytes', used + size) // touch the expiry clock: blobs keep a room alive the same way ops do - await this.state.storage.setAlarm(Date.now() + IDLE_TTL_MS) + await this.schedule('idle', Date.now() + IDLE_TTL_MS) return new Response(JSON.stringify({ reserved: size }), { status: 200 }) } if (req.headers.get('Upgrade') !== 'websocket') { @@ -369,9 +385,20 @@ export class Room { const url = new URL(req.url) const tok = url.searchParams.get('tok') || '' if (!/^[A-Za-z0-9_-]{10,64}$/.test(tok)) return new Response('bad token', { status: 400 }) + // AUDIENCE sockets never compare tokens. `?tok=` is a hash of the ROOM key, + // trust-on-first-use per room; an audience copy holds the SHOW key (its + // `collab.key` is Ke, not the room key) and derives a different value, so + // the compare would 403 it at the door. Its admission proof is the + // owner-signed `audience` invite on the chain below — strictly stronger + // than a token (signed, role-typed, expirable, revocable), and it keeps the + // relay blind: nothing about Ke, not even a hash of it, is ever presented. + const ivrEarly = url.searchParams.get('ivr') || '' + const isAudience = ivrEarly === 'audience' const saved = await this.state.storage.get('tok') - if (saved === undefined) await this.state.storage.put('tok', tok) - else if (saved !== tok) return new Response('forbidden', { status: 403 }) + if (!isAudience) { + if (saved === undefined) await this.state.storage.put('tok', tok) + else if (saved !== tok) return new Response('forbidden', { status: 403 }) + } // Signed rooms: the room name commits to a pubkey (v1.0.2: the shared // writer key; v1.0.3: the OWNER key). A writer socket presents ?w= (the key @@ -383,7 +410,38 @@ export class Room { if ((await this.state.storage.get('name')) === undefined) await this.state.storage.put('name', name) const signed = name[0] === 'w' let sockW = null - if (signed) { + // An audience socket: `w` rooms only, admitted on the chain with role + // `audience`, RECEIVE-ONLY (every frame it sends is dropped), pinned with + // no writer key, and admitted only while a show is live. + let audienceIvp = null + if (isAudience) { + if (!signed) return new Response('forbidden', { status: 403 }) + const w = url.searchParams.get('w') || '' + const o = url.searchParams.get('o') || '' + const ivp = url.searchParams.get('ivp') || '' + const ive = parseInt(url.searchParams.get('ive') || '0', 10) || 0 + const ivs = url.searchParams.get('ivs') || '' + const dg = url.searchParams.get('dg') || '' + if (!/^[A-Za-z0-9_-]{80,200}$/.test(w)) return new Response('bad key', { status: 400 }) + const rev = (await this.state.storage.get('rev')) || [] + const ok = !!(o && ipvOk(ivp) && ivs && dg) + && (!ive || Date.now() < ive) + && !rev.includes(ivp) + && !rev.includes(w) + && 'w' + (await sha256b64u(b64uDec(o))) === name + && (await this.verifyWith(o, ivs, `inv.${ivp}.audience.${ive}`)) + && (await this.verifyWith(ivp, dg, `dlg.${w}`)) + if (!ok) return new Response('forbidden', { status: 403 }) + if (!(await this.state.storage.get('show:live'))) { + // Not live: no socket is held waiting on the relay. The client shows + // "waiting for the presenter" and reconnects with backoff. + return this.refuseUpgrade(4002, 'not-live') + } + if (await this.state.storage.get('show:full')) { + return this.refuseUpgrade(4003, 'show-full') + } + audienceIvp = ivp + } else if (signed) { const w = url.searchParams.get('w') || '' if (w) { if (!/^[A-Za-z0-9_-]{80,200}$/.test(w)) return new Response('bad writer key', { status: 400 }) @@ -450,21 +508,76 @@ export class Room { nonce = b64uEnc(b) wantsTicket = url.searchParams.get('bt') === '1' } + // `sid` names the socket in storage — the presenter of a show is "the + // socket that sent `live`", and an attachment has no identity of its own. + const sidBytes = new Uint8Array(8) + crypto.getRandomValues(sidBytes) + const sid = b64uEnc(sidBytes) server.serializeAttachment({ - count: 0, windowStart: Date.now(), signed, w: sockW, + count: 0, windowStart: Date.now(), signed, w: sockW, sid, nonce, bt: wantsTicket, proven: false, + audience: !!audienceIvp, ivp: audienceIvp, }) - await this.replay(server, since, nonce) - await this.state.storage.setAlarm(Date.now() + IDLE_TTL_MS) + if (audienceIvp) { + await this.serveAudience(server) + await this.bumpCount() + } else { + await this.replay(server, since, nonce) + } + await this.schedule('idle', Date.now() + IDLE_TTL_MS) + return new Response(null, { status: 101, webSocket: client }) + } + + /** Refuse a websocket upgrade with a close code the client can act on. The + * pair is accepted and closed at once, so the client sees the CODE rather + * than an HTTP error it would have to guess at. */ + refuseUpgrade(code, reason) { + const pair = new WebSocketPair() + const [client, server] = Object.values(pair) + // Hibernation-accepted even though it closes at once: a plain accept() is + // the one call this file must never contain (it is how v0.9.7 broke every + // live room), and a reviewer grepping for it should find nothing. + this.state.acceptWebSocket(server) + try { server.close(code, reason) } catch { /* gone */ } return new Response(null, { status: 101, webSocket: client }) } + /** What a joining audience socket receives: the show as the relay holds it. + * Never the op log or the persisted snapshot — those are room-key + * ciphertext it cannot read, and their count and timing are metadata about + * the collaborators. The show state lives in DO STORAGE, not memory: the + * Hibernation API evicts this object while sockets stay open, and an + * in-memory show would vanish on the first eviction, mid-show, silently. */ + async serveAudience(ws) { + try { + const snap = await this.state.storage.get('show:snap') + if (snap) ws.send(JSON.stringify({ ctl: 'audsnap', i: snap.i, d: snap.d })) + const ops = await this.state.storage.list({ prefix: 'show:op:' }) + for (const [, f] of ops) ws.send(JSON.stringify({ s: 'aud', i: f.i, d: f.d })) + for (const ctl of ['nav', 'black']) { + const last = await this.state.storage.get(`show:${ctl}:aud`) + if (last) ws.send(JSON.stringify({ ctl, s: 'aud', i: last.i, d: last.d, g: last.g })) + } + ws.send(JSON.stringify({ ctl: 'ready', bc: 1, v: RELAY_V })) + } catch { /* socket died mid-serve */ } + } + // --- hibernation handlers (fire on wake; replace addEventListener) --------- async webSocketMessage(ws, data) { await this.onMessage(ws, data).catch(() => {}) } - webSocketClose(ws) { + async webSocketClose(ws) { try { ws.close() } catch { /* already closed */ } + try { + const m = ws.deserializeAttachment() || {} + if (m.audience) await this.bumpCount() + // The presenter's socket went without `end`. One reconnect window of + // grace, then the show ends the same way `end` would. Only a WRITER's + // `live` cancels it — audience activity never extends a show. + if (m.sid && m.sid === (await this.state.storage.get('show:presenter'))) { + await this.schedule('grace', Date.now() + GRACE_MS) + } + } catch { /* nothing to clean up */ } } webSocketError() { /* the runtime drops the socket; nothing to clean up */ } @@ -488,7 +601,9 @@ export class Room { ws.send(JSON.stringify({ q: parseInt(key.slice(3), 10), i: f.i, d: f.d })) } } - ws.send(JSON.stringify(nonce ? { ctl: 'ready', q: seq, c: nonce } : { ctl: 'ready', q: seq })) + // bc:1 — this relay speaks broadcast. A client that does not see it + // reports "relay does not support broadcast" and presents locally. + ws.send(JSON.stringify(nonce ? { ctl: 'ready', q: seq, c: nonce, bc: 1, v: RELAY_V } : { ctl: 'ready', q: seq, bc: 1, v: RELAY_V })) } catch { /* socket died mid-replay */ } @@ -526,6 +641,14 @@ export class Room { } catch { return } + // AUDIENCE SOCKETS ARE RECEIVE-ONLY. Not "may send presence but not ops" — + // nothing. The audience invite is owner-signed and on the same chain as a + // writer's, so any check phrased as "is this key in the chain" would let a + // ticket drive the show; the check is by ROLE, and the role sends nothing. + if (meta.audience) return + + if (await this.onBroadcastFrame(ws, meta, f)) return + // Possession proof (see the connect path for why a hash-match is not one). // A certified socket signs `prove..` with the private half of // the key it presented; the name is in the text so the signature cannot be @@ -569,7 +692,9 @@ export class Room { const noteW = JSON.stringify({ ctl: 'revoked', p: f.p, wt: fresh }) for (const peer of this.state.getWebSockets()) { const m = peer.deserializeAttachment() || {} - if (m.w === f.p) { + if (m.w === f.p || (m.audience && m.ivp === f.p)) { + // "Issue new tickets" revokes the old audience invite: every socket + // admitted on it goes now, and the chain check refuses it hereafter try { peer.send(note) } catch { /* gone */ } try { peer.close(1008, 'revoked') } catch { /* gone */ } continue @@ -596,11 +721,27 @@ export class Room { // the fork snapshot is ephemeral yet replaces a peer's whole document, so // it signs itself, and the echo below must never carry a `g` this relay // did not check — a stamp you don't verify is worse than no stamp. + // The audience stream — checked BEFORE the signature gate below, because + // that gate would drop a `p:1` frame without `g`, and `p:1` is ignored on + // this stream. A frame tagged s:'aud' is an op batch under the SHOW key: + // writer sockets only (a reader must not feed the audience — the socket's + // pinned key is the proof, as for laser), while live only, NEVER persisted + // whatever else it carries, routed to audience sockets only, and appended + // to the relay-held show state so a late joiner is served the same ops. + if (f.s === 'aud') { + if (!meta.w || !meta.signed || !(await this.state.storage.get('show:live'))) return + if (typeof f.i !== 'string' || typeof f.d !== 'string') return + await this.appendShowOp(ws, f) + this.fanTo('aud', JSON.stringify({ s: 'aud', i: f.i, d: f.d }), ws) + return + } + const claimed = typeof f.g === 'string' if (meta.signed && (f.p === 1 || f.snap === 1 || claimed)) { if (!(await this.verifySig(f, ws))) return } + const out = { i: f.i, d: f.d } if (meta.signed && claimed) out.g = f.g const weight = (f.i?.length || 0) + (f.d?.length || 0) @@ -667,17 +808,199 @@ export class Room { return // snapshots are storage-only, never fanned out } - const text = JSON.stringify(out) + // ROOM-stream fan-out: never to audience sockets. Presence in particular + // — 100 joiners on a full-mesh presence path is ~10,000 messages, and the + // audience gets none of it in either direction. + this.fanTo('room', JSON.stringify(out), ws) + await this.schedule('idle', Date.now() + IDLE_TTL_MS) + } + + /** Deliver `text` to every socket on `stream` except `from`. */ + fanTo(stream, text, from = null) { for (const peer of this.state.getWebSockets()) { - if (peer === ws) continue - try { - peer.send(text) - } catch { /* runtime reaps dead sockets */ } + if (peer === from) continue + const m = peer.deserializeAttachment() || {} + if ((stream === 'aud') !== !!m.audience) continue + try { peer.send(text) } catch { /* runtime reaps dead sockets */ } + } + } + + /** Writer sockets only — the presenter and co-presenters. */ + fanToWriters(text) { + for (const peer of this.state.getWebSockets()) { + const m = peer.deserializeAttachment() || {} + if (!m.w) continue + try { peer.send(text) } catch { /* gone */ } } - await this.state.storage.setAlarm(Date.now() + IDLE_TTL_MS) } + // --------------------------------------------------------------------------- + // BROADCAST — a show is a special case of collaboration, not a second + // transport. An audience member is a collaborator holding a TICKET (an + // owner-signed `audience` invite) whose `collab.key` is a per-show SHOW KEY, + // not the room key. While live the presenter sends each op batch and each + // control frame twice — room key for collaborators, show key tagged s:'aud' + // for the audience — and the relay routes by stream, verifies control frames + // by ROLE, holds the show state for late joiners, and forgets all of it at + // `end`. Design: handoffs/broadcast-design.md (private) → docs at release. + // --------------------------------------------------------------------------- + + /** Returns true if `f` was a broadcast control frame (handled or dropped). */ + async onBroadcastFrame(ws, meta, f) { + const ctl = f.ctl + if (ctl !== 'live' && ctl !== 'end' && ctl !== 'nav' && ctl !== 'black' && ctl !== 'laser' && ctl !== 'audsnap') return false + // Every broadcast verb is a WRITER's. Verified by role (the pinned `w` on + // this socket), never by chain membership — the audience invite is on the + // same chain, and membership would let a ticket drive. + if (!meta.w || !meta.signed) return true + + if (ctl === 'live' || ctl === 'end') { + if (typeof f.g !== 'string' || !(await this.verifyWith(meta.w, f.g, ctl))) return true + if (ctl === 'live') { + await this.state.storage.put('show:live', 1) + await this.state.storage.put('show:presenter', meta.sid) + await this.unschedule('grace') + await this.schedule('count', Date.now() + COUNT_TICK_MS) + } else { + await this.endShow('end') + } + return true + } + + if (ctl === 'audsnap') { + // The presenter's checkpoint: the whole document under Ke. Replaces the + // held snapshot, drops the aud ops it supersedes, resets the byte count, + // and reaches every audience socket so they converge on it. + if (typeof f.i !== 'string' || typeof f.d !== 'string') return true + if (!(await this.state.storage.get('show:live'))) return true + await this.state.storage.put('show:snap', { i: f.i, d: f.d }) + const old = await this.state.storage.list({ prefix: 'show:op:' }) + for (const k of old.keys()) await this.state.storage.delete(k) + await this.state.storage.put('show:bytes', 0) + await this.state.storage.put('show:opn', 0) + await this.state.storage.delete('show:full') + await this.state.storage.delete('show:ckpt-asked') + this.fanTo('aud', JSON.stringify({ ctl: 'audsnap', i: f.i, d: f.d }), ws) + return true + } + + // nav / black / laser: encrypted control frames on a named stream. The + // relay never sees a slide id — it verifies, retains, routes. + const stream = f.s === 'aud' ? 'aud' : 'room' + if (typeof f.i !== 'string' || typeof f.d !== 'string') return true + if (stream === 'aud' && !(await this.state.storage.get('show:live'))) return true + if (ctl === 'laser') { + // Fire-and-forget from an already-authenticated writer socket: the + // socket binding is the proof, and 20 signatures a second buy nothing. + this.fanTo(stream, JSON.stringify({ ctl, s: stream, i: f.i, d: f.d }), ws) + return true + } + // nav and black are LATEST-STATE and replayed to later joiners — they + // outlive the socket that sent them, so the proof travels with the frame. + // The stream is inside the signed text so a room-stream ciphertext cannot + // be replayed as the audience's. + if (typeof f.g !== 'string' || !(await this.verifyWith(meta.w, f.g, `${ctl}.${stream}.${f.i}.${f.d}`))) return true + await this.state.storage.put(`show:${ctl}:${stream}`, { i: f.i, d: f.d, g: f.g }) + this.fanTo(stream, JSON.stringify({ ctl, s: stream, i: f.i, d: f.d, g: f.g }), ws) + return true + } + + /** Append an aud op to the held show state, and police its size. */ + async appendShowOp(ws, f) { + const n = ((await this.state.storage.get('show:opn')) || 0) + 1 + const bytes = ((await this.state.storage.get('show:bytes')) || 0) + f.i.length + f.d.length + await this.state.storage.put(`show:op:${String(n).padStart(8, '0')}`, { i: f.i, d: f.d }) + await this.state.storage.put('show:opn', n) + await this.state.storage.put('show:bytes', bytes) + // Past the ceiling, ask the presenter to checkpoint (once per crossing). + // Past twice the ceiling with no checkpoint, stop admitting joiners + // rather than serve them an ever-growing, possibly incoherent state; the + // audience already present keeps streaming. + if (bytes > SHOW_OPS_CAP && !(await this.state.storage.get('show:ckpt-asked'))) { + await this.state.storage.put('show:ckpt-asked', 1) + try { ws.send(JSON.stringify({ ctl: 'audckpt' })) } catch { /* gone */ } + } + if (bytes > 2 * SHOW_OPS_CAP) await this.state.storage.put('show:full', 1) + } + + /** The show is over — by `end`, or by the grace period running out. */ + async endShow(why) { + for (const peer of this.state.getWebSockets()) { + const m = peer.deserializeAttachment() || {} + if (m.audience) { try { peer.close(4001, why) } catch { /* gone */ } } + } + const keys = [...(await this.state.storage.list({ prefix: 'show:' })).keys()] + for (const k of keys) await this.state.storage.delete(k) + await this.unschedule('grace') + await this.unschedule('count') + } + + /** Something changed the audience size: make sure a count tick is pending. + * Coarse by design — one message per tick, not one per join. */ + async bumpCount() { + if (!(await this.state.storage.get('show:live'))) return + if ((await this.state.storage.get('al:count')) === undefined) { + await this.schedule('count', Date.now() + COUNT_TICK_MS) + } + } + + /** The count tick: tell writer sockets how many are watching, only when it + * changed, and only while live. */ + async pushCount() { + if (!(await this.state.storage.get('show:live'))) return + let n = 0 + for (const peer of this.state.getWebSockets()) { + const m = peer.deserializeAttachment() || {} + if (m.audience) n++ + } + const last = await this.state.storage.get('show:count') + if (last !== n) { + await this.state.storage.put('show:count', n) + this.fanToWriters(JSON.stringify({ ctl: 'audcount', n })) + } + // keep ticking while live: joins and leaves re-arm, but a steady audience + // still wants the tick to exist so a missed bump cannot go stale forever + await this.schedule('count', Date.now() + COUNT_TICK_MS) + } + + /** + * THE DURABLE OBJECT HAS ONE ALARM. Before broadcast, that alarm meant one + * thing — the 30-day idle wipe — and every `setAlarm` in this file re-armed + * it. A show needs two more timers (the presenter-loss grace period and the + * coarse audience-count tick), and arming either with a bare `setAlarm` + * would REPLACE the idle alarm; worse, when the grace alarm fired, `alarm()` + * would have run the wipe and evaporated the room mid-show. So every timer + * goes through here: each kind stores its own due time, the DO alarm is + * armed to the earliest, and `alarm()` runs whichever are due and re-arms. + */ + async schedule(kind, at) { + await this.state.storage.put(`al:${kind}`, at) + await this.rearm() + } + async unschedule(kind) { + await this.state.storage.delete(`al:${kind}`) + await this.rearm() + } + async rearm() { + const due = [...(await this.state.storage.list({ prefix: 'al:' })).values()] + if (!due.length) { try { await this.state.storage.deleteAlarm() } catch { /* older runtime */ } ; return } + await this.state.storage.setAlarm(Math.min(...due)) + } async alarm() { + const now = Date.now() + const all = await this.state.storage.list({ prefix: 'al:' }) + for (const [k, at] of all) { + if (at > now) continue + await this.state.storage.delete(k) + const kind = k.slice(3) + if (kind === 'idle') { await this.onIdle(); return } // the room is gone; nothing else to run + if (kind === 'grace') await this.endShow('grace') + if (kind === 'count') await this.pushCount() + } + await this.rearm() + } + + async onIdle() { // ~30 days idle: the room evaporates. Files reopen fine — the document // itself is the durable artifact; a fresh room re-forms on next join. // From 907996a8b68354ce51a52c65d7fb627ea55d335c Mon Sep 17 00:00:00 2001 From: nyblnet <306268465+nyblnet@users.noreply.github.com> Date: Sun, 13 Sep 2026 03:00:12 +0100 Subject: [PATCH 2/3] relay: a writer is a proven socket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The broadcast paths gate on `proven` rather than the pinned key alone — the same rule the write ticket already follows. The checkpoint request goes to the presenter's socket; past the hard cap nothing more is appended and the sender is refused with a code; audience sockets hear nothing of the room's membership. test-relay-broadcast 100 (was 85), the new checks mutation-tested five more ways (4/5/5/2/1 red). test-relay-auth 49, test-relay-protocol 19, test-ci-registered 295. --- scripts/test-relay-broadcast.ts | 68 +++++++++++++++++++++++++++++++- server/sync-worker/src/worker.js | 57 +++++++++++++++++++------- 2 files changed, 109 insertions(+), 16 deletions(-) diff --git a/scripts/test-relay-broadcast.ts b/scripts/test-relay-broadcast.ts index b566377e..f7855a3d 100644 --- a/scripts/test-relay-broadcast.ts +++ b/scripts/test-relay-broadcast.ts @@ -179,6 +179,45 @@ const frames = (s: Sock) => s.sent.map(parse) ok(a2.server.closeCode === 4002, 'after `end` the audience is not admitted') } +// --------------------------------------------------------------------------- +{ + console.log('a certified-but-unproven socket is a reader to the show…') + const { state, room, presenter } = await stage() + await goLive(room, presenter) + const a = await connect(room, NAME, await audQuery(viewerA)) + // The socket every reader copy can open: `?w=` is the owner's PUBLIC key, + // which hash-matches the room name and pins `w` — without the private half + // it can never answer the nonce, so it is never `proven`. + const imp = await connect(room, NAME, `&bt=1&w=${owner.pub}`) + ok(typeof imp.ready?.c === 'string' && !(imp.server.deserializeAttachment() as { proven?: boolean }).proven, 'a reader presenting the owner’s public key is certified and challenged, not proven') + a.server.sent.length = 0; presenter.sent.length = 0 + const held = (await state.storage.list({ prefix: 'show:op:' })).size + + await room.onMessage(imp.server, JSON.stringify({ s: 'aud', p: 1, i: IV, d: CT })) + await room.onMessage(imp.server, JSON.stringify({ ctl: 'laser', s: 'aud', i: IV, d: CT })) + await room.onMessage(imp.server, JSON.stringify({ ctl: 'audsnap', i: IV, d: CT })) + await room.onMessage(imp.server, JSON.stringify({ ctl: 'end', g: await owner.sign('end') })) + ok(a.server.sent.length === 0, 'its aud op, laser and audsnap reach no audience socket') + ok((await state.storage.list({ prefix: 'show:op:' })).size === held, 'nothing it sent is held for late joiners') + ok((await state.storage.get('show:snap')) === undefined, 'it cannot replace the held snapshot') + ok((await state.storage.get('show:live')) === 1, 'and it cannot end the show — even carrying the owner’s real `end` signature, on an unproven socket') + + // nor can it fill the show + const big = 'x'.repeat(300 * 1024) + await room.onMessage(imp.server, JSON.stringify({ s: 'aud', i: big, d: big })) + await room.onMessage(imp.server, JSON.stringify({ s: 'aud', i: big, d: big })) + ok((await state.storage.get('show:full')) === undefined, 'two oversize frames from it do not mark the show full') + const late = await connect(room, NAME, await audQuery(viewerB)) + ok(late.status === 101 && !late.server.closed, 'and joiners are still admitted') + + // a chain-admitted co-presenter who DOES prove is not locked out + const co = await connect(room, NAME, await chainQuery(owner, writerInvite, member, 'writer')) + await prove(room, co, member, NAME) + a.server.sent.length = 0 + await room.onMessage(co.server, JSON.stringify({ ctl: 'laser', s: 'aud', i: IV, d: CT })) + ok(frames(a.server).some((f) => f.ctl === 'laser'), 'a proven co-presenter’s laser reaches the audience') +} + // --------------------------------------------------------------------------- { console.log('control verbs verify by role; nav/black retained, laser not…') @@ -286,12 +325,26 @@ const frames = (s: Sock) => s.sent.map(parse) } ok(asked === 1, `the presenter was asked to checkpoint exactly once (after ${n} frames)`) ok((await state.storage.get('show:full')) === 1, 'past twice the cap the show is marked full') + const heldAtCap = await state.storage.get('show:bytes') as number + presenter.sent.length = 0 + await room.onMessage(presenter, JSON.stringify({ s: 'aud', k: 'k-over', i: big, d: big })) + ok((await state.storage.get('show:bytes')) === heldAtCap, 'past the hard cap nothing more is appended') + ok(frames(presenter).some((f) => f.ctl === 'refused' && f.code === 'show-full' && f.k === 'k-over'), 'and the sender is refused with a code naming the frame') const refused = await connect(room, NAME, await audQuery(viewerA)) ok(refused.server.closeCode === 4003 && refused.server.closeReason === 'show-full', 'a joiner is refused 4003 show-full') await room.onMessage(presenter, JSON.stringify({ ctl: 'audsnap', i: 'AUDI', d: 'AUDD' })) ok((await state.storage.get('show:full')) === undefined && (await state.storage.get('show:bytes')) === 0, 'a checkpoint resets the cap') const admitted = await connect(room, NAME, await audQuery(viewerA)) ok(admitted.status === 101 && !admitted.server.closed, 'and joiners are admitted again') + + // a co-presenter pushing the show over the SOFT line: the request still goes + // to the socket that sent `live`, not to the sender + const co = await connect(room, NAME, await chainQuery(owner, writerInvite, member, 'writer')) + await prove(room, co, member, NAME) + presenter.sent.length = 0; co.server.sent.length = 0 + for (let i = 0; i < 40; i++) await room.onMessage(co.server, JSON.stringify({ s: 'aud', i: big, d: big })) + ok(frames(presenter).some((f) => f.ctl === 'audckpt'), 'a co-presenter’s ops ask the PRESENTER to checkpoint') + ok(!frames(co.server).some((f) => f.ctl === 'audckpt'), 'not the co-presenter') } // --------------------------------------------------------------------------- @@ -312,8 +365,12 @@ const frames = (s: Sock) => s.sent.map(parse) // a writer's `live` cancels it — here a co-presenter on a member chain const co = await connect(room, NAME, await chainQuery(owner, writerInvite, member, 'writer')) + // unproven, the co-presenter's `live` is nothing — a certified socket is a reader to the show + await room.onMessage(co.server, JSON.stringify({ ctl: 'live', g: await member.sign('live') })) + ok((await state.storage.get('al:grace')) !== undefined, 'an UNPROVEN co-presenter’s `live` does not cancel the grace') + await prove(room, co, member, NAME) await room.onMessage(co.server, JSON.stringify({ ctl: 'live', g: await member.sign('live') })) - ok((await state.storage.get('al:grace')) === undefined, 'a writer’s `live` cancels the grace') + ok((await state.storage.get('al:grace')) === undefined, 'a PROVEN writer’s `live` cancels the grace') ok((await state.storage.get('show:live')) === 1, 'and the show continues') // grace runs out: the show ends and THE ROOM SURVIVES @@ -361,6 +418,15 @@ const frames = (s: Sock) => s.sent.map(parse) const { state, room, presenter } = await stage() await goLive(room, presenter) const a = await connect(room, NAME, await audQuery(viewerA)) + + // revoking a COLLABORATOR's member key: the audience hears nothing + a.server.sent.length = 0 + await room.onMessage(presenter, JSON.stringify({ + ctl: 'revoke', p: member.pub, o: owner.pub, g: await owner.sign(`rev.${member.pub}`), + })) + ok(a.server.sent.length === 0 && !a.server.closed, 'a collaborator revocation reaches no audience socket') + + // "Issue new tickets": revoking the AUDIENCE invite await room.onMessage(presenter, JSON.stringify({ ctl: 'revoke', p: audInvite.pub, o: owner.pub, g: await owner.sign(`rev.${audInvite.pub}`), })) diff --git a/server/sync-worker/src/worker.js b/server/sync-worker/src/worker.js index 158d2cd3..4cbb10c8 100644 --- a/server/sync-worker/src/worker.js +++ b/server/sync-worker/src/worker.js @@ -699,6 +699,8 @@ export class Room { try { peer.close(1008, 'revoked') } catch { /* gone */ } continue } + // the audience is told nothing about the room's membership + if (m.audience) continue // PROVEN, not merely certified: a reader that presented the owner's // public key has a pinned `w` too, and must not receive the fresh ticket try { peer.send(m.proven ? noteW : note) } catch { /* gone */ } @@ -729,9 +731,19 @@ export class Room { // whatever else it carries, routed to audience sockets only, and appended // to the relay-held show state so a late joiner is served the same ops. if (f.s === 'aud') { - if (!meta.w || !meta.signed || !(await this.state.storage.get('show:live'))) return + // PROVEN, not merely certified. `w` is pinned at connect on a hash-match + // of the owner's PUBLIC key, which every reader copy carries; `proven` + // is the socket that answered the nonce with the private half. The room + // stream can gate on `w` because its persisted frames sign themselves; + // this stream is unsigned by design, so the socket must be the proof. + if (!meta.proven || !(await this.state.storage.get('show:live'))) return if (typeof f.i !== 'string' || typeof f.d !== 'string') return - await this.appendShowOp(ws, f) + if (!(await this.appendShowOp(f))) { + // past the hard cap with no checkpoint: refuse, and say so to the + // sender — storage stops growing, and the presenter is told twice + // (audckpt earlier, this refusal now) rather than silently dropped + return refuse(ws, 'show-full', { k: f.k }) + } this.fanTo('aud', JSON.stringify({ s: 'aud', i: f.i, d: f.d }), ws) return } @@ -849,10 +861,14 @@ export class Room { async onBroadcastFrame(ws, meta, f) { const ctl = f.ctl if (ctl !== 'live' && ctl !== 'end' && ctl !== 'nav' && ctl !== 'black' && ctl !== 'laser' && ctl !== 'audsnap') return false - // Every broadcast verb is a WRITER's. Verified by role (the pinned `w` on - // this socket), never by chain membership — the audience invite is on the - // same chain, and membership would let a ticket drive. - if (!meta.w || !meta.signed) return true + // Every broadcast verb is a PROVEN writer's. `proven` — the socket answered + // the possession nonce with the private half of the key it presented — and + // never `w` alone, which a reader earns by presenting the owner's public + // key; never chain membership either, since the audience invite is on the + // same chain and membership would let a ticket drive. Chain-admitted + // co-presenters are challenged like everyone else, so nothing legitimate + // is locked out by this. + if (!meta.proven) return true if (ctl === 'live' || ctl === 'end') { if (typeof f.g !== 'string' || !(await this.verifyWith(meta.w, f.g, ctl))) return true @@ -905,22 +921,33 @@ export class Room { return true } - /** Append an aud op to the held show state, and police its size. */ - async appendShowOp(ws, f) { + /** Append an aud op to the held show state, and police its size. Returns + * false — nothing appended — once the hard cap is reached. */ + async appendShowOp(f) { + const held = (await this.state.storage.get('show:bytes')) || 0 + const bytes = held + f.i.length + f.d.length + // Past twice the ceiling with no checkpoint: STOP. Joiners are refused + // (show:full) and so is this frame — otherwise storage keeps growing with + // every op the presenter sends while the checkpoint request is unanswered. + if (bytes > 2 * SHOW_OPS_CAP) { + await this.state.storage.put('show:full', 1) + return false + } const n = ((await this.state.storage.get('show:opn')) || 0) + 1 - const bytes = ((await this.state.storage.get('show:bytes')) || 0) + f.i.length + f.d.length await this.state.storage.put(`show:op:${String(n).padStart(8, '0')}`, { i: f.i, d: f.d }) await this.state.storage.put('show:opn', n) await this.state.storage.put('show:bytes', bytes) - // Past the ceiling, ask the presenter to checkpoint (once per crossing). - // Past twice the ceiling with no checkpoint, stop admitting joiners - // rather than serve them an ever-growing, possibly incoherent state; the - // audience already present keeps streaming. + // Past the ceiling, ask the PRESENTER to checkpoint (once per crossing) — + // the socket that sent `live`, not whichever writer sent this frame. if (bytes > SHOW_OPS_CAP && !(await this.state.storage.get('show:ckpt-asked'))) { await this.state.storage.put('show:ckpt-asked', 1) - try { ws.send(JSON.stringify({ ctl: 'audckpt' })) } catch { /* gone */ } + const presenter = await this.state.storage.get('show:presenter') + for (const peer of this.state.getWebSockets()) { + const m = peer.deserializeAttachment() || {} + if (m.sid === presenter) { try { peer.send(JSON.stringify({ ctl: 'audckpt' })) } catch { /* gone */ } } + } } - if (bytes > 2 * SHOW_OPS_CAP) await this.state.storage.put('show:full', 1) + return true } /** The show is over — by `end`, or by the grace period running out. */ From 2a0c053087211a6c75648b088aafa66198a2612c Mon Sep 17 00:00:00 2001 From: nyblnet <306268465+nyblnet@users.noreply.github.com> Date: Sun, 13 Sep 2026 03:02:57 +0100 Subject: [PATCH 3/3] relay: say what show-full means to the sender --- server/sync-worker/src/worker.js | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/server/sync-worker/src/worker.js b/server/sync-worker/src/worker.js index 4cbb10c8..b7da93d1 100644 --- a/server/sync-worker/src/worker.js +++ b/server/sync-worker/src/worker.js @@ -739,9 +739,13 @@ export class Room { if (!meta.proven || !(await this.state.storage.get('show:live'))) return if (typeof f.i !== 'string' || typeof f.d !== 'string') return if (!(await this.appendShowOp(f))) { - // past the hard cap with no checkpoint: refuse, and say so to the - // sender — storage stops growing, and the presenter is told twice - // (audckpt earlier, this refusal now) rather than silently dropped + // Past the hard cap with no checkpoint: refuse, and say so to the + // sender. The refused frame is neither held NOR fanned to the live + // audience — held state and the live stream stay equal, so a late + // joiner never sees a different show from someone already watching. + // The consequence for the CLIENT: `show-full` means "checkpoint now + // and resend", not "one frame was dropped" — every aud op from here + // is invisible to everyone until a fresh audsnap arrives. return refuse(ws, 'show-full', { k: f.k }) } this.fanTo('aud', JSON.stringify({ s: 'aud', i: f.i, d: f.d }), ws)