diff --git a/rust/core/src/clients/local.rs b/rust/core/src/clients/local.rs index 7350194..7732b8d 100644 --- a/rust/core/src/clients/local.rs +++ b/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); diff --git a/rust/core/src/clients/mod.rs b/rust/core/src/clients/mod.rs index 6caa286..fad40db 100644 --- a/rust/core/src/clients/mod.rs +++ b/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>, + /// 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>, 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, 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); diff --git a/rust/core/src/clients/telnet.rs b/rust/core/src/clients/telnet.rs index d136591..1a3bd62 100644 --- a/rust/core/src/clients/telnet.rs +++ b/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, + him: std::collections::HashSet, } impl TelnetClient { @@ -56,7 +61,7 @@ impl TelnetClient { pub fn new(profile: &Profile, cb: Callbacks, on_debug: Option) -> 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_NAWS => { + OPT_TERM_TYPE | OPT_SGA | OPT_NAWS => { + if !self.state.lock().us.insert(opt) { + return; + } self.negotiate(WILL, opt); - let (cols, rows) = { - let st = self.state.lock(); - (st.cols, st.rows) - }; - let _ = self.send_naws(cols, rows); + if opt == OPT_NAWS { + let (cols, rows) = { + let st = self.state.lock(); + (st.cols, st.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(); - let [r0, r1] = rows.to_be_bytes(); - self.conn.write(&[IAC, SB, OPT_NAWS, c0, c1, r0, r1, IAC, SE]) + // 0xFF inside the subnegotiation must be doubled (RFC 1073), or a 255-column window + // sends a bare IAC the server parses as a command. + 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 = [ diff --git a/rust/core/src/clients/vnc.rs b/rust/core/src/clients/vnc.rs index a49935d..28c5bb7 100644 --- a/rust/core/src/clients/vnc.rs +++ b/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 { 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 { - 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()?; + self.pump_out()?; } } } @@ -523,21 +539,45 @@ impl Default for Decoders { } } -/// Inflates one server chunk (which ends at a sync flush) into `out`. -fn inflate(z: &mut Decompress, mut input: &[u8], out: &mut Vec) -> crate::Result<()> { +/// Inflates one server chunk (which ends at a sync flush) into `out`, refusing to grow past +/// `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, 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) { + 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]