open-standards/joule-code/joule-code-rs/crates/joule-moq-web/web/room/e2ee.js

// The member of an end-to-end encrypted room (WAI extensions/confidentiality.md), in a page.
//
// The room service is the room's carrier of MLS messages (confidentiality §5.1): its log
// holds the clock's lifecycle records and every proposal, GroupInfo, commit and Welcome, in
// one order. The member reads it from its first entry. It checks every lifecycle record as
// the clock's (its signature, its room, its place in the clock's chain) before the room's
// MLS model takes it, and hands every MLS message to the wasm member
// (joule_moq_web_e2ee's E2eeMember), which checks each as §5 requires. When the room calls
// on it, it creates the group or commits (§5.4, §5.5), and applies what it made only once
// the log returns it in its order. The member's MLS state and every key it derives live in
// the wasm member's memory alone; nothing of them is written to storage or sent anywhere.

const hex = (b) => Array.from(b, (x) => x.toString(16).padStart(2, "0")).join("");
const unhex = (h) => new Uint8Array((h.match(/../g) || []).map((x) => parseInt(x, 16)));
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// An epoch's keys are deleted this long after the next epoch opens (§5.8): objects of an
// epoch still in flight open meanwhile, and nothing of it opens after.
export const FORGET_AFTER_MS = 15000;
// A post whose answer was lost is dropped after this, if the log has not returned it.
const STAGED_MAX_MS = 15000;

export class E2ee {
  // o: { moq, wai, api, log, room, clock, seed, suite, role, grantOf, refreshGrants,
  //      onRecord, onChange }
  constructor(o) {
    this.moq = o.moq;
    this.wai = o.wai;
    this.api = o.api;
    this.log = o.log;
    this.room = o.room;
    this.clock = o.clock;
    this.role = o.role;
    this.grantOf = o.grantOf;
    this.refreshGrants = o.refreshGrants;
    this.onRecord = o.onRecord;
    this.onChange = o.onChange || (() => {});
    this.suite = o.suite;
    this.member = new o.moq.E2eeMember(unhex(o.room), unhex(o.clock), o.seed, o.suite);
    this.at = 0; // the next entry of the log
    this.chain = { seq: 0, prev: "00".repeat(32) }; // the clock's chain as the log carried it
    this.stopped = null;
    this.posting = false;
    this.stagedAt = null;
    this.stats = {
      log: 0, events: 0, mls: {}, refused: {}, posts: { ok: 0, refused: 0 },
      epochs: [], joined: null, removed: null, forgotten: [],
      sealed: { audio: 0, video: 0 }, sealed_by_epoch: {}, unkeyed: { audio: 0, video: 0 },
    };
  }

  // The KeyPackage the clock proposes this principal with (the keying draft §4.7), left
  // with the room service before the page publishes.
  async keyPackage() {
    const kp = this.member.keyPackage();
    const r = await this.api("/mls/key-package", { room: this.room, key_package: hex(kp) });
    if (r.status !== 200) throw new Error(`the room kept no KeyPackage: ${r.status}${r.body && r.body.error ? ` (${r.body.error})` : ""}`);
  }

  stop(why) {
    this.stopped = why;
    this.log(`end-to-end encryption: ${why}`);
  }

  // Take the log's entries from `this.at`, in order; false once the member has stopped.
  async take(entries) {
    for (const e of entries) {
      if (e.seq !== this.at) { this.stop(`the log skipped from ${this.at} to ${e.seq}`); return false; }
      if (e.event) await this.onEvent(unhex(e.event)); else this.onMls(unhex(e.mls));
      if (this.stopped) return false;
      this.at++;
      this.stats.log++;
    }
    return true;
  }

  // Read the room's log once to its present end, without waiting for more and without acting.
  async readToEnd() {
    for (;;) {
      const r = await this.api(`/mls?room=${this.room}&from=${this.at}&wait_ms=0`);
      if (r.status === 410) return this.stop("the room was replaced");
      if (r.status !== 200 || !r.body) throw new Error(`the room's log could not be read: ${r.status}`);
      if (!(await this.take(r.body.entries))) return;
      if (!r.body.entries.length || this.at >= r.body.end) return;
    }
  }

  // Read the room's log to its end, again and again, while `alive()`.
  async follow(alive) {
    while (alive() && !this.stopped) {
      let r;
      try { r = await this.api(`/mls?room=${this.room}&from=${this.at}`); } catch { await sleep(1000); continue; }
      if (r.status === 410) { this.stop("the room was replaced"); break; }
      if (r.status !== 200 || !r.body) { await sleep(1000); continue; }
      if (!(await this.take(r.body.entries))) return;
      if (this.member.staged() && this.stagedAt != null && performance.now() - this.stagedAt > STAGED_MAX_MS) {
        this.member.withdraw();
        this.stagedAt = null;
        this.log("a post the log never returned: dropped");
      }
      await this.act();
      this.onChange();
    }
  }

  // A lifecycle record on the log: the clock's, in its chain's order, then the room's
  // model takes what it says of who holds keys (§5.4).
  async onEvent(bytes) {
    const v = JSON.parse(this.wai.eventRead(bytes));
    if (!v.ok) return this.stop(`a lifecycle record on the log does not verify: ${v.reason}`);
    if (v.principal !== this.clock || v.room !== this.room) return this.stop("a record on the log is not the room clock's");
    if (v.seq !== this.chain.seq || v.prev !== this.chain.prev) return this.stop(`the clock's chain breaks on the log at ${v.seq}`);
    this.chain = { seq: v.seq + 1, prev: v.digest };
    this.stats.events++;
    this.onRecord(bytes);
    const p = v.payload || {};
    let ev = null;
    if (v.kind_name === "join") {
      let g = this.grantOf(p.detail.grant);
      if (!g) { await this.refreshGrants(); g = this.grantOf(p.detail.grant); }
      if (!g) return this.stop(`a join under a grant the room does not list (${String(p.detail.grant).slice(0, 16)})`);
      // A keyed grant may subscribe or may publish (§2).
      ev = { event: "join", subject: p.subject, keyed: !!g.subscribe || g.publish !== 0 };
    } else if (v.kind_name === "leave") {
      ev = { event: "leave", subject: p.subject, reason: p.detail.reason };
    } else if (v.kind_name === "grant_exchange") {
      return this.stop("a grant exchange, which this page does not follow");
    }
    if (ev) this.member.event(JSON.stringify(ev));
  }

  // An MLS message on the log, checked and applied by the wasm member.
  onMls(frame) {
    const v = JSON.parse(this.member.receive(frame));
    if (!v.ok) {
      const why = v.reason || v.error;
      this.stats.refused[why] = (this.stats.refused[why] || 0) + 1;
      this.log(`an MLS message refused: ${why}${v.detail ? ` (${v.detail})` : ""}`);
      return;
    }
    this.stats.mls[v.received] = (this.stats.mls[v.received] || 0) + 1;
    if (v.received === "epoch" || v.received === "joined" || v.received === "created") {
      this.stagedAt = null;
      this.entered(v.epoch, v.received);
    } else if (v.received === "removed") {
      this.stats.removed = v.epoch;
      this.log(`removed from the room's group by the commit that opens epoch ${v.epoch}: no later epoch opens here`);
    }
  }

  entered(epoch, how) {
    if (this.stats.joined == null) this.stats.joined = epoch;
    const last = this.stats.epochs[this.stats.epochs.length - 1];
    this.stats.epochs.push({ epoch, how, at: Math.round(performance.now()) });
    this.log(`epoch ${epoch} (${how})`);
    if (last) {
      setTimeout(() => {
        this.member.forget(BigInt(last.epoch));
        this.stats.forgotten.push(last.epoch);
      }, FORGET_AFTER_MS);
    }
  }

  // Create the group, or commit, when the room calls on this member; the log's order
  // decides whether it is applied.
  async act() {
    if (this.role !== "publisher" || this.stats.removed != null || this.posting) return;
    const m = this.member;
    let frames = null;
    let what = "";
    if (m.owesCreate()) {
      frames = [m.stageCreate()];
      what = "GroupInfo";
    } else if (m.shouldCommit()) {
      const c = m.stageCommit();
      const w = c.welcome();
      frames = [c.commit(), ...(w ? [w] : [])];
      what = `commit opening epoch ${c.epoch}`;
      c.free();
    }
    if (!frames) return;
    this.posting = true;
    this.stagedAt = performance.now();
    try {
      const r = await this.api("/mls/post", { room: this.room, frames: frames.map(hex) });
      if (r.status === 200) {
        this.stats.posts.ok++;
        this.log(`posted the ${what}`);
      } else {
        this.stats.posts.refused++;
        m.withdraw();
        this.stagedAt = null;
        this.log(`the ${what} was not carried: ${r.status}${r.body && r.body.error ? ` (${r.body.error})` : ""}`);
      }
    } catch (e) {
      // The answer was lost: the log says whether it was carried.
      this.log(`posting the ${what}: ${e.message || e}`);
    } finally {
      this.posting = false;
    }
  }

  get joined() {
    return this.member.epoch() !== undefined;
  }

  // Whether an object refused epoch_not_joined may yet open: a publisher that has not
  // joined the group, and was not removed from it, holds it until its Welcome.
  get awaitingWelcome() {
    return this.role === "publisher" && !this.joined && this.stats.removed == null && !this.stopped;
  }

  // Seal an encoded frame's envelope as object 0 of group `group` of this member's track
  // in namespace `ns` (§4.9), or null while it holds no epoch.
  seal(ns, group, priority, envelope, digest, timeoutMs) {
    if (!this.joined) return null;
    const s = this.member.seal(JSON.stringify(ns), "0", BigInt(group), 0n, priority, envelope, digest, true, BigInt(timeoutMs));
    const kid = s.kid;
    const out = { props: s.properties(), payload: s.payload(), kid, epoch: Number(kid >> 24n) };
    s.free();
    this.stats.sealed_by_epoch[out.epoch] = (this.stats.sealed_by_epoch[out.epoch] || 0) + 1;
    return out;
  }

  // Open an object through the receiving checks (§6): { ok, payload, epoch, leaf } or
  // { ok: false, reason } or { ok: false, pending: <epoch> }.
  open(ns, name, group, object, priority, props, payload) {
    const o = this.member.open(JSON.stringify(ns), name, BigInt(group), BigInt(object), priority, props || new Uint8Array(0), payload);
    const v = JSON.parse(o.json);
    if (v.ok) v.payload = o.payload();
    o.free();
    return v;
  }

  state() {
    return JSON.parse(this.member.state());
  }
}