From 82c0ea68ba6d85c34c4b9501ea75c9c35c36e5c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Fri, 7 Aug 2026 16:22:00 +0200 Subject: [PATCH 1/2] AVRO-4328: [python] Bound decode recursion depth Recursive schemas (e.g. a linked list or tree) let a small, hostile payload drive arbitrarily deep nesting during binary decoding, exhausting the Python call stack (RecursionError, or a fatal interpreter crash once the C stack is exhausted) before any allocation limit is reached. Bound the decode nesting depth by counting structural descents into records, arrays, maps and unions and rejecting input that nests deeper than the limit with a clean AvroException. The default is 100 (matching Protocol Buffers and the Java SDK) and is configurable via the AVRO_MAX_DECODE_DEPTH environment variable. The depth is tracked per DatumReader for the current datum and reset at the start of each top-level read(), and restored on exit so a reader can be reused. --- lang/py/avro/io.py | 79 ++++++++++++++++- .../avro/test/test_decode_recursion_depth.py | 84 +++++++++++++++++++ 2 files changed, 159 insertions(+), 4 deletions(-) create mode 100644 lang/py/avro/test/test_decode_recursion_depth.py diff --git a/lang/py/avro/io.py b/lang/py/avro/io.py index f5063a6686f..1b5398fec07 100644 --- a/lang/py/avro/io.py +++ b/lang/py/avro/io.py @@ -85,6 +85,7 @@ """ import collections +import contextlib import datetime import decimal import os @@ -742,6 +743,41 @@ def _collection_limits() -> Tuple[int, int]: return parsed, parsed +# Environment variable overriding the maximum decode nesting depth. +MAX_DECODE_DEPTH_ENV = "AVRO_MAX_DECODE_DEPTH" + +# Default maximum decode nesting depth. A recursive schema (e.g. a linked list or +# a tree) lets a small, hostile payload drive arbitrarily deep nesting during +# decoding, exhausting the Python call stack (RecursionError, or a fatal +# interpreter crash once the C stack is exhausted) before any allocation limit is +# reached. Bounding the depth turns such input into a clean, catchable Avro error. +# The default comfortably exceeds any realistic schema nesting while staying well +# below the interpreter's recursion ceiling, and aligns with the value used by +# Protocol Buffers. Configure with the ``AVRO_MAX_DECODE_DEPTH`` environment +# variable. +DEFAULT_MAX_DECODE_DEPTH = 100 + + +def _max_decode_depth() -> int: + """Return the configured maximum decode nesting depth. + + Overridable with ``AVRO_MAX_DECODE_DEPTH`` (a positive integer). Invalid or + non-positive values are ignored with a warning and the default is used. + """ + value = os.environ.get(MAX_DECODE_DEPTH_ENV) + if value is None: + return DEFAULT_MAX_DECODE_DEPTH + try: + parsed = int(value) + except ValueError: + warnings.warn(avro.errors.AvroWarning(f"Ignoring invalid {MAX_DECODE_DEPTH_ENV} value: {value!r}")) + return DEFAULT_MAX_DECODE_DEPTH + if parsed <= 0: + warnings.warn(avro.errors.AvroWarning(f"Ignoring non-positive {MAX_DECODE_DEPTH_ENV} value: {value!r}")) + return DEFAULT_MAX_DECODE_DEPTH + return parsed + + def _max_collection_items() -> int: """Return the configured zero-byte-element collection limit.""" return _collection_limits()[0] @@ -816,6 +852,12 @@ def __init__(self, writers_schema: Optional[avro.schema.Schema] = None, readers_ # limit but together unbounded. The cap is therefore applied across the # whole datum. See _ensure_collection_available. self._zero_byte_items_read = 0 + # Current decode nesting depth (structural descents into records, arrays, + # maps and unions) for the datum being read. Reset at the start of each + # top-level read() and bounded by _max_decode_depth() so a recursive + # schema fed deeply nested data fails with a clean error instead of + # exhausting the call stack. See _nested_read. + self._read_depth = 0 @property def writers_schema(self) -> Optional[avro.schema.Schema]: @@ -845,8 +887,33 @@ def read(self, decoder: "BinaryDecoder") -> object: # many small collection fields cannot bypass it (see # _ensure_collection_available). self._zero_byte_items_read = 0 + # Start a fresh nesting-depth budget for this datum (see _nested_read). + self._read_depth = 0 return self.read_data(self.writers_schema, reader_schema, decoder) + @contextlib.contextmanager + def _nested_read(self) -> Generator[None, None, None]: + """Track one level of structural nesting while decoding. + + Entering a record, array, map or union grows the (recursive) Python call + stack. Bounding the nesting depth turns a recursive schema fed a deeply + nested payload into a clean, catchable error instead of a RecursionError + or a fatal interpreter crash from C-stack exhaustion. The check runs + before incrementing so the counter stays balanced as the exception + unwinds the enclosing (already-entered) levels, and the depth is restored + on exit so a reader instance can be reused. + """ + if self._read_depth >= _max_decode_depth(): + raise avro.errors.AvroException( + f"Decode nesting depth exceeds the maximum allowed of {_max_decode_depth()} " + f"(configure with the {MAX_DECODE_DEPTH_ENV} environment variable)" + ) + self._read_depth += 1 + try: + yield + finally: + self._read_depth -= 1 + def read_data(self, writers_schema: avro.schema.Schema, readers_schema: avro.schema.Schema, decoder: "BinaryDecoder") -> object: # schema matching if not readers_schema.match(writers_schema): @@ -856,7 +923,8 @@ def read_data(self, writers_schema: avro.schema.Schema, readers_schema: avro.sch # function dispatch for reading data based on type of writer's schema if isinstance(writers_schema, avro.schema.UnionSchema) and isinstance(readers_schema, avro.schema.UnionSchema): - return self.read_union(writers_schema, readers_schema, decoder) + with self._nested_read(): + return self.read_union(writers_schema, readers_schema, decoder) if isinstance(readers_schema, avro.schema.UnionSchema): # schema resolution: reader's schema is a union, writer's schema is not @@ -918,12 +986,15 @@ def read_data(self, writers_schema: avro.schema.Schema, readers_schema: avro.sch if isinstance(writers_schema, avro.schema.EnumSchema) and isinstance(readers_schema, avro.schema.EnumSchema): return self.read_enum(writers_schema, readers_schema, decoder) if isinstance(writers_schema, avro.schema.ArraySchema) and isinstance(readers_schema, avro.schema.ArraySchema): - return self.read_array(writers_schema, readers_schema, decoder) + with self._nested_read(): + return self.read_array(writers_schema, readers_schema, decoder) if isinstance(writers_schema, avro.schema.MapSchema) and isinstance(readers_schema, avro.schema.MapSchema): - return self.read_map(writers_schema, readers_schema, decoder) + with self._nested_read(): + return self.read_map(writers_schema, readers_schema, decoder) if isinstance(writers_schema, avro.schema.RecordSchema) and isinstance(readers_schema, avro.schema.RecordSchema): # .type in ["record", "error", "request"]: - return self.read_record(writers_schema, readers_schema, decoder) + with self._nested_read(): + return self.read_record(writers_schema, readers_schema, decoder) raise avro.errors.AvroException(f"Cannot read unknown schema type: {writers_schema.type}") def skip_data(self, writers_schema: avro.schema.Schema, decoder: BinaryDecoder) -> None: diff --git a/lang/py/avro/test/test_decode_recursion_depth.py b/lang/py/avro/test/test_decode_recursion_depth.py new file mode 100644 index 00000000000..f3653dcdacd --- /dev/null +++ b/lang/py/avro/test/test_decode_recursion_depth.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 + +## +# 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 +# +# https://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. + +"""AVRO-4302: bound decode recursion depth to prevent stack exhaustion.""" + +import io +import json +import unittest + +import avro.errors +import avro.io +import avro.schema + + +def _encode_long(value: int) -> bytes: + """Zig-zag + varint encode a long, matching BinaryEncoder.write_long.""" + datum = (value << 1) ^ (value >> 63) + out = bytearray() + while (datum & ~0x7F) != 0: + out.append((datum & 0x7F) | 0x80) + datum >>= 7 + out.append(datum) + return bytes(out) + + +class TestDecodeRecursionDepth(unittest.TestCase): + # A self-referencing linked-list schema: the classic recursion-bomb shape. + NODE = avro.schema.parse(json.dumps({"type": "record", "name": "Node", "fields": [{"name": "next", "type": ["null", "Node"]}]})) + + @staticmethod + def _linked_list(depth: int) -> bytes: + """Encode a Node linked list nested ``depth`` levels deep. + + Each level selects the ``Node`` union branch (index 1); the final level + selects ``null`` (index 0) to terminate. ~1 byte per level. + """ + return _encode_long(1) * depth + _encode_long(0) + + def _read(self, data: bytes) -> object: + reader = avro.io.DatumReader(self.NODE, self.NODE) + return reader.read(avro.io.BinaryDecoder(io.BytesIO(data))) + + def test_deeply_nested_input_rejected_with_bounded_error(self) -> None: + # ~100k levels: far beyond the default depth limit and enough to overflow + # the stack if it were left unbounded, yet only ~100kB of input. Must fail + # with a bounded AvroException rather than a RecursionError / crash. + bomb = self._linked_list(100_000) + self.assertRaises(avro.errors.AvroException, self._read, bomb) + + def test_moderately_nested_input_within_limit_still_decodes(self) -> None: + # Two structural descents (union + record) are counted per list level, so + # keep the level count well under half the default limit. + result = self._read(self._linked_list(20)) + self.assertIsInstance(result, dict) + + def test_custom_depth_limit_env_is_honored(self) -> None: + import os + + os.environ[avro.io.MAX_DECODE_DEPTH_ENV] = "6" + try: + # 6 allows only 3 list levels (2 descents each); 10 levels must fail. + self.assertRaises(avro.errors.AvroException, self._read, self._linked_list(10)) + finally: + del os.environ[avro.io.MAX_DECODE_DEPTH_ENV] + + +if __name__ == "__main__": + unittest.main() From 710eace4f3dad0d1dc3d26b7db4312f426739a3e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Fri, 7 Aug 2026 18:52:16 +0200 Subject: [PATCH 2/2] AVRO-4328: [python] Bound recursion depth on the skip path Address review feedback: - Guard the structural skip path (skip_array/skip_map/skip_union/ skip_record) with the same _nested_read depth bound, so skipping a writer-only field for a recursive schema during resolution cannot overflow the stack. Add a regression test. - Compute _max_decode_depth() once in _nested_read so the check and the error message cannot disagree if the environment changes between calls. - Preserve and restore any pre-existing AVRO_MAX_DECODE_DEPTH value in the env-override test instead of unconditionally deleting it. --- lang/py/avro/io.py | 18 +++++++++++------- .../avro/test/test_decode_recursion_depth.py | 15 ++++++++++++++- 2 files changed, 25 insertions(+), 8 deletions(-) diff --git a/lang/py/avro/io.py b/lang/py/avro/io.py index 1b5398fec07..2bac06e77e1 100644 --- a/lang/py/avro/io.py +++ b/lang/py/avro/io.py @@ -903,10 +903,10 @@ def _nested_read(self) -> Generator[None, None, None]: unwinds the enclosing (already-entered) levels, and the depth is restored on exit so a reader instance can be reused. """ - if self._read_depth >= _max_decode_depth(): + max_depth = _max_decode_depth() + if self._read_depth >= max_depth: raise avro.errors.AvroException( - f"Decode nesting depth exceeds the maximum allowed of {_max_decode_depth()} " - f"(configure with the {MAX_DECODE_DEPTH_ENV} environment variable)" + f"Decode nesting depth exceeds the maximum allowed of {max_depth} (configure with the {MAX_DECODE_DEPTH_ENV} environment variable)" ) self._read_depth += 1 try: @@ -1019,13 +1019,17 @@ def skip_data(self, writers_schema: avro.schema.Schema, decoder: BinaryDecoder) if isinstance(writers_schema, avro.schema.EnumSchema): return self.skip_enum(writers_schema, decoder) if isinstance(writers_schema, avro.schema.ArraySchema): - return self.skip_array(writers_schema, decoder) + with self._nested_read(): + return self.skip_array(writers_schema, decoder) if isinstance(writers_schema, avro.schema.MapSchema): - return self.skip_map(writers_schema, decoder) + with self._nested_read(): + return self.skip_map(writers_schema, decoder) if isinstance(writers_schema, avro.schema.UnionSchema): - return self.skip_union(writers_schema, decoder) + with self._nested_read(): + return self.skip_union(writers_schema, decoder) if isinstance(writers_schema, avro.schema.RecordSchema): - return self.skip_record(writers_schema, decoder) + with self._nested_read(): + return self.skip_record(writers_schema, decoder) raise avro.errors.AvroException(f"Unknown schema type: {writers_schema.type}") def read_fixed(self, writers_schema: avro.schema.FixedSchema, readers_schema: avro.schema.Schema, decoder: BinaryDecoder) -> bytes: diff --git a/lang/py/avro/test/test_decode_recursion_depth.py b/lang/py/avro/test/test_decode_recursion_depth.py index f3653dcdacd..39170ee13e1 100644 --- a/lang/py/avro/test/test_decode_recursion_depth.py +++ b/lang/py/avro/test/test_decode_recursion_depth.py @@ -69,15 +69,28 @@ def test_moderately_nested_input_within_limit_still_decodes(self) -> None: result = self._read(self._linked_list(20)) self.assertIsInstance(result, dict) + def test_deeply_nested_input_rejected_when_skipped(self) -> None: + # The skip path (a writer-only field during resolution) descends + # recursively too and must be bounded as well. + bomb = self._linked_list(100_000) + reader = avro.io.DatumReader(self.NODE, self.NODE) + decoder = avro.io.BinaryDecoder(io.BytesIO(bomb)) + self.assertRaises(avro.errors.AvroException, reader.skip_data, self.NODE, decoder) + def test_custom_depth_limit_env_is_honored(self) -> None: import os + prior = os.environ.get(avro.io.MAX_DECODE_DEPTH_ENV) os.environ[avro.io.MAX_DECODE_DEPTH_ENV] = "6" try: # 6 allows only 3 list levels (2 descents each); 10 levels must fail. self.assertRaises(avro.errors.AvroException, self._read, self._linked_list(10)) finally: - del os.environ[avro.io.MAX_DECODE_DEPTH_ENV] + # Restore any pre-existing value instead of unconditionally clearing it. + if prior is None: + del os.environ[avro.io.MAX_DECODE_DEPTH_ENV] + else: + os.environ[avro.io.MAX_DECODE_DEPTH_ENV] = prior if __name__ == "__main__":