WIP: toybox: OCapN vats on workerd, with ingress and egress as capabilities #4
No files matched your search
@@ -0,0 +1,33 @@
|
||||
/**
|
||||
* @file An OCapN client for the admin socket, from Node: it registers its
|
||||
* own gateway with the router there and reaches every other peer through
|
||||
* it. Run under SES (see test/lockdown.js).
|
||||
*/
|
||||
import { makeClient, swissnumFromBytes } from "@endo/ocapn";
|
||||
import { newWebSocketRpcSession } from "capnweb";
|
||||
import WebSocket from "ws";
|
||||
import { locationOf, makeRpcNetLayer } from "./netlayer.js";
|
||||
|
||||
/**
|
||||
* An OCapN client on the admin socket: registers its own gateway with the
|
||||
* router there, and reaches every other peer through it.
|
||||
*
|
||||
* @param {string} adminSock
|
||||
* @param {string} designator
|
||||
* @param {{ verbose?: boolean }} [opts] OCapN logging
|
||||
*/
|
||||
export const connectClient = async (adminSock, designator, { verbose = false } = {}) => {
|
||||
const ws = new WebSocket(`ws+unix:${adminSock}:/`);
|
||||
await new Promise((resolve, reject) => (ws.once("open", resolve), ws.once("error", reject)));
|
||||
const session = newWebSocketRpcSession(ws);
|
||||
const ocapn = makeClient({ debugLabel: designator, verbose });
|
||||
const netlayer = await ocapn.registerNetlayer(
|
||||
makeRpcNetLayer({ designator, openGateway: (d) => session.gateway(d) }),
|
||||
);
|
||||
await session.register(designator, netlayer.gateway);
|
||||
const vatAdmin = await ocapn.enlivenSturdyRef(
|
||||
ocapn.makeSturdyRef(locationOf("vatAdmin"), swissnumFromBytes(new TextEncoder().encode("vatAdmin"))),
|
||||
);
|
||||
return { vatAdmin, close: () => (ocapn.shutdown(), ws.close()) };
|
||||
};
|
||||
|
||||
+4
-3
@@ -4,10 +4,11 @@
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"scripts": {
|
||||
"start": "node build.js && mkdir -p state && rm -f admin.sock && systemd-socket-activate -l 127.0.0.1:8787 -l ./admin.sock -- node_modules/workerd/bin/workerd serve --experimental config.capnp -S http=3 -S admin=4 -d state=state -e switchboard=unix:./switchboard-public.sock",
|
||||
"switchboard": "rm -f switchboard-public.sock switchboard-local.sock && systemd-socket-activate -l ./switchboard-public.sock -l ./switchboard-local.sock -- node_modules/workerd/bin/workerd serve --experimental switchboard.capnp -S public=3 -S local=4",
|
||||
"start": "node build.js && mkdir -p state && rm -f admin.sock && systemd-socket-activate -l 127.0.0.1:8787 -l \"$PWD/admin.sock\" -- node_modules/workerd/bin/workerd serve --experimental config.capnp -S http=3 -S admin=4 -d state=state -e \"switchboard=unix:$PWD/switchboard-public.sock\"",
|
||||
"switchboard": "rm -f switchboard-public.sock switchboard-local.sock && systemd-socket-activate -l \"$PWD/switchboard-public.sock\" -l \"$PWD/switchboard-local.sock\" -- node_modules/workerd/bin/workerd serve --experimental switchboard.capnp -S public=3 -S local=4",
|
||||
"test": "node build.js && node --import ./test/lockdown.js --test test/*.test.js",
|
||||
"build": "node build.js"
|
||||
"build": "node build.js",
|
||||
"hello": "node --import ./test/lockdown.js scripts/hello.js"
|
||||
},
|
||||
"devDependencies": {
|
||||
"esbuild": "^0.28.2",
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
/**
|
||||
* @file Create the hello-world vat on a running toybox and mount it.
|
||||
*
|
||||
* Usage: node --import ./test/lockdown.js scripts/hello.js [name]
|
||||
* then visit http://127.0.0.1:8787/?vat=<name> (default name: hello).
|
||||
*/
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { E } from "@endo/far";
|
||||
import { connectClient } from "../client.js";
|
||||
|
||||
const name = process.argv[2] ?? "hello";
|
||||
const client = await connectClient("admin.sock", `hello-script-${process.pid}`);
|
||||
try {
|
||||
const ingress = await E(client.vatAdmin).makeIngress({ vat: name });
|
||||
const { id } = await E(client.vatAdmin).createVat(await readFile("vats/hello.js", "utf8"), { ingress });
|
||||
console.log(`vat ${id}\nhttp://127.0.0.1:8787/?vat=${name}`);
|
||||
} finally {
|
||||
client.close();
|
||||
}
|
||||
@@ -178,6 +178,8 @@ const encodeUtf8 = harden(TextEncoder.prototype.encode.bind(new TextEncoder()));
|
||||
*/
|
||||
export class VatAdmin extends DurableObject {
|
||||
#clients = new Map();
|
||||
/** mount key → the handler serving it (a live reference, for now) */
|
||||
#handlers = new Map();
|
||||
#ready;
|
||||
|
||||
constructor(ctx, env) {
|
||||
@@ -200,6 +202,21 @@ export class VatAdmin extends DurableObject {
|
||||
const root = await E(vatBootstrap).start(params);
|
||||
return harden({ id: id.toString(), root });
|
||||
},
|
||||
/**
|
||||
* Reserve a mount at `labels` and return the ingress for it: whoever
|
||||
* holds the ingress (a vat it is passed to, say) serves requests
|
||||
* that carry those labels.
|
||||
*
|
||||
* @param {Record<string, string>} labels
|
||||
*/
|
||||
makeIngress: async (labels) => {
|
||||
const key = `ingress:${crypto.randomUUID()}`;
|
||||
await this.env.DIRECTORY.getByName("directory").mount(key, harden({ ...labels }));
|
||||
return Far("ingress", {
|
||||
/** @param {{ handle: (request: object) => object }} handler */
|
||||
serve: (handler) => void this.#handlers.set(key, handler),
|
||||
});
|
||||
},
|
||||
});
|
||||
// workerd runs FinalizationRegistry callbacks outside any request, where
|
||||
// sending the resulting GC message is forbidden I/O. So imports are
|
||||
@@ -240,6 +257,19 @@ export class VatAdmin extends DurableObject {
|
||||
return this.env.VATS.get(id).gateway();
|
||||
}
|
||||
|
||||
/**
|
||||
* One HTTP request, as passable data, for the handler serving `key`;
|
||||
* `null` when nothing serves it.
|
||||
*
|
||||
* @param {string} key
|
||||
* @param {object} request
|
||||
*/
|
||||
async serveHttp(key, request) {
|
||||
await this.#ready;
|
||||
const handler = this.#handlers.get(key);
|
||||
return handler ? E(handler).handle(harden(request)) : null;
|
||||
}
|
||||
|
||||
/** 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));
|
||||
@@ -361,6 +391,7 @@ export class Public extends WorkerEntrypoint {
|
||||
}
|
||||
const mount = await this.env.DIRECTORY.getByName("directory").lookup(labels);
|
||||
if (!mount) return new Response("nothing mounted here\n", { status: 404 });
|
||||
if (mount.vat.startsWith("ingress:")) return this.#serveOcapn(mount, labels, request);
|
||||
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);
|
||||
@@ -369,6 +400,18 @@ export class Public extends WorkerEntrypoint {
|
||||
sealed.headers.set("Referrer-Policy", "no-referrer");
|
||||
return sealed;
|
||||
}
|
||||
|
||||
/** A mount made by `makeIngress`: the request goes, as data, to whoever serves it. */
|
||||
async #serveOcapn(mount, labels, request) {
|
||||
const { pathname } = new URL(request.url);
|
||||
const body = ["GET", "HEAD"].includes(request.method) ? "" : await request.text();
|
||||
const req = { method: request.method, path: pathname, labels: mergeLabels(mount.labels, labels), body };
|
||||
const res = await this.env.VAT_ADMIN.getByName(VAT_ADMIN).serveHttp(mount.vat, req);
|
||||
if (!res) return new Response("mounted, but nothing serves it now\n", { status: 503 });
|
||||
const resp = new Response(res.body ?? "", { status: res.status ?? 200, headers: res.headers ?? {} });
|
||||
resp.headers.set("Referrer-Policy", "no-referrer");
|
||||
return resp;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
+18
-29
@@ -5,42 +5,17 @@ import { join } from "node:path";
|
||||
import { after, before, test } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { E, Far } from "@endo/far";
|
||||
import { makeClient, swissnumFromBytes } from "@endo/ocapn";
|
||||
import { newWebSocketRpcSession } from "capnweb";
|
||||
import WebSocket from "ws";
|
||||
import { locationOf, makeRpcNetLayer } from "../netlayer.js";
|
||||
import { connectClient } from "../client.js";
|
||||
import { freePort, here, startWorkerd } from "./workerd.js";
|
||||
|
||||
const vatSrc = (name) => readFile(join(here, "vats", name), "utf8");
|
||||
|
||||
/**
|
||||
* An OCapN client on the admin socket: registers its own gateway with the
|
||||
* router there, and reaches every other peer through it.
|
||||
*
|
||||
* @param {string} adminSock
|
||||
* @param {string} designator
|
||||
*/
|
||||
const connectClient = async (adminSock, designator) => {
|
||||
const ws = new WebSocket(`ws+unix:${adminSock}:/`);
|
||||
await new Promise((resolve, reject) => (ws.once("open", resolve), ws.once("error", reject)));
|
||||
const session = newWebSocketRpcSession(ws);
|
||||
const ocapn = makeClient({ debugLabel: designator, verbose: !!process.env.OCAPN_LOG });
|
||||
const netlayer = await ocapn.registerNetlayer(
|
||||
makeRpcNetLayer({ designator, openGateway: (d) => session.gateway(d) }),
|
||||
);
|
||||
await session.register(designator, netlayer.gateway);
|
||||
const vatAdmin = await ocapn.enlivenSturdyRef(
|
||||
ocapn.makeSturdyRef(locationOf("vatAdmin"), swissnumFromBytes(new TextEncoder().encode("vatAdmin"))),
|
||||
);
|
||||
return { vatAdmin, close: () => (ocapn.shutdown(), ws.close()) };
|
||||
};
|
||||
|
||||
let dir, server, client;
|
||||
let dir, port, server, client;
|
||||
|
||||
before(async () => {
|
||||
dir = await mkdtemp(join(tmpdir(), "toybox-ocapn-"));
|
||||
await mkdir(join(dir, "state"));
|
||||
const port = await freePort();
|
||||
port = await freePort();
|
||||
server = await startWorkerd(
|
||||
"config.capnp",
|
||||
[
|
||||
@@ -50,7 +25,7 @@ before(async () => {
|
||||
{ port, host: "127.0.0.1" },
|
||||
[join(dir, "admin.sock")],
|
||||
);
|
||||
client = await connectClient(join(dir, "admin.sock"), "client-1");
|
||||
client = await connectClient(join(dir, "admin.sock"), "client-1", { verbose: !!process.env.OCAPN_LOG });
|
||||
});
|
||||
|
||||
after(async () => {
|
||||
@@ -77,3 +52,17 @@ test("a counter vat counts; two counter vats count apart", async () => {
|
||||
assert.equal(await E(a).read(), 2);
|
||||
assert.equal(await E(b).read(), 1);
|
||||
});
|
||||
|
||||
test("a vat serves the web through the ingress it is given", async () => {
|
||||
const ingress = await E(client.vatAdmin).makeIngress({ vat: "hello" });
|
||||
await E(client.vatAdmin).createVat(await vatSrc("hello.js"), { ingress });
|
||||
const resp = await fetch(`http://127.0.0.1:${port}/greet?vat=hello`);
|
||||
assert.equal(resp.status, 200);
|
||||
assert.equal(resp.headers.get("content-type"), "text/html; charset=utf-8");
|
||||
assert.match(await resp.text(), /<h1>Hello, world<\/h1><p>from a toybox vat, at \/greet<\/p>/);
|
||||
});
|
||||
|
||||
test("an ingress nobody serves is a 503", async () => {
|
||||
await E(client.vatAdmin).makeIngress({ vat: "idle" });
|
||||
assert.equal((await fetch(`http://127.0.0.1:${port}/?vat=idle`)).status, 503);
|
||||
});
|
||||
@@ -0,0 +1,12 @@
|
||||
/** Hello, world, served through the `ingress` it was created with. */
|
||||
export const buildRootObject = ({ E, Far }, { ingress }) => {
|
||||
const handler = Far("helloHandler", {
|
||||
handle: ({ path }) => ({
|
||||
status: 200,
|
||||
headers: { "content-type": "text/html; charset=utf-8" },
|
||||
body: `<!doctype html><title>hello</title><h1>Hello, world</h1><p>from a toybox vat, at ${path}</p>\n`,
|
||||
}),
|
||||
});
|
||||
E(ingress).serve(handler);
|
||||
return Far("helloRoot", {});
|
||||
};
|
||||
Reference in new issue
Block a user