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..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, @@ -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 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..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, @@ -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 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..3d61e4d55bba 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,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