From fa88c3c64f9af56b362fe0b2d2a05ee2924edd7c Mon Sep 17 00:00:00 2001 From: Carter Tinney Date: Fri, 21 Aug 2026 16:08:26 -0700 Subject: [PATCH 1/3] fix: validate encoded telemetry message size Measure IoT Hub messages from their encoded payload and logical property contribution instead of CPython object layout, and make boundary coverage deterministic across supported Python versions. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../iot/device/iothub/aio/async_clients.py | 4 +- .../azure/iot/device/iothub/models/message.py | 86 +++++++++-- .../iothub/pipeline/mqtt_topic_iothub.py | 54 ++----- .../azure/iot/device/iothub/sync_clients.py | 4 +- tests/unit/iothub/aio/test_async_clients.py | 136 +++++++----------- tests/unit/iothub/models/test_message.py | 134 ++++++++++++++++- tests/unit/iothub/test_sync_clients.py | 130 +++++++---------- 7 files changed, 316 insertions(+), 232 deletions(-) diff --git a/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py b/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py index 6eaca9c1a..08cb3288d 100644 --- a/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py +++ b/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py @@ -648,11 +648,11 @@ async def send_message_to_output(self, message: Union[Message, str], output_name if not isinstance(message, Message): message = Message(message) + message.output_name = output_name + if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: raise ValueError("Size of message can not exceed 256 KB.") - message.output_name = output_name - logger.info("Sending message to output:" + output_name + "...") send_output_message_async = async_adapter.emulate_async( self._mqtt_pipeline.send_output_message diff --git a/azure-iot-device/azure/iot/device/iothub/models/message.py b/azure-iot-device/azure/iot/device/iothub/models/message.py index 8332ec090..a7033ed96 100644 --- a/azure-iot-device/azure/iot/device/iothub/models/message.py +++ b/azure-iot-device/azure/iot/device/iothub/models/message.py @@ -5,8 +5,64 @@ # -------------------------------------------------------------------------- """This module contains a class representing messages that are sent or received. """ +from datetime import date + from azure.iot.device import constant -import sys + + +def _encode_message_data(data): + if isinstance(data, str): + return data.encode("utf-8") + if isinstance(data, (int, float)): + return str(data).encode("ascii") + if data is None: + return b"" + if not isinstance(data, (bytes, bytearray)): + raise TypeError("Message data must be a string, bytes, bytearray, int, float, or None.") + return data + + +def _get_system_properties(message): + properties = [] + if message.output_name: + properties.append(("$.on", str(message.output_name))) + if message.message_id: + properties.append(("$.mid", str(message.message_id))) + if message.correlation_id: + properties.append(("$.cid", str(message.correlation_id))) + if message.user_id: + properties.append(("$.uid", str(message.user_id))) + if message.content_type: + properties.append(("$.ct", str(message.content_type))) + if message.content_encoding: + properties.append(("$.ce", str(message.content_encoding))) + if message.iothub_interface_id: + properties.append(("$.ifid", str(message.iothub_interface_id))) + if message.expiry_time_utc: + expiry_time = ( + message.expiry_time_utc.isoformat() + if isinstance(message.expiry_time_utc, date) + else message.expiry_time_utc + ) + properties.append(("$.exp", str(expiry_time))) + return properties + + +def _get_custom_properties(message): + if not message.custom_properties: + return [] + + properties = [(str(key), str(value)) for key, value in message.custom_properties.items()] + properties.sort() + + keys = [key for key, _ in properties] + if len(keys) != len(set(keys)): + raise ValueError("Duplicate keys in custom properties!") + return properties + + +def _get_string_size(value): + return len(value.encode("utf-8")) class Message(object): @@ -65,14 +121,22 @@ def __str__(self): return str(self.data) def get_size(self) -> int: - total = 0 - total = total + sum( - sys.getsizeof(v) - for v in self.__dict__.values() - if v is not None and v is not self.custom_properties + """Return the message size in bytes as measured by IoT Hub. + + The size is the encoded body plus system property values and application property names + and values. Strings are measured as UTF-8, matching the MQTT transport; bytes and + bytearrays are measured as-is. MQTT topic and packet overhead are not included. + + :raises TypeError: If the message data is not a payload type supported by the MQTT + transport. + :raises ValueError: If custom property keys are duplicated after string conversion. + """ + payload_size = len(_encode_message_data(self.data)) + system_property_size = sum( + _get_string_size(value) for _, value in _get_system_properties(self) + ) + application_property_size = sum( + _get_string_size(key) + _get_string_size(value) + for key, value in _get_custom_properties(self) ) - if self.custom_properties: - total = total + sum( - sys.getsizeof(v) for v in self.custom_properties.values() if v is not None - ) - return total + return payload_size + system_property_size + application_property_size diff --git a/azure-iot-device/azure/iot/device/iothub/pipeline/mqtt_topic_iothub.py b/azure-iot-device/azure/iot/device/iothub/pipeline/mqtt_topic_iothub.py index 9db638882..3538625d3 100644 --- a/azure-iot-device/azure/iot/device/iothub/pipeline/mqtt_topic_iothub.py +++ b/azure-iot-device/azure/iot/device/iothub/pipeline/mqtt_topic_iothub.py @@ -5,9 +5,13 @@ # -------------------------------------------------------------------------- import logging -from datetime import date import urllib +from azure.iot.device.iothub.models.message import ( + _get_custom_properties, + _get_system_properties, +) + logger = logging.getLogger(__name__) # NOTE: Whenever using standard URL encoding via the urllib.parse.quote() API @@ -343,59 +347,17 @@ def encode_message_properties_in_topic(message_to_send, topic): "devices//modules//messages/events/ :return: The topic which has been uri-encoded """ - system_properties = [] - if message_to_send.output_name: - system_properties.append(("$.on", str(message_to_send.output_name))) - if message_to_send.message_id: - system_properties.append(("$.mid", str(message_to_send.message_id))) - - if message_to_send.correlation_id: - system_properties.append(("$.cid", str(message_to_send.correlation_id))) - - if message_to_send.user_id: - system_properties.append(("$.uid", str(message_to_send.user_id))) - - if message_to_send.content_type: - system_properties.append(("$.ct", str(message_to_send.content_type))) - - if message_to_send.content_encoding: - system_properties.append(("$.ce", str(message_to_send.content_encoding))) - - if message_to_send.iothub_interface_id: - system_properties.append(("$.ifid", str(message_to_send.iothub_interface_id))) - - if message_to_send.expiry_time_utc: - system_properties.append( - ( - "$.exp", - message_to_send.expiry_time_utc.isoformat() # returns string - if isinstance(message_to_send.expiry_time_utc, date) - else message_to_send.expiry_time_utc, - ) - ) - + system_properties = _get_system_properties(message_to_send) system_properties_encoded = urllib.parse.urlencode( system_properties, quote_via=urllib.parse.quote ) topic += system_properties_encoded - if message_to_send.custom_properties and len(message_to_send.custom_properties) > 0: + custom_prop_seq = _get_custom_properties(message_to_send) + if custom_prop_seq: if system_properties and len(system_properties) > 0: topic += "&" - # Convert the custom properties to a sorted list in order to ensure the - # resulting ordering in the topic string is consistent across versions of Python. - # Convert to the properties to strings for safety. - custom_prop_seq = [ - (str(i[0]), str(i[1])) for i in list(message_to_send.custom_properties.items()) - ] - custom_prop_seq.sort() - - # Validate that string conversion has not created duplicate keys - keys = [i[0] for i in custom_prop_seq] - if len(keys) != len(set(keys)): - raise ValueError("Duplicate keys in custom properties!") - user_properties_encoded = urllib.parse.urlencode( custom_prop_seq, quote_via=urllib.parse.quote ) diff --git a/azure-iot-device/azure/iot/device/iothub/sync_clients.py b/azure-iot-device/azure/iot/device/iothub/sync_clients.py index 4088e0b6f..6023f31e6 100644 --- a/azure-iot-device/azure/iot/device/iothub/sync_clients.py +++ b/azure-iot-device/azure/iot/device/iothub/sync_clients.py @@ -669,11 +669,11 @@ def send_message_to_output(self, message: Union[Message, str], output_name: str) if not isinstance(message, Message): message = Message(message) + message.output_name = output_name + if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: raise ValueError("Size of message can not exceed 256 KB.") - message.output_name = output_name - logger.info("Sending message to output:" + output_name + "...") callback = EventedCallback() diff --git a/tests/unit/iothub/aio/test_async_clients.py b/tests/unit/iothub/aio/test_async_clients.py index 3796f494a..dfa0cc8e0 100644 --- a/tests/unit/iothub/aio/test_async_clients.py +++ b/tests/unit/iothub/aio/test_async_clients.py @@ -9,7 +9,6 @@ import asyncio import time import urllib -import sys from azure.iot.device import exceptions as client_exceptions from azure.iot.device.common.auth import sastoken as st from azure.iot.device.iothub.aio import IoTHubDeviceClient, IoTHubModuleClient @@ -659,10 +658,10 @@ def fail_send_message(message, callback): [ pytest.param("message", id="String input"), pytest.param(222, id="Integer input"), - pytest.param(object(), id="Object input"), + pytest.param(1.5, id="Float input"), + pytest.param(b"message", id="Bytes input"), + pytest.param(bytearray(b"message"), id="Bytearray input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) async def test_wraps_data_in_message_and_calls_pipeline_send_message( @@ -674,44 +673,27 @@ async def test_wraps_data_in_message_and_calls_pipeline_send_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input - @pytest.mark.it("Raises error when message data size is greater than 256 KB") - async def test_raises_error_when_message_data_greater_than_256(self, client, mqtt_pipeline): - data_input = "serpensortia" * 256000 - message = Message(data_input) - with pytest.raises(ValueError) as e_info: - await client.send_message(message) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_message.call_count == 0 - - @pytest.mark.it("Raises error when message size is greater than 256 KB") - async def test_raises_error_when_message_size_greater_than_256(self, client, mqtt_pipeline): - data_input = "serpensortia" - message = Message(data_input) - message.custom_properties["spell"] = data_input * 256000 - with pytest.raises(ValueError) as e_info: - await client.send_message(message) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_message.call_count == 0 - - @pytest.mark.skipif( - sys.version_info >= (3, 12), - reason="Python 3.12 appears to have an issue. Investigate further.", - ) - @pytest.mark.it("Does not raises error when message data size is equal to 256 KB") - async def test_raises_error_when_message_data_equal_to_256(self, client, mqtt_pipeline): - data_input = "a" * 262095 - message = Message(data_input) - # This check was put as message class may undergo the default content type encoding change - # and the above calculation will change. - if message.get_size() != device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: - assert False - - await client.send_message(message) + @pytest.mark.it("Validates message size at the 256 KB boundary") + @pytest.mark.parametrize( + "size_delta, raises_error", + [ + pytest.param(-1, False, id="Below limit"), + pytest.param(0, False, id="At limit"), + pytest.param(1, True, id="Above limit"), + ], + ) + async def test_validates_message_size(self, client, mqtt_pipeline, size_delta, raises_error): + message_size = device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT + size_delta + message = Message(b"a" * message_size) - assert mqtt_pipeline.send_message.call_count == 1 - sent_message = mqtt_pipeline.send_message.call_args[0][0] - assert isinstance(sent_message, Message) - assert sent_message.data == data_input + if raises_error: + with pytest.raises(ValueError, match="256 KB"): + await client.send_message(message) + assert mqtt_pipeline.send_message.call_count == 0 + else: + await client.send_message(message) + assert mqtt_pipeline.send_message.call_count == 1 + assert mqtt_pipeline.send_message.call_args[0][0] is message class SharedClientReceiveMethodRequestTests(object): @@ -2061,10 +2043,10 @@ def fail_send_output_message(message, callback): [ pytest.param("message", id="String input"), pytest.param(222, id="Integer input"), - pytest.param(object(), id="Object input"), + pytest.param(1.5, id="Float input"), + pytest.param(b"message", id="Bytes input"), + pytest.param(bytearray(b"message"), id="Bytearray input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) async def test_send_message_to_output_calls_pipeline_wraps_data_in_message( @@ -2077,53 +2059,31 @@ async def test_send_message_to_output_calls_pipeline_wraps_data_in_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input - @pytest.mark.it("Raises error when message data size is greater than 256 KB") - async def test_raises_error_when_message_to_output_data_greater_than_256( - self, client, mqtt_pipeline - ): - output_name = "some_output" - data_input = "serpensortia" * 256000 - message = Message(data_input) - with pytest.raises(ValueError) as e_info: - await client.send_message_to_output(message, output_name) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_output_message.call_count == 0 - - @pytest.mark.it("Raises error when message size is greater than 256 KB") - async def test_raises_error_when_message_to_output_size_greater_than_256( - self, client, mqtt_pipeline - ): - output_name = "some_output" - data_input = "serpensortia" - message = Message(data_input) - message.custom_properties["spell"] = data_input * 256000 - with pytest.raises(ValueError) as e_info: - await client.send_message_to_output(message, output_name) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_output_message.call_count == 0 - - @pytest.mark.skipif( - sys.version_info >= (3, 12), - reason="Python 3.12 appears to have an issue. Investigate further.", + @pytest.mark.it("Validates message size at the 256 KB boundary") + @pytest.mark.parametrize( + "size_delta, raises_error", + [ + pytest.param(-1, False, id="Below limit"), + pytest.param(0, False, id="At limit"), + pytest.param(1, True, id="Above limit"), + ], ) - @pytest.mark.it("Does not raises error when message data size is equal to 256 KB") - async def test_raises_error_when_message_to_output_data_equal_to_256( - self, client, mqtt_pipeline - ): + async def test_validates_message_size(self, client, mqtt_pipeline, size_delta, raises_error): output_name = "some_output" - data_input = "a" * 262095 - message = Message(data_input) - # This check was put as message class may undergo the default content type encoding change - # and the above calculation will change. - if message.get_size() != device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: - assert False - - await client.send_message_to_output(message, output_name) + message_size = device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT + size_delta + output_name_size = len(output_name.encode("utf-8")) + message = Message(b"a" * (message_size - output_name_size)) + + if raises_error: + with pytest.raises(ValueError, match="256 KB"): + await client.send_message_to_output(message, output_name) + assert mqtt_pipeline.send_output_message.call_count == 0 + else: + await client.send_message_to_output(message, output_name) + assert mqtt_pipeline.send_output_message.call_count == 1 + assert mqtt_pipeline.send_output_message.call_args[0][0] is message - assert mqtt_pipeline.send_output_message.call_count == 1 - sent_message = mqtt_pipeline.send_output_message.call_args[0][0] - assert isinstance(sent_message, Message) - assert sent_message.data == data_input + assert message.get_size() == message_size @pytest.mark.describe("IoTHubModuleClient (Asynchronous) - .receive_message_on_input()") diff --git a/tests/unit/iothub/models/test_message.py b/tests/unit/iothub/models/test_message.py index d444a55fd..c9aa15b0d 100644 --- a/tests/unit/iothub/models/test_message.py +++ b/tests/unit/iothub/models/test_message.py @@ -4,8 +4,11 @@ # license information. # -------------------------------------------------------------------------- -import pytest +import datetime import logging + +import pytest + from azure.iot.device.iothub.models import Message from azure.iot.device import constant @@ -16,6 +19,50 @@ data_obj = Message(data_str) +def _make_unicode_message(target_size): + multibyte_character = "\N{SNAKE}" + remaining_size = target_size - len(multibyte_character.encode("utf-8")) + return Message(multibyte_character + ("a" * remaining_size)) + + +def _make_custom_property_message(target_size): + key = "custom" + value = "property" + property_size = len(key.encode("utf-8")) + len(value.encode("utf-8")) + message = Message(b"a" * (target_size - property_size)) + message.custom_properties[key] = value + return message + + +def _set_system_properties(message): + message.output_name = "output" + message.message_id = 1234 + message.correlation_id = 5678 + message.user_id = 4000 + message.content_type = "application/json" + message.content_encoding = "utf-8" + message.expiry_time_utc = datetime.datetime(2026, 8, 21, 15, 33, 42) + message.set_as_security_message() + return [ + "output", + "1234", + "5678", + "4000", + "application/json", + "utf-8", + constant.SECURITY_MESSAGE_INTERFACE_ID, + "2026-08-21T15:33:42", + ] + + +def _make_system_property_message(target_size): + message = Message(b"") + values = _set_system_properties(message) + property_size = sum(len(value.encode("utf-8")) for value in values) + message.data = b"a" * (target_size - property_size) + return message + + @pytest.mark.describe("Message") class TestMessage(object): @pytest.mark.it("Instantiates from data type") @@ -105,3 +152,88 @@ def test_setting_message_as_security_message(self): assert msg.iothub_interface_id is None msg.set_as_security_message() assert msg.iothub_interface_id == constant.SECURITY_MESSAGE_INTERFACE_ID + + @pytest.mark.it("Measures payload using the MQTT transport encoding") + @pytest.mark.parametrize( + "data, expected_size", + [ + pytest.param("message", 7, id="String"), + pytest.param("\N{LATIN SMALL LETTER E WITH ACUTE}\N{SNAKE}", 6, id="Unicode string"), + pytest.param(b"\x00\xff", 2, id="Bytes"), + pytest.param(bytearray(b"\x00\xff"), 2, id="Bytearray"), + pytest.param(1234, 4, id="Integer"), + pytest.param(-1.25, 5, id="Float"), + pytest.param(None, 0, id="None"), + ], + ) + def test_get_size_payload_types(self, data, expected_size): + assert Message(data).get_size() == expected_size + + @pytest.mark.it("Raises TypeError for payload types not supported by the MQTT transport") + @pytest.mark.parametrize( + "data", + [ + pytest.param({"a": 1}, id="Dictionary"), + pytest.param([1, 2, 3], id="List"), + pytest.param(object(), id="Object"), + ], + ) + def test_get_size_invalid_payload_type(self, data): + with pytest.raises(TypeError): + Message(data).get_size() + + @pytest.mark.it("Counts custom property names and values after string conversion") + def test_get_size_custom_properties(self): + message = Message(b"body") + message.custom_properties = {1: 23, "custom": "property"} + + expected_property_size = len("1") + len("23") + len("custom") + len("property") + assert message.get_size() == len(b"body") + expected_property_size + + @pytest.mark.it("Counts system property values but not their names") + def test_get_size_system_properties(self): + message = Message(b"body") + values = _set_system_properties(message) + + expected_property_size = sum(len(value.encode("utf-8")) for value in values) + assert message.get_size() == len(b"body") + expected_property_size + + @pytest.mark.it("Does not count receive-only properties") + def test_get_size_receive_only_properties(self): + message = Message("body") + message.input_name = "input" + message.ack = "full" + + assert message.get_size() == len("body") + + @pytest.mark.it("Does not count MQTT topic encoding or system property names") + def test_get_size_excludes_protocol_overhead(self): + message = Message(b"") + message.message_id = "#" + message.custom_properties["#"] = "#" + + assert message.get_size() == 3 + + @pytest.mark.it("Is deterministic below, at, and above the 256 KB limit") + @pytest.mark.parametrize( + "message_factory", + [ + pytest.param(lambda target_size: Message(b"a" * target_size), id="Bytes payload"), + pytest.param(_make_unicode_message, id="Unicode payload"), + pytest.param(_make_custom_property_message, id="Custom properties"), + pytest.param(_make_system_property_message, id="System properties"), + ], + ) + @pytest.mark.parametrize( + "size_delta", + [ + pytest.param(-1, id="Below limit"), + pytest.param(0, id="At limit"), + pytest.param(1, id="Above limit"), + ], + ) + def test_get_size_boundary(self, message_factory, size_delta): + expected_size = constant.TELEMETRY_MESSAGE_SIZE_LIMIT + size_delta + message = message_factory(expected_size) + + assert message.get_size() == expected_size diff --git a/tests/unit/iothub/test_sync_clients.py b/tests/unit/iothub/test_sync_clients.py index 6f5385994..72da23b37 100644 --- a/tests/unit/iothub/test_sync_clients.py +++ b/tests/unit/iothub/test_sync_clients.py @@ -9,7 +9,6 @@ import threading import time import urllib -import sys from azure.iot.device.iothub import IoTHubDeviceClient, IoTHubModuleClient from azure.iot.device import exceptions as client_exceptions from azure.iot.device.common.auth import sastoken as st @@ -648,10 +647,10 @@ def test_raises_error_on_pipeline_op_error( [ pytest.param("message", id="String input"), pytest.param(222, id="Integer input"), - pytest.param(object(), id="Object input"), + pytest.param(1.5, id="Float input"), + pytest.param(b"message", id="Bytes input"), + pytest.param(bytearray(b"message"), id="Bytearray input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) def test_wraps_data_in_message_and_calls_pipeline_send_message( @@ -663,44 +662,27 @@ def test_wraps_data_in_message_and_calls_pipeline_send_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input - @pytest.mark.it("Raises error when message data size is greater than 256 KB") - def test_raises_error_when_message_data_greater_than_256(self, client, mqtt_pipeline): - data_input = "serpensortia" * 25600 - message = Message(data_input) - with pytest.raises(ValueError) as e_info: - client.send_message(message) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_message.call_count == 0 - - @pytest.mark.it("Raises error when message size is greater than 256 KB") - def test_raises_error_when_message_size_greater_than_256(self, client, mqtt_pipeline): - data_input = "serpensortia" - message = Message(data_input) - message.custom_properties["spell"] = data_input * 25600 - with pytest.raises(ValueError) as e_info: - client.send_message(message) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_message.call_count == 0 - - @pytest.mark.skipif( - sys.version_info >= (3, 12), - reason="Python 3.12 appears to have an issue. Investigate further.", - ) - @pytest.mark.it("Does not raises error when message data size is equal to 256 KB") - def test_raises_error_when_message_data_equal_to_256(self, client, mqtt_pipeline): - data_input = "a" * 262095 - message = Message(data_input) - # This check was put as message class may undergo the default content type encoding change - # and the above calculation will change. - if message.get_size() != device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: - assert False - - client.send_message(message) + @pytest.mark.it("Validates message size at the 256 KB boundary") + @pytest.mark.parametrize( + "size_delta, raises_error", + [ + pytest.param(-1, False, id="Below limit"), + pytest.param(0, False, id="At limit"), + pytest.param(1, True, id="Above limit"), + ], + ) + def test_validates_message_size(self, client, mqtt_pipeline, size_delta, raises_error): + message_size = device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT + size_delta + message = Message(b"a" * message_size) - assert mqtt_pipeline.send_message.call_count == 1 - sent_message = mqtt_pipeline.send_message.call_args[0][0] - assert isinstance(sent_message, Message) - assert sent_message.data == data_input + if raises_error: + with pytest.raises(ValueError, match="256 KB"): + client.send_message(message) + assert mqtt_pipeline.send_message.call_count == 0 + else: + client.send_message(message) + assert mqtt_pipeline.send_message.call_count == 1 + assert mqtt_pipeline.send_message.call_args[0][0] is message class SharedClientReceiveMethodRequestTests(object): @@ -2303,10 +2285,10 @@ def test_raises_error_on_pipeline_op_error( [ pytest.param("message", id="String input"), pytest.param(222, id="Integer input"), - pytest.param(object(), id="Object input"), + pytest.param(1.5, id="Float input"), + pytest.param(b"message", id="Bytes input"), + pytest.param(bytearray(b"message"), id="Bytearray input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) def test_send_message_to_output_calls_pipeline_wraps_data_in_message( @@ -2319,47 +2301,31 @@ def test_send_message_to_output_calls_pipeline_wraps_data_in_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input - @pytest.mark.it("Raises error when message data size is greater than 256 KB") - def test_raises_error_when_message_to_output_data_greater_than_256(self, client, mqtt_pipeline): - output_name = "some_output" - data_input = "serpensortia" * 256000 - message = Message(data_input) - with pytest.raises(ValueError) as e_info: - client.send_message_to_output(message, output_name) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_output_message.call_count == 0 - - @pytest.mark.it("Raises error when message size is greater than 256 KB") - def test_raises_error_when_message_to_output_size_greater_than_256(self, client, mqtt_pipeline): - output_name = "some_output" - data_input = "serpensortia" - message = Message(data_input) - message.custom_properties["spell"] = data_input * 256000 - with pytest.raises(ValueError) as e_info: - client.send_message_to_output(message, output_name) - assert "256 KB" in e_info.value.args[0] - assert mqtt_pipeline.send_output_message.call_count == 0 - - @pytest.mark.skipif( - sys.version_info >= (3, 12), - reason="Python 3.12 appears to have an issue. Investigate further.", + @pytest.mark.it("Validates message size at the 256 KB boundary") + @pytest.mark.parametrize( + "size_delta, raises_error", + [ + pytest.param(-1, False, id="Below limit"), + pytest.param(0, False, id="At limit"), + pytest.param(1, True, id="Above limit"), + ], ) - @pytest.mark.it("Does not raises error when message data size is equal to 256 KB") - def test_raises_error_when_message_to_output_data_equal_to_256(self, client, mqtt_pipeline): + def test_validates_message_size(self, client, mqtt_pipeline, size_delta, raises_error): output_name = "some_output" - data_input = "a" * 262095 - message = Message(data_input) - # This check was put as message class may undergo the default content type encoding change - # and the above calculation will change. - if message.get_size() != device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: - assert False - - client.send_message_to_output(message, output_name) + message_size = device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT + size_delta + output_name_size = len(output_name.encode("utf-8")) + message = Message(b"a" * (message_size - output_name_size)) + + if raises_error: + with pytest.raises(ValueError, match="256 KB"): + client.send_message_to_output(message, output_name) + assert mqtt_pipeline.send_output_message.call_count == 0 + else: + client.send_message_to_output(message, output_name) + assert mqtt_pipeline.send_output_message.call_count == 1 + assert mqtt_pipeline.send_output_message.call_args[0][0] is message - assert mqtt_pipeline.send_output_message.call_count == 1 - sent_message = mqtt_pipeline.send_output_message.call_args[0][0] - assert isinstance(sent_message, Message) - assert sent_message.data == data_input + assert message.get_size() == message_size @pytest.mark.describe("IoTHubModuleClient (Synchronous) - .receive_message_on_input()") From c8fdf162348280fac5da2faaf4594be15762a29f Mon Sep 17 00:00:00 2001 From: Carter Tinney Date: Mon, 31 Aug 2026 18:08:44 -0700 Subject: [PATCH 2/3] fix: preserve invalid message error handling Defer unsupported payload types to the MQTT transport so send_message continues to surface ClientError with the original TypeError cause, while supported payloads retain deterministic size validation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../azure/iot/device/iothub/aio/async_clients.py | 4 ++-- .../azure/iot/device/iothub/models/message.py | 10 +++++++++- .../azure/iot/device/iothub/sync_clients.py | 4 ++-- tests/unit/iothub/aio/test_async_clients.py | 6 ++++++ tests/unit/iothub/test_sync_clients.py | 6 ++++++ 5 files changed, 25 insertions(+), 5 deletions(-) diff --git a/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py b/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py index 08cb3288d..ee6533967 100644 --- a/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py +++ b/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py @@ -348,7 +348,7 @@ async def send_message(self, message: Union[Message, str]) -> None: if not isinstance(message, Message): message = Message(message) - if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: + if message._is_size_exceeded(device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT): raise ValueError("Size of telemetry message can not exceed 256 KB.") logger.info("Sending message to Hub...") @@ -650,7 +650,7 @@ async def send_message_to_output(self, message: Union[Message, str], output_name message.output_name = output_name - if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: + if message._is_size_exceeded(device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT): raise ValueError("Size of message can not exceed 256 KB.") logger.info("Sending message to output:" + output_name + "...") diff --git a/azure-iot-device/azure/iot/device/iothub/models/message.py b/azure-iot-device/azure/iot/device/iothub/models/message.py index a7033ed96..f77a51f45 100644 --- a/azure-iot-device/azure/iot/device/iothub/models/message.py +++ b/azure-iot-device/azure/iot/device/iothub/models/message.py @@ -10,6 +10,10 @@ from azure.iot.device import constant +def _is_message_data_supported(data): + return data is None or isinstance(data, (str, bytes, bytearray, int, float)) + + def _encode_message_data(data): if isinstance(data, str): return data.encode("utf-8") @@ -17,7 +21,7 @@ def _encode_message_data(data): return str(data).encode("ascii") if data is None: return b"" - if not isinstance(data, (bytes, bytearray)): + if not _is_message_data_supported(data): raise TypeError("Message data must be a string, bytes, bytearray, int, float, or None.") return data @@ -120,6 +124,10 @@ def set_as_security_message(self): def __str__(self): return str(self.data) + def _is_size_exceeded(self, size_limit: int) -> bool: + # Unsupported payloads are deferred to the MQTT transport to preserve client error handling. + return _is_message_data_supported(self.data) and self.get_size() > size_limit + def get_size(self) -> int: """Return the message size in bytes as measured by IoT Hub. diff --git a/azure-iot-device/azure/iot/device/iothub/sync_clients.py b/azure-iot-device/azure/iot/device/iothub/sync_clients.py index 6023f31e6..1fa8949db 100644 --- a/azure-iot-device/azure/iot/device/iothub/sync_clients.py +++ b/azure-iot-device/azure/iot/device/iothub/sync_clients.py @@ -343,7 +343,7 @@ def send_message(self, message: Union[Message, str]) -> None: if not isinstance(message, Message): message = Message(message) - if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: + if message._is_size_exceeded(device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT): raise ValueError("Size of telemetry message can not exceed 256 KB.") logger.info("Sending message to Hub...") @@ -671,7 +671,7 @@ def send_message_to_output(self, message: Union[Message, str], output_name: str) message.output_name = output_name - if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: + if message._is_size_exceeded(device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT): raise ValueError("Size of message can not exceed 256 KB.") logger.info("Sending message to output:" + output_name + "...") diff --git a/tests/unit/iothub/aio/test_async_clients.py b/tests/unit/iothub/aio/test_async_clients.py index dfa0cc8e0..5d1946a6c 100644 --- a/tests/unit/iothub/aio/test_async_clients.py +++ b/tests/unit/iothub/aio/test_async_clients.py @@ -661,7 +661,10 @@ def fail_send_message(message, callback): pytest.param(1.5, id="Float input"), pytest.param(b"message", id="Bytes input"), pytest.param(bytearray(b"message"), id="Bytearray input"), + pytest.param(object(), id="Object input"), pytest.param(None, id="None input"), + pytest.param([1, "str"], id="List input"), + pytest.param({"a": 2}, id="Dictionary input"), ], ) async def test_wraps_data_in_message_and_calls_pipeline_send_message( @@ -2046,7 +2049,10 @@ def fail_send_output_message(message, callback): pytest.param(1.5, id="Float input"), pytest.param(b"message", id="Bytes input"), pytest.param(bytearray(b"message"), id="Bytearray input"), + pytest.param(object(), id="Object input"), pytest.param(None, id="None input"), + pytest.param([1, "str"], id="List input"), + pytest.param({"a": 2}, id="Dictionary input"), ], ) async def test_send_message_to_output_calls_pipeline_wraps_data_in_message( diff --git a/tests/unit/iothub/test_sync_clients.py b/tests/unit/iothub/test_sync_clients.py index 72da23b37..bfddb73b0 100644 --- a/tests/unit/iothub/test_sync_clients.py +++ b/tests/unit/iothub/test_sync_clients.py @@ -650,7 +650,10 @@ def test_raises_error_on_pipeline_op_error( pytest.param(1.5, id="Float input"), pytest.param(b"message", id="Bytes input"), pytest.param(bytearray(b"message"), id="Bytearray input"), + pytest.param(object(), id="Object input"), pytest.param(None, id="None input"), + pytest.param([1, "str"], id="List input"), + pytest.param({"a": 2}, id="Dictionary input"), ], ) def test_wraps_data_in_message_and_calls_pipeline_send_message( @@ -2288,7 +2291,10 @@ def test_raises_error_on_pipeline_op_error( pytest.param(1.5, id="Float input"), pytest.param(b"message", id="Bytes input"), pytest.param(bytearray(b"message"), id="Bytearray input"), + pytest.param(object(), id="Object input"), pytest.param(None, id="None input"), + pytest.param([1, "str"], id="List input"), + pytest.param({"a": 2}, id="Dictionary input"), ], ) def test_send_message_to_output_calls_pipeline_wraps_data_in_message( From eeb762646acfa7ba5606ac16bfe798f9dbc041a2 Mon Sep 17 00:00:00 2001 From: Carter Tinney Date: Mon, 31 Aug 2026 20:21:07 -0700 Subject: [PATCH 3/3] fix: reject unsupported message data directly Keep telemetry size checks as direct get_size comparisons and surface invalid caller-provided payload types as TypeError before starting a pipeline operation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../iot/device/iothub/aio/async_clients.py | 6 ++- .../azure/iot/device/iothub/models/message.py | 10 +---- .../azure/iot/device/iothub/sync_clients.py | 6 ++- tests/e2e/iothub_e2e/aio/test_send_message.py | 5 +-- .../iothub_e2e/sync/test_sync_send_message.py | 5 +-- tests/unit/iothub/aio/test_async_clients.py | 40 ++++++++++++++++--- tests/unit/iothub/test_sync_clients.py | 40 ++++++++++++++++--- 7 files changed, 81 insertions(+), 31 deletions(-) diff --git a/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py b/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py index ee6533967..f16fb89f1 100644 --- a/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py +++ b/azure-iot-device/azure/iot/device/iothub/aio/async_clients.py @@ -343,12 +343,13 @@ async def send_message(self, message: Union[Message, str]) -> None: connected (and there is no auto-connect enabled) :raises: :class:`azure.iot.device.exceptions.ClientError` if there is an unexpected failure during execution. + :raises: TypeError if the message data type is not supported by the MQTT transport. :raises: ValueError if the message fails size validation. """ if not isinstance(message, Message): message = Message(message) - if message._is_size_exceeded(device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT): + if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: raise ValueError("Size of telemetry message can not exceed 256 KB.") logger.info("Sending message to Hub...") @@ -643,6 +644,7 @@ async def send_message_to_output(self, message: Union[Message, str], output_name connected (and there is no auto-connect enabled) :raises: :class:`azure.iot.device.exceptions.ClientError` if there is an unexpected failure during execution. + :raises: TypeError if the message data type is not supported by the MQTT transport. :raises: ValueError if the message fails size validation. """ if not isinstance(message, Message): @@ -650,7 +652,7 @@ async def send_message_to_output(self, message: Union[Message, str], output_name message.output_name = output_name - if message._is_size_exceeded(device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT): + if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: raise ValueError("Size of message can not exceed 256 KB.") logger.info("Sending message to output:" + output_name + "...") diff --git a/azure-iot-device/azure/iot/device/iothub/models/message.py b/azure-iot-device/azure/iot/device/iothub/models/message.py index f77a51f45..a7033ed96 100644 --- a/azure-iot-device/azure/iot/device/iothub/models/message.py +++ b/azure-iot-device/azure/iot/device/iothub/models/message.py @@ -10,10 +10,6 @@ from azure.iot.device import constant -def _is_message_data_supported(data): - return data is None or isinstance(data, (str, bytes, bytearray, int, float)) - - def _encode_message_data(data): if isinstance(data, str): return data.encode("utf-8") @@ -21,7 +17,7 @@ def _encode_message_data(data): return str(data).encode("ascii") if data is None: return b"" - if not _is_message_data_supported(data): + if not isinstance(data, (bytes, bytearray)): raise TypeError("Message data must be a string, bytes, bytearray, int, float, or None.") return data @@ -124,10 +120,6 @@ def set_as_security_message(self): def __str__(self): return str(self.data) - def _is_size_exceeded(self, size_limit: int) -> bool: - # Unsupported payloads are deferred to the MQTT transport to preserve client error handling. - return _is_message_data_supported(self.data) and self.get_size() > size_limit - def get_size(self) -> int: """Return the message size in bytes as measured by IoT Hub. diff --git a/azure-iot-device/azure/iot/device/iothub/sync_clients.py b/azure-iot-device/azure/iot/device/iothub/sync_clients.py index 1fa8949db..a3d8c70f3 100644 --- a/azure-iot-device/azure/iot/device/iothub/sync_clients.py +++ b/azure-iot-device/azure/iot/device/iothub/sync_clients.py @@ -338,12 +338,13 @@ def send_message(self, message: Union[Message, str]) -> None: connected (and there is no auto-connect enabled) :raises: :class:`azure.iot.device.exceptions.ClientError` if there is an unexpected failure during execution. + :raises: TypeError if the message data type is not supported by the MQTT transport. :raises: ValueError if the message fails size validation. """ if not isinstance(message, Message): message = Message(message) - if message._is_size_exceeded(device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT): + if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: raise ValueError("Size of telemetry message can not exceed 256 KB.") logger.info("Sending message to Hub...") @@ -664,6 +665,7 @@ def send_message_to_output(self, message: Union[Message, str], output_name: str) connected (and there is no auto-connect enabled) :raises: :class:`azure.iot.device.exceptions.ClientError` if there is an unexpected failure during execution. + :raises: TypeError if the message data type is not supported by the MQTT transport. :raises: ValueError if the message fails size validation. """ if not isinstance(message, Message): @@ -671,7 +673,7 @@ def send_message_to_output(self, message: Union[Message, str], output_name: str) message.output_name = output_name - if message._is_size_exceeded(device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT): + if message.get_size() > device_constant.TELEMETRY_MESSAGE_SIZE_LIMIT: raise ValueError("Size of message can not exceed 256 KB.") logger.info("Sending message to output:" + output_name + "...") diff --git a/tests/e2e/iothub_e2e/aio/test_send_message.py b/tests/e2e/iothub_e2e/aio/test_send_message.py index bc609e9c9..4af584fd2 100644 --- a/tests/e2e/iothub_e2e/aio/test_send_message.py +++ b/tests/e2e/iothub_e2e/aio/test_send_message.py @@ -6,7 +6,7 @@ import logging import json import dev_utils -from azure.iot.device.exceptions import OperationCancelled, ClientError +from azure.iot.device.exceptions import OperationCancelled logger = logging.getLogger(__name__) logger.setLevel(level=logging.INFO) @@ -53,9 +53,8 @@ async def test_bad_payload_raises(self, client, leak_tracker): def thing_that_cant_serialize(): pass - with pytest.raises(ClientError) as e_info: + with pytest.raises(TypeError, match="Message data must be"): await client.send_message(thing_that_cant_serialize) - assert isinstance(e_info.value.__cause__, TypeError) # TODO: investigate leak # leak_tracker.check_for_leaks() diff --git a/tests/e2e/iothub_e2e/sync/test_sync_send_message.py b/tests/e2e/iothub_e2e/sync/test_sync_send_message.py index 78136019b..2fbd5b371 100644 --- a/tests/e2e/iothub_e2e/sync/test_sync_send_message.py +++ b/tests/e2e/iothub_e2e/sync/test_sync_send_message.py @@ -6,7 +6,7 @@ import json import time import dev_utils -from azure.iot.device.exceptions import OperationCancelled, ClientError +from azure.iot.device.exceptions import OperationCancelled logger = logging.getLogger(__name__) logger.setLevel(level=logging.INFO) @@ -50,9 +50,8 @@ def test_sync_bad_payload_raises(self, client, leak_tracker): def thing_that_cant_serialize(): pass - with pytest.raises(ClientError) as e_info: + with pytest.raises(TypeError, match="Message data must be"): client.send_message(thing_that_cant_serialize) - assert isinstance(e_info.value.__cause__, TypeError) # TODO; investigate this leak # leak_tracker.check_for_leaks() diff --git a/tests/unit/iothub/aio/test_async_clients.py b/tests/unit/iothub/aio/test_async_clients.py index 5d1946a6c..14898f84b 100644 --- a/tests/unit/iothub/aio/test_async_clients.py +++ b/tests/unit/iothub/aio/test_async_clients.py @@ -661,10 +661,7 @@ def fail_send_message(message, callback): pytest.param(1.5, id="Float input"), pytest.param(b"message", id="Bytes input"), pytest.param(bytearray(b"message"), id="Bytearray input"), - pytest.param(object(), id="Object input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) async def test_wraps_data_in_message_and_calls_pipeline_send_message( @@ -676,6 +673,23 @@ async def test_wraps_data_in_message_and_calls_pipeline_send_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input + @pytest.mark.it("Raises TypeError when the message data type is unsupported") + @pytest.mark.parametrize( + "message_input", + [ + pytest.param(object(), id="Object input"), + pytest.param([1, "str"], id="List input"), + pytest.param({"a": 2}, id="Dictionary input"), + ], + ) + async def test_raises_type_error_for_unsupported_message_data( + self, client, mqtt_pipeline, message_input + ): + with pytest.raises(TypeError, match="Message data must be"): + await client.send_message(message_input) + + assert mqtt_pipeline.send_message.call_count == 0 + @pytest.mark.it("Validates message size at the 256 KB boundary") @pytest.mark.parametrize( "size_delta, raises_error", @@ -2049,10 +2063,7 @@ def fail_send_output_message(message, callback): pytest.param(1.5, id="Float input"), pytest.param(b"message", id="Bytes input"), pytest.param(bytearray(b"message"), id="Bytearray input"), - pytest.param(object(), id="Object input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) async def test_send_message_to_output_calls_pipeline_wraps_data_in_message( @@ -2065,6 +2076,23 @@ async def test_send_message_to_output_calls_pipeline_wraps_data_in_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input + @pytest.mark.it("Raises TypeError when the message data type is unsupported") + @pytest.mark.parametrize( + "message_input", + [ + pytest.param(object(), id="Object input"), + pytest.param([1, "str"], id="List input"), + pytest.param({"a": 2}, id="Dictionary input"), + ], + ) + async def test_raises_type_error_for_unsupported_message_data( + self, client, mqtt_pipeline, message_input + ): + with pytest.raises(TypeError, match="Message data must be"): + await client.send_message_to_output(message_input, "some_output") + + assert mqtt_pipeline.send_output_message.call_count == 0 + @pytest.mark.it("Validates message size at the 256 KB boundary") @pytest.mark.parametrize( "size_delta, raises_error", diff --git a/tests/unit/iothub/test_sync_clients.py b/tests/unit/iothub/test_sync_clients.py index bfddb73b0..fc4f4abbd 100644 --- a/tests/unit/iothub/test_sync_clients.py +++ b/tests/unit/iothub/test_sync_clients.py @@ -650,10 +650,7 @@ def test_raises_error_on_pipeline_op_error( pytest.param(1.5, id="Float input"), pytest.param(b"message", id="Bytes input"), pytest.param(bytearray(b"message"), id="Bytearray input"), - pytest.param(object(), id="Object input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) def test_wraps_data_in_message_and_calls_pipeline_send_message( @@ -665,6 +662,23 @@ def test_wraps_data_in_message_and_calls_pipeline_send_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input + @pytest.mark.it("Raises TypeError when the message data type is unsupported") + @pytest.mark.parametrize( + "message_input", + [ + pytest.param(object(), id="Object input"), + pytest.param([1, "str"], id="List input"), + pytest.param({"a": 2}, id="Dictionary input"), + ], + ) + def test_raises_type_error_for_unsupported_message_data( + self, client, mqtt_pipeline, message_input + ): + with pytest.raises(TypeError, match="Message data must be"): + client.send_message(message_input) + + assert mqtt_pipeline.send_message.call_count == 0 + @pytest.mark.it("Validates message size at the 256 KB boundary") @pytest.mark.parametrize( "size_delta, raises_error", @@ -2291,10 +2305,7 @@ def test_raises_error_on_pipeline_op_error( pytest.param(1.5, id="Float input"), pytest.param(b"message", id="Bytes input"), pytest.param(bytearray(b"message"), id="Bytearray input"), - pytest.param(object(), id="Object input"), pytest.param(None, id="None input"), - pytest.param([1, "str"], id="List input"), - pytest.param({"a": 2}, id="Dictionary input"), ], ) def test_send_message_to_output_calls_pipeline_wraps_data_in_message( @@ -2307,6 +2318,23 @@ def test_send_message_to_output_calls_pipeline_wraps_data_in_message( assert isinstance(sent_message, Message) assert sent_message.data == message_input + @pytest.mark.it("Raises TypeError when the message data type is unsupported") + @pytest.mark.parametrize( + "message_input", + [ + pytest.param(object(), id="Object input"), + pytest.param([1, "str"], id="List input"), + pytest.param({"a": 2}, id="Dictionary input"), + ], + ) + def test_raises_type_error_for_unsupported_message_data( + self, client, mqtt_pipeline, message_input + ): + with pytest.raises(TypeError, match="Message data must be"): + client.send_message_to_output(message_input, "some_output") + + assert mqtt_pipeline.send_output_message.call_count == 0 + @pytest.mark.it("Validates message size at the 256 KB boundary") @pytest.mark.parametrize( "size_delta, raises_error",