fix: bound VNC zlib inflation, telnet negotiation state, unblockable socket close

- VNC: ZRLE and Tight inflation is capped at what the rectangle can
  legitimately decode to; a ~64 KiB zlib bomb previously expanded to
  gigabytes and the OOM killer took the whole app. Large scratch buffers
  are released after use. Queued input and a local close are now handled
  after every read, so a server that streams without pause can't starve
  keystrokes or keep a closed tab's connection alive.
- Telnet: option replies are sent only when the option's state changes
  (RFC 854), ending DONT/WONT ping-pong and 1:1 reply amplification; NAWS
  doubles 0xFF so 255/511-column windows don't emit a bare IAC.
- Telnet/RAW: writes time out after 10 s and close() shuts the socket via
  a second handle without waiting for the stream lock, so a peer that
  stops reading can no longer wedge the UI thread or the tab's close.
- Local shells skip non-UTF-8 environment entries instead of panicking.

rust/core/src/clients/local.rs

@@ -60,7 +60,8 @@ impl LocalShell {
             .map_err(|e| format!("failed to start shell {:?}: {e}", self.shell))?;
         let mut cmd = CommandBuilder::new(&self.shell);
         cmd.env_clear();
-        let env = crate::terminalenv::local_shell_env(std::env::vars().map(|(k, v)| format!("{k}={v}")).collect());
+        // `vars()` panics on a non-UTF-8 entry (a launcher can plant one); skip those instead.
+        let env = crate::terminalenv::local_shell_env(std::env::vars_os().filter_map(|(k, v)| Some(format!("{}={}", k.to_str()?, v.to_str()?))).collect());
         for kv in env {
             if let Some((k, v)) = kv.split_once('=') {
                 cmd.env(k, v);

rust/core/src/clients/mod.rs

@@ -105,22 +105,31 @@ pub(crate) fn dial(host: &str, port: i64, timeout_secs: i64) -> (String, std::io
 /// reports `read error: …` then closes; `close` is idempotent and fires `on_close` once).
 pub(crate) struct TcpConn {
     stream: parking_lot::Mutex<Option<std::net::TcpStream>>,
+    /// A second handle on the socket, so `close` can shut it down without waiting for the
+    /// stream lock (a write blocked on a peer that stopped reading holds that lock).
+    shutdown: parking_lot::Mutex<Option<std::net::TcpStream>>,
     on_error: ErrorFn,
     on_close: CloseFn,
 }
 
+/// Longest a single write may block on a peer that has stopped reading; the writes happen on
+/// the UI thread (keystrokes) and on the reader thread (negotiation replies).
+const WRITE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
 impl TcpConn {
     pub(crate) fn new(on_error: ErrorFn, on_close: CloseFn) -> TcpConn {
-        TcpConn { stream: parking_lot::Mutex::new(None), on_error, on_close }
+        TcpConn { stream: parking_lot::Mutex::new(None), shutdown: parking_lot::Mutex::new(None), on_error, on_close }
     }
 
     pub(crate) fn connect(self: &Arc<Self>, stream: std::net::TcpStream, mut on_read: impl FnMut(&[u8]) + Send + 'static) -> crate::Result<()> {
         let mut reader = stream.try_clone()?;
+        let _ = stream.set_write_timeout(Some(WRITE_TIMEOUT));
         {
             let mut guard = self.stream.lock();
             if guard.is_some() {
                 return Err("already connected".into());
             }
+            *self.shutdown.lock() = stream.try_clone().ok();
             *guard = Some(stream);
         }
         let me = self.clone();
@@ -161,6 +170,9 @@ impl TcpConn {
     }
 
     pub(crate) fn close(&self) {
+        if let Some(s) = self.shutdown.lock().take() {
+            let _ = s.shutdown(std::net::Shutdown::Both);
+        }
         let taken = self.stream.lock().take();
         if let Some(s) = taken {
             let _ = s.shutdown(std::net::Shutdown::Both);

rust/core/src/clients/telnet.rs

@@ -48,6 +48,11 @@ struct State {
     local_echo: bool,
     cols: u16,
     rows: u16,
+    /// Options we have agreed to perform (WILL) and the peer has agreed to (DO), so a request
+    /// for a state we're already in isn't answered (RFC 854 forbids it; answering unconditionally
+    /// lets a hostile or naive peer drive an endless DONT/WONT ping-pong at line rate).
+    us: std::collections::HashSet<u8>,
+    him: std::collections::HashSet<u8>,
 }
 
 impl TelnetClient {
@@ -56,7 +61,7 @@ impl TelnetClient {
     pub fn new(profile: &Profile, cb: Callbacks, on_debug: Option<DebugFn>) -> TelnetClient {
         let cfg = profile.telnet.clone().unwrap_or_default();
         let encoding = if cfg.encoding.is_empty() { "utf-8".to_string() } else { cfg.encoding.clone() };
-        let state = State { local_echo: cfg.local_echo, cols: 80, rows: 24 };
+        let state = State { local_echo: cfg.local_echo, cols: 80, rows: 24, us: Default::default(), him: Default::default() };
         TelnetClient {
             inner: Arc::new(Inner {
                 conn: Arc::new(TcpConn::new(cb.on_error, cb.on_close)),
@@ -193,14 +198,18 @@ impl Inner {
     fn handle_do(&self, opt: u8) {
         self.debug(format!("received DO {opt}"));
         match opt {
-            OPT_TERM_TYPE | OPT_SGA => self.negotiate(WILL, opt),
+            OPT_TERM_TYPE | OPT_SGA | OPT_NAWS => {
-            OPT_NAWS => {
+                if !self.state.lock().us.insert(opt) {
+                    return;
+                }
                 self.negotiate(WILL, opt);
-                let (cols, rows) = {
+                if opt == OPT_NAWS {
-                    let st = self.state.lock();
+                    let (cols, rows) = {
-                    (st.cols, st.rows)
+                        let st = self.state.lock();
-                };
+                        (st.cols, st.rows)
-                let _ = self.send_naws(cols, rows);
+                    };
+                    let _ = self.send_naws(cols, rows);
+                }
             }
             _ => self.negotiate(WONT, opt),
         }
@@ -208,17 +217,23 @@ impl Inner {
 
     fn handle_dont(&self, opt: u8) {
         self.debug(format!("received DONT {opt}"));
-        self.negotiate(WONT, opt);
+        if self.state.lock().us.remove(&opt) {
+            self.negotiate(WONT, opt);
+        }
     }
 
     fn handle_will(&self, opt: u8) {
         self.debug(format!("received WILL {opt}"));
         match opt {
-            OPT_ECHO => {
+            OPT_ECHO | OPT_SGA => {
+                if !self.state.lock().him.insert(opt) {
+                    return;
+                }
                 self.negotiate(DO, opt);
-                self.state.lock().local_echo = false;
+                if opt == OPT_ECHO {
+                    self.state.lock().local_echo = false;
+                }
             }
-            OPT_SGA => self.negotiate(DO, opt),
             _ => self.negotiate(DONT, opt),
         }
     }
@@ -228,7 +243,9 @@ impl Inner {
         if opt == OPT_ECHO {
             self.state.lock().local_echo = true;
         }
-        self.negotiate(DONT, opt);
+        if self.state.lock().him.remove(&opt) {
+            self.negotiate(DONT, opt);
+        }
     }
 
     fn process_subnegotiation(&self, payload: &[u8]) {
@@ -251,9 +268,17 @@ impl Inner {
 
     fn send_naws(&self, cols: u16, rows: u16) -> crate::Result<()> {
         self.debug(format!("sending NAWS: {cols}x{rows}"));
-        let [c0, c1] = cols.to_be_bytes();
+        // 0xFF inside the subnegotiation must be doubled (RFC 1073), or a 255-column window
-        let [r0, r1] = rows.to_be_bytes();
+        // sends a bare IAC the server parses as a command.
-        self.conn.write(&[IAC, SB, OPT_NAWS, c0, c1, r0, r1, IAC, SE])
+        let mut msg = vec![IAC, SB, OPT_NAWS];
+        for b in cols.to_be_bytes().into_iter().chain(rows.to_be_bytes()) {
+            msg.push(b);
+            if b == IAC {
+                msg.push(IAC);
+            }
+        }
+        msg.extend_from_slice(&[IAC, SE]);
+        self.conn.write(&msg)
     }
 
     fn debug(&self, msg: String) {
@@ -346,6 +371,38 @@ mod tests {
         }
     }
 
+    #[test]
+    fn negotiation_replies_only_on_state_change() {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let profile = Profile { host: "127.0.0.1".into(), port: listener.local_addr().unwrap().port() as i64, ..Profile::new(crate::models::Protocol::Telnet) };
+        let client = TelnetClient::new(&profile, Callbacks::noop(), None);
+        client.connect().unwrap();
+        let (mut srv, _) = listener.accept().unwrap();
+        srv.set_read_timeout(Some(Duration::from_millis(500))).unwrap();
+        // DONT/WONT for options that are already off, and repeated WILL/DO, get no reply;
+        // only the first WILL SGA and the first DO SGA are answered.
+        let mut flood = vec![];
+        for _ in 0..100 {
+            flood.extend_from_slice(&[IAC, DONT, 1, IAC, WONT, 1, IAC, WILL, OPT_SGA, IAC, DO, OPT_SGA]);
+        }
+        srv.write_all(&flood).unwrap();
+        let mut got = vec![];
+        let mut buf = [0u8; 4096];
+        while let Ok(n) = srv.read(&mut buf) {
+            if n == 0 {
+                break;
+            }
+            got.extend_from_slice(&buf[..n]);
+        }
+        assert_eq!(got, [IAC, DO, OPT_SGA, IAC, WILL, OPT_SGA]);
+        // A 255-column window doubles the 0xFF inside the NAWS subnegotiation.
+        srv.write_all(&[IAC, DO, OPT_NAWS]).unwrap();
+        let _ = read_n(&mut srv, 3 + 9);
+        client.resize(255, 24).unwrap();
+        assert_eq!(read_n(&mut srv, 10), [IAC, SB, OPT_NAWS, 0, 255, 255, 0, 24, IAC, SE]);
+        client.close();
+    }
+
     #[test]
     fn line_endings() {
         let cases = [

rust/core/src/clients/vnc.rs

@@ -461,23 +461,39 @@ impl Conn {
     }
 }
 
+impl Conn {
+    /// Writes queued input (keys, pointer, clipboard) and notices a local close. Runs after every
+    /// read, not only on read timeouts, so a server that streams without pause can't starve
+    /// our output or keep a closed tab's connection alive.
+    fn pump_out(&mut self) -> std::io::Result<()> {
+        let mut wrote = false;
+        loop {
+            match self.out_rx.try_recv() {
+                Ok(msg) => {
+                    self.io.write_all(&msg)?;
+                    wrote = true;
+                }
+                Err(TryRecvError::Empty) => break,
+                Err(TryRecvError::Disconnected) => return Err(std::io::Error::other("closed")),
+            }
+        }
+        if wrote { self.io.flush() } else { Ok(()) }
+    }
+}
+
 impl Read for Conn {
     fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
         loop {
             match self.io.read(buf) {
                 Ok(0) => return Err(std::io::Error::new(ErrorKind::UnexpectedEof, "connection closed by server")),
-                Ok(n) => return Ok(n),
+                Ok(n) => {
+                    self.pump_out()?;
+                    return Ok(n);
+                }
                 Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut | ErrorKind::Interrupted) => {}
                 Err(e) => return Err(e),
             }
-            loop {
+            self.pump_out()?;
-                match self.out_rx.try_recv() {
-                    Ok(msg) => self.io.write_all(&msg)?,
-                    Err(TryRecvError::Empty) => break,
-                    Err(TryRecvError::Disconnected) => return Err(std::io::Error::other("closed")),
-                }
-            }
-            self.io.flush()?;
         }
     }
 }
@@ -523,21 +539,45 @@ impl Default for Decoders {
     }
 }
 
-/// Inflates one server chunk (which ends at a sync flush) into `out`.
+/// Inflates one server chunk (which ends at a sync flush) into `out`, refusing to grow past
-fn inflate(z: &mut Decompress, mut input: &[u8], out: &mut Vec<u8>) -> crate::Result<()> {
+/// `max_out`: the caller knows how many bytes a rectangle can legitimately need, and a zlib bomb
+/// would otherwise expand a few hundred KiB from the wire into gigabytes.
+fn inflate(z: &mut Decompress, mut input: &[u8], out: &mut Vec<u8>, max_out: usize) -> crate::Result<()> {
     out.clear();
     loop {
-        out.reserve(64 << 10);
+        if out.len() > max_out {
+            return Err(format!("compressed rectangle inflates past {max_out} bytes").into());
+        }
+        // Never ask for more room than the cap allows (plus one byte to detect overflow).
+        out.reserve((64 << 10).min(max_out + 1 - out.len()));
         let (i0, o0) = (z.total_in(), z.total_out());
         z.decompress_vec(input, out, FlushDecompress::Sync).map_err(|e| format!("zlib: {e}"))?;
         input = &input[(z.total_in() - i0) as usize..];
         let stalled = z.total_in() == i0 && z.total_out() == o0;
+        if out.len() > max_out {
+            return Err(format!("compressed rectangle inflates past {max_out} bytes").into());
+        }
         if (input.is_empty() && out.len() < out.capacity()) || stalled {
             return Ok(());
         }
     }
 }
 
+/// A rectangle's decoded size can't exceed this, whatever the encoding: raw pixels plus, per
+/// 64×64 ZRLE tile, a subencoding byte and a 128-entry palette.
+fn max_inflated(w: usize, h: usize) -> usize {
+    let tiles = w.div_ceil(64) * h.div_ceil(64);
+    w * h * 4 + tiles * (1 + 128 * 4)
+}
+
+/// Returns a decode scratch buffer to a modest size after a big rectangle, so one large update
+/// doesn't pin hundreds of MiB for the rest of the session.
+fn shrink_scratch(buf: &mut Vec<u8>) {
+    if buf.capacity() > (8 << 20) {
+        *buf = Vec::new();
+    }
+}
+
 /// Reads one FramebufferUpdate (after the type byte) into `fb`. Returns the changed row range and
 /// whether the desktop was resized.
 fn read_update(r: &mut impl Read, fb: &mut Framebuffer, dec: &mut Decoders) -> crate::Result<(usize, usize, bool)> {
@@ -592,8 +632,9 @@ fn read_update(r: &mut impl Read, fb: &mut Framebuffer, dec: &mut Decoders) -> c
                     return Err(format!("ZRLE rectangle of {len} bytes exceeds {MAX_ZRLE_BYTES}").into());
                 }
                 let data = read_n(r, len)?;
-                inflate(&mut dec.zrle, &data, &mut dec.buf)?;
+                inflate(&mut dec.zrle, &data, &mut dec.buf, max_inflated(w as usize, ht as usize))?;
                 zrle(&dec.buf, fb, x, y, w, ht)?;
+                shrink_scratch(&mut dec.buf);
             }
             e => return Err(format!("unsupported encoding {e}").into()),
         }
@@ -770,7 +811,7 @@ fn tight(r: &mut impl Read, fb: &mut Framebuffer, dec: &mut Decoders, x: u16, y:
     } else {
         let len = compact_len(r)?;
         let compressed = read_n(r, len)?;
-        inflate(&mut dec.tight[(ctl >> 4 & 3) as usize], &compressed, &mut dec.buf)?;
+        inflate(&mut dec.tight[(ctl >> 4 & 3) as usize], &compressed, &mut dec.buf, size)?;
         std::mem::take(&mut dec.buf)
     };
     if data.len() < size {
@@ -877,6 +918,25 @@ pub fn fuzz_updates(data: &[u8]) {
 #[cfg(test)]
 mod tests {
     use super::*;
+
+    #[test]
+    fn inflate_refuses_zlib_bombs() {
+        use flate2::{Compress, Compression, FlushCompress};
+        // 64 MiB of zeros compresses to ~64 KiB; a 4x4 rectangle may only need a few hundred bytes.
+        let mut z = Compress::new(Compression::best(), true);
+        let mut wire = Vec::with_capacity(1 << 20);
+        z.compress_vec(&vec![0u8; 64 << 20], &mut wire, FlushCompress::Sync).unwrap();
+        let (mut d, mut out) = (Decompress::new(true), Vec::new());
+        assert!(inflate(&mut d, &wire, &mut out, max_inflated(4, 4)).is_err());
+        assert!(out.capacity() < 1 << 20, "never grew far past the cap: {}", out.capacity());
+        // A legitimate rectangle still inflates.
+        let mut z = Compress::new(Compression::fast(), true);
+        let mut wire = Vec::with_capacity(4096);
+        z.compress_vec(&[7u8; 64], &mut wire, FlushCompress::Sync).unwrap();
+        let (mut d, mut out) = (Decompress::new(true), Vec::new());
+        inflate(&mut d, &wire, &mut out, max_inflated(4, 4)).unwrap();
+        assert_eq!(out, [7u8; 64]);
+    }
     use std::net::TcpListener;
 
     #[test]