This is an automated email from the ASF dual-hosted git repository. martin-g pushed a commit to branch recursion-depth-check-decoding-deserializing in repository https://gitbox.apache.org/repos/asf/avro-rs.git
commit 658b8a32fb34d7c3eb0f44e7784ab3d9ac473695 Author: Martin Tzvetanov Grigorov <[email protected]> AuthorDate: Tue Aug 25 15:43:48 2026 +0300 fix: No recursion depth limit in decode_internal / resolve_internal / serde deserializer A recursive schema can lead to stack overflow (i.e. an abort) in OCF decoding, Value resolving and Schema deserialization. Add `util::max_decode_recursion_depth(usize)` to be able to control the recursipn depth Reported-by: Security scans --- avro/src/decode.rs | 62 ++++++++++++++++++++++- avro/src/error.rs | 5 ++ avro/src/reader/block.rs | 1 + avro/src/reader/datum.rs | 2 + avro/src/reader/single_object.rs | 1 + avro/src/serde/deser_schema/mod.rs | 18 ++++++- avro/src/types.rs | 78 +++++++++++++++++++++++------ avro/src/util.rs | 34 +++++++++++++ avro/tests/recursion_depth_deser.rs | 51 +++++++++++++++++++ avro/tests/recursion_depth_value_resolve.rs | 53 ++++++++++++++++++++ 10 files changed, 285 insertions(+), 20 deletions(-) diff --git a/avro/src/decode.rs b/avro/src/decode.rs index fe71a5d..4d61ac8 100644 --- a/avro/src/decode.rs +++ b/avro/src/decode.rs @@ -25,8 +25,8 @@ use crate::{ schema::{DecimalSchema, EnumSchema, FixedSchema, Name, RecordSchema, ResolvedSchema, Schema}, types::Value, util::{ - DEFAULT_MAX_ALLOCATION_BYTES, max_allocation_bytes, safe_collection_len, safe_len, zag_i32, - zag_i64, + DEFAULT_MAX_ALLOCATION_BYTES, decode_recursion_limit, max_allocation_bytes, + safe_collection_len, safe_len, zag_i32, zag_i64, }, }; use std::{ @@ -78,9 +78,12 @@ fn decode_seq_len<R: Read>(reader: &mut R) -> AvroResult<usize> { /// 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. +#[derive(Debug)] pub(crate) struct DecodeContext { /// Bytes still available for allocations while decoding the current datum. remaining_budget: usize, + /// Current recursion depth of `decode_internal`. + depth: usize, } impl DecodeContext { @@ -88,9 +91,27 @@ impl DecodeContext { pub(crate) fn new() -> Self { Self { remaining_budget: max_allocation_bytes(DEFAULT_MAX_ALLOCATION_BYTES), + depth: 0, } } + /// Track one level of decoding recursion, erroring once the configured + /// maximum depth is exceeded. + fn enter(&mut self) -> AvroResult<()> { + self.depth += 1; + let maximum = decode_recursion_limit(); + if self.depth > maximum { + Err(Details::DecodeRecursionLimit { maximum }.into()) + } else { + Ok(()) + } + } + + /// Leave one level of decoding recursion. + fn leave(&mut self) { + self.depth -= 1; + } + /// 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<()> { @@ -129,6 +150,19 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>( enclosing_namespace: NamespaceRef, reader: &mut R, ctx: &mut DecodeContext, +) -> AvroResult<Value> { + ctx.enter()?; + let value = decode_internal_body(schema, names, enclosing_namespace, reader, ctx); + ctx.leave(); + value +} + +fn decode_internal_body<R: Read, S: Borrow<Schema>>( + schema: &Schema, + names: &HashMap<Name, S>, + enclosing_namespace: NamespaceRef, + reader: &mut R, + ctx: &mut DecodeContext, ) -> AvroResult<Value> { match schema { Schema::Null => Ok(Value::Null), @@ -452,6 +486,7 @@ pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>( #[allow(clippy::expect_fun_call)] mod tests { use crate::schema::{InnerDecimalSchema, UuidSchema}; + use crate::util::decode_recursion_limit; use crate::{ Decimal, decode::decode, @@ -625,6 +660,29 @@ mod tests { Ok(()) } + #[test] + fn avro_rs_641_test_decode_recursion_depth_is_bounded() -> TestResult { + // With a recursive schema, one wire byte per level drives unbounded + // recursion; the decoder must return an error instead of overflowing + // the stack (which would abort the process). + let schema = Schema::parse_str( + r#"{ + "type": "record", + "name": "Node", + "fields": [ + {"name": "next", "type": ["null", "Node"]} + ] + }"#, + )?; + let recursion_depth_trigger = decode_recursion_limit() + 1; + // Each 0x02 byte selects the "Node" union branch, one level deeper. + let payload = vec![0x02u8; recursion_depth_trigger]; + let result = decode(&schema, &mut payload.as_slice()); + assert!(result.is_err(), "unbounded recursion must be rejected"); + + Ok(()) + } + #[test] fn test_decode_map_without_size() -> TestResult { let mut input: &[u8] = &[0x02, 0x08, 0x74, 0x65, 0x73, 0x74, 0x02, 0x00]; diff --git a/avro/src/error.rs b/avro/src/error.rs index 2acb217..6723340 100644 --- a/avro/src/error.rs +++ b/avro/src/error.rs @@ -569,6 +569,11 @@ pub enum Details { #[error("Overflow when decoding integer value")] IntegerOverflow, + #[error( + "Maximum decode recursion depth reached (maximum: {maximum}). Change the limit using `apache_avro::util::max_decode_recursion_depth`" + )] + DecodeRecursionLimit { maximum: usize }, + #[error("Failed to read bytes for decoding variable length integer: {0}")] ReadVariableIntegerBytes(#[source] std::io::Error), diff --git a/avro/src/reader/block.rs b/avro/src/reader/block.rs index 8c9041a..0111b45 100644 --- a/avro/src/reader/block.rs +++ b/avro/src/reader/block.rs @@ -238,6 +238,7 @@ impl<'r, R: Read> Block<'r, R> { let config = Config { names: &self.names_refs, human_readable: self.human_readable, + recursion_depth: 0, }; T::deserialize(SchemaAwareDeserializer::new( &mut block_bytes, diff --git a/avro/src/reader/datum.rs b/avro/src/reader/datum.rs index 36c8fc5..2bf9e39 100644 --- a/avro/src/reader/datum.rs +++ b/avro/src/reader/datum.rs @@ -162,6 +162,7 @@ impl<'s> GenericDatumReader<'s> { Config { names: self.resolved.get_names(), human_readable: self.human_readable, + recursion_depth: 0, }, )?) } @@ -204,6 +205,7 @@ impl<T: AvroSchema + DeserializeOwned> SpecificDatumReader<T> { Config { names: self.resolved.get_names(), human_readable: self.human_readable, + recursion_depth: 0, }, )?) } diff --git a/avro/src/reader/single_object.rs b/avro/src/reader/single_object.rs index bf91238..c8a923e 100644 --- a/avro/src/reader/single_object.rs +++ b/avro/src/reader/single_object.rs @@ -75,6 +75,7 @@ impl GenericSingleObjectReader { let config = Config { names: self.write_schema.get_names(), human_readable: self.human_readable, + recursion_depth: 0, }; T::deserialize(SchemaAwareDeserializer::new( reader, diff --git a/avro/src/serde/deser_schema/mod.rs b/avro/src/serde/deser_schema/mod.rs index 4ebfa25..4b510a5 100644 --- a/avro/src/serde/deser_schema/mod.rs +++ b/avro/src/serde/deser_schema/mod.rs @@ -24,7 +24,7 @@ use crate::{ decode::decode_len, error::Details, schema::{DecimalSchema, InnerDecimalSchema, Name, UnionSchema, UuidSchema}, - util::{safe_len, zag_i32, zag_i64}, + util::{decode_recursion_limit, safe_len, zag_i32, zag_i64}, }; mod block; @@ -50,6 +50,12 @@ pub struct Config<'s, S: Borrow<Schema>> { pub names: &'s HashMap<Name, S>, /// Was the data serialized with `human_readable`. pub human_readable: bool, + /// Current recursion depth of the deserializer. + /// + /// Every nesting level of the deserialized value creates a new + /// [`SchemaAwareDeserializer`], which increments this; the depth is + /// bounded by [`crate::util::max_decode_recursion_depth`]. + pub(crate) recursion_depth: usize, } impl<'s, S: Borrow<Schema>> Config<'s, S> { @@ -88,8 +94,16 @@ impl<'s, 'r, R: Read, S: Borrow<Schema>> SchemaAwareDeserializer<'s, 'r, R, S> { pub fn new( reader: &'r mut R, schema: &'s Schema, - config: Config<'s, S>, + mut config: Config<'s, S>, ) -> Result<Self, Error> { + // Bound the recursion depth so deeply nested (possibly hostile) data + // yields an error instead of exhausting the stack. A recursive schema + // lets roughly one wire byte drive one nesting level. + config.recursion_depth += 1; + let maximum = decode_recursion_limit(); + if config.recursion_depth > maximum { + return Err(Details::DecodeRecursionLimit { maximum }.into()); + } if let Schema::Ref { name } = schema { let schema = config.get_schema(name)?; Ok(Self { diff --git a/avro/src/types.rs b/avro/src/types.rs index c2e0b3a..124fe64 100644 --- a/avro/src/types.rs +++ b/avro/src/types.rs @@ -27,6 +27,7 @@ use crate::{ DecimalSchema, EnumSchema, FixedSchema, Name, Precision, RecordField, RecordSchema, ResolvedSchema, Scale, Schema, SchemaKind, UnionSchema, }, + util::decode_recursion_limit, }; use bigdecimal::BigDecimal; use log::{debug, error}; @@ -744,12 +745,31 @@ impl Value { } pub(crate) fn resolve_internal<S: Borrow<Schema> + Debug>( + self, + schema: &Schema, + names: &HashMap<Name, S>, + enclosing_namespace: NamespaceRef, + field_default: Option<&JsonValue>, + ) -> AvroResult<Self> { + self.resolve_internal_depth(schema, names, enclosing_namespace, field_default, 0) + } + + fn resolve_internal_depth<S: Borrow<Schema> + Debug>( mut self, schema: &Schema, names: &HashMap<Name, S>, enclosing_namespace: NamespaceRef, field_default: Option<&JsonValue>, + depth: usize, ) -> AvroResult<Self> { + // Resolution recurses over both the value and the (possibly + // attacker-supplied) schema; bound the depth so hostile input yields + // an error instead of exhausting the stack. + let depth = depth + 1; + let maximum = decode_recursion_limit(); + if depth > maximum { + return Err(Details::DecodeRecursionLimit { maximum }.into()); + } // Check if this schema is a union, and if the reader schema is not. if SchemaKind::from(&self) == SchemaKind::Union && SchemaKind::from(schema) != SchemaKind::Union @@ -767,7 +787,13 @@ impl Value { if let Some(resolved) = names.get(&name) { debug!("Resolved {name:?}"); - self.resolve_internal(resolved.borrow(), names, name.namespace(), field_default) + self.resolve_internal_depth( + resolved.borrow(), + names, + name.namespace(), + field_default, + depth, + ) } else { error!("Failed to resolve schema {name:?}"); Err(Details::SchemaResolutionError(name.into_owned()).into()) @@ -783,16 +809,21 @@ impl Value { Schema::String => self.resolve_string(), Schema::Fixed(FixedSchema { size, .. }) => self.resolve_fixed(*size), Schema::Union(inner) => { - self.resolve_union(inner, names, enclosing_namespace, field_default) + self.resolve_union(inner, names, enclosing_namespace, field_default, depth) } Schema::Enum(EnumSchema { symbols, default, .. }) => self.resolve_enum(symbols, default, field_default), - Schema::Array(inner) => self.resolve_array(&inner.items, names, enclosing_namespace), - Schema::Map(inner) => self.resolve_map(&inner.types, names, enclosing_namespace), - Schema::Record(RecordSchema { fields, name, .. }) => { - self.resolve_record(fields, names, name.namespace().or(enclosing_namespace)) + Schema::Array(inner) => { + self.resolve_array(&inner.items, names, enclosing_namespace, depth) } + Schema::Map(inner) => self.resolve_map(&inner.types, names, enclosing_namespace, depth), + Schema::Record(RecordSchema { fields, name, .. }) => self.resolve_record( + fields, + names, + name.namespace().or(enclosing_namespace), + depth, + ), Schema::Decimal(DecimalSchema { scale, precision, @@ -1166,6 +1197,7 @@ impl Value { names: &HashMap<Name, S>, enclosing_namespace: NamespaceRef, field_default: Option<&JsonValue>, + depth: usize, ) -> Result<Self, Error> { let v = match self { // Both are unions case. @@ -1182,7 +1214,13 @@ impl Value { Ok(Value::Union( i as u32, - Box::new(v.resolve_internal(inner, names, enclosing_namespace, field_default)?), + Box::new(v.resolve_internal_depth( + inner, + names, + enclosing_namespace, + field_default, + depth, + )?), )) } @@ -1191,12 +1229,15 @@ impl Value { schema: &Schema, names: &HashMap<Name, S>, enclosing_namespace: NamespaceRef, + depth: usize, ) -> Result<Self, Error> { match self { Value::Array(items) => Ok(Value::Array( items .into_iter() - .map(|item| item.resolve_internal(schema, names, enclosing_namespace, None)) + .map(|item| { + item.resolve_internal_depth(schema, names, enclosing_namespace, None, depth) + }) .collect::<Result<_, _>>()?, )), other => Err(Details::GetArray { @@ -1212,6 +1253,7 @@ impl Value { schema: &Schema, names: &HashMap<Name, S>, enclosing_namespace: NamespaceRef, + depth: usize, ) -> Result<Self, Error> { match self { Value::Map(items) => Ok(Value::Map( @@ -1219,7 +1261,7 @@ impl Value { .into_iter() .map(|(key, value)| { value - .resolve_internal(schema, names, enclosing_namespace, None) + .resolve_internal_depth(schema, names, enclosing_namespace, None, depth) .map(|value| (key, value)) }) .collect::<Result<_, _>>()?, @@ -1237,6 +1279,7 @@ impl Value { fields: &[RecordField], names: &HashMap<Name, S>, enclosing_namespace: NamespaceRef, + depth: usize, ) -> Result<Self, Error> { let mut items = match self { Value::Map(items) => Ok(items), @@ -1275,12 +1318,14 @@ impl Value { _ => Value::Union( 0, Box::new( - Value::try_from(value.clone())?.resolve_internal( - first, - names, - enclosing_namespace, - field.default.as_ref(), - )?, + Value::try_from(value.clone())? + .resolve_internal_depth( + first, + names, + enclosing_namespace, + field.default.as_ref(), + depth, + )?, ), ), } @@ -1293,11 +1338,12 @@ impl Value { }, }; value - .resolve_internal( + .resolve_internal_depth( &field.schema, names, enclosing_namespace, field.default.as_ref(), + depth, ) .map(|value| (field.name.clone(), value)) }) diff --git a/avro/src/util.rs b/avro/src/util.rs index 4e4b106..61d44b5 100644 --- a/avro/src/util.rs +++ b/avro/src/util.rs @@ -18,6 +18,7 @@ //! Utility functions, like configuring various global settings. use crate::{AvroResult, error::Details, schema::Documentation}; +use log::warn; use serde_json::{Map, Value}; use std::{ io::{Read, Write}, @@ -32,6 +33,15 @@ use std::{ pub const DEFAULT_MAX_ALLOCATION_BYTES: usize = 512 * 1024 * 1024; static MAX_ALLOCATION_BYTES: OnceLock<usize> = OnceLock::new(); +/// Maximum recursion depth when decoding or resolving Avro-encoded values. +/// +/// This protects against stack exhaustion (an abort, not a catchable error) from deeply nested +/// data: a recursive schema lets an attacker drive one recursion level with roughly one wire byte. +/// +/// See [`max_decode_recursion_depth`] to change this limit. +pub const DEFAULT_MAX_DECODE_RECURSION_DEPTH: usize = 32; +static MAX_DECODE_RECURSION_DEPTH: OnceLock<usize> = OnceLock::new(); + /// Whether to set serialization & deserialization traits as `human_readable` or not. /// /// See [`set_serde_human_readable`] to change this value. @@ -172,6 +182,30 @@ pub fn max_allocation_bytes(num_bytes: usize) -> usize { *MAX_ALLOCATION_BYTES.get_or_init(|| num_bytes) } +/// Set the maximum recursion depth used when decoding or resolving data. +/// +/// This function only changes the setting once. On subsequent calls the value will stay the same +/// as the first time it is called. It is automatically called on first decode and defaults to +/// [`DEFAULT_MAX_DECODE_RECURSION_DEPTH`]. +/// +/// # Returns +/// The configured maximum, which might be different from what the function was called with if the +/// value was already set before. In that case a warning is logged. +pub fn max_decode_recursion_depth(depth: usize) -> usize { + let configured = *MAX_DECODE_RECURSION_DEPTH.get_or_init(|| depth); + if configured != depth { + warn!( + "max_decode_recursion_depth({depth}) has no effect: the limit was already set to {configured}" + ); + } + configured +} + +/// The effective decode recursion limit, initializing the default if unset. +pub(crate) fn decode_recursion_limit() -> usize { + *MAX_DECODE_RECURSION_DEPTH.get_or_init(|| DEFAULT_MAX_DECODE_RECURSION_DEPTH) +} + pub(crate) fn safe_len(len: usize) -> AvroResult<usize> { let max_bytes = max_allocation_bytes(DEFAULT_MAX_ALLOCATION_BYTES); diff --git a/avro/tests/recursion_depth_deser.rs b/avro/tests/recursion_depth_deser.rs new file mode 100644 index 0000000..8ef69e8 --- /dev/null +++ b/avro/tests/recursion_depth_deser.rs @@ -0,0 +1,51 @@ +// 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. + +use apache_avro::Schema; +use apache_avro::reader::datum::GenericDatumReader; +use apache_avro::util::max_decode_recursion_depth; +use apache_avro_test_helper::TestResult; +use serde::{Deserialize, Serialize}; + +// This is an IT test because it sets the default recursion depth limit (OnceLock). + +#[test] +fn avro_rs_641_recursion_depth_is_bounded() -> TestResult { + #[derive(Debug, Deserialize, Serialize, Clone, PartialEq)] + struct Node { + next: Option<Box<Node>>, + } + + let schema = Schema::parse_str( + r#"{ + "type": "record", + "name": "Node", + "fields": [ + {"name": "next", "type": ["null", "Node"]} + ] + }"#, + )?; + let recursion_depth_trigger = max_decode_recursion_depth(16) + 1; + // Each 0x02 byte selects the "Node" union branch, one level deeper. + let payload = vec![0x02u8; recursion_depth_trigger]; + let result = GenericDatumReader::builder(&schema) + .build()? + .read_deser::<Node>(&mut payload.as_slice()); + assert!(result.is_err(), "unbounded recursion must be rejected"); + + Ok(()) +} diff --git a/avro/tests/recursion_depth_value_resolve.rs b/avro/tests/recursion_depth_value_resolve.rs new file mode 100644 index 0000000..b47c0ce --- /dev/null +++ b/avro/tests/recursion_depth_value_resolve.rs @@ -0,0 +1,53 @@ +// 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. + +use apache_avro::Schema; +use apache_avro::types::Value; +use apache_avro::util::max_decode_recursion_depth; +use apache_avro_test_helper::TestResult; + +// This is an IT test because it sets the default recursion depth limit (OnceLock). + +#[test] +fn avro_rs_641_resolve_recursion_depth_is_bounded() -> TestResult { + let schema = Schema::parse_str( + r#"{ + "type": "record", + "name": "Node", + "fields": [ + {"name": "next", "type": ["null", "Node"]} + ] + }"#, + )?; + + let recursion_depth_trigger = max_decode_recursion_depth(16) + 1; + + // Build a value that nests deeper than the recursion limit but is + // still shallow enough for the test itself to drop safely. + let mut value = Value::Record(vec![( + "next".into(), + Value::Union(0, Box::new(Value::Null)), + )]); + for _ in 0..recursion_depth_trigger { + value = Value::Record(vec![("next".into(), Value::Union(1, Box::new(value)))]); + } + + let result = value.resolve(&schema); + assert!(result.is_err(), "unbounded resolution must be rejected"); + + Ok(()) +}
