//! 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}");
    }
}