use std::io;
#[cfg(unix)]
use std::io::{Read, Write};
use std::sync::{Arc, Mutex};
#[cfg(unix)]
use std::time::Duration;
use crossterm::event::{KeyCode, KeyEvent, KeyModifiers};
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;
use crate::sandbox::Sandbox;
use crate::ui::ansi::{StripPolicy, strip_escapes};
#[cfg(unix)]
use crate::ui::pty_relay::stdio_from_fd;
use crate::ui::shell_phase::ShellKind;
pub(crate) enum ShellEvent {
Output(Vec<u8>),
Exited { outcome: ShellOutcome },
}
#[derive(Clone, Debug)]
pub(crate) struct ShellOutcome {
pub exit_code: Option<i32>,
pub captured: String,
pub interrupted: bool,
}
pub(crate) struct ShellSession {
pub input_tx: mpsc::UnboundedSender<Vec<u8>>,
pub events_rx: mpsc::UnboundedReceiver<ShellEvent>,
pub interrupt: Option<oneshot::Sender<()>>,
pub join: Option<JoinHandle<()>>,
pub kind: ShellKind,
pub command: String,
}
#[cfg(unix)]
pub(crate) fn spawn(
command: &str,
sandbox: &Sandbox,
kind: ShellKind,
cols: u16,
rows: u16,
) -> io::Result<ShellSession> {
let (primary, secondary) = open_pty_pair()?;
let secondary_fd = secondary.as_raw_fd();
set_cooked(secondary_fd);
set_winsize(secondary_fd, cols, rows);
let mut cmd = sandbox.command_for_interactive(command);
let din = secondary.try_clone()?;
let dout = secondary.try_clone()?;
let derr = secondary.try_clone()?;
unsafe {
cmd.stdin(stdio_from_fd(din));
cmd.stdout(stdio_from_fd(dout));
cmd.stderr(stdio_from_fd(derr));
}
cmd.kill_on_drop(true);
unsafe {
use std::os::unix::process::CommandExt;
cmd.as_std_mut().pre_exec(|| {
if libc::setsid() == -1 {
return Err(io::Error::last_os_error());
}
let _ = libc::ioctl(0, libc::TIOCSCTTY as _, 0);
Ok(())
});
}
let mut child = cmd.spawn()?;
let pid = child.id();
let primary_read = primary.try_clone()?;
let primary_write = primary.try_clone()?;
drop(secondary);
drop(primary);
let (events_tx, events_rx) = mpsc::unbounded_channel::<ShellEvent>();
let (input_tx, mut input_rx) = mpsc::unbounded_channel::<Vec<u8>>();
let (interrupt_tx, interrupt_rx) = oneshot::channel::<()>();
let captured: Arc<Mutex<Vec<u8>>> = Arc::new(Mutex::new(Vec::new()));
let captured_r = captured.clone();
let events_tx_r = events_tx.clone();
let reader = std::thread::Builder::new()
.name("shell-pty-read".into())
.spawn(move || {
let mut f = primary_read;
let mut buf = [0u8; 4096];
loop {
match f.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
let chunk = &buf[..n];
if let Ok(mut c) = captured_r.lock() {
c.extend_from_slice(chunk);
}
let _ = events_tx_r.send(ShellEvent::Output(chunk.to_vec()));
}
Err(e) if e.raw_os_error() == Some(libc::EIO) => break,
Err(_) => break,
}
}
})?;
let _writer = std::thread::Builder::new()
.name("shell-pty-write".into())
.spawn(move || {
let mut f = primary_write;
while let Some(bytes) = input_rx.blocking_recv() {
if f.write_all(&bytes).is_err() {
break;
}
let _ = f.flush();
}
})?;
let join = tokio::spawn(async move {
let (status, interrupted) = tokio::select! {
biased;
Ok(()) = interrupt_rx => { kill_group(pid); (child.wait().await, true) }
s = child.wait() => (s, false),
};
let _ = tokio::time::timeout(
Duration::from_secs(3),
tokio::task::spawn_blocking(move || {
let _ = reader.join();
}),
)
.await;
let raw = captured
.lock()
.map(|c| String::from_utf8_lossy(&c).into_owned())
.unwrap_or_default();
let exit_code = status.ok().and_then(|s| s.code());
let captured = strip_escapes(&raw, StripPolicy::KEEP_NEWLINE);
let outcome = ShellOutcome {
exit_code,
captured,
interrupted,
};
let _ = events_tx.send(ShellEvent::Exited { outcome });
});
Ok(ShellSession {
input_tx,
events_rx,
interrupt: Some(interrupt_tx),
join: Some(join),
kind,
command: command.to_string(),
})
}
pub(crate) fn key_to_bytes(key: KeyEvent) -> Option<Vec<u8>> {
let KeyEvent {
code, modifiers, ..
} = key;
if modifiers.contains(KeyModifiers::CONTROL) {
if let KeyCode::Char(c) = code {
let b = c.to_ascii_lowercase() as u32;
if (b'@' as u32..=b'_' as u32).contains(&b) {
return Some(vec![(b as u8) & 0o37]);
}
}
return None;
}
match code {
KeyCode::Char(c) => Some(c.encode_utf8(&mut [0u8; 4]).as_bytes().to_vec()),
KeyCode::Enter => Some(vec![0o015]), KeyCode::Backspace => Some(vec![0o177]), KeyCode::Tab => Some(vec![0o011]),
KeyCode::Up => Some(b"\x1b[A".to_vec()),
KeyCode::Down => Some(b"\x1b[B".to_vec()),
KeyCode::Right => Some(b"\x1b[C".to_vec()),
KeyCode::Left => Some(b"\x1b[D".to_vec()),
KeyCode::Home => Some(b"\x1b[H".to_vec()),
KeyCode::End => Some(b"\x1b[F".to_vec()),
KeyCode::Delete => Some(b"\x1b[3~".to_vec()),
KeyCode::PageUp => Some(b"\x1b[5~".to_vec()),
KeyCode::PageDown => Some(b"\x1b[6~".to_vec()),
_ => None,
}
}
#[cfg(not(unix))]
async fn pump<R>(
mut reader: R,
events_tx: mpsc::UnboundedSender<ShellEvent>,
captured: Arc<Mutex<Vec<u8>>>,
) where
R: tokio::io::AsyncRead + Unpin,
{
use tokio::io::AsyncReadExt;
let mut buf = [0u8; 4096];
loop {
match reader.read(&mut buf).await {
Ok(0) | Err(_) => break,
Ok(n) => {
let chunk = &buf[..n];
if let Ok(mut c) = captured.lock() {
c.extend_from_slice(chunk);
}
let _ = events_tx.send(ShellEvent::Output(chunk.to_vec()));
}
}
}
}
#[cfg(not(unix))]
pub(crate) fn spawn(
command: &str,
sandbox: &Sandbox,
kind: ShellKind,
_cols: u16,
_rows: u16,
) -> io::Result<ShellSession> {
use std::process::Stdio;
let mut cmd = sandbox.command_for_interactive(command);
cmd.stdin(Stdio::null());
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::piped());
cmd.kill_on_drop(true);
let mut child = cmd.spawn()?;
let stdout = child.stdout.take();
let stderr = child.stderr.take();
let (events_tx, events_rx) = mpsc::unbounded_channel::<ShellEvent>();
let (input_tx, mut input_rx) = mpsc::unbounded_channel::<Vec<u8>>();
let (interrupt_tx, interrupt_rx) = oneshot::channel::<()>();
let join = tokio::spawn(async move {
let _drain = tokio::spawn(async move { while input_rx.recv().await.is_some() {} });
let captured: Arc<Mutex<Vec<u8>>> = Arc::new(Mutex::new(Vec::new()));
let cap = captured.clone();
let out_task = stdout.map(|s| tokio::spawn(pump(s, events_tx.clone(), cap)));
let cap = captured.clone();
let err_task = stderr.map(|s| tokio::spawn(pump(s, events_tx.clone(), cap)));
let (status, interrupted) = tokio::select! {
biased;
Ok(()) = interrupt_rx => {
let _ = child.kill().await;
(child.wait().await, true)
}
s = child.wait() => (s, false),
};
if let Some(t) = out_task {
let _ = t.await;
}
if let Some(t) = err_task {
let _ = t.await;
}
let raw = captured
.lock()
.map(|c| String::from_utf8_lossy(&c).into_owned())
.unwrap_or_default();
let exit_code = status.ok().and_then(|s| s.code());
let stripped = strip_escapes(&raw, StripPolicy::KEEP_NEWLINE);
let outcome = ShellOutcome {
exit_code,
captured: stripped,
interrupted,
};
let _ = events_tx.send(ShellEvent::Exited { outcome });
});
Ok(ShellSession {
input_tx,
events_rx,
interrupt: Some(interrupt_tx),
join: Some(join),
kind,
command: command.to_string(),
})
}
#[cfg(unix)]
fn kill_group(pid: Option<u32>) {
if let Some(pid) = pid {
unsafe {
if libc::kill(-(pid as libc::pid_t), libc::SIGKILL) == -1 {
let _ = libc::kill(pid as libc::pid_t, libc::SIGKILL);
}
}
}
}
#[cfg(unix)]
use std::os::unix::io::{AsRawFd, FromRawFd};
#[cfg(unix)]
fn open_pty_pair() -> io::Result<(std::fs::File, std::fs::File)> {
let master_fd = unsafe { libc::posix_openpt(libc::O_RDWR | libc::O_NOCTTY) };
if master_fd < 0 {
return Err(io::Error::last_os_error());
}
macro_rules! fail {
() => {{
let e = io::Error::last_os_error();
unsafe {
libc::close(master_fd);
}
return Err(e);
}};
}
if unsafe { libc::grantpt(master_fd) } < 0 || unsafe { libc::unlockpt(master_fd) } < 0 {
fail!();
}
let name_ptr = unsafe { libc::ptsname(master_fd) };
if name_ptr.is_null() {
fail!();
}
let slave_fd = unsafe { libc::open(name_ptr, libc::O_RDWR | libc::O_NOCTTY) };
if slave_fd < 0 {
fail!();
}
Ok((unsafe { std::fs::File::from_raw_fd(master_fd) }, unsafe {
std::fs::File::from_raw_fd(slave_fd)
}))
}
#[cfg(unix)]
fn set_cooked(fd: libc::c_int) {
let mut t: libc::termios = unsafe { std::mem::zeroed() };
if unsafe { libc::tcgetattr(fd, &mut t) } < 0 {
return;
}
t.c_lflag |= libc::ECHO | libc::ICANON | libc::ISIG | libc::IEXTEN;
t.c_lflag &= !libc::ECHOCTL;
t.c_iflag |= libc::ICRNL | libc::IXON;
t.c_oflag |= libc::OPOST | libc::ONLCR;
t.c_cc[libc::VINTR] = 0o03;
t.c_cc[libc::VQUIT] = 0o34;
t.c_cc[libc::VEOF] = 0o04;
unsafe {
libc::tcsetattr(fd, libc::TCSANOW, &t);
}
}
#[cfg(unix)]
fn set_winsize(fd: libc::c_int, cols: u16, rows: u16) {
let mut ws: libc::winsize = unsafe { std::mem::zeroed() };
ws.ws_col = cols;
ws.ws_row = rows;
unsafe {
libc::ioctl(fd, libc::TIOCSWINSZ, &ws);
}
}
#[cfg(all(test, unix))]
mod tests {
use super::*;
use crate::sandbox::{Sandbox, SandboxMode};
use std::time::Duration;
use tokio::time::timeout;
fn sb() -> Sandbox {
Sandbox::new(SandboxMode::Off)
}
async fn collect_outcome(rx: &mut mpsc::UnboundedReceiver<ShellEvent>) -> ShellOutcome {
while let Some(ev) = rx.recv().await {
if let ShellEvent::Exited { outcome } = ev {
return outcome;
}
}
panic!("events stream closed without Exited");
}
#[tokio::test]
async fn captures_stdout_and_stderr() {
let s = spawn(
"echo hello; echo oops 1>&2",
&sb(),
ShellKind::Visible,
80,
24,
)
.unwrap();
let ShellSession {
input_tx,
mut events_rx,
..
} = s;
drop(input_tx);
let out = collect_outcome(&mut events_rx).await;
assert_eq!(out.exit_code, Some(0));
assert!(out.captured.contains("hello"), "got: {:?}", out.captured);
assert!(out.captured.contains("oops"), "got: {:?}", out.captured);
}
#[tokio::test]
async fn child_sees_a_real_tty() {
let s = spawn(
"if [ -t 0 ]; then echo ISATTY; fi",
&sb(),
ShellKind::Visible,
80,
24,
)
.unwrap();
let ShellSession {
input_tx,
mut events_rx,
..
} = s;
drop(input_tx);
let out = collect_outcome(&mut events_rx).await;
assert!(
out.captured.contains("ISATTY"),
"stdin should be a tty, got: {:?}",
out.captured
);
}
#[tokio::test]
async fn forwards_stdin_to_child() {
let s = spawn("cat", &sb(), ShellKind::Visible, 80, 24).unwrap();
let ShellSession {
input_tx,
mut events_rx,
..
} = s;
input_tx.send(b"ping\n".to_vec()).unwrap();
tokio::time::sleep(Duration::from_millis(150)).await;
input_tx.send(b"\x04".to_vec()).unwrap();
drop(input_tx);
let out = timeout(Duration::from_secs(5), collect_outcome(&mut events_rx))
.await
.expect("cat did not exit on EOF")
.clone();
assert!(out.captured.contains("ping"), "got: {:?}", out.captured);
assert_eq!(out.exit_code, Some(0));
}
#[tokio::test]
async fn stdin_drives_a_read_echo_command() {
let s = spawn(
"read line; echo \"you said: $line\"",
&sb(),
ShellKind::Visible,
80,
24,
)
.unwrap();
let ShellSession {
input_tx,
mut events_rx,
..
} = s;
tokio::time::sleep(Duration::from_millis(150)).await;
input_tx.send(b"world\n".to_vec()).unwrap();
tokio::time::sleep(Duration::from_millis(150)).await;
drop(input_tx);
let out = collect_outcome(&mut events_rx).await;
assert!(
out.captured.contains("you said: world"),
"got: {:?}",
out.captured
);
}
#[tokio::test]
async fn propagates_nonzero_exit_code() {
let s = spawn("exit 7", &sb(), ShellKind::Visible, 80, 24).unwrap();
let ShellSession {
input_tx,
mut events_rx,
..
} = s;
drop(input_tx);
let out = collect_outcome(&mut events_rx).await;
assert_eq!(out.exit_code, Some(7));
}
#[tokio::test]
async fn strips_ansi_escapes() {
let s = spawn(
"printf '\\033[31mred\\033[0m\\n'",
&sb(),
ShellKind::Visible,
80,
24,
)
.unwrap();
let ShellSession {
input_tx,
mut events_rx,
..
} = s;
drop(input_tx);
let out = collect_outcome(&mut events_rx).await;
assert_eq!(out.captured, "red\n", "got: {:?}", out.captured);
}
#[tokio::test]
async fn ctrl_c_delivers_sigint() {
let s = spawn(
"trap 'echo CAUGHT; exit 0' INT; sleep 5",
&sb(),
ShellKind::Visible,
80,
24,
)
.unwrap();
let ShellSession {
input_tx,
mut events_rx,
..
} = s;
tokio::time::sleep(Duration::from_millis(300)).await;
input_tx.send(b"\x03".to_vec()).unwrap();
let out = timeout(Duration::from_secs(6), collect_outcome(&mut events_rx))
.await
.expect("Ctrl+C did not terminate the child via SIGINT")
.clone();
assert!(
out.captured.contains("CAUGHT"),
"SIGINT trap should fire, got: {:?}",
out.captured
);
assert_eq!(out.exit_code, Some(0));
assert!(!out.interrupted, "Ctrl+C is graceful, not a hard kill");
}
#[tokio::test]
async fn interrupt_kills_long_running_process() {
let s = spawn("sleep 30", &sb(), ShellKind::Visible, 80, 24).unwrap();
let mut session = s;
tokio::time::sleep(Duration::from_millis(200)).await;
if let Some(tx) = session.interrupt.take() {
let _ = tx.send(());
}
drop(session.input_tx);
let out = timeout(
Duration::from_secs(5),
collect_outcome(&mut session.events_rx),
)
.await
.expect("interrupt did not kill the child")
.clone();
assert!(out.interrupted);
}
}