From a7df9582e134ce92cac15814326d768104b3be18 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Wed, 24 Jun 2026 21:38:48 -0500 Subject: [PATCH 1/6] fix(pyamqp): pad decoded performatives to full field count AMQP 1.0 section 1.4 lets a sender omit trailing null fields, so an incoming performative described-list can be shorter than the full field count. The pyAMQP decoder built the field list from the wire count, and consumers then accessed fixed indices (frame[9], OpenFrame(*frame)), raising IndexError/TypeError on a short list. decode_frame now pads the decoded list up to the performative's full field count, so omitted trailing fields read back as None. Applied to both the Event Hubs and Service Bus vendored copies, with regression tests and changelog entries. --- sdk/eventhub/azure-eventhub/CHANGELOG.md | 1 + .../azure/eventhub/_pyamqp/_decode.py | 36 ++++++++++ .../pyamqp_tests/unittest/test_decode.py | 69 ++++++++++++++++++- sdk/servicebus/azure-servicebus/CHANGELOG.md | 1 + .../azure/servicebus/_pyamqp/_decode.py | 36 ++++++++++ .../tests/unittests/test_pyamqp_decode.py | 62 +++++++++++++++++ 6 files changed, 204 insertions(+), 1 deletion(-) create mode 100644 sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py diff --git a/sdk/eventhub/azure-eventhub/CHANGELOG.md b/sdk/eventhub/azure-eventhub/CHANGELOG.md index 7ebacae7524f..56f6223011dd 100644 --- a/sdk/eventhub/azure-eventhub/CHANGELOG.md +++ b/sdk/eventhub/azure-eventhub/CHANGELOG.md @@ -5,6 +5,7 @@ ### Bugs Fixed - 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 `None`. ## 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 32ef0ddd9c12..d64132f645d9 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py @@ -24,6 +24,7 @@ from . import described from .message import Message, Header, Properties +from . import performatives if TYPE_CHECKING: from .message import MessageDict @@ -416,6 +417,36 @@ def decode_payload(buffer: memoryview) -> Message: return Message(**message_properties) +# Number of fields encoded on the wire for each performative, keyed by its +# frame-type code. The AMQP 1.0 spec (section 1.4) lets a sender omit trailing +# null fields, so an incoming performative list can be shorter than the full +# field count. Padding the decoded list up to this count keeps positional +# (frame[N]) access and namedtuple unpacking safe; the missing trailing fields +# read back as None, which is the spec-defined meaning of an omitted field. +# The transfer performative (code 20) carries a trailing payload that is not a +# wire field, so its _definition uses a None sentinel for that slot, which is +# excluded from the count here. +_PERFORMATIVE_FIELD_COUNT: Dict[int, int] = { + performative._code: sum(1 for field in performative._definition if field is not None) + for performative in ( + performatives.OpenFrame, + performatives.BeginFrame, + performatives.AttachFrame, + performatives.FlowFrame, + performatives.TransferFrame, + performatives.DispositionFrame, + performatives.DetachFrame, + performatives.EndFrame, + performatives.CloseFrame, + performatives.SASLMechanism, + performatives.SASLInit, + performatives.SASLChallenge, + performatives.SASLResponse, + performatives.SASLOutcome, + ) +} + + def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: # Ignore the first two bytes, they will always be the constructors for # described type then ulong. @@ -439,6 +470,11 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: fields: List[Optional[memoryview]] = [None] * count for i in range(count): buffer, fields[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:]) + # A sender may omit trailing null fields, so pad the decoded list up to the + # performative's full field count before any positional access or unpacking. + full_field_count = _PERFORMATIVE_FIELD_COUNT.get(frame_type) + if full_field_count is not None and count < full_field_count: + fields.extend([None] * (full_field_count - count)) if frame_type == 20: fields.append(buffer) return frame_type, fields diff --git a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py index 4254715abffa..101e39698fdd 100644 --- a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py +++ b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py @@ -1,5 +1,13 @@ import pytest -from azure.eventhub._pyamqp._decode import _decode_decimal128, _decode_described, _decode_array_small, _decode_array_large +from azure.eventhub._pyamqp._decode import ( + _decode_decimal128, + _decode_described, + _decode_array_small, + _decode_array_large, + decode_frame, + _PERFORMATIVE_FIELD_COUNT, +) +from azure.eventhub._pyamqp import performatives from decimal import Decimal @@ -45,3 +53,62 @@ def test_array_of_described_large(): for i in range(256): assert output[i] == [b'n', b'v'] assert output[i].descriptor == 1335734831060 + + +def _list8_frame(code, count, encoded_fields, payload=b""): + # Build a described performative frame using a list8 (0xc0) body: + # described-type ctor (0x00), ulong ctor (0x53), descriptor code, list8 (0xc0), + # size, count, then the encoded field bytes and any trailing payload. + header = bytes([0x00, 0x53, code, 0xC0, len(encoded_fields) + 1, count]) + return memoryview(header + encoded_fields + payload) + + +# A sender may omit trailing null fields (AMQP 1.0 section 1.4), so an incoming +# performative list can be shorter than the full field count. The decoder must +# pad it back to the full count so positional access and namedtuple unpacking +# stay safe and omitted fields read back as None. +def test_short_open_is_padded_to_full_field_count(): + # Open with only container_id set ("x"), 1 field on the wire out of 10. + frame = _list8_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78])) + frame_type, fields = decode_frame(frame) + assert frame_type == performatives.OpenFrame._code + assert len(fields) == 10 + # Unpacking and fixed-index access must not raise on the omitted fields. + open_frame = performatives.OpenFrame(*fields) + assert open_frame.container_id == b"x" + assert open_frame.properties is None + assert fields[9] is None + + +def test_short_transfer_pads_fields_and_preserves_payload(): + # Transfer with only handle (0) set, plus a message payload. The payload is + # appended after the fields and must survive the padding. + frame = _list8_frame(performatives.TransferFrame._code, 1, bytes([0x52, 0x00]), payload=b"\xde\xad") + frame_type, fields = decode_frame(frame) + assert frame_type == performatives.TransferFrame._code + # 11 wire fields padded out, then the trailing payload appended (12 total). + assert len(fields) == 12 + transfer = performatives.TransferFrame(*fields) + assert transfer.handle == 0 + assert transfer.batchable is None + assert bytes(transfer.payload) == b"\xde\xad" + + +@pytest.mark.parametrize( + "frame_cls,expected_count", + [ + (performatives.OpenFrame, 10), + (performatives.BeginFrame, 8), + (performatives.AttachFrame, 14), + (performatives.FlowFrame, 11), + (performatives.TransferFrame, 11), + (performatives.DispositionFrame, 6), + (performatives.DetachFrame, 3), + (performatives.EndFrame, 1), + (performatives.CloseFrame, 1), + ], +) +def test_performative_field_count_matches_spec(frame_cls, expected_count): + # The padding target is the number of wire fields defined for each + # performative (the trailing transfer payload slot is excluded). + assert _PERFORMATIVE_FIELD_COUNT[frame_cls._code] == expected_count diff --git a/sdk/servicebus/azure-servicebus/CHANGELOG.md b/sdk/servicebus/azure-servicebus/CHANGELOG.md index 2f38caeea96f..8ab63608ac9a 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md @@ -16,6 +16,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`. ([#45394](https://github.com/Azure/azure-sdk-for-python/issues/45394)) - Fixed a bug where passing a `fully_qualified_namespace` that included a port and/or trailing path (for example the `https://.servicebus.windows.net:443/` form that Azure returns when provisioning a namespace) raised `ServiceBusAuthenticationError`. The namespace is now normalized to its bare host, matching the .NET and JavaScript SDKs. ([#44034](https://github.com/Azure/azure-sdk-for-python/issues/44034)) - Fixed a bug where iterating over a `ServiceBusReceiver` suppressed automatic HTTP instrumentation (e.g. from `opentelemetry-instrumentation-httpx`/`requests`) while user code processed a received message, causing the user's own outbound HTTP spans to be dropped. The receive tracing span is now closed before the message is yielded to the caller, so suppression no longer leaks into message processing. ([#42755](https://github.com/Azure/azure-sdk-for-python/issues/42755)) +- 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. ### Other Changes diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py index 32ef0ddd9c12..d64132f645d9 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py @@ -24,6 +24,7 @@ from . import described from .message import Message, Header, Properties +from . import performatives if TYPE_CHECKING: from .message import MessageDict @@ -416,6 +417,36 @@ def decode_payload(buffer: memoryview) -> Message: return Message(**message_properties) +# Number of fields encoded on the wire for each performative, keyed by its +# frame-type code. The AMQP 1.0 spec (section 1.4) lets a sender omit trailing +# null fields, so an incoming performative list can be shorter than the full +# field count. Padding the decoded list up to this count keeps positional +# (frame[N]) access and namedtuple unpacking safe; the missing trailing fields +# read back as None, which is the spec-defined meaning of an omitted field. +# The transfer performative (code 20) carries a trailing payload that is not a +# wire field, so its _definition uses a None sentinel for that slot, which is +# excluded from the count here. +_PERFORMATIVE_FIELD_COUNT: Dict[int, int] = { + performative._code: sum(1 for field in performative._definition if field is not None) + for performative in ( + performatives.OpenFrame, + performatives.BeginFrame, + performatives.AttachFrame, + performatives.FlowFrame, + performatives.TransferFrame, + performatives.DispositionFrame, + performatives.DetachFrame, + performatives.EndFrame, + performatives.CloseFrame, + performatives.SASLMechanism, + performatives.SASLInit, + performatives.SASLChallenge, + performatives.SASLResponse, + performatives.SASLOutcome, + ) +} + + def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: # Ignore the first two bytes, they will always be the constructors for # described type then ulong. @@ -439,6 +470,11 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: fields: List[Optional[memoryview]] = [None] * count for i in range(count): buffer, fields[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:]) + # A sender may omit trailing null fields, so pad the decoded list up to the + # performative's full field count before any positional access or unpacking. + full_field_count = _PERFORMATIVE_FIELD_COUNT.get(frame_type) + if full_field_count is not None and count < full_field_count: + fields.extend([None] * (full_field_count - count)) if frame_type == 20: fields.append(buffer) return frame_type, fields diff --git a/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py new file mode 100644 index 000000000000..326449a9a164 --- /dev/null +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py @@ -0,0 +1,62 @@ +import pytest +from azure.servicebus._pyamqp._decode import decode_frame, _PERFORMATIVE_FIELD_COUNT +from azure.servicebus._pyamqp import performatives + + +def _list8_frame(code, count, encoded_fields, payload=b""): + # Build a described performative frame using a list8 (0xc0) body: + # described-type ctor (0x00), ulong ctor (0x53), descriptor code, list8 (0xc0), + # size, count, then the encoded field bytes and any trailing payload. + header = bytes([0x00, 0x53, code, 0xC0, len(encoded_fields) + 1, count]) + return memoryview(header + encoded_fields + payload) + + +# A sender may omit trailing null fields (AMQP 1.0 section 1.4), so an incoming +# performative list can be shorter than the full field count. The decoder must +# pad it back to the full count so positional access and namedtuple unpacking +# stay safe and omitted fields read back as None. +def test_short_open_is_padded_to_full_field_count(): + # Open with only container_id set ("x"), 1 field on the wire out of 10. + frame = _list8_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78])) + frame_type, fields = decode_frame(frame) + assert frame_type == performatives.OpenFrame._code + assert len(fields) == 10 + # Unpacking and fixed-index access must not raise on the omitted fields. + open_frame = performatives.OpenFrame(*fields) + assert open_frame.container_id == b"x" + assert open_frame.properties is None + assert fields[9] is None + + +def test_short_transfer_pads_fields_and_preserves_payload(): + # Transfer with only handle (0) set, plus a message payload. The payload is + # appended after the fields and must survive the padding. + frame = _list8_frame(performatives.TransferFrame._code, 1, bytes([0x52, 0x00]), payload=b"\xde\xad") + frame_type, fields = decode_frame(frame) + assert frame_type == performatives.TransferFrame._code + # 11 wire fields padded out, then the trailing payload appended (12 total). + assert len(fields) == 12 + transfer = performatives.TransferFrame(*fields) + assert transfer.handle == 0 + assert transfer.batchable is None + assert bytes(transfer.payload) == b"\xde\xad" + + +@pytest.mark.parametrize( + "frame_cls,expected_count", + [ + (performatives.OpenFrame, 10), + (performatives.BeginFrame, 8), + (performatives.AttachFrame, 14), + (performatives.FlowFrame, 11), + (performatives.TransferFrame, 11), + (performatives.DispositionFrame, 6), + (performatives.DetachFrame, 3), + (performatives.EndFrame, 1), + (performatives.CloseFrame, 1), + ], +) +def test_performative_field_count_matches_spec(frame_cls, expected_count): + # The padding target is the number of wire fields defined for each + # performative (the trailing transfer payload slot is excluded). + assert _PERFORMATIVE_FIELD_COUNT[frame_cls._code] == expected_count From a2aea4e0102ec0b942ffb4444d89b3824309cc68 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Wed, 15 Jul 2026 17:55:36 -0500 Subject: [PATCH 2/6] fix(pyamqp): silence protected-access lint and handle list0 performatives Add the # type: ignore # pylint: disable=protected-access suppressions on the _PERFORMATIVE_FIELD_COUNT comprehension, matching every other access to the protected _code/_definition attributes in this engine (see _encode.py), so the pylint and mypy CI gates stay green. Also handle the AMQP list0 (0x45) body encoding in decode_frame: a performative whose fields are all omitted may arrive as list0, which carries no count byte. Previously this read data[5] out of range and raised IndexError before the padding ran. It is now treated as zero fields and padded to the full field count, closing the "omitted trailing null fields" case for End and Close. Add list0 regression tests to both the eventhub and servicebus suites. --- sdk/eventhub/azure-eventhub/CHANGELOG.md | 2 +- .../azure/eventhub/_pyamqp/_decode.py | 9 +++++++- .../pyamqp_tests/unittest/test_decode.py | 21 +++++++++++++++++++ sdk/servicebus/azure-servicebus/CHANGELOG.md | 2 +- .../azure/servicebus/_pyamqp/_decode.py | 9 +++++++- .../tests/unittests/test_pyamqp_decode.py | 21 +++++++++++++++++++ 6 files changed, 60 insertions(+), 4 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/CHANGELOG.md b/sdk/eventhub/azure-eventhub/CHANGELOG.md index 56f6223011dd..dad31c7dcaa2 100644 --- a/sdk/eventhub/azure-eventhub/CHANGELOG.md +++ b/sdk/eventhub/azure-eventhub/CHANGELOG.md @@ -5,7 +5,7 @@ ### Bugs Fixed - 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 `None`. +- 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 `None`, including the compact `list0` encoding where every field is omitted. ## 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 d64132f645d9..b1829bf6f63e 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py @@ -427,7 +427,8 @@ def decode_payload(buffer: memoryview) -> Message: # wire field, so its _definition uses a None sentinel for that slot, which is # excluded from the count here. _PERFORMATIVE_FIELD_COUNT: Dict[int, int] = { - performative._code: sum(1 for field in performative._definition if field is not None) + # pylint: disable=protected-access + performative._code: sum(1 for field in performative._definition if field is not None) # type: ignore for performative in ( performatives.OpenFrame, performatives.BeginFrame, @@ -463,6 +464,12 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: f"AMQP frame field count {count} exceeds maximum {_MAX_COMPOUND_COUNT}" ) buffer = data[12:] + elif compound_list_type == 0x45: + # list0 0x45: an empty list with no size or count bytes. A sender may + # encode a performative whose fields are all omitted this way, so treat + # it as zero fields and let the padding below fill in the nulls. + count = 0 + buffer = data[4:] else: # list8 0xc0: data[4] is size, data[5] is count (1 byte, bounded at 255). count = data[5] diff --git a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py index 101e39698fdd..ba52746d4c91 100644 --- a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py +++ b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py @@ -94,6 +94,27 @@ def test_short_transfer_pads_fields_and_preserves_payload(): assert bytes(transfer.payload) == b"\xde\xad" +def _list0_frame(code): + # Build a described performative whose body is an AMQP list0 (0x45): the most + # compact "all fields omitted" encoding, carrying no size or count bytes. + # described-type ctor (0x00), ulong ctor (0x53), descriptor code, list0 (0x45). + return memoryview(bytes([0x00, 0x53, code, 0x45])) + + +# A performative with every field omitted may arrive as a list0 (0x45) body, +# which has no count byte. The decoder must treat it as zero fields and pad up +# to the full field count instead of indexing past the end of the buffer. +@pytest.mark.parametrize("frame_cls", [performatives.EndFrame, performatives.CloseFrame]) +def test_list0_performative_pads_to_full_field_count(frame_cls): + frame = _list0_frame(frame_cls._code) + frame_type, fields = decode_frame(frame) + assert frame_type == frame_cls._code + assert len(fields) == _PERFORMATIVE_FIELD_COUNT[frame_cls._code] + # Every omitted field reads back as None and unpacking must not raise. + performative = frame_cls(*fields) + assert performative.error is None + + @pytest.mark.parametrize( "frame_cls,expected_count", [ diff --git a/sdk/servicebus/azure-servicebus/CHANGELOG.md b/sdk/servicebus/azure-servicebus/CHANGELOG.md index 8ab63608ac9a..251484fcc6f7 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md @@ -16,7 +16,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`. ([#45394](https://github.com/Azure/azure-sdk-for-python/issues/45394)) - Fixed a bug where passing a `fully_qualified_namespace` that included a port and/or trailing path (for example the `https://.servicebus.windows.net:443/` form that Azure returns when provisioning a namespace) raised `ServiceBusAuthenticationError`. The namespace is now normalized to its bare host, matching the .NET and JavaScript SDKs. ([#44034](https://github.com/Azure/azure-sdk-for-python/issues/44034)) - Fixed a bug where iterating over a `ServiceBusReceiver` suppressed automatic HTTP instrumentation (e.g. from `opentelemetry-instrumentation-httpx`/`requests`) while user code processed a received message, causing the user's own outbound HTTP spans to be dropped. The receive tracing span is now closed before the message is yielded to the caller, so suppression no longer leaks into message processing. ([#42755](https://github.com/Azure/azure-sdk-for-python/issues/42755)) -- 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. +- 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. ### Other Changes diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py index d64132f645d9..b1829bf6f63e 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py @@ -427,7 +427,8 @@ def decode_payload(buffer: memoryview) -> Message: # wire field, so its _definition uses a None sentinel for that slot, which is # excluded from the count here. _PERFORMATIVE_FIELD_COUNT: Dict[int, int] = { - performative._code: sum(1 for field in performative._definition if field is not None) + # pylint: disable=protected-access + performative._code: sum(1 for field in performative._definition if field is not None) # type: ignore for performative in ( performatives.OpenFrame, performatives.BeginFrame, @@ -463,6 +464,12 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: f"AMQP frame field count {count} exceeds maximum {_MAX_COMPOUND_COUNT}" ) buffer = data[12:] + elif compound_list_type == 0x45: + # list0 0x45: an empty list with no size or count bytes. A sender may + # encode a performative whose fields are all omitted this way, so treat + # it as zero fields and let the padding below fill in the nulls. + count = 0 + buffer = data[4:] else: # list8 0xc0: data[4] is size, data[5] is count (1 byte, bounded at 255). count = data[5] 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 326449a9a164..54aa678ba3db 100644 --- a/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py @@ -42,6 +42,27 @@ def test_short_transfer_pads_fields_and_preserves_payload(): assert bytes(transfer.payload) == b"\xde\xad" +def _list0_frame(code): + # Build a described performative whose body is an AMQP list0 (0x45): the most + # compact "all fields omitted" encoding, carrying no size or count bytes. + # described-type ctor (0x00), ulong ctor (0x53), descriptor code, list0 (0x45). + return memoryview(bytes([0x00, 0x53, code, 0x45])) + + +# A performative with every field omitted may arrive as a list0 (0x45) body, +# which has no count byte. The decoder must treat it as zero fields and pad up +# to the full field count instead of indexing past the end of the buffer. +@pytest.mark.parametrize("frame_cls", [performatives.EndFrame, performatives.CloseFrame]) +def test_list0_performative_pads_to_full_field_count(frame_cls): + frame = _list0_frame(frame_cls._code) + frame_type, fields = decode_frame(frame) + assert frame_type == frame_cls._code + assert len(fields) == _PERFORMATIVE_FIELD_COUNT[frame_cls._code] + # Every omitted field reads back as None and unpacking must not raise. + performative = frame_cls(*fields) + assert performative.error is None + + @pytest.mark.parametrize( "frame_cls,expected_count", [ From 6348af63f9a9d39802185518ed253e5e3c9981bb Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Wed, 15 Jul 2026 18:28:47 -0500 Subject: [PATCH 3/6] test(pyamqp): cover more short performatives, list32, SASL, and copy drift Extend the decode regression tests for the trailing-null-omission fix: - Parametrized short-list8 decode over Begin, Attach, Disposition, and Detach, the no-default namedtuples that raised TypeError on a short unpack pre-fix. Only Open was previously covered. - Short performative encoded as a list32 (0xd0) body, exercising the count/offset branch that no existing test reached (all fixtures used list8). - Short SASLInit and SASLOutcome frames, which also have required fields. - A skip-guarded test asserting the eventhub and servicebus _pyamqp copies of _decode.py, _encode.py, and performatives.py stay byte-identical, so a fix applied to one copy but not the other is caught. It skips when the sibling package source is not present (the packages ship separately). Applied identically to both the eventhub and servicebus decode test suites. --- .../pyamqp_tests/unittest/test_decode.py | 92 +++++++++++++++++++ .../tests/unittests/test_pyamqp_decode.py | 92 +++++++++++++++++++ 2 files changed, 184 insertions(+) diff --git a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py index ba52746d4c91..e5de9145141f 100644 --- a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py +++ b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py @@ -1,3 +1,4 @@ +import pathlib import pytest from azure.eventhub._pyamqp._decode import ( _decode_decimal128, @@ -101,6 +102,66 @@ def _list0_frame(code): return memoryview(bytes([0x00, 0x53, code, 0x45])) +def _list32_frame(code, count, encoded_fields, payload=b""): + # Build a described performative frame using a list32 (0xd0) body, whose size + # and count are 4-byte big-endian. decode_frame ignores the size field and + # reads the count from data[8:12], so only the count must be accurate. + size = (len(encoded_fields) + 4).to_bytes(4, "big") + header = bytes([0x00, 0x53, code, 0xD0]) + size + count.to_bytes(4, "big") + return memoryview(header + encoded_fields + payload) + + +# Begin/Attach/Disposition/Detach are namedtuples with no field defaults, so a +# short positional unpack raised TypeError before the decoder padded to the full +# field count. Only Open was covered above; exercise the rest of the no-default +# performatives directly. +@pytest.mark.parametrize( + "frame_cls", + [ + performatives.AttachFrame, + performatives.BeginFrame, + performatives.DispositionFrame, + performatives.DetachFrame, + ], +) +def test_short_no_default_performative_is_padded(frame_cls): + full_field_count = _PERFORMATIVE_FIELD_COUNT[frame_cls._code] + # A single null field on the wire, the rest omitted. + frame = _list8_frame(frame_cls._code, 1, bytes([0x40])) + frame_type, fields = decode_frame(frame) + assert frame_type == frame_cls._code + assert len(fields) == full_field_count + assert fields[-1] is None + # Namedtuple construction must not raise on the omitted (now padded) fields. + frame_cls(*fields) + + +# The list32 (0xd0) body path is only taken for large frames and is otherwise +# unexercised; confirm a short performative encoded as list32 pads identically to +# the list8 case. +def test_short_list32_is_padded_to_full_field_count(): + frame = _list32_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78])) + frame_type, fields = decode_frame(frame) + assert frame_type == performatives.OpenFrame._code + assert len(fields) == 10 + open_frame = performatives.OpenFrame(*fields) + assert open_frame.container_id == b"x" + assert fields[9] is None + + +# SASLInit/SASLOutcome have required (no-default) fields, so a short SASL frame +# crashed pre-fix. They are in the field-count map and must pad too. +@pytest.mark.parametrize("frame_cls", [performatives.SASLInit, performatives.SASLOutcome]) +def test_short_sasl_frame_is_padded(frame_cls): + full_field_count = _PERFORMATIVE_FIELD_COUNT[frame_cls._code] + frame = _list8_frame(frame_cls._code, 1, bytes([0x40])) + frame_type, fields = decode_frame(frame) + assert frame_type == frame_cls._code + assert len(fields) == full_field_count + assert fields[-1] is None + frame_cls(*fields) + + # A performative with every field omitted may arrive as a list0 (0x45) body, # which has no count byte. The decoder must treat it as zero fields and pad up # to the full field count instead of indexing past the end of the buffer. @@ -133,3 +194,34 @@ def test_performative_field_count_matches_spec(frame_cls, expected_count): # The padding target is the number of wire fields defined for each # performative (the trailing transfer payload slot is excluded). assert _PERFORMATIVE_FIELD_COUNT[frame_cls._code] == expected_count + + +# 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 +# ship separately, so skip when the sibling source is not present rather than +# fail on a package-isolated checkout. +_PYAMQP_COPIES = ( + "sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp", + "sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp", +) + + +def _repo_root(): + for parent in pathlib.Path(__file__).resolve().parents: + if (parent / "sdk").is_dir(): + return parent + return None + + +@pytest.mark.parametrize("filename", ["_decode.py", "_encode.py", "performatives.py"]) +def test_pyamqp_copies_are_byte_identical(filename): + root = _repo_root() + assert root is not None, "could not locate the repo root (no ancestor contains sdk/)" + eventhub_copy = root / _PYAMQP_COPIES[0] / filename + servicebus_copy = root / _PYAMQP_COPIES[1] / filename + if not (eventhub_copy.exists() and servicebus_copy.exists()): + pytest.skip("both _pyamqp copies are not present in this checkout") + assert ( + eventhub_copy.read_bytes() == servicebus_copy.read_bytes() + ), f"{filename} has drifted between the eventhub and servicebus _pyamqp copies; apply to both." 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 54aa678ba3db..ec7100316b2c 100644 --- a/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py @@ -1,3 +1,4 @@ +import pathlib import pytest from azure.servicebus._pyamqp._decode import decode_frame, _PERFORMATIVE_FIELD_COUNT from azure.servicebus._pyamqp import performatives @@ -49,6 +50,66 @@ def _list0_frame(code): return memoryview(bytes([0x00, 0x53, code, 0x45])) +def _list32_frame(code, count, encoded_fields, payload=b""): + # Build a described performative frame using a list32 (0xd0) body, whose size + # and count are 4-byte big-endian. decode_frame ignores the size field and + # reads the count from data[8:12], so only the count must be accurate. + size = (len(encoded_fields) + 4).to_bytes(4, "big") + header = bytes([0x00, 0x53, code, 0xD0]) + size + count.to_bytes(4, "big") + return memoryview(header + encoded_fields + payload) + + +# Begin/Attach/Disposition/Detach are namedtuples with no field defaults, so a +# short positional unpack raised TypeError before the decoder padded to the full +# field count. Only Open was covered above; exercise the rest of the no-default +# performatives directly. +@pytest.mark.parametrize( + "frame_cls", + [ + performatives.AttachFrame, + performatives.BeginFrame, + performatives.DispositionFrame, + performatives.DetachFrame, + ], +) +def test_short_no_default_performative_is_padded(frame_cls): + full_field_count = _PERFORMATIVE_FIELD_COUNT[frame_cls._code] + # A single null field on the wire, the rest omitted. + frame = _list8_frame(frame_cls._code, 1, bytes([0x40])) + frame_type, fields = decode_frame(frame) + assert frame_type == frame_cls._code + assert len(fields) == full_field_count + assert fields[-1] is None + # Namedtuple construction must not raise on the omitted (now padded) fields. + frame_cls(*fields) + + +# The list32 (0xd0) body path is only taken for large frames and is otherwise +# unexercised; confirm a short performative encoded as list32 pads identically to +# the list8 case. +def test_short_list32_is_padded_to_full_field_count(): + frame = _list32_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78])) + frame_type, fields = decode_frame(frame) + assert frame_type == performatives.OpenFrame._code + assert len(fields) == 10 + open_frame = performatives.OpenFrame(*fields) + assert open_frame.container_id == b"x" + assert fields[9] is None + + +# SASLInit/SASLOutcome have required (no-default) fields, so a short SASL frame +# crashed pre-fix. They are in the field-count map and must pad too. +@pytest.mark.parametrize("frame_cls", [performatives.SASLInit, performatives.SASLOutcome]) +def test_short_sasl_frame_is_padded(frame_cls): + full_field_count = _PERFORMATIVE_FIELD_COUNT[frame_cls._code] + frame = _list8_frame(frame_cls._code, 1, bytes([0x40])) + frame_type, fields = decode_frame(frame) + assert frame_type == frame_cls._code + assert len(fields) == full_field_count + assert fields[-1] is None + frame_cls(*fields) + + # A performative with every field omitted may arrive as a list0 (0x45) body, # which has no count byte. The decoder must treat it as zero fields and pad up # to the full field count instead of indexing past the end of the buffer. @@ -81,3 +142,34 @@ def test_performative_field_count_matches_spec(frame_cls, expected_count): # The padding target is the number of wire fields defined for each # performative (the trailing transfer payload slot is excluded). assert _PERFORMATIVE_FIELD_COUNT[frame_cls._code] == expected_count + + +# 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 +# ship separately, so skip when the sibling source is not present rather than +# fail on a package-isolated checkout. +_PYAMQP_COPIES = ( + "sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp", + "sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp", +) + + +def _repo_root(): + for parent in pathlib.Path(__file__).resolve().parents: + if (parent / "sdk").is_dir(): + return parent + return None + + +@pytest.mark.parametrize("filename", ["_decode.py", "_encode.py", "performatives.py"]) +def test_pyamqp_copies_are_byte_identical(filename): + root = _repo_root() + assert root is not None, "could not locate the repo root (no ancestor contains sdk/)" + eventhub_copy = root / _PYAMQP_COPIES[0] / filename + servicebus_copy = root / _PYAMQP_COPIES[1] / filename + if not (eventhub_copy.exists() and servicebus_copy.exists()): + pytest.skip("both _pyamqp copies are not present in this checkout") + assert ( + eventhub_copy.read_bytes() == servicebus_copy.read_bytes() + ), f"{filename} has drifted between the eventhub and servicebus _pyamqp copies; apply to both." From 3868ff9adb360494988cebf7ac6ec93daf7be3d6 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 16 Jul 2026 14:18:55 -0400 Subject: [PATCH 4/6] fix(pyamqp): pad omitted performative fields with their AMQP default Decoding a short performative padded every omitted trailing field with None. For a field whose AMQP-defined default is non-null this is wrong: a minimal Open frame legitimately omits max_frame_size (default 4294967295), and _connection._incoming_open reads it positionally and numerically (`frame[2] < 512`), which raises TypeError on None. Pad each omitted field with its field default from the performative _definition instead of None, via a _PERFORMATIVE_FIELD_DEFAULTS map; _PERFORMATIVE_FIELD_COUNT is now derived from it so the two stay in lockstep. Applied to both the Event Hubs and Service Bus copies of the vendored _pyamqp engine. Tests exercise the incoming-Open numeric path (max_frame_size/channel_max materialize to their defaults and the `< 512` comparison does not raise), and the no-default-performative and transfer tests now assert the padded tail equals each field's spec default (e.g. Disposition.batchable is False, Transfer.message_format is 0). --- sdk/eventhub/azure-eventhub/CHANGELOG.md | 2 +- .../azure/eventhub/_pyamqp/_decode.py | 39 ++++++++++++------- .../pyamqp_tests/unittest/test_decode.py | 38 ++++++++++++++---- .../azure/servicebus/_pyamqp/_decode.py | 39 ++++++++++++------- .../tests/unittests/test_pyamqp_decode.py | 38 ++++++++++++++---- 5 files changed, 111 insertions(+), 45 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/CHANGELOG.md b/sdk/eventhub/azure-eventhub/CHANGELOG.md index dad31c7dcaa2..ed9c93366315 100644 --- a/sdk/eventhub/azure-eventhub/CHANGELOG.md +++ b/sdk/eventhub/azure-eventhub/CHANGELOG.md @@ -5,7 +5,7 @@ ### Bugs Fixed - 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 `None`, including the compact `list0` encoding where every field is omitted. +- 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. ## 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 b1829bf6f63e..cc36dd249205 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py @@ -417,18 +417,21 @@ def decode_payload(buffer: memoryview) -> Message: return Message(**message_properties) -# Number of fields encoded on the wire for each performative, keyed by its -# frame-type code. The AMQP 1.0 spec (section 1.4) lets a sender omit trailing -# null fields, so an incoming performative list can be shorter than the full -# field count. Padding the decoded list up to this count keeps positional -# (frame[N]) access and namedtuple unpacking safe; the missing trailing fields -# read back as None, which is the spec-defined meaning of an omitted field. +# The AMQP-defined default of each wire field of a performative, keyed by its +# frame-type code and ordered as the fields appear on the wire. The AMQP 1.0 +# spec (section 1.4) lets a sender omit trailing fields whose value is the +# default, so an incoming performative list can be shorter than the full field +# count. Padding the decoded list back up to the full count with these defaults +# keeps positional (frame[N]) access and namedtuple unpacking safe, and makes an +# omitted field read back as its default rather than None. That distinction +# matters: an Open that omits max_frame_size means 4294967295, and the connection +# compares it numerically (frame[2] < 512), which would raise on None. # The transfer performative (code 20) carries a trailing payload that is not a # wire field, so its _definition uses a None sentinel for that slot, which is -# excluded from the count here. -_PERFORMATIVE_FIELD_COUNT: Dict[int, int] = { +# excluded here. +_PERFORMATIVE_FIELD_DEFAULTS: Dict[int, List[Any]] = { # pylint: disable=protected-access - performative._code: sum(1 for field in performative._definition if field is not None) # type: ignore + performative._code: [field.default for field in performative._definition if field is not None] # type: ignore for performative in ( performatives.OpenFrame, performatives.BeginFrame, @@ -447,6 +450,12 @@ def decode_payload(buffer: memoryview) -> Message: ) } +# The number of wire fields for each performative, derived from the defaults +# above so the two stay in lockstep. +_PERFORMATIVE_FIELD_COUNT: Dict[int, int] = { + code: len(defaults) for code, defaults in _PERFORMATIVE_FIELD_DEFAULTS.items() +} + def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: # Ignore the first two bytes, they will always be the constructors for @@ -477,11 +486,13 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: fields: List[Optional[memoryview]] = [None] * count for i in range(count): buffer, fields[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:]) - # A sender may omit trailing null fields, so pad the decoded list up to the - # performative's full field count before any positional access or unpacking. - full_field_count = _PERFORMATIVE_FIELD_COUNT.get(frame_type) - if full_field_count is not None and count < full_field_count: - fields.extend([None] * (full_field_count - count)) + # A sender may omit trailing fields whose value is the default (AMQP 1.0 + # section 1.4), so pad the decoded list back up to the performative's full + # field count with each omitted field's default before any positional access + # or unpacking. + field_defaults = _PERFORMATIVE_FIELD_DEFAULTS.get(frame_type) + if field_defaults is not None and count < len(field_defaults): + fields.extend(field_defaults[count:]) if frame_type == 20: fields.append(buffer) return frame_type, fields diff --git a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py index e5de9145141f..6d066c8082f0 100644 --- a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py +++ b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py @@ -64,10 +64,10 @@ def _list8_frame(code, count, encoded_fields, payload=b""): return memoryview(header + encoded_fields + payload) -# A sender may omit trailing null fields (AMQP 1.0 section 1.4), so an incoming -# performative list can be shorter than the full field count. The decoder must -# pad it back to the full count so positional access and namedtuple unpacking -# stay safe and omitted fields read back as None. +# A sender may omit trailing fields whose value is the default (AMQP 1.0 section +# 1.4), so an incoming performative list can be shorter than the full field +# count. The decoder must pad it back to the full count so positional access and +# namedtuple unpacking stay safe and omitted fields read back as their default. def test_short_open_is_padded_to_full_field_count(): # Open with only container_id set ("x"), 1 field on the wire out of 10. frame = _list8_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78])) @@ -81,6 +81,21 @@ def test_short_open_is_padded_to_full_field_count(): assert fields[9] is None +def test_short_open_materializes_non_null_field_defaults(): + # An Open that omits max_frame_size/channel_max means their AMQP defaults, + # not null. _connection._incoming_open reads them positionally and numerically + # (frame[2] < 512, frame[3]); padding with None would raise TypeError there. + frame = _list8_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78])) + _, fields = decode_frame(frame) + assert fields[2] == 4294967295 # max_frame_size default + assert fields[3] == 65535 # channel_max default + # Exercise the exact comparison _incoming_open performs; must not raise. + assert not fields[2] < 512 + open_frame = performatives.OpenFrame(*fields) + assert open_frame.max_frame_size == 4294967295 + assert open_frame.channel_max == 65535 + + def test_short_transfer_pads_fields_and_preserves_payload(): # Transfer with only handle (0) set, plus a message payload. The payload is # appended after the fields and must survive the padding. @@ -91,7 +106,9 @@ def test_short_transfer_pads_fields_and_preserves_payload(): assert len(fields) == 12 transfer = performatives.TransferFrame(*fields) assert transfer.handle == 0 - assert transfer.batchable is None + # Omitted boolean/uint fields read back as their AMQP defaults, not None. + assert transfer.message_format == 0 + assert transfer.batchable is False assert bytes(transfer.payload) == b"\xde\xad" @@ -125,13 +142,18 @@ def _list32_frame(code, count, encoded_fields, payload=b""): ], ) def test_short_no_default_performative_is_padded(frame_cls): - full_field_count = _PERFORMATIVE_FIELD_COUNT[frame_cls._code] + # Each field's AMQP default, in wire order (the transfer payload sentinel + # is excluded, but these performatives have none). + defaults = [f.default for f in frame_cls._definition if f is not None] # pylint: disable=protected-access # A single null field on the wire, the rest omitted. frame = _list8_frame(frame_cls._code, 1, bytes([0x40])) frame_type, fields = decode_frame(frame) assert frame_type == frame_cls._code - assert len(fields) == full_field_count - assert fields[-1] is None + assert len(fields) == len(defaults) + # The one wire field decoded as an explicit null; the omitted trailing fields + # are padded with their defaults (e.g. Disposition.batchable is False). + assert fields[0] is None + assert fields[1:] == defaults[1:] # Namedtuple construction must not raise on the omitted (now padded) fields. frame_cls(*fields) diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py index b1829bf6f63e..cc36dd249205 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py @@ -417,18 +417,21 @@ def decode_payload(buffer: memoryview) -> Message: return Message(**message_properties) -# Number of fields encoded on the wire for each performative, keyed by its -# frame-type code. The AMQP 1.0 spec (section 1.4) lets a sender omit trailing -# null fields, so an incoming performative list can be shorter than the full -# field count. Padding the decoded list up to this count keeps positional -# (frame[N]) access and namedtuple unpacking safe; the missing trailing fields -# read back as None, which is the spec-defined meaning of an omitted field. +# The AMQP-defined default of each wire field of a performative, keyed by its +# frame-type code and ordered as the fields appear on the wire. The AMQP 1.0 +# spec (section 1.4) lets a sender omit trailing fields whose value is the +# default, so an incoming performative list can be shorter than the full field +# count. Padding the decoded list back up to the full count with these defaults +# keeps positional (frame[N]) access and namedtuple unpacking safe, and makes an +# omitted field read back as its default rather than None. That distinction +# matters: an Open that omits max_frame_size means 4294967295, and the connection +# compares it numerically (frame[2] < 512), which would raise on None. # The transfer performative (code 20) carries a trailing payload that is not a # wire field, so its _definition uses a None sentinel for that slot, which is -# excluded from the count here. -_PERFORMATIVE_FIELD_COUNT: Dict[int, int] = { +# excluded here. +_PERFORMATIVE_FIELD_DEFAULTS: Dict[int, List[Any]] = { # pylint: disable=protected-access - performative._code: sum(1 for field in performative._definition if field is not None) # type: ignore + performative._code: [field.default for field in performative._definition if field is not None] # type: ignore for performative in ( performatives.OpenFrame, performatives.BeginFrame, @@ -447,6 +450,12 @@ def decode_payload(buffer: memoryview) -> Message: ) } +# The number of wire fields for each performative, derived from the defaults +# above so the two stay in lockstep. +_PERFORMATIVE_FIELD_COUNT: Dict[int, int] = { + code: len(defaults) for code, defaults in _PERFORMATIVE_FIELD_DEFAULTS.items() +} + def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: # Ignore the first two bytes, they will always be the constructors for @@ -477,11 +486,13 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: fields: List[Optional[memoryview]] = [None] * count for i in range(count): buffer, fields[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:]) - # A sender may omit trailing null fields, so pad the decoded list up to the - # performative's full field count before any positional access or unpacking. - full_field_count = _PERFORMATIVE_FIELD_COUNT.get(frame_type) - if full_field_count is not None and count < full_field_count: - fields.extend([None] * (full_field_count - count)) + # A sender may omit trailing fields whose value is the default (AMQP 1.0 + # section 1.4), so pad the decoded list back up to the performative's full + # field count with each omitted field's default before any positional access + # or unpacking. + field_defaults = _PERFORMATIVE_FIELD_DEFAULTS.get(frame_type) + if field_defaults is not None and count < len(field_defaults): + fields.extend(field_defaults[count:]) if frame_type == 20: fields.append(buffer) return frame_type, fields 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 ec7100316b2c..7796fd6dcbf9 100644 --- a/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py @@ -12,10 +12,10 @@ def _list8_frame(code, count, encoded_fields, payload=b""): return memoryview(header + encoded_fields + payload) -# A sender may omit trailing null fields (AMQP 1.0 section 1.4), so an incoming -# performative list can be shorter than the full field count. The decoder must -# pad it back to the full count so positional access and namedtuple unpacking -# stay safe and omitted fields read back as None. +# A sender may omit trailing fields whose value is the default (AMQP 1.0 section +# 1.4), so an incoming performative list can be shorter than the full field +# count. The decoder must pad it back to the full count so positional access and +# namedtuple unpacking stay safe and omitted fields read back as their default. def test_short_open_is_padded_to_full_field_count(): # Open with only container_id set ("x"), 1 field on the wire out of 10. frame = _list8_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78])) @@ -29,6 +29,21 @@ def test_short_open_is_padded_to_full_field_count(): assert fields[9] is None +def test_short_open_materializes_non_null_field_defaults(): + # An Open that omits max_frame_size/channel_max means their AMQP defaults, + # not null. _connection._incoming_open reads them positionally and numerically + # (frame[2] < 512, frame[3]); padding with None would raise TypeError there. + frame = _list8_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78])) + _, fields = decode_frame(frame) + assert fields[2] == 4294967295 # max_frame_size default + assert fields[3] == 65535 # channel_max default + # Exercise the exact comparison _incoming_open performs; must not raise. + assert not fields[2] < 512 + open_frame = performatives.OpenFrame(*fields) + assert open_frame.max_frame_size == 4294967295 + assert open_frame.channel_max == 65535 + + def test_short_transfer_pads_fields_and_preserves_payload(): # Transfer with only handle (0) set, plus a message payload. The payload is # appended after the fields and must survive the padding. @@ -39,7 +54,9 @@ def test_short_transfer_pads_fields_and_preserves_payload(): assert len(fields) == 12 transfer = performatives.TransferFrame(*fields) assert transfer.handle == 0 - assert transfer.batchable is None + # Omitted boolean/uint fields read back as their AMQP defaults, not None. + assert transfer.message_format == 0 + assert transfer.batchable is False assert bytes(transfer.payload) == b"\xde\xad" @@ -73,13 +90,18 @@ def _list32_frame(code, count, encoded_fields, payload=b""): ], ) def test_short_no_default_performative_is_padded(frame_cls): - full_field_count = _PERFORMATIVE_FIELD_COUNT[frame_cls._code] + # Each field's AMQP default, in wire order (the transfer payload sentinel + # is excluded, but these performatives have none). + defaults = [f.default for f in frame_cls._definition if f is not None] # pylint: disable=protected-access # A single null field on the wire, the rest omitted. frame = _list8_frame(frame_cls._code, 1, bytes([0x40])) frame_type, fields = decode_frame(frame) assert frame_type == frame_cls._code - assert len(fields) == full_field_count - assert fields[-1] is None + assert len(fields) == len(defaults) + # The one wire field decoded as an explicit null; the omitted trailing fields + # are padded with their defaults (e.g. Disposition.batchable is False). + assert fields[0] is None + assert fields[1:] == defaults[1:] # Namedtuple construction must not raise on the omitted (now padded) fields. frame_cls(*fields) From 8b193ee76f69b6cae607badaa4ced93f82b58dff Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 16 Jul 2026 17:02:10 -0400 Subject: [PATCH 5/6] ci(pyamqp): silence pylint no-member on performative _code/_definition The _PERFORMATIVE_FIELD_DEFAULTS comprehension reads _code and _definition off the performative classes directly. Those attributes are assigned at import time in performatives.py, so pylint 4.0.4 with azure-pylint-guidelines-checker cannot resolve them statically and raised 28 E1101(no-member) errors, failing the Analyze job with exit 2. Extend the existing protected-access disable on that line to also cover no-member. Comment-only, behavior unchanged; applied to both byte-identical _pyamqp copies. --- sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py | 4 +++- .../azure-servicebus/azure/servicebus/_pyamqp/_decode.py | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py index cc36dd249205..9866afd0cac3 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py @@ -430,7 +430,9 @@ def decode_payload(buffer: memoryview) -> Message: # wire field, so its _definition uses a None sentinel for that slot, which is # excluded here. _PERFORMATIVE_FIELD_DEFAULTS: Dict[int, List[Any]] = { - # pylint: disable=protected-access + # _code and _definition are assigned onto the performative classes at import + # time (see performatives.py), so pylint cannot see them statically. + # pylint: disable=protected-access,no-member performative._code: [field.default for field in performative._definition if field is not None] # type: ignore for performative in ( performatives.OpenFrame, diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py index cc36dd249205..9866afd0cac3 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py @@ -430,7 +430,9 @@ def decode_payload(buffer: memoryview) -> Message: # wire field, so its _definition uses a None sentinel for that slot, which is # excluded here. _PERFORMATIVE_FIELD_DEFAULTS: Dict[int, List[Any]] = { - # pylint: disable=protected-access + # _code and _definition are assigned onto the performative classes at import + # time (see performatives.py), so pylint cannot see them statically. + # pylint: disable=protected-access,no-member performative._code: [field.default for field in performative._definition if field is not None] # type: ignore for performative in ( performatives.OpenFrame, From 3accb70a26b38908ee455f78428705e0c20eb397 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 16 Jul 2026 18:16:20 -0400 Subject: [PATCH 6/6] fix(pyamqp): normalize explicit-null performative fields to their default Only trailing fields of a performative may be omitted, so a sender that sets a later field while wanting an earlier one's default must encode that earlier field as an explicit null. A decoded null for a field whose AMQP default is non-null therefore also means that default. Decoding now normalizes those nulls so an explicit null reads back identically to an omitted field: an Open that nulls max_frame_size comes back as 4294967295 rather than None, and _connection._incoming_open's frame[2] < 512 no longer raises TypeError. Fields whose declared default is null are left as None. Applied to both byte-identical _pyamqp copies with a regression test that nulls max_frame_size while setting channel_max after it. --- sdk/eventhub/azure-eventhub/CHANGELOG.md | 2 +- .../azure/eventhub/_pyamqp/_decode.py | 16 ++++++++++++++-- .../pyamqp_tests/unittest/test_decode.py | 19 +++++++++++++++++++ sdk/servicebus/azure-servicebus/CHANGELOG.md | 2 +- .../azure/servicebus/_pyamqp/_decode.py | 16 ++++++++++++++-- .../tests/unittests/test_pyamqp_decode.py | 19 +++++++++++++++++++ 6 files changed, 68 insertions(+), 6 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/CHANGELOG.md b/sdk/eventhub/azure-eventhub/CHANGELOG.md index ed9c93366315..52f00add630f 100644 --- a/sdk/eventhub/azure-eventhub/CHANGELOG.md +++ b/sdk/eventhub/azure-eventhub/CHANGELOG.md @@ -5,7 +5,7 @@ ### Bugs Fixed - 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. +- 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. ## 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 9866afd0cac3..68e571a45c56 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py @@ -493,8 +493,20 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: # field count with each omitted field's default before any positional access # or unpacking. field_defaults = _PERFORMATIVE_FIELD_DEFAULTS.get(frame_type) - if field_defaults is not None and count < len(field_defaults): - fields.extend(field_defaults[count:]) + if field_defaults is not None: + if count < len(field_defaults): + fields.extend(field_defaults[count:]) + # Only trailing fields may be omitted, so a sender that sets a later + # field while wanting an earlier one's default must encode that earlier + # field as an explicit null. A decoded null for a field whose AMQP + # default is non-null therefore also means that default; normalize it so + # it reads back identically to the omitted case. For example, an Open + # that nulls max_frame_size must still compare as 4294967295, not None, + # when _incoming_open evaluates frame[2] < 512. Fields whose declared + # default is null are left as None. + for index, default in enumerate(field_defaults): + if default is not None and fields[index] is None: + fields[index] = default if frame_type == 20: fields.append(buffer) return frame_type, fields diff --git a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py index 6d066c8082f0..884021e5f16c 100644 --- a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py +++ b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py @@ -96,6 +96,25 @@ def test_short_open_materializes_non_null_field_defaults(): assert open_frame.channel_max == 65535 +def test_open_with_explicit_null_field_uses_default(): + # Only trailing fields may be omitted, so an Open that sets a later field + # (channel_max) while wanting the default max_frame_size must encode + # max_frame_size as an explicit null. That null must still read back as the + # 4294967295 default, or _incoming_open's frame[2] < 512 raises TypeError. + # container_id="x", hostname=null, max_frame_size=null, channel_max=100. + frame = _list8_frame( + performatives.OpenFrame._code, 4, bytes([0xA1, 0x01, 0x78, 0x40, 0x40, 0x52, 0x64]) + ) + _, fields = decode_frame(frame) + assert fields[2] == 4294967295 # explicit null normalized to the default + assert not fields[2] < 512 # the comparison _incoming_open performs + assert fields[3] == 100 # the explicitly set later field is preserved + assert fields[1] is None # a null-default field (hostname) stays None + open_frame = performatives.OpenFrame(*fields) + assert open_frame.max_frame_size == 4294967295 + assert open_frame.channel_max == 100 + + def test_short_transfer_pads_fields_and_preserves_payload(): # Transfer with only handle (0) set, plus a message payload. The payload is # appended after the fields and must survive the padding. diff --git a/sdk/servicebus/azure-servicebus/CHANGELOG.md b/sdk/servicebus/azure-servicebus/CHANGELOG.md index 251484fcc6f7..be9d3ecef980 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md @@ -16,7 +16,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`. ([#45394](https://github.com/Azure/azure-sdk-for-python/issues/45394)) - Fixed a bug where passing a `fully_qualified_namespace` that included a port and/or trailing path (for example the `https://.servicebus.windows.net:443/` form that Azure returns when provisioning a namespace) raised `ServiceBusAuthenticationError`. The namespace is now normalized to its bare host, matching the .NET and JavaScript SDKs. ([#44034](https://github.com/Azure/azure-sdk-for-python/issues/44034)) - Fixed a bug where iterating over a `ServiceBusReceiver` suppressed automatic HTTP instrumentation (e.g. from `opentelemetry-instrumentation-httpx`/`requests`) while user code processed a received message, causing the user's own outbound HTTP spans to be dropped. The receive tracing span is now closed before the message is yielded to the caller, so suppression no longer leaks into message processing. ([#42755](https://github.com/Azure/azure-sdk-for-python/issues/42755)) -- 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. +- 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. ### Other Changes diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py index 9866afd0cac3..68e571a45c56 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py @@ -493,8 +493,20 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]: # field count with each omitted field's default before any positional access # or unpacking. field_defaults = _PERFORMATIVE_FIELD_DEFAULTS.get(frame_type) - if field_defaults is not None and count < len(field_defaults): - fields.extend(field_defaults[count:]) + if field_defaults is not None: + if count < len(field_defaults): + fields.extend(field_defaults[count:]) + # Only trailing fields may be omitted, so a sender that sets a later + # field while wanting an earlier one's default must encode that earlier + # field as an explicit null. A decoded null for a field whose AMQP + # default is non-null therefore also means that default; normalize it so + # it reads back identically to the omitted case. For example, an Open + # that nulls max_frame_size must still compare as 4294967295, not None, + # when _incoming_open evaluates frame[2] < 512. Fields whose declared + # default is null are left as None. + for index, default in enumerate(field_defaults): + if default is not None and fields[index] is None: + fields[index] = default if frame_type == 20: fields.append(buffer) return frame_type, fields 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 7796fd6dcbf9..b22f6c7238be 100644 --- a/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py @@ -44,6 +44,25 @@ def test_short_open_materializes_non_null_field_defaults(): assert open_frame.channel_max == 65535 +def test_open_with_explicit_null_field_uses_default(): + # Only trailing fields may be omitted, so an Open that sets a later field + # (channel_max) while wanting the default max_frame_size must encode + # max_frame_size as an explicit null. That null must still read back as the + # 4294967295 default, or _incoming_open's frame[2] < 512 raises TypeError. + # container_id="x", hostname=null, max_frame_size=null, channel_max=100. + frame = _list8_frame( + performatives.OpenFrame._code, 4, bytes([0xA1, 0x01, 0x78, 0x40, 0x40, 0x52, 0x64]) + ) + _, fields = decode_frame(frame) + assert fields[2] == 4294967295 # explicit null normalized to the default + assert not fields[2] < 512 # the comparison _incoming_open performs + assert fields[3] == 100 # the explicitly set later field is preserved + assert fields[1] is None # a null-default field (hostname) stays None + open_frame = performatives.OpenFrame(*fields) + assert open_frame.max_frame_size == 4294967295 + assert open_frame.channel_max == 100 + + def test_short_transfer_pads_fields_and_preserves_payload(): # Transfer with only handle (0) set, plus a message payload. The payload is # appended after the fields and must survive the padding.