johnandersen777.bsky.social/atlibs-iroh
iroh transport for atlibs hosts: com.fedfork.exec.iroh.connect in Rust (atlibs-host) and JavaScript
0
0
//! atlibs-iroh: com.fedfork.exec.iroh.connect and the iroh: scheme fetcher (HTTP/1.1 to a guest's unix
//! socket over `ssh root@guest socialweb-guest-agent relay <socket>`) for atlibs hosts, on atlibs-host's
//! transport seam (atlibs_host::transport) and nothing else of it. The atlibs CLI registers it behind its
//! cargo feature `iroh` (default on).
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use atlibs_host::host::Response;
use atlibs_host::run::{Call, Request, Sub, SubCmd};
use atlibs_host::transport::Transport;
use atlibs_host::{BoxFut, Signal, js_string};
use serde_json::{Value, json};
pub const NSID: &str = "com.fedfork.exec.iroh.connect";
/// The URL schemes (no colon) the fetcher serves: Call::schemes (js/index.js fetchers' "iroh:").
pub const SCHEMES: &[&str] = &["iroh"];
/// The guest agent's shell service socket, the URL's socket "shell" (vm-mcp contract v2.2 §7).
pub const SHELL_SOCKET: &str = "/run/socialweb-guest-agent/shell.sock";
// One request's cap (a guest exec runs up to 600 s); a response's head and body caps.
const FETCH_MS: u64 = 660_000;
const HEAD_MAX: usize = 1 << 16;
pub const BODY_MAX: usize = 1 << 24;
/// The transport: com.fedfork.exec.iroh.connect and the iroh: fetch with the host's ssh private key (a
/// path; the key never enters wasm), shelling out to dumbpipe and ssh as js/index.js irohCall and irohFetch
/// do. A connect request is a one-shot exchange (iroh), subscribe a long-lived session (iroh_session); a
/// fetch goes over an idle Relay of its (ticket, hostKeys, socket), or a new one.
pub struct Iroh {
pub key: PathBuf,
// the ssh binary ("ssh"; tests: js/fake-bin/ssh)
ssh: String,
// idle relays by "<ticket> <hostKeys> <socket>"
relays: Mutex<HashMap<String, Vec<Relay>>>,
}
impl Iroh {
pub fn new(key: PathBuf) -> Self {
Self { key, ssh: "ssh".into(), relays: Default::default() }
}
fn input(&self, nsid: &str, input: Option<Value>) -> Result<Value, String> {
if nsid != NSID {
return Err(format!("{nsid} is not available on this host"));
}
let input = input.unwrap_or_else(|| json!({}));
if input.get("alpn").is_none() && std::fs::metadata(&self.key).is_err() {
return Err(format!("{nsid} ssh needs the host's ssh private key: set ATLIBS_IROH_KEY or put it at {}", self.key.display()));
}
Ok(input)
}
/// iroh:///<ref>/<socket>/<path...> (js/index.js irohFetch): relays reused per (ref, socket) with
/// HTTP keep-alive, one request in flight on each; a request whose relay failed before any response
/// byte is retried once on a fresh one. Response headers: lowercase names, in order.
pub async fn http(&self, method: &str, url: &str, headers: Vec<(String, String)>, body: Vec<u8>) -> Result<Response, String> {
if !token(method) {
return Err(format!("iroh: bad method {method}"));
}
let to = match raw_url(url) {
Some(to) => to?,
None => iroh_url(url)?,
};
let mut sent = vec![];
for (k, v) in headers {
if !token(&k) || v.contains(['\r', '\n', '\0']) {
return Err(format!("iroh: bad header {k}"));
}
if !["host", "content-length", "transfer-encoding", "connection"].iter().any(|h| h.eq_ignore_ascii_case(&k)) {
sent.push((k, v));
}
}
if !to.raw && std::fs::metadata(&self.key).is_err() {
return Err(format!("iroh: needs the host's ssh private key: set ATLIBS_IROH_KEY or put it at {}", self.key.display()));
}
let name = if to.raw { format!("raw {}", to.ticket) } else { format!("{} {} {}", to.ticket, to.host_keys, to.socket) };
let mut fresh = false;
loop {
let idle = if fresh { None } else { self.idle(&name) };
let mut relay = match idle {
Some(relay) => relay,
None => Relay::open(&self.ssh, &self.key, &to)?,
};
match tokio::time::timeout(Duration::from_millis(FETCH_MS), relay.request(method, &to.target, &sent, &body)).await {
Err(_) => return Err(format!("iroh: no response in {} s", FETCH_MS / 1000)),
Ok(Ok((response, keep))) => {
if keep {
self.relays.lock().unwrap().entry(name).or_default().push(relay);
}
return Ok(response);
}
Ok(Err(why)) => {
let answered = relay.got;
let why = relay.failed(why).await;
if fresh || answered {
return Err(why);
}
fresh = true;
}
}
}
}
// An idle relay of name still running.
fn idle(&self, name: &str) -> Option<Relay> {
let mut relays = self.relays.lock().unwrap();
let pool = relays.get_mut(name)?;
while let Some(mut relay) = pool.pop() {
if matches!(relay.child.try_wait(), Ok(None)) {
return Some(relay);
}
}
None
}
}
impl Call for Iroh {
fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
Box::pin(async move {
let input = self.input(&nsid, request.input)?;
iroh(&self.key, &input).await.map(Some)
})
}
fn subscribe(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Sub, String>> {
Box::pin(async move {
let input = self.input(&nsid, request.input)?;
iroh_session(&self.key, &input, request.signal).await
})
}
/// The iroh: scheme fetch (vm-mcp contract v2.2 §3): com.fedfork.exec.http.fetch to iroh:/// URLs.
fn schemes(&self) -> &'static [&'static str] {
SCHEMES
}
fn fetch(&self, method: &str, url: &str, headers: Vec<(String, String)>, body: Option<Vec<u8>>) -> BoxFut<'_, Result<Response, String>> {
let (method, url) = (method.to_string(), url.to_string());
Box::pin(async move { self.http(&method, &url, headers, body.unwrap_or_default()).await })
}
}
/// The transport a host registers (atlibs_host::transport): NSID, and iroh: URLs for http.fetch
/// (transport::fetchers), served with the ssh key at `key`.
pub fn transport(key: PathBuf) -> Transport {
Transport { provides: &[NSID], call: Arc::new(Iroh::new(key)) }
}
fn token(s: &str) -> bool {
!s.is_empty() && s.chars().all(|c| c.is_ascii_alphanumeric() || "!#$%&'*+.^_`|~-".contains(c))
}
/// Where an iroh: URL goes: the guest (ticket, pinned host keys), its unix socket, the HTTP request target.
#[derive(Debug, PartialEq)]
pub struct Target {
pub ticket: String,
pub host_keys: Value,
pub socket: String,
pub target: String,
/// iroh://<ticket>/<path>: HTTP/1.1 over `dumbpipe connect <ticket>` itself (a market endpoint), no ssh.
pub raw: bool,
}
/// iroh://<ticket>/<path...> (js/index.js rawUrl): a market endpoint's short ticket; None for any other URL.
pub fn raw_url(url: &str) -> Option<Result<Target, String>> {
let rest = url.strip_prefix("iroh://")?.split('#').next().unwrap_or_default();
let (t, path) = rest.split_at(rest.find('/').unwrap_or(rest.len()));
let body = t.strip_prefix("endpoint")?;
if body.is_empty() || t.len() > 4096 || !body.bytes().all(|b| b.is_ascii_lowercase() || (b'2'..=b'7').contains(&b)) {
return None;
}
let target = if path.is_empty() { "/".to_string() } else { path.to_string() };
if target.bytes().any(|b| b <= 0x20 || b == 0x7f) {
return Some(Err(format!("{url}: the path has spaces or control characters")));
}
Some(Ok(Target { ticket: t.to_string(), host_keys: json!([]), socket: String::new(), target, raw: true }))
}
/// iroh:///<ref>/<socket>/<path...>: ref base64url JSON {ticket, hostKeys}; socket "shell" (SHELL_SOCKET) or a
/// percent-encoded absolute unix path; target "/" + path, query kept (js/index.js irohUrl).
pub fn iroh_url(url: &str) -> Result<Target, String> {
use base64::Engine;
let bad = || format!("not an iroh:///<ref>/<socket>/<path> URL: {url}");
let rest = url.strip_prefix("iroh://").ok_or_else(bad)?;
let rest = rest.strip_prefix('/').unwrap_or(rest).split('#').next().unwrap_or_default();
let (r, rest) = rest.split_once('/').ok_or_else(bad)?;
let (sock, target) = rest.split_at(rest.find(['/', '?']).unwrap_or(rest.len()));
let b64 = r.trim_end_matches('=');
if b64.is_empty() || r.len() - b64.len() > 2 || sock.is_empty() || !b64.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') {
return Err(bad());
}
let refs: Value = base64::engine::general_purpose::URL_SAFE_NO_PAD
.decode(b64)
.ok()
.and_then(|j| serde_json::from_slice(&j).ok())
.ok_or_else(|| format!("{url}: the ref is not base64url JSON"))?;
let socket = match sock {
"shell" => SHELL_SOCKET.to_string(),
s => percent_encoding::percent_decode_str(s).decode_utf8().map_err(|_| format!("{url}: bad percent-encoding in the socket"))?.into_owned(),
};
if !socket.starts_with('/') || socket.chars().any(|c| c.is_ascii_control()) {
return Err(format!("{url}: the socket must be \"shell\" or an absolute unix path"));
}
let target = if target.starts_with('/') { target.to_string() } else { format!("/{target}") };
if target.bytes().any(|b| b <= 0x20 || b == 0x7f) {
return Err(format!("{url}: the path has spaces or control characters"));
}
Ok(Target { ticket: ticket(refs.get("ticket"))?, host_keys: refs.get("hostKeys").cloned().unwrap_or(json!([])), socket, target, raw: false })
}
/// The guest's login shell parses ssh's command: the path single-quoted.
pub fn relay_command(path: &str) -> String {
format!("socialweb-guest-agent relay '{}'", path.replace('\'', "'\\''"))
}
// A read that hit the relay's end (stdout closed): failed() reports its exit instead.
const EOF: &str = "eof";
/// One `ssh root@guest socialweb-guest-agent relay <socket>` through dumbpipe (js/index.js relay): an
/// HTTP/1.1 connection to the guest's unix socket, one request at a time (its holder's). Dropped: ssh is
/// killed (kill_on_drop), and its temp dir removed.
struct Relay {
child: tokio::process::Child,
stdin: tokio::process::ChildStdin,
stdout: tokio::io::BufReader<tokio::process::ChildStdout>,
stderr: Arc<Mutex<String>>,
errors: tokio::task::JoinHandle<()>,
// a response byte arrived for the current request
got: bool,
_dir: Option<TempDir>,
}
fn io(e: std::io::Error) -> String {
if e.kind() == std::io::ErrorKind::UnexpectedEof { EOF.into() } else { e.to_string() }
}
impl Relay {
fn open(ssh: &str, key: &Path, to: &Target) -> Result<Self, String> {
use tokio::io::AsyncReadExt;
let (cmd, args, dir) = if to.raw {
("dumbpipe", vec!["connect".to_string(), to.ticket.clone()], None)
} else {
let (_, args, dir) = session_command(key, &json!({ "ticket": to.ticket, "hostKeys": to.host_keys, "command": relay_command(&to.socket) }))?;
(ssh, args, dir)
};
let mut child = iroh_command(cmd, &args).spawn().map_err(|e| format!("spawn {cmd} {e}"))?;
let (stdin, stdout, mut err) = (child.stdin.take().expect("piped"), child.stdout.take().expect("piped"), child.stderr.take().expect("piped"));
let stderr = Arc::new(Mutex::new(String::new()));
let tail_of = stderr.clone();
let errors = tokio::spawn(async move {
let (mut buf, mut carry) = (vec![0u8; 8192], Vec::new());
while let Ok(n @ 1..) = err.read(&mut buf).await {
let mut text = tail_of.lock().unwrap();
text.push_str(&decode_utf8(&mut carry, &buf[..n]));
tail(&mut text);
}
});
Ok(Self { child, stdin, stdout: tokio::io::BufReader::new(stdout), stderr, errors, got: false, _dir: dir })
}
// A CRLF-terminated line, without it.
async fn line(&mut self) -> Result<String, String> {
use tokio::io::{AsyncBufReadExt, AsyncReadExt};
let mut line = Vec::new();
(&mut self.stdout).take(HEAD_MAX as u64 + 2).read_until(b'\n', &mut line).await.map_err(io)?;
self.got |= !line.is_empty();
if !line.ends_with(b"\r\n") {
return Err(if line.len() > HEAD_MAX { format!("iroh: a response line over {HEAD_MAX} bytes") } else { EOF.into() });
}
line.truncate(line.len() - 2);
Ok(String::from_utf8_lossy(&line).into_owned())
}
async fn take(&mut self, n: usize) -> Result<Vec<u8>, String> {
use tokio::io::AsyncReadExt;
let mut bytes = vec![0u8; n];
self.stdout.read_exact(&mut bytes).await.map_err(io)?;
Ok(bytes)
}
// -> (response, keep): keep, the connection can take another request
async fn request(&mut self, method: &str, target: &str, headers: &[(String, String)], body: &[u8]) -> Result<(Response, bool), String> {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
self.got = false;
let mut head = format!("{method} {target} HTTP/1.1\r\nhost: guest\r\n");
for (k, v) in headers {
head.push_str(&format!("{k}: {v}\r\n"));
}
head.push_str(&format!("content-length: {}\r\n\r\n", body.len()));
let mut bytes = head.into_bytes();
bytes.extend_from_slice(body);
self.stdin.write_all(&bytes).await.map_err(io)?;
self.stdin.flush().await.map_err(io)?;
let (minor, status, fields) = loop {
let line = self.line().await?;
let status = line.strip_prefix("HTTP/1.").and_then(|s| Some((s.get(..1)?.to_string(), s.get(2..5)?.parse::<u16>().ok()?)));
let Some((minor, status)) = status.filter(|_| line.as_bytes().get(8) == Some(&b' ')) else {
return Err("iroh: not an HTTP/1.x response".into());
};
let (mut fields, mut size) = (vec![], 0);
loop {
let line = self.line().await?;
if line.is_empty() {
break;
}
size += line.len();
if size > HEAD_MAX {
return Err(format!("iroh: response headers over {HEAD_MAX} bytes"));
}
if let Some((k, v)) = line.split_once(':').filter(|(k, _)| !k.trim().is_empty()) {
fields.push((k.trim().to_ascii_lowercase(), v.trim().to_string()));
}
}
if !(100..200).contains(&status) {
break (minor, status, fields);
}
};
let field = |name: &str| fields.iter().filter(|(k, _)| k == name).map(|(_, v)| v.as_str()).collect::<Vec<_>>().join(", ");
let has = |name: &str, word: &str| field(name).split(',').any(|w| w.trim().eq_ignore_ascii_case(word));
let mut keep = minor == "1" && !has("connection", "close");
let body = if method == "HEAD" || status == 204 || status == 304 {
vec![]
} else if has("transfer-encoding", "chunked") {
let mut body = vec![];
loop {
let size = self.line().await?;
let size = size.split(';').next().unwrap_or_default().trim();
let n = Some(size).filter(|s| (1..=8).contains(&s.len())).and_then(|s| usize::from_str_radix(s, 16).ok());
let Some(n) = n.filter(|n| body.len() + n <= BODY_MAX) else {
return Err(format!("iroh: a bad chunk size, or a response body over {BODY_MAX} bytes"));
};
if n == 0 {
break;
}
body.extend(self.take(n).await?);
self.line().await?;
}
while !self.line().await?.is_empty() {} // trailers
body
} else if !field("content-length").is_empty() {
let n = field("content-length").split(',').next().unwrap_or_default().trim().parse::<usize>().ok().filter(|n| *n <= BODY_MAX);
self.take(n.ok_or_else(|| format!("iroh: a bad content-length, or a response body over {BODY_MAX} bytes"))?).await?
} else {
keep = false;
let mut body = vec![];
(&mut self.stdout).take(BODY_MAX as u64 + 1).read_to_end(&mut body).await.map_err(io)?;
if body.len() > BODY_MAX {
return Err(format!("iroh: a response body over {BODY_MAX} bytes"));
}
body
};
Ok((Response { status, headers: fields, body }, keep))
}
// The error for a failed request: the relay stopped, its exit (at its end) and stderr tail added.
async fn failed(mut self, why: String) -> String {
let why = match why.as_str() {
EOF => match tokio::time::timeout(Duration::from_secs(2), self.child.wait()).await {
Ok(Ok(status)) => format!("exit {}", status.code().map_or("by a signal".into(), |c| c.to_string())),
_ => "its output ended".into(),
},
_ => why,
};
stop(&mut self.child).await;
let _ = tokio::time::timeout(Duration::from_millis(500), &mut self.errors).await;
let stderr = self.stderr.lock().unwrap().trim().to_string();
format!("iroh: the relay ended ({why}){}", if stderr.is_empty() { String::new() } else { format!(": {stderr}") })
}
}
/// vm.onNetwork address -> dumbpipe ticket.
pub fn ticket(address: Option<&Value>) -> Result<String, String> {
let raw = match address {
None | Some(Value::Null) => String::new(),
Some(v) => js_string(Some(v)),
};
let t = raw.trim().trim_start_matches("iroh://").to_string();
if t.is_empty()
|| t.len() > 4096
|| !t
.bytes()
.all(|b| b.is_ascii_lowercase() || b.is_ascii_digit())
{
return Err(format!("not an iroh ticket: {}", js_string(address)));
}
Ok(t)
}
/// vm.hostKeys -> known_hosts lines (bare base64 with a type, or authorized_keys strings).
pub fn known_hosts(alias: &str, host_keys: Option<&Value>) -> Result<String, String> {
let lines: Vec<String> = host_keys
.and_then(Value::as_array)
.map(|keys| {
keys.iter()
.map(|k| {
let text = k
.as_str()
.map(String::from)
.unwrap_or_else(|| js_string(k.get("publicKey")));
let f: Vec<&str> = text.split_whitespace().collect();
if f.len() > 1 {
f[..2].join(" ")
} else {
format!("{} {}", js_string(k.get("type")), f.first().unwrap_or(&""))
}
})
.collect()
})
.unwrap_or_default();
if lines.is_empty() {
return Err("no host keys: refusing to trust an unpinned guest".into());
}
Ok(lines.iter().map(|l| format!("{alias} {l}\n")).collect())
}
// stderr is kept only for an error message: its last STDERR_MAX bytes, however long the process runs.
const STDERR_MAX: usize = 16384;
// 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.
pub const OUTPUT_MAX: usize = 1 << 19;
fn random_hex<const N: usize>() -> String {
let mut bytes = [0u8; N];
rand_core::RngCore::fill_bytes(&mut rand_core::OsRng, &mut bytes);
bytes.iter().map(|b| format!("{b:02x}")).collect()
}
fn tail(text: &mut String) {
if text.len() > STDERR_MAX {
let mut cut = text.len() - STDERR_MAX;
while !text.is_char_boundary(cut) {
cut += 1;
}
text.drain(..cut);
}
}
// 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.
fn iroh_command(cmd: &str, args: &[String]) -> tokio::process::Command {
let mut command = tokio::process::Command::new(cmd);
command.env_clear();
for key in ["PATH", "HOME", "LANG"] {
if let Some(value) = std::env::var_os(key) {
command.env(key, value);
}
}
command
.args(args)
.env("IROH_SECRET", random_hex::<32>())
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.kill_on_drop(true);
command
}
/// A temp dir removed when dropped: on every exit path, and when the runtime drops the task.
#[derive(Debug)]
struct TempDir(PathBuf);
impl TempDir {
fn new() -> Result<Self, String> {
let dir = std::env::temp_dir().join(format!("iroh-ssh-{}", random_hex::<16>()));
std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?;
Ok(Self(dir))
}
}
impl Drop for TempDir {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.0);
}
}
// SIGTERM (ssh then takes its dumbpipe ProxyCommand down with it), SIGKILL if still up after 2 s.
// ponytail: `kill -TERM` as a process, so no libc dependency for one signal.
async fn stop(child: &mut tokio::process::Child) {
if let Some(pid) = child.id() {
let _ = tokio::process::Command::new("kill").args(["-TERM", &pid.to_string()]).status().await;
if tokio::time::timeout(Duration::from_secs(2), child.wait()).await.is_ok() {
return;
}
}
let _ = child.start_kill();
let _ = child.wait().await;
}
/// What a one-shot exchange read: code None when it was killed (`timed_out` after ms, `truncated` at OUTPUT_MAX).
struct Exchanged {
stdout: String,
stderr: String,
code: Option<i32>,
timed_out: bool,
truncated: bool,
}
// Runs cmd with input on stdin; on timeout, or past OUTPUT_MAX bytes of stdout, the process is killed and
// what it printed (up to OUTPUT_MAX) is kept.
async fn exchange(cmd: &str, args: &[String], input: &str, ms: u64) -> Result<Exchanged, String> {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let mut child = iroh_command(cmd, args).spawn().map_err(|e| format!("spawn {cmd} {e}"))?;
let mut stdin = child.stdin.take().expect("piped");
let input = input.to_string();
tokio::spawn(async move {
let _ = stdin.write_all(input.as_bytes()).await; // the peer may close before reading everything
});
let (mut out_pipe, mut err_pipe) = (child.stdout.take().expect("piped"), child.stderr.take().expect("piped"));
let (mut stdout, mut stderr, mut truncated) = (Vec::new(), String::new(), false);
let read = async {
let (mut out_buf, mut err_buf, mut open) = (vec![0u8; 8192], vec![0u8; 8192], (true, true));
let mut carry = Vec::new();
while open.0 || open.1 {
tokio::select! {
n = out_pipe.read(&mut out_buf), if open.0 => match n {
Ok(n) if n > 0 => {
stdout.extend_from_slice(&out_buf[..n]);
if stdout.len() > OUTPUT_MAX {
stdout.truncate(OUTPUT_MAX);
truncated = true;
let _ = child.start_kill();
break;
}
}
_ => open.0 = false,
},
n = err_pipe.read(&mut err_buf), if open.1 => match n {
Ok(n) if n > 0 => {
stderr.push_str(&decode_utf8(&mut carry, &err_buf[..n]));
tail(&mut stderr);
}
_ => open.1 = false,
},
}
}
child.wait().await.map_err(|e| e.to_string())
};
let (code, timed_out) = match tokio::time::timeout(Duration::from_millis(ms), read).await {
Ok(status) => (status?.code(), false),
Err(_) => {
let _ = child.start_kill();
let _ = child.wait().await;
(None, true)
}
};
Ok(Exchanged { stdout: String::from_utf8_lossy(&stdout).into(), stderr, code, timed_out, truncated })
}
pub fn ssh_args(key: &Path, ticket: &str, known_hosts_file: &Path, command: &str) -> Vec<String> {
[
// no ~/.ssh/config or system ssh_config: no ForwardAgent, RemoteForward, LocalCommand or ControlMaster
// toward a guest the bidder controls; only these options
"-F",
"/dev/null",
"-i",
&key.display().to_string(),
"-o",
"IdentitiesOnly=yes",
"-o",
"BatchMode=yes",
"-o",
&format!("ProxyCommand=dumbpipe connect {ticket}"),
"-o",
"HostKeyAlias=guest",
"-o",
&format!("UserKnownHostsFile={}", known_hosts_file.display()),
"-o",
"StrictHostKeyChecking=yes",
"-o",
"ConnectTimeout=60",
"--", // ssh re-parses options after the host otherwise: a command "-oProxyCommand=..." would run locally
"root@guest",
command,
]
.iter()
.map(|s| s.to_string())
.collect()
}
/// ssh only: what the command reads on stdin. Secrets go here, never in `command`: the command is
/// an argument of ssh on the host and of the login shell on the guest, so `ps` shows it on both.
fn ssh_stdin(input: &Value) -> Result<String, String> {
match input.get("stdin") {
None | Some(Value::Null) => Ok(String::new()),
Some(Value::String(s)) if s.len() <= 65536 => Ok(s.clone()),
Some(_) => Err("stdin must be a string of at most 65536 bytes".into()),
}
}
// The connect input -> (cmd, args, temp dir): 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).
fn session_command(
key: &Path,
input: &Value,
) -> Result<(&'static str, Vec<String>, Option<TempDir>), String> {
let t = ticket(input.get("ticket"))?;
if let Some(alpn) = input.get("alpn") {
let alpn = alpn
.as_str()
.filter(|a| {
(1..=64).contains(&a.len()) && a.bytes().all(|b| (0x21..=0x7e).contains(&b))
})
.ok_or("alpn must be 1 to 64 printable ASCII bytes")?;
let args = vec!["connect".into(), "--custom-alpn".into(), format!("utf8:{alpn}"), t];
return Ok(("dumbpipe", args, None));
}
let command = match input.get("command") {
None => None,
Some(Value::String(c)) if !c.is_empty() => Some(c.as_str()),
Some(_) => return Err("command must be a non-empty string".into()),
};
let lines = known_hosts("guest", input.get("hostKeys"))?;
let dir = TempDir::new()?;
std::fs::write(dir.0.join("known_hosts"), lines).map_err(|e| e.to_string())?;
let mut args = ssh_args(key, &t, &dir.0.join("known_hosts"), command.unwrap_or_default());
if command.is_none() {
args.pop();
}
Ok(("ssh", args, Some(dir)))
}
// Streaming UTF-8: the valid prefix of carry + bytes; an incomplete trailing sequence waits for more.
fn decode_utf8(carry: &mut Vec<u8>, bytes: &[u8]) -> String {
carry.extend_from_slice(bytes);
let valid = match std::str::from_utf8(carry) {
Ok(_) => carry.len(),
Err(e) if e.error_len().is_none() => e.valid_up_to(),
Err(_) => {
let text = String::from_utf8_lossy(carry).into_owned();
carry.clear();
return text;
}
};
let text = String::from_utf8_lossy(&carry[..valid]).into_owned();
carry.drain(..valid);
text
}
/// com.fedfork.exec.iroh.connect with subscribe: true: the one-shot input,
/// but the process stays up. Each stdout / stderr chunk is a message { stdout } / { stderr }, the exit
/// a last message { exitCode }, then the stream ends; it fails instead when ssh itself fails (255) or
/// dumbpipe exits non-zero having printed nothing. Send writes stdin (a string as is, else JSON);
/// Close and the signal kill the process. Same as js/index.js irohSession.
pub async fn iroh_session(key: &Path, input: &Value, signal: Signal) -> Result<Sub, String> {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::mpsc;
let (cmd, args, dir) = session_command(key, input)?;
let mut child = iroh_command(cmd, &args).spawn().map_err(|e| format!("spawn {cmd} {e}"))?;
let (msg_tx, messages) = mpsc::unbounded_channel();
let (commands, mut cmd_rx) = mpsc::unbounded_channel::<SubCmd>();
let (chunk_tx, mut chunk_rx) = mpsc::unbounded_channel::<(&'static str, Vec<u8>)>();
let readers: [(&'static str, Option<Box<dyn tokio::io::AsyncRead + Unpin + Send>>); 2] = [
("stdout", child.stdout.take().map(|r| Box::new(r) as _)),
("stderr", child.stderr.take().map(|r| Box::new(r) as _)),
];
for (name, reader) in readers {
let (Some(mut reader), tx) = (reader, chunk_tx.clone()) else { continue };
tokio::spawn(async move {
let mut buf = vec![0u8; 8192];
while let Ok(n) = reader.read(&mut buf).await {
if n == 0 || tx.send((name, buf[..n].to_vec())).is_err() {
break;
}
}
});
}
drop(chunk_tx);
// stdin has its own writer: a write blocked on a peer that stopped reading never holds up Close
let (write_tx, mut write_rx) = mpsc::unbounded_channel::<(String, tokio::sync::oneshot::Sender<Result<(), String>>)>();
if let Some(mut w) = child.stdin.take() {
tokio::spawn(async move {
while let Some((data, ack)) = write_rx.recv().await {
let sent = match w.write_all(data.as_bytes()).await {
Ok(()) => w.flush().await.map_err(|e| e.to_string()),
Err(e) => Err(e.to_string()),
};
let _ = ack.send(sent);
}
});
}
let stdin = Some(write_tx);
tokio::spawn(async move {
let _dir = dir; // removed when this task ends, or is dropped with the runtime
let (mut out_carry, mut err_carry) = (Vec::new(), Vec::new());
let (mut printed, mut killed, mut stderr) = (false, false, String::new());
loop {
tokio::select! {
chunk = chunk_rx.recv() => {
let Some((name, bytes)) = chunk else { break };
let text = if name == "stdout" {
printed = true;
decode_utf8(&mut out_carry, &bytes)
} else {
let text = decode_utf8(&mut err_carry, &bytes);
stderr.push_str(&text);
tail(&mut stderr);
text
};
if !text.is_empty() {
let _ = msg_tx.send(Ok(json!({ name: text })));
}
}
command = cmd_rx.recv() => match command {
Some(SubCmd::Send(data, ack)) => {
let data = match data { Value::String(s) => s, other => other.to_string() };
let refused = match &stdin { Some(w) => w.send((data, ack)).err().map(|e| e.0.1), None => Some(ack) };
if let Some(ack) = refused {
let _ = ack.send(Err(format!("{NSID} session is closed")));
}
}
Some(SubCmd::Close) | None => { killed = true; break; }
},
_ = signal.aborted() => { killed = true; break; }
}
}
if killed {
drop(stdin);
stop(&mut child).await;
} else {
drop(stdin);
let code = child.wait().await.ok().and_then(|s| s.code());
let message = match (cmd, code) {
("ssh", Some(255)) => Err(format!("ssh failed: {}", stderr.trim())),
("dumbpipe", Some(c)) if c != 0 && !printed => {
Err(format!("dumbpipe exited {c}: {}", stderr.trim()))
}
_ => Ok(json!({ "exitCode": code })),
};
let _ = msg_tx.send(message);
}
});
Ok(Sub { messages, commands })
}
async fn iroh(key: &Path, input: &Value) -> Result<Value, String> {
let t = ticket(input.get("ticket"))?;
if let Some(alpn) = input.get("alpn") {
let alpn = alpn
.as_str()
.filter(|a| {
(1..=64).contains(&a.len()) && a.bytes().all(|b| (0x21..=0x7e).contains(&b))
})
.ok_or("alpn must be 1 to 64 printable ASCII bytes")?;
let send = input
.get("send")
.map(|s| js_string(Some(s)))
.unwrap_or_default();
let args = ["connect".into(), "--custom-alpn".into(), format!("utf8:{alpn}"), t];
let out = exchange("dumbpipe", &args, &send, 30_000).await?;
if let Some(c) = out.code.filter(|c| *c != 0)
&& out.stdout.is_empty()
{
return Err(format!("dumbpipe exited {c}: {}", out.stderr.trim()));
}
// a peer that never closes, or sends too much: what it sent so far, marked
let mut output = json!({ "output": out.stdout });
if out.truncated {
output["truncated"] = json!(true);
} else if out.timed_out {
output["timedOut"] = json!(true);
}
return Ok(output);
}
let command = input
.get("command")
.and_then(Value::as_str)
.filter(|c| !c.is_empty())
.ok_or("command is required for ssh")?;
let stdin = ssh_stdin(input)?;
let dir = TempDir::new()?;
std::fs::write(dir.0.join("known_hosts"), known_hosts("guest", input.get("hostKeys"))?)
.map_err(|e| e.to_string())?;
let out = exchange("ssh", &ssh_args(key, &t, &dir.0.join("known_hosts"), command), &stdin, 120_000).await?;
if out.truncated {
return Ok(json!({ "output": out.stdout, "truncated": true }));
}
// 255 is ssh's own failure (host key mismatch, auth, dial); None is the timeout.
match out.code {
Some(255) => Err(format!("ssh failed: {}", out.stderr.trim())),
None => Err(format!("ssh timed out: {}", out.stderr.trim())),
Some(code) => Ok(json!({ "output": out.stdout, "exitCode": code })),
}
}
#[cfg(test)]
mod tests {
use super::*;
// The loopback rule: real dumbpipe, a local HTTP service behind `dumbpipe listen-tcp`, the raw fetch (no
// ssh, no key) over `dumbpipe connect <ticket>`; skipped without dumbpipe on PATH.
#[tokio::test]
async fn a_market_endpoint_fetch_goes_over_raw_dumbpipe() {
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt};
if std::process::Command::new("dumbpipe").arg("--version").output().is_err() {
eprintln!("skipped: no dumbpipe");
return;
}
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
tokio::spawn(async move {
while let Ok((mut tcp, _)) = listener.accept().await {
tokio::spawn(async move {
let mut buf = vec![0u8; 8192];
let mut got = Vec::new();
while !String::from_utf8_lossy(&got).contains("{\"a\":1}") {
let n = tcp.read(&mut buf).await.unwrap_or(0);
if n == 0 {
return;
}
got.extend_from_slice(&buf[..n]);
}
let text = String::from_utf8_lossy(&got).to_string();
let body = json!({ "line": text.lines().next(), "auth": text.contains("authorization: Bearer jwt1") }).to_string();
let _ = tcp.write_all(format!("HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\r\n{body}", body.len()).as_bytes()).await;
});
}
});
let mut listen = iroh_command("dumbpipe", &["listen-tcp".into(), "--host".into(), format!("127.0.0.1:{port}")]).spawn().unwrap();
let mut lines = tokio::io::BufReader::new(listen.stderr.take().unwrap()).lines();
let ticket = loop {
let line = lines.next_line().await.unwrap().expect("dumbpipe printed no ticket");
if let Some(t) = line.split_whitespace().find(|w| w.starts_with("endpoint")) {
break t.to_string();
}
};
let iroh = Iroh::new("/nonexistent".into());
let url = format!("iroh://{ticket}/xrpc/com.publicdomainrelay.temp.market.submitRfp");
let headers = vec![("authorization".to_string(), "Bearer jwt1".to_string()), ("content-type".into(), "application/json".into())];
let response = iroh.http("POST", &url, headers, br#"{"a":1}"#.to_vec()).await.unwrap();
let answer: Value = serde_json::from_slice(&response.body).unwrap();
assert_eq!((response.status, answer), (200, json!({ "line": "POST /xrpc/com.publicdomainrelay.temp.market.submitRfp HTTP/1.1", "auth": true })));
let _ = listen.kill().await;
}
#[test]
fn tickets_and_known_hosts() {
assert_eq!(ticket(Some(&json!("iroh://abc123\n"))).unwrap(), "abc123");
assert!(ticket(Some(&json!("ABC"))).is_err());
assert_eq!(
known_hosts(
"guest",
Some(&json!([{ "type": "ssh-ed25519", "publicKey": "AAAA" }, "ssh-rsa BBBB c"]))
)
.unwrap(),
"guest ssh-ed25519 AAAA\nguest ssh-rsa BBBB\n"
);
assert!(known_hosts("guest", Some(&json!([]))).is_err());
}
#[tokio::test]
async fn ssh_secrets_go_over_stdin() {
assert_eq!(ssh_stdin(&json!({})).unwrap(), "");
assert_eq!(ssh_stdin(&json!({ "stdin": "tok\n" })).unwrap(), "tok\n");
assert!(ssh_stdin(&json!({ "stdin": 5 })).is_err());
assert!(ssh_stdin(&json!({ "stdin": "x".repeat(65537) })).is_err());
// what the guest runs: a builtin read, so the token is in no process's argv
let args = ["-c".to_string(), "IFS= read -r t; printf 'got:%s' \"$t\"".to_string()];
let out = exchange("sh", &args, "tok\n", 5000).await.unwrap();
assert_eq!((out.stdout.as_str(), out.code), ("got:tok", Some(0)));
}
#[tokio::test]
async fn exchanges_keep_partial_output_on_timeout_a_stderr_tail_and_a_fresh_iroh_secret() {
let sh = |script: &str| ["-c".to_string(), script.to_string()];
let out = exchange("sh", &sh("printf partial; sleep 5"), "", 300).await.unwrap();
assert_eq!((out.stdout.as_str(), out.code, out.timed_out), ("partial", None, true));
let out = exchange("sh", &sh("head -c 1000000 /dev/zero | tr '\\0' x >&2; printf '\\nlast' >&2; exit 1"), "", 5000).await.unwrap();
assert_eq!((out.stderr.len(), out.code), (STDERR_MAX, Some(1)));
assert!(out.stderr.ends_with("x\nlast"));
let out = exchange("sh", &sh("printf %s \"$IROH_SECRET\""), "", 5000).await.unwrap();
assert!(out.stdout.len() == 64 && out.stdout.bytes().all(|b| b.is_ascii_hexdigit()), "{}", out.stdout);
assert_ne!(exchange("sh", &sh("printf %s \"$IROH_SECRET\""), "", 5000).await.unwrap().stdout, out.stdout, "one per process");
}
#[tokio::test]
async fn children_get_a_cleared_env_and_at_most_output_max_of_stdout() {
// env itself (no shell adding PWD, SHLVL): this test process's ATPROTO_*, CARGO_*, SSH_AUTH_SOCK stay out
let out = exchange("/usr/bin/env", &[], "", 5000).await.unwrap();
let keys: Vec<&str> = out.stdout.lines().filter_map(|l| l.split_once('=').map(|(k, _)| k)).collect();
assert!(keys.contains(&"IROH_SECRET") && keys.iter().all(|k| ["PATH", "HOME", "LANG", "IROH_SECRET"].contains(k)), "{keys:?}");
let sh = |script: &str| ["-c".to_string(), script.to_string()];
let out = exchange("sh", &sh("head -c 10000000 /dev/zero; sleep 5"), "", 5000).await.unwrap();
assert_eq!((out.stdout.len(), out.truncated, out.timed_out), (OUTPUT_MAX, true, false));
let out = exchange("sh", &sh(&format!("head -c {OUTPUT_MAX} /dev/zero")), "", 5000).await.unwrap();
assert_eq!((out.stdout.len(), out.truncated, out.code), (OUTPUT_MAX, false, Some(0)), "exactly the cap is all of it");
}
#[test]
fn utf8_split_across_chunks_waits_for_the_rest() {
let mut carry = vec![];
let bytes = "héllo".as_bytes();
assert_eq!(decode_utf8(&mut carry, &bytes[..2]), "h");
assert_eq!(decode_utf8(&mut carry, &bytes[2..]), "éllo");
assert!(carry.is_empty());
}
#[test]
fn session_command_matches_the_one_shot_inputs() {
let t = "abc123";
let (cmd, args, dir) = session_command(Path::new("/k"), &json!({ "ticket": t, "alpn": "echo/1" })).unwrap();
assert_eq!((cmd, args, dir.is_none()), ("dumbpipe", vec!["connect".to_string(), "--custom-alpn".into(), "utf8:echo/1".into(), t.into()], true));
let keys = json!([{ "type": "ssh-ed25519", "publicKey": "AAAA" }]);
let (cmd, args, dir) = session_command(Path::new("/k"), &json!({ "ticket": t, "hostKeys": keys })).unwrap();
assert_eq!(cmd, "ssh");
assert_eq!(args.last().map(String::as_str), Some("root@guest"), "no command: a remote shell");
let dir = dir.unwrap();
assert!(std::fs::read_to_string(dir.0.join("known_hosts")).unwrap().starts_with("guest ssh-ed25519 AAAA"));
let path = dir.0.clone();
drop(dir);
assert!(!path.exists(), "the temp dir goes with its guard");
assert_eq!(session_command(Path::new("/k"), &json!({ "ticket": t })).unwrap_err(), "no host keys: refusing to trust an unpinned guest");
assert_eq!(session_command(Path::new("/k"), &json!({ "ticket": t, "hostKeys": keys, "command": "" })).unwrap_err(), "command must be a non-empty string");
}
// js/fetch.test.mjs's cases against the same stand-ins (js/fake-bin/ssh runs relay.mjs; http-stub.mjs)
#[tokio::test]
async fn fetches_reuse_relays_open_more_for_concurrency_and_retry_once() {
use tokio::io::AsyncBufReadExt;
let dir = TempDir::new().unwrap();
let key = dir.0.join("key");
std::fs::write(&key, "not a real key\n").unwrap();
let sock = dir.0.join("it's.sock").display().to_string();
let fake = concat!(env!("CARGO_MANIFEST_DIR"), "/js/fake-bin");
let mut stub = tokio::process::Command::new("node").args([format!("{fake}/http-stub.mjs"), sock.clone()]).stdout(std::process::Stdio::piped()).kill_on_drop(true).spawn().unwrap();
tokio::io::BufReader::new(stub.stdout.take().unwrap()).read_line(&mut String::new()).await.unwrap();
let iroh = Iroh { key, ssh: format!("{fake}/ssh"), relays: Default::default() };
let refs = base64::Engine::encode(&base64::engine::general_purpose::URL_SAFE_NO_PAD, json!({ "ticket": "abc123", "hostKeys": [{ "type": "ssh-ed25519", "publicKey": "AAAAC3Bare" }] }).to_string());
let base = format!("iroh:///{refs}/{}", percent_encoding::utf8_percent_encode(&sock, percent_encoding::NON_ALPHANUMERIC));
let shared = &iroh;
let get = |path: &str| {
let url = format!("{base}{path}");
async move { shared.http("GET", &url, vec![], vec![]).await }
};
let body = |r: Response| serde_json::from_slice::<Value>(&r.body).unwrap();
let idle = |iroh: &Iroh| iroh.relays.lock().unwrap().values().map(Vec::len).sum::<usize>();
let posted = iroh.http("POST", &format!("{base}/mcp"), vec![("content-type".into(), "application/json".into()), ("Host".into(), "evil".into())], b"{}".to_vec()).await.unwrap();
assert_eq!(posted.status, 200);
assert!(posted.headers.contains(&("content-type".into(), "application/json".into())));
let posted = body(posted);
assert_eq!((posted["method"].clone(), posted["url"].clone(), posted["body"].clone(), posted["headers"]["host"].clone(), posted["n"].clone()), (json!("POST"), json!("/mcp"), json!("{}"), json!("guest"), json!(1)));
assert_eq!(body(get("/x").await.unwrap())["n"], 2, "keep-alive: the same relay");
// the host's way in: transport::fetchers (run::Http::schemes) -> Transport -> Call::fetch, its own relays
let host = atlibs_host::transport::fetchers(&[Transport { provides: &[NSID], call: Arc::new(Iroh { key: iroh.key.clone(), ssh: iroh.ssh.clone(), relays: Default::default() }) }]);
let via = host["iroh:"].fetch("PUT", &format!("{base}/put"), vec![], Some(b"x".to_vec())).await.ok().unwrap();
assert_eq!((body(via)["method"].clone(), host.len()), (json!("PUT"), 1));
let (slow, fast) = tokio::join!(get("/slow?ms=300"), get("/fast?chunked=1"));
let (slow, fast) = (body(slow.unwrap()), body(fast.unwrap()));
assert_eq!((slow["url"].clone(), fast["url"].clone()), (json!("/slow?ms=300"), json!("/fast?chunked=1")));
assert_eq!(idle(&iroh), 2, "a second relay for the second in-flight request");
assert_eq!(get("/c?close=1").await.unwrap().status, 200);
assert_eq!(idle(&iroh), 1, "connection: close is not kept");
let marker = dir.0.join("died").display().to_string();
let again = get(&format!("/mcp?die={}", percent_encoding::utf8_percent_encode(&marker, percent_encoding::NON_ALPHANUMERIC))).await.unwrap();
assert_eq!(body(again)["n"], 1, "retried on a fresh relay");
assert!(std::path::Path::new(&marker).exists());
let absent = format!("iroh:///{refs}/{}/", percent_encoding::utf8_percent_encode(&format!("{}/absent.sock", dir.0.display()), percent_encoding::NON_ALPHANUMERIC));
let failed = iroh.http("GET", &absent, vec![], vec![]).await.err().unwrap();
assert!(failed.starts_with("iroh: the relay ended (exit 1)") && failed.contains("ENOENT"), "{failed}");
assert!(iroh.http("G T", &base, vec![], vec![]).await.err().unwrap().contains("bad method"));
assert!(iroh.http("GET", &format!("{base}/"), vec![("x".into(), "a\r\nb: c".into())], vec![]).await.err().unwrap().contains("bad header"));
let no_keys = base64::Engine::encode(&base64::engine::general_purpose::URL_SAFE_NO_PAD, r#"{"ticket":"abc"}"#);
assert_eq!(iroh.http("GET", &format!("iroh:///{no_keys}/shell/mcp"), vec![], vec![]).await.err().unwrap(), "no host keys: refusing to trust an unpinned guest");
}
#[test]
fn iroh_urls_and_the_relay_command() {
let refs = |v: Value| base64::Engine::encode(&base64::engine::general_purpose::URL_SAFE_NO_PAD, v.to_string());
let keys = json!([{ "type": "ssh-ed25519", "publicKey": "AAAA" }]);
let r = refs(json!({ "ticket": "iroh://abc", "hostKeys": keys }));
assert_eq!(iroh_url(&format!("iroh:///{r}/shell/mcp?x=1#f")).unwrap(), Target { ticket: "abc".into(), host_keys: keys.clone(), socket: SHELL_SOCKET.into(), target: "/mcp?x=1".into(), raw: false });
let raw = raw_url("iroh://endpointabc234/xrpc/a.b.c?x=1").unwrap().unwrap();
assert_eq!((raw.ticket.as_str(), raw.target.as_str(), raw.raw), ("endpointabc234", "/xrpc/a.b.c?x=1", true));
assert!(raw_url(&format!("iroh://{r}/shell/mcp")).is_none());
assert_eq!(iroh_url(&format!("iroh://{r}/%2Ftmp%2Fs.sock")).unwrap().socket, "/tmp/s.sock");
assert_eq!(iroh_url(&format!("iroh://{r}/%2Ftmp%2Fs.sock")).unwrap().target, "/");
assert!(iroh_url("iroh:///abc").unwrap_err().starts_with("not an iroh"));
assert!(iroh_url("iroh:///!!/shell/mcp").unwrap_err().starts_with("not an iroh"));
assert!(iroh_url(&format!("iroh:///{r}/tmp/x")).unwrap_err().contains("absolute unix path"));
assert!(iroh_url(&format!("iroh:///{}/shell/mcp", refs(json!({ "ticket": "ABC" })))).unwrap_err().starts_with("not an iroh ticket"));
assert!(iroh_url(&format!("iroh:///{r}/shell/a b")).unwrap_err().contains("spaces"));
assert_eq!(relay_command("/a'b"), "socialweb-guest-agent relay '/a'\\''b'");
}
#[test]
fn ssh_command_cannot_be_an_option() {
let evil = "-oKnownHostsCommand=/nonexistent/pwned";
let args = ssh_args(Path::new("/k/key"), "abc", Path::new("/k/kh"), evil);
assert_eq!(&args[args.len() - 3..], ["--", "root@guest", evil]);
assert_eq!(&args[..2], ["-F", "/dev/null"], "no user or system ssh config");
// ssh -G resolves the config without connecting
let Ok(out) = std::process::Command::new("ssh").arg("-G").args(&args).output() else {
return; // no ssh binary here
};
let out = String::from_utf8_lossy(&out.stdout);
assert!(out.contains("proxycommand dumbpipe connect abc"), "{out}");
assert!(!out.contains("pwned"), "{out}");
}
}