This is an automated email from the ASF dual-hosted git repository.
Kriskras99 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/avro-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 661bbd9 fix: bound decompressed size to guard against decompression
bombs (#582)
661bbd9 is described below
commit 661bbd9df5e73126c9a52bb2e1698039a728a8d4
Author: Ismaël Mejía <[email protected]>
AuthorDate: Tue Jul 14 11:45:52 2026 +0200
fix: bound decompressed size to guard against decompression bombs (#582)
* fix: bound decompressed size to guard against decompression bombs
`Codec::decompress` grew the output buffer without limit, so a small
compressed
block could inflate to an enormous buffer and exhaust memory (a
"decompression
bomb"): deflate/zstd/bzip2/xz decompressed the whole stream into a `Vec`,
and
snappy allocated `vec![0; decompressed_size]` from the (untrusted) block
header.
Bound every codec's output by the configured `max_allocation_bytes`:
- deflate uses `decompress_to_vec_with_limit`, mapping the resulting
`HasMoreOutput` status to a memory-allocation error;
- snappy validates the header-declared size with `safe_len` before
allocating;
- zstd/bzip2/xz read through `Read::take(max + 1)` and reject an output
that
exceeds the budget.
The limit stays configurable through `max_allocation_bytes`, consistent
with the
existing byte-length guards. Legitimate blocks within the budget still
round-trip.
Assisted-by: GitHub Copilot:claude-opus-4.8
* refactor: address review on decompression bounds
- bzip2/xz: propagate a decode error instead of panicking. read_to_end can
fail on malformed/corrupt input, so replace unwrap_or_else(unreachable!)
with
dedicated Details::Bzip2Decompress / Details::XzDecompress errors, mapped
via
map_err, so a bad block returns an AvroResult error rather than aborting.
- decompression_bomb test: assert the failure is specifically
Details::MemoryAllocation, so a regression that fails for another reason
cannot mask the missing bound.
Assisted-by: GitHub Copilot:claude-opus-4.8
* refactor: gate new decompress error variants; fix stale deflate message
Addresses review:
- Feature-gate Details::Bzip2Decompress and Details::XzDecompress behind the
`bzip` and `xz` features (like the existing Snappy variants), so adding
them
is not a breaking change for downstream crates that exhaustively match on
Details without those features enabled.
- Update the deflate BadParam message: it no longer references
decompress_to_vec
("invalid output buffer size" / "not possible for _to_vec()") now that the
code uses decompress_to_vec_with_limit.
Assisted-by: GitHub Copilot:claude-opus-4.8
* fix: reject short Snappy blocks instead of panicking
A Snappy block ends with a 4-byte CRC32. Decompression sliced the block with
stream.len() - 4, which underflows (usize) and panics if a truncated or
corrupt
block is shorter than 4 bytes -- a decode failure turned into a process
abort.
Compute the data end with checked_sub(4) and return a new (snappy-gated)
Details::BadSnappyLength error for a too-short block, slicing via the
checked
offset. Adds a regression test over 0..4 byte inputs.
Assisted-by: GitHub Copilot:claude-opus-4.8
* chore: cleanup code and improve error message
---------
Co-authored-by: Kriskras99 <[email protected]>
---
avro/src/codec.rs | 95 ++++++++++++++++++++++++++++----------
avro/src/error.rs | 24 +++++++++-
avro/src/util.rs | 2 +-
avro/tests/decompression_bomb.rs | 98 ++++++++++++++++++++++++++++++++++++++++
4 files changed, 191 insertions(+), 28 deletions(-)
diff --git a/avro/src/codec.rs b/avro/src/codec.rs
index 31aa92c..661558e 100644
--- a/avro/src/codec.rs
+++ b/avro/src/codec.rs
@@ -16,6 +16,7 @@
// under the License.
//! Logic for all supported compression codecs in Avro.
+
use crate::{AvroResult, Error, error::Details, types::Value};
use strum::{EnumIter, EnumString, IntoStaticStr};
@@ -144,36 +145,50 @@ impl Codec {
/// Decompress a stream of bytes in-place.
pub fn decompress(self, stream: &mut Vec<u8>) -> AvroResult<()> {
+ // Cap the decompressed output at the configured allocation budget so a
+ // small compressed block cannot inflate to an enormous buffer (a
+ // "decompression bomb") and exhaust memory.
+ let max_bytes =
+
crate::util::max_allocation_bytes(crate::util::DEFAULT_MAX_ALLOCATION_BYTES);
*stream = match self {
Codec::Null => return Ok(()),
- Codec::Deflate(_settings) =>
miniz_oxide::inflate::decompress_to_vec(stream).map_err(|e| {
- let err = {
- use
miniz_oxide::inflate::TINFLStatus::{FailedCannotMakeProgress, BadParam,
Adler32Mismatch, Failed, Done, NeedsMoreInput, HasMoreOutput};
- use std::io::{Error,ErrorKind};
- match e.status {
- FailedCannotMakeProgress =>
Error::from(ErrorKind::UnexpectedEof),
- BadParam => Error::other("Unexpected error:
miniz_oxide reported invalid output buffer size. Please report this to avro-rs
developers."), // not possible for _to_vec()
- Adler32Mismatch => Error::from(ErrorKind::InvalidData),
- Failed => Error::from(ErrorKind::InvalidData),
- Done => Error::other("Unexpected error: miniz_oxide
reported an error with a success status. Please report this to avro-rs
developers."),
- NeedsMoreInput =>
Error::from(ErrorKind::UnexpectedEof),
- HasMoreOutput => Error::other("Unexpected error:
miniz_oxide has more data than the output buffer can hold. Please report this
to avro-rs developers."), // not possible for _to_vec()
- other => Error::other(format!("Unexpected error:
{other:?}"))
- }
+ Codec::Deflate(_settings) =>
miniz_oxide::inflate::decompress_to_vec_with_limit(stream,
max_bytes).map_err(|e| {
+ use std::io::ErrorKind;
+ use miniz_oxide::inflate::TINFLStatus;
+
+ let details = match e.status {
+ TINFLStatus::FailedCannotMakeProgress |
TINFLStatus::NeedsMoreInput =>
Details::DeflateDecompress(ErrorKind::UnexpectedEof.into()),
+ TINFLStatus::Adler32Mismatch | TINFLStatus::Failed |
TINFLStatus::BadParam =>
Details::DeflateDecompress(ErrorKind::InvalidData.into()),
+ TINFLStatus::Done =>
Details::DeflateDecompress(std::io::Error::other("Unexpected error: miniz_oxide
reported an error with a success status. Please report this to avro-rs
developers.")),
+ // Output is larger than max allocation allowed
+ TINFLStatus::HasMoreOutput => Details::MemoryAllocation {
+ desired: None,
+ maximum: max_bytes,
+ },
+ other =>
Details::DeflateDecompress(std::io::Error::other(format!("Unexpected error:
{other:?}")))
};
- Error::new(Details::DeflateDecompress(err))
+ Error::new(details)
})?,
#[cfg(feature = "snappy")]
Codec::Snappy => {
- let decompressed_size =
snap::raw::decompress_len(&stream[..stream.len() - 4])
+ // The block ends with a 4-byte CRC32; a truncated/corrupt
block
+ // shorter than that must error rather than underflow the
slice.
+ let data_end = stream
+ .len()
+ .checked_sub(4)
+ .ok_or(Details::BadSnappyLength(stream.len()))?;
+ let decompressed_size =
snap::raw::decompress_len(&stream[..data_end])
.map_err(Details::GetSnappyDecompressLen)?;
+ // The decompressed size is taken from the (untrusted) block
+ // header, so bound it before allocating for it.
+ let decompressed_size =
crate::util::safe_len(decompressed_size)?;
let mut decoded = vec![0; decompressed_size];
snap::raw::Decoder::new()
- .decompress(&stream[..stream.len() - 4], &mut decoded[..])
+ .decompress(&stream[..data_end], &mut decoded[..])
.map_err(Details::SnappyDecompress)?;
let mut last_four: [u8; 4] = [0; 4];
- last_four.copy_from_slice(&stream[(stream.len() - 4)..]);
+ last_four.copy_from_slice(&stream[data_end..]);
let expected: u32 = u32::from_be_bytes(last_four);
let mut hasher = crc32fast::Hasher::new();
@@ -187,14 +202,22 @@ impl Codec {
}
#[cfg(feature = "zstandard")]
Codec::Zstandard(_settings) => {
- use std::io::BufReader;
+ use std::io::{BufReader, Read};
use zstd::zstd_safe;
let mut decoded = Vec::new();
let buffer_size = zstd_safe::DCtx::in_size();
let buffer = BufReader::with_capacity(buffer_size,
&stream[..]);
- let mut decoder =
zstd::Decoder::new(buffer).map_err(Details::ZstdDecompress)?;
- std::io::copy(&mut decoder, &mut
decoded).map_err(Details::ZstdDecompress)?;
+ let decoder =
zstd::Decoder::new(buffer).map_err(Details::ZstdDecompress)?;
+ // Read one byte past the budget so an output that exactly
fills
+ // it is allowed, while a larger (bomb) output is detected.
+ decoder
+ .take((max_bytes as u64).saturating_add(1))
+ .read_to_end(&mut decoded)
+ .map_err(Details::ZstdDecompress)?;
+ if decoded.len() > max_bytes {
+ return Err(Details::MemoryAllocation { desired: None,
maximum: max_bytes }.into());
+ }
decoded
}
#[cfg(feature = "bzip")]
@@ -202,9 +225,14 @@ impl Codec {
use bzip2::read::BzDecoder;
use std::io::Read;
- let mut decoder = BzDecoder::new(&stream[..]);
let mut decoded = Vec::new();
- decoder.read_to_end(&mut decoded).unwrap_or_else(|_|
unreachable!("No I/O errors possible with Vec<u8>"));
+ BzDecoder::new(&stream[..])
+ .take((max_bytes as u64).saturating_add(1))
+ .read_to_end(&mut decoded)
+ .map_err(Details::Bzip2Decompress)?;
+ if decoded.len() > max_bytes {
+ return Err(Details::MemoryAllocation { desired: None,
maximum: max_bytes }.into());
+ }
decoded
}
#[cfg(feature = "xz")]
@@ -212,9 +240,14 @@ impl Codec {
use liblzma::read::XzDecoder;
use std::io::Read;
- let mut decoder = XzDecoder::new(&stream[..]);
let mut decoded: Vec<u8> = Vec::new();
- decoder.read_to_end(&mut decoded).unwrap_or_else(|_|
unreachable!("No I/O errors possible with Vec<u8>"));
+ XzDecoder::new(&stream[..])
+ .take((max_bytes as u64).saturating_add(1))
+ .read_to_end(&mut decoded)
+ .map_err(Details::XzDecompress)?;
+ if decoded.len() > max_bytes {
+ return Err(Details::MemoryAllocation { desired: None,
maximum: max_bytes }.into());
+ }
decoded
}
};
@@ -321,6 +354,18 @@ mod tests {
compress_and_decompress(Codec::Snappy)
}
+ #[cfg(feature = "snappy")]
+ #[test]
+ fn snappy_decompress_short_block_errors_without_panicking() {
+ // A block shorter than the trailing 4-byte CRC must return an error
+ // rather than underflowing `stream.len() - 4` and panicking.
+ for len in 0..4usize {
+ let mut stream = vec![0u8; len];
+ let result = Codec::Snappy.decompress(&mut stream);
+ assert!(result.is_err(), "len={len} should error, got {result:?}");
+ }
+ }
+
#[cfg(feature = "zstandard")]
#[test]
fn zstd_compress_and_decompress() -> TestResult {
diff --git a/avro/src/error.rs b/avro/src/error.rs
index bd9cbcb..2acb217 100644
--- a/avro/src/error.rs
+++ b/avro/src/error.rs
@@ -111,9 +111,17 @@ pub enum Details {
},
#[error(
- "Unable to allocate {desired} bytes (maximum allowed: {maximum}).
Change the limit using `apache_avro::util::max_allocation_bytes`"
+ "{} (maximum allowed: {maximum}). Change the limit using
`apache_avro::util::max_allocation_bytes`",
+ if let Some(desired) = desired {
+ format!("Unable to allocate {desired} bytes")
+ } else {
+ "Allocation limit reached with unknown amount of bytes
remaining".to_string()
+ },
)]
- MemoryAllocation { desired: usize, maximum: usize },
+ MemoryAllocation {
+ desired: Option<usize>,
+ maximum: usize,
+ },
/// Describe a specific error happening with decimal representation
#[error(
@@ -503,6 +511,10 @@ pub enum Details {
#[error("Failed to get snappy decompression length: {0}")]
GetSnappyDecompressLen(#[source] snap::Error),
+ #[cfg(feature = "snappy")]
+ #[error("Snappy-compressed block is {0} bytes, too short to contain the
trailing CRC32")]
+ BadSnappyLength(usize),
+
#[cfg(feature = "snappy")]
#[error("Failed to decompress with snappy: {0}")]
SnappyDecompress(#[source] snap::Error),
@@ -513,6 +525,14 @@ pub enum Details {
#[error("Failed to decompress with zstd: {0}")]
ZstdDecompress(#[source] std::io::Error),
+ #[cfg(feature = "bzip")]
+ #[error("Failed to decompress with bzip2: {0}")]
+ Bzip2Decompress(#[source] std::io::Error),
+
+ #[cfg(feature = "xz")]
+ #[error("Failed to decompress with xz: {0}")]
+ XzDecompress(#[source] std::io::Error),
+
#[error("Failed to read header: {0}")]
ReadHeader(#[source] std::io::Error),
diff --git a/avro/src/util.rs b/avro/src/util.rs
index 090d5c3..a0bcf29 100644
--- a/avro/src/util.rs
+++ b/avro/src/util.rs
@@ -179,7 +179,7 @@ pub(crate) fn safe_len(len: usize) -> AvroResult<usize> {
Ok(len)
} else {
Err(Details::MemoryAllocation {
- desired: len,
+ desired: Some(len),
maximum: max_bytes,
}
.into())
diff --git a/avro/tests/decompression_bomb.rs b/avro/tests/decompression_bomb.rs
new file mode 100644
index 0000000..b66197e
--- /dev/null
+++ b/avro/tests/decompression_bomb.rs
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Decompression must be bounded by `max_allocation_bytes` so a small
+//! compressed block cannot inflate to an enormous buffer and exhaust memory.
+//!
+//! This is a dedicated integration test so it runs in its own process, where
+//! the (write-once) allocation limit can be set to a small value; the payloads
+//! are then only a few MiB, keeping the test cheap and deterministic.
+
+use apache_avro::Codec;
+use apache_avro::error::Details;
+
+const BUDGET: usize = 1024; // 1 KiB
+const BOMB_PLAINTEXT: usize = 8 * 1024; // 8 KiB, well over the budget
+
+fn set_budget() {
+ apache_avro::util::max_allocation_bytes(BUDGET);
+}
+
+/// Compresses `plaintext` bytes with `codec` and asserts the decompressed
+/// output (larger than the budget) is rejected rather than allocated, then
that
+/// a small payload still round-trips.
+fn assert_bounded(codec: Codec) {
+ set_budget();
+
+ let mut bomb = vec![0u8; BOMB_PLAINTEXT];
+ codec.compress(&mut bomb).expect("compress bomb");
+ assert!(
+ bomb.len() <= BUDGET,
+ "compressed bomb ({} bytes) should be far smaller than the budget",
+ bomb.len()
+ );
+ let result = codec.decompress(&mut bomb);
+ // Assert it fails *specifically* because the output exceeded the budget,
so
+ // a regression that fails decompression for another reason cannot pass.
+ match result {
+ Err(e) => assert!(
+ matches!(e.details(), Details::MemoryAllocation { .. }),
+ "expected a MemoryAllocation error, got {e:?}"
+ ),
+ Ok(()) => panic!("a decompressed output exceeding the budget must be
rejected"),
+ }
+
+ // A payload within the budget must still round-trip.
+ let original = b"apache avro decompression within budget".to_vec();
+ let mut data = original.clone();
+ codec.compress(&mut data).expect("compress small");
+ codec.decompress(&mut data).expect("decompress small");
+ assert_eq!(data, original);
+}
+
+#[test]
+fn deflate_decompression_is_bounded() {
+ use apache_avro::DeflateSettings;
+ assert_bounded(Codec::Deflate(DeflateSettings::default()));
+}
+
+#[test]
+#[cfg(feature = "snappy")]
+fn snappy_decompression_is_bounded() {
+ assert_bounded(Codec::Snappy);
+}
+
+#[test]
+#[cfg(feature = "zstandard")]
+fn zstandard_decompression_is_bounded() {
+ use apache_avro::ZstandardSettings;
+ assert_bounded(Codec::Zstandard(ZstandardSettings::default()));
+}
+
+#[test]
+#[cfg(feature = "bzip")]
+fn bzip2_decompression_is_bounded() {
+ use apache_avro::Bzip2Settings;
+ assert_bounded(Codec::Bzip2(Bzip2Settings::default()));
+}
+
+#[test]
+#[cfg(feature = "xz")]
+fn xz_decompression_is_bounded() {
+ use apache_avro::XzSettings;
+ assert_bounded(Codec::Xz(XzSettings::default()));
+}
