captoys/toybox/supervisor.js

/** @file toybox supervisor: vats with code installed at runtime, each app a facet; ingress by mount, egress by grant. */
import { DurableObject, WorkerEntrypoint } from "cloudflare:workers";
import { newWebSocketRpcSession } from "capnweb";
import { carries, conflict, mergeLabels, urlLabels } from "./labels.js";

/**
 * A stored grant: labels as JSON, or (before labels) a bare `host[:port]`.
 *
 * @param {string} stored
 */
const grantLabels = (stored) => {
	if (stored.startsWith("{")) return JSON.parse(stored);
	const { host, port } = urlLabels(`http://${stored}`);
	return stored.includes(":") ? { host, port } : { host };
};

const hex = (buf) => [...new Uint8Array(buf)].map((b) => b.toString(16).padStart(2, "0")).join("");

/**
 * One vat, by DO id: holds its installed code and egress grant; the app
 * facet holds the app's state. Its mount, if any, is in the {@link Directory}.
 *
 * Code is stored by hash so the facet can be rebuilt after hibernation or a
 * restart. The app's global `fetch` throws; a granted host arrives as
 * `env.egress`, an {@link Egress} the app can pass along like any object.
 *
 * Each `start` begins a new incarnation, which names the loaded Worker: the
 * same code with a different grant must not share a Worker.
 */
export class Vat extends DurableObject {
	constructor(ctx, env) {
		super(ctx, env);
		ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS bundle (hash TEXT PRIMARY KEY, src TEXT)");
		ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS current (k INTEGER PRIMARY KEY CHECK (k = 0), hash TEXT)");
		ctx.storage.sql.exec(
			"CREATE TABLE IF NOT EXISTS incarnation (k INTEGER PRIMARY KEY CHECK (k = 0), n INTEGER, egress TEXT)",
		);
	}

	/**
	 * Install `src` and run the app on it; the app's storage carries over.
	 *
	 * @param {string} src an ES module exporting `class App extends DurableObject`
	 * @param {string | null} egress the app's grant: egress labels as JSON (see labels.js), or none
	 * @returns {Promise<string>} the code's SHA-256, hex
	 */
	async start(src, egress) {
		const hash = hex(await crypto.subtle.digest("SHA-256", new TextEncoder().encode(src)));
		const { sql } = this.ctx.storage;
		sql.exec("INSERT OR IGNORE INTO bundle VALUES (?, ?)", hash, src);
		sql.exec("INSERT OR REPLACE INTO current VALUES (0, ?)", hash);
		const [prev] = sql.exec("SELECT n FROM incarnation").toArray();
		sql.exec("INSERT OR REPLACE INTO incarnation VALUES (0, ?, ?)", (prev?.n ?? 0) + 1, egress);
		this.ctx.facets.abort("app", new Error("upgraded"));
		return hash;
	}

	#app({ hash, src, n, egress }) {
		return this.ctx.facets.get("app", () => {
			const worker = this.env.LOADER.get(`${this.ctx.id}:${n}`, () => ({
				compatibilityDate: "2026-10-01",
				mainModule: "app.js",
				modules: { "app.js": src },
				env: egress ? { egress: this.ctx.exports.Egress({ props: { labels: grantLabels(egress) } }) } : {},
				globalOutbound: null,
			}));
			return { class: worker.getDurableObjectClass("App") };
		});
	}

	fetch(request) {
		const [row] = this.ctx.storage.sql
			.exec(
				`SELECT b.hash, b.src, coalesce(i.n, 0) AS n, i.egress
				 FROM current c JOIN bundle b USING (hash) LEFT JOIN incarnation i ON i.k = 0`,
			)
			.toArray();
		if (!row) return new Response("no code installed\n", { status: 404 });
		return this.#app(row).fetch(request);
	}
}

/**
 * A vat's egress grant: a switchboard handle constrained to the labels in
 * `ctx.props`, which no holder can change. The switchboard enforces them;
 * a refusal comes back as a 403.
 */
export class Egress extends WorkerEntrypoint {
	/** A narrower grant: these labels too (conflicting ones throw). */
	constrain(more) {
		return this.ctx.exports.Egress({ props: { labels: mergeLabels(this.ctx.props.labels, more) } });
	}

	async #handle() {
		const resp = await this.env.SWITCHBOARD.fetch("http://switchboard/", { headers: { Upgrade: "websocket" } });
		const ws = resp.webSocket;
		if (!ws) throw new Error(`switchboard refused a session: ${resp.status}`);
		ws.accept();
		return newWebSocketRpcSession(ws).constrain(this.ctx.props.labels);
	}

	async fetch(request) {
		try {
			const handle = await this.#handle();
			if (request.headers.get("Upgrade")?.toLowerCase() !== "websocket") return await handle.fetch(request);
			const protocols = (request.headers.get("Sec-WebSocket-Protocol") ?? "").split(",").map((p) => p.trim()).filter(Boolean);
			const { readable, writable, protocol } = await handle.webSocket(request.url, protocols);
			return this.#bridge(readable, writable, protocol);
		} catch (e) {
			return new Response(`egress refused: ${e.message}\n`, { status: 403 });
		}
	}

	/** A WebSocket for the vat whose messages travel the switchboard's streams. */
	#bridge(readable, writable, protocol) {
		const [client, server] = Object.values(new WebSocketPair());
		server.accept();
		const writer = writable.getWriter();
		server.addEventListener("message", (e) => writer.write(e.data).catch(() => server.close(1011)));
		server.addEventListener("close", () => writer.close().catch(() => {}));
		this.ctx.waitUntil(
			(async () => {
				for await (const m of readable) server.send(m);
				server.close();
			})().catch(() => server.close(1011)),
		);
		const headers = protocol ? { "Sec-WebSocket-Protocol": protocol } : {};
		return new Response(null, { status: 101, webSocket: client, headers });
	}
}

/**
 * Mounts: which vat serves which label set. One instance, by name.
 *
 * Registration follows Flower: a mount must conflict with every other, so
 * no request can carry two mounts' labels. Lookup is stricter than
 * Flower's: the request must carry all of a mount's labels.
 */
export class Directory extends DurableObject {
	constructor(ctx, env) {
		super(ctx, env);
		ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS mount (vat TEXT PRIMARY KEY, labels TEXT)");
	}

	#mounts() {
		return this.ctx.storage.sql
			.exec("SELECT vat, labels FROM mount")
			.toArray()
			.map(({ vat, labels }) => ({ vat, labels: JSON.parse(labels) }));
	}

	/**
	 * Mount `vat` at `labels`, replacing its mount; `null` unmounts it.
	 *
	 * @param {string} vat a Vat's DO id, hex
	 * @param {Record<string, string> | null} labels
	 */
	mount(vat, labels) {
		if (labels) {
			const overlap = this.#mounts().find((m) => m.vat !== vat && !conflict(m.labels, labels));
			if (overlap) throw new Error("mount overlaps another: add a label whose value tells them apart");
			this.ctx.storage.sql.exec("INSERT OR REPLACE INTO mount VALUES (?, ?)", vat, JSON.stringify(labels));
		} else {
			this.ctx.storage.sql.exec("DELETE FROM mount WHERE vat = ?", vat);
		}
	}

	/**
	 * The mount whose labels `request` carries, if any.
	 *
	 * @param {Record<string, string>} request
	 */
	lookup(request) {
		return this.#mounts().find((m) => carries(m.labels, request)) ?? null;
	}
}

/**
 * A request's labels: its query parameters, plus a `Toybox-Labels` header
 * (JSON) for labels too secret for a URL, which ends up in logs and
 * `Referer`s. A webkey page keeps such a secret in its fragment and sends
 * it in the header.
 *
 * @param {Request} request
 */
const requestLabels = (request) => {
	const query = {};
	for (const [k, v] of new URL(request.url).searchParams) {
		if (k in query) throw new Error(`label ${k} given twice`);
		query[k] = v;
	}
	const header = request.headers.get("Toybox-Labels");
	return mergeLabels(query, header ? JSON.parse(header) : {});
};

/**
 * The http socket: the ingress. A request reaches the vat mounted at labels
 * it carries, which sees the mount's and the request's labels together in
 * a `Toybox-Connection-Labels` header (JSON), as a Flower server gets its
 * connection labels.
 */
export class Public extends WorkerEntrypoint {
	async fetch(request) {
		let labels;
		try {
			labels = requestLabels(request);
		} catch (e) {
			return new Response(`bad labels: ${e.message}\n`, { status: 400 });
		}
		const mount = await this.env.DIRECTORY.getByName("directory").lookup(labels);
		if (!mount) return new Response("nothing mounted here\n", { status: 404 });
		const forwarded = new Request(request);
		forwarded.headers.set("Toybox-Connection-Labels", JSON.stringify(mergeLabels(mount.labels, labels)));
		const resp = await this.env.VATS.get(this.env.VATS.idFromString(mount.vat)).fetch(forwarded);
		if (resp.webSocket) return resp;
		const sealed = new Response(resp.body, resp);
		sealed.headers.set("Referrer-Policy", "no-referrer");
		return sealed;
	}
}

/**
 * Grants in a query parameter, as JSON labels; absent means none.
 *
 * @param {URL} url
 * @param {string} name
 */
const grantParam = (url, name) => {
	const v = url.searchParams.get(name);
	return v ? JSON.parse(v) : null;
};

/**
 * The admin socket: `createVat` and its `adminNode`, over HTTP.
 *
 * - `POST /` with an ES module body creates a vat and starts it; the reply
 *   is `{"vat": <id>, "hash": <code hash>}`, and the id addresses it here.
 * - `PUT /<id>` with a body upgrades it; its storage carries over.
 * - `PUT /by-name/<name>` does the same for a vat from before ids, when
 *   vats were named.
 *
 * Every start states the vat's grants, absent meaning none:
 * `?egress=<labels>` (see {@link Egress}) and `?mount=<labels>` (ingress:
 * see {@link Public}).
 */
export class Admin extends WorkerEntrypoint {
	async fetch(request) {
		const url = new URL(request.url);
		const [, first, second] = url.pathname.split("/");
		let id;
		if (request.method === "POST" && !first) id = this.env.VATS.newUniqueId();
		else if (request.method === "PUT" && first === "by-name" && second) id = this.env.VATS.idFromName(second);
		else if (request.method === "PUT" && first && !second) {
			try {
				id = this.env.VATS.idFromString(first);
			} catch {
				return new Response("no such vat\n", { status: 404 });
			}
		} else return new Response("POST / or PUT /<id> or PUT /by-name/<name>, with code\n", { status: 405 });
		const vat = id.toString();
		try {
			await this.env.DIRECTORY.getByName("directory").mount(vat, grantParam(url, "mount"));
		} catch (e) {
			return new Response(`${e.message}\n`, { status: 409 });
		}
		const egress = url.searchParams.get("egress");
		const hash = await this.env.VATS.get(id).start(await request.text(), egress);
		return Response.json({ vat, hash });
	}
}

export default Public;