AVRO-4328: [python] Bound decode recursion depth

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:

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()