From 2a6f28d529894be555928e855897920f1e896dbd Mon Sep 17 00:00:00 2001 From: DuProcess <273172371+DuProcess@users.noreply.github.com> Date: Sun, 14 Jun 2026 10:06:27 -0400 Subject: [PATCH] Fix native-v1 connection break, server panic, and resource leaks The native-v1 work changed the UDP wire format (HEADER_LEN 42->58) without a version bump, so a current client silently timed out against the older still-deployed server. The server just needed redeploying; alongside that, fix the robustness bugs surfaced while restoring the connection: - Server: clamp terminal size to >=1 at every PTY/parser resize site. A client reporting a 0x0 terminal made vt100::Parser::set_size panic ("attempt to subtract with overflow") and took down the whole single-threaded daemon. - Client: treat size() == Ok((0,0)) like a missing size and fall back to 80x24 instead of sending 0x0 to the server. - Reap abandoned sessions and their shells. PtyHandle now retains the child and kills it on drop, and the cleanup task removes clientless, non-prewarmed sessions after the grace period. Previously the shell leaked forever (one zsh per abandoned session); prewarmed sessions still stay hot. - Evict half-finished native handshakes (pending_native) after 30s so an unauthenticated ClientHello flood can't grow the map without bound. - Drop the per-client, per-packet vt100 Screen clone on the output hot path and remove the write-only last_screen field. - Guard load_or_create_server_secret against a <32-byte secret file (was an index-out-of-range panic at startup). - Reconcile stream flow-control credit (client and server) when StreamData hits a missing writer, so a closed/unknown stream can't wedge the peer's window. Adds a unit test for session reaping. cargo fmt + cargo test green. Co-Authored-By: Claude Opus 4.8 --- src/auth.rs | 6 ++ src/bin/dosh-client.rs | 39 ++++++-- src/bin/dosh-server.rs | 202 +++++++++++++++++++++++++++++------------ src/pty.rs | 25 ++++- 4 files changed, 201 insertions(+), 71 deletions(-) diff --git a/src/auth.rs b/src/auth.rs index 184e71c..8618f09 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -68,6 +68,12 @@ pub fn load_or_create_server_secret(config: &ServerConfig) -> Result<[u8; 32]> { .decode(String::from_utf8_lossy(&raw).trim()) .context("decode server secret")? }; + anyhow::ensure!( + decoded.len() == 32, + "server secret at {} must be 32 bytes, got {}", + path.display(), + decoded.len() + ); let mut out = [0u8; 32]; out.copy_from_slice(&decoded[..32]); return Ok(out); diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index f9526e4..e247746 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -44,6 +44,20 @@ use tokio::sync::mpsc; const STREAM_INITIAL_WINDOW: usize = 1024 * 1024; +/// Current terminal size, with a sane fallback. +/// +/// `crossterm::size()` returns `Err` when stdout is not a TTY (piped), but it +/// can also return `Ok((0, 0))` for a real PTY that has not been sized yet +/// (e.g. launched under `script` with no controlling window). Sending `0` to +/// the server is meaningless and used to crash older servers, so treat any +/// zero dimension the same as a missing size and fall back to 80x24. +fn terminal_size() -> (u16, u16) { + match size() { + Ok((cols, rows)) if cols > 0 && rows > 0 => (cols, rows), + _ => (80, 24), + } +} + #[derive(Debug, Parser)] #[command(name = "dosh-client")] struct Args { @@ -218,7 +232,7 @@ async fn main() -> Result<()> { } let dosh_port = args.dosh_port.or(host.port).unwrap_or(config.dosh_port); let cache_path = cache_path(&config.credential_cache, &requested_server, &session, &mode); - let (cols, rows) = size().unwrap_or((80, 24)); + let (cols, rows) = terminal_size(); let auth_preference = args .auth .clone() @@ -2072,7 +2086,7 @@ async fn run_terminal( let mut last_packet_at = Instant::now(); let mut status_tick = tokio::time::interval(Duration::from_secs(1)); let mut resize_tick = tokio::time::interval(Duration::from_millis(250)); - let mut last_size = size().unwrap_or((80, 24)); + let mut last_size = terminal_size(); let mut frame_buffer = FrameBuffer::default(); let mut predictor = Predictor::new(predict && cred.mode != "view-only" && !forward_only); let (forward_tx, mut forward_rx) = mpsc::channel::(1024); @@ -2151,6 +2165,8 @@ async fn run_terminal( } _ = resize_tick.tick() => { if let Ok((cols, rows)) = size() + && cols > 0 + && rows > 0 && (cols, rows) != last_size { last_size = (cols, rows); @@ -2337,17 +2353,20 @@ async fn run_terminal( continue; }; last_packet_at = Instant::now(); + let len = data.bytes.len(); if let Some(writer) = stream_writers.get_mut(&data.stream_id) { let _ = writer.write_all(&data.bytes).await; - send_stream_window_adjust( - &socket, - addr, - &cred, - &mut send_seq, - data.stream_id, - data.bytes.len(), - ).await?; } + // Always return flow-control credit so a stream whose local + // writer is already gone cannot wedge the server's send window. + send_stream_window_adjust( + &socket, + addr, + &cred, + &mut send_seq, + data.stream_id, + len, + ).await?; } PacketKind::StreamWindowAdjust => { let Ok(plain) = protocol::decrypt_body(&packet, &cred.session_key, SERVER_TO_CLIENT) else { diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index 5b43a75..4fffdc2 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -28,6 +28,9 @@ use tokio::net::{TcpListener, TcpStream, UdpSocket}; use tokio::sync::mpsc; const STREAM_INITIAL_WINDOW: usize = 1024 * 1024; +/// How long a pending native handshake (ClientHello awaiting UserAuth) is kept +/// before it is swept by the cleanup task. +const NATIVE_HANDSHAKE_TTL_SECS: u64 = 30; #[derive(Debug, Parser)] #[command(name = "dosh-server")] @@ -176,6 +179,10 @@ struct Session { clients: HashMap<[u8; 16], ClientState>, output_seq: u64, recent: VecDeque>, + /// When the session last dropped to zero clients, used to reap abandoned + /// sessions (and their shells) after a grace period. `None` while at least + /// one client is attached. + empty_since: Option, } #[derive(Clone)] @@ -190,7 +197,6 @@ struct ClientState { rows: u16, last_seen: Instant, pending: VecDeque, - last_screen: Option, allowed_forwardings: Vec, stream_writers: HashMap>>, opened_streams: HashSet, @@ -272,6 +278,7 @@ impl ServerState { clients: HashMap::new(), output_seq: 0, recent: VecDeque::with_capacity(self.config.scrollback), + empty_since: None, }, ); Ok(()) @@ -282,6 +289,27 @@ fn mode_uses_pty(mode: &str) -> bool { mode != "forward-only" } +/// Resize the PTY and terminal parser for a session. +/// +/// `cols`/`rows` come straight off the wire from the client, where a terminal +/// that has not been sized yet (or a buggy/hostile client) can report `0`. +/// `vt100::Parser::set_size(0, _)` panics with a subtract overflow and would +/// take down the whole single-threaded daemon, so clamp to at least `1` here, +/// the same way `Parser::new` is constructed in `ensure_session`. +fn apply_terminal_size( + pty: Option<&PtyHandle>, + parser: &mut vt100::Parser, + cols: u16, + rows: u16, +) -> Result<()> { + let cols = cols.max(1); + let rows = rows.max(1); + pty.context("terminal session has no pty")? + .resize(cols, rows)?; + parser.set_size(rows, cols); + Ok(()) +} + fn mode_allows_terminal_updates(mode: &str) -> bool { mode != "view-only" && mode != "forward-only" } @@ -604,16 +632,10 @@ async fn handle_native_user_auth( .get_mut(&session_name) .expect("session exists"); if mode_allows_terminal_updates(&mode) { - session - .pty - .as_ref() - .context("terminal session has no pty")? - .resize(cols, rows)?; - session.parser.set_size(rows, cols); + apply_terminal_size(session.pty.as_ref(), &mut session.parser, cols, rows)?; } let client_id = crypto::random_16(); let snapshot = screen_snapshot(session.parser.screen()); - let screen = session.parser.screen().clone(); let output_seq = session.output_seq; session.clients.insert( client_id, @@ -628,7 +650,6 @@ async fn handle_native_user_auth( rows, last_seen: Instant::now(), pending: VecDeque::new(), - last_screen: Some(screen), allowed_forwardings: req.auth.requested_forwardings.clone(), stream_writers: HashMap::new(), opened_streams: HashSet::new(), @@ -734,16 +755,15 @@ async fn handle_bootstrap_attach( .get_mut(&req.bootstrap.session) .expect("session exists"); if mode_allows_terminal_updates(&req.bootstrap.mode) { - session - .pty - .as_ref() - .context("terminal session has no pty")? - .resize(req.cols, req.rows)?; - session.parser.set_size(req.rows, req.cols); + apply_terminal_size( + session.pty.as_ref(), + &mut session.parser, + req.cols, + req.rows, + )?; } let client_id = crypto::random_16(); let snapshot = screen_snapshot(session.parser.screen()); - let screen = session.parser.screen().clone(); let output_seq = session.output_seq; session.clients.insert( client_id, @@ -758,7 +778,6 @@ async fn handle_bootstrap_attach( rows: req.rows, last_seen: Instant::now(), pending: VecDeque::new(), - last_screen: Some(screen), allowed_forwardings: Vec::new(), stream_writers: HashMap::new(), opened_streams: HashSet::new(), @@ -849,16 +868,15 @@ async fn handle_ticket_attach( .get_mut(&req.session) .expect("session exists"); if mode_allows_terminal_updates(&req.mode) { - session - .pty - .as_ref() - .context("terminal session has no pty")? - .resize(req.cols, req.rows)?; - session.parser.set_size(req.rows, req.cols); + apply_terminal_size( + session.pty.as_ref(), + &mut session.parser, + req.cols, + req.rows, + )?; } let client_id = crypto::random_16(); let snapshot = screen_snapshot(session.parser.screen()); - let screen = session.parser.screen().clone(); let output_seq = session.output_seq; session.clients.insert( client_id, @@ -873,7 +891,6 @@ async fn handle_ticket_attach( rows: req.rows, last_seen: Instant::now(), pending: VecDeque::new(), - last_screen: Some(screen), allowed_forwardings: Vec::new(), stream_writers: HashMap::new(), opened_streams: HashSet::new(), @@ -965,15 +982,14 @@ async fn handle_resume( client.last_seen = Instant::now(); client.send_seq += 1; if mode_allows_terminal_updates(&client.mode) { - session - .pty - .as_ref() - .context("terminal session has no pty")? - .resize(req.cols, req.rows)?; - session.parser.set_size(req.rows, req.cols); + apply_terminal_size( + session.pty.as_ref(), + &mut session.parser, + req.cols, + req.rows, + )?; } let snapshot = screen_snapshot(session.parser.screen()); - client.last_screen = Some(session.parser.screen().clone()); (client.send_seq, session.output_seq, snapshot) }; let frame = Frame { @@ -1056,12 +1072,12 @@ async fn handle_resize( client.endpoint = peer; client.cols = resize.cols; client.rows = resize.rows; - session - .pty - .as_ref() - .context("terminal session has no pty")? - .resize(resize.cols, resize.rows)?; - session.parser.set_size(resize.rows, resize.cols); + apply_terminal_size( + session.pty.as_ref(), + &mut session.parser, + resize.cols, + resize.rows, + )?; } Ok(()) } @@ -1280,18 +1296,15 @@ async fn handle_stream_data( client.last_seen = Instant::now(); client.stream_writers.get(&data.stream_id).cloned() }; + let len = data.bytes.len(); if let Some(writer) = writer { - let len = data.bytes.len(); let _ = writer.send(data.bytes).await; - send_stream_window_adjust_to_client( - state, - socket, - packet.header.conn_id, - data.stream_id, - len, - ) - .await?; } + // Always return flow-control credit, even when the stream's writer is already + // gone (closed/unknown). The peer debited its send window for these bytes, so + // skipping the adjust would wedge the stream at a permanent credit deficit. + send_stream_window_adjust_to_client(state, socket, packet.header.conn_id, data.stream_id, len) + .await?; Ok(()) } @@ -1847,10 +1860,12 @@ async fn broadcast_output( let mut sends = Vec::new(); for (client_id, client) in session.clients.iter_mut() { client.send_seq += 1; - let current_screen = session.parser.screen().clone(); + // Only materialize a screen snapshot when this client has fallen too + // far behind; the common case sends the raw output bytes and must not + // clone the whole vt100 grid per client per packet. let (bytes, snapshot) = if client.pending.len() >= retransmit_window { client.pending.clear(); - (screen_snapshot(¤t_screen), true) + (screen_snapshot(session.parser.screen()), true) } else { (output.bytes.clone(), false) }; @@ -1874,7 +1889,6 @@ async fn broadcast_output( while client.pending.len() >= retransmit_window { client.pending.pop_front(); } - client.last_screen = Some(current_screen); client.pending.push_back(PendingFrame { output_seq, packet: packet.clone(), @@ -1977,14 +1991,49 @@ async fn retransmit_pending( } fn cleanup_disconnected_clients(state: &Arc>) { - let mut locked = state.lock().expect("server state poisoned"); - let timeout = Duration::from_secs(locked.config.client_timeout_secs.max(1)); - let now = Instant::now(); - for session in locked.sessions.values_mut() { - session - .clients - .retain(|_, client| now.duration_since(client.last_seen) <= timeout); - } + // Sessions removed here are dropped after the lock is released so their + // shells are killed (PtyHandle::drop) without holding up the event loop. + let reaped: Vec = { + let mut locked = state.lock().expect("server state poisoned"); + let timeout = Duration::from_secs(locked.config.client_timeout_secs.max(1)); + let now = Instant::now(); + let prewarm: HashSet = locked.config.prewarm_sessions.iter().cloned().collect(); + for session in locked.sessions.values_mut() { + session + .clients + .retain(|_, client| now.duration_since(client.last_seen) <= timeout); + if session.clients.is_empty() { + session.empty_since.get_or_insert(now); + } else { + session.empty_since = None; + } + } + // Evict half-finished native handshakes. Each unauthenticated ClientHello + // inserts a PendingNativeAuth that is otherwise only removed by a matching + // UserAuth, so without this a flood of hellos grows the map without bound. + let handshake_ttl = Duration::from_secs(NATIVE_HANDSHAKE_TTL_SECS); + locked + .pending_native + .retain(|_, pending| now.duration_since(pending.created_at) < handshake_ttl); + // Reap sessions that have been clientless past the grace period. Prewarmed + // sessions stay hot on purpose, even with no clients attached. + let to_reap: Vec = locked + .sessions + .iter() + .filter(|(name, session)| { + !prewarm.contains(name.as_str()) + && session + .empty_since + .is_some_and(|since| now.duration_since(since) >= timeout) + }) + .map(|(name, _)| name.clone()) + .collect(); + to_reap + .into_iter() + .filter_map(|name| locked.sessions.remove(&name)) + .collect() + }; + drop(reaped); } fn find_client_key( @@ -2030,6 +2079,41 @@ mod tests { assert!(state.sessions["forward"].pty.is_none()); } + #[test] + fn cleanup_reaps_abandoned_sessions_but_keeps_prewarmed() { + let (pty_tx, _pty_rx) = mpsc::unbounded_channel(); + let config = ServerConfig { + client_timeout_secs: 1, + prewarm_sessions: vec!["default".to_string()], + ..ServerConfig::default() + }; + let mut state = ServerState::new(config, [0u8; 32], pty_tx); + // Both sessions are clientless; use forward-only so no real shell spawns. + state + .ensure_session("default", 80, 24, "forward-only", &[]) + .unwrap(); + state + .ensure_session("abandoned", 80, 24, "forward-only", &[]) + .unwrap(); + // Mark both as empty long enough ago to be past the grace period. + let stale = Instant::now() - Duration::from_secs(60); + state.sessions.get_mut("default").unwrap().empty_since = Some(stale); + state.sessions.get_mut("abandoned").unwrap().empty_since = Some(stale); + + let state = Arc::new(Mutex::new(state)); + cleanup_disconnected_clients(&state); + + let locked = state.lock().unwrap(); + assert!( + locked.sessions.contains_key("default"), + "prewarmed session must stay hot" + ); + assert!( + !locked.sessions.contains_key("abandoned"), + "clientless non-prewarmed session must be reaped" + ); + } + #[test] fn accepted_env_filters_by_server_patterns() { let config = ServerConfig { @@ -2107,7 +2191,6 @@ mod tests { rows: 24, last_seen: Instant::now(), pending: VecDeque::new(), - last_screen: None, allowed_forwardings: Vec::new(), stream_writers: HashMap::new(), opened_streams: HashSet::new(), @@ -2117,6 +2200,7 @@ mod tests { )]), output_seq: 0, recent: VecDeque::new(), + empty_since: None, }, ); let state = Arc::new(Mutex::new(state)); diff --git a/src/pty.rs b/src/pty.rs index e20f571..2f1361c 100644 --- a/src/pty.rs +++ b/src/pty.rs @@ -1,5 +1,5 @@ use anyhow::{Context, Result}; -use portable_pty::{CommandBuilder, MasterPty, NativePtySystem, PtySize, PtySystem}; +use portable_pty::{Child, CommandBuilder, MasterPty, NativePtySystem, PtySize, PtySystem}; use std::io::{Read, Write}; use std::path::Path; use std::sync::{Arc, Mutex}; @@ -8,6 +8,7 @@ use tokio::sync::mpsc; pub struct PtyHandle { writer: Arc>>, + child: Mutex>, _master: Box, } @@ -28,6 +29,25 @@ impl PtyHandle { })?; Ok(()) } + + /// Terminate the shell process backing this PTY and reap it. + /// + /// Without this the child shell outlives the session: dropping the master + /// alone is not guaranteed to take the process down, and the `Child` handle + /// used to be discarded at spawn time, which leaked one shell per + /// abandoned session. + pub fn kill(&self) { + if let Ok(mut child) = self.child.lock() { + let _ = child.kill(); + let _ = child.wait(); + } + } +} + +impl Drop for PtyHandle { + fn drop(&mut self) { + self.kill(); + } } #[derive(Debug)] @@ -68,7 +88,7 @@ pub fn spawn_pty_session( } else if let Some(parent) = Path::new(shell).parent() { cmd.env("PWD", parent.as_os_str()); } - let _child = pair.slave.spawn_command(cmd).context("spawn shell")?; + let child = pair.slave.spawn_command(cmd).context("spawn shell")?; drop(pair.slave); let writer = pair.master.take_writer().context("take pty writer")?; @@ -110,6 +130,7 @@ pub fn spawn_pty_session( Ok(PtyHandle { writer: Arc::new(Mutex::new(writer)), + child: Mutex::new(child), _master: pair.master, }) }