fix: don't truncate compressed packets that outgrow the output reserve

Compress::compress_into reserves input.len() + 10 bytes and grew that
buffer only on flate2::Status::BufError, which deflate returns when it
cannot make progress at all, not when it filled the buffer after
consuming input - that comes back as Status::Ok. The loop broke there
with output still pending inside zlib. The packet went out short and
the leftover bytes desynced the stream for every later packet.

Incompressible payloads are what outgrow the reserve: deflate has to
store them, so past 32 KiB a packet needs more room than input.len() +
10 leaves it. Peers with compression on lose bulk binary transfers -
git pack data is the easy way to hit it - which OpenSSH reports as
channel 0: get data: incomplete message.

Break out of the loop on the output having room to spare instead, which
is what deflate signals completion with, and grow whenever it comes
back full. compress now delegates to compress_into so the two loops
cannot drift apart again. The regression tests round-trip packets that
outgrow the reserve, including through compress_into at a non-zero
start_len - the shape the packet writer uses, which the existing
compressible under-4096-byte tests don't cover.

Author: luizribeiro

russh/src/compression.rs

@@ -236,6 +236,77 @@ mod tests {
             assert_eq!(out, payload.as_slice(), "payload of {len} bytes came back wrong");
         }
     }
+
+    /// Deterministic bytes with no redundancy for deflate to exploit, so the
+    /// packet comes back out larger than it went in.
+    fn incompressible(seed: u64, len: usize) -> Vec<u8> {
+        let mut state = seed
+            .wrapping_mul(6364136223846793005)
+            .wrapping_add(1442695040888963407);
+        (0..len)
+            .map(|_| {
+                state ^= state << 13;
+                state ^= state >> 7;
+                state ^= state << 17;
+                (state >> 24) as u8
+            })
+            .collect()
+    }
+
+    #[test]
+    fn compressed_packets_larger_than_the_reserve_round_trip() {
+        // Deflate grows incompressible input rather than shrinking it, so a
+        // large enough packet needs more output than the reserve holds.
+        for len in [1024usize, 32768, 32769, 40000, 65536, 131072, 262144] {
+            let mut comp = Compress::None;
+            let mut decomp = Decompress::None;
+            Compression::Zlib.init_compress(&mut comp);
+            Compression::Zlib.init_decompress(&mut decomp);
+
+            let payload = incompressible(len as u64, len);
+            let mut cbuf = Vec::new();
+            let compressed = comp.compress(&payload, &mut cbuf).unwrap().to_vec();
+            let mut dbuf = Vec::new();
+            let out = decomp.decompress(&compressed, &mut dbuf).unwrap();
+            assert!(
+                out == payload.as_slice(),
+                "payload of {len} bytes came back as {} bytes",
+                out.len()
+            );
+        }
+    }
+
+    #[test]
+    fn compress_into_round_trips_oversized_packets_after_a_prefix() {
+        // The packet writer compresses into a buffer that already holds the
+        // packet prefix, so the payload lands partway into the output.
+        const PREFIX: &[u8] = b"packet prefix";
+
+        let mut comp = Compress::None;
+        let mut decomp = Decompress::None;
+        Compression::Zlib.init_compress(&mut comp);
+        Compression::Zlib.init_decompress(&mut decomp);
+
+        for packet in 0..4u64 {
+            let payload = incompressible(packet, 65536);
+            let mut buf = PREFIX.to_vec();
+            let n = comp
+                .compress_into(&payload, &mut buf, PREFIX.len())
+                .unwrap();
+
+            assert_eq!(&buf[..PREFIX.len()], PREFIX, "prefix was clobbered");
+            assert_eq!(buf.len(), PREFIX.len() + n, "output length disagrees");
+
+            let mut dbuf = Vec::new();
+            let out = decomp.decompress(&buf[PREFIX.len()..], &mut dbuf).unwrap();
+            assert!(
+                out == payload.as_slice(),
+                "packet {packet} came back as {} of {} bytes",
+                out.len(),
+                payload.len()
+            );
+        }
+    }
 }
 
 #[cfg(feature = "flate2")]
@@ -249,31 +320,12 @@ impl Compress {
         input: &'a [u8],
         output: &'a mut Vec<u8>,
     ) -> Result<&'a [u8], crate::Error> {
-        match *self {
-            Compress::None => Ok(input),
-            Compress::Zlib(ref mut z) => {
-                output.clear();
-                let n_in = z.total_in() as usize;
-                let n_out = z.total_out() as usize;
-                output.resize(input.len() + 10, 0);
-                let flush = flate2::FlushCompress::Partial;
-                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 c = z.compress(&input[n_in_..], &mut output[n_out_..], flush)?;
-                    match c {
-                        flate2::Status::BufError => {
-                            output.resize(output.len() * 2, 0);
-                        }
-                        _ => break,
-                    }
-                }
-                let n_out_ = z.total_out() as usize - n_out;
-                #[allow(clippy::indexing_slicing)] // length checked
-                Ok(&output[..n_out_])
-            }
+        if let Compress::None = *self {
+            return Ok(input);
         }
+        let n_out_ = self.compress_into(input, output, 0)?;
+        #[allow(clippy::indexing_slicing)] // length checked
+        Ok(&output[..n_out_])
     }
 
     pub fn compress_into(
@@ -300,9 +352,17 @@ impl Compress {
                     let n_out_ = z.total_out() as usize - n_out;
                     #[allow(clippy::indexing_slicing)] // length checked
                     let c = z.compress(&input[n_in_..], &mut output[start_len + n_out_..], flush)?;
+
+                    // deflate leaves room to spare only once it has emitted
+                    // everything it holds, so a buffer filled to the brim may
+                    // still have more to come.
+                    let room = output.len().saturating_sub(start_len);
+                    if z.total_out() as usize - n_out < room {
+                        break;
+                    }
                     match c {
-                        flate2::Status::BufError => {
-                            let growth = output.len().saturating_sub(start_len).max(1);
+                        flate2::Status::Ok | flate2::Status::BufError => {
+                            let growth = room.max(1);
                             output.resize(output.len() + growth, 0);
                         }
                         _ => break,