//! Session persistence across server restarts. //! //! A persistent dosh session's shell does not run as a child of `dosh-server`. //! Instead the server spawns a tiny per-session *holder* process (the same //! `dosh-server` binary re-exec'd as `dosh-server hold ...`). The holder calls //! `setsid()` to leave the server's process group/session, opens a PTY, spawns //! the shell as ITS own child, and listens on a Unix socket in a per-session //! runtime directory. The server connects to that socket and the holder hands it //! the PTY master fd over `SCM_RIGHTS`. //! //! Because the shell belongs to the holder (which is in its own session and is //! not waited on by the server), killing or restarting `dosh-server` leaves the //! holder + shell alive. On startup the server scans the runtime directory, //! reconnects to each live holder, receives the master fd again, and rebuilds the //! in-memory session so clients reattach to the very same shell. //! //! Screen / scrollback state lives only in server memory, so it is also mirrored //! to disk (atomically) and restored on re-adoption, letting a reattaching client //! repaint the screen exactly as it was before the restart. use anyhow::{Context, Result, anyhow, bail}; use std::io::{Read, Write}; use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, RawFd}; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Path, PathBuf}; use std::sync::Mutex; use std::time::Duration; /// One-byte commands a server sends to a holder over its control socket after /// the holder has handed back the master fd. const HOLDER_CMD_SHUTDOWN: u8 = b'X'; const HOLDER_STARTUP_TIMEOUT: Duration = Duration::from_millis(750); static SCREEN_WRITE_LOCK: Mutex<()> = Mutex::new(()); /// Magic written into a holder's `meta` file, bumped if the on-disk layout /// changes so a stale holder from an incompatible build is ignored. const META_MAGIC: &str = "dosh-holder-1"; /// Per-session runtime metadata persisted next to the holder socket. #[derive(Debug, Clone)] pub struct HolderMeta { pub session: String, pub shell_pid: i32, } /// Root runtime directory holding one subdirectory per persistent session. /// Lives under `sessions_dir/run` so it shares the session storage location and /// can be wiped with it. pub fn runtime_root(sessions_dir: &Path) -> PathBuf { sessions_dir.join("run") } /// Map a session name to a filesystem-safe directory name. Session names are /// user-controlled, so hex-encode them rather than trusting them as path /// components (avoids traversal, slashes, NULs, length issues). fn session_dir_name(session: &str) -> String { let mut out = String::with_capacity(session.len() * 2); for byte in session.as_bytes() { out.push(char::from_digit((byte >> 4) as u32, 16).unwrap()); out.push(char::from_digit((byte & 0xf) as u32, 16).unwrap()); } out } fn decode_session_dir_name(name: &str) -> Option { if !name.len().is_multiple_of(2) { return None; } let mut bytes = Vec::with_capacity(name.len() / 2); let raw = name.as_bytes(); let mut i = 0; while i < raw.len() { let hi = (raw[i] as char).to_digit(16)?; let lo = (raw[i + 1] as char).to_digit(16)?; bytes.push(((hi << 4) | lo) as u8); i += 2; } String::from_utf8(bytes).ok() } /// Directory for one session's holder runtime state. pub fn session_runtime_dir(sessions_dir: &Path, session: &str) -> PathBuf { runtime_root(sessions_dir).join(session_dir_name(session)) } fn holder_sock_path(dir: &Path) -> PathBuf { dir.join("holder.sock") } fn meta_path(dir: &Path) -> PathBuf { dir.join("meta") } fn screen_path(dir: &Path) -> PathBuf { dir.join("screen") } /// Create the runtime directory tree with private (0700) permissions. pub fn ensure_runtime_dir(sessions_dir: &Path, session: &str) -> Result { use std::os::unix::fs::PermissionsExt; let root = runtime_root(sessions_dir); std::fs::create_dir_all(&root).with_context(|| format!("create {}", root.display()))?; let _ = std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o700)); let dir = session_runtime_dir(sessions_dir, session); std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?; std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)) .with_context(|| format!("chmod {}", dir.display()))?; Ok(dir) } /// Atomically write `data` to `path` (write temp + rename), so a concurrent /// reader (e.g. a restarting server) never sees a half-written file. fn atomic_write(path: &Path, data: &[u8]) -> Result<()> { use std::os::unix::fs::PermissionsExt; let tmp = path.with_extension("tmp"); { let mut file = std::fs::File::create(&tmp).with_context(|| format!("create {}", tmp.display()))?; let _ = file.set_permissions(std::fs::Permissions::from_mode(0o600)); file.write_all(data) .with_context(|| format!("write {}", tmp.display()))?; file.sync_all().ok(); } std::fs::rename(&tmp, path) .with_context(|| format!("rename {} -> {}", tmp.display(), path.display()))?; Ok(()) } /// Persist the vt100 screen snapshot plus recent raw output for a session so a /// post-restart reattach repaints correctly. `snapshot` is the bytes a fresh /// attach would receive (alt-screen toggle + `state_formatted`). pub fn save_screen( sessions_dir: &Path, session: &str, cols: u16, rows: u16, output_seq: u64, snapshot: &[u8], ) -> Result<()> { let dir = session_runtime_dir(sessions_dir, session); if !dir.exists() { // No holder runtime for this session (non-persistent): nothing to do. return Ok(()); } let _guard = SCREEN_WRITE_LOCK .lock() .map_err(|_| anyhow!("screen persistence lock poisoned"))?; let path = screen_path(&dir); if existing_screen_output_seq(&path).is_some_and(|existing| existing >= output_seq) { return Ok(()); } let mut buf = Vec::with_capacity(snapshot.len() + 32); buf.extend_from_slice(&cols.to_be_bytes()); buf.extend_from_slice(&rows.to_be_bytes()); buf.extend_from_slice(&output_seq.to_be_bytes()); buf.extend_from_slice(&(snapshot.len() as u32).to_be_bytes()); buf.extend_from_slice(snapshot); atomic_write(&path, &buf) } fn existing_screen_output_seq(path: &Path) -> Option { let data = std::fs::read(path).ok()?; if data.len() < 12 { return None; } Some(u64::from_be_bytes(data[4..12].try_into().ok()?)) } /// A restored screen for a re-adopted session. pub struct SavedScreen { pub cols: u16, pub rows: u16, pub output_seq: u64, pub snapshot: Vec, } /// Load a previously persisted screen, if any. pub fn load_screen(sessions_dir: &Path, session: &str) -> Option { let dir = session_runtime_dir(sessions_dir, session); let data = std::fs::read(screen_path(&dir)).ok()?; if data.len() < 16 { return None; } let cols = u16::from_be_bytes(data[0..2].try_into().ok()?); let rows = u16::from_be_bytes(data[2..4].try_into().ok()?); let output_seq = u64::from_be_bytes(data[4..12].try_into().ok()?); let len = u32::from_be_bytes(data[12..16].try_into().ok()?) as usize; if data.len() < 16 + len { return None; } Some(SavedScreen { cols, rows, output_seq, snapshot: data[16..16 + len].to_vec(), }) } fn write_meta(dir: &Path, meta: &HolderMeta) -> Result<()> { let body = format!("{META_MAGIC}\n{}\n{}\n", meta.shell_pid, meta.session); atomic_write(&meta_path(dir), body.as_bytes()) } fn read_meta(dir: &Path) -> Option { let body = std::fs::read_to_string(meta_path(dir)).ok()?; let mut lines = body.lines(); if lines.next()? != META_MAGIC { return None; } let shell_pid: i32 = lines.next()?.parse().ok()?; let session = lines.next()?.to_string(); Some(HolderMeta { session, shell_pid }) } /// Spawn a holder process for `session`: re-exec this binary as `dosh-server /// hold` with the runtime dir, shell, terminal size, and accepted env. The /// holder daemonizes (setsid, owns the PTY) and listens on its control socket. /// Returns once the holder has signalled readiness by creating its socket. #[allow(clippy::too_many_arguments)] pub fn spawn_holder( sessions_dir: &Path, session: &str, shell: &str, cols: u16, rows: u16, env: &[(String, String)], ) -> Result<()> { let dir = ensure_runtime_dir(sessions_dir, session)?; let sock = holder_sock_path(&dir); // Clear any stale socket left by a crashed holder with the same name. let _ = std::fs::remove_file(&sock); let exe = std::env::current_exe().context("locate current executable")?; let mut cmd = std::process::Command::new(exe); cmd.arg("hold") .arg("--runtime-dir") .arg(&dir) .arg("--session") .arg(session) .arg("--shell") .arg(shell) .arg("--cols") .arg(cols.to_string()) .arg("--rows") .arg(rows.to_string()); for (name, value) in env { cmd.arg("--env").arg(format!("{name}={value}")); } cmd.stdin(std::process::Stdio::null()) .stdout(std::process::Stdio::null()) .stderr(std::process::Stdio::null()); let mut child = cmd.spawn().context("spawn holder process")?; // Wait (briefly) for the holder to come up. The holder forks/setsids before // creating the socket; once the socket exists we can connect. We also reap // the short-lived launcher child so it never becomes a zombie. let deadline = std::time::Instant::now() + HOLDER_STARTUP_TIMEOUT; loop { if sock.exists() { break; } if let Some(status) = child.try_wait().context("wait holder launcher")? { bail!("holder launcher for session {session} exited before socket was ready: {status}"); } if std::time::Instant::now() >= deadline { let _ = child.kill(); let _ = child.wait(); bail!("holder for session {session} did not come up in time"); } std::thread::sleep(Duration::from_millis(20)); } // The launcher exits immediately after the grandchild setsids; reap it. let _ = child.wait(); Ok(()) } /// Connect to a session's holder and receive the PTY master fd over SCM_RIGHTS. /// Returns the raw fd (caller owns it) and the holder control socket, kept open /// so the server can later ask the holder to shut down. Returns an error if the /// holder is gone (caller then degrades to a fresh, non-persistent session). pub fn adopt_holder(sessions_dir: &Path, session: &str) -> Result<(RawFd, UnixStream)> { let dir = session_runtime_dir(sessions_dir, session); let sock = holder_sock_path(&dir); let mut stream = UnixStream::connect(&sock) .with_context(|| format!("connect holder socket {}", sock.display()))?; stream.set_read_timeout(Some(Duration::from_secs(5))).ok(); let fd = recv_fd(&mut stream).context("receive master fd from holder")?; Ok((fd, stream)) } /// Ask a holder to terminate its shell and exit, then clean its runtime dir. /// Used when reaping a truly-abandoned persistent session. pub fn request_shutdown(sessions_dir: &Path, session: &str, control: Option<&mut UnixStream>) { if let Some(stream) = control { let _ = stream.write_all(&[HOLDER_CMD_SHUTDOWN]); let _ = stream.flush(); } else { let dir = session_runtime_dir(sessions_dir, session); if let Ok(mut stream) = UnixStream::connect(holder_sock_path(&dir)) { // Drain the fd the holder sends on connect, then send shutdown. let _ = recv_fd(&mut stream); let _ = stream.write_all(&[HOLDER_CMD_SHUTDOWN]); let _ = stream.flush(); } } // Give the holder a moment to tear down, then remove its runtime dir. std::thread::sleep(Duration::from_millis(50)); let dir = session_runtime_dir(sessions_dir, session); let _ = std::fs::remove_dir_all(&dir); } /// Remove a session's runtime directory unconditionally (best effort). pub fn remove_runtime_dir(sessions_dir: &Path, session: &str) { let dir = session_runtime_dir(sessions_dir, session); let _ = std::fs::remove_dir_all(&dir); } /// Scan the runtime root for holders left behind by a previous server. Returns /// `(session_name, meta)` for each one whose holder process still appears alive. /// Stale entries (no live process) are cleaned up. pub fn scan_existing_holders(sessions_dir: &Path) -> Vec { let root = runtime_root(sessions_dir); let mut found = Vec::new(); let Ok(entries) = std::fs::read_dir(&root) else { return found; }; for entry in entries.flatten() { let dir = entry.path(); if !dir.is_dir() { continue; } let Some(name) = dir.file_name().and_then(|n| n.to_str()) else { continue; }; // Decode the on-disk name back to a session name; skip junk dirs. if decode_session_dir_name(name).is_none() { continue; } match read_meta(&dir) { Some(meta) if process_alive(meta.shell_pid) && holder_sock_path(&dir).exists() => { found.push(meta); } _ => { // Holder gone or meta unreadable: clean up so we don't try to // adopt a dead session (degrade to fresh on next attach). let _ = std::fs::remove_dir_all(&dir); } } } found } /// Whether a pid refers to a live process (signal 0 probe). fn process_alive(pid: i32) -> bool { if pid <= 0 { return false; } unsafe { libc::kill(pid, 0) == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) } } // --------------------------------------------------------------------------- // Holder process entry point // --------------------------------------------------------------------------- /// Run as a holder process. Daemonizes (double-fork + setsid), opens a PTY, /// spawns the shell as its own child, and serves the control socket: every /// accepted connection is handed the master fd (SCM_RIGHTS); a subsequent /// SHUTDOWN byte (or the shell exiting) tears everything down. /// /// This function does not return on success — it `exit`s the process. Errors /// before the daemonization point are returned to the launcher. #[allow(clippy::too_many_arguments)] pub fn run_holder( runtime_dir: &Path, session: &str, shell: &str, cols: u16, rows: u16, env: &[(String, String)], ) -> Result<()> { use portable_pty::{NativePtySystem, PtySize, PtySystem}; // Detach from the launching server: own session + process group so a server // exit (even a process-group kill of the service) does not take us down. daemonize().context("daemonize holder")?; let pty_system = NativePtySystem::default(); let pair = pty_system .openpty(PtySize { rows, cols, pixel_width: 0, pixel_height: 0, }) .context("holder open pty")?; let cmd = crate::pty::build_shell_command(shell, env); let mut child = pair .slave .spawn_command(cmd) .context("holder spawn shell")?; drop(pair.slave); let master_fd = pair .master .as_raw_fd() .ok_or_else(|| anyhow!("holder master has no raw fd"))?; let shell_pid = child.process_id().map(|p| p as i32).unwrap_or(-1); write_meta( runtime_dir, &HolderMeta { session: session.to_string(), shell_pid, }, )?; let sock = holder_sock_path(runtime_dir); let _ = std::fs::remove_file(&sock); let listener = UnixListener::bind(&sock).with_context(|| format!("holder bind {}", sock.display()))?; { use std::os::unix::fs::PermissionsExt; let _ = std::fs::set_permissions(&sock, std::fs::Permissions::from_mode(0o600)); } listener .set_nonblocking(false) .context("holder listener blocking")?; // A watcher thread reaps the shell: when it exits, the holder cleans up and // exits too, so an abandoned shell does not linger forever. let runtime_owned = runtime_dir.to_path_buf(); std::thread::spawn(move || { let _ = child.wait(); // Shell exited: remove runtime dir and exit the holder process. let _ = std::fs::remove_dir_all(&runtime_owned); std::process::exit(0); }); // Accept loop: each connecting server gets the master fd; a SHUTDOWN byte // from any of them tears the holder down. Multiple servers never run at once // in practice (one service), but serving repeated connections lets a server // restart re-adopt cleanly. for stream in listener.incoming() { let Ok(mut stream) = stream else { continue }; if send_fd(&mut stream, master_fd).is_err() { continue; } // Wait for an optional command byte. EOF (server dropped the control // socket on its own exit) just means "keep running, await re-adoption". let mut cmd = [0u8; 1]; match stream.read(&mut cmd) { Ok(1) if cmd[0] == HOLDER_CMD_SHUTDOWN => { if shell_pid > 0 { let _ = unsafe { libc::kill(shell_pid, libc::SIGHUP) }; let _ = unsafe { libc::kill(shell_pid, libc::SIGTERM) }; } let _ = std::fs::remove_dir_all(runtime_dir); std::process::exit(0); } _ => { // Server detached (exit/restart) or sent nothing actionable: // loop back and wait for the next server to re-adopt us. } } } Ok(()) } /// Double-fork + `setsid` so the holder runs in its own session, detached from /// the server's controlling terminal and process group. Without this a /// `systemctl restart` (which signals the whole service cgroup/process group) /// could take the holder down with the server. fn daemonize() -> Result<()> { // First fork: parent (launcher) returns to reap; child continues. match unsafe { libc::fork() } { -1 => bail!("fork: {}", std::io::Error::last_os_error()), 0 => {} _ => { // Parent process: exit so the launcher's wait() returns promptly and // the grandchild is reparented to init. std::process::exit(0); } } // New session: detaches from controlling tty and the server's process group. if unsafe { libc::setsid() } == -1 { bail!("setsid: {}", std::io::Error::last_os_error()); } // Second fork: ensures we are not a session leader, so we can never // re-acquire a controlling terminal. match unsafe { libc::fork() } { -1 => bail!("fork2: {}", std::io::Error::last_os_error()), 0 => Ok(()), _ => std::process::exit(0), } } // --------------------------------------------------------------------------- // SCM_RIGHTS file-descriptor passing over a Unix domain socket // --------------------------------------------------------------------------- /// Send a single fd over `stream` using SCM_RIGHTS, with one byte of normal data /// (sendmsg requires at least one iovec byte for the ancillary data to ride on). fn send_fd(stream: &mut UnixStream, fd: RawFd) -> Result<()> { let dummy: [u8; 1] = [0]; let mut iov = libc::iovec { iov_base: dummy.as_ptr() as *mut libc::c_void, iov_len: 1, }; let mut cmsg_buf = [0u8; cmsg_space_one_fd()]; let mut msg: libc::msghdr = unsafe { std::mem::zeroed() }; msg.msg_iov = &mut iov; msg.msg_iovlen = 1; msg.msg_control = cmsg_buf.as_mut_ptr() as *mut libc::c_void; msg.msg_controllen = cmsg_buf.len() as _; unsafe { let cmsg = libc::CMSG_FIRSTHDR(&msg); if cmsg.is_null() { bail!("CMSG_FIRSTHDR null"); } (*cmsg).cmsg_level = libc::SOL_SOCKET; (*cmsg).cmsg_type = libc::SCM_RIGHTS; (*cmsg).cmsg_len = libc::CMSG_LEN(std::mem::size_of::() as u32) as _; std::ptr::copy_nonoverlapping( &fd as *const RawFd as *const u8, libc::CMSG_DATA(cmsg), std::mem::size_of::(), ); let n = libc::sendmsg(stream.as_raw_fd(), &msg, 0); if n < 0 { return Err(std::io::Error::last_os_error()).context("sendmsg SCM_RIGHTS"); } } Ok(()) } /// Receive a single fd sent via SCM_RIGHTS. Returns a fresh fd owned by the /// caller. fn recv_fd(stream: &mut UnixStream) -> Result { let mut dummy = [0u8; 1]; let mut iov = libc::iovec { iov_base: dummy.as_mut_ptr() as *mut libc::c_void, iov_len: 1, }; let mut cmsg_buf = [0u8; cmsg_space_one_fd()]; let mut msg: libc::msghdr = unsafe { std::mem::zeroed() }; msg.msg_iov = &mut iov; msg.msg_iovlen = 1; msg.msg_control = cmsg_buf.as_mut_ptr() as *mut libc::c_void; msg.msg_controllen = cmsg_buf.len() as _; unsafe { let n = libc::recvmsg(stream.as_raw_fd(), &mut msg, 0); if n < 0 { return Err(std::io::Error::last_os_error()).context("recvmsg SCM_RIGHTS"); } if n == 0 { bail!("holder closed connection before sending fd"); } let cmsg = libc::CMSG_FIRSTHDR(&msg); if cmsg.is_null() { bail!("no ancillary data (fd) received from holder"); } if (*cmsg).cmsg_level != libc::SOL_SOCKET || (*cmsg).cmsg_type != libc::SCM_RIGHTS { bail!("unexpected ancillary message from holder"); } let mut fd: RawFd = -1; std::ptr::copy_nonoverlapping( libc::CMSG_DATA(cmsg), &mut fd as *mut RawFd as *mut u8, std::mem::size_of::(), ); if fd < 0 { bail!("invalid fd received from holder"); } Ok(fd) } } /// Space, in bytes, needed for a control-message buffer carrying exactly one fd. const fn cmsg_space_one_fd() -> usize { // CMSG_SPACE is not const in libc; this is the equivalent for one RawFd. // cmsghdr is aligned to size_of::(); add data length rounded up. let data = std::mem::size_of::(); let hdr = std::mem::size_of::(); let align = std::mem::size_of::(); // round(hdr) + round(data) ((hdr + align - 1) & !(align - 1)) + ((data + align - 1) & !(align - 1)) } /// Helper used by `adopt_holder` callers to turn the received raw fd into an /// owned `File`-like object if they need RAII (the server hands it to /// `pty::adopt_pty_from_fd`, which takes ownership of the raw fd). pub fn fd_into_file(fd: RawFd) -> std::fs::File { unsafe { std::fs::File::from_raw_fd(fd) } } /// Convert a `File` back into a raw fd it no longer owns (so it can be handed to /// `adopt_pty_from_fd`). Currently unused outside tests but kept symmetric. pub fn file_into_fd(file: std::fs::File) -> RawFd { file.into_raw_fd() } #[cfg(test)] mod tests { use super::*; #[test] fn session_dir_name_round_trips() { for name in ["default", "work", "a/b/../c", "weird name", "日本語"] { let encoded = session_dir_name(name); assert!(encoded.bytes().all(|b| b.is_ascii_hexdigit())); assert_eq!(decode_session_dir_name(&encoded).as_deref(), Some(name)); } } #[test] fn decode_rejects_non_hex() { assert!(decode_session_dir_name("zz").is_none()); assert!(decode_session_dir_name("abc").is_none()); } #[test] fn save_and_load_screen_round_trips() { let tmp = tempfile::tempdir().unwrap(); let sessions_dir = tmp.path(); ensure_runtime_dir(sessions_dir, "work").unwrap(); let snap = b"\x1b[?1049lhello world".to_vec(); save_screen(sessions_dir, "work", 100, 40, 7, &snap).unwrap(); let loaded = load_screen(sessions_dir, "work").expect("screen restored"); assert_eq!(loaded.cols, 100); assert_eq!(loaded.rows, 40); assert_eq!(loaded.output_seq, 7); assert_eq!(loaded.snapshot, snap); } #[test] fn save_screen_never_overwrites_newer_snapshot_with_older_seq() { let tmp = tempfile::tempdir().unwrap(); let sessions_dir = tmp.path(); ensure_runtime_dir(sessions_dir, "work").unwrap(); save_screen(sessions_dir, "work", 80, 24, 10, b"new-screen").unwrap(); save_screen(sessions_dir, "work", 80, 24, 9, b"old-screen").unwrap(); let loaded = load_screen(sessions_dir, "work").expect("screen restored"); assert_eq!(loaded.output_seq, 10); assert_eq!(loaded.snapshot, b"new-screen"); } #[test] fn load_screen_absent_is_none() { let tmp = tempfile::tempdir().unwrap(); assert!(load_screen(tmp.path(), "missing").is_none()); } #[test] fn scan_skips_dead_and_junk_entries() { let tmp = tempfile::tempdir().unwrap(); let sessions_dir = tmp.path(); // A meta pointing at a definitely-dead pid is cleaned up, not returned. let dir = ensure_runtime_dir(sessions_dir, "ghost").unwrap(); write_meta( &dir, &HolderMeta { session: "ghost".to_string(), shell_pid: 2_000_000_000, // not a live pid }, ) .unwrap(); // A junk (non-hex) directory is ignored. std::fs::create_dir_all(runtime_root(sessions_dir).join("not-hex")).unwrap(); let found = scan_existing_holders(sessions_dir); assert!(found.is_empty(), "dead/junk holders must be skipped"); assert!(!dir.exists(), "dead holder dir should be cleaned up"); } #[test] fn send_and_recv_fd_round_trips() { use std::io::Seek; // Pass a temp file's fd across a socketpair and confirm both ends point // at the same open file (write via one, read via the other). let (mut a, mut b) = UnixStream::pair().unwrap(); let mut file = tempfile::tempfile().unwrap(); writeln!(file, "shared-fd-marker").unwrap(); file.flush().unwrap(); send_fd(&mut a, file.as_raw_fd()).unwrap(); let received = recv_fd(&mut b).unwrap(); let mut got = fd_into_file(received); got.rewind().unwrap(); let mut contents = String::new(); got.read_to_string(&mut contents).unwrap(); assert!(contents.contains("shared-fd-marker")); } }