use std::process::Command;
use sysinfo::{Pid, ProcessRefreshKind, ProcessesToUpdate, System};
#[cfg(unix)]
const SUBMIT: char = '\r';
#[cfg(unix)]
const SETTLE: std::time::Duration = std::time::Duration::from_millis(60);
pub fn send_line(pid: u32, text: &str) -> Result<(), String> {
for backend in [shim_send, tmux_send, tiocsti_send] {
if let Some(result) = backend(pid, text) {
return result;
}
}
Err(format!(
"no way to type into session {pid}: start the agent with `cctop run <agent>` \
or inside tmux, or run cctop as root with dev.tty.legacy_tiocsti=1"
))
}
#[cfg(unix)]
fn shim_send(pid: u32, text: &str) -> Option<Result<(), String>> {
let path = crate::shim::socket_path(pid)?;
let mut stream = std::os::unix::net::UnixStream::connect(path).ok()?;
Some(write_then_submit(&mut stream, text).map_err(|e| format!("cctop run socket: {e}")))
}
#[cfg(unix)]
fn write_then_submit(out: &mut impl std::io::Write, text: &str) -> std::io::Result<()> {
out.write_all(text.as_bytes())?;
out.flush()?;
std::thread::sleep(SETTLE);
out.write_all(&[SUBMIT as u8])?;
out.flush()
}
#[cfg(not(unix))]
fn shim_send(_pid: u32, _text: &str) -> Option<Result<(), String>> {
None
}
fn tmux_send(pid: u32, text: &str) -> Option<Result<(), String>> {
let pane = pane_for(pid)?;
Some(send(&pane, text))
}
#[cfg(target_os = "linux")]
fn tiocsti_send(pid: u32, text: &str) -> Option<Result<(), String>> {
use std::os::fd::AsRawFd;
let tty = std::fs::read_link(format!("/proc/{pid}/fd/0")).ok()?;
if !tty.starts_with("/dev/pts/") {
return None;
}
let file = match std::fs::OpenOptions::new().write(true).open(&tty) {
Ok(f) => f,
Err(e) => return Some(Err(format!("{}: {e}", tty.display()))),
};
let push = |byte: u8| {
if unsafe { libc::ioctl(file.as_raw_fd(), libc::TIOCSTI as _, &byte) } != -1 {
return Ok(());
}
let err = std::io::Error::last_os_error();
Err(match err.raw_os_error() {
Some(libc::EIO) => "the kernel has TIOCSTI disabled: run cctop as root, or \
sysctl -w dev.tty.legacy_tiocsti=1"
.into(),
Some(libc::EPERM) => format!(
"typing into {} needs cctop as root (or start the agent with `cctop run`)",
tty.display()
),
_ => format!("TIOCSTI: {err}"),
})
};
for byte in text.bytes() {
if let Err(e) = push(byte) {
return Some(Err(e));
}
}
std::thread::sleep(SETTLE);
Some(push(SUBMIT as u8))
}
#[cfg(not(target_os = "linux"))]
fn tiocsti_send(_pid: u32, _text: &str) -> Option<Result<(), String>> {
None
}
const MAX_DEPTH: usize = 32;
fn pane_for(pid: u32) -> Option<String> {
let panes = list_panes()?;
let mut sys = System::new();
sys.refresh_processes_specifics(ProcessesToUpdate::All, false, ProcessRefreshKind::nothing());
let mut current = pid;
for _ in 0..MAX_DEPTH {
if let Some((_, pane)) = panes.iter().find(|(pane_pid, _)| *pane_pid == current) {
return Some(pane.clone());
}
let parent = sys.process(Pid::from_u32(current))?.parent()?.as_u32();
if parent == 0 || parent == current {
return None;
}
current = parent;
}
None
}
fn send(pane: &str, text: &str) -> Result<(), String> {
tmux(&["send-keys", "-t", pane, "-l", "--", text])?;
tmux(&["send-keys", "-t", pane, "Enter"])
}
fn list_panes() -> Option<Vec<(u32, String)>> {
let out = Command::new("tmux")
.args(["list-panes", "-a", "-F", "#{pane_pid} #{pane_id}"])
.output()
.ok()?;
if !out.status.success() {
return None;
}
Some(
String::from_utf8_lossy(&out.stdout)
.lines()
.filter_map(|line| {
let (pid, id) = line.split_once(' ')?;
Some((pid.parse().ok()?, id.to_string()))
})
.collect(),
)
}
fn tmux(args: &[&str]) -> Result<(), String> {
let out = Command::new("tmux")
.args(args)
.output()
.map_err(|e| format!("tmux: {e}"))?;
if out.status.success() {
return Ok(());
}
let err = String::from_utf8_lossy(&out.stderr).trim().to_string();
Err(if err.is_empty() {
"tmux rejected the keys".into()
} else {
err
})
}
#[cfg(test)]
mod tests {
use super::*;
#[cfg(unix)]
#[test]
fn the_submit_key_is_written_apart_from_the_line() {
use std::io::Write;
struct Writes(Vec<Vec<u8>>);
impl Write for Writes {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.0.push(buf.to_vec());
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
let mut out = Writes(Vec::new());
let start = std::time::Instant::now();
write_then_submit(&mut out, "continue").unwrap();
assert_eq!(out.0, vec![b"continue".to_vec(), vec![b'\r']]);
assert!(
start.elapsed() >= SETTLE,
"the two writes went out back to back, which is the race this avoids"
);
}
#[test]
fn types_into_the_pane_holding_a_pid() {
if Command::new("tmux").arg("-V").output().is_err() {
eprintln!("skipping: tmux not installed");
return;
}
let _turn = crate::tmux::test_lock();
let out = std::env::temp_dir().join("cctop-mux-test.txt");
let _ = std::fs::remove_file(&out);
let session = "cctop-mux-test";
let _ = Command::new("tmux")
.args(["kill-session", "-t", &format!("={session}")])
.status();
let script = format!("tee {} >/dev/null; :", out.display());
assert!(
Command::new("tmux")
.args(["new-session", "-d", "-s", session, "sh", "-c", &script])
.status()
.unwrap()
.success()
);
let reader = wait_for(|| {
let sys = {
let mut s = System::new();
s.refresh_processes_specifics(
ProcessesToUpdate::All,
true,
ProcessRefreshKind::nothing().with_cmd(sysinfo::UpdateKind::Always),
);
s
};
sys.processes()
.values()
.find(|p| {
p.name().to_string_lossy().starts_with("tee")
&& p.cmd()
.iter()
.any(|a| a.to_string_lossy().contains("cctop-mux-test.txt"))
})
.map(|p| p.pid().as_u32())
});
let pane = reader.and_then(pane_for);
let text = pane.as_ref().and_then(|pane| {
send(pane, "continue").unwrap();
wait_for(|| std::fs::read_to_string(&out).ok().filter(|t| !t.is_empty()))
});
let _ = Command::new("tmux")
.args(["kill-session", "-t", session])
.status();
let _ = std::fs::remove_file(&out);
assert!(pane.is_some(), "no pane found for the reader process");
assert_eq!(
text.expect("nothing reached the child as input").trim(),
"continue"
);
}
fn wait_for<T>(mut f: impl FnMut() -> Option<T>) -> Option<T> {
for _ in 0..50 {
if let Some(v) = f() {
return Some(v);
}
std::thread::sleep(std::time::Duration::from_millis(100));
}
None
}
}