From f0a2baf6f59c67d04e124d84cb437d438b32878c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mat=C3=ADas=20D=2E?= <9351115+decode2@users.noreply.github.com> Date: Sat, 18 Jul 2026 17:04:53 +0000 Subject: [PATCH] feat(pty): add Unix PTY spawn and I/O --- Cargo.lock | 104 ++++++++- crates/splice-pty/Cargo.toml | 4 + crates/splice-pty/src/lib.rs | 332 +++++++++++++++++++++++++--- crates/splice-pty/tests/unix_pty.rs | 64 ++++++ 4 files changed, 475 insertions(+), 29 deletions(-) create mode 100644 crates/splice-pty/tests/unix_pty.rs diff --git a/Cargo.lock b/Cargo.lock index f20d479..3cdd198 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -205,7 +205,7 @@ checksum = "2f3f6da4992df95bbcd9af42a6c7dcb994498fc9048230405f3b36ff7cd3f145" dependencies = [ "borsh-derive", "bytes", - "cfg_aliases", + "cfg_aliases 0.2.1", ] [[package]] @@ -419,6 +419,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd16c4719339c4530435d38e511904438d07cce7950afa3718a84ac36c10e89e" + [[package]] name = "cfg_aliases" version = "0.2.1" @@ -759,6 +765,12 @@ dependencies = [ "tendril", ] +[[package]] +name = "downcast-rs" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2" + [[package]] name = "dpi" version = "0.1.2" @@ -821,7 +833,7 @@ dependencies = [ "rustc_version", "toml 1.1.2+spec-1.1.0", "vswhom", - "winreg", + "winreg 0.55.0", ] [[package]] @@ -901,6 +913,17 @@ dependencies = [ "rustc_version", ] +[[package]] +name = "filedescriptor" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e40758ed24c9b2eeb76c35fb0aebc66c626084edd827e07e1552279814c6682d" +dependencies = [ + "libc", + "thiserror 1.0.69", + "winapi", +] + [[package]] name = "filetime" version = "0.2.29" @@ -1824,6 +1847,12 @@ dependencies = [ "unicode-segmentation", ] +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + [[package]] name = "libappindicator" version = "0.9.0" @@ -2032,6 +2061,18 @@ version = "1.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "650eef8c711430f1a879fdd01d4745a7deea475becfb90269c06775983bbf086" +[[package]] +name = "nix" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab2156c4fce2f8df6c499cc1c763e4394b7482525bf2a9701c9d79d215f519e4" +dependencies = [ + "bitflags 2.13.0", + "cfg-if", + "cfg_aliases 0.1.1", + "libc", +] + [[package]] name = "num-conv" version = "0.1.0" @@ -2476,6 +2517,27 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "portable-pty" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4a596a2b3d2752d94f51fac2d4a96737b8705dddd311a32b9af47211f08671e" +dependencies = [ + "anyhow", + "bitflags 1.3.2", + "downcast-rs", + "filedescriptor", + "lazy_static", + "libc", + "log", + "nix", + "serial2", + "shared_library", + "shell-words", + "winapi", + "winreg 0.10.1", +] + [[package]] name = "potential_utf" version = "0.1.5" @@ -3209,6 +3271,17 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "serial2" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9eb6ea5562eeaed6936b8b54e086aa0f88b9e5b1bef45beb038e2519fa1185b1" +dependencies = [ + "cfg-if", + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "serialize-to-javascript" version = "0.1.2" @@ -3251,6 +3324,22 @@ dependencies = [ "digest", ] +[[package]] +name = "shared_library" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a9e7e0f2bfae24d8a5b5a66c5b257a83c7412304311512a0c054cd5e619da11" +dependencies = [ + "lazy_static", + "libc", +] + +[[package]] +name = "shell-words" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc6fe69c597f9c37bfeeeeeb33da3530379845f10be461a66d16d03eca2ded77" + [[package]] name = "shlex" version = "2.0.1" @@ -3380,7 +3469,9 @@ version = "0.1.0" name = "splice-pty" version = "0.1.0" dependencies = [ + "libc", "log", + "portable-pty", "windows", ] @@ -5083,6 +5174,15 @@ dependencies = [ "memchr", ] +[[package]] +name = "winreg" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "80d0f4e272c85def139476380b12f9ac60926689dd2e01d4923222f40580869d" +dependencies = [ + "winapi", +] + [[package]] name = "winreg" version = "0.55.0" diff --git a/crates/splice-pty/Cargo.toml b/crates/splice-pty/Cargo.toml index b6eb7fe..ce2eb3f 100644 --- a/crates/splice-pty/Cargo.toml +++ b/crates/splice-pty/Cargo.toml @@ -12,6 +12,10 @@ edition.workspace = true # previously went to a stderr the windowed release build does not have. log = "0.4" +[target.'cfg(unix)'.dependencies] +libc = "0.2" +portable-pty = "0.9.0" + [target.'cfg(windows)'.dependencies] windows = { version = "0.61.3", features = [ "Win32_Foundation", diff --git a/crates/splice-pty/src/lib.rs b/crates/splice-pty/src/lib.rs index 2c5afe8..b267b0b 100644 --- a/crates/splice-pty/src/lib.rs +++ b/crates/splice-pty/src/lib.rs @@ -113,6 +113,8 @@ pub enum PtyError { Windows(windows::core::Error), Io(std::io::Error), CommandContainsNul, + InvalidWorkingDirectory, + InvalidEnvironment, InvalidOutput, SessionClosed, UnsupportedPlatform, @@ -125,6 +127,10 @@ impl std::fmt::Display for PtyError { Self::Windows(error) => write!(f, "Windows API error: {error}"), Self::Io(error) => write!(f, "I/O error: {error}"), Self::CommandContainsNul => write!(f, "command contains a NUL byte"), + Self::InvalidWorkingDirectory => { + write!(f, "working directory must be an absolute directory") + } + Self::InvalidEnvironment => write!(f, "environment override is invalid"), Self::InvalidOutput => write!(f, "PTY output was not valid UTF-8"), Self::SessionClosed => write!(f, "PTY session is closed"), Self::UnsupportedPlatform => write!(f, "ConPTY is only supported on Windows"), @@ -195,73 +201,261 @@ pub fn run_conpty_command_with_resize( Err(PtyError::UnsupportedPlatform) } -#[cfg(not(windows))] -pub struct PtySession; +#[cfg(unix)] +#[derive(Debug, Default)] +pub struct PtySpawnOptions { + pub cwd: Option, + pub env: Vec<(String, String)>, +} -#[cfg(not(windows))] +#[cfg(unix)] +pub struct PtySession { + lifecycle: std::sync::Arc, + master: std::sync::Mutex>, + writer: std::sync::Mutex>>, + killer: std::sync::Mutex>, + pid: Option, + completed: std::sync::Mutex>, + reader: std::sync::Mutex>>, + waiter: std::sync::Mutex>>, + running: std::sync::Arc, + process_name: String, + on_closing: std::sync::Arc, +} + +#[cfg(unix)] impl PtySession { pub fn spawn( - _program: &str, - _args: &[&str], - _size: TerminalSize, - _on_output: F, - _on_exit: G, + program: &str, + args: &[&str], + size: TerminalSize, + on_output: F, + on_exit: G, ) -> Result where F: FnMut(u64, String) + Send + 'static, G: FnOnce(u64) + Send + 'static, { - Err(PtyError::UnsupportedPlatform) + Self::spawn_with_close_hook(program, args, size, on_output, on_exit, || {}) } pub fn spawn_with_close_hook( - _program: &str, - _args: &[&str], - _size: TerminalSize, - _on_output: F, - _on_exit: G, - _on_closing: H, + program: &str, + args: &[&str], + size: TerminalSize, + on_output: F, + on_exit: G, + on_closing: H, ) -> Result where F: FnMut(u64, String) + Send + 'static, G: FnOnce(u64) + Send + 'static, H: Fn() + Send + Sync + 'static, { - Err(PtyError::UnsupportedPlatform) + Self::spawn_with_options_and_close_hook( + program, + args, + PtySpawnOptions::default(), + size, + on_output, + on_exit, + on_closing, + ) + } + + pub fn spawn_with_options( + program: &str, + args: &[&str], + options: PtySpawnOptions, + size: TerminalSize, + on_output: F, + on_exit: G, + ) -> Result + where + F: FnMut(u64, String) + Send + 'static, + G: FnOnce(u64) + Send + 'static, + { + Self::spawn_with_options_and_close_hook( + program, + args, + options, + size, + on_output, + on_exit, + || {}, + ) + } + + fn spawn_with_options_and_close_hook( + program: &str, + args: &[&str], + options: PtySpawnOptions, + size: TerminalSize, + on_output: F, + on_exit: G, + on_closing: H, + ) -> Result + where + F: FnMut(u64, String) + Send + 'static, + G: FnOnce(u64) + Send + 'static, + H: Fn() + Send + Sync + 'static, + { + use portable_pty::{native_pty_system, CommandBuilder, PtySize}; + + if program == "cmd.exe" { + return Err(PtyError::UnsupportedPlatform); + } + validate_spawn_inputs(program, args, &options)?; + + let lifecycle = std::sync::Arc::new(PtySessionLifecycle::new(next_pty_session_id())); + let running = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)); + let pair = native_pty_system() + .openpty(PtySize { + rows: size.rows, + cols: size.columns, + pixel_width: 0, + pixel_height: 0, + }) + .map_err(|error| PtyError::Io(std::io::Error::other(error)))?; + let mut command = CommandBuilder::new(program); + command.args(args); + if let Some(cwd) = &options.cwd { + command.cwd(cwd); + } + for (key, value) in &options.env { + command.env(key, value); + } + + let mut child = pair + .slave + .spawn_command(command) + .map_err(|error| PtyError::Io(std::io::Error::other(error)))?; + let pid = child.process_id(); + let killer = child.clone_killer(); + drop(pair.slave); + let reader = pair + .master + .try_clone_reader() + .map_err(|error| PtyError::Io(std::io::Error::other(error)))?; + let writer = pair + .master + .take_writer() + .map_err(|error| PtyError::Io(std::io::Error::other(error)))?; + let id = lifecycle.id(); + let reader_lifecycle = std::sync::Arc::clone(&lifecycle); + + let reader = + std::thread::spawn(move || read_output(reader, id, reader_lifecycle, on_output)); + + let waiter_lifecycle = std::sync::Arc::clone(&lifecycle); + let waiter_running = std::sync::Arc::clone(&running); + let (completed_tx, completed) = std::sync::mpsc::channel(); + let waiter = std::thread::spawn(move || { + let exited = child.wait().is_ok(); + waiter_running.store(false, std::sync::atomic::Ordering::SeqCst); + let _ = completed_tx.send(()); + if exited && waiter_lifecycle.should_emit_natural_exit() { + on_exit(waiter_lifecycle.id()); + } + }); + + Ok(Self { + lifecycle, + master: std::sync::Mutex::new(pair.master), + writer: std::sync::Mutex::new(Some(writer)), + killer: std::sync::Mutex::new(killer), + pid, + completed: std::sync::Mutex::new(completed), + reader: std::sync::Mutex::new(Some(reader)), + waiter: std::sync::Mutex::new(Some(waiter)), + running, + process_name: std::path::Path::new(program) + .file_name() + .unwrap_or_default() + .to_string_lossy() + .into_owned(), + on_closing: std::sync::Arc::new(on_closing), + }) } pub fn id(&self) -> u64 { - 0 + self.lifecycle.id() } - pub fn write(&self, _data: &str) -> Result<(), PtyError> { - Err(PtyError::UnsupportedPlatform) + pub fn write(&self, data: &str) -> Result<(), PtyError> { + if !self.lifecycle.should_emit_natural_exit() { + return Err(PtyError::SessionClosed); + } + + let mut writer = self.writer.lock().map_err(|_| PtyError::SessionClosed)?; + let writer = writer.as_mut().ok_or(PtyError::SessionClosed)?; + writer.write_all(data.as_bytes())?; + writer.flush()?; + Ok(()) } pub fn interrupt(&self) -> Result<(), PtyError> { - Err(PtyError::UnsupportedPlatform) + self.write("\u{3}") } - pub fn resize(&self, _size: TerminalSize) -> Result<(), PtyError> { - Err(PtyError::UnsupportedPlatform) + pub fn resize(&self, size: TerminalSize) -> Result<(), PtyError> { + use portable_pty::PtySize; + self.master + .lock() + .map_err(|_| PtyError::SessionClosed)? + .resize(PtySize { + rows: size.rows, + cols: size.columns, + pixel_width: 0, + pixel_height: 0, + }) + .map_err(|error| PtyError::Io(std::io::Error::other(error))) } pub fn is_running(&self) -> Result { - Err(PtyError::UnsupportedPlatform) + Ok(self.running.load(std::sync::atomic::Ordering::SeqCst)) } pub fn active_process_name(&self) -> Result { - Err(PtyError::UnsupportedPlatform) + Ok(self.process_name.clone()) } pub fn active_process_candidates(&self) -> Result, PtyError> { - Err(PtyError::UnsupportedPlatform) + Ok(vec![self.active_process_name()?]) } - pub fn close(&self) {} + pub fn close(&self) { + if self.lifecycle.begin_close() { + (self.on_closing)(); + if let Ok(mut killer) = self.killer.lock() { + let _ = killer.kill(); + } + let completed = self.completed.lock().is_ok_and(|completed| { + completed + .recv_timeout(std::time::Duration::from_millis(250)) + .is_ok() + }); + if !completed { + if let Some(pid) = self.pid { + // Root-only escalation; process-group teardown belongs to PR4. + unsafe { libc::kill(pid as libc::pid_t, libc::SIGKILL) }; + } + } + if let Ok(mut writer) = self.writer.lock() { + writer.take(); + } + for thread in [&self.waiter, &self.reader] { + if let Ok(mut thread) = thread.lock() { + if let Some(thread) = thread.take() { + let _ = thread.join(); + } + } + } + } + } } -#[cfg(not(windows))] +#[cfg(unix)] impl PtySessionContract for PtySession { fn id(&self) -> u64 { self.id() @@ -296,6 +490,90 @@ impl PtySessionContract for PtySession { } } +#[cfg(unix)] +impl Drop for PtySession { + fn drop(&mut self) { + self.close(); + } +} + +#[cfg(unix)] +fn next_pty_session_id() -> u64 { + static NEXT_PTY_SESSION_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); + NEXT_PTY_SESSION_ID.fetch_add(1, std::sync::atomic::Ordering::SeqCst) +} + +#[cfg(unix)] +fn validate_spawn_inputs( + program: &str, + args: &[&str], + options: &PtySpawnOptions, +) -> Result<(), PtyError> { + if program.contains('\0') || args.iter().any(|arg| arg.contains('\0')) { + return Err(PtyError::CommandContainsNul); + } + if options + .cwd + .as_ref() + .is_some_and(|cwd| !cwd.is_absolute() || !cwd.is_dir()) + { + return Err(PtyError::InvalidWorkingDirectory); + } + if options + .env + .iter() + .any(|(key, value)| key.is_empty() || key.contains('\0') || value.contains('\0')) + { + return Err(PtyError::InvalidEnvironment); + } + Ok(()) +} + +#[cfg(unix)] +fn read_output( + mut reader: Box, + id: u64, + lifecycle: std::sync::Arc, + mut on_output: F, +) where + F: FnMut(u64, String), +{ + use std::io::Read; + + let mut bytes = [0; 4096]; + let mut pending = Vec::new(); + while lifecycle.should_emit_natural_exit() { + let Ok(count) = reader.read(&mut bytes) else { + break; + }; + if count == 0 { + break; + } + pending.extend_from_slice(&bytes[..count]); + + while !pending.is_empty() { + match std::str::from_utf8(&pending) { + Ok(output) => { + on_output(id, output.to_owned()); + pending.clear(); + } + Err(error) if error.valid_up_to() > 0 => { + let valid = error.valid_up_to(); + let output = std::str::from_utf8(&pending[..valid]) + .expect("valid UTF-8 prefix reported by Utf8Error"); + on_output(id, output.to_owned()); + pending.drain(..valid); + } + Err(error) if error.error_len().is_none() => break, + Err(error) => { + on_output(id, "\u{fffd}".to_owned()); + pending.drain(..error.error_len().unwrap()); + } + } + } + } +} + #[cfg(windows)] pub fn run_conpty_smoke_command( program: &str, diff --git a/crates/splice-pty/tests/unix_pty.rs b/crates/splice-pty/tests/unix_pty.rs new file mode 100644 index 0000000..032c459 --- /dev/null +++ b/crates/splice-pty/tests/unix_pty.rs @@ -0,0 +1,64 @@ +#![cfg(unix)] + +#[rustfmt::skip] +mod tests { +use std::{path::PathBuf, sync::mpsc, time::{Duration, Instant}}; +use splice_pty::{PtyError, PtySession, PtySpawnOptions, TerminalSize}; + +fn size() -> TerminalSize { TerminalSize::new(80, 24).unwrap() } fn receive_until(receiver: &mpsc::Receiver, marker: &str) -> String { + let deadline = Instant::now() + Duration::from_secs(5); let mut output = String::new(); + while Instant::now() < deadline && !output.contains(marker) { + if let Ok(chunk) = receiver.recv_timeout(Duration::from_millis(250)) { output.push_str(&chunk); } + } output +} +fn spawn(command: &str) -> (PtySession, mpsc::Receiver) { + let (sender, receiver) = mpsc::channel(); let session = PtySession::spawn("/bin/sh", &[ + "-c", command], size(), move |_, output| { let _ = sender.send(output); }, |_| {}).unwrap(); + (session, receiver) +} +fn pid(output: &str) -> String { output.split("pid=").nth(1).unwrap().trim().to_owned() } fn is_alive(pid: &str) -> bool { std::process::Command::new("kill").args(["-0", pid]).status().unwrap().success() } +#[test] fn unix_pty_uses_structured_argv_absolute_cwd_env_and_utf8_output() { + let cwd = std::env::temp_dir().join(format!("splice-pty-unix-{}", std::process::id())); + std::fs::create_dir_all(&cwd).unwrap(); let (sender, receiver) = mpsc::channel(); + let session = PtySession::spawn_with_options("/bin/sh", &["-c", "printf 'cwd=%s env=%s utf8=%s\\n' \"$PWD\" \"$SPLICE_PTY_ENV\" 'café'"], + PtySpawnOptions { cwd: Some(cwd.clone()), env: vec![("SPLICE_PTY_ENV".into(), "configured".into())] }, size(), + move |_, output| { let _ = sender.send(output); }, |_| {}).unwrap(); + let output = receive_until(&receiver, "utf8=café"); session.close(); std::fs::remove_dir_all(&cwd).unwrap(); + assert!(output.contains(&format!("cwd={}", cwd.display())) && output.contains("env=configured") && output.contains("utf8=café")); } +#[test] fn unix_pty_starts_shell_accepts_input_and_reports_liveness_and_name() { + let (session, receiver) = spawn("read line; printf 'input=%s\\n' \"$line\""); + assert!(session.is_running().unwrap()); assert_eq!(session.active_process_name().unwrap(), "sh"); session.write("accepted\n").unwrap(); + assert!(receive_until(&receiver, "input=accepted").contains("input=accepted")); session.close(); } +#[test] fn unix_pty_rejects_a_non_absolute_working_directory() { + let result = PtySession::spawn_with_options("/bin/sh", &["-c", "true"], + PtySpawnOptions { cwd: Some(PathBuf::from("relative")), ..PtySpawnOptions::default() }, size(), |_, _| {}, |_| {}); + assert!(matches!(result, Err(PtyError::InvalidWorkingDirectory))); } +#[test] fn unix_pty_close_and_drop_terminate_children_without_natural_exit() { + let (exit_sender, exit_receiver) = mpsc::channel(); let (sender, receiver) = mpsc::channel(); + let session = PtySession::spawn("/bin/sh", &["-c", "printf 'pid=%s\\n' $$; sleep 30"], size(), + move |_, output| { let _ = sender.send(output); }, move |_| { let _ = exit_sender.send(()); }).unwrap(); + let child = pid(&receive_until(&receiver, "pid=")); session.close(); session.close(); + assert!(!is_alive(&child) && exit_receiver.recv_timeout(Duration::from_millis(100)).is_err()); + let (session, receiver) = spawn("printf 'pid=%s\\n' $$; sleep 30"); let child = pid(&receive_until(&receiver, "pid=")); + drop(session); assert!(!is_alive(&child)); } +#[test] fn unix_pty_interrupt_writes_etx() { + let (session, receiver) = spawn("trap 'printf interrupted; exit' INT; printf ready; while :; do sleep 1; done"); + assert!(receive_until(&receiver, "ready").contains("ready")); session.interrupt().unwrap(); + assert!(receive_until(&receiver, "interrupted").contains("interrupted")); session.close(); } +#[test] fn unix_pty_replaces_malformed_utf8_and_preserves_split_sequences() { + let (session, receiver) = spawn("printf '\\303'; sleep 0.1; printf '\\251\\377ok'"); + assert!(receive_until(&receiver, "ok").contains("é�ok")); session.close(); } +#[test] fn unix_pty_rejects_exact_windows_default_but_runs_unix_programs() { + let result = PtySession::spawn("cmd.exe", &[], size(), |_, _| {}, |_| {}); + assert!(matches!(result, Err(PtyError::UnsupportedPlatform))); let (session, receiver) = spawn("printf explicit-unix"); + assert!(receive_until(&receiver, "explicit-unix").contains("explicit-unix")); session.close(); } +#[test] fn unix_pty_close_is_bounded_when_child_ignores_hup() { + let (session, receiver) = spawn("trap '' HUP; printf 'pid=%s\\n' $$; while :; do sleep 1; done"); + let child = pid(&receive_until(&receiver, "pid=")); let started = Instant::now(); session.close(); + assert!(started.elapsed() < Duration::from_secs(2) && !is_alive(&child)); } +#[test] fn unix_pty_resize_changes_live_terminal_dimensions() { + let (session, receiver) = spawn("stty size; sleep 0.2; stty size"); + assert!(receive_until(&receiver, "24 80").contains("24 80")); + session.resize(TerminalSize::new(132, 43).unwrap()).unwrap(); + assert!(receive_until(&receiver, "43 132").contains("43 132")); session.close(); } +}