[EventHub][ServiceBus] Bound pyAMQP decoder recursion depth

sdk/eventhub/azure-eventhub/CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Fixed a bug where the async pure-Python AMQP transport failed to connect with `[Errno 22] Invalid argument` (`amqp:socket-error`) inside containerized/virtualized environments such as Docker Desktop on macOS. The transport no longer reads back and re-applies platform-negotiated TCP options (e.g. `TCP_MAXSEG`) that some platforms reject via `setsockopt`. Also fixed the async transport to apply default TCP socket settings even when no custom `socket_settings` are provided. ([#45394](https://github.com/Azure/azure-sdk-for-python/issues/45394))
 - Fixed a bug in the pyAMQP transport where decoding an incoming performative whose trailing null fields were omitted by the sender (permitted by AMQP 1.0 section 1.4) raised `IndexError`/`TypeError`. The decoded field list is now padded to the performative's full field count so omitted trailing fields read back as their AMQP-defined default, including the compact `list0` encoding where every field is omitted. A field encoded as an explicit null but whose declared default is non-null (for example a `max_frame_size` set to null so the connection would compare `None < 512`) now also reads back as that default.
+- Bounded the recursion depth of the pyAMQP decoder so a deeply nested message cannot exhaust the interpreter stack (`RecursionError`) during decode. Nesting beyond a fixed limit now raises a clear `ValueError`, complementing the existing element-count cap.
 
 ## 5.15.1 (2025-11-11)
 

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py

@@ -7,6 +7,7 @@
 import uuid
 import logging
 import decimal
+import threading
 from typing import (
     Callable,
     List,
@@ -74,6 +75,32 @@
 DECIMAL128_MAX_DIGITS = 34
 DECIMAL128_BIAS = 6176
 
+# --- Nesting-depth guard: bound recursion of the recursive-descent decoder.
+# Complements the _MAX_COMPOUND_COUNT element-count cap. A malicious peer can nest
+# compound types (list/map/array/described) arbitrarily deep; each level is ~1-3 wire
+# bytes but one Python recursion frame, so a small message can exhaust the interpreter
+# stack (RecursionError) before any count cap applies. ---
+_MAX_NESTED_DEPTH = 64
+_decode_depth = threading.local()
+
+
+def _depth_guarded(fn):
+    """Bound the recursion depth of a compound-type decoder. Thread-local so concurrent
+    connections do not race; balanced by try/finally (decode is synchronous, no awaits)."""
+    def _wrapper(buffer, *args, **kwargs):
+        depth = getattr(_decode_depth, "value", 0) + 1
+        if depth > _MAX_NESTED_DEPTH:
+            raise ValueError(
+                f"AMQP nested compound depth exceeds maximum {_MAX_NESTED_DEPTH}"
+            )
+        _decode_depth.value = depth
+        try:
+            return fn(buffer, *args, **kwargs)
+        finally:
+            _decode_depth.value = depth - 1
+    return _wrapper
+
+
 
 
 def _decode_null(buffer: memoryview) -> Tuple[memoryview, None]:
@@ -232,6 +259,7 @@ def _decode_decimal128(buffer: memoryview) -> Tuple[memoryview, decimal.Decimal]
     with decimal.localcontext(decimal_ctx) as ctx:
         return buffer[16:], ctx.create_decimal((sign, digits, exponent))
 
+@_depth_guarded
 def _decode_list_small(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     count = buffer[1]
     buffer = buffer[2:]
@@ -241,6 +269,7 @@ def _decode_list_small(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     return buffer, values
 
 
+@_depth_guarded
 def _decode_list_large(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     count = c_unsigned_long.unpack(buffer[4:8])[0]
     # Validate the wire-supplied count before allocating `[None] * count`,
@@ -256,6 +285,7 @@ def _decode_list_large(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     return buffer, values
 
 
+@_depth_guarded
 def _decode_map_small(buffer: memoryview) -> Tuple[memoryview, Dict[Any, Any]]:
     raw_count = buffer[1]
     if raw_count % 2 != 0:
@@ -272,6 +302,7 @@ def _decode_map_small(buffer: memoryview) -> Tuple[memoryview, Dict[Any, Any]]:
     return buffer, values
 
 
+@_depth_guarded
 def _decode_map_large(buffer: memoryview) -> Tuple[memoryview, Dict[Any, Any]]:
     # Validate the raw on-wire count *before* halving it (the AMQP encoding
     # stores total entries; pairs = entries / 2). Checking pre-halve keeps
@@ -298,6 +329,7 @@ def _decode_map_large(buffer: memoryview) -> Tuple[memoryview, Dict[Any, Any]]:
     return buffer, values
 
 
+@_depth_guarded
 def _decode_array_small(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     count = buffer[1]  # Ignore first byte (size) and just rely on count
     if count:
@@ -319,6 +351,7 @@ def _decode_array_small(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     return buffer[2:], []
 
 
+@_depth_guarded
 def _decode_array_large(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     count = c_unsigned_long.unpack(buffer[4:8])[0]
     # Validate the wire-supplied count before allocating `[None] * count`.
@@ -347,6 +380,7 @@ def _decode_array_large(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     return buffer[8:], []
 
 
+@_depth_guarded
 def _decode_described(buffer: memoryview) -> Tuple[memoryview, object]:
     # TODO: to move the cursor of the buffer to the described value based on size of the
     #  descriptor without decoding descriptor value

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -8,6 +8,8 @@
 
 ### Bugs Fixed
 
+- Bounded the recursion depth of the pyAMQP decoder so a deeply nested message cannot exhaust the interpreter stack (`RecursionError`) during decode. Nesting beyond a fixed limit now raises a clear `ValueError`, complementing the existing element-count cap.
+
 ### Other Changes
 
 ## 7.15.0b2 (2026-08-21)

sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py

@@ -7,6 +7,7 @@
 import uuid
 import logging
 import decimal
+import threading
 from typing import (
     Callable,
     List,
@@ -74,6 +75,32 @@
 DECIMAL128_MAX_DIGITS = 34
 DECIMAL128_BIAS = 6176
 
+# --- Nesting-depth guard: bound recursion of the recursive-descent decoder.
+# Complements the _MAX_COMPOUND_COUNT element-count cap. A malicious peer can nest
+# compound types (list/map/array/described) arbitrarily deep; each level is ~1-3 wire
+# bytes but one Python recursion frame, so a small message can exhaust the interpreter
+# stack (RecursionError) before any count cap applies. ---
+_MAX_NESTED_DEPTH = 64
+_decode_depth = threading.local()
+
+
+def _depth_guarded(fn):
+    """Bound the recursion depth of a compound-type decoder. Thread-local so concurrent
+    connections do not race; balanced by try/finally (decode is synchronous, no awaits)."""
+    def _wrapper(buffer, *args, **kwargs):
+        depth = getattr(_decode_depth, "value", 0) + 1
+        if depth > _MAX_NESTED_DEPTH:
+            raise ValueError(
+                f"AMQP nested compound depth exceeds maximum {_MAX_NESTED_DEPTH}"
+            )
+        _decode_depth.value = depth
+        try:
+            return fn(buffer, *args, **kwargs)
+        finally:
+            _decode_depth.value = depth - 1
+    return _wrapper
+
+
 
 
 def _decode_null(buffer: memoryview) -> Tuple[memoryview, None]:
@@ -232,6 +259,7 @@ def _decode_decimal128(buffer: memoryview) -> Tuple[memoryview, decimal.Decimal]
     with decimal.localcontext(decimal_ctx) as ctx:
         return buffer[16:], ctx.create_decimal((sign, digits, exponent))
 
+@_depth_guarded
 def _decode_list_small(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     count = buffer[1]
     buffer = buffer[2:]
@@ -241,6 +269,7 @@ def _decode_list_small(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     return buffer, values
 
 
+@_depth_guarded
 def _decode_list_large(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     count = c_unsigned_long.unpack(buffer[4:8])[0]
     # Validate the wire-supplied count before allocating `[None] * count`,
@@ -256,6 +285,7 @@ def _decode_list_large(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     return buffer, values
 
 
+@_depth_guarded
 def _decode_map_small(buffer: memoryview) -> Tuple[memoryview, Dict[Any, Any]]:
     raw_count = buffer[1]
     if raw_count % 2 != 0:
@@ -272,6 +302,7 @@ def _decode_map_small(buffer: memoryview) -> Tuple[memoryview, Dict[Any, Any]]:
     return buffer, values
 
 
+@_depth_guarded
 def _decode_map_large(buffer: memoryview) -> Tuple[memoryview, Dict[Any, Any]]:
     # Validate the raw on-wire count *before* halving it (the AMQP encoding
     # stores total entries; pairs = entries / 2). Checking pre-halve keeps
@@ -298,6 +329,7 @@ def _decode_map_large(buffer: memoryview) -> Tuple[memoryview, Dict[Any, Any]]:
     return buffer, values
 
 
+@_depth_guarded
 def _decode_array_small(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     count = buffer[1]  # Ignore first byte (size) and just rely on count
     if count:
@@ -319,6 +351,7 @@ def _decode_array_small(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     return buffer[2:], []
 
 
+@_depth_guarded
 def _decode_array_large(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     count = c_unsigned_long.unpack(buffer[4:8])[0]
     # Validate the wire-supplied count before allocating `[None] * count`.
@@ -347,6 +380,7 @@ def _decode_array_large(buffer: memoryview) -> Tuple[memoryview, List[Any]]:
     return buffer[8:], []
 
 
+@_depth_guarded
 def _decode_described(buffer: memoryview) -> Tuple[memoryview, object]:
     # TODO: to move the cursor of the buffer to the described value based on size of the
     #  descriptor without decoding descriptor value

sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py

@@ -1,6 +1,11 @@
 import pathlib
 import pytest
-from azure.servicebus._pyamqp._decode import decode_frame, _PERFORMATIVE_FIELD_COUNT
+from azure.servicebus._pyamqp._decode import (
+    decode_frame,
+    decode_payload,
+    _PERFORMATIVE_FIELD_COUNT,
+    _MAX_NESTED_DEPTH,
+)
 from azure.servicebus._pyamqp import performatives
 
 
@@ -185,6 +190,36 @@ def test_performative_field_count_matches_spec(frame_cls, expected_count):
     assert _PERFORMATIVE_FIELD_COUNT[frame_cls._code] == expected_count
 
 
+def _nested_value(levels):
+    # An amqp-value section (descriptor 0x77) wrapping `levels` nested list8
+    # compounds, each holding exactly one element: 0xc0 (list8), size, count=1,
+    # with an innermost list0 (0x45) leaf. decode_payload dispatches the value
+    # constructor directly and the list0 leaf is not a guarded decoder, so the
+    # decode reaches exactly `levels` guarded compound frames. Every level passes
+    # the _MAX_COMPOUND_COUNT check (count == 1); only the nesting depth grows,
+    # ~3 wire bytes per level.
+    inner = bytes([0x45])  # innermost empty list (list0)
+    for _ in range(levels):
+        inner = bytes([0xC0, 0, 1]) + inner
+    return memoryview(bytes([0x00, 0x53, 0x77]) + inner)
+
+
+def test_decode_allows_nesting_at_the_depth_cap():
+    # Exactly at the cap must still decode: the guard rejects only depth > cap.
+    decode_payload(_nested_value(_MAX_NESTED_DEPTH))
+
+
+def test_decode_rejects_excessive_nesting_depth():
+    # One level past the cap, and a pathologically deep message, must each raise
+    # a clean ValueError rather than exhausting the interpreter stack with a
+    # RecursionError. The element-count cap does not catch this: every level has
+    # count == 1.
+    with pytest.raises(ValueError, match="nested compound depth"):
+        decode_payload(_nested_value(_MAX_NESTED_DEPTH + 1))
+    with pytest.raises(ValueError, match="nested compound depth"):
+        decode_payload(_nested_value(5000))
+
+
 # The _pyamqp engine is vendored identically into azure-eventhub and
 # azure-servicebus; a fix (like the padding above) must be applied to both
 # copies. Guard against the two copies silently drifting apart. The packages