johnandersen777.bsky.social/atlibs-iroh
iroh transport for atlibs hosts: com.fedfork.exec.iroh.connect in Rust (atlibs-host) and JavaScript
0
0
// The com.fedfork.exec.iroh.connect transport for Node hosts. No deps; shells out to dumbpipe
// and ssh. A host passes it in (atlibs javascript/src/server/cli.mjs: ATLIBS_TRANSPORTS=<this repo>/js/index.js,
// or main's transports; hostCall's calls); no core path imports it.
import { spawn } from "node:child_process";
import { randomBytes } from "node:crypto";
import { access, mkdtemp, rm, writeFile } from "node:fs/promises";
import { homedir, tmpdir } from "node:os";
import { join } from "node:path";
export const IROH_NSID = "com.fedfork.exec.iroh.connect";
const EXCHANGE_MS = 30000;
// stderr is kept only for an error message: its last STDERR_MAX bytes, however long the process runs.
const STDERR_MAX = 16384;
const tail = (text) => (text.length > STDERR_MAX ? text.slice(-STDERR_MAX) : text);
// A one-shot exchange keeps at most OUTPUT_MAX bytes of stdout (half a module's 1 MiB input, room for the
// event around it): past it the process is killed and the result says truncated.
export const OUTPUT_MAX = 1 << 19;
// dumbpipe without IROH_SECRET makes a key and prints "using secret key <hex>" to stderr, which would
// reach the module. A fresh key per process (ssh's ProxyCommand inherits the env) keeps it quiet.
// The rest of the environment stays out (ATPROTO_PASSWORD, SSH_AUTH_SOCK, ...): PATH, HOME and LANG only.
const spawnOptions = () => {
const env = Object.fromEntries(["PATH", "HOME", "LANG"].filter((k) => process.env[k] !== undefined).map((k) => [k, process.env[k]]));
return { stdio: ["pipe", "pipe", "pipe"], env: { ...env, IROH_SECRET: randomBytes(32).toString("hex") } };
};
// vm.onNetwork address -> dumbpipe ticket (swck requester.ValidateTicket).
export function ticket(address) {
const t = String(address ?? "").trim().replace(/^iroh:\/\//, "");
if (!/^[a-z0-9]{1,4096}$/.test(t)) throw new Error(`not an iroh ticket: ${address}`);
return t;
}
// vm.hostKeys keys -> known_hosts lines. hat-7 sends publicKey as bare base64 (no type prefix),
// though the lexicon says authorized_keys form; take both, and plain authorized_keys strings.
export function knownHosts(alias, hostKeys) {
const lines = (hostKeys ?? []).map((k) => {
const f = (typeof k === "string" ? k : k.publicKey).trim().split(/\s+/);
return (f.length > 1 ? f.slice(0, 2) : [k.type, f[0]]).join(" ");
});
if (!lines.length) throw new Error("no host keys: refusing to trust an unpinned guest");
return lines.map((l) => `${alias} ${l}\n`).join("");
}
// ssh to root@guest through dumbpipe, the host key pinned by a known_hosts file. "--" ends option
// parsing: ssh re-parses options after the host otherwise, so a command like "-oProxyCommand=..." would run locally.
// -F /dev/null: no ~/.ssh/config or system ssh_config (ForwardAgent, RemoteForward, LocalCommand, ControlMaster)
// toward a guest the bidder controls; only these options.
export const sshArgs = (key, address, knownHostsFile, command) => [
"-F", "/dev/null", "-i", key, "-o", "IdentitiesOnly=yes", "-o", "BatchMode=yes",
"-o", `ProxyCommand=dumbpipe connect ${ticket(address)}`,
"-o", "HostKeyAlias=guest", "-o", `UserKnownHostsFile=${knownHostsFile}`,
"-o", "StrictHostKeyChecking=yes", "-o", "ConnectTimeout=60",
"--", "root@guest", ...command,
];
// Runs cmd with input on stdin; resolves {stdout, stderr, code, truncated?} (code null: killed after ms, or
// past OUTPUT_MAX bytes of stdout, truncated). Exported for tests.
export function exchange(cmd, args, input = "", ms = EXCHANGE_MS) {
return new Promise((resolve, reject) => {
const child = spawn(cmd, args, spawnOptions());
const chunks = [];
let size = 0, stderr = "", truncated = false;
child.stdout.on("data", (d) => {
if (truncated) return;
chunks.push(d);
size += d.length;
if (size <= OUTPUT_MAX) return;
truncated = true;
// our ends of the pipes too: a grandchild still holding them (ssh's ProxyCommand) never delays close
child.stdout.destroy();
child.stderr.destroy();
child.kill("SIGKILL");
});
child.stderr.on("data", (d) => (stderr = tail(stderr + d)));
const timer = setTimeout(() => child.kill(), ms);
child.on("error", (e) => (clearTimeout(timer), reject(e)));
child.on("close", (code) => {
clearTimeout(timer);
resolve({ stdout: Buffer.concat(chunks).subarray(0, OUTPUT_MAX).toString(), stderr, code, ...(truncated && { truncated }) });
});
child.stdin.on("error", () => {}); // the peer may close before reading everything
child.stdin.end(input);
});
}
// The com.fedfork.exec.iroh.connect host pseudo-call (lexicons/com/fedfork/exec/iroh/connect.json),
// for a Node host's call(): if (nsid === IROH_NSID) return iroh(input). key: the ssh private key
// the host holds for guests (its .pub goes in user_data). run: exchange, swappable for tests.
export function irohHost({ key, run = exchange }) {
return async ({ ticket: address, alpn, hostKeys, command, stdin, send }) => {
const t = ticket(address);
if (alpn !== undefined) {
if (typeof alpn !== "string" || !/^[\x21-\x7e]{1,64}$/.test(alpn)) throw new Error("alpn must be 1 to 64 printable ASCII bytes");
const { stdout, stderr, code, truncated } = await run("dumbpipe", ["connect", "--custom-alpn", `utf8:${alpn}`, t], send ?? "");
if (code !== 0 && code !== null && !stdout) throw new Error(`dumbpipe exited ${code}: ${stderr.trim()}`);
// null: killed at the timeout (a peer that never closes) or at OUTPUT_MAX; what it sent so far, flagged (as the Rust host).
return { output: stdout, ...(truncated ? { truncated } : code === null && { timedOut: true }) };
}
if (typeof command !== "string" || !command) throw new Error("command is required for ssh");
// Secrets go over stdin, never in command: ps shows the command on the host (ssh's argv) and the guest (the shell's).
if (stdin !== undefined && !(typeof stdin === "string" && stdin.length <= 65536)) throw new Error("stdin must be a string of at most 65536 bytes");
const dir = await mkdtemp(join(tmpdir(), "iroh-ssh-"));
try {
await writeFile(join(dir, "known_hosts"), knownHosts("guest", hostKeys));
const { stdout, stderr, code, truncated } = await run("ssh", sshArgs(key, t, join(dir, "known_hosts"), [command]), stdin ?? "", 120000);
if (truncated) return { output: stdout, truncated };
// 255 is ssh's own failure (host key mismatch, auth, dial); null is the timeout.
if (code === 255 || code === null) throw new Error(`ssh ${code === null ? "timed out" : "failed"}: ${stderr.trim()}`);
return { output: stdout, exitCode: code };
} finally {
await rm(dir, { recursive: true, force: true });
}
};
}
// The connect input -> [cmd, args, cleanup]: dumbpipe for a raw ALPN, else ssh through dumbpipe with
// the host keys pinned (command optional for a session: none is a remote shell reading stdin).
async function sessionCommand(key, { ticket: address, alpn, hostKeys, command }) {
const t = ticket(address);
if (alpn !== undefined) {
if (typeof alpn !== "string" || !/^[\x21-\x7e]{1,64}$/.test(alpn)) throw new Error("alpn must be 1 to 64 printable ASCII bytes");
return ["dumbpipe", ["connect", "--custom-alpn", `utf8:${alpn}`, t], async () => {}];
}
if (command !== undefined && (typeof command !== "string" || !command)) throw new Error("command must be a non-empty string");
const dir = await mkdtemp(join(tmpdir(), "iroh-ssh-"));
try {
await writeFile(join(dir, "known_hosts"), knownHosts("guest", hostKeys));
} catch (error) {
await rm(dir, { recursive: true, force: true });
throw error;
}
return ["ssh", sshArgs(key, t, join(dir, "known_hosts"), command === undefined ? [] : [command]), () => rm(dir, { recursive: true, force: true })];
}
/**
* com.fedfork.exec.iroh.connect with subscribe: true: the same input as the
* one-shot call, but the process stays up. Each stdout / stderr chunk is a message { stdout } / { stderr }
* (UTF-8 text; chunk boundaries are wherever the pipe split them), the exit a last message { exitCode },
* then the stream ends. It fails instead (closed with the error) when ssh itself fails (255) or dumbpipe
* exits non-zero having printed nothing. send(data) writes stdin (a string as is, anything else as JSON);
* close() and the request signal kill the process. src/lib.rs's iroh_session does the same.
* @param {{ key: string, spawn?: typeof spawn }} options
*/
export function irohSession({ key, spawn: spawnImpl = spawn }) {
return async (input, { signal } = {}) => {
const [cmd, args, cleanup] = await sessionCommand(key, input ?? {});
const child = spawnImpl(cmd, args, spawnOptions());
const queue = [];
let wake;
let finished = false;
let failure;
let printed = false;
let killed = false;
let stderr = "";
const push = (message) => (queue.push(message), wake?.());
const decoders = { stdout: new TextDecoder(), stderr: new TextDecoder() };
for (const name of ["stdout", "stderr"]) {
child[name].on("data", (chunk) => {
const text = decoders[name].decode(chunk, { stream: true });
if (name === "stdout") printed = true;
else stderr = tail(stderr + text);
if (text) push({ [name]: text });
});
}
const end = () => void cleanup().catch(() => {}).finally(() => ((finished = true), wake?.()));
child.on("error", (error) => ((failure ??= new Error(`spawn ${cmd} ${error.message}`)), end()));
child.on("close", (code) => {
if (killed || failure) return end(); // failure: spawn failed (Node closes with -2 after error)
if (cmd === "ssh" && code === 255) failure ??= new Error(`ssh failed: ${stderr.trim()}`);
else if (cmd === "dumbpipe" && code !== 0 && code !== null && !printed) failure ??= new Error(`dumbpipe exited ${code}: ${stderr.trim()}`);
else push({ exitCode: code });
end();
});
child.stdin.on("error", () => {}); // the peer may close first
const close = () => {
if (finished || killed) return;
killed = true;
child.kill(); // SIGTERM, so ssh takes its dumbpipe ProxyCommand down; SIGKILL if it lingers
setTimeout(() => finished || child.kill("SIGKILL"), 2000).unref();
};
signal?.addEventListener("abort", close, { once: true });
if (signal?.aborted) close();
return {
close,
send: (data) =>
new Promise((resolve, reject) => {
if (finished || killed || !child.stdin.writable) return reject(new Error(`${IROH_NSID} session is closed`));
child.stdin.write(typeof data === "string" ? data : JSON.stringify(data), (error) => (error ? reject(error) : resolve()));
}),
async *[Symbol.asyncIterator]() {
for (;;) {
while (!queue.length && !finished) await new Promise((resolve) => (wake = resolve));
if (queue.length) yield queue.shift();
else if (failure) throw failure;
else return;
}
},
};
};
}
// The iroh: scheme fetcher (vm-mcp contract v2.2 §3): iroh:///<ref>/<socket>/<path...>, ref base64url JSON
// {ticket, hostKeys}; socket "shell" (SHELL_SOCKET) or a percent-encoded absolute unix path. HTTP/1.1 over
// `ssh root@guest socialweb-guest-agent relay <socket>` (stdin/stdout <-> the guest's unix socket) through dumbpipe.
export const SHELL_SOCKET = "/run/socialweb-guest-agent/shell.sock";
// The guest's login shell parses ssh's command: the path single-quoted.
export const relayCommand = (path) => `socialweb-guest-agent relay '${path.replaceAll("'", "'\\''")}'`;
// One request's cap (a guest exec runs up to 600 s); a response's head and body caps.
const FETCH_MS = 660000;
const HEAD_MAX = 1 << 16;
export const BODY_MAX = 1 << 24;
const TOKEN = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/;
/** iroh://<ticket>/<path...> (a market endpoint: HTTP/1.1 over `dumbpipe connect <ticket>`, no ssh) -> { ticket, raw, target }; else undefined. */
export function rawUrl(url) {
const m = /^iroh:\/\/(endpoint[a-z2-7]{1,4088})(\/[^#]*)?(#.*)?$/.exec(String(url));
if (!m) return undefined;
const target = m[2] || "/";
if (/[\0-\x20\x7f]/.test(target)) throw new Error(`${url}: the path has spaces or control characters`);
return { ticket: m[1], raw: true, target };
}
/** iroh:///<ref>/<socket>/<path...> -> { ticket, hostKeys, socket, target } (target: "/" + path, query kept). */
export function irohUrl(url) {
const m = /^iroh:\/\/\/?([A-Za-z0-9_-]+={0,2})\/([^/?#]+)([^#]*)/.exec(String(url));
if (!m) throw new Error(`not an iroh:///<ref>/<socket>/<path> URL: ${url}`);
let ref, socket;
try {
ref = JSON.parse(Buffer.from(m[1], "base64url").toString());
} catch {
throw new Error(`${url}: the ref is not base64url JSON`);
}
try {
socket = m[2] === "shell" ? SHELL_SOCKET : decodeURIComponent(m[2]);
} catch {
throw new Error(`${url}: bad percent-encoding in the socket`);
}
if (!socket.startsWith("/") || /[\0-\x1f\x7f]/.test(socket)) throw new Error(`${url}: the socket must be "shell" or an absolute unix path`);
const target = m[3].startsWith("/") ? m[3] : `/${m[3]}`;
if (/[\0-\x20\x7f]/.test(target)) throw new Error(`${url}: the path has spaces or control characters`);
return { ticket: ticket(ref?.ticket), hostKeys: ref?.hostKeys, socket, target };
}
// One relay: an HTTP/1.1 connection to the guest's socket, one request at a time (its caller holds it).
// Unref'd: an idle relay never keeps the host process up (a request's timer does).
async function relay(key, { ticket: t, hostKeys, socket, raw }, spawnImpl) {
const [cmd, args, cleanup] = raw ? ["dumbpipe", ["connect", t], async () => {}] : await sessionCommand(key, { ticket: t, hostKeys, command: relayCommand(socket) });
const child = spawnImpl(cmd, args, spawnOptions());
let buf = Buffer.alloc(0), more = [], size = 0, stderr = "", ended, wake;
const conn = { got: 0 };
child.stdout.on("data", (d) => (more.push(d), (size += d.length), (conn.got += d.length), wake?.()));
child.stderr.on("data", (d) => (stderr = tail(stderr + d)));
const end = (why) => {
if (ended) return;
ended = why;
void cleanup().catch(() => {});
if (child.exitCode === null) child.kill();
wake?.();
};
child.on("error", (error) => end(`spawn ${cmd} ${error.message}`));
child.on("close", (code) => end(`exit ${code}`));
child.stdin.on("error", () => {});
child.unref();
for (const pipe of [child.stdin, child.stdout, child.stderr]) pipe.unref?.();
const failure = () => new Error(`iroh: the relay ended (${ended})${stderr.trim() ? `: ${stderr.trim()}` : ""}`);
const wait = () => (ended ? Promise.reject(failure()) : new Promise((resolve) => (wake = resolve)));
const merge = () => more.length && ((buf = Buffer.concat([buf, ...more])), (more = []), (size = 0));
const line = async () => {
for (;;) {
merge();
const at = buf.indexOf("\r\n");
if (at >= 0) {
const text = buf.subarray(0, at).toString("latin1");
buf = buf.subarray(at + 2);
return text;
}
if (buf.length > HEAD_MAX) throw new Error(`iroh: a response line over ${HEAD_MAX} bytes`);
await wait();
}
};
const take = async (n) => {
while (buf.length + size < n) await wait();
merge();
const bytes = Buffer.from(buf.subarray(0, n));
buf = buf.subarray(n);
return bytes;
};
const rest = async () => {
while (!ended) {
if (buf.length + size > BODY_MAX) throw new Error(`iroh: a response body over ${BODY_MAX} bytes`);
await new Promise((resolve) => (wake = resolve));
}
merge();
if (buf.length > BODY_MAX) throw new Error(`iroh: a response body over ${BODY_MAX} bytes`);
return buf;
};
const head = async () => {
const status = /^HTTP\/1\.([01]) (\d{3})/.exec(await line());
if (!status) throw new Error("iroh: not an HTTP/1.x response");
const fields = [];
for (let bytes = 0, l; (l = await line()); ) {
if ((bytes += l.length) > HEAD_MAX) throw new Error(`iroh: response headers over ${HEAD_MAX} bytes`);
const colon = l.indexOf(":");
if (colon > 0) fields.push([l.slice(0, colon).trim().toLowerCase(), l.slice(colon + 1).trim()]);
}
return { minor: status[1], status: Number(status[2]), fields };
};
return Object.assign(conn, {
get ended() {
return ended;
},
close: (why = "closed") => end(why),
// -> { response: {status, headers, body}, keep }: keep, the connection can take another request
async request(method, target, headers, body) {
conn.got = 0;
const lines = [`${method} ${target} HTTP/1.1`, "host: guest", ...headers.map(([k, v]) => `${k}: ${v}`), `content-length: ${body.length}`];
child.stdin.write(Buffer.concat([Buffer.from(`${lines.join("\r\n")}\r\n\r\n`), body]));
let h;
do h = await head();
while (h.status >= 100 && h.status < 200);
const field = (name) => h.fields.filter(([k]) => k === name).map(([, v]) => v).join(", ");
let data, keep = h.minor === "1" && !/\bclose\b/i.test(field("connection"));
if (method === "HEAD" || h.status === 204 || h.status === 304) data = Buffer.alloc(0);
else if (/\bchunked\b/i.test(field("transfer-encoding"))) {
const parts = [];
for (let total = 0; ; ) {
const size = (await line()).split(";")[0].trim();
const n = /^[0-9a-fA-F]{1,8}$/.test(size) ? parseInt(size, 16) : NaN;
if (!(n >= 0) || (total += n) > BODY_MAX) throw new Error(`iroh: a bad chunk size, or a response body over ${BODY_MAX} bytes`);
if (n === 0) break;
parts.push(await take(n));
await line();
}
while (await line()); // trailers
data = Buffer.concat(parts);
} else if (field("content-length")) {
const n = Number(field("content-length").split(",")[0]);
if (!Number.isSafeInteger(n) || n < 0 || n > BODY_MAX) throw new Error(`iroh: a bad content-length, or a response body over ${BODY_MAX} bytes`);
data = await take(n);
} else {
keep = false;
data = Buffer.from(await rest());
}
const out = {};
for (const [k, v] of h.fields) out[k] = k in out ? `${out[k]}, ${v}` : v;
return { response: { status: h.status, headers: out, body: data.toString() }, keep };
},
});
}
/**
* The iroh: fetcher: ({method, url, headers, body, signal}) -> {status, headers (lowercase names), body (text)}.
* Relays per (ref, socket) are reused (HTTP keep-alive), one request in flight on each (more are opened for
* concurrent requests); a request whose relay failed before any response byte is retried once on a fresh
* one. fetch.close() ends the idle relays.
* @param {{ key: string, spawn?: typeof spawn }} options
*/
export function irohFetch({ key, spawn: spawnImpl = spawn }) {
const idle = new Map();
const fetch = async ({ method = "GET", url, headers = {}, body, signal } = {}) => {
if (!TOKEN.test(method)) throw new Error(`iroh: bad method ${method}`);
const where = rawUrl(url) ?? irohUrl(url);
const sent = [];
for (const [k, v] of headers instanceof Headers ? headers : Object.entries(headers ?? {})) {
if (!TOKEN.test(k) || typeof v !== "string" || /[\r\n\0]/.test(v)) throw new Error(`iroh: bad header ${k}`);
if (!["host", "content-length", "transfer-encoding", "connection"].includes(k.toLowerCase())) sent.push([k, v]);
}
const bytes = body == null ? Buffer.alloc(0) : Buffer.from(body);
if (!where.raw) await access(key).catch(() => Promise.reject(new Error(`iroh: needs the host's ssh private key: set ATLIBS_IROH_KEY or put it at ${key}`)));
const name = where.raw ? `raw ${where.ticket}` : `${where.ticket} ${JSON.stringify(where.hostKeys ?? [])} ${where.socket}`;
const attempt = async (fresh) => {
signal?.throwIfAborted();
const pool = idle.get(name) ?? [];
let conn = fresh ? undefined : pool.pop();
while (conn?.ended) conn = pool.pop();
conn ??= await relay(key, where, spawnImpl);
const timer = setTimeout(() => conn.close(`no response in ${FETCH_MS / 1000} s`), FETCH_MS);
const abort = () => conn.close("aborted");
signal?.addEventListener("abort", abort, { once: true });
try {
const { response, keep } = await conn.request(method, where.target, sent, bytes);
if (keep && !conn.ended) idle.set(name, [...(idle.get(name) ?? []), conn]);
else conn.close();
return response;
} catch (error) {
conn.close();
signal?.throwIfAborted();
throw Object.assign(error, { retry: !fresh && conn.got === 0 });
} finally {
clearTimeout(timer);
signal?.removeEventListener("abort", abort);
}
};
try {
return await attempt(false);
} catch (error) {
if (!error.retry) throw error;
return attempt(true);
}
};
fetch.close = () => {
for (const pool of idle.values()) for (const conn of pool) conn.close();
idle.clear();
};
return fetch;
}
// The transport: com.fedfork.exec.iroh.connect with the host's ssh private key (a path; the key never
// enters wasm), for hostCall's calls. A request is irohHost, subscribe: true an irohSession (output
// streamed as messages, send writes stdin). run, spawn: swappable for tests.
export function irohCall({ key, run, spawn: spawnImpl }) {
const iroh = irohHost({ key, ...(run && { run }) });
const session = irohSession({ key, ...(spawnImpl && { spawn: spawnImpl }) });
return async (request = {}) => {
const input = request.input ?? {};
if (input.alpn === undefined) await access(key).catch(() => Promise.reject(new Error(`${IROH_NSID} ssh needs the host's ssh private key: set ATLIBS_IROH_KEY or put it at ${key}`)));
return request.subscribe ? session(input, request) : iroh(input);
};
}
// What a host loads (javascript/src/server/cli.mjs ATLIBS_TRANSPORTS, or main's transports): connect, and the
// iroh: fetcher (runModule's fetchers); the key ATLIBS_IROH_KEY, else <state>/iroh_ed25519.
const keyOf = ({ env = process.env, stateDir = env.ATLIBS_SERVER_STATE || join(homedir(), ".atlibs-server") } = {}) => env.ATLIBS_IROH_KEY || join(stateDir, "iroh_ed25519");
export const transport = { [IROH_NSID]: (host) => irohCall({ key: keyOf(host) }) };
export const fetchers = (host) => ({ "iroh:": irohFetch({ key: keyOf(host) }) });