johnandersen777.bsky.social/atlibs-webrtc

Switch repository

Loading…

// node --test js/fetch.test.mjs
// The webrtc: fetcher on a fake peer connection (each datachannel a TCP connection to a Node HTTP server, as
// the guest agent bridges a tag; age and signalling faked): URL parsing, one peer per (receipt, tag), a
// channel per concurrent request, keep-alive reuse, a redial after the peer fails.
import assert from "node:assert/strict";
import { createServer } from "node:http";
import { connect } from "node:net";
import { after, test } from "node:test";

import { LISTEN_COLLECTION, WebrtcHost, endpointUrl, parseWebrtcUrl, webrtcUrl } from "./index.js";

const seen = { connections: 0, requests: [] };
const http = createServer((req, res) => {
	let body = "";
	req.on("data", (d) => (body += d)).on("end", () => {
		seen.requests.push({ method: req.method, url: req.url, body, session: req.headers["mcp-session-id"] ?? null });
		const wait = Number(new URL(req.url, "http://x").searchParams.get("wait") ?? 0);
		setTimeout(() => res.writeHead(200, { "content-type": "application/json", "x-echo": req.headers["x-echo"] ?? "" }).end(JSON.stringify({ url: req.url, body })), wait);
	});
}).on("connection", () => seen.connections++).listen(0, "127.0.0.1");
await new Promise((r) => http.once("listening", r));
const sockets = [];
after(() => (http.close(), sockets.forEach((s) => s.destroy())));

const FINGERPRINT = "sha-256 AA:BB";
// Identity "age": the sealed offer is its plaintext, every answer opens.
const age = {
	generateX25519Identity: async () => "identity",
	identityToRecipient: async () => "recipient",
	Encrypter: class { addRecipient() {} encrypt = async (text) => text; },
	Decrypter: class { addIdentity() {} decrypt = async (text) => text; },
	armor: { encode: (x) => x, decode: (x) => x },
};
const peers = [];
// A peer connection whose datachannels are TCP connections to the HTTP server.
function makePeer() {
	const pc = new EventTarget();
	Object.assign(pc, {
		connectionState: "connected",
		iceGatheringState: "new",
		// A channel opens once the answer is in (the first one is made before the offer).
		pending: [],
		createDataChannel() {
			const dc = { readyState: "connecting" };
			const open = () => {
				const socket = connect(http.address().port, "127.0.0.1");
				sockets.push(socket);
				Object.assign(dc, { send: (b) => socket.write(b), close: () => socket.end() });
				socket.on("connect", () => ((dc.readyState = "open"), dc.onopen?.()));
				socket.on("data", (d) => dc.onmessage?.({ data: d.buffer.slice(d.byteOffset, d.byteOffset + d.length) }));
				socket.on("close", () => ((dc.readyState = "closed"), dc.onclose?.()));
			};
			this.pending ? this.pending.push(open) : open();
			return dc;
		},
		createOffer: async () => ({ type: "offer", sdp: "v=0" }),
		async setLocalDescription(d) {
			this.localDescription = d;
			setTimeout(() => ((this.iceGatheringState = "complete"), this.dispatchEvent(new Event("icegatheringstatechange"))));
		},
		async setRemoteDescription() {
			this.pending.splice(0).forEach((open) => open());
			this.pending = null;
		},
		close() {
			this.connectionState = "closed";
		},
	});
	peers.push(pc);
	return pc;
}
const signals = [];
const host = new WebrtcHost({
	events: async ({ receipt, tag }) => [
		{ kind: "guest.publicKey", value: { tag, key: "age1guest" } },
		{ kind: "vm.onNetwork", value: { tag, kind: "webrtc", address: FINGERPRINT } },
		...signals.filter((s) => s.receipt.uri === receipt.uri && s.tag === tag).map(() => ({ kind: "guest.signalWebRTC", value: { tag, kind: "answer", ciphertext: JSON.stringify({ sdp: `a=fingerprint:${FINGERPRINT}` }) } })),
	],
	signal: async (input) => signals.push(input),
	makePeer,
	age,
	sleep: async () => {},
});
const fetchWebrtc = host.fetchers()["webrtc:"];
const RECEIPT = { uri: "at://did:plc:bidder/com.publicdomainrelay.temp.market.event/r1", cid: "bafyr1" };
const url = (path, receipt = RECEIPT) => webrtcUrl({ receipt }, "shell", path);

test("webrtc URLs: ref, tag, path with the query; bad ones refused", () => {
	assert.deepEqual(parseWebrtcUrl(url("mcp?x=1")), { receipt: { ...RECEIPT, bidder: "did:plc:bidder" }, tag: "shell", path: "/mcp?x=1" });
	assert.equal(parseWebrtcUrl(webrtcUrl({ receipt: RECEIPT, bidder: "did:web:b.example" }, "shell", "mcp")).receipt.bidder, "did:web:b.example");
	assert.throws(() => parseWebrtcUrl("webrtc://host/x/shell/mcp"), /host must be base64url JSON \{did, listen\}/);
	assert.throws(() => parseWebrtcUrl("webrtc:///"), /webrtc:\/\/\/<ref>/);
	assert.throws(() => parseWebrtcUrl("webrtc:///bm90LWpzb24/shell/mcp"), /base64url JSON/);
	assert.throws(() => parseWebrtcUrl(webrtcUrl({ receipt: RECEIPT }, "ssh", "mcp")), /not ssh/);
	assert.throws(() => parseWebrtcUrl(webrtcUrl({ receipt: { uri: "nope", cid: "x" } }, "shell", "")), /strong ref/);
});

test("fetch: one peer per (receipt, tag), keep-alive, a channel per concurrent request", async () => {
	const post = (path, body) => fetchWebrtc({ method: "POST", url: url(path), headers: { "content-type": "application/json", "mcp-session-id": "s1", "content-length": "999" }, body });
	const one = await post("mcp", '{"a":1}');
	assert.deepEqual([one.status, one.headers["content-type"], JSON.parse(one.body)], [200, "application/json", { url: "/mcp", body: '{"a":1}' }]);
	assert.equal((await fetchWebrtc({ url: url("mcp"), headers: { "x-echo": "hi" } })).headers["x-echo"], "hi", "GET by default; headers ride through");
	assert.equal(seen.connections, 1, "keep-alive on the first channel");
	const [slow, fast] = await Promise.all([post("mcp?wait=300", "slow"), post("mcp", "fast")]);
	assert.deepEqual([JSON.parse(slow.body).body, JSON.parse(fast.body).body], ["slow", "fast"]);
	assert.equal(seen.connections, 2, "a second channel for the concurrent request");
	await Promise.all([post("mcp", "a"), post("mcp", "b")]);
	assert.equal(seen.connections, 2, "both pooled channels reused");
	assert.equal(peers.length, 1, "one peer connection");
	assert.equal(signals.length, 1, "one offer");
	assert.ok(seen.requests.filter((r) => r.method === "POST").every((r) => r.session === "s1"));
});

test("fetch: a failed peer is redialled; another receipt has its own peer; abort", async () => {
	peers[0].connectionState = "failed";
	assert.equal((await fetchWebrtc({ url: url("again") })).status, 200);
	assert.deepEqual([peers.length, peers[0].connectionState], [2, "closed"], "dialled again, the old one closed");
	await fetchWebrtc({ url: url("x", { ...RECEIPT, uri: RECEIPT.uri.replace("r1", "r2") }) });
	assert.equal(peers.length, 3);
	const controller = new AbortController();
	const pending = fetchWebrtc({ url: url("mcp?wait=2000"), signal: controller.signal });
	setTimeout(() => controller.abort(), 50);
	await assert.rejects(pending, /webrtc fetch: aborted/);
	await assert.rejects(fetchWebrtc({ url: url("x"), headers: { "x-bad": "a\r\nb" } }), /a bad header/);
	host.close();
	await new Promise((r) => setTimeout(r));
	assert.ok(peers.every((p) => p.connectionState === "closed"), "close closes every peer");
});

test("an endpoint URL (a bidder's offering webrtc endpoint) is signalled over records: offer in our repo, answer in theirs", async () => {
	const { recordSignals, OFFER, ANSWER } = await import("./record-signals.js");
	const BIDDER = "did:plc:bidder";
	const repos = new Map([[`${BIDDER} ${LISTEN_COLLECTION} market`, { value: { key: "age1bidder", fingerprint: FINGERPRINT }, cid: "c" }]]);
	const writes = [];
	const repo = {
		did: "did:plc:page",
		getRecord: async ({ repo: did, collection, rkey }) => repos.get(`${did} ${collection} ${rkey}`) ?? null,
		putRecord: async ({ collection, rkey, record }) => {
			writes.push(["put", collection, rkey, record.to]);
			repos.set(`did:plc:page ${collection} ${rkey}`, { value: record, cid: "o" });
			// the bidder: one answer per offer, in its own repo at <offerer DID>.<tid>
			const tid = rkey.slice(BIDDER.length + 1);
			repos.set(`${BIDDER} ${ANSWER} did:plc:page.${tid}`, { value: { ciphertext: JSON.stringify({ sdp: `a=fingerprint:${FINGERPRINT}` }) }, cid: "a" });
		},
		deleteRecord: async ({ collection, rkey }) => void writes.push(["delete", collection, rkey]),
	};
	const records = recordSignals({ repo, subhandle: "page.test" });
	const bidderHost = new WebrtcHost({ records, makePeer, age, sleep: async () => {} });
	const at = endpointUrl({ did: BIDDER, listen: "market" }, "xrpc/com.publicdomainrelay.temp.market.submitRfp");
	assert.deepEqual(parseWebrtcUrl(at), { receipt: { uri: `at://${BIDDER}/${LISTEN_COLLECTION}/market`, cid: "listen", bidder: BIDDER }, tag: "market", path: "/xrpc/com.publicdomainrelay.temp.market.submitRfp", records: true });
	const answered = await bidderHost.fetchers()["webrtc:"]({ method: "POST", url: at, headers: { authorization: "Bearer t" }, body: "{}" });
	assert.deepEqual([answered.status, JSON.parse(answered.body).url], [200, "/xrpc/com.publicdomainrelay.temp.market.submitRfp"]);
	await new Promise((r) => setTimeout(r));
	assert.deepEqual(writes.map((w) => w.slice(0, 2)), [["put", OFFER], ["delete", OFFER]], "the offer is written to our repo, then deleted once answered");
	assert.match(writes[0][2], new RegExp(`^${BIDDER}\\.[2-7a-z]{13}$`));
	assert.equal(writes[0][3], "market");
	await assert.rejects(host.fetchers()["webrtc:"]({ url: at }), /needs the host's records signalling/);
	assert.throws(() => parseWebrtcUrl("webrtc://bm90LWpzb24/x"), /base64url JSON/);
	bidderHost.close();
});