264 lines
13 KiB
JavaScript
264 lines
13 KiB
JavaScript
// 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());
|
|
}
|
|
|
|
// 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=<id>` 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 = {} }, 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\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`;
|
|
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)]);
|
|
},
|
|
})));
|
|
});
|
|
}
|