// What the end-to-end tests share: a static server for the pages, the relay on loopback // from a test configuration, the host's composition (wai_live_web), and the browser. // // Every address is loopback, and every wait has a hard bound. import http from "node:http"; import net from "node:net"; import fs from "node:fs"; import path from "node:path"; import os from "node:os"; import { spawn, execFileSync } from "node:child_process"; import crypto from "node:crypto"; export const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); export async function until(what, ms, f, every = 250) { const end = Date.now() + ms; let last; for (;;) { try { last = await f(); if (last) return last; } catch (e) { last = e; } if (Date.now() > end) throw new Error(`${what}: not within ${ms} ms (last: ${last instanceof Error ? last.message : JSON.stringify(last)?.slice(0, 600)})`); await sleep(every); } } const MIME = { ".html": "text/html; charset=utf-8", ".js": "text/javascript; charset=utf-8", ".mjs": "text/javascript; charset=utf-8", ".wasm": "application/wasm", ".json": "application/json", ".css": "text/css" }; // A static server on 127.0.0.1: `root` at /, and each of `mounts` (a URL path to a file). export function serve(root, mounts = {}) { const server = http.createServer((req, res) => { const u = new URL(req.url, "http://127.0.0.1"); let file = mounts[u.pathname]; if (!file) { const p = path.normalize(path.join(root, decodeURIComponent(u.pathname))); if (!p.startsWith(root)) { res.writeHead(403); return res.end(); } file = fs.existsSync(p) && fs.statSync(p).isDirectory() ? path.join(p, "index.html") : p; } fs.readFile(file, (err, b) => { if (err) { res.writeHead(404); return res.end("not found"); } res.writeHead(200, { "content-type": MIME[path.extname(file)] || "application/octet-stream", "cache-control": "no-store" }); res.end(b); }); }); return new Promise((ok) => server.listen(0, "127.0.0.1", () => ok({ server, url: `http://127.0.0.1:${server.address().port}`, close: () => new Promise((r) => server.close(r)) }))); } const strip = (s) => s.replace(/\x1b\[[0-9;]*m/g, ""); // The relay, from a configuration this writes, on loopback ports it reads back from its // log: the UDP port of QUIC and WebTransport, the TCP port of the WebSocket binding, and // its development certificate's SHA-256. export async function startRelay(bin, cfg, dir, name) { const toml = path.join(dir, `${name}.toml`); fs.writeFileSync(toml, relayToml(cfg)); const log = path.join(dir, `${name}.log`); const out = fs.openSync(log, "w"); const proc = spawn(bin, ["--config", toml], { stdio: ["ignore", "pipe", "pipe"], env: { ...process.env, NO_COLOR: "1", JOULE_MOQ_RELAY_LOG: process.env.JOULE_MOQ_RELAY_LOG || "info" } }); const lines = []; const take = (b) => { fs.writeSync(out, b); for (const l of strip(b.toString()).split("\n")) if (l) lines.push(l); }; proc.stdout.on("data", take); proc.stderr.on("data", take); const exited = new Promise((r) => proc.on("exit", (code, signal) => r({ code, signal }))); const r = await until(`relay ${name} listening`, 30000, () => { if (proc.exitCode != null) throw new Error(`relay ${name} exited: ${lines.slice(-5).join(" | ")}`); const q = lines.map((l) => l.match(/on udp 127\.0\.0\.1:(\d+)/)).find(Boolean); const w = lines.map((l) => l.match(/on tcp 127\.0\.0\.1:(\d+)/)).find(Boolean); const h = lines.map((l) => l.match(/certificate sha-256 \(hex\):\s+([0-9a-f]{64})/)).find(Boolean); if (cfg.no_quic) return w && lines.some((l) => l.includes("session summaries signed by")) && { quic: null, ws: Number(w[1]), hash: null }; return q && w && h && { quic: Number(q[1]), ws: Number(w[1]), hash: h[1] }; }, 100); const path_ = cfg.path || "/moq"; return { ...r, name, proc, lines, log, exited, wt: `https://127.0.0.1:${r.quic}${path_}`, wsUrl: `ws://127.0.0.1:${r.ws}${path_}`, // `signal` null: wait for an exit already asked for. async stop(signal = "SIGINT", ms = 20000) { if (signal && proc.exitCode == null && proc.signalCode == null) proc.kill(signal); const e = await Promise.race([exited, sleep(ms).then(() => null)]); if (!e) { proc.kill("SIGKILL"); await exited; } if (!this.closedLog) { this.closedLog = true; fs.closeSync(out); } return e; }, }; } function relayToml(c) { const q = (s) => JSON.stringify(s); const arr = (a) => `[${a.map(q).join(", ")}]`; let t = `${c.no_quic ? "" : 'quic_listen = "127.0.0.1:0"\n'}websocket_listen = "127.0.0.1:0"\npath = ${q(c.path || "/moq")}\ngoaway_uri = ${q(c.goaway_uri || "")}\n\n`; t += `[tls]\ndev_names = ["localhost", "127.0.0.1"]\n\n[limits]\ngoaway_timeout_ms = ${c.goaway_timeout_ms || 8000}\n\n`; t += `[receipts]\nreceipts_track_suffix = "_receipts"\n\n`; for (const d of c.door || []) t += `[[receipts.door]]\nprefix = ${arr(d.prefix)}\ntrust = ${arr(d.trust || [])}\n${d.trust_field != null ? `trust_field = ${d.trust_field}\n` : ""}hold_ms = ${d.hold_ms}\n\n`; if (c.limits) for (const [k, v] of Object.entries(c.limits)) t = t.replace("[limits]\n", `[limits]\n${k} = ${v}\n`); if (c.cache_bytes) t += `[cache]\nmax_bytes = ${c.cache_bytes}\n\n`; // A DVR window of every track (WAI live-recording §8). if (c.dvr) t += `[dvr]\npath = ${q(c.dvr.path)}\nroom_keys = ${arr(c.dvr.room_keys || [])}\n\n[[dvr.window]]\nprefix = ["*"]\nmax_age_s = ${c.dvr.max_age_s}\nmax_bytes = ${c.dvr.max_bytes}\n\n`; t += `[summary]\nout_path = ${q(c.summary_out)}\n\n[auth]\nrequired = ${c.auth ? "true" : "false"}\n\n`; // Lines a room service prints for the relay that trusts it (`joule-moq-room relay-lines`). if (c.extra) t += c.extra + "\n"; // A grant that may subscribe may also seek back below the live edge (live-rooms §10.6), // as a late joiner does, unless the test says otherwise. for (const g of c.grants || []) t += `[[auth.grants]]\nprefix = ${arr(g.prefix)}\ntoken_sha256 = ${q(g.token_sha256)}\npublish = ${g.publish}\nsubscribe = ${g.subscribe}\nproof = ${g.proof === true}\nseek = ${g.seek != null ? g.seek : g.subscribe}\n\n`; return t; } // The test room, from wai_live_web on the host. export function makeFixture(waiBin, names, dir) { const seed = crypto.randomBytes(32).toString("hex"); const t0 = BigInt(Date.now()) * 1000n; const out = execFileSync(waiBin, ["fixture", seed, t0.toString(), names.join(",")], { maxBuffer: 16 << 20 }).toString(); const f = JSON.parse(out); const file = path.join(dir, "fixture.json"); fs.writeFileSync(file, JSON.stringify(f, null, 1)); return { fixture: f, file }; } // The composition the host computes at `t` for the room as a page holds it. export function hostFrame(waiBin, roomJson, compose, t, dir) { const file = path.join(dir, `room-${t}.json`); fs.writeFileSync(file, roomJson); return JSON.parse(execFileSync(waiBin, ["frame", file, compose, String(t)], { maxBuffer: 16 << 20 }).toString()); } // A room verified on the host as co-performance §9 verifies it (`wai_live_web co-verify`), from // the room as a page holds it and `{policies, objects, payloads}`. export function hostCoVerify(waiBin, roomJson, coDoc, dir, name = "co") { const r = path.join(dir, `${name}-room.json`); const c = path.join(dir, `${name}-co.json`); fs.writeFileSync(r, roomJson); fs.writeFileSync(c, JSON.stringify(coDoc)); return JSON.parse(execFileSync(waiBin, ["co-verify", r, c], { maxBuffer: 64 << 20 }).toString()); } // The browser, headless, with a fake camera and microphone, through memguard where the host // has it. Every process this run starts carries `--moq-web-run=` so that strays can be // found and ended afterwards. export const RUN_ID = crypto.randomBytes(6).toString("hex"); export function browserArgs() { return [ "--use-fake-device-for-media-stream", "--use-fake-ui-for-media-stream", "--autoplay-policy=no-user-gesture-required", "--disable-background-timer-throttling", "--disable-renderer-backgrounding", "--disable-backgrounding-occluded-windows", `--moq-web-run=${RUN_ID}`, ]; } // A wrapper that runs the browser under memguard with its own bound (MOQ_WEB_BROWSER_GB, // 8 by default), so that one runaway browser is killed, not the host. export function memguardedBrowser(bin, dir) { const mg = process.env.RB_ROOT ? path.join(process.env.RB_ROOT, "bin", "memguard") : process.env.MEMGUARD || ""; if (!mg || !fs.existsSync(mg)) return { executable: bin, guarded: false }; const w = path.join(dir, "browser-memguard.sh"); fs.writeFileSync(w, `#!/bin/bash\nexec ${JSON.stringify(mg)} --max-gb ${process.env.MOQ_WEB_BROWSER_GB || 8} -- ${JSON.stringify(bin)} "$@"\n`, { mode: 0o755 }); return { executable: w, guarded: true }; } export function killStrays() { // The pattern without its leading dashes: pkill would read them as an option. try { execFileSync("pkill", ["-f", `moq-web-run=${RUN_ID}`]); } catch { /* none left */ } } export function tmpdir(label) { return fs.mkdtempSync(path.join(process.env.TMPDIR || os.tmpdir(), `moq-web-${label}-`)); } // A room service (joule-moq-room) for a relay: its keys made in `dir` (mode 0600), the // relay's lines that trust them, then, once the relay listens, the service itself. export function roomKeys(bin, dir) { const keys = path.join(dir, "room-keys.toml"); const lines = execFileSync(bin, ["keygen", keys], { stdio: ["ignore", "pipe", "pipe"] }).toString(); return { keys, lines, authority: lines.match(/^# authority ([0-9a-f]{64})/m)[1], clock: lines.match(/^# clock\s+([0-9a-f]{64})/m)[1] }; } export async function startRoomService(bin, { keys, relayWs, room = {}, rate = {}, co = null, top = {}, extra = "" }, dir) { const port = await freePort(); const toml = path.join(dir, "room.toml"); let t = `listen = "127.0.0.1:${port}"\nkeys = ${JSON.stringify(keys)}\nrelay_url = ${JSON.stringify(relayWs)}\n`; // Top-level fields (`public_prefix`), each value already in TOML's form. for (const [k, v] of Object.entries(top)) t += `${k} = ${v}\n`; t += "\n[room]\n"; for (const [k, v] of Object.entries(room)) t += `${k} = ${v}\n`; t += "\n[rate]\n"; for (const [k, v] of Object.entries(rate)) t += `${k} = ${v}\n`; // Co-performance (WAI co-performance): off unless given. if (co) { t += "\n[co]\n"; for (const [k, v] of Object.entries(co)) t += `${k} = ${v}\n`; } // Tables of their own after it: `[age]` and its issuers and ratings. t += extra ? `\n${extra}\n` : ""; fs.writeFileSync(toml, t); const log = path.join(dir, "room-service.log"); const out = fs.openSync(log, "w"); const proc = spawn(bin, ["--config", toml], { stdio: ["ignore", out, out], env: { ...process.env, NO_COLOR: "1" } }); const exited = new Promise((r) => proc.on("exit", (code, signal) => r({ code, signal }))); const url = `http://127.0.0.1:${port}`; await until("the room service's clock on the relay", 30000, async () => { if (proc.exitCode != null) throw new Error(`the room service exited: ${fs.readFileSync(log, "utf8").slice(-400)}`); try { const r = await fetch(`${url}/health`); return r.ok && (await r.json()).relay === true; } catch { return false; } }, 200); return { url, proc, log, exited, lines: () => fs.readFileSync(log, "utf8").split("\n"), async stop() { if (proc.exitCode == null) proc.kill("SIGTERM"); const e = await Promise.race([exited, sleep(10000).then(() => null)]); if (!e) { proc.kill("SIGKILL"); await exited; } fs.closeSync(out); return e; }, }; } function freePort() { return new Promise((ok) => { const s = net.createServer(); s.listen(0, "127.0.0.1", () => { const p = s.address().port; s.close(() => ok(p)); }); }); } // The pages, the room service and the relay on one origin, as the production front serves // them: `root` at /, `/room-api/*` to the service with the prefix stripped, and the // WebSocket upgrade of `/moq` to the relay, its request (Host and all) passed as it came. export function serveSameOrigin(root, { service, relayPort }) { const files = serve(root); return files.then(({ server: fileServer, close: closeFiles }) => { const server = http.createServer((req, res) => { if (req.url.startsWith("/room-api/")) { const u = new URL(service); const up = http.request({ host: u.hostname, port: u.port, method: req.method, path: req.url.slice("/room-api".length), headers: { ...req.headers, "cf-connecting-ip": "127.0.0.1" } }, (r) => { res.writeHead(r.statusCode, r.headers); r.pipe(res); }); up.on("error", () => { res.writeHead(502); res.end(); }); req.pipe(up); return; } fileServer.emit("request", req, res); }); // Every socket, the upgraded ones and their relay ends too, is ended when the server // closes: an upgraded socket otherwise holds `close` open for as long as it lives. const sockets = new Set(); const keep = (x) => { sockets.add(x); x.on("close", () => sockets.delete(x)); }; server.on("connection", keep); server.on("upgrade", (req, socket, head) => { if (!req.url.startsWith("/moq")) return socket.destroy(); const up = net.connect(relayPort, "127.0.0.1", () => { let raw = `${req.method} ${req.url} HTTP/1.1\r\n`; for (let i = 0; i < req.rawHeaders.length; i += 2) raw += `${req.rawHeaders[i]}: ${req.rawHeaders[i + 1]}\r\n`; up.write(raw + "\r\n"); if (head && head.length) up.write(head); up.pipe(socket); socket.pipe(up); }); keep(up); up.on("error", () => socket.destroy()); socket.on("error", () => up.destroy()); up.on("close", () => socket.destroy()); socket.on("close", () => up.destroy()); }); return new Promise((ok) => server.listen(0, "127.0.0.1", () => ok({ server, url: `http://127.0.0.1:${server.address().port}`, close: () => { for (const x of sockets) x.destroy(); return Promise.race([Promise.all([new Promise((r) => server.close(r)), closeFiles()]), sleep(5000)]); }, }))); }); }