johnandersen777.bsky.social/atlibs-iroh

Switch repository

Loading…

// node --test tests/loopback.test.mjs   (needs ../atlibs: `cargo wasm` and `cargo host`, or ATLIBS_BIN; no VM, no market, no account)
// The pre-commit iroh check: the real transport on this machine. A throwaway sshd
// (random 127.0.0.1 port, temp host key + authorized key, no system config) behind `dumbpipe listen-tcp`, and a
// TCP echo service behind a raw-ALPN listener; the echo module's `connect` (one-shot) and `session` (subscribe)
// entries through `atlibs run`, the Node CLI (atlibs javascript/src/server/cli.mjs with ATLIBS_TRANSPORTS=
// js/index.js) and `atlibs serve` POST /run, host keys pinned; and the iroh: fetcher through a relay to a local
// unix socket (js/fake-bin/relay.mjs standing in for `socialweb-guest-agent relay`). Skips (with the reason) when
// dumbpipe, sshd or ssh-keygen is missing. Full market rounds are only for requester / market flow changes.
import assert from "node:assert/strict";
import { spawn } from "node:child_process";
import { createServer as createHttpServer } from "node:http";
import { createServer } from "node:net";
import { mkdtemp, readFile, writeFile } from "node:fs/promises";
import { tmpdir, userInfo } from "node:os";
import { join } from "node:path";
import { after, before, describe, test } from "node:test";

import { hostBin } from "../../atlibs/rust/host/conformance.mjs";
import { irohFetch } from "../js/index.js";
import { ROOT, cleanup, exec, missing, servers, spawnUntil, sshd, temps, which } from "../../atlibs/rust/host/loopback.mjs";

after(cleanup);

describe("iroh: dumbpipe + a throwaway sshd", async () => {
	const absent = await missing("dumbpipe", "sshd", "ssh", "ssh-keygen");
	const skip = absent.length ? `missing ${absent.join(", ")} (brew install dumbpipe; OpenSSH)` : false;
	let env, ssh, alpn, bidder, hostKeys, served, bin, socket;
	const stubs = [];
	after(() => stubs.forEach((s) => s.kill()));

	before(async () => {
		if (skip) return;
		const dir = await mkdtemp(join(tmpdir(), "atlibs-loopback-iroh-"));
		temps.push(dir);
		await exec("ssh-keygen", ["-q", "-t", "ed25519", "-N", "", "-C", "loopback", "-f", join(dir, "iroh_ed25519")]);
		await writeFile(join(dir, "authorized_keys"), await readFile(join(dir, "iroh_ed25519.pub")));
		const { port, hostKey } = await sshd(dir, "ed25519");
		// The test's stand-ins: ssh with root@guest -> <user>@guest, and the guest agent's relay -> js/fake-bin/relay.mjs
		// (stdin/stdout <-> a unix socket, run by the real sshd's shell, which parses the command's quoting) in front of
		// js/fake-bin/http-stub.mjs; every other argument (-F /dev/null, ProxyCommand dumbpipe connect, the pinned
		// known_hosts, StrictHostKeyChecking=yes) is what the hosts built.
		bin = join(dir, "bin");
		await exec("mkdir", [bin]);
		const relay = `${process.execPath} ${new URL("../js/fake-bin/relay.mjs", import.meta.url).pathname}`;
		await writeFile(join(bin, "ssh"), `#!/bin/sh\nfor a; do shift; [ "$a" = root@guest ] && a=${userInfo().username}@guest; case "$a" in "socialweb-guest-agent relay "*) a="${relay} \${a#socialweb-guest-agent relay }";; esac; set -- "$@" "$a"; done\nexec ${await which("ssh")} "$@"\n`, { mode: 0o755 });
		socket = join(dir, "shell.sock");
		const stub = spawn(process.execPath, [new URL("../js/fake-bin/http-stub.mjs", import.meta.url).pathname, socket], { stdio: ["ignore", "pipe", "inherit"] });
		stubs.push(stub);
		await new Promise((r) => stub.stdout.once("data", r));
		const echo = createServer((socket) => {
			socket.write("ready\n");
			socket.once("data", (line) => socket.end(`got ${line}`));
		}).listen(0, "127.0.0.1");
		servers.push(echo);
		await new Promise((r) => echo.once("listening", r));
		const ticket = /endpoint[a-z0-9]+(?=\s)/; // the whole ticket: stderr may arrive split mid-ticket
		const market = createHttpServer((req, res) => {
			let body = "";
			req.on("data", (d) => (body += d)).on("end", () => res.end(JSON.stringify({ method: req.method, url: req.url, auth: req.headers.authorization ?? null, body })));
		}).listen(0, "127.0.0.1");
		servers.push(market);
		await new Promise((r) => market.once("listening", r));
		[ssh, alpn, bidder] = await Promise.all([
			spawnUntil("dumbpipe", ["listen-tcp", "--host", `127.0.0.1:${port}`], ticket),
			spawnUntil("dumbpipe", ["listen-tcp", "--custom-alpn", "utf8:loopback/1", "--host", `127.0.0.1:${echo.address().port}`], ticket),
			spawnUntil("dumbpipe", ["listen-tcp", "--host", `127.0.0.1:${market.address().port}`], ticket),
		]).then((r) => r.map(({ match }) => match[0]));
		const [type, publicKey] = hostKey.split(" ");
		hostKeys = [{ type, publicKey: `${type} ${publicKey}` }]; // the vm.hostKeys form seen live
		env = { PATH: `${bin}:${process.env.PATH}`, HOME: process.env.HOME, ATLIBS_SERVER_STATE: join(dir, "state"), ATLIBS_IROH_KEY: join(dir, "iroh_ed25519"), ATLIBS_TRANSPORTS: new URL("../js/index.js", import.meta.url).pathname };
		const { match } = await spawnUntil(hostBin(), ["serve", "examples/rust/echo", "--port", "0"], /http:\/\/(127\.0\.0\.1:\d+)\//, { cwd: ROOT, env });
		served = `http://${match[1]}`;
	});

	// dumbpipe's own stderr lines (its ephemeral client key, iroh's log lines) reach a session's stderr
	// through ssh's ProxyCommand: different every run, so they are dropped before comparing.
	const quiet = (result) => (typeof result?.stderr === "string" ? { ...result, stderr: result.stderr.replace(/^(using secret key [0-9a-f]+|\x1b\[2m\d{4}-.*)\n/gm, "") } : result);
	// One input through the three hosts, one after another (parallel dumbpipe clients to one listener
	// hung or lost the connection, dumbpipe 0.40.1); resolves the run's result, the same from each.
	async function run(entry, input) {
		const json = JSON.stringify(input);
		const options = { cwd: ROOT, env, timeout: 30000 };
		const rust = JSON.parse((await exec(hostBin(), ["run", "examples/rust/echo", entry, json], options)).stdout);
		const js = JSON.parse((await exec(process.execPath, [join(ROOT, "javascript/src/server/cli.mjs"), "run", "target/wasm32-unknown-unknown/release/echo.wasm", "examples/rust/echo/module.json", entry, json], options)).stdout);
		const serve = await fetch(`${served}/run/${entry}`, { method: "POST", headers: { "content-type": "application/json" }, body: json, signal: AbortSignal.timeout(30000) }).then((r) => r.json());
		for (const [name, ran] of [["atlibs run", rust], ["cli.mjs", js], ["atlibs serve", serve]]) assert.equal(ran.error, undefined, `${name}: ${ran.error}`);
		assert.deepEqual(quiet(js.result), quiet(rust.result), "cli.mjs vs atlibs run");
		assert.deepEqual(quiet(serve.result), quiet(rust.result), "atlibs serve vs atlibs run");
		return quiet(rust.result);
	}

	test("one-shot ssh: stdin to the command, its output and exit code", { skip }, async () => {
		assert.deepEqual(await run("connect", { ticket: `iroh://${ssh}`, hostKeys, command: "cat; echo done", stdin: "secret\n" }), { output: "secret\ndone\n", exitCode: 0 });
		assert.deepEqual(await run("connect", { ticket: ssh, hostKeys, command: "exit 7" }), { output: "", exitCode: 7 });
	});

	test("one-shot raw alpn: send, then everything until the peer closes", { skip }, async () => {
		assert.deepEqual(await run("connect", { ticket: alpn, alpn: "loopback/1", send: "hello\n" }), { output: "ready\ngot hello\n" });
	});

	// No raw-alpn session here: real `dumbpipe connect` (0.40.1) exits on stdin EOF only, not when the
	// peer closes, so a session's module must cancel it (fake-bin's dumbpipe exits; conformance.test.mjs).
	test("ssh sessions: output as it arrives, send to stdin, the exit code", { skip }, async () => {
		assert.deepEqual(await run("session", { ticket: ssh, hostKeys, command: "echo one; echo two >&2; exit 3" }), { stdout: "one\n", stderr: "two\n", exitCode: 3, error: null });
		assert.deepEqual(await run("session", { ticket: ssh, hostKeys, command: "echo ready; read line; echo got $line", send: "hello\n" }), { stdout: "ready\ngot hello\n", stderr: "", exitCode: 0, error: null });
	});

	test("a host key that is not the pinned one fails as ssh's own failure", { skip }, async () => {
		const other = [{ type: "ssh-ed25519", publicKey: "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIMTlQfhHkM2ZD3wPKn0WSxVbbhgWMvVnLsVqH4NM8x4G" }];
		const args = ["run", "examples/rust/echo", "connect", JSON.stringify({ ticket: ssh, hostKeys: other, command: "true" })];
		const failed = await exec(hostBin(), args, { cwd: ROOT, env }).then(() => assert.fail("a run error exits 1"), (e) => e);
		const ran = JSON.parse(failed.stdout); // the run is printed either way
		assert.match(ran.error, /ssh failed: .*Host key verification failed/s);
	});

	test("iroh: fetch: HTTP over real dumbpipe, ssh and a relay to a unix socket, the relay kept alive", { skip }, async () => {
		const fetch = irohFetch({ key: env.ATLIBS_IROH_KEY, spawn: (cmd, args, options) => spawn(join(bin, cmd), args, options) });
		const url = (path) => `iroh:///${Buffer.from(JSON.stringify({ ticket: `iroh://${ssh}`, hostKeys })).toString("base64url")}/${encodeURIComponent(socket)}${path}`;
		try {
			const started = performance.now();
			const first = await fetch({ method: "POST", url: url("/mcp"), headers: { "content-type": "application/json" }, body: '{"jsonrpc":"2.0","id":1,"method":"initialize"}' });
			console.log(`iroh fetch (dial + ssh + relay + first request): ${(performance.now() - started).toFixed(0)} ms`);
			assert.equal(first.status, 200);
			assert.deepEqual([JSON.parse(first.body).body, JSON.parse(first.body).n], ['{"jsonrpc":"2.0","id":1,"method":"initialize"}', 1]);
			const chunked = JSON.parse((await fetch({ url: url("/again?chunked=1") })).body);
			assert.deepEqual([chunked.url, chunked.n, chunked.conns], ["/again?chunked=1", 2, 1], "the same relay, kept alive");
		} finally {
			fetch.close();
		}
	});

	test("iroh: fetch to a market endpoint: HTTP/1.1 over raw dumbpipe connect <ticket> (no ssh, no key), kept alive", { skip }, async () => {
		const fetch = irohFetch({ key: "/nonexistent" });
		try {
			const sent = { method: "POST", url: `iroh://${bidder}/xrpc/com.publicdomainrelay.temp.market.submitRfp`, headers: { authorization: "Bearer jwt1", "content-type": "application/json" }, body: '{"a":1}' };
			const first = await fetch(sent);
			assert.deepEqual([first.status, JSON.parse(first.body)], [200, { method: "POST", url: "/xrpc/com.publicdomainrelay.temp.market.submitRfp", auth: "Bearer jwt1", body: '{"a":1}' }]);
			assert.equal(JSON.parse((await fetch({ url: `iroh://${bidder}/xrpc/a.b.c?x=1` })).body).url, "/xrpc/a.b.c?x=1");
		} finally {
			fetch.close();
		}
	});
});