You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@avro.apache.org by mg...@apache.org on 2026/08/25 11:23:58 UTC
(avro-rs) branch track-allocated-bytes-during-decoding created (now fecaf4b)
This is an automated email from the ASF dual-hosted git repository.
martin-g pushed a change to branch track-allocated-bytes-during-decoding
in repository https://gitbox.apache.org/repos/asf/avro-rs.git
at fecaf4b fix: Nested collections multiply the 512MB allocation budget arbitrarily
This branch includes the following new commits:
new fecaf4b fix: Nested collections multiply the 512MB allocation budget arbitrarily
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(avro-rs) 01/01: fix: Nested collections multiply the 512MB allocation budget arbitrarily
Posted by mg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martin-g pushed a commit to branch track-allocated-bytes-during-decoding
in repository https://gitbox.apache.org/repos/asf/avro-rs.git
commit fecaf4bd8f0d5341867da8e3166f019fab683168
Author: Martin Tzvetanov Grigorov <mg...@apache.org>
AuthorDate: Tue Aug 25 14:17:39 2026 +0300
fix: Nested collections multiply the 512MB allocation budget arbitrarily
decode.rs::decode_internal (for Schema::Array & Map): safe_collection_len bounds each individual Vec/HashMap against max_allocation_bytes, but no state tracks the running total across the decoded value tree. Elements that cost zero wire bytes (null, empty records, zero-size fixed) let each ~7-byte inner array<null> block materialize ~512MB of live Value slots, so array<array<null>> from an attacker OCF header holds K x 512MB simultaneously from ~130 datum bytes.
Reported-by: Security scans
---
avro/src/decode.rs | 135 ++++++++++++++++++++++++++++++++++++---
avro/src/reader/block.rs | 4 +-
avro/src/reader/datum.rs | 11 +++-
avro/src/reader/single_object.rs | 4 +-
4 files changed, 140 insertions(+), 14 deletions(-)
diff --git a/avro/src/decode.rs b/avro/src/decode.rs
index d78da71..483c5e9 100644
--- a/avro/src/decode.rs
+++ b/avro/src/decode.rs
@@ -24,7 +24,10 @@ use crate::{
error::Details,
schema::{DecimalSchema, EnumSchema, FixedSchema, Name, RecordSchema, ResolvedSchema, Schema},
types::Value,
- util::{safe_collection_len, safe_len, zag_i32, zag_i64},
+ util::{
+ DEFAULT_MAX_ALLOCATION_BYTES, max_allocation_bytes, safe_collection_len, safe_len, zag_i32,
+ zag_i64,
+ },
};
use std::{
borrow::Borrow,
@@ -68,10 +71,56 @@ fn decode_seq_len<R: Read>(reader: &mut R) -> AvroResult<usize> {
)
}
+/// Per-datum decoding state.
+///
+/// Tracks the cumulative number of bytes allocated on behalf of a single
+/// datum, so that nested collections cannot multiply the allocation budget:
+/// every allocation performed while decoding one datum is debited from a
+/// shared budget of [`max_allocation_bytes`] bytes, instead of each
+/// collection only being checked in isolation.
+pub(crate) struct DecodeContext {
+ /// Bytes still available for allocations while decoding the current datum.
+ remaining_budget: usize,
+}
+
+impl DecodeContext {
+ /// Create a fresh context. Call once per datum.
+ pub(crate) fn new() -> Self {
+ Self {
+ remaining_budget: max_allocation_bytes(DEFAULT_MAX_ALLOCATION_BYTES),
+ }
+ }
+
+ /// Debit `bytes` from the per-datum allocation budget, erroring when the
+ /// cumulative allocations for this datum would exceed it.
+ fn debit(&mut self, bytes: usize) -> AvroResult<()> {
+ match self.remaining_budget.checked_sub(bytes) {
+ Some(remaining) => {
+ self.remaining_budget = remaining;
+ Ok(())
+ }
+ None => Err(Details::MemoryAllocation {
+ desired: Some(bytes),
+ maximum: max_allocation_bytes(DEFAULT_MAX_ALLOCATION_BYTES),
+ }
+ .into()),
+ }
+ }
+
+ /// Debit the budget for `items` collection elements of type `T`.
+ fn debit_items<T>(&mut self, items: usize) -> AvroResult<()> {
+ let bytes = items
+ .checked_mul(size_of::<T>())
+ .ok_or(Details::IntegerOverflow)?;
+ self.debit(bytes)
+ }
+}
+
/// Decode a `Value` from avro format given its `Schema`.
pub fn decode<R: Read>(schema: &Schema, reader: &mut R) -> AvroResult<Value> {
let rs = ResolvedSchema::try_from(schema)?;
- decode_internal(schema, rs.get_names(), None, reader)
+ let mut ctx = DecodeContext::new();
+ decode_internal(schema, rs.get_names(), None, reader, &mut ctx)
}
pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
@@ -79,6 +128,7 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
names: &HashMap<Name, S>,
enclosing_namespace: NamespaceRef,
reader: &mut R,
+ ctx: &mut DecodeContext,
) -> AvroResult<Value> {
match schema {
Schema::Null => Ok(Value::Null),
@@ -106,27 +156,28 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
names,
enclosing_namespace,
reader,
+ ctx,
)? {
Value::Fixed(_, bytes) => Ok(Value::Decimal(Decimal::from(bytes))),
value => Err(Details::FixedValue(value).into()),
}
}
InnerDecimalSchema::Bytes => {
- match decode_internal(&Schema::Bytes, names, enclosing_namespace, reader)? {
+ match decode_internal(&Schema::Bytes, names, enclosing_namespace, reader, ctx)? {
Value::Bytes(bytes) => Ok(Value::Decimal(Decimal::from(bytes))),
value => Err(Details::BytesValue(value).into()),
}
}
},
Schema::BigDecimal => {
- match decode_internal(&Schema::Bytes, names, enclosing_namespace, reader)? {
+ match decode_internal(&Schema::Bytes, names, enclosing_namespace, reader, ctx)? {
Value::Bytes(bytes) => deserialize_big_decimal(&bytes).map(Value::BigDecimal),
value => Err(Details::BytesValue(value).into()),
}
}
Schema::Uuid(UuidSchema::String) => {
let Value::String(string) =
- decode_internal(&Schema::String, names, enclosing_namespace, reader)?
+ decode_internal(&Schema::String, names, enclosing_namespace, reader, ctx)?
else {
// decoding a String can also return a Null, indicating EOF
return Err(Error::new(Details::ReadBytes(std::io::Error::from(
@@ -138,7 +189,7 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
}
Schema::Uuid(UuidSchema::Bytes) => {
let Value::Bytes(bytes) =
- decode_internal(&Schema::Bytes, names, enclosing_namespace, reader)?
+ decode_internal(&Schema::Bytes, names, enclosing_namespace, reader, ctx)?
else {
unreachable!(
"decode_internal(Schema::Bytes) can only return a Value::Bytes or an error"
@@ -153,6 +204,7 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
names,
enclosing_namespace,
reader,
+ ctx,
)?
else {
unreachable!(
@@ -205,12 +257,14 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
}
Schema::Bytes => {
let len = decode_len(reader)?;
+ ctx.debit(len)?;
let mut buf = vec![0u8; len];
reader.read_exact(&mut buf).map_err(Details::ReadBytes)?;
Ok(Value::Bytes(buf))
}
Schema::String => {
let len = decode_len(reader)?;
+ ctx.debit(len)?;
let mut buf = vec![0u8; len];
match reader.read_exact(&mut buf) {
Ok(_) => Ok(Value::String(
@@ -226,6 +280,10 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
}
}
Schema::Fixed(FixedSchema { size, .. }) => {
+ // The size is schema-declared, not wire-declared, but the schema
+ // itself may be attacker-supplied (e.g. an OCF header), so it must
+ // be debited from the allocation budget like any other length.
+ ctx.debit(*size)?;
let mut buf = vec![0u8; *size];
reader
.read_exact(&mut buf)
@@ -247,6 +305,10 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
.checked_add(len)
.ok_or(Details::IntegerOverflow)?;
safe_collection_len::<Value>(total)?;
+ // Elements can be arbitrarily cheap on the wire (e.g. null),
+ // so also debit the per-datum budget: nested collections must
+ // not multiply the allocation budget.
+ ctx.debit_items::<Value>(len)?;
// Use reserve_exact as reserve can allocate more than needed defeating the purpose
// of the previous check
items.reserve_exact(len);
@@ -256,6 +318,7 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
names,
enclosing_namespace,
reader,
+ ctx,
)?);
}
}
@@ -279,13 +342,21 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
.checked_add(len)
.ok_or(Details::IntegerOverflow)?;
safe_collection_len::<(String, Value)>(total)?;
+ // See the Array arm: nested collections share one budget.
+ ctx.debit_items::<(String, Value)>(len)?;
items.reserve(len);
for _ in 0..len {
- match decode_internal(&Schema::String, names, enclosing_namespace, reader)? {
+ match decode_internal(&Schema::String, names, enclosing_namespace, reader, ctx)?
+ {
Value::String(key) => {
- let value =
- decode_internal(&inner.types, names, enclosing_namespace, reader)?;
+ let value = decode_internal(
+ &inner.types,
+ names,
+ enclosing_namespace,
+ reader,
+ ctx,
+ )?;
items.insert(key, value);
}
value => return Err(Details::MapKeyType(value.into()).into()),
@@ -304,7 +375,7 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
index,
num_variants: variants.len(),
})?;
- let value = decode_internal(variant, names, enclosing_namespace, reader)?;
+ let value = decode_internal(variant, names, enclosing_namespace, reader, ctx)?;
Ok(Value::Union(index as u32, Box::new(value)))
}
Err(Details::ReadVariableIntegerBytes(io_err)) => {
@@ -318,9 +389,13 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
},
Schema::Record(RecordSchema { name, fields, .. }) => {
let fully_qualified_name = name.fully_qualified_name(enclosing_namespace);
+ // Records can consume zero wire bytes (e.g. all-null fields), so
+ // debit the budget for the field vector and the cloned names.
+ ctx.debit_items::<(String, Value)>(fields.len())?;
// Benchmarks indicate ~10% improvement using this method.
let mut items = Vec::with_capacity(fields.len());
for field in fields {
+ ctx.debit(field.name.len())?;
// TODO: This clone is also expensive. See if we can do away with it...
items.push((
field.name.clone(),
@@ -329,6 +404,7 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
names,
fully_qualified_name.namespace(),
reader,
+ ctx,
)?,
));
}
@@ -339,6 +415,9 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
let index = usize::try_from(raw_index)
.map_err(|e| Details::ConvertI32ToUsize(e, raw_index))?;
if (0..symbols.len()).contains(&index) {
+ // Cloning the symbol allocates without consuming wire
+ // bytes, so it counts against the per-datum budget.
+ ctx.debit(symbols[index].len())?;
let symbol = symbols[index].clone();
Value::Enum(raw_index as u32, symbol)
} else {
@@ -360,6 +439,7 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
names,
fully_qualified_name.namespace(),
reader,
+ ctx,
)
} else {
Err(Details::SchemaResolutionError(fully_qualified_name.into_owned()).into())
@@ -472,6 +552,41 @@ mod tests {
Ok(())
}
+ #[test]
+ fn avro_rs_639_test_nested_collections_share_one_allocation_budget() -> TestResult {
+ use crate::util::{DEFAULT_MAX_ALLOCATION_BYTES, max_allocation_bytes};
+
+ // Each inner array<null> block passes the per-collection check on its
+ // own, but the shared per-datum budget must reject the cumulative
+ // total: elements of type null cost zero wire bytes, so without a
+ // cumulative budget a handful of ~10-byte inner arrays would pin an
+ // unbounded multiple of the allocation limit in memory at once.
+ let budget = max_allocation_bytes(DEFAULT_MAX_ALLOCATION_BYTES);
+ let inner_count = (budget / 2) / size_of::<Value>() + 1;
+
+ let inner_arrays_count = 2;
+ let mut payload = Vec::new();
+ // Outer array: a single block declaring two inner arrays.
+ crate::util::zig_i64(inner_arrays_count, &mut payload)?;
+ for _ in 0..inner_arrays_count {
+ payload.extend(create_block(inner_count as i64));
+ }
+ // Outer array terminator.
+ payload.push(0x00);
+
+ let result = decode(
+ &Schema::array(Schema::array(Schema::Null).build()).build(),
+ &mut payload.as_slice(),
+ );
+
+ assert!(
+ result.is_err(),
+ "nested collections must share one allocation budget, got {result:?}"
+ );
+
+ Ok(())
+ }
+
#[test]
fn test_decode_array_int64_min_block_count_is_rejected() -> TestResult {
// i64::MIN as a negative block count cannot be negated (checked_neg
diff --git a/avro/src/reader/block.rs b/avro/src/reader/block.rs
index 5e9ae1b..8c9041a 100644
--- a/avro/src/reader/block.rs
+++ b/avro/src/reader/block.rs
@@ -27,7 +27,7 @@ use serde_json::from_slice;
use crate::{
AvroResult, Codec, Error,
- decode::{decode, decode_internal},
+ decode::{DecodeContext, decode, decode_internal},
error::Details,
schema::{Names, Schema, resolve_names, resolve_names_with_schemata},
serde::deser_schema::{Config, SchemaAwareDeserializer},
@@ -195,11 +195,13 @@ impl<'r, R: Read> Block<'r, R> {
let mut block_bytes = &self.buf[self.buf_idx..];
let b_original = block_bytes.len();
+ let mut ctx = DecodeContext::new();
let item = decode_internal(
&self.writer_schema,
&self.names_refs,
None,
&mut block_bytes,
+ &mut ctx,
)?;
let item = match read_schema {
Some(schema) => item.resolve(schema)?,
diff --git a/avro/src/reader/datum.rs b/avro/src/reader/datum.rs
index 4f0c31e..36c8fc5 100644
--- a/avro/src/reader/datum.rs
+++ b/avro/src/reader/datum.rs
@@ -22,7 +22,7 @@ use serde::de::DeserializeOwned;
use crate::{
AvroResult, AvroSchema, Schema,
- decode::decode_internal,
+ decode::{DecodeContext, decode_internal},
schema::{ResolvedOwnedSchema, ResolvedSchema},
serde::deser_schema::{Config, SchemaAwareDeserializer},
types::Value,
@@ -130,7 +130,14 @@ impl<'s, S: generic_datum_reader_builder::State> GenericDatumReaderBuilder<'s, S
impl<'s> GenericDatumReader<'s> {
/// Read a Avro datum from the reader.
pub fn read_value<R: Read>(&self, reader: &mut R) -> AvroResult<Value> {
- let value = decode_internal(self.writer, self.resolved.get_names(), None, reader)?;
+ let mut ctx = DecodeContext::new();
+ let value = decode_internal(
+ self.writer,
+ self.resolved.get_names(),
+ None,
+ reader,
+ &mut ctx,
+ )?;
if let Some((reader, resolved)) = &self.reader {
value.resolve_internal(reader, resolved.get_names(), None, None)
} else {
diff --git a/avro/src/reader/single_object.rs b/avro/src/reader/single_object.rs
index d65ca2f..bf91238 100644
--- a/avro/src/reader/single_object.rs
+++ b/avro/src/reader/single_object.rs
@@ -22,7 +22,7 @@ use serde::de::DeserializeOwned;
use crate::{
AvroResult, AvroSchema, Schema,
- decode::decode_internal,
+ decode::{DecodeContext, decode_internal},
error::Details,
headers::{HeaderBuilder, RabinFingerprintHeader},
schema::ResolvedOwnedSchema,
@@ -60,11 +60,13 @@ impl GenericSingleObjectReader {
impl GenericSingleObjectReader {
pub fn read_value<R: Read>(&self, reader: &mut R) -> AvroResult<Value> {
self.read_header(reader)?;
+ let mut ctx = DecodeContext::new();
decode_internal(
self.write_schema.get_root_schema(),
self.write_schema.get_names(),
None,
reader,
+ &mut ctx,
)
}