From de7cba3c8bdcdb826a847948b4ddf8d31d04762b Mon Sep 17 00:00:00 2001 From: Mark Karpeles Date: Tue, 7 Jul 2026 03:12:22 +0900 Subject: [PATCH 1/2] fix: large-file corruption in snappy/lzss/lzs/xpress; add stress harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Streaming ~1-16 GiB through each codec (new examples/stress.rs, bounded memory) surfaced several corruption bugs that KB-scale unit tests and fuzzing never hit. None affect xz/lzma2/zstd/lz4/deflate/brotli/bzip2, which round-trip cleanly to 16 GiB (xz also verified against native `xz -d`). - snappy, lzss: the uncompressed length is written as a 32-bit field (`len as u32` / 4-byte LE header). At >= 4 GiB it silently truncated, making lzss decode the wrong length (MISMATCH) and snappy emit an undecodable stream. Reject over-length input with a clean `Error::Unsupported` instead of corrupting. - lzs, xpress: a fixed `SANITY_MATCH_LEN` bomb guard (1<<24 = 16 MiB for lzs, 1<<28 = 256 MiB for xpress) wrongly rejected VALID streams whose longest back-reference exceeded it — trivially reached on repetitive input (one match can span most of the output). Bound the guard by the declared total length instead. Their match finders also address positions with `u32`, which degrades/corrupts above 4 GiB, so the encoders now reject >= 4 GiB input with a clean error. Adds a fast regression test for the lzs guard (a ~20 MiB match that the old code rejected) and examples/stress.rs (generate -> encode -> decode -> 128-bit hash compare, plus an our-encode -> native `xz -d` mode). Full suite 1706 green; fmt + clippy clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.toml | 5 + examples/stress.rs | 383 ++++++++++++++++++++++++++++++++++++++++++ src/lzs/decoder.rs | 13 +- src/lzs/encoder.rs | 7 + src/lzss/mod.rs | 7 + src/snappy/mod.rs | 8 + src/xpress/decoder.rs | 11 +- src/xpress/encoder.rs | 7 + tests/lzs.rs | 14 ++ 9 files changed, 452 insertions(+), 3 deletions(-) create mode 100644 examples/stress.rs diff --git a/Cargo.toml b/Cargo.toml index 8ef45dd..f8ba541 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -300,3 +300,8 @@ required-features = ["factory"] name = "micro" path = "examples/micro.rs" required-features = ["factory"] + +[[example]] +name = "stress" +path = "examples/stress.rs" +required-features = ["factory"] diff --git a/examples/stress.rs b/examples/stress.rs new file mode 100644 index 0000000..4fe67f4 --- /dev/null +++ b/examples/stress.rs @@ -0,0 +1,383 @@ +//! Large-file streaming stress test — bounded memory at any scale. +//! +//! Generates up to many GiB of deterministic data, streams it through an +//! encoder and (interleaved) a decoder, and checks the decoded stream hashes +//! and lengths identically to the input. Holds only fixed-size buffers, so it +//! runs at 16 GiB+ without buffering the whole stream. +//! +//! Usage: +//! stress +//! kind: text | random | mixed | runs | seq +//! mode: self our encode -> our decode (round-trip hash check) +//! enc-native our encode -> native `xz -d` (xz/lzma2 only) +//! dec-native native `xz -z` -> our decode (xz only) +//! +//! Exit code 0 = OK, 1 = MISMATCH/error. + +use std::io::{Read, Write}; +use std::process::{Command, Stdio}; +use std::time::Instant; + +use compcol::{Status, factory}; + +// ─── deterministic, position-addressable data generator ────────────────── +// `fill(kind, abs_offset, buf)` fills `buf` with the bytes of the stream at +// absolute offset `abs_offset`, so any range is reproducible (needed to hash +// the input independently of how it is chunked, and for the native modes). + +const LOREM: &[u8] = b"Lorem ipsum dolor sit amet, consectetur adipiscing elit, \ +sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad \ +minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea. "; + +#[inline] +fn lcg(state: &mut u64) -> u64 { + // 64-bit LCG (Knuth MMIX constants). + *state = state + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + *state +} + +fn fill(kind: &str, abs: u64, buf: &mut [u8]) { + match kind { + "text" => { + for (i, b) in buf.iter_mut().enumerate() { + let p = (abs + i as u64) % LOREM.len() as u64; + *b = LOREM[p as usize]; + } + } + "seq" => { + for (i, b) in buf.iter_mut().enumerate() { + *b = (abs + i as u64) as u8; + } + } + "random" => { + // Seed the LCG from the absolute offset so each byte is a pure + // function of its position (streams reproducibly, still looks + // incompressible). + for (i, b) in buf.iter_mut().enumerate() { + let pos = abs + i as u64; + let mut s = pos.wrapping_mul(0x9E3779B97F4A7C15) ^ 0xD1B54A32D192ED03; + s = lcg(&mut s); + *b = (s >> 33) as u8; + } + } + "runs" => { + // Long runs of a byte value that changes every ~4 KiB — stresses + // RLE / overlap copies and offset==1 splats at scale. + for (i, b) in buf.iter_mut().enumerate() { + let pos = abs + i as u64; + *b = (pos >> 12) as u8; + } + } + "mixed" => { + // Alternate 64 KiB compressible (text) and 64 KiB incompressible + // (random) blocks — repeatedly crosses the compressed/uncompressed + // chunk transition the lzma2 encoder switches on. + for (i, b) in buf.iter_mut().enumerate() { + let pos = abs + i as u64; + if (pos >> 16) & 1 == 0 { + *b = LOREM[(pos % LOREM.len() as u64) as usize]; + } else { + let mut s = pos.wrapping_mul(0x9E3779B97F4A7C15) ^ 0xD1B54A32D192ED03; + s = lcg(&mut s); + *b = (s >> 33) as u8; + } + } + } + _ => panic!("unknown kind {kind}"), + } +} + +// ─── rolling 128-bit hash (two independent FNV-1a lanes) ────────────────── +#[derive(Clone, Copy)] +struct Hash { + a: u64, + b: u64, + len: u64, +} +impl Hash { + fn new() -> Self { + Self { + a: 0xcbf29ce484222325, + b: 0x100000001b3 ^ 0x9E3779B97F4A7C15, + len: 0, + } + } + #[inline] + fn update(&mut self, data: &[u8]) { + let mut a = self.a; + let mut b = self.b; + for &x in data { + a = (a ^ x as u64).wrapping_mul(0x100000001b3); + b = (b ^ x as u64) + .wrapping_mul(0x9E3779B97F4A7C15) + .rotate_left(27); + } + self.a = a; + self.b = b; + self.len += data.len() as u64; + } + fn eq(&self, o: &Hash) -> bool { + self.a == o.a && self.b == o.b && self.len == o.len + } + fn show(&self) -> String { + format!("{:016x}{:016x}/{}", self.a, self.b, self.len) + } +} + +const CHUNK: usize = 1 << 20; // 1 MiB generator chunk +const BUF: usize = 1 << 18; // 256 KiB codec buffers + +fn mb(bytes: u64) -> f64 { + bytes as f64 / 1e6 +} + +fn run_self(algo: &str, kind: &str, total: u64) -> Result<(), String> { + let mut enc = factory::encoder_by_name(algo).ok_or("unknown algo (encode)")?; + let mut dec = factory::decoder_by_name(algo).ok_or("unknown algo (decode)")?; + let mut hin = Hash::new(); + let mut hout = Hash::new(); + let mut comp_bytes: u64 = 0; + let mut enc_buf = vec![0u8; BUF]; + let mut dec_buf = vec![0u8; BUF]; + let mut genbuf = vec![0u8; CHUNK]; + + // Feed compressed bytes straight into the decoder, hashing decoded output. + let feed = |dec: &mut Box, + comp: &[u8], + hout: &mut Hash, + dec_buf: &mut [u8]| + -> Result<(), String> { + let mut c = 0; + while c < comp.len() { + let (p, status) = dec + .decode(&comp[c..], dec_buf) + .map_err(|e| format!("decode: {e}"))?; + c += p.consumed; + hout.update(&dec_buf[..p.written]); + match status { + Status::InputEmpty | Status::StreamEnd => { + if p.consumed == 0 && p.written == 0 { + break; + } + } + Status::OutputFull => {} + } + } + Ok(()) + }; + + let t0 = Instant::now(); + let mut produced: u64 = 0; + let mut next_report = 1u64 << 30; + while produced < total { + let n = CHUNK.min((total - produced) as usize); + fill(kind, produced, &mut genbuf[..n]); + hin.update(&genbuf[..n]); + let mut consumed = 0; + while consumed < n { + let (p, status) = enc + .encode(&genbuf[consumed..n], &mut enc_buf) + .map_err(|e| format!("encode: {e}"))?; + consumed += p.consumed; + comp_bytes += p.written as u64; + feed(&mut dec, &enc_buf[..p.written], &mut hout, &mut dec_buf)?; + match status { + Status::InputEmpty | Status::StreamEnd => { + if p.consumed == 0 && p.written == 0 { + break; + } + } + Status::OutputFull => {} + } + } + produced += n as u64; + if produced >= next_report { + let s = t0.elapsed().as_secs_f64(); + eprintln!( + " .. {} GiB in, {:.1} MB comp, {:.0} MB/s", + produced >> 30, + mb(comp_bytes), + mb(produced) / s + ); + next_report += 1 << 30; + } + } + // Flush encoder, feeding the tail into the decoder. + loop { + let (p, status) = enc + .finish(&mut enc_buf) + .map_err(|e| format!("finish enc: {e}"))?; + comp_bytes += p.written as u64; + feed(&mut dec, &enc_buf[..p.written], &mut hout, &mut dec_buf)?; + if matches!(status, Status::StreamEnd) { + break; + } + if p.written == 0 { + break; + } + } + // Drain any decoder-buffered output, then finish it. + loop { + let (p, _s) = dec + .decode(&[], &mut dec_buf) + .map_err(|e| format!("drain dec: {e}"))?; + hout.update(&dec_buf[..p.written]); + if p.written == 0 { + break; + } + } + loop { + let (p, status) = dec + .finish(&mut dec_buf) + .map_err(|e| format!("finish dec: {e}"))?; + hout.update(&dec_buf[..p.written]); + if matches!(status, Status::StreamEnd) || p.written == 0 { + break; + } + } + + let s = t0.elapsed().as_secs_f64(); + let ok = hin.eq(&hout); + println!( + "{algo} {kind} {}B comp={}B ratio={:.4} {:.0} MB/s {}", + total, + comp_bytes, + comp_bytes as f64 / total as f64, + mb(total) / s, + if ok { "OK" } else { "MISMATCH" } + ); + if !ok { + println!(" in ={}", hin.show()); + println!(" out={}", hout.show()); + return Err(format!( + "ROUND-TRIP MISMATCH: in.len={} out.len={}", + hin.len, hout.len + )); + } + Ok(()) +} + +// Our encode piped to native `xz -d`; verify its output hashes to the input. +fn run_enc_native(algo: &str, kind: &str, total: u64) -> Result<(), String> { + if algo != "xz" { + return Err("enc-native only supports xz".into()); + } + let mut hin = Hash::new(); + let mut genbuf = vec![0u8; CHUNK]; + let mut produced = 0u64; + while produced < total { + let n = CHUNK.min((total - produced) as usize); + fill(kind, produced, &mut genbuf[..n]); + hin.update(&genbuf[..n]); + produced += n as u64; + } + + let mut child = Command::new("xz") + .args(["-d", "-c", "-T", "1"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .map_err(|e| format!("spawn xz: {e}"))?; + let mut xz_in = child.stdin.take().unwrap(); + let mut xz_out = child.stdout.take().unwrap(); + + // Thread: generate + our-encode, write compressed to xz stdin. + let algo_t = algo.to_string(); + let kind_s = kind.to_string(); + let writer = std::thread::spawn(move || -> Result { + let mut enc = factory::encoder_by_name(&algo_t).ok_or("unknown algo")?; + let mut enc_buf = vec![0u8; BUF]; + let mut genbuf = vec![0u8; CHUNK]; + let mut comp = 0u64; + let mut produced = 0u64; + while produced < total { + let n = CHUNK.min((total - produced) as usize); + fill(&kind_s, produced, &mut genbuf[..n]); + let mut consumed = 0; + while consumed < n { + let (p, status) = enc + .encode(&genbuf[consumed..n], &mut enc_buf) + .map_err(|e| format!("encode: {e}"))?; + consumed += p.consumed; + comp += p.written as u64; + xz_in + .write_all(&enc_buf[..p.written]) + .map_err(|e| format!("pipe: {e}"))?; + if matches!(status, Status::InputEmpty | Status::StreamEnd) + && p.consumed == 0 + && p.written == 0 + { + break; + } + } + produced += n as u64; + } + loop { + let (p, status) = enc + .finish(&mut enc_buf) + .map_err(|e| format!("finish: {e}"))?; + comp += p.written as u64; + xz_in + .write_all(&enc_buf[..p.written]) + .map_err(|e| format!("pipe: {e}"))?; + if matches!(status, Status::StreamEnd) || p.written == 0 { + break; + } + } + drop(xz_in); + Ok(comp) + }); + + // Main: hash xz's decompressed output. + let mut hout = Hash::new(); + let mut rbuf = vec![0u8; BUF]; + loop { + let k = xz_out + .read(&mut rbuf) + .map_err(|e| format!("read xz: {e}"))?; + if k == 0 { + break; + } + hout.update(&rbuf[..k]); + } + let comp = writer.join().map_err(|_| "writer panicked")??; + let status = child.wait().map_err(|e| format!("wait xz: {e}"))?; + if !status.success() { + return Err("native xz -d rejected our output".into()); + } + let ok = hin.eq(&hout); + println!( + "{algo} {kind} {}B (our-enc -> native xz -d) comp={}B {}", + total, + comp, + if ok { "OK" } else { "MISMATCH" } + ); + if !ok { + return Err(format!("MISMATCH: in.len={} out.len={}", hin.len, hout.len)); + } + Ok(()) +} + +fn main() { + let a: Vec = std::env::args().collect(); + if a.len() < 5 { + eprintln!( + "usage: stress " + ); + std::process::exit(2); + } + let (algo, kind, mode) = (&a[1], &a[2], &a[4]); + let total: u64 = a[3].parse().expect("total_bytes"); + let r = match mode.as_str() { + "self" => run_self(algo, kind, total), + "enc-native" => run_enc_native(algo, kind, total), + _ => Err(format!("unknown mode {mode}")), + }; + if let Err(e) = r { + eprintln!("FAIL: {e}"); + std::process::exit(1); + } +} diff --git a/src/lzs/decoder.rs b/src/lzs/decoder.rs index f473d78..26224c8 100644 --- a/src/lzs/decoder.rs +++ b/src/lzs/decoder.rs @@ -120,7 +120,16 @@ impl Decoder { 0b10 => Ok(Some(7)), 0b11 => { // Chain: read nibbles until one is != 1111. Each - // 1111 adds 15. + // 1111 adds 15. A single match can legitimately be as + // long as the whole declared output (highly repetitive + // input compresses to one long back-reference), so bound + // the accumulator by the declared total length rather + // than a fixed constant — otherwise valid streams whose + // longest match exceeds ~16 MiB are wrongly rejected. + let cap = match self.header { + HeaderPhase::Active { target } => target.min(u32::MAX as u64) as u32, + _ => SANITY_MATCH_LEN, + }; let mut acc: u32 = 8; loop { let nib = match self.bits.read_bits(4) { @@ -141,7 +150,7 @@ impl Decoder { Some(v) => v, None => return Err(Error::Corrupt), }; - if acc > SANITY_MATCH_LEN { + if acc > cap { return Err(Error::Corrupt); } } diff --git a/src/lzs/encoder.rs b/src/lzs/encoder.rs index 33ad0ff..49a26f2 100644 --- a/src/lzs/encoder.rs +++ b/src/lzs/encoder.rs @@ -103,6 +103,13 @@ impl RawEncoder for Encoder { done: false, }); } + // The match finder addresses positions with `u32`, so above 4 GiB the + // hash table truncates positions and can emit wrong-distance matches. + // Reject a stream at that limit with a clean error rather than risk + // silently corrupt output. + if self.raw.len() as u64 + input.len() as u64 > u32::MAX as u64 { + return Err(Error::Unsupported); + } self.raw.extend_from_slice(input); Ok(RawProgress { consumed: input.len(), diff --git a/src/lzss/mod.rs b/src/lzss/mod.rs index 44ec09e..f1e6c08 100644 --- a/src/lzss/mod.rs +++ b/src/lzss/mod.rs @@ -262,6 +262,13 @@ impl Default for Encoder { impl RawEncoder for Encoder { fn raw_encode(&mut self, input: &[u8], _output: &mut [u8]) -> Result { + // The stream is prefixed with a fixed 4-byte little-endian uncompressed + // length, so it can represent at most `u32::MAX` bytes. Reject overflow + // with a clean error rather than silently truncating the header (which + // would make the decoder emit the wrong number of bytes). + if self.input.len() as u64 + input.len() as u64 > u32::MAX as u64 { + return Err(Error::Unsupported); + } self.input.extend_from_slice(input); Ok(RawProgress { consumed: input.len(), diff --git a/src/snappy/mod.rs b/src/snappy/mod.rs index 7134285..9ee3c08 100644 --- a/src/snappy/mod.rs +++ b/src/snappy/mod.rs @@ -131,6 +131,14 @@ impl Encoder { impl RawEncoder for Encoder { fn raw_encode(&mut self, input: &[u8], _output: &mut [u8]) -> Result { + // Snappy's raw block format prefixes the stream with the uncompressed + // length as a 32-bit varint, and the match finder addresses positions + // with `u32`, so a block can hold at most `u32::MAX` bytes. Reject an + // over-length stream with a clean error instead of silently truncating + // the length field into a corrupt, undecodable stream. + if self.input.len() as u64 + input.len() as u64 > u32::MAX as u64 { + return Err(Error::Unsupported); + } // Buffer the whole input — Snappy needs the total length up front, // and back-references can reach anywhere within the block, so we // can't emit anything until the caller signals end of input. diff --git a/src/xpress/decoder.rs b/src/xpress/decoder.rs index c9935ac..f9be9cf 100644 --- a/src/xpress/decoder.rs +++ b/src/xpress/decoder.rs @@ -461,7 +461,16 @@ fn try_read_length_at( if dw < 22 { return Err(Error::Corrupt); } - if dw > SANITY_MATCH_LEN { + // A single match can legitimately be as long as the whole declared output + // (highly repetitive input collapses to one long back-reference), so bound + // the tier-3 length by the declared total rather than a fixed constant — + // otherwise valid streams whose longest match exceeds 256 MiB are wrongly + // rejected. + let cap = match dec.header { + HeaderPhase::Active { target } => target.min(u32::MAX as u64) as u32, + _ => SANITY_MATCH_LEN, + }; + if dw > cap { return Err(Error::Corrupt); } Ok(Some((dw + 3, hb_consumed + 1 + 2 + 4, hb_op))) diff --git a/src/xpress/encoder.rs b/src/xpress/encoder.rs index ac511f0..3c8d2ad 100644 --- a/src/xpress/encoder.rs +++ b/src/xpress/encoder.rs @@ -107,6 +107,13 @@ impl RawEncoder for Encoder { done: false, }); } + // The match finder addresses positions with `u32`, so above 4 GiB the + // hash table truncates positions and can emit wrong-distance matches. + // Reject a stream at that limit with a clean error rather than risk + // silently corrupt output. + if self.raw.len() as u64 + input.len() as u64 > u32::MAX as u64 { + return Err(Error::Unsupported); + } self.raw.extend_from_slice(input); Ok(RawProgress { consumed: input.len(), diff --git a/tests/lzs.rs b/tests/lzs.rs index 1f8a0fc..c4b4ad5 100644 --- a/tests/lzs.rs +++ b/tests/lzs.rs @@ -165,6 +165,20 @@ fn round_trip_repeated_64kib() { round_trip(&input); } +#[test] +fn round_trip_match_exceeding_16mib_guard() { + // Regression: a single back-reference longer than the old fixed 16 MiB + // (1<<24) "sanity" cap must still decode. ~20 MiB of a short repeating + // phrase collapses to one very long match; the decoder previously rejected + // it as a suspected decompression bomb (Error::Corrupt). + let phrase = b"the quick brown fox jumps over the lazy dog. "; + let mut input = Vec::with_capacity(20 << 20); + while input.len() < (20 << 20) { + input.extend_from_slice(phrase); + } + round_trip(&input); +} + #[test] fn round_trip_mixed_short_runs() { let mut input = Vec::new(); From 07f9fd7e1e454341c3c1a5f871aa005d71e8d62f Mon Sep 17 00:00:00 2001 From: Mark Karpeles Date: Tue, 7 Jul 2026 19:19:47 +0900 Subject: [PATCH 2/2] fix(xpress): cap tier-3 length at u32::MAX-3 to avoid dw+3 overflow The previous commit bounded the tier-3 match-length guard by the declared `target`, but `target` is attacker-controlled (8-byte header), so a malformed stream declaring a huge target let `dw` reach `u32::MAX` and the `dw + 3` return overflowed (panic under debug assertions). Found by the decoder_xpress fuzz smoke job. Cap at `u32::MAX - 3` so the add can never overflow while still admitting legitimately long matches. Adds the libFuzzer crash input as a regression test. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/xpress/decoder.rs | 4 +++- tests/xpress.rs | 13 +++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/src/xpress/decoder.rs b/src/xpress/decoder.rs index f9be9cf..2b9c771 100644 --- a/src/xpress/decoder.rs +++ b/src/xpress/decoder.rs @@ -466,8 +466,10 @@ fn try_read_length_at( // the tier-3 length by the declared total rather than a fixed constant — // otherwise valid streams whose longest match exceeds 256 MiB are wrongly // rejected. + // Cap at `u32::MAX - 3` so the `dw + 3` below cannot overflow on a + // malformed stream that declares a huge target and a huge length field. let cap = match dec.header { - HeaderPhase::Active { target } => target.min(u32::MAX as u64) as u32, + HeaderPhase::Active { target } => target.min((u32::MAX - 3) as u64) as u32, _ => SANITY_MATCH_LEN, }; if dw > cap { diff --git a/tests/xpress.rs b/tests/xpress.rs index 80ab1a1..7d662db 100644 --- a/tests/xpress.rs +++ b/tests/xpress.rs @@ -312,6 +312,19 @@ fn decoder_reset_allows_reuse() { // ─── decoder error rejection ─────────────────────────────────────────── +#[test] +fn oversized_length_field_rejected_without_panic() { + // Regression (libFuzzer decoder_xpress crash-9bfdce...): a malformed + // stream declaring a huge target plus a huge tier-3 length field used to + // overflow `dw + 3` and panic under debug assertions. It must now fail + // cleanly (the declared output is never satisfiable) instead of crashing. + let crash: &[u8] = &[ + 7, 0, 0, 125, 253, 0, 0, 0, 10, 38, 189, 40, 189, 0, 0, 0, 0, 0, 0, 38, 255, 236, 255, 0, + 255, 255, 0, 0, 255, 255, 255, 255, 189, 38, 17, 96, + ]; + assert!(decode_chunked(crash, 4096, 4096).is_err()); +} + #[test] fn truncated_header_rejected() { // Less than 8 bytes ⇒ finish must error.