diff --git a/lang/py/avro/io.py b/lang/py/avro/io.py index f5063a6686f..2bac06e77e1 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. + """ + 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_depth} (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: @@ -948,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 new file mode 100644 index 00000000000..39170ee13e1 --- /dev/null +++ b/lang/py/avro/test/test_decode_recursion_depth.py @@ -0,0 +1,97 @@ +#!/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_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: + # 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__": + unittest.main()