open-standards/wai/wai-web/peer/run.mjs

// Peer delivery in a browser: pages that hold objects and serve them over WebRTC data channels,
// a holder reached over a local HTTP link, and pages that ask, with faults planted on the
// holders' side.
//
//   PLAYWRIGHT_HOST=<dir with node_modules/playwright>/ node run.mjs <wai root> [<session file>]
//
// It serves this directory (the bundle in pkg/, built as verify_peer.mjs says) and the corpus's
// world from one local origin, which is also the pages' origin service and the mailbox that
// introduces them; the HTTP holder is another origin. Every service binds port 0 and says its
// port, and all are closed before exit. The browser is the installed one (WAI_PEER_CHANNEL,
// default `chrome`; empty for the engine Playwright carries). Nothing is fetched from outside
// this machine.
//
// A browser offers its local addresses to a page as names that resolve on the local network
// (§9). On a machine where those names do not resolve between two profiles (a session started
// over SSH, with no multicast), no data channel opens, and the run says so: each failed
// connection names the kinds of candidate it offered. WAI_PEER_BROWSER_ARGS passes the browser
// arguments for such a machine; `--disable-features=WebRtcHideLocalIpsWithMdns` has it offer
// the addresses themselves.
//
// The pages, each a profile of its own:
//   h0  a holder whose viewer leaves serving off
//   hm  holds A0 to A2, and serves A0 with a byte changed
//   hc  holds A whole, and states the group one object larger than it is
//   h1  holds A0 to A3 and B1, and serves what it holds
//   hl  holds A but for A4, and lists A whole
//   a0  an asker whose viewer leaves asking off
//   a1  an asker with asking on: hm, hc, h1 and hl in that order, then the HTTP holder (B whole)
//
// Exit 0 iff every check holds.

import { existsSync, readFileSync, statSync, writeFileSync } from "node:fs";
import { createRequire } from "node:module";
import { dirname, extname, join, normalize, resolve, sep } from "node:path";
import { fileURLToPath } from "node:url";
import { PeerServer, memoryStore } from "./peer.js";
import { checks, httpHolder, listen, loadBundle, loadWorld, sha256 } from "./harness.mjs";

const HERE = dirname(fileURLToPath(import.meta.url));
const [rootArg, sessionOut] = process.argv.slice(2);
if (!rootArg) {
  console.log("usage: PLAYWRIGHT_HOST=<dir>/ node run.mjs <wai root> [<session file>]");
  process.exit(2);
}
let chromium;
try {
  ({ chromium } = createRequire(process.env.PLAYWRIGHT_HOST || HERE + "/")("playwright"));
} catch (e) {
  console.log(`  FAIL run.mjs: playwright is not here (set PLAYWRIGHT_HOST): ${e.message}`);
  process.exit(1);
}
const w = await loadBundle(join(HERE, "pkg"));
const world = loadWorld(resolve(rootArg));
const { expect, done } = checks();
const [A, B] = world.groups;
const POLICY = { ask: true, serve: true, max_object_bytes: 1 << 22, relay_only: false };
const closing = [];
let browser;
// Whatever happens, the browser and every service are closed.
const cleanup = async () => {
  await browser?.close().catch(() => {});
  for (const s of closing) s.close();
};
const guard = setTimeout(async () => {
  console.log("  FAIL the run did not finish in ten minutes");
  await cleanup();
  process.exit(1);
}, 600000);
for (const sig of ["SIGINT", "SIGTERM", "SIGHUP"]) process.on(sig, () => cleanup().then(() => process.exit(143)));

// ── The HTTP holder: B whole, on an origin of its own.
const hh = new PeerServer(w, { seed: crypto.getRandomValues(new Uint8Array(32)), policy: POLICY, setting: { serve: true }, store: memoryStore() });
for (const [i, o] of B.objects.entries()) await hh.store.put(B.link, i, o.envelope);
hh.publish(Date.now() - 5000, Date.now() + 3600000, [{ group: B.link, leaf_count: B.objects.length, held: B.objects.map((_, i) => i) }]);
const httpLog = [];
const http = await listen(httpHolder(hh, (l) => httpLog.push(l)), "the HTTP holder");
closing.push(http);
const httpBase = `http://127.0.0.1:${http.address().port}/`;

// ── The pages' origin: this directory, the configuration, the world and the mailbox.
const TYPES = { ".html": "text/html; charset=utf-8", ".js": "text/javascript", ".mjs": "text/javascript", ".wasm": "application/wasm", ".json": "application/json" };
const originLog = [];
const mail = new Map();
const senders = new Set();
const site = await listen(async (req, res) => {
  const url = new URL(req.url, "http://x");
  const send = (status, type, body) => res.writeHead(status, { "content-type": type, "cache-control": "no-store" }).end(body);
  const path = url.pathname;
  if (path === "/config.json") return send(200, TYPES[".json"], JSON.stringify({ policy: POLICY, origin: "/origin/", signal: "/signal/", http_holders: [httpBase], relays: [] }));
  if (path === "/origin/world.json") {
    // Receipts only: the envelopes are fetched one by one.
    return send(200, TYPES[".json"], JSON.stringify({ groups: world.groups.map((g) => ({ link: g.link, receipt: g.receipt, objects: g.objects.map((o) => ({ receipt: o.receipt })) })) }));
  }
  const object = /^\/origin\/object\/([0-9a-f]{64})\/(\d+)$/.exec(path);
  if (object) {
    const g = world.groups.find((x) => x.link === object[1]);
    const o = g?.objects[Number(object[2])];
    if (!o) return send(404, "text/plain", "");
    originLog.push(`${url.searchParams.get("c")} ${g === A ? "A" : "B"}${object[2]}`);
    return send(200, "application/octet-stream", o.envelope);
  }
  const signal = /^\/signal\/(send|recv)\/([a-z0-9]+)$/.exec(path);
  if (signal && signal[1] === "recv") return send(200, TYPES[".json"], JSON.stringify(mail.get(signal[2])?.splice(0) || []));
  if (signal && req.method === "POST") {
    let text = "";
    for await (const c of req) text += c;
    const m = JSON.parse(text);
    senders.add(m.from);
    if (!mail.has(signal[2])) mail.set(signal[2], []);
    mail.get(signal[2]).push(m);
    return send(200, TYPES[".json"], "{}");
  }
  const file = normalize(join(HERE, decodeURIComponent(path)));
  if (req.method !== "GET" || !file.startsWith(HERE + sep) || !existsSync(file) || !statSync(file).isFile()) return send(404, "text/plain", "");
  return send(200, TYPES[extname(file)] || "application/octet-stream", readFileSync(file));
}, "the pages' origin");
closing.push(site);
const base = `http://127.0.0.1:${site.address().port}`;

const channel = process.env.WAI_PEER_CHANNEL ?? "chrome";
browser = await chromium.launch({ ...(channel ? { channel } : {}), args: (process.env.WAI_PEER_BROWSER_ARGS || "").split(" ").filter(Boolean) });
console.log(`  the browser: ${browser.browserType().name()} ${browser.version()}${channel ? ` (channel ${channel})` : ""}`);

const errors = [];
/** A page in a profile of its own, with a count of the connections it constructs and the URLs it fetches. */
async function open(id, query, { ask = false, serve = false } = {}) {
  const context = await browser.newContext();
  await context.addInitScript(() => {
    const probe = (window.__probe = { connections: 0, fetches: [] });
    const Native = window.RTCPeerConnection;
    window.RTCPeerConnection = class extends Native {
      constructor(...a) {
        super(...a);
        probe.connections++;
      }
    };
    const nativeFetch = window.fetch;
    window.fetch = (...a) => {
      probe.fetches.push(String(a[0] instanceof Request ? a[0].url : a[0]));
      return nativeFetch(...a);
    };
  });
  const page = await context.newPage();
  page.on("pageerror", (e) => errors.push(`${id}: ${e}`));
  await page.goto(`${base}/index.html?id=${id}&${query}`);
  await page.waitForFunction(() => window.__wai_peer && window.__wai_peer.state !== "loading", null, { timeout: 60000 });
  // The viewer's setting: both boxes are off when the page opens, and only a click turns one on.
  const boxes = await page.evaluate(() => [document.getElementById("ask").checked, document.getElementById("serve").checked]);
  if (boxes[0] || boxes[1]) errors.push(`${id}: a setting was on before anyone turned it on`);
  if (ask) await page.check("#ask");
  if (serve) await page.check("#serve");
  await page.click("#start");
  const state = async (...states) => {
    await page.waitForFunction((s) => s.includes(window.__wai_peer.state) || window.__wai_peer.state === "failed", states, { timeout: 120000 });
    return page.evaluate(() => JSON.parse(JSON.stringify({ hook: window.__wai_peer, probe: window.__probe, status: document.getElementById("status").textContent })));
  };
  return { id, page, state };
}

let status = 1;
try {
  // ── A holder whose viewer leaves serving off.
  const h0 = await (await open("h0", "role=holder&holds=0:0-5")).state("off", "serving");
  expect(h0.hook.state === "off" && h0.probe.connections === 0 && !h0.hook.record && !originLog.some((l) => l.startsWith("h0 ")) && !senders.has("h0"), "serving off: the page signs no record, opens no connection and holds nothing for its neighbours");

  // ── The holders.
  const holders = {
    hm: await open("hm", "role=holder&holds=0:0-2&alter=0:0", { serve: true }),
    hc: await open("hc", "role=holder&holds=0:0-5&count=1", { serve: true }),
    h1: await open("h1", "role=holder&holds=0:0-3;1:1", { serve: true }),
    hl: await open("hl", "role=holder&holds=0:0-3,5&lists=0:4", { serve: true }),
  };
  const serving = {};
  for (const [name, h] of Object.entries(holders)) serving[name] = await h.state("serving");
  expect(Object.values(serving).every((s) => s.hook.state === "serving" && s.hook.record), `four holder pages hold their objects in the Cache API and serve${Object.values(serving).find((s) => s.hook.error) ? ": " + Object.values(serving).map((s) => s.hook.error).join(" | ") : ""}`);

  // ── An asker whose viewer leaves asking off.
  const a0 = await (await open("a0", "role=asker&peers=hm,hc,h1,hl")).state("done");
  const quiet = async () => (await Promise.all(Object.values(holders).map((h) => h.page.evaluate(() => window.__wai_peer.stats.wants + window.__wai_peer.links)))).reduce((x, y) => x + y, 0);
  expect(a0.hook.state === "done" && a0.hook.mayAsk === false && a0.probe.connections === 0, `asking off: the page constructs no peer connection (${a0.hook.error || a0.probe.connections})`);
  expect(a0.hook.objects?.length === 9 && a0.hook.objects.every((o) => o.from === "origin" && o.route === "origin reason=peer_not_permitted" && o.verified), "asking off: every object is the origin's, and verified");
  expect(!a0.probe.fetches.some((u) => u.startsWith(httpBase)) && httpLog.length === 0 && !senders.has("a0") && (await quiet()) === 0 && a0.hook.acks.length === 0, "asking off: no holder is sent a request, an introduction or an acknowledgement");

  // ── The asker, asking on.
  const a1 = await (await open("a1", "role=asker&peers=hm,hc,h1,hl", { ask: true })).state("done");
  const hook = a1.hook;
  expect(hook.state === "done" && Object.values(hook.peers || {}).every(Boolean) && a1.probe.connections === 4, `asking on: four data channels and the HTTP link are up${hook.error || hook.failures?.length ? ": " + [hook.error, ...(hook.failures || [])].filter(Boolean).join(" | ") : ""}`);
  const objects = hook.objects || [];
  const at = (name) => objects.find((o) => `${"AB"[o.group]}${o.index}` === name) || { attempts: [] };
  const tried = (name) => at(name).attempts.map((x) => `${x.holder}:${x.outcome}`).join(", ");
  let same = objects.length === 9;
  for (const o of objects) same &&= o.verified && o.sha256 === (await sha256(world.groups[o.group].objects[o.index].envelope));
  expect(same, "every object the page shows verified under its receipt and is the publisher's bytes");
  expect(JSON.stringify(hook.before) === JSON.stringify(["stands", "reject reason=holdings_leaf_count_mismatch", "stands", "stands", "stands"]), `a record that overstates its group is refused whole (${JSON.stringify(hook.before)})`);
  expect(tried("A0") === "hm:reject step=4 reason=content_hash_mismatch, h1:taken", `altered bytes are refused, and the next holder serves (A0: ${tried("A0")})`);
  expect(["A1", "A2", "A3"].every((k) => tried(k) === "h1:taken"), "the peer that served altered bytes is asked nothing more (A1 to A3 from h1 alone)");
  expect(tried("A4") === "hl:miss reason=holdings_overstated" && at("A4").from === "origin", `a peer that lists what it lacks answers miss, and the origin serves (A4: ${tried("A4")})`);
  expect(at("A5").attempts.length === 0 && at("A5").from === "origin", "its record is set aside: the next object it listed is not asked of it (A5)");
  expect(at("B0").from === httpBase && at("B2").from === httpBase && tried("B1") === "h1:taken", `objects come over the HTTP link, and one past a data frame over a data channel (B0 ${at("B0").from}, B1 ${tried("B1")})`);
  expect(JSON.stringify(hook.after) === JSON.stringify(["set-aside reason=peer_bytes_refused", "reject reason=holdings_leaf_count_mismatch", "stands", "set-aside reason=holdings_overstated", "stands"]), `the records' labels after the session (${JSON.stringify(hook.after)})`);
  expect(JSON.stringify(hook.asks) === JSON.stringify({ ...hook.asks, hm: 1, h1: 5, hl: 1 }) && hook.asks.hc === undefined && hook.asks[httpBase] === 2, `each holder was asked what its route gave it, and the overstating record's holder nothing (${JSON.stringify(hook.asks)})`);
  const mine = originLog.filter((l) => l.startsWith("a1 ")).join(",");
  expect(mine === "a1 A4,a1 A5", `the origin served the asker only what no peer did (${mine})`);
  expect(httpLog.length > 0 && httpLog.every((l) => !l.cookie && !l.authorization && !l.referer), "requests to the HTTP holder carried no cookie, credential or referrer");

  // ── The acknowledgements, and what each holder made of them.
  const acks = hook.acks || [];
  const final = {};
  for (const [name, h] of Object.entries(holders)) final[name] = await h.page.evaluate(() => JSON.parse(JSON.stringify(window.__wai_peer)));
  const size = (names) => names.reduce((n, k) => n + at(k).length, 0);
  const to = (name) => acks.find((x) => x.upstream_name === name);
  expect(acks.length === 2 && to("h1") && to(httpBase) && acks.every((x) => x.sent && x.taken), `one acknowledgement each to the two holders that served, each taken (${acks.map((x) => `${x.upstream_name}:${x.taken}`).join(", ")})`);
  expect(!acks.some((x) => [final.hm.key, final.hc.key, final.hl.key].includes(x.upstream)), "no acknowledgement names the altering peer, the overstating one or the miscounting one");
  const taken = (stats) => stats.acks.filter((x) => x.outcome === "taken");
  const t1 = taken(final.h1.stats)[0] || {};
  expect(taken(final.h1.stats).length === 1 && t1.objects === 5 && t1.bytes === size(["A0", "A1", "A2", "A3", "B1"]) && t1.downstream === hook.key, `h1 checked its acknowledgement: five objects, ${t1.bytes} bytes, signed by the asker`);
  const th = taken(hh.stats)[0] || {};
  expect(taken(hh.stats).length === 1 && th.objects === 2 && th.bytes === size(["B0", "B2"]), `the HTTP holder checked its own: two objects, ${th.bytes} bytes`);
  expect(final.hm.stats.wants === 1 && final.hm.stats.acks.length === 0 && final.hc.stats.wants === 0 && final.hl.stats.wants === 1 && final.hl.stats.acks.length === 0, "the altering peer and the overstating peer were each asked once and the miscounting peer never, and none was sent an acknowledgement");
  expect(errors.length === 0, `no page error${errors.length ? "s: " + errors.join(" | ") : ""}`);

  if (sessionOut) {
    writeFileSync(sessionOut, JSON.stringify({ asker: hook.key, acks: acks.map(({ upstream, record, items }) => ({ upstream, record, items })), expected: { [final.h1.key]: { objects: 5, bytes: size(["A0", "A1", "A2", "A3", "B1"]) }, [hh.key]: { objects: 2, bytes: size(["B0", "B2"]) } } }, null, 1) + "\n");
    console.log(`  the session's records: ${sessionOut}`);
  }
  status = done(23);
} catch (e) {
  console.log(`  FAIL the run stopped: ${e?.stack || e}`);
} finally {
  clearTimeout(guard);
  await cleanup();
}
process.exit(status);