captoys/toybox/supervisor.js

/** @file toybox supervisor: vats with code installed at runtime, each app a facet; ingress by mount, egress by grant. */
import "@endo/init";
import { DurableObject, WorkerEntrypoint } from "cloudflare:workers";
import { RpcTarget, newWebSocketRpcSession, newWorkersRpcResponse } from "capnweb";
import { E, Far } from "@endo/far";
import { makeClient, swissnumFromBytes } from "@endo/ocapn";
import { mergeLabels, urlLabels } from "./labels.js";
import { LOOKUP, MOUNT, MOUNT_LABELS, MOUNT_SCHEMA, OVERLAP, UNMOUNT, UNMOUNT_LABELS } from "./mounts-sql.js";
import { locationOf, makeRpcNetLayer } from "./netlayer.js";
import VAT_RUNTIME from "vat-runtime.txt";

/**
 * 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)",
		);
		// absent: an app vat (exports class App), from before OCapN vats
		ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS kind (k INTEGER PRIMARY KEY CHECK (k = 0), kind 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
	 * @param {"app" | "ocapn"} kind `app`: `src` exports `class App`; `ocapn`:
	 *   `src` exports `buildRootObject`, run by vat-runtime.js
	 * @returns {Promise<string>} the code's SHA-256, hex
	 */
	async start(src, egress, kind = "app") {
		const hash = hex(await crypto.subtle.digest("SHA-256", encodeUtf8(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);
		sql.exec("INSERT OR REPLACE INTO kind VALUES (0, ?)", kind);
		this.ctx.facets.abort("app", new Error("upgraded"));
		return hash;
	}

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

	#current() {
		const [row] = this.ctx.storage.sql
			.exec(
				`SELECT b.hash, b.src, coalesce(i.n, 0) AS n, i.egress, coalesce(k.kind, 'app') AS kind
				 FROM current c JOIN bundle b USING (hash)
				 LEFT JOIN incarnation i ON i.k = 0 LEFT JOIN kind k ON k.k = 0`,
			)
			.toArray();
		return row;
	}

	fetch(request) {
		const row = this.#current();
		if (!row) return new Response("no code installed\n", { status: 404 });
		return this.#app(row).fetch(request);
	}

	/** An OCapN vat's gateway, for the router. */
	gateway() {
		const row = this.#current();
		if (row?.kind !== "ocapn") throw new Error("not an OCapN vat");
		return this.#app(row).gateway();
	}
}

/**
 * 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 });
	}
}

/** The vatAdmin's designator, and the swissnum of its bootstrap object. */
const VAT_ADMIN = "vatAdmin";
/** The swissnum of an OCapN vat's bootstrap (see vat-runtime.js). */
const VAT_SWISSNUM = "vat";
/** UTF-8 bytes of a string: the method alone, so no encoder object is in module scope. */
const encodeUtf8 = harden(TextEncoder.prototype.encode.bind(new TextEncoder()));

/**
 * The vatAdmin: an OCapN peer, and the router through which every OCapN
 * peer here reaches every other. One instance, by name.
 *
 * Peers find each other by designator: the vatAdmin's is `vatAdmin`, a
 * vat's is its id, and a client registers its own when it connects to
 * the admin socket. Nothing here can dial out, so a handoff's receiver
 * reaches the exporter through this router.
 */
export class VatAdmin extends DurableObject {
	#clients = new Map();
	#ready;

	constructor(ctx, env) {
		super(ctx, env);
		this.#ready = this.#boot();
	}

	async #boot() {
		const vatAdmin = Far("vatAdmin", {
			/**
			 * Create an OCapN vat running `src`; `params` (any passable,
			 * remote objects included) reach its `buildRootObject`.
			 *
			 * @param {string} src an ES module exporting `buildRootObject(vatPowers, params)`
			 */
			createVat: async (src, params = {}) => {
				const id = this.env.VATS.newUniqueId();
				await this.env.VATS.get(id).start(src, null, "ocapn");
				const vatBootstrap = await this.#enliven(id.toString(), VAT_SWISSNUM);
				const root = await E(vatBootstrap).start(params);
				return harden({ id: id.toString(), root });
			},
		});
		// workerd runs FinalizationRegistry callbacks outside any request, where
		// sending the resulting GC message is forbidden I/O. So imports are
		// never collected here; they go when their session does.
		this.ocapn = makeClient({
			debugLabel: VAT_ADMIN,
			swissnumTable: new Map([[VAT_ADMIN, vatAdmin]]),
			enableImportCollection: false,
		});
		const netlayer = await this.ocapn.registerNetlayer(
			makeRpcNetLayer({ designator: VAT_ADMIN, openGateway: (designator) => this.gateway(designator) }),
		);
		this.own = netlayer.gateway;
	}

	#enliven(designator, swissnum) {
		return this.ocapn.enlivenSturdyRef(
			this.ocapn.makeSturdyRef(locationOf(designator), swissnumFromBytes(encodeUtf8(swissnum))),
		);
	}

	/**
	 * The gateway of the peer with this designator.
	 *
	 * @param {string} designator
	 */
	async gateway(designator) {
		await this.#ready;
		if (designator === VAT_ADMIN) return this.own;
		const client = this.#clients.get(designator);
		if (client) return client;
		let id;
		try {
			id = this.env.VATS.idFromString(designator);
		} catch {
			throw new Error("no such peer");
		}
		return this.env.VATS.get(id).gateway();
	}

	/** A WebSocket here starts a Cap'n Web session whose main object is an {@link AdminSession}. */
	fetch(request) {
		return newWorkersRpcResponse(request, new AdminSession(this, this.#clients));
	}
}

/**
 * What an OCapN client on the admin socket gets: register its own
 * gateway under a designator, and reach any peer's.
 */
class AdminSession extends RpcTarget {
	#vatAdmin;
	#clients;

	constructor(vatAdmin, clients) {
		super();
		this.#vatAdmin = vatAdmin;
		this.#clients = clients;
	}

	/** @param {string} designator @param {any} gateway */
	register(designator, gateway) {
		if (designator === VAT_ADMIN || /^[0-9a-f]{64}$/.test(designator) || this.#clients.has(designator)) {
			throw new Error("designator taken");
		}
		this.#clients.set(designator, gateway.dup?.() ?? gateway);
	}

	/** @param {string} designator */
	gateway(designator) {
		return this.#vatAdmin.gateway(designator);
	}
}

/** A vat's way to the router: `env.router` in an OCapN vat. */
export class Router extends WorkerEntrypoint {
	gateway(designator) {
		return this.env.VAT_ADMIN.getByName(VAT_ADMIN).gateway(designator);
	}
}

/**
 * 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);
		for (const statement of MOUNT_SCHEMA) ctx.storage.sql.exec(statement);
	}

	/**
	 * 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) {
		const { sql } = this.ctx.storage;
		this.ctx.storage.transactionSync(() => {
			sql.exec(UNMOUNT_LABELS, vat);
			if (!labels) return void sql.exec(UNMOUNT, vat);
			const json = JSON.stringify(labels);
			if (sql.exec(OVERLAP, json, vat).toArray().length) {
				throw new Error("mount overlaps another: add a label whose value tells them apart");
			}
			sql.exec(MOUNT, vat, json);
			sql.exec(MOUNT_LABELS, vat, json);
		});
	}

	/**
	 * The mount whose labels `request` carries, if any. Registration keeps
	 * mounts pairwise conflicting, so at most one can match.
	 *
	 * @param {Record<string, string>} request
	 */
	lookup(request) {
		const [row] = this.ctx.storage.sql.exec(LOOKUP, JSON.stringify(request)).toArray();
		return row ? { vat: row.vat, labels: JSON.parse(row.labels) } : null;
	}
}

/**
 * A request's labels: its query parameters.
 *
 * Routing carries no secrets: a URL ends up in logs and `Referer`s. A
 * secret travels in-band instead, as on a webkey page: the page keeps it
 * in its fragment, reaches the vat by public labels, and sends it in the
 * body of a call, which the vat checks like a swissnum.
 *
 * @param {Request} request
 */
const requestLabels = (request) => {
	const labels = {};
	for (const [k, v] of new URL(request.url).searchParams) {
		if (k in labels) throw new Error(`label ${k} given twice`);
		labels[k] = v;
	}
	return Object.freeze(labels);
};

/**
 * 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: a WebSocket upgrade reaches the {@link VatAdmin} (OCapN
 * over Cap'n Web); otherwise, app vats 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) {
		if (request.headers.get("Upgrade")?.toLowerCase() === "websocket") {
			return this.env.VAT_ADMIN.getByName(VAT_ADMIN).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;