feat(remote): add network-efficient thin client transport

Add SSH stdio remote attach, negotiate terminal-ANSI render encoding, move blit encoding server-side for remote clients, and split control/render writer delivery for safer backpressure.

Fixes #76

1 files changed, 363 insertions, 0 deletions

diff --git a/src/remote.rs b/src/remote.rs
new file mode 100644
index 00000000..a1344031
--- /dev/null
+++ b/src/remote.rs
@@ -0,0 +1,363 @@
+//! Remote thin-client launcher over SSH command stdio.
+
+use std::io;
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::path::{Path, PathBuf};
+use std::process::{Command, Stdio};
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
+use std::thread::{self, JoinHandle};
+use std::time::Duration;
+
+const BRIDGE_ACCEPT_POLL: Duration = Duration::from_millis(50);
+const BRIDGE_SOCKET_PERMISSION_MODE: u32 = 0o600;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct RemoteLaunch {
+    pub(crate) target: String,
+}
+
+pub(crate) fn extract_remote_args(
+    args: &[String],
+) -> Result<(Vec<String>, Option<RemoteLaunch>), String> {
+    let mut cleaned = Vec::with_capacity(args.len());
+    if let Some(program) = args.first() {
+        cleaned.push(program.clone());
+    }
+
+    let mut remote = None;
+    let mut index = 1;
+    while index < args.len() {
+        let arg = &args[index];
+        if arg == "--remote" {
+            if remote.is_some() {
+                return Err("--remote can only be specified once".to_string());
+            }
+            let Some(value) = args.get(index + 1) else {
+                return Err("missing value for --remote".to_string());
+            };
+            remote = Some(RemoteLaunch {
+                target: validate_remote_target(value)?.to_owned(),
+            });
+            index += 2;
+            continue;
+        }
+        if let Some(value) = arg.strip_prefix("--remote=") {
+            if remote.is_some() {
+                return Err("--remote can only be specified once".to_string());
+            }
+            remote = Some(RemoteLaunch {
+                target: validate_remote_target(value)?.to_owned(),
+            });
+            index += 1;
+            continue;
+        }
+
+        cleaned.push(arg.clone());
+        index += 1;
+    }
+
+    Ok((cleaned, remote))
+}
+
+fn validate_remote_target(target: &str) -> Result<&str, String> {
+    if target.is_empty() {
+        return Err("missing value for --remote".to_string());
+    }
+    if target.starts_with('-') {
+        return Err("--remote target must not start with '-'".to_string());
+    }
+    Ok(target)
+}
+
+pub(crate) fn run_remote(remote: RemoteLaunch) -> io::Result<()> {
+    let session_name = crate::session::active_name()
+        .unwrap_or_else(|| crate::session::DEFAULT_SESSION_NAME.to_string());
+    let local_socket = local_forward_socket_path(&remote.target, &session_name);
+
+    let _bridge = SshStdioBridge::start(remote.target, local_socket.clone(), session_name)?;
+
+    run_client_process(&local_socket)
+}
+
+pub(crate) fn run_remote_client_bridge() -> io::Result<()> {
+    let socket_path = crate::server::headless::client_socket_path();
+    let stream = UnixStream::connect(&socket_path).map_err(|err| {
+        io::Error::new(
+            err.kind(),
+            format!(
+                "failed to connect to remote Herdr client socket {}: {err}",
+                socket_path.display()
+            ),
+        )
+    })?;
+
+    let mut stdout = io::stdout().lock();
+    let mut socket_to_stdout = stream.try_clone()?;
+    let mut stdin_to_socket = stream;
+
+    let _upload = thread::spawn(move || {
+        let mut stdin = io::stdin();
+        let _ = copy_flush(&mut stdin, &mut stdin_to_socket);
+        let _ = stdin_to_socket.shutdown(std::net::Shutdown::Write);
+    });
+
+    copy_flush(&mut socket_to_stdout, &mut stdout).map(|_| ())
+}
+
+struct SshStdioBridge {
+    local_socket: PathBuf,
+    should_stop: Arc<AtomicBool>,
+    thread: Option<JoinHandle<()>>,
+}
+
+impl SshStdioBridge {
+    fn start(target: String, local_socket: PathBuf, session_name: String) -> io::Result<Self> {
+        let _ = std::fs::remove_file(&local_socket);
+        let listener = UnixListener::bind(&local_socket)?;
+        crate::ipc::restrict_socket_permissions(&local_socket, BRIDGE_SOCKET_PERMISSION_MODE)?;
+        listener.set_nonblocking(true)?;
+
+        let should_stop = Arc::new(AtomicBool::new(false));
+        let thread_stop = Arc::clone(&should_stop);
+        let thread = thread::spawn(move || {
+            while !thread_stop.load(Ordering::Acquire) {
+                match listener.accept() {
+                    Ok((stream, _addr)) => {
+                        if let Err(err) = stream.set_nonblocking(false) {
+                            eprintln!(
+                                "herdr: remote bridge failed to prepare client socket: {err}"
+                            );
+                            continue;
+                        }
+                        if let Err(err) = bridge_connection(stream, &target, &session_name) {
+                            eprintln!("herdr: remote bridge failed: {err}");
+                        }
+                    }
+                    Err(err) if err.kind() == io::ErrorKind::WouldBlock => {
+                        thread::sleep(BRIDGE_ACCEPT_POLL);
+                    }
+                    Err(err) => {
+                        eprintln!("herdr: remote bridge listener failed: {err}");
+                        break;
+                    }
+                }
+            }
+        });
+
+        Ok(Self {
+            local_socket,
+            should_stop,
+            thread: Some(thread),
+        })
+    }
+}
+
+impl Drop for SshStdioBridge {
+    fn drop(&mut self) {
+        self.should_stop.store(true, Ordering::Release);
+        let _ = std::fs::remove_file(&self.local_socket);
+        if let Some(thread) = self.thread.take() {
+            let _ = thread.join();
+        }
+    }
+}
+
+fn bridge_connection(stream: UnixStream, target: &str, session_name: &str) -> io::Result<()> {
+    let mut command = Command::new("ssh");
+    command.arg("-T").arg(target).arg("herdr");
+    if session_name != crate::session::DEFAULT_SESSION_NAME {
+        command.arg("--session").arg(session_name);
+    }
+    command
+        .arg("remote-client-bridge")
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::inherit());
+
+    let mut child = command
+        .spawn()
+        .map_err(|err| io::Error::new(err.kind(), format!("failed to start ssh bridge: {err}")))?;
+    let mut child_stdin = child
+        .stdin
+        .take()
+        .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "ssh bridge stdin missing"))?;
+    let mut child_stdout = child
+        .stdout
+        .take()
+        .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "ssh bridge stdout missing"))?;
+    let mut stream_to_child = stream.try_clone()?;
+    let mut child_to_stream = stream;
+
+    let upload = thread::spawn(move || {
+        let _ = copy_flush(&mut stream_to_child, &mut child_stdin);
+    });
+    let download = thread::spawn(move || {
+        let _ = copy_flush(&mut child_stdout, &mut child_to_stream);
+        let _ = child_to_stream.shutdown(std::net::Shutdown::Write);
+    });
+
+    let status = child.wait()?;
+    let _ = upload.join();
+    let _ = download.join();
+
+    if status.success() {
+        Ok(())
+    } else {
+        Err(io::Error::new(
+            io::ErrorKind::ConnectionAborted,
+            format!("ssh bridge exited with {status}"),
+        ))
+    }
+}
+
+fn copy_flush<R: io::Read, W: io::Write>(reader: &mut R, writer: &mut W) -> io::Result<u64> {
+    let mut buffer = [0_u8; 16 * 1024];
+    let mut total = 0;
+
+    loop {
+        let bytes_read = match reader.read(&mut buffer) {
+            Ok(0) => return Ok(total),
+            Ok(bytes_read) => bytes_read,
+            Err(err) if err.kind() == io::ErrorKind::Interrupted => continue,
+            Err(err) => return Err(err),
+        };
+
+        writer.write_all(&buffer[..bytes_read])?;
+        writer.flush()?;
+        total += bytes_read as u64;
+    }
+}
+
+fn run_client_process(local_socket: &Path) -> io::Result<()> {
+    let exe = std::env::current_exe()?;
+    let status = Command::new(exe)
+        .arg("client")
+        .env(
+            crate::server::headless::CLIENT_SOCKET_PATH_ENV_VAR,
+            local_socket,
+        )
+        .env("HERDR_RENDER_ENCODING", "terminal-ansi")
+        .env_remove(crate::api::SOCKET_PATH_ENV_VAR)
+        .stdin(Stdio::inherit())
+        .stdout(Stdio::inherit())
+        .stderr(Stdio::inherit())
+        .status()?;
+
+    if status.success() {
+        Ok(())
+    } else {
+        Err(io::Error::new(
+            io::ErrorKind::Interrupted,
+            format!("remote client exited with {status}"),
+        ))
+    }
+}
+
+fn local_forward_socket_path(target: &str, session_name: &str) -> PathBuf {
+    std::env::temp_dir().join(format!(
+        "herdr-remote-{}-{}-{}.sock",
+        std::process::id(),
+        sanitize_path_component(target),
+        sanitize_path_component(session_name)
+    ))
+}
+
+fn sanitize_path_component(input: &str) -> String {
+    let sanitized: String = input
+        .chars()
+        .map(|ch| {
+            if ch.is_ascii_alphanumeric() || matches!(ch, '.' | '_' | '-') {
+                ch
+            } else {
+                '-'
+            }
+        })
+        .collect();
+
+    sanitized.trim_matches('-').chars().take(32).collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn bridge_socket_is_user_only() {
+        use std::os::unix::fs::PermissionsExt;
+
+        let socket = std::env::temp_dir().join(format!(
+            "herdr-bridge-permissions-test-{}.sock",
+            std::process::id()
+        ));
+        let bridge =
+            SshStdioBridge::start("example".to_string(), socket.clone(), "default".to_string())
+                .expect("start bridge listener");
+
+        let mode = std::fs::metadata(&socket).unwrap().permissions().mode() & 0o777;
+        assert_eq!(mode, BRIDGE_SOCKET_PERMISSION_MODE);
+
+        drop(bridge);
+        let _ = std::fs::remove_file(socket);
+    }
+
+    #[test]
+    fn extract_remote_args_removes_space_form() {
+        let args = vec![
+            "herdr".into(),
+            "--remote".into(),
+            "dev".into(),
+            "--help".into(),
+        ];
+        let (cleaned, remote) = extract_remote_args(&args).unwrap();
+        assert_eq!(cleaned, vec!["herdr", "--help"]);
+        assert_eq!(remote.unwrap().target, "dev");
+    }
+
+    #[test]
+    fn extract_remote_args_removes_equals_form() {
+        let args = vec!["herdr".into(), "--remote=user@host".into()];
+        let (cleaned, remote) = extract_remote_args(&args).unwrap();
+        assert_eq!(cleaned, vec!["herdr"]);
+        assert_eq!(remote.unwrap().target, "user@host");
+    }
+
+    #[test]
+    fn extract_remote_args_requires_value() {
+        let args = vec!["herdr".into(), "--remote".into()];
+        let err = extract_remote_args(&args).unwrap_err();
+        assert_eq!(err, "missing value for --remote");
+    }
+
+    #[test]
+    fn extract_remote_args_rejects_empty_value() {
+        let args = vec!["herdr".into(), "--remote=".into()];
+        let err = extract_remote_args(&args).unwrap_err();
+        assert_eq!(err, "missing value for --remote");
+    }
+
+    #[test]
+    fn extract_remote_args_rejects_duplicate_values() {
+        let args = vec![
+            "herdr".into(),
+            "--remote=dev".into(),
+            "--remote=prod".into(),
+        ];
+        let err = extract_remote_args(&args).unwrap_err();
+        assert_eq!(err, "--remote can only be specified once");
+    }
+
+    #[test]
+    fn extract_remote_args_rejects_option_like_target() {
+        let args = vec!["herdr".into(), "--remote".into(), "-oProxyCommand=x".into()];
+        let err = extract_remote_args(&args).unwrap_err();
+        assert_eq!(err, "--remote target must not start with '-'");
+    }
+
+    #[test]
+    fn sanitize_path_component_removes_shell_sensitive_chars() {
+        assert_eq!(sanitize_path_component("user@host:22"), "user-host-22");
+    }
+}