johnandersen777.bsky.social/atlibs-webrtc
WebRTC transport for atlibs: com.fedfork.exec.webrtc.connect in the browser, and webrtc-peer, its guest side
0
0
// node --test tests/loopback.test.mjs (needs `cargo build --release` and ../atlibs; no VM, no market, no account)
// The WebRTC check: `webrtc-peer` (this crate's binary: str0m, the guest
// side: age-sealed offer -> answer, datachannel label -> TCP port) in front of a throwaway sshd (ecdsa host
// key: the page's ssh library has no ed25519), and web/loopback.html in headless Chrome: the
// page runs com.fedfork.exec.webrtc.connect {tag: ssh, command, stdin} through the local signal
// endpoint (same sealed messages as the records), then ssh exec over the datachannel; and the webrtc:
// fetcher to a stub MCP server (Streamable HTTP) behind tag shell. Skips (with the reason)
// when Chrome, sshd or ssh-keygen is missing, or there is no `cargo build --release`.
import assert from "node:assert/strict";
import { access, mkdtemp, readFile, writeFile } from "node:fs/promises";
import { createServer as createHttpServer } from "node:http";
import { tmpdir, userInfo } from "node:os";
import { extname, join, normalize } from "node:path";
import { after, before, describe, test } from "node:test";
// The core's loopback helpers, from the sibling checkout (../atlibs).
import { cleanup, exec, missing, servers, spawnUntil, sshd, temps, which } from "../../atlibs/rust/host/loopback.mjs";
const ROOT = new URL("../", import.meta.url).pathname;
after(cleanup);
// Headless Chrome over the DevTools protocol (one browser WebSocket, flattened target sessions).
const CHROMES = ["/Applications/Google Chrome.app/Contents/MacOS/Google Chrome", "google-chrome", "google-chrome-stable", "chromium", "chromium-browser"];
async function chrome() {
for (const c of [process.env.CHROME, ...CHROMES].filter(Boolean)) {
if (c.startsWith("/") ? await access(c).then(() => true, () => false) : await which(c)) return c;
}
return null;
}
async function devtools(bin, profile) {
const args = ["--headless=new", "--remote-debugging-port=0", `--user-data-dir=${profile}`, "--no-first-run", "--no-default-browser-check", "--disable-gpu",
// loopback ICE candidates, and real addresses instead of mDNS names (the peer cannot resolve .local)
"--allow-loopback-in-peer-connection", "--disable-features=WebRtcHideLocalIpsWithMdns", "about:blank"];
const { match } = await spawnUntil(bin, args, /DevTools listening on (ws:\/\/\S+)/, {}, 30000);
const ws = new WebSocket(match[1]);
await new Promise((resolve, reject) => ((ws.onopen = resolve), (ws.onerror = reject)));
let id = 0;
const pending = new Map();
ws.onmessage = ({ data }) => {
const m = JSON.parse(data);
const p = pending.get(m.id);
if (!p) return;
pending.delete(m.id);
m.error ? p.reject(new Error(`${m.error.message} ${m.error.data ?? ""}`)) : p.resolve(m.result);
};
const send = (method, params = {}, sessionId) =>
new Promise((resolve, reject) => {
pending.set(++id, { resolve, reject });
ws.send(JSON.stringify({ id, method, params, sessionId }));
});
return {
close: () => ws.close(),
async open(url) {
const { targetId } = await send("Target.createTarget", { url });
const { sessionId } = await send("Target.attachToTarget", { targetId, flatten: true });
// An expression in the page, awaited; its value or the page's exception.
return async (expression) => {
const { result, exceptionDetails } = await send("Runtime.evaluate", { expression, awaitPromise: true, returnByValue: true }, sessionId);
if (exceptionDetails) throw new Error(exceptionDetails.exception?.description ?? exceptionDetails.text);
return result.value;
};
},
};
}
// web/ as files (symlinks: atlibs/ is ../atlibs/javascript/src, webrtc/ is js/), as the page is served in production.
function staticServer(dir) {
const types = { ".html": "text/html", ".js": "text/javascript", ".json": "application/json", ".wasm": "application/wasm" };
return createHttpServer(async (req, res) => {
const path = normalize(decodeURIComponent(new URL(req.url, "http://x").pathname));
if (path.includes("..")) return res.writeHead(400).end();
try {
const body = await readFile(join(dir, path));
res.writeHead(200, { "content-type": types[extname(path)] ?? "application/octet-stream" }).end(body);
} catch {
res.writeHead(404).end();
}
}).listen(0, "127.0.0.1");
}
// A stub MCP server (Streamable HTTP, JSON responses): what the guest agent serves on tag shell. tools/call
// echoes arguments.command after arguments.waitMs. seen: TCP connections (one per datachannel) and the
// Mcp-Session-Id of each request.
function mcpServer() {
const seen = { connections: 0, sessions: [] };
const server = createHttpServer((req, res) => {
let body = "";
req.on("data", (d) => (body += d)).on("end", () => {
const m = JSON.parse(body);
seen.sessions.push(req.headers["mcp-session-id"] ?? null);
if (req.method !== "POST" || req.url !== "/mcp") return res.writeHead(404).end();
if (m.id === undefined) return res.writeHead(202).end();
const reply = (result) => res.writeHead(200, { "content-type": "application/json", ...(m.method === "initialize" && { "mcp-session-id": "s1" }) }).end(JSON.stringify({ jsonrpc: "2.0", id: m.id, result }));
if (m.method === "initialize") return reply({ protocolVersion: "2025-11-25", capabilities: { tools: {} }, serverInfo: { name: "stub" } });
if (m.method === "tools/list") return reply({ tools: [{ name: "exec", inputSchema: { type: "object" } }] });
const { command, waitMs = 0 } = m.params.arguments;
setTimeout(() => reply({ content: [{ type: "text", text: command }] }), waitMs);
});
}).on("connection", () => seen.connections++);
return { server: server.listen(0, "127.0.0.1"), seen };
}
describe("webrtc: headless Chrome <-> webrtc-peer + a throwaway sshd", async () => {
const absent = await missing("sshd", "ssh-keygen");
const bin = await chrome();
const peerBin = process.env.WEBRTC_PEER ?? join(ROOT, "target/release/webrtc-peer");
const built = await access(peerBin).then(() => true, () => false);
const skip = !bin ? "no Chrome (set CHROME to its binary)" : absent.length ? `missing ${absent.join(", ")} (OpenSSH)` : !built ? `no ${peerBin}: cargo build --release` : false;
let dir, peer, evaluate, browser, hostFingerprint, page, mcp;
const RECEIPT = { uri: "at://did:plc:loopback/com.publicdomainrelay.temp.market.event/receipt", cid: "bafyreiloopback" };
before(async () => {
if (skip) return;
dir = await mkdtemp(join(tmpdir(), "atlibs-loopback-webrtc-"));
temps.push(dir);
await writeFile(join(dir, "authorized_keys"), "");
const { port } = await sshd(dir, "ecdsa");
hostFingerprint = (await exec("ssh-keygen", ["-lf", join(dir, "host_key.pub")])).stdout.split(" ")[1];
const files = staticServer(join(ROOT, "web"));
servers.push(files);
await new Promise((r) => files.once("listening", r));
page = `http://127.0.0.1:${files.address().port}`;
// --origin: the page is the one origin CORS lets read the signal endpoint (and POST to it)
mcp = mcpServer();
servers.push(mcp.server);
await new Promise((r) => mcp.server.once("listening", r));
const started = await spawnUntil(peerBin, ["--tunnel", `ssh=${port}`, "--tunnel", `shell=${mcp.server.address().port}`, "--origin", page], /^\{.*\}$/m, { env: { PATH: process.env.PATH, HOME: process.env.HOME } });
peer = JSON.parse(started.match[0]);
// A kept profile under target/: the page's esm.sh modules (pinned versions) come from its cache after the first run.
browser = await devtools(bin, join(ROOT, "target/loopback-chrome"));
evaluate = await browser.open(`${page}/loopback.html?signal=${encodeURIComponent(peer.signal)}&user=${userInfo().username}`);
for (const end = Date.now() + 30000; (await evaluate("document.title")) !== "ready"; await new Promise((r) => setTimeout(r, 100))) {
if (Date.now() > end) throw new Error("loopback.html did not load (its modules come from esm.sh: offline with an empty cache?)");
}
});
after(() => browser?.close());
test("the local signal endpoint has the guest's age key and DTLS fingerprint for the tunnel's tag", { skip }, async () => {
const { events } = await (await fetch(new URL("events", peer.signal))).json();
assert.deepEqual(events.map((e) => [e.kind, e.value.tag]).sort(), [["guest.publicKey", "shell"], ["guest.publicKey", "ssh"], ["vm.onNetwork", "shell"], ["vm.onNetwork", "ssh"]]);
assert.match(events.find((e) => e.kind === "guest.publicKey").value.key, /^age1[a-z0-9]{58}$/);
assert.match(events.find((e) => e.kind === "vm.onNetwork").value.address, /^sha-256 ([0-9A-F]{2}:){31}[0-9A-F]{2}$/);
});
// review 4a L5: only the page's origin and this listener's Host get in (another site's page, DNS rebinding)
test("the signal endpoint answers CORS for the page only and refuses other Origins and Hosts", { skip }, async () => {
const ours = await fetch(new URL("events", peer.signal), { headers: { origin: page } });
assert.deepEqual([ours.status, ours.headers.get("access-control-allow-origin")], [200, page]);
const evil = await fetch(new URL("signal", peer.signal), { method: "POST", headers: { origin: "https://evil.example" }, body: "{}" });
assert.equal(evil.status, 403);
const { port } = new URL(peer.signal);
const { stdout } = await exec("curl", ["-s", "-o", "/dev/null", "-w", "%{http_code}", "-H", `Host: evil.example:${port}`, new URL("events", peer.signal).href]);
assert.equal(stdout, "403", "a rebound Host");
});
test("webrtc.connect from the page: sealed offer, answer, datachannel, ssh exec with stdin", { skip }, async () => {
await writeFile(join(dir, "authorized_keys"), `${await evaluate("webrtcCubit.sshKey()")}\n`);
const started = performance.now();
const result = await evaluate(`webrtcCubit.call("com.fedfork.exec.webrtc.connect", { input: { receipt: ${JSON.stringify(RECEIPT)}, tag: "ssh", command: "echo over-webrtc; cat", stdin: "from-stdin" } })`);
const ms = performance.now() - started;
assert.deepEqual(result, { tag: "ssh", fingerprint: peer.fingerprint, output: "over-webrtc\nfrom-stdin", hostKey: { type: "ecdsa-sha2-nistp256", fingerprint: hostFingerprint, pinned: false } });
assert.equal(await evaluate(`webrtcCubit.sshExec("echo second; exit 3")`), "second\n", "a second exec on the same ssh session");
const { events } = await (await fetch(new URL("events", peer.signal))).json();
assert.equal(events.filter((e) => e.kind === "guest.signalWebRTC" && e.value.kind === "answer").length, 1);
console.log(`webrtc.connect + ssh exec: ${ms.toFixed(0)} ms`);
});
test("the webrtc: fetcher from the page: MCP over HTTP, a notification, two concurrent requests on two datachannels", { skip }, async () => {
let session;
const post = async (message) => {
const headers = { "content-type": "application/json", accept: "application/json, text/event-stream", ...(session && { "mcp-session-id": session }) };
const response = await evaluate(`webrtcCubit.fetch(${JSON.stringify({ receipt: RECEIPT, tag: "shell", path: "mcp", method: "POST", headers, body: JSON.stringify(message) })})`);
session ??= response.headers["mcp-session-id"];
return response;
};
const started = performance.now();
const init = await post({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-11-25", capabilities: {}, clientInfo: { name: "loopback" } } });
console.log(`webrtc: fetch initialize (connect + first request): ${(performance.now() - started).toFixed(0)} ms`);
assert.deepEqual([init.status, JSON.parse(init.body).result.serverInfo.name, session], [200, "stub", "s1"]);
assert.equal((await post({ jsonrpc: "2.0", method: "notifications/initialized" })).status, 202);
assert.equal(JSON.parse((await post({ jsonrpc: "2.0", id: 2, method: "tools/list" })).body).result.tools[0].name, "exec");
assert.equal(mcp.seen.connections, 1, "keep-alive: one datachannel so far");
const call = async (id, command, waitMs) => JSON.parse((await post({ jsonrpc: "2.0", id, method: "tools/call", params: { name: "exec", arguments: { command, waitMs } } })).body);
const [slow, fast] = await Promise.all([call(3, "slow", 1500), call(4, "fast", 0)]);
assert.deepEqual([slow.id, slow.result.content[0].text, fast.id, fast.result.content[0].text], [3, "slow", 4, "fast"]);
assert.equal(mcp.seen.connections, 2, "a second datachannel for the concurrent request");
assert.deepEqual(mcp.seen.sessions, [null, "s1", "s1", "s1", "s1"], "the client's Mcp-Session-Id rides through");
const { events } = await (await fetch(new URL("events", peer.signal))).json();
assert.equal(events.filter((e) => e.kind === "guest.signalWebRTC" && e.value.tag === "shell").length, 1, "one peer connection for (receipt, shell)");
});
});