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 <noreply@anthropic.com>

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);

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::<ForwardEvent>(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,18 +2353,21 @@ 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;
+                        }
+                        // 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,
-                                data.bytes.len(),
+                            len,
                         ).await?;
                     }
-                    }
                     PacketKind::StreamWindowAdjust => {
                         let Ok(plain) = protocol::decrypt_body(&packet, &cred.session_key, SERVER_TO_CLIENT) else {
                             continue;

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<Vec<u8>>,
+    /// 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<Instant>,
 }
 
 #[derive(Clone)]
@@ -190,7 +197,6 @@ struct ClientState {
     rows: u16,
     last_seen: Instant,
     pending: VecDeque<PendingFrame>,
-    last_screen: Option<vt100::Screen>,
     allowed_forwardings: Vec<ForwardingRequest>,
     stream_writers: HashMap<u64, mpsc::Sender<Vec<u8>>>,
     opened_streams: HashSet<u64>,
@@ -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()
     };
-    if let Some(writer) = writer {
     let len = data.bytes.len();
+    if let Some(writer) = writer {
         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(&current_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<Mutex<ServerState>>) {
+    // 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<Session> = {
         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<String> = 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<String> = 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));

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<Mutex<Box<dyn Write + Send>>>,
+    child: Mutex<Box<dyn Child + Send + Sync>>,
     _master: Box<dyn MasterPty + Send>,
 }
 
@@ -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,
     })
 }