russh: Post-decompression SSH packet size was not bounded, allowing remote oversized compressed packets

@@ -363,6 +363,9 @@ const MAXIMUM_PADDING_LEN: usize = 19;
 const MAXIMUM_PACKET_LEN_HEADROOM: usize =
     PADDING_LENGTH_LEN + CHANNEL_EXTENDED_DATA_PACKET_OVERHEAD + MAXIMUM_PADDING_LEN;
 const MAXIMUM_PACKET_LEN: usize = MAXIMUM_PACKET_LEN_BASELINE + MAXIMUM_PACKET_LEN_HEADROOM;
+// Keep post-decompression growth within the same packet-acceptance model as
+// the transport read path.
+pub(crate) const MAXIMUM_DECOMPRESSED_PACKET_LEN: usize = MAXIMUM_PACKET_LEN;
 
 #[cfg(feature = "_bench")]
 pub mod benchmark;

@@ -1666,7 +1666,9 @@ async fn reply<H: Handler>(
 
 #[cfg(test)]
 mod tests {
-    use std::collections::HashMap;
+    use std::collections::{HashMap, VecDeque};
+    #[cfg(feature = "flate2")]
+    use std::io::Write;
     use std::num::Wrapping;
     use std::sync::Arc;
 
@@ -1679,6 +1681,7 @@ mod tests {
     use crate::compression::{Compression, Decompress};
     use crate::kex::{KEXES, NONE};
     use crate::session::{CommonSession, Encrypted, EncryptedState, Exchange};
+    use crate::sshbuffer::{IncomingSshPacket, PacketWriter, SSHBuffer};
     use crate::{CryptoVec, cipher, mac};
 
     struct TestHandler;
@@ -1750,6 +1753,56 @@ mod tests {
         (session, sender, reply_receiver)
     }
 
+    #[cfg(feature = "flate2")]
+    fn authenticated_session() -> Session {
+        let config = Arc::new(Config::default());
+        let (receiver_sender, receiver) = channel(config.channel_buffer_size);
+        let (reply_sender, _) = unbounded_channel();
+        let mut session = Session::new(
+            config.window_size,
+            CommonSession {
+                auth_user: String::new(),
+                auth_attempts: 0,
+                auth_method: None,
+                remote_to_local: Box::new(cipher::clear::Key),
+                encrypted: Some(Encrypted {
+                    state: EncryptedState::Authenticated,
+                    exchange: Some(Exchange::default()),
+                    kex: KEXES.get(&NONE).unwrap().make(),
+                    key: 0,
+                    client_mac: mac::NONE,
+                    server_mac: mac::NONE,
+                    session_id: CryptoVec::new(),
+                    channels: HashMap::new(),
+                    last_channel_id: Wrapping(0),
+                    write: Vec::new(),
+                    write_cursor: 0,
+                    last_rekey: russh_util::time::Instant::now(),
+                    server_compression: Compression::None,
+                    client_compression: Compression::None,
+                    decompress: Decompress::Zlib(flate2::Decompress::new(true)),
+                    rekey_wanted: false,
+                    received_extensions: Vec::new(),
+                    extension_info_awaiters: HashMap::new(),
+                }),
+                config,
+                wants_reply: false,
+                disconnected: false,
+                buffer: Vec::new(),
+                strict_kex: false,
+                alive_timeouts: 0,
+                received_data: false,
+                remote_sshid: b"SSH-2.0-test".to_vec(),
+                packet_writer: PacketWriter::clear(),
+            },
+            receiver,
+            reply_sender,
+        );
+        session.open_global_requests = VecDeque::new();
+        let _ = receiver_sender;
+        session
+    }
+
     fn oversized_prompt_count_packet() -> Vec<u8> {
         let mut packet = Vec::new();
         msg::USERAUTH_INFO_REQUEST_OR_USERAUTH_PK_OK
@@ -1772,7 +1825,59 @@ mod tests {
             .expect_err("malformed prompt count must fail");
 
         assert!(matches!(err, crate::Error::Inconsistent));
-        assert!(replies.try_recv().is_err(), "malformed packet must not emit a reply");
+        assert!(
+            replies.try_recv().is_err(),
+            "malformed packet must not emit a reply"
+        );
+    }
+
+    #[cfg(feature = "flate2")]
+    fn compressed_debug_payload(payload_len: usize) -> Vec<u8> {
+        let mut payload = vec![b'A'; payload_len];
+        payload[0] = crate::msg::DEBUG;
+
+        let mut encoder = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::best());
+        encoder.write_all(&payload).unwrap();
+        let compressed = encoder.finish().unwrap();
+        assert!(compressed.len() < 256 * 1024);
+        compressed
+    }
+
+    #[cfg(feature = "flate2")]
+    fn incoming_packet(compressed: Vec<u8>) -> SSHBuffer {
+        let mut buffer = SSHBuffer::new();
+        buffer.buffer.extend_from_slice(&[0; 5]);
+        buffer.buffer.extend_from_slice(&compressed);
+        buffer
+    }
+
+    #[cfg(feature = "flate2")]
+    #[tokio::test]
+    async fn compressed_debug_is_ignored_after_client_parses_it() {
+        let mut session = authenticated_session();
+        let mut handler = TestHandler;
+        let mut kex_done_signal = None;
+        let buffer = incoming_packet(compressed_debug_payload(200 * 1024));
+        let mut pkt: IncomingSshPacket = session.maybe_decompress(&buffer).unwrap();
+
+        super::reply(&mut session, &mut handler, &mut kex_done_signal, &mut pkt)
+            .await
+            .unwrap();
+
+        assert!(!session.common.disconnected);
+    }
+
+    #[cfg(feature = "flate2")]
+    #[test]
+    fn oversized_compressed_debug_is_rejected_before_client_ignores_it() {
+        let mut session = authenticated_session();
+        let oversized = crate::cipher::MAXIMUM_DECOMPRESSED_PACKET_LEN + 1024;
+        let buffer = incoming_packet(compressed_debug_payload(oversized));
+
+        let err = session.maybe_decompress(&buffer).unwrap_err();
+        assert!(
+            matches!(err, crate::Error::PacketSize(len) if len > crate::cipher::MAXIMUM_DECOMPRESSED_PACKET_LEN)
+        );
     }
 }
 

@@ -3,6 +3,9 @@ use std::convert::TryFrom;
 use delegate::delegate;
 use ssh_encoding::Encode;
 
+#[cfg(feature = "flate2")]
+use crate::cipher::MAXIMUM_DECOMPRESSED_PACKET_LEN;
+
 #[derive(Debug, Clone)]
 pub enum Compression {
     None,
@@ -166,6 +169,55 @@ impl Decompress {
     }
 }
 
+#[cfg(all(test, feature = "flate2"))]
+mod tests {
+    use std::io::Write;
+
+    use flate2::write::ZlibEncoder;
+
+    use super::*;
+
+    #[test]
+    fn decompressed_packet_at_limit_is_accepted() {
+        let payload = vec![b'A'; MAXIMUM_DECOMPRESSED_PACKET_LEN];
+        let mut encoder = ZlibEncoder::new(Vec::new(), flate2::Compression::best());
+        encoder.write_all(&payload).unwrap();
+        let compressed = encoder.finish().unwrap();
+
+        let mut decompressor = Decompress::Zlib(flate2::Decompress::new(true));
+        let mut output = Vec::new();
+
+        let decompressed = decompressor.decompress(&compressed, &mut output).unwrap();
+        assert_eq!(decompressed.len(), MAXIMUM_DECOMPRESSED_PACKET_LEN);
+    }
+
+    #[test]
+    fn oversized_decompressed_packet_is_rejected() {
+        let payload = vec![b'A'; MAXIMUM_DECOMPRESSED_PACKET_LEN + 1024];
+        let mut encoder = ZlibEncoder::new(Vec::new(), flate2::Compression::best());
+        encoder.write_all(&payload).unwrap();
+        let compressed = encoder.finish().unwrap();
+
+        let mut decompressor = Decompress::Zlib(flate2::Decompress::new(true));
+        let mut output = Vec::new();
+
+        let err = decompressor.decompress(&compressed, &mut output).unwrap_err();
+        assert!(
+            matches!(err, crate::Error::PacketSize(len) if len > MAXIMUM_DECOMPRESSED_PACKET_LEN)
+        );
+    }
+
+    #[test]
+    fn empty_compressed_packet_does_not_spin() {
+        let compressed = Vec::new();
+        let mut decompressor = Decompress::Zlib(flate2::Decompress::new(true));
+        let mut output = Vec::new();
+
+        let decompressed = decompressor.decompress(&compressed, &mut output).unwrap();
+        assert!(decompressed.is_empty());
+    }
+}
+
 #[cfg(feature = "flate2")]
 impl Compress {
     fn zlib_output_reserve_bound(input_len: usize) -> usize {
@@ -257,21 +309,44 @@ impl Decompress {
                 output.clear();
                 let n_in = z.total_in() as usize;
                 let n_out = z.total_out() as usize;
-                output.resize(input.len(), 0);
+                let max_output_len = MAXIMUM_DECOMPRESSED_PACKET_LEN
+                    .checked_add(1)
+                    .ok_or(crate::Error::PacketSize(usize::MAX))?;
+                output.resize(input.len().clamp(1, max_output_len), 0);
                 let flush = flate2::FlushDecompress::None;
                 loop {
                     let n_in_ = z.total_in() as usize - n_in;
                     let n_out_ = z.total_out() as usize - n_out;
                     #[allow(clippy::indexing_slicing)] // length checked
                     let d = z.decompress(&input[n_in_..], &mut output[n_out_..], flush);
                     match d? {
-                        flate2::Status::Ok => {
-                            output.resize(output.len() * 2, 0);
+                        flate2::Status::Ok | flate2::Status::BufError => {
+                            let consumed_all_input = n_in_ == input.len();
+                            let output_full = n_out_ == output.len();
+
+                            if !output_full && consumed_all_input {
+                                break;
+                            }
+
+                            if output_full {
+                                if output.len() == max_output_len {
+                                    break;
+                                }
+                                let next_len = output
+                                    .len()
+                                    .checked_mul(2)
+                                    .map(|len| len.min(max_output_len))
+                                    .ok_or(crate::Error::PacketSize(usize::MAX))?;
+                                output.resize(next_len, 0);
+                            }
                         }
                         _ => break,
                     }
                 }
                 let n_out_ = z.total_out() as usize - n_out;
+                if n_out_ > MAXIMUM_DECOMPRESSED_PACKET_LEN {
+                    return Err(crate::Error::PacketSize(n_out_));
+                }
                 #[allow(clippy::indexing_slicing)] // length checked
                 Ok(&output[..n_out_])
             }

@@ -1317,3 +1317,124 @@ impl Session {
         Ok(())
     }
 }
+
+#[cfg(all(test, feature = "flate2"))]
+mod tests {
+    use std::collections::{HashMap, VecDeque};
+    use std::io::Write;
+    use std::num::Wrapping;
+    use std::sync::Arc;
+
+    use super::*;
+    use crate::compression::{Compression, Decompress};
+    use crate::kex::{KEXES, NONE, SessionKexState};
+    use crate::session::{CommonSession, Encrypted, EncryptedState, Exchange};
+    use crate::sshbuffer::{IncomingSshPacket, PacketWriter, SSHBuffer};
+    use crate::{CryptoVec, cipher, mac};
+
+    struct TestHandler;
+
+    impl crate::server::Handler for TestHandler {
+        type Error = crate::Error;
+    }
+
+    fn authenticated_session() -> Session {
+        let config = Arc::new(crate::server::Config::default());
+        let (sender, receiver) = tokio::sync::mpsc::channel(config.event_buffer_size);
+        let handle = Handle {
+            sender,
+            channel_buffer_size: config.channel_buffer_size,
+        };
+
+        Session {
+            common: CommonSession {
+                auth_user: String::new(),
+                remote_sshid: b"SSH-2.0-test".to_vec(),
+                config: config.clone(),
+                encrypted: Some(Encrypted {
+                    state: EncryptedState::Authenticated,
+                    exchange: Some(Exchange::default()),
+                    kex: KEXES.get(&NONE).unwrap().make(),
+                    key: 0,
+                    client_mac: mac::NONE,
+                    server_mac: mac::NONE,
+                    session_id: CryptoVec::new(),
+                    channels: HashMap::new(),
+                    last_channel_id: Wrapping(0),
+                    write: Vec::new(),
+                    write_cursor: 0,
+                    last_rekey: russh_util::time::Instant::now(),
+                    server_compression: Compression::None,
+                    client_compression: Compression::None,
+                    decompress: Decompress::Zlib(flate2::Decompress::new(true)),
+                    rekey_wanted: false,
+                    received_extensions: Vec::new(),
+                    extension_info_awaiters: HashMap::new(),
+                }),
+                auth_method: None,
+                auth_attempts: 0,
+                packet_writer: PacketWriter::clear(),
+                remote_to_local: Box::new(cipher::clear::Key),
+                wants_reply: false,
+                disconnected: false,
+                buffer: Vec::new(),
+                strict_kex: false,
+                alive_timeouts: 0,
+                received_data: false,
+            },
+            sender: handle,
+            receiver,
+            target_window_size: config.window_size,
+            pending_reads: Vec::new(),
+            pending_len: 0,
+            channels: HashMap::new(),
+            open_global_requests: VecDeque::new(),
+            kex: SessionKexState::Idle,
+        }
+    }
+
+    fn compressed_debug_payload(payload_len: usize) -> Vec<u8> {
+        let mut payload = vec![b'A'; payload_len];
+        payload[0] = crate::msg::DEBUG;
+
+        let mut encoder = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::best());
+        encoder.write_all(&payload).unwrap();
+        let compressed = encoder.finish().unwrap();
+        assert!(compressed.len() < 256 * 1024);
+        compressed
+    }
+
+    fn incoming_packet(compressed: Vec<u8>) -> SSHBuffer {
+        let mut buffer = SSHBuffer::new();
+        buffer.buffer.extend_from_slice(&[0; 5]);
+        buffer.buffer.extend_from_slice(&compressed);
+        buffer
+    }
+
+    #[tokio::test]
+    async fn compressed_debug_is_ignored_after_server_parses_it() {
+        let mut session = authenticated_session();
+        let mut handler = TestHandler;
+        let buffer = incoming_packet(compressed_debug_payload(200 * 1024));
+        let mut pkt: IncomingSshPacket = session.maybe_decompress(&buffer).unwrap();
+
+        super::super::reply(&mut session, &mut handler, &mut pkt)
+            .await
+            .unwrap();
+
+        assert!(!session.common.disconnected);
+    }
+
+    #[test]
+    fn oversized_compressed_debug_is_rejected_before_server_ignores_it() {
+        let mut session = authenticated_session();
+        let buffer = incoming_packet(compressed_debug_payload(
+            crate::cipher::MAXIMUM_DECOMPRESSED_PACKET_LEN + 1024,
+        ));
+
+        let err = session.maybe_decompress(&buffer).unwrap_err();
+        assert!(
+            matches!(err, crate::Error::PacketSize(len) if len > crate::cipher::MAXIMUM_DECOMPRESSED_PACKET_LEN)
+        );
+    }
+}

@@ -4,8 +4,11 @@ use futures::Future;
 
 use super::*;
 
+#[cfg(feature = "flate2")]
 mod compress {
+    use std::borrow::Cow;
     use std::collections::HashMap;
+    use std::io::Write;
     use std::sync::{Arc, Mutex};
 
     use keys::PrivateKeyWithHashAlg;
@@ -14,15 +17,18 @@ mod compress {
 
     use super::server::{Server as _, Session};
     use super::*;
+    use crate::cipher::MAXIMUM_DECOMPRESSED_PACKET_LEN;
     use crate::server::Msg;
 
+    const OVERSIZED_DEBUG_MESSAGE_LEN: usize = MAXIMUM_DECOMPRESSED_PACKET_LEN + 1024;
+
     #[tokio::test]
     async fn compress_local_test() {
         let _ = env_logger::try_init();
 
         let client_key = PrivateKey::random(&mut rand::rng(), ssh_key::Algorithm::Ed25519).unwrap();
         let mut config = server::Config::default();
-        config.preferred = Preferred::COMPRESSED;
+        config.preferred = preferred_zlib();
         config.inactivity_timeout = None; // Some(std::time::Duration::from_secs(3));
         config.auth_rejection_time = std::time::Duration::from_secs(3);
         config
@@ -44,7 +50,7 @@ mod compress {
         });
 
         let mut config = client::Config::default();
-        config.preferred = Preferred::COMPRESSED;
+        config.preferred = preferred_zlib();
         let config = Arc::new(config);
 
         let mut session = client::connect(config, addr, Client {}).await.unwrap();
@@ -73,6 +79,19 @@ mod compress {
         }
     }
 
+    #[test]
+    fn oversized_debug_payload_can_stay_below_wire_cap() {
+        let payload = vec![b'A'; OVERSIZED_DEBUG_MESSAGE_LEN];
+        let mut encoder = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::best());
+        encoder.write_all(&payload).unwrap();
+        let compressed = encoder.finish().unwrap();
+
+        assert!(
+            compressed.len() < 256 * 1024,
+            "attacker-crafted compressed payload should stay below the normal SSH wire cap"
+        );
+    }
+
     #[derive(Clone)]
     struct Server {
         clients: Arc<Mutex<HashMap<(usize, ChannelId), super::server::Handle>>>,
@@ -135,6 +154,17 @@ mod compress {
             Ok(true)
         }
     }
+
+    fn preferred_zlib() -> Preferred {
+        Preferred {
+            compression: Cow::Borrowed(&[
+                crate::compression::ZLIB,
+                crate::compression::ZLIB_LEGACY,
+                crate::compression::NONE,
+            ]),
+            ..Preferred::DEFAULT
+        }
+    }
 }
 
 mod channels {

@@ -9,7 +9,7 @@ license = "Apache-2.0"
 name = "russh"
 readme = "../README.md"
 repository = "https://github.com/warp-tech/russh"
-version = "0.60.1"
+version = "0.60.2"
 rust-version = "1.85"
 
 [features]