diff --git a/sdk/eventhub/azure-eventhub/CHANGELOG.md b/sdk/eventhub/azure-eventhub/CHANGELOG.md index f287cb0b8fde..fb61ee13307a 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 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) ### Bugs Fixed 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 diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py index 32ef0ddd9c12..68e571a45c56 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,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. @@ -432,6 +475,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] @@ -439,6 +488,25 @@ 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 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 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..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 @@ -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 @@ -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." diff --git a/sdk/servicebus/azure-servicebus/CHANGELOG.md b/sdk/servicebus/azure-servicebus/CHANGELOG.md index 0d105a35dc45..78c4e8adfc72 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, 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 32ef0ddd9c12..68e571a45c56 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,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. @@ -432,6 +475,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] @@ -439,6 +488,25 @@ 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 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 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..b22f6c7238be --- /dev/null +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py @@ -0,0 +1,216 @@ +import pathlib +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 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."