From f4a39df603cf8c368581420bbfcb0a85bddc9554 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Wed, 24 Jun 2026 21:38:48 -0500 Subject: [PATCH 1/7] 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 | 6 ++ .../azure/eventhub/_pyamqp/_decode.py | 36 ++++++++++ .../azure-eventhub/azure/eventhub/_version.py | 2 +- .../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 +++++++++++++++++ 7 files changed, 210 insertions(+), 2 deletions(-) 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 f287cb0b8fde..c823a834bdc5 100644 --- a/sdk/eventhub/azure-eventhub/CHANGELOG.md +++ b/sdk/eventhub/azure-eventhub/CHANGELOG.md @@ -1,5 +1,11 @@ # Release History +## 5.15.2 (Unreleased) + +### Bugs Fixed + +- 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) ### Bugs Fixed 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/azure/eventhub/_version.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_version.py index 5bb5c30266da..5d660bbf8ea3 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_version.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_version.py @@ -3,4 +3,4 @@ # Licensed under the MIT License. # ------------------------------------ -VERSION = "5.15.1" +VERSION = "5.15.2" 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 0d105a35dc45..7fc66848ed5a 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md @@ -13,6 +13,7 @@ - Fixed a bug where sending a batched or multi-message payload with `uamqp_transport=True` raised `TypeError: 'BatchMessage' object is not subscriptable` (and a masked `AttributeError` on the list path) when the first message carried a `message_id`, `session_id`, or `partition_key`. The batch envelope properties are now set through the transport-appropriate code path. (regression from [#42598](https://github.com/Azure/azure-sdk-for-python/pull/42598)) - Fixed a bug where the async receiver factory methods on `azure.servicebus.aio.ServiceBusClient` (`get_queue_receiver`, `get_subscription_receiver`) and the async `ServiceBusReceiver` annotated the `auto_lock_renewer` keyword with the synchronous `AutoLockRenewer`, causing static type checkers to reject the documented `azure.servicebus.aio.AutoLockRenewer` usage. The annotation now references the async `AutoLockRenewer`, matching the docstrings and runtime behavior. ([#47948](https://github.com/Azure/azure-sdk-for-python/issues/47948)) - Fixed a bug where closing a `PEEK_LOCK` receiver did not release messages that had been prefetched into the client buffer or were still in flight, so they remained locked at the broker until lock expiry — delaying their redelivery and inflating their delivery count. On close, a non-session `PEEK_LOCK` receiver now drains the link (stopping the broker and flushing in-flight transfers) and releases the buffered messages (`released` disposition), so the broker can redeliver them immediately without incrementing the delivery count. ([#42917](https://github.com/Azure/azure-sdk-for-python/issues/42917)) +- 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 247a45fac00288349ae7b0764e34b60ab3cbe863 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Wed, 15 Jul 2026 17:55:36 -0500 Subject: [PATCH 2/7] 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 c823a834bdc5..b78839d9481b 100644 --- a/sdk/eventhub/azure-eventhub/CHANGELOG.md +++ b/sdk/eventhub/azure-eventhub/CHANGELOG.md @@ -4,7 +4,7 @@ ### Bugs Fixed -- 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 7fc66848ed5a..88f5c178a6db 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md @@ -13,7 +13,7 @@ - Fixed a bug where sending a batched or multi-message payload with `uamqp_transport=True` raised `TypeError: 'BatchMessage' object is not subscriptable` (and a masked `AttributeError` on the list path) when the first message carried a `message_id`, `session_id`, or `partition_key`. The batch envelope properties are now set through the transport-appropriate code path. (regression from [#42598](https://github.com/Azure/azure-sdk-for-python/pull/42598)) - Fixed a bug where the async receiver factory methods on `azure.servicebus.aio.ServiceBusClient` (`get_queue_receiver`, `get_subscription_receiver`) and the async `ServiceBusReceiver` annotated the `auto_lock_renewer` keyword with the synchronous `AutoLockRenewer`, causing static type checkers to reject the documented `azure.servicebus.aio.AutoLockRenewer` usage. The annotation now references the async `AutoLockRenewer`, matching the docstrings and runtime behavior. ([#47948](https://github.com/Azure/azure-sdk-for-python/issues/47948)) - Fixed a bug where closing a `PEEK_LOCK` receiver did not release messages that had been prefetched into the client buffer or were still in flight, so they remained locked at the broker until lock expiry — delaying their redelivery and inflating their delivery count. On close, a non-session `PEEK_LOCK` receiver now drains the link (stopping the broker and flushing in-flight transfers) and releases the buffered messages (`released` disposition), so the broker can redeliver them immediately without incrementing the delivery count. ([#42917](https://github.com/Azure/azure-sdk-for-python/issues/42917)) -- 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 8ff392f6b726676ad315c56f81a3dd79ee14b828 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Wed, 15 Jul 2026 18:28:47 -0500 Subject: [PATCH 3/7] 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 fb636b93e696af5d8d9332cbea38c79283f8218d Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 16 Jul 2026 14:18:55 -0400 Subject: [PATCH 4/7] 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 b78839d9481b..cd57402203fa 100644 --- a/sdk/eventhub/azure-eventhub/CHANGELOG.md +++ b/sdk/eventhub/azure-eventhub/CHANGELOG.md @@ -4,7 +4,7 @@ ### Bugs Fixed -- 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 50c8857a971471ce8a68aa2f78fa507f726ca372 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 16 Jul 2026 17:02:10 -0400 Subject: [PATCH 5/7] 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 f446c2358ba86667dd1bbdeb541ba61669bc4790 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 16 Jul 2026 17:56:20 -0400 Subject: [PATCH 6/7] chore(eventhub): add generated api.md for apiview consistency check The api-consistency workflow requires a committed api.md and api.metadata.yml for every package a PR touches. azure-eventhub had neither, so the check reported "Missing required API files" once this branch touched the package. Generate both with azpysdk apistub (apiview-stub-generator 0.3.28, the version eng/apiview_reqs.txt pins), which the workflow diff-gates against a fresh regeneration. The change touches only the private _pyamqp engine, so the public API surface is unchanged; this is the package's existing public API captured for the first time. Verified the local toolchain reproduces the committed azure-servicebus api.md byte-for-byte (matching apiMdSha256), so the generated eventhub file matches what CI regenerates. --- sdk/eventhub/azure-eventhub/api.md | 1032 ++++++++++++++++++ sdk/eventhub/azure-eventhub/api.metadata.yml | 3 + 2 files changed, 1035 insertions(+) create mode 100644 sdk/eventhub/azure-eventhub/api.md create mode 100644 sdk/eventhub/azure-eventhub/api.metadata.yml diff --git a/sdk/eventhub/azure-eventhub/api.md b/sdk/eventhub/azure-eventhub/api.md new file mode 100644 index 000000000000..2465799687a8 --- /dev/null +++ b/sdk/eventhub/azure-eventhub/api.md @@ -0,0 +1,1032 @@ +```py +namespace azure.eventhub + + def azure.eventhub.parse_connection_string(conn_str: str) -> EventHubConnectionStringProperties: ... + + + class azure.eventhub.CheckpointStore: + + @abstractmethod + def claim_ownership( + self, + ownership_list: Iterable[Dict[str, Any]], + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + def list_checkpoints( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + def list_ownership( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + def update_checkpoint( + self, + checkpoint: Dict[str, Optional[Union[str, int]]], + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.CloseReason(Enum): + OWNERSHIP_LOST = 1 + SHUTDOWN = 0 + + + class azure.eventhub.EventData: + property body: PrimitiveTypes # Read-only + property body_type: AmqpMessageBodyType # Read-only + property content_type: Optional[str] + property correlation_id: Optional[str] + property enqueued_time: Optional[datetime] # Read-only + property message: Union[Message, LegacyMessage] + property message_id: Optional[str] + property offset: Optional[str] # Read-only + property partition_key: Optional[bytes] # Read-only + property properties: Dict[Union[str, bytes], Any] + property raw_amqp_message: AmqpAnnotatedMessage # Read-only + property sequence_number: Optional[int] # Read-only + property system_properties: Dict[bytes, Any] # Read-only + + def __init__(self, body: Optional[Union[str, bytes, List[AnyStr]]] = None) -> None: ... + + def __message_content__(self) -> MessageContent: ... + + def __repr__(self) -> str: ... + + def __str__(self) -> str: ... + + @classmethod + def from_bytes(cls, message: bytes) -> EventData: ... + + @classmethod + def from_message_content( + cls, + content: bytes, + content_type: str, + **kwargs: Any + ) -> EventData: ... + + def body_as_json(self, encoding: str = "UTF-8") -> Dict[str, Any]: ... + + def body_as_str(self, encoding: str = "UTF-8") -> str: ... + + + class azure.eventhub.EventDataBatch: + property message: Union[BatchMessage, LegacyBatchMessage] + property size_in_bytes: int # Read-only + + def __init__( + self, + max_size_in_bytes: Optional[int] = None, + partition_id: Optional[str] = None, + partition_key: Optional[Union[str, bytes]] = None, + **kwargs: Any + ) -> None: ... + + def __len__(self) -> int: ... + + def __repr__(self) -> str: ... + + def add(self, event_data: Union[EventData, AmqpAnnotatedMessage]) -> None: ... + + + class azure.eventhub.EventHubConnectionStringProperties(DictMixin): + property endpoint: str # Read-only + property eventhub_name: Optional[str] # Read-only + property fully_qualified_namespace: str # Read-only + property shared_access_key: Optional[str] # Read-only + property shared_access_key_name: Optional[str] # Read-only + property shared_access_signature: Optional[str] # Read-only + + def __contains__(self, key: str) -> bool: ... + + def __delitem__(self, key: str) -> None: ... + + def __eq__(self, other: Any) -> bool: ... + + def __getitem__(self, key: str) -> Any: ... + + def __init__( + self, + *, + endpoint: str, + eventhub_name: Optional[str] = ..., + fully_qualified_namespace: str, + shared_access_key: Optional[str] = ..., + shared_access_key_name: Optional[str] = ..., + shared_access_signature: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + def __len__(self) -> int: ... + + def __ne__(self, other: Any) -> bool: ... + + def __repr__(self) -> str: ... + + def __setitem__( + self, + key: str, + item: Any + ) -> None: ... + + def __str__(self) -> str: ... + + def get( + self, + key: str, + default: Optional[Any] = None + ) -> Any: ... + + def has_key(self, k: str) -> bool: ... + + def items(self) -> List[Tuple[str, Any]]: ... + + def keys(self) -> List[str]: ... + + def update( + self, + *args: Any, + **kwargs: Any + ) -> None: ... + + def values(self) -> List[Any]: ... + + + class azure.eventhub.EventHubConsumerClient(ClientBase): implements ContextManager + + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + checkpoint_store: Union[CheckpointStore, None] = ..., + connection_verify: Union[str, None] = ..., + custom_endpoint_address: Union[str, None] = ..., + idle_timeout: Optional[float] = ..., + load_balancing_interval: Optional[float] = ..., + load_balancing_strategy: Union[str, LoadBalancingStrategy] = ..., + logging_enable: Optional[bool] = ..., + partition_ownership_expiration_interval: Optional[float] = ..., + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @classmethod + def from_connection_string( + cls, + conn_str: str, + consumer_group: str, + *, + auth_timeout: float = 60, + checkpoint_store: Optional[CheckpointStore] = ..., + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + eventhub_name: Optional[str] = ..., + http_proxy: Optional[Dict[str, Union[str, int]]] = ..., + idle_timeout: Optional[float] = ..., + load_balancing_interval: float = 30, + load_balancing_strategy: Union[str, LoadBalancingStrategy] = LoadBalancingStrategy.GREEDY, + logging_enable: bool = False, + partition_ownership_expiration_interval: Optional[float] = ..., + retry_backoff_factor: float = 0.8, + retry_backoff_max: float = 120, + retry_mode: Literal["exponential", "fixed"] = "exponential", + retry_total: int = 3, + ssl_context: Optional[SSLContext] = ..., + transport_type: TransportType = TransportType.Amqp, + uamqp_transport: bool = False, + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> EventHubConsumerClient: ... + + def close(self) -> None: ... + + def get_eventhub_properties(self) -> Dict[str, Any]: ... + + def get_partition_ids(self) -> List[str]: ... + + def get_partition_properties(self, partition_id: str) -> Dict[str, Any]: ... + + def receive( + self, + on_event: Callable[[PartitionContext, Optional[EventData]], None], + *, + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[PartitionContext, Exception], None]] = ..., + on_partition_close: Optional[Callable[[PartitionContext, CloseReason], None]] = ..., + on_partition_initialize: Optional[Callable[[PartitionContext], None]] = ..., + owner_level: Optional[int] = ..., + partition_id: Optional[str] = ..., + prefetch: int = 300, + starting_position: Optional[Union[str, int, datetime, Dict[str, Any]]] = ..., + starting_position_inclusive: Union[bool, Dict[str, bool]] = False, + track_last_enqueued_event_properties: bool = False, + **kwargs: Any + ) -> None: ... + + def receive_batch( + self, + on_event_batch: Callable[[PartitionContext, List[EventData]], None], + *, + max_batch_size: int = 300, + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[PartitionContext, Exception], None]] = ..., + on_partition_close: Optional[Callable[[PartitionContext, CloseReason], None]] = ..., + on_partition_initialize: Optional[Callable[[PartitionContext], None]] = ..., + owner_level: Optional[int] = ..., + partition_id: Optional[str] = ..., + prefetch: int = 300, + starting_position: Optional[Union[str, int, datetime, Dict[str, Any]]] = ..., + starting_position_inclusive: Union[bool, Dict[str, bool]] = False, + track_last_enqueued_event_properties: bool = False, + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.EventHubProducerClient(ClientBase): implements ContextManager + property total_buffered_event_count: Optional[int] # Read-only + + @overload + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + buffer_concurrency: Union[ThreadPoolExecutor, int, None] = ..., + buffered_mode: Literal[False] = False, + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + http_proxy: Optional[Dict] = ..., + idle_timeout: Optional[float] = ..., + logging_enable: Optional[bool] = ..., + max_buffer_length: Optional[int] = ..., + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[SendEventTypes, Optional[str], Exception], None]] = ..., + on_success: Optional[Callable[[SendEventTypes, Optional[str]], None]] = ..., + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @overload + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + buffer_concurrency: Optional[Union[ThreadPoolExecutor, int]] = ..., + buffered_mode: Literal[True], + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + http_proxy: Optional[Dict] = ..., + idle_timeout: Optional[float] = ..., + logging_enable: Optional[bool] = ..., + max_buffer_length: int = 1500, + max_wait_time: float = 1, + on_error: Callable[[SendEventTypes, Optional[str], Exception], None], + on_success: Callable[[SendEventTypes, Optional[str]], None], + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @classmethod + @overload + def from_connection_string( + cls, + conn_str: str, + *, + buffered_mode: Literal[False] = False, + eventhub_name: Optional[str] = ..., + **kwargs: Any + ) -> EventHubProducerClient: ... + + @classmethod + @overload + def from_connection_string( + cls, + conn_str: str, + *, + buffer_concurrency: Optional[Union[ThreadPoolExecutor, int]] = ..., + buffered_mode: Literal[True], + eventhub_name: Optional[str] = ..., + max_buffer_length: int = 1500, + max_wait_time: float = 1, + on_error: Callable[[SendEventTypes, Optional[str], Exception], None], + on_success: Callable[[SendEventTypes, Optional[str]], None], + **kwargs: Any + ) -> EventHubProducerClient: ... + + def close( + self, + *, + flush: bool = True, + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + def create_batch( + self, + *, + max_size_in_bytes: Optional[int] = ..., + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ..., + **kwargs: Any + ) -> EventDataBatch: ... + + def flush( + self, + *, + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + def get_buffered_event_count(self, partition_id: str) -> Optional[int]: ... + + def get_eventhub_properties(self) -> Dict[str, Any]: ... + + def get_partition_ids(self) -> List[str]: ... + + def get_partition_properties(self, partition_id: str) -> Dict[str, Any]: ... + + def send_batch( + self, + event_data_batch: Union[EventDataBatch, SendEventTypes], + *, + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ..., + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + def send_event( + self, + event_data: Union[EventData, AmqpAnnotatedMessage], + *, + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ..., + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.EventHubSharedKeyCredential: + + def __init__( + self, + policy: str, + key: str + ) -> None: ... + + def get_token( + self, + *scopes: str, + **kwargs: Any + ) -> AccessToken: ... + + + class azure.eventhub.LoadBalancingStrategy(Enum): + BALANCED = "balanced" + GREEDY = "greedy" + + + class azure.eventhub.PartitionContext: + property last_enqueued_event_properties: Optional[Dict[str, Any]] # Read-only + + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + partition_id: str, + checkpoint_store: Optional[CheckpointStore] = None + ) -> None: ... + + def update_checkpoint( + self, + event: Optional[EventData] = None, + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.TransportType(Enum): + Amqp = 1 + AmqpOverWebsocket = 2 + + +namespace azure.eventhub.aio + + class azure.eventhub.aio.CheckpointStore(ABC): + + @abstractmethod + async def claim_ownership( + self, + ownership_list: Iterable[Dict[str, Any]], + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + async def list_checkpoints( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + async def list_ownership( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + async def update_checkpoint( + self, + checkpoint: Dict[str, Optional[Union[str, int]]], + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.aio.EventHubConsumerClient(ClientBaseAsync): implements AsyncContextManager + + def __enter__(self) -> None: ... + + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + checkpoint_store: Optional[CheckpointStore] = ..., + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + http_proxy: Union[dict[str, str], dict[str, int], None] = ..., + idle_timeout: Optional[float] = ..., + load_balancing_interval: Optional[float] = ..., + load_balancing_strategy: Union[str, LoadBalancingStrategy] = ..., + logging_enable: Optional[bool] = ..., + partition_ownership_expiration_interval: Optional[float] = ..., + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @classmethod + def from_connection_string( + cls, + conn_str: str, + consumer_group: str, + *, + auth_timeout: float = 60, + checkpoint_store: Optional[CheckpointStore] = ..., + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + eventhub_name: Optional[str] = ..., + http_proxy: Optional[Dict[str, Union[str, int]]] = ..., + idle_timeout: Optional[float] = ..., + load_balancing_interval: float = 30, + load_balancing_strategy: Union[str, LoadBalancingStrategy] = LoadBalancingStrategy.GREEDY, + logging_enable: bool = False, + partition_ownership_expiration_interval: Optional[float] = ..., + retry_backoff_factor: float = 0.8, + retry_backoff_max: float = 120, + retry_mode: Literal["exponential", "fixed"] = "exponential", + retry_total: int = 3, + ssl_context: Optional[SSLContext] = ..., + transport_type: TransportType = TransportType.Amqp, + uamqp_transport: bool = False, + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> EventHubConsumerClient: ... + + async def close(self) -> None: ... + + async def get_eventhub_properties(self) -> Dict[str, Any]: ... + + async def get_partition_ids(self) -> List[str]: ... + + async def get_partition_properties(self, partition_id: str) -> Dict[str, Any]: ... + + async def receive( + self, + on_event: Callable[[PartitionContext, Optional[EventData]], Awaitable[None]], + *, + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[PartitionContext, Exception], Awaitable[None]]] = ..., + on_partition_close: Optional[Callable[[PartitionContext, CloseReason], Awaitable[None]]] = ..., + on_partition_initialize: Optional[Callable[[PartitionContext], Awaitable[None]]] = ..., + owner_level: Optional[int] = ..., + partition_id: Optional[str] = ..., + prefetch: int = 300, + starting_position: Optional[Union[str, int, datetime, Dict[str, Any]]] = ..., + starting_position_inclusive: Union[bool, Dict[str, bool]] = False, + track_last_enqueued_event_properties: bool = False + ) -> None: ... + + async def receive_batch( + self, + on_event_batch: Callable[[PartitionContext, List[EventData]], Awaitable[None]], + *, + max_batch_size: int = 300, + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[PartitionContext, Exception], Awaitable[None]]] = ..., + on_partition_close: Optional[Callable[[PartitionContext, CloseReason], Awaitable[None]]] = ..., + on_partition_initialize: Optional[Callable[[PartitionContext], Awaitable[None]]] = ..., + owner_level: Optional[int] = ..., + partition_id: Optional[str] = ..., + prefetch: int = 300, + starting_position: Optional[Union[str, int, datetime, Dict[str, Any]]] = ..., + starting_position_inclusive: Union[bool, Dict[str, bool]] = False, + track_last_enqueued_event_properties: bool = False + ) -> None: ... + + + class azure.eventhub.aio.EventHubProducerClient(ClientBaseAsync): implements AsyncContextManager + property total_buffered_event_count: Optional[int] # Read-only + + def __enter__(self) -> None: ... + + @overload + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + buffered_mode: Literal[False] = False, + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + http_proxy: Optional[dict] = ..., + idle_timeout: Optional[float] = ..., + logging_enable: Optional[bool] = ..., + max_buffer_length: Optional[int] = ..., + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[SendEventTypes, Optional[str], Exception], Awaitable[None]]] = ..., + on_success: Optional[Callable[[SendEventTypes, Optional[str]], Awaitable[None]]] = ..., + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @overload + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + buffered_mode: Literal[True], + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + http_proxy: Optional[dict] = ..., + idle_timeout: Optional[float] = ..., + logging_enable: Optional[bool] = ..., + max_buffer_length: int = 1500, + max_wait_time: float = 1, + on_error: Callable[[SendEventTypes, Optional[str], Exception], Awaitable[None]], + on_success: Callable[[SendEventTypes, Optional[str]], Awaitable[None]], + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @classmethod + @overload + def from_connection_string( + cls, + conn_str: str, + *, + buffered_mode: Literal[False] = False, + eventhub_name: Optional[str] = ..., + **kwargs: Any + ) -> EventHubProducerClient: ... + + @classmethod + @overload + def from_connection_string( + cls, + conn_str: str, + *, + buffered_mode: Literal[True], + eventhub_name: Optional[str] = ..., + max_buffer_length: int = 1500, + max_wait_time: float = 1, + on_error: Callable[[SendEventTypes, Optional[str], Exception], Awaitable[None]], + on_success: Callable[[SendEventTypes, Optional[str]], Awaitable[None]], + **kwargs: Any + ) -> EventHubProducerClient: ... + + async def close( + self, + *, + flush: bool = True, + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + async def create_batch( + self, + *, + max_size_in_bytes: Optional[int] = ..., + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ... + ) -> EventDataBatch: ... + + async def flush( + self, + *, + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + def get_buffered_event_count(self, partition_id: str) -> Optional[int]: ... + + async def get_eventhub_properties(self) -> Dict[str, Any]: ... + + async def get_partition_ids(self) -> List[str]: ... + + async def get_partition_properties(self, partition_id: str) -> Dict[str, Any]: ... + + async def send_batch( + self, + event_data_batch: Union[EventDataBatch, SendEventTypes], + *, + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ..., + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + async def send_event( + self, + event_data: Union[EventData, AmqpAnnotatedMessage], + *, + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ..., + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.aio.EventHubSharedKeyCredential: + + def __init__( + self, + policy: str, + key: str + ): ... + + async def get_token( + self, + *scopes: str, + **kwargs: Any + ) -> AccessToken: ... + + + class azure.eventhub.aio.PartitionContext: + property last_enqueued_event_properties: Optional[Dict[str, Any]] # Read-only + + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + partition_id: str, + checkpoint_store: Optional[CheckpointStore] = None + ) -> None: ... + + async def update_checkpoint( + self, + event: Optional[EventData] = None, + **kwargs: Any + ) -> None: ... + + +namespace azure.eventhub.amqp + + class azure.eventhub.amqp.AmqpAnnotatedMessage: + property annotations: Optional[Dict[Union[str, bytes], Any]] + property application_properties: Optional[Dict[Union[str, bytes], Any]] + property body: Any # Read-only + property body_type: AmqpMessageBodyType # Read-only + property delivery_annotations: Optional[Dict[Union[str, bytes], Any]] + property footer: Optional[Dict[Any, Any]] + property header: Optional[AmqpMessageHeader] + property properties: Optional[AmqpMessageProperties] + + def __init__( + self, + *, + annotations: Optional[Dict] = ..., + application_properties: Optional[Dict] = ..., + data_body: Union[str, bytes, List[Union[str, bytes]]] = ..., + delivery_annotations: Optional[Dict] = ..., + footer: Optional[Dict] = ..., + header: Optional[AmqpMessageHeader] = ..., + properties: Optional[AmqpMessageProperties] = ..., + sequence_body: List[Any] = ..., + value_body: Any = ..., + **kwargs: Any + ) -> None: ... + + def __repr__(self) -> str: ... + + def __str__(self) -> str: ... + + + class azure.eventhub.amqp.AmqpMessageBodyType(str, Enum, metaclass=CaseInsensitiveEnumMeta): + DATA = "data" + SEQUENCE = "sequence" + VALUE = "value" + + + class azure.eventhub.amqp.AmqpMessageHeader(DictMixin): + delivery_count: Optional[int] + durable: Optional[bool] + first_acquirer: Optional[bool] + priority: Optional[int] + time_to_live: Optional[int] + + def __contains__(self, key: str) -> bool: ... + + def __delitem__(self, key: str) -> None: ... + + def __eq__(self, other: Any) -> bool: ... + + def __getitem__(self, key: str) -> Any: ... + + def __init__( + self, + *, + delivery_count: Optional[int] = ..., + durable: Optional[bool] = ..., + first_acquirer: Optional[bool] = ..., + priority: Optional[int] = ..., + time_to_live: Optional[int] = ..., + **kwargs + ): ... + + def __len__(self) -> int: ... + + def __ne__(self, other: Any) -> bool: ... + + def __repr__(self) -> str: ... + + def __setitem__( + self, + key: str, + item: Any + ) -> None: ... + + def __str__(self) -> str: ... + + def get( + self, + key: str, + default: Optional[Any] = None + ) -> Any: ... + + def has_key(self, k: str) -> bool: ... + + def items(self) -> List[Tuple[str, Any]]: ... + + def keys(self) -> List[str]: ... + + def update( + self, + *args: Any, + **kwargs: Any + ) -> None: ... + + def values(self) -> List[Any]: ... + + + class azure.eventhub.amqp.AmqpMessageProperties(DictMixin): + absolute_expiry_time: Optional[int] + content_encoding: Optional[bytes] + content_type: Optional[bytes] + correlation_id: Optional[bytes] + creation_time: Optional[int] + group_id: Optional[bytes] + group_sequence: Optional[int] + message_id: Optional[bytes] + reply_to: Optional[bytes] + reply_to_group_id: Optional[bytes] + subject: Optional[bytes] + to: Optional[bytes] + user_id: Optional[bytes] + + def __contains__(self, key: str) -> bool: ... + + def __delitem__(self, key: str) -> None: ... + + def __eq__(self, other: Any) -> bool: ... + + def __getitem__(self, key: str) -> Any: ... + + def __init__( + self, + *, + absolute_expiry_time: Optional[int] = ..., + content_encoding: Optional[Union[str, bytes]] = ..., + content_type: Optional[Union[str, bytes]] = ..., + correlation_id: Optional[Union[str, bytes]] = ..., + creation_time: Optional[int] = ..., + group_id: Optional[Union[str, bytes]] = ..., + group_sequence: Optional[int] = ..., + message_id: Optional[Union[str, bytes, uuid.UUID]] = ..., + reply_to: Optional[Union[str, bytes]] = ..., + reply_to_group_id: Optional[Union[str, bytes]] = ..., + subject: Optional[Union[str, bytes]] = ..., + to: Optional[Union[str, bytes]] = ..., + user_id: Optional[Union[str, bytes]] = ..., + **kwargs + ): ... + + def __len__(self) -> int: ... + + def __ne__(self, other: Any) -> bool: ... + + def __repr__(self) -> str: ... + + def __setitem__( + self, + key: str, + item: Any + ) -> None: ... + + def __str__(self) -> str: ... + + def get( + self, + key: str, + default: Optional[Any] = None + ) -> Any: ... + + def has_key(self, k: str) -> bool: ... + + def items(self) -> List[Tuple[str, Any]]: ... + + def keys(self) -> List[str]: ... + + def update( + self, + *args: Any, + **kwargs: Any + ) -> None: ... + + def values(self) -> List[Any]: ... + + +namespace azure.eventhub.exceptions + + class azure.eventhub.exceptions.AuthenticationError(ConnectError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.ClientClosedError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.ConnectError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.ConnectionLostError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.EventDataError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.EventDataSendError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.EventHubError(Exception): + details: list[str] + error: str + message: str + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.OperationTimeoutError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.OwnershipLostError(Exception): + + +``` \ No newline at end of file diff --git a/sdk/eventhub/azure-eventhub/api.metadata.yml b/sdk/eventhub/azure-eventhub/api.metadata.yml new file mode 100644 index 000000000000..cd20a8c5db35 --- /dev/null +++ b/sdk/eventhub/azure-eventhub/api.metadata.yml @@ -0,0 +1,3 @@ +apiMdSha256: c54b169f6a986c1d8193fc3420e451cc005b078a2c347782b9ef349b66633663 +parserVersion: 0.3.28 +pythonVersion: 3.12.13 From 61bf6dc9badd14abd6cb78947d89ccd40d72296b Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 16 Jul 2026 18:16:20 -0400 Subject: [PATCH 7/7] 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 cd57402203fa..fb61ee13307a 100644 --- a/sdk/eventhub/azure-eventhub/CHANGELOG.md +++ b/sdk/eventhub/azure-eventhub/CHANGELOG.md @@ -4,7 +4,7 @@ ### Bugs Fixed -- 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 88f5c178a6db..78c4e8adfc72 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md @@ -13,7 +13,7 @@ - Fixed a bug where sending a batched or multi-message payload with `uamqp_transport=True` raised `TypeError: 'BatchMessage' object is not subscriptable` (and a masked `AttributeError` on the list path) when the first message carried a `message_id`, `session_id`, or `partition_key`. The batch envelope properties are now set through the transport-appropriate code path. (regression from [#42598](https://github.com/Azure/azure-sdk-for-python/pull/42598)) - Fixed a bug where the async receiver factory methods on `azure.servicebus.aio.ServiceBusClient` (`get_queue_receiver`, `get_subscription_receiver`) and the async `ServiceBusReceiver` annotated the `auto_lock_renewer` keyword with the synchronous `AutoLockRenewer`, causing static type checkers to reject the documented `azure.servicebus.aio.AutoLockRenewer` usage. The annotation now references the async `AutoLockRenewer`, matching the docstrings and runtime behavior. ([#47948](https://github.com/Azure/azure-sdk-for-python/issues/47948)) - Fixed a bug where closing a `PEEK_LOCK` receiver did not release messages that had been prefetched into the client buffer or were still in flight, so they remained locked at the broker until lock expiry — delaying their redelivery and inflating their delivery count. On close, a non-session `PEEK_LOCK` receiver now drains the link (stopping the broker and flushing in-flight transfers) and releases the buffered messages (`released` disposition), so the broker can redeliver them immediately without incrementing the delivery count. ([#42917](https://github.com/Azure/azure-sdk-for-python/issues/42917)) -- 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.