johnandersen777.bsky.social/atlibs-socialweb-computer-requester
socialweb.computer market requester and policy modules for atlibs, with its web pages and lexicons
0
0
// node --test web/webrtc/webrtc-cubit.test.mjs (after `cargo wasm`)
// The cubit with the real requester module and runModule; the bidder's PDS, age and the
// peer connection are stubbed (Node has no RTCPeerConnection).
import assert from "node:assert/strict";
import { createHash } from "node:crypto";
import { readFile } from "node:fs/promises";
import { registerHooks } from "node:module";
import { mock, test } from "node:test";
import { load } from "../../../atlibs/rust/src/harness.mjs";
registerHooks({
resolve(specifier, context, next) {
// atlibs/webrtc/ is the atlibs-webrtc sibling checkout's js/; the rest of atlibs/ is core's javascript/src
const lib = specifier.startsWith("atlibs/webrtc/") ? new URL(`../../../atlibs-webrtc/js/${specifier.slice("atlibs/webrtc/".length)}`, import.meta.url).href
: specifier.startsWith("atlibs/") && new URL(`../../../atlibs/javascript/src/${specifier.slice("atlibs/".length)}`, import.meta.url).href;
return next(lib || specifier, context);
},
});
const { WebrtcCubit } = await import("./webrtc-cubit.js");
const { checkFingerprint, guestOf, parseReceipt } = await import("atlibs/webrtc/index.js");
const { run } = await load(new URL("../../target/wasm32-unknown-unknown/release/requester.wasm", import.meta.url));
// What runModule and pdsOf use of a PluginContext.
const context = (fetch) => ({ wasm: { run: async ({ entry, input }) => run(entry, input) }, http: { fetch } });
const T = "com.publicdomainrelay.temp";
const HAT9 = "did:plc:4vcig7s43z3nyqrnepmcfzqt";
const PDS = "https://pds.example";
const receipt = { uri: `at://${HAT9}/${T}.market.receipt/receipt-1`, cid: "bafyrc" };
const FP = "sha-256 AB:CD:EF";
const spec = JSON.parse(await readFile(new URL("../../modules/requester/module.json", import.meta.url), "utf8"));
const module = {
module: new Uint8Array(await readFile(new URL("../../target/wasm32-unknown-unknown/release/requester.wasm", import.meta.url))),
calls: spec.calls,
};
// "age" that seals by wrapping: sealed to r opens only with the identity whose recipient is r.
const age = {
generateX25519Identity: async () => "AGE-SECRET-KEY-ME",
identityToRecipient: async (id) => id.replace("AGE-SECRET-KEY-", "age1"),
Encrypter: class { addRecipient(r) { this.r = r } async encrypt(text) { return JSON.stringify({ to: this.r, text }) } },
Decrypter: class {
addIdentity(id) { this.r = id.replace("AGE-SECRET-KEY-", "age1") }
async decrypt(bytes, _) { const b = JSON.parse(bytes); if (b.to !== this.r) throw new Error("no identity matched"); return b.text }
},
armor: { encode: (s) => `-----BEGIN AGE ENCRYPTED FILE-----\n${s}`, decode: (s) => s.split("\n").slice(1).join("\n") },
};
const sealed = (to, body) => age.armor.encode(JSON.stringify({ to, text: JSON.stringify(body) }));
function peer() {
const pc = new EventTarget();
Object.assign(pc, {
iceGatheringState: "new",
createDataChannel: (label, init) => (pc.dc = { label, init, readyState: "open", sent: [], send: (b) => pc.dc.sent.push(new TextDecoder().decode(b)) }),
createOffer: async () => ({ type: "offer", sdp: "v=0 offer" }),
// gather: "complete" (default), or a list of candidate events the test fires itself
setLocalDescription: async (d) => {
pc.localDescription = d;
if (pc.gather !== "manual") pc.iceGatheringState = "complete", pc.dispatchEvent(new Event("icegatheringstatechange"));
},
setRemoteDescription: async (d) => { pc.remoteDescription = d },
close() {},
});
return pc;
}
// The bidder's repo: each payload at the key the bidder gives it (kcp bidder syncs.go): a report at
// <prefix>-payload-<hex(sha256(key))[0..32]>, key onNetwork:<tag>:<receipt> | hostKeys:<receipt>; a
// guest event the n-th of its stream, key guest:<nsid>:<tag>:<kind>:<n>:<receipt>.
const hex = (key) => createHash("sha256").update(key).digest("hex").slice(0, 32);
function bidder(payloads) {
const records = new Map();
const counts = new Map();
for (const value of Object.values(payloads)) {
const c = value.$collection;
const nsid = `${T}.compute.events.${c}`;
let key;
if (c === "vm.onNetwork") key = ["onnetwork", value.tag ? `onNetwork:${value.tag}:${receipt.uri}` : `onNetwork:${receipt.uri}`];
else if (c === "vm.hostKeys") key = ["hostkeys", `hostKeys:${receipt.uri}`];
else {
const stream = `guest:${nsid}:${value.tag}:${value.kind}:`;
const n = counts.get(stream) ?? 0;
counts.set(stream, n + 1);
key = ["guest", `${stream}${n}:${receipt.uri}`];
}
records.set(`at://${HAT9}/${nsid}/${key[0]}-payload-${hex(key[1])}`, value);
}
return async (url) => {
url = new URL(url);
if (url.hostname === "plc.directory") return Response.json({ service: [{ id: "#atproto_pds", serviceEndpoint: PDS }] });
const p = Object.fromEntries(url.searchParams);
assert.ok(!url.pathname.endsWith("listRecords"), "never listRecords");
const value = url.pathname.endsWith("getRecord") && records.get(`at://${p.repo}/${p.collection}/${p.rkey}`);
if (value) return Response.json({ value });
return Response.json({ error: "RecordNotFound" }, { status: 400 });
};
}
test("find, offer, answer, send: a sealed offer to the guest key and the answer that opens", async () => {
const repo = {
k: { $collection: "guest.publicKey", kind: "age-x25519", tag: "shell", key: "age1GUEST", createdAt: "1" },
n: { $collection: "vm.onNetwork", kind: "webrtc", tag: "shell", address: FP, createdAt: "1" },
ssh: { $collection: "vm.onNetwork", kind: "webrtc", tag: "ssh", address: "sha-256 00", createdAt: "2" },
};
const ctx = context(bidder(repo));
let pc;
const cubit = new WebrtcCubit({ age, module: Promise.resolve(module), ctx, makePeer: (config) => ((pc = peer()), (pc.config = config), pc), sleep: async () => {} });
await cubit.find(JSON.stringify(receipt), "shell");
assert.equal(cubit.state.error, null);
assert.deepEqual(cubit.state.guest, { key: "age1GUEST", fingerprint: FP });
await cubit.offer();
assert.deepEqual(pc.config, { iceServers: [{ urls: "stun:stun.cloudflare.com:3478" }] });
assert.equal(pc.dc.label, "shell");
assert.equal(cubit.state.phase, "offered");
assert.deepEqual(cubit.state.signal.receipt, receipt);
assert.deepEqual(JSON.parse(JSON.parse(age.armor.decode(cubit.state.signal.offer)).text), { sdp: "v=0 offer", replyTo: "age1ME" });
// An answer for someone else, then ours.
repo.a1 = { $collection: "guest.signalWebRTC", kind: "answer", tag: "shell", ciphertext: sealed("age1OTHER", { sdp: "other" }), createdAt: "3" };
repo.a2 = { $collection: "guest.signalWebRTC", kind: "answer", tag: "shell", ciphertext: sealed("age1ME", { sdp: `v=0\r\na=fingerprint:${FP.toLowerCase()}\r\n` }), createdAt: "4" };
ctx.http.fetch = bidder(repo);
await cubit.answer();
assert.equal(cubit.state.error, null);
assert.equal(cubit.state.phase, "connecting");
assert.match(pc.remoteDescription.sdp, /a=fingerprint/);
pc.dc.onopen();
assert.equal(cubit.state.phase, "open");
cubit.send("echo hi");
assert.deepEqual(pc.dc.sent, ["echo hi\n"]);
pc.dc.onmessage({ data: new TextEncoder().encode("hi\n").buffer });
assert.equal(cubit.state.output, "hi\n");
});
// The prod round of 2026-11-10 21:06Z: hat-9 published vm.onNetwork for shell only; a find for ssh
// must still see it (the guest's one DTLS certificate), so the watch reads the default tags too.
test("find for a tag whose vm.onNetwork is missing takes another tag's fingerprint", async () => {
const ctx = context(bidder({
k: { $collection: "guest.publicKey", kind: "age-x25519", tag: "ssh", key: "age1GUEST", createdAt: "1" },
n: { $collection: "vm.onNetwork", kind: "webrtc", tag: "shell", address: FP, createdAt: "1" },
}));
const cubit = new WebrtcCubit({ age, module: Promise.resolve(module), ctx, makePeer: peer });
await cubit.find(JSON.stringify(receipt), "ssh");
assert.equal(cubit.state.error, null);
assert.deepEqual(cubit.state.guest, { key: "age1GUEST", fingerprint: FP });
});
test("find reports a guest that has not published its key or fingerprint", async () => {
const ctx = context(bidder({}));
const cubit = new WebrtcCubit({ age, module: Promise.resolve(module), ctx, makePeer: peer });
await cubit.find(JSON.stringify(receipt), "shell");
assert.equal(cubit.state.phase, "error");
assert.match(cubit.state.error, /guest.publicKey .*kind webrtc/);
// contract's mcp URLs go to the webrtc: fetcher, which signals as the session.
assert.deepEqual(Object.keys(cubit.fetchers), ["webrtc:"]);
await assert.rejects(cubit.fetchers["webrtc:"]({ url: "webrtc:///x/shell/mcp" }), /needs a signed-in session/);
});
test("fingerprint check and receipt parsing", () => {
checkFingerprint("a=fingerprint:SHA-256 ab:cd:ef\r\n", FP);
assert.throws(() => checkFingerprint("v=0\r\n", FP), /no a=fingerprint/);
assert.throws(() => checkFingerprint("a=fingerprint:sha-256 00\r\n", FP), /is not the guest's/);
assert.deepEqual(parseReceipt(JSON.stringify(receipt)), { ...receipt, bidder: HAT9 });
assert.throws(() => parseReceipt('{"uri":"at://x"}'), /strong ref/);
// hat-9 published vm.onNetwork for the first tag only; the agent's one fingerprint serves all.
const onNetwork = { kind: "vm.onNetwork", value: { kind: "webrtc", tag: "ssh", address: FP } };
assert.equal(guestOf([onNetwork], "shell").fingerprint, FP);
assert.equal(guestOf([onNetwork, { ...onNetwork, value: { ...onNetwork.value, tag: "shell", address: "sha-256 01" } }], "shell").fingerprint, "sha-256 01");
});
// Signed in: the OAuth agent is stubbed at fetchHandler, the curve is a fake (attest.test.mjs
// covers real signing); checks the host contract routes: attest, repo writes, https service.
// The page's forms give text; the module reads numbers and booleans (hold_ms as text was no hold).
test("form text goes as each field's module.json type; lists and other text as typed", async () => {
const inputs = [];
const ctx = { ...context(bidder({})), wasm: { run: async ({ input }) => (inputs.push(input), Promise.reject(new Error("stop"))) } };
const cubit = new WebrtcCubit({ age, module: Promise.resolve({ ...module, actions: spec.actions }), ctx, makePeer: peer });
await assert.rejects(cubit.run("contract", { hold_ms: "60000", release: "false", shell: "true", cpus: "2", bidders: "https://a https://b", tags: '["x"]', user_data: "7" }));
const { hold_ms, release, shell, cpus, bidders, tags, user_data } = inputs[0];
assert.deepEqual({ hold_ms, release, shell, cpus, bidders, tags, user_data }, { hold_ms: 60000, release: false, shell: true, cpus: 2, bidders: "https://a https://b", tags: ["x"], user_data: "7" });
});
test("signed in: run plans through attest, the session's PDS, and service-auth to an https bidder", async () => {
const alice = "did:plc:lpfuqerea3deuoyrn7ojser4";
const xrpc = [];
const agent = {
did: alice,
com: { atproto: { server: { getSession: async () => ({ success: true, data: { handle: "alice.test" } }) } } },
async fetchHandler(path, init) {
const [nsid, query] = path.replace("/xrpc/", "").split("?");
const input = init.body && JSON.parse(init.body);
xrpc.push({ nsid, query, input, headers: init.headers });
if (nsid === "com.atproto.server.getServiceAuth") return Response.json({ token: "svc-jwt" });
if (nsid === "com.atproto.repo.createRecord") return Response.json({ uri: `at://${alice}/${input.collection}/r${xrpc.length}`, cid: `bafy${xrpc.length}` });
if (nsid === "com.atproto.repo.deleteRecord") return Response.json({});
return new Response("{}", { status: 400 });
},
};
const bidderCalls = [];
const fetch = async (url, init) => (bidderCalls.push({ url, init }), Response.json({ ok: true, name: "bid-1" }));
const store = new Map();
const ctx = { ...context(fetch), settings: { get: async (k) => store.get(k), set: async (k, v) => store.set(k, v) } };
const secp256k1 = { utils: { randomSecretKey: () => new Uint8Array(32).fill(1) }, getPublicKey: () => new Uint8Array(33).fill(2), sign: () => new Uint8Array(64) };
const cubit = new WebrtcCubit({ client: Promise.resolve({ init: async () => ({ session: {} }) }), makeAgent: () => agent, secp256k1, age, module: Promise.resolve(module), ctx, makePeer: peer });
await cubit.init();
assert.equal(cubit.state.session, "signed-in");
const bound = await cubit.run("bind_key");
assert.equal(xrpc[0].input.collection, `${T}.badgeBlueKeys`);
assert.match(xrpc[0].input.record.keyId, /^did:key:z/);
assert.ok(store.get("attestKeys")[alice]);
assert.deepEqual(cubit.state.results.bind_key, bound);
const endpoint = "https://bidder.example";
await cubit.run("request", { user_data: "#cloud-config\n", bidders: [endpoint] });
const rfp = xrpc.find((c) => c.input?.collection === `${T}.market.rfp`).input.record;
assert.equal(rfp.signatures[0].issuer, alice);
assert.deepEqual(xrpc.find((c) => c.nsid === "com.atproto.server.getServiceAuth").query, `aud=did%3Aweb%3Abidder.example&lxm=${T}.market.submitRfp`);
assert.equal(bidderCalls[0].url, `${endpoint}/xrpc/${T}.market.submitRfp`);
assert.equal(bidderCalls[0].init.headers.authorization, "Bearer svc-jwt");
await assert.rejects(cubit.deleteRecords([`at://${HAT9}/${T}.market.rfp/x`]), /not in your repo/);
assert.deepEqual(await cubit.deleteRecords([`at://${alice}/${T}.market.rfp/x`]), [`at://${alice}/${T}.market.rfp/x`]);
assert.deepEqual(xrpc.at(-1).input, { repo: alice, collection: `${T}.market.rfp`, rkey: "x" });
});
// The requester's shell entry plans com.fedfork.exec.webrtc.connect; the page's call() runs
// find/offer/signal/answer and resolves once the channel opens. Other pseudo-calls never reach the PDS.
test("signed in: shell -> webrtc.connect opens the channel; iroh.connect is refused here", async () => {
const alice = "did:plc:lpfuqerea3deuoyrn7ojser4";
const repo = {
k: { $collection: "guest.publicKey", kind: "age-x25519", tag: "ssh", key: "age1GUEST", createdAt: "1" },
n: { $collection: "vm.onNetwork", kind: "webrtc", tag: "ssh", address: FP, createdAt: "1" },
};
const store = new Map();
const ctx = { ...context(bidder(repo)), settings: { get: async (k) => store.get(k), set: async (k, v) => store.set(k, v) } };
const written = [];
const agent = {
did: alice,
com: { atproto: { server: { getSession: async () => ({ success: true, data: { handle: "alice.test" } }) } } },
async fetchHandler(path, init) {
const nsid = path.replace("/xrpc/", "").split("?")[0];
const input = init.body && JSON.parse(init.body);
written.push(nsid === "com.atproto.repo.createRecord" ? input.collection : nsid);
if (input?.collection === `${T}.compute.events.guest.signalWebRTC`) {
// The guest answers once the offer is published.
repo.a = { $collection: "guest.signalWebRTC", kind: "answer", tag: "ssh", ciphertext: sealed("age1ME", { sdp: `a=fingerprint:${FP}\r\n` }), createdAt: "2" };
ctx.http.fetch = bidder(repo);
}
return Response.json({ uri: `at://${alice}/${input?.collection}/r${written.length}`, cid: `bafy${written.length}` });
},
};
const secp256k1 = { utils: { randomSecretKey: () => new Uint8Array(32).fill(1) }, getPublicKey: () => new Uint8Array(33).fill(2), sign: () => new Uint8Array(64) };
let pc;
const makePeer = () => {
pc = peer();
pc.setRemoteDescription = async (d) => { pc.remoteDescription = d; queueMicrotask(() => pc.dc.onopen()) };
return pc;
};
const cubit = new WebrtcCubit({ client: Promise.resolve({ init: async () => ({ session: {} }) }), makeAgent: () => agent, secp256k1, age, module: Promise.resolve(module), ctx, makePeer, sleep: async () => {} });
await cubit.init();
const result = await cubit.run("shell", { receipt: JSON.stringify(receipt) });
assert.deepEqual(result, { via: "webrtc", result: { tag: "ssh", fingerprint: FP } });
assert.equal(cubit.state.phase, "open");
assert.equal(pc.dc.label, "ssh");
assert.deepEqual(written, [`${T}.compute.events.guest.signalWebRTC`, `${T}.market.event`]);
await assert.rejects(cubit.run("shell", { receipt: JSON.stringify(receipt), command: "id" }), /only over tag ssh, with an ssh client/);
repo.i = { $collection: "vm.onNetwork", address: "iroh://endpointabc", createdAt: "3" };
repo.h = { $collection: "vm.hostKeys", keys: [{ type: "ssh-ed25519", publicKey: "AAAA" }], createdAt: "3" };
ctx.http.fetch = bidder(repo);
const before = written.length;
await assert.rejects(cubit.run("shell", { receipt: JSON.stringify(receipt), via: "iroh", command: "id" }), /com\.fedfork\.exec\.iroh\.connect: not available on this host \(built without iroh\)/);
assert.equal(written.length, before);
});
// Offer creation waits only as long as gathering needs: complete, the null candidate, 2 s after
// the first srflx, or the 20 s cap. Fake timers; the peer fires candidates by hand.
test("offer stops gathering on complete, the null candidate, 2 s after the first srflx, or 20 s", async (t) => {
mock.timers.enable({ apis: ["setTimeout"] });
t.after(() => mock.timers.reset());
const repo = {
k: { $collection: "guest.publicKey", kind: "age-x25519", tag: "shell", key: "age1GUEST", createdAt: "1" },
n: { $collection: "vm.onNetwork", kind: "webrtc", tag: "shell", address: FP, createdAt: "1" },
};
const candidate = (type) => Object.assign(new Event("icecandidate"), { candidate: type && { candidate: `candidate:1 1 udp 1 192.0.2.1 5000 typ ${type} generation 0` } });
const flush = () => new Promise((r) => setImmediate(r));
async function offering(fire) {
let pc;
const cubit = new WebrtcCubit({ age, module: Promise.resolve(module), ctx: context(bidder(repo)), makePeer: () => ((pc = peer()), (pc.gather = "manual"), pc) });
await cubit.find(JSON.stringify(receipt), "shell");
const offered = cubit.offer();
while (!pc?.localDescription) await flush();
await fire(pc);
await flush();
return { cubit, offered };
}
let { cubit, offered } = await offering((pc) => pc.dispatchEvent(candidate(null)));
await offered;
assert.equal(cubit.state.phase, "offered", "null candidate");
({ cubit, offered } = await offering((pc) => { pc.dispatchEvent(candidate("host")); pc.dispatchEvent(candidate("srflx")); pc.dispatchEvent(candidate("srflx")) }));
mock.timers.tick(1999);
await flush();
assert.equal(cubit.state.phase, "offering", "still gathering 1999 ms after srflx");
mock.timers.tick(1);
await offered;
assert.equal(cubit.state.phase, "offered", "2 s after the first srflx");
({ cubit, offered } = await offering((pc) => pc.dispatchEvent(candidate("host"))));
mock.timers.tick(19999);
await flush();
assert.equal(cubit.state.phase, "offering", "host candidates only: still gathering");
mock.timers.tick(1);
await offered;
assert.equal(cubit.state.phase, "offered", "20 s cap");
});
// A subscribe to webrtc.connect is a channel session; messages in, sends out.
test("signed in: a subscribe to webrtc.connect streams the datachannel both ways", async () => {
const alice = "did:plc:lpfuqerea3deuoyrn7ojser4";
const repo = {
k: { $collection: "guest.publicKey", kind: "age-x25519", tag: "tty", key: "age1GUEST", createdAt: "1" },
n: { $collection: "vm.onNetwork", kind: "webrtc", tag: "tty", address: FP, createdAt: "1" },
};
let pc;
const ctx = {
http: { fetch: bidder(repo) },
settings: (() => {
const store = new Map();
return { get: async (k) => store.get(k), set: async (k, v) => void store.set(k, v) };
})(),
// The handler stands in for a module: subscribe, echo the first message back, finish.
wasm: {
run: async ({ entry, input }) => {
// The requester runs every other entry (find, signal); "tty" is the test's own handler.
if (entry !== "tty") return run(entry, input);
const { event, receipt: r } = input;
if (event.type === "start") return { calls: [{ id: "tty", nsid: "com.fedfork.exec.webrtc.connect", params: { receipt: r, tag: "tty" }, subscribe: true }] };
if (event.type === "message") return { send: [{ id: "tty", data: `echo ${event.data}` }], result: event.data, done: true };
return { result: event };
},
},
};
const agent = {
did: alice,
com: { atproto: { server: { getSession: async () => ({ success: true, data: { handle: "alice.test" } }) } } },
async fetchHandler(path, init) {
const input = init.body && JSON.parse(init.body);
if (input?.collection === `${T}.compute.events.guest.signalWebRTC`) {
repo.a = { $collection: "guest.signalWebRTC", kind: "answer", tag: "tty", ciphertext: sealed("age1ME", { sdp: `a=fingerprint:${FP}\r\n` }), createdAt: "2" };
ctx.http.fetch = bidder(repo);
}
return Response.json({ uri: `at://${alice}/${input?.collection}/r1`, cid: "bafy1" });
},
};
const makePeer = () => {
pc = peer();
pc.setRemoteDescription = async (d) => {
pc.remoteDescription = d;
queueMicrotask(() => pc.dc.onopen());
setTimeout(() => pc.dc.onmessage({ data: new TextEncoder().encode("login: ").buffer }), 5);
};
return pc;
};
const secp256k1 = { utils: { randomSecretKey: () => new Uint8Array(32).fill(1) }, getPublicKey: () => new Uint8Array(33).fill(2), sign: () => new Uint8Array(64) };
const cubit = new WebrtcCubit({ client: Promise.resolve({ init: async () => ({ session: {} }) }), makeAgent: () => agent, secp256k1, age, module: Promise.resolve(module), ctx, makePeer, sleep: async () => {} });
await cubit.init();
assert.equal(await cubit.run("tty", { receipt: JSON.stringify(receipt) }), "login: ");
assert.deepEqual(pc.dc.sent, ["echo login: "]);
assert.equal(cubit.state.phase, "open", "closing the subscription leaves the channel open");
});
// A peer connection failing before the channel opens is the NAT case; after it opened (live:
// the contract released the VM) it is the guest going away, so the page closes, no NAT error.
test("a failed peer connection is a NAT error before open and a close after", async () => {
const repo = {
k: { $collection: "guest.publicKey", kind: "age-x25519", tag: "shell", key: "age1GUEST", createdAt: "1" },
n: { $collection: "vm.onNetwork", kind: "webrtc", tag: "shell", address: FP, createdAt: "1" },
};
const fail = (pc) => ((pc.connectionState = "failed"), (pc.iceConnectionState = "disconnected"), pc.dispatchEvent(new Event("connectionstatechange")));
for (const opened of [false, true]) {
let pc;
const cubit = new WebrtcCubit({ age, module: Promise.resolve(module), ctx: context(bidder(repo)), makePeer: () => (pc = peer()) });
await cubit.find(JSON.stringify(receipt), "shell");
await cubit.offer();
if (opened) pc.dc.onopen();
fail(pc);
if (opened) assert.deepEqual([cubit.state.phase, cubit.state.error], ["closed", null]);
else assert.match(cubit.state.error, /symmetric NAT/);
}
});