From 3ce1e43866ecafc48ea19f52b11fc789c602302c Mon Sep 17 00:00:00 2001 From: Anna Tchijova Date: Sat, 29 Aug 2026 14:54:28 -0300 Subject: [PATCH 1/2] [EventHub][ServiceBus] Bound pyAMQP decoder recursion depth The pyAMQP decoder is recursive-descent: each nested compound type (list/map/array/described) re-enters the decoder for its elements. Each nesting level is only ~1-3 wire bytes but costs one Python recursion frame, so a small but deeply nested message exhausts the interpreter stack and raises an uncaught RecursionError inside decode_payload. The existing _MAX_COMPOUND_COUNT element-count cap does not catch this because every nesting level has count == 1. Add a thread-local nesting-depth guard (_MAX_NESTED_DEPTH = 64) applied via an @_depth_guarded decorator to the seven recursive compound decoders; nesting beyond the cap raises a clear ValueError instead of a RecursionError. The guard is thread-local so concurrent connections do not race and is balanced by try/finally (decode is synchronous). Applied byte-identically to both vendored _pyamqp copies (eventhub + servicebus) and covered by unit tests. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01VWmkpCnuhCPKFPD2fexY1a --- sdk/eventhub/azure-eventhub/CHANGELOG.md | 1 + .../azure/eventhub/_pyamqp/_decode.py | 34 +++++++++++++++++++ sdk/servicebus/azure-servicebus/CHANGELOG.md | 2 ++ .../azure/servicebus/_pyamqp/_decode.py | 34 +++++++++++++++++++ .../tests/unittests/test_pyamqp_decode.py | 33 +++++++++++++++++- 5 files changed, 103 insertions(+), 1 deletion(-) diff --git a/sdk/eventhub/azure-eventhub/CHANGELOG.md b/sdk/eventhub/azure-eventhub/CHANGELOG.md index 52f00add630f..fa16ce39c336 100644 --- a/sdk/eventhub/azure-eventhub/CHANGELOG.md +++ b/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) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py index 68e571a45c56..9e0c03afc24e 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py @@ -74,6 +74,33 @@ 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. --- +import threading as _threading +_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 diff --git a/sdk/servicebus/azure-servicebus/CHANGELOG.md b/sdk/servicebus/azure-servicebus/CHANGELOG.md index 92fc9dd99d37..d0d6a8746ca6 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/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) diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py index 68e571a45c56..9e0c03afc24e 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py @@ -74,6 +74,33 @@ 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. --- +import threading as _threading +_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 diff --git a/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py index b22f6c7238be..61d1e12d825f 100644 --- a/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py +++ b/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,32 @@ def test_performative_field_count_matches_spec(frame_cls, expected_count): assert _PERFORMATIVE_FIELD_COUNT[frame_cls._code] == expected_count +def _nested_value(depth): + # An amqp-value section (descriptor 0x77) wrapping `depth` nested list8 + # compounds, each holding exactly one element: 0xc0 (list8), size, count=1. + # 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 + for _ in range(depth): + inner = bytes([0xC0, 0, 1]) + inner + return memoryview(bytes([0x00, 0x53, 0x77]) + inner) + + +def test_decode_allows_nesting_up_to_the_depth_cap(): + # Nesting within the cap decodes without error. + decode_payload(_nested_value(_MAX_NESTED_DEPTH - 2)) + + +def test_decode_rejects_excessive_nesting_depth(): + # A deeply nested message must 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 + 5)) + 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 From 2ad26072bfe13d670327f9af0d37eae09fea2f61 Mon Sep 17 00:00:00 2001 From: Anna Tchijova Date: Sat, 29 Aug 2026 15:11:24 -0300 Subject: [PATCH 2/2] Address review feedback: import position and exact depth-cap boundary - Move `import threading` into the top standard-library import block; it was after module-level assignments, which pylint flags as C0413 (wrong-import-position) and C0411 (wrong-import-order). - Tighten the depth-guard tests to exercise the exact boundary: nesting at _MAX_NESTED_DEPTH must decode, and _MAX_NESTED_DEPTH + 1 must raise ValueError (the previous test stopped short of the cap). Both _pyamqp copies stay byte-identical. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01VWmkpCnuhCPKFPD2fexY1a --- .../azure/eventhub/_pyamqp/_decode.py | 4 +-- .../azure/servicebus/_pyamqp/_decode.py | 4 +-- .../tests/unittests/test_pyamqp_decode.py | 32 +++++++++++-------- 3 files changed, 22 insertions(+), 18 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py index 9e0c03afc24e..38ab98026f85 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py +++ b/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, @@ -79,9 +80,8 @@ # 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. --- -import threading as _threading _MAX_NESTED_DEPTH = 64 -_decode_depth = _threading.local() +_decode_depth = threading.local() def _depth_guarded(fn): diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py index 9e0c03afc24e..38ab98026f85 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py +++ b/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, @@ -79,9 +80,8 @@ # 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. --- -import threading as _threading _MAX_NESTED_DEPTH = 64 -_decode_depth = _threading.local() +_decode_depth = threading.local() def _depth_guarded(fn): diff --git a/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py index 61d1e12d825f..3d61e4d55bba 100644 --- a/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py @@ -190,28 +190,32 @@ def test_performative_field_count_matches_spec(frame_cls, expected_count): assert _PERFORMATIVE_FIELD_COUNT[frame_cls._code] == expected_count -def _nested_value(depth): - # An amqp-value section (descriptor 0x77) wrapping `depth` nested list8 - # compounds, each holding exactly one element: 0xc0 (list8), size, count=1. - # 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 - for _ in range(depth): +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_up_to_the_depth_cap(): - # Nesting within the cap decodes without error. - decode_payload(_nested_value(_MAX_NESTED_DEPTH - 2)) +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(): - # A deeply nested message must 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. + # 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 + 5)) + decode_payload(_nested_value(_MAX_NESTED_DEPTH + 1)) with pytest.raises(ValueError, match="nested compound depth"): decode_payload(_nested_value(5000))