Skip to content
1 change: 1 addition & 0 deletions sdk/eventhub/azure-eventhub/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
### Bugs Fixed

- Fixed a bug where the async pure-Python AMQP transport failed to connect with `[Errno 22] Invalid argument` (`amqp:socket-error`) inside containerized/virtualized environments such as Docker Desktop on macOS. The transport no longer reads back and re-applies platform-negotiated TCP options (e.g. `TCP_MAXSEG`) that some platforms reject via `setsockopt`. Also fixed the async transport to apply default TCP socket settings even when no custom `socket_settings` are provided. ([#45394](https://github.com/Azure/azure-sdk-for-python/issues/45394))
- Fixed a bug in the pyAMQP transport where decoding an incoming performative whose trailing null fields were omitted by the sender (permitted by AMQP 1.0 section 1.4) raised `IndexError`/`TypeError`. The decoded field list is now padded to the performative's full field count so omitted trailing fields read back as 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)

Expand Down
68 changes: 68 additions & 0 deletions sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@

from . import described
from .message import Message, Header, Properties
from . import performatives

if TYPE_CHECKING:
from .message import MessageDict
Expand Down Expand Up @@ -416,6 +417,48 @@ def decode_payload(buffer: memoryview) -> Message:
return Message(**message_properties)


# 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 here.
_PERFORMATIVE_FIELD_DEFAULTS: Dict[int, List[Any]] = {
# _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,
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,
)
}

# 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
# described type then ulong.
Expand All @@ -432,13 +475,38 @@ 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]
buffer = data[6:]
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 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:
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
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,14 @@
import pathlib
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


Expand Down Expand Up @@ -45,3 +54,215 @@ 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 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]))
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_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_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.
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
# 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"


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]))


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):
# 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) == 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)


# 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.
@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",
[
(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


# 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."
1 change: 1 addition & 0 deletions sdk/servicebus/azure-servicebus/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
- Fixed a bug where the async pure-Python AMQP transport failed to connect with `[Errno 22] Invalid argument` (`amqp:socket-error`) inside containerized/virtualized environments such as Docker Desktop on macOS. The transport no longer reads back and re-applies platform-negotiated TCP options (e.g. `TCP_MAXSEG`) that some platforms reject via `setsockopt`. ([#45394](https://github.com/Azure/azure-sdk-for-python/issues/45394))
- Fixed a bug where passing a `fully_qualified_namespace` that included a port and/or trailing path (for example the `https://<namespace>.servicebus.windows.net:443/` form that Azure returns when provisioning a namespace) raised `ServiceBusAuthenticationError`. The namespace is now normalized to its bare host, matching the .NET and JavaScript SDKs. ([#44034](https://github.com/Azure/azure-sdk-for-python/issues/44034))
- Fixed a bug where iterating over a `ServiceBusReceiver` suppressed automatic HTTP instrumentation (e.g. from `opentelemetry-instrumentation-httpx`/`requests`) while user code processed a received message, causing the user's own outbound HTTP spans to be dropped. The receive tracing span is now closed before the message is yielded to the caller, so suppression no longer leaks into message processing. ([#42755](https://github.com/Azure/azure-sdk-for-python/issues/42755))
- Fixed a bug in the pyAMQP transport where decoding an incoming performative whose trailing null fields were omitted by the sender (permitted by AMQP 1.0 section 1.4) raised `IndexError`/`TypeError`. The decoded field list is now padded to the performative's full field count so omitted trailing fields read back as their AMQP-defined default, including the compact `list0` encoding where every field is omitted. 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

Expand Down
Loading
Loading