From 34c6c83fd1361bfd699a755346b59434eef95f28 Mon Sep 17 00:00:00 2001 From: clydebarrow <2366188+clydebarrow@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:40:48 +1100 Subject: [PATCH] [packet_interface] Check at validation that fragments fit the reliability queue - Work out each layer's packet size during validation, as the C++ does, and reject a fragmentation max_packet_size whose fragments do not all fit in the reliability queue at once, or need more than 128 fragments - Fit the test peer's largest packet in its queue - Reuse interface_layers() in final_validate_interface() - Keep the earlier tests in place in test_packet_interface.py Co-Authored-By: Claude Opus 5.5 --- .../components/packet_interface/__init__.py | 25 ++- .../packet_interface/layers/__init__.py | 44 +++++ .../layers/basic_addressing.py | 10 +- .../layers/basic_fragmentation.py | 28 ++- .../layers/hdlc_reliability.py | 13 +- .../layers/xxtea_encryption.py | 11 +- tests/components/packet_interface/common.yaml | 3 +- tests/unit_tests/test_packet_interface.py | 181 ++++++++++++------ 8 files changed, 240 insertions(+), 75 deletions(-) diff --git a/esphome/components/packet_interface/__init__.py b/esphome/components/packet_interface/__init__.py index 4cd2f031a25..e1894f34144 100644 --- a/esphome/components/packet_interface/__init__.py +++ b/esphome/components/packet_interface/__init__.py @@ -22,6 +22,7 @@ from .layers import ( STACK_ORDER, PacketInterface, build_stack, + check_packet_sizes, check_session, check_single_peer, packet_interface_ns, @@ -74,8 +75,11 @@ def AUTO_LOAD(config: list[ConfigType] | None = None) -> list[str]: def _validate_layers(config: ConfigType) -> ConfigType: """With peers, the layers go in each peer; otherwise check them against the one destination.""" check_session(config) + size = config[CONF_MAX_PACKET_SIZE] if (addressing := config.get(CONF_ADDRESSING)) is None: + check_packet_sizes(config, size) return config + size = ADDRESSING.get_type(addressing).packet_size(addressing, size) if CONF_PEERS in addressing: for kind in STACK_ORDER: if kind.key in config: @@ -83,9 +87,16 @@ def _validate_layers(config: ConfigType) -> ConfigType: f"With '{CONF_PEERS}', '{kind.key}' goes in each peer", path=[kind.key], ) + for index, peer in enumerate(addressing[CONF_PEERS]): + try: + check_packet_sizes(peer, size) + except cv.Invalid as err: + err.prepend([CONF_ADDRESSING, CONF_PEERS, index]) + raise return config addressing.setdefault(CONF_DESTINATION, BROADCAST_ADDRESS) check_single_peer(config, addressing[CONF_DESTINATION]) + check_packet_sizes(config, size) return config @@ -108,14 +119,12 @@ def final_validate_interface(key: str) -> Callable[[ConfigType], ConfigType]: def validator(config: ConfigType) -> ConfigType: interface_id: ID = config[key] - for interface in fv.full_config.get().get(DOMAIN, []): - if interface[CONF_ID].id == interface_id.id and CONF_PEERS in interface.get( - CONF_ADDRESSING, {} - ): - raise cv.Invalid( - f"'{interface_id.id}' has several peers, so use the id of one of them", - path=[key], - ) + interface = interface_layers(interface_id) + if interface is not None and CONF_PEERS in interface.get(CONF_ADDRESSING, {}): + raise cv.Invalid( + f"'{interface_id.id}' has several peers, so use the id of one of them", + path=[key], + ) return config return validator diff --git a/esphome/components/packet_interface/layers/__init__.py b/esphome/components/packet_interface/layers/__init__.py index 51bd5d93963..ab88010a509 100644 --- a/esphome/components/packet_interface/layers/__init__.py +++ b/esphome/components/packet_interface/layers/__init__.py @@ -44,6 +44,18 @@ def _nothing(config: ConfigType) -> list[str]: return [] +def _same_size(config: ConfigType, size: int) -> int: + return size + + +def _no_limit(config: ConfigType) -> int | None: + return None + + +def _fits(config: ConfigType, size: int, capacity: int | None) -> None: + """Accept any layer below.""" + + @dataclass(frozen=True) class LayerType: """One implementation of a kind of layer.""" @@ -57,6 +69,12 @@ class LayerType: needs_session: Callable[[ConfigType], bool] # The components this configuration needs loaded auto_load: Callable[[ConfigType], list[str]] + # The largest packet this configuration takes, from the largest that the layer below takes, as the C++ works it out + packet_size: Callable[[ConfigType, int], int] + # How many packets this configuration takes at once, or None for no limit + send_capacity: Callable[[ConfigType], int | None] + # Raises cv.Invalid if this configuration cannot work above a layer with this packet size and send capacity + check_lower: Callable[[ConfigType, int, int | None], None] class LayerKind: @@ -82,6 +100,9 @@ class LayerKind: single_peer: Callable[[ConfigType], bool] = _never, needs_session: Callable[[ConfigType], bool] = _never, auto_load: Callable[[ConfigType], list[str]] = _nothing, + packet_size: Callable[[ConfigType, int], int] = _same_size, + send_capacity: Callable[[ConfigType], int | None] = _no_limit, + check_lower: Callable[[ConfigType, int, int | None], None] = _fits, ) -> Callable[[LayerBuilder], LayerBuilder]: """Register a type with its schema, or with a function returning a schema that contains other layers.""" @@ -104,6 +125,9 @@ class LayerKind: single_peer, needs_session, auto_load, + packet_size, + send_capacity, + check_lower, ) return build @@ -178,6 +202,26 @@ def check_single_peer(config: ConfigType, destination: int) -> None: ) +def remaining_packet_size(size: int, used: int) -> int: + return max(size - used, 0) + + +def check_packet_sizes(config: ConfigType, size: int) -> None: + """Check each layer in config against the layer below it; size is the largest packet that the stack sits on takes.""" + capacity = None + for kind in STACK_ORDER: + if (layer := config.get(kind.key)) is None: + continue + layer_type = kind.get_type(layer) + try: + layer_type.check_lower(layer, size, capacity) + except cv.Invalid as err: + err.prepend([kind.key]) + raise + size = layer_type.packet_size(layer, size) + capacity = layer_type.send_capacity(layer) + + def stack_auto_load(config: ConfigType) -> set[str]: """The components that the layers configured in config need loaded.""" loads: set[str] = set() diff --git a/esphome/components/packet_interface/layers/basic_addressing.py b/esphome/components/packet_interface/layers/basic_addressing.py index 76dc4e150d4..d810dfb375a 100644 --- a/esphome/components/packet_interface/layers/basic_addressing.py +++ b/esphome/components/packet_interface/layers/basic_addressing.py @@ -18,11 +18,15 @@ from . import ( check_session, check_single_peer, packet_interface_ns, + remaining_packet_size, stack_schema, ) CONF_PORT_ID = "port_id" +# The destination and source addresses +HEADER_SIZE = 4 + BasicAddressing = packet_interface_ns.class_("BasicAddressing", cg.Component) AddressingPort = packet_interface_ns.class_("AddressingPort", PacketInterface) @@ -103,7 +107,11 @@ async def _new_port( return port -@ADDRESSING.register("basic", schema_factory=_basic_addressing_schema) +@ADDRESSING.register( + "basic", + schema_factory=_basic_addressing_schema, + packet_size=lambda config, size: remaining_packet_size(size, HEADER_SIZE), +) async def build_basic_addressing( config: ConfigType, link: MockObj ) -> list[tuple[MockObj, ConfigType | None]]: diff --git a/esphome/components/packet_interface/layers/basic_fragmentation.py b/esphome/components/packet_interface/layers/basic_fragmentation.py index 7440ae38765..c4d1f5271bc 100644 --- a/esphome/components/packet_interface/layers/basic_fragmentation.py +++ b/esphome/components/packet_interface/layers/basic_fragmentation.py @@ -1,5 +1,7 @@ """Fragmentation layer that splits packets larger than the layers below allow, and joins them again.""" +import math + import esphome.codegen as cg from esphome.components.const import CONF_MAX_PACKET_SIZE import esphome.config_validation as cv @@ -9,6 +11,10 @@ from esphome.types import ConfigType from . import FRAGMENTATION, PacketInterface, packet_interface_ns +# The packet number, and the fragment's index with the flag on the last fragment +HEADER_SIZE = 2 +MAX_FRAGMENTS = 128 + BasicFragmentation = packet_interface_ns.class_("BasicFragmentation", PacketInterface) BASIC_FRAGMENTATION_SCHEMA = cv.Schema( @@ -20,9 +26,29 @@ BASIC_FRAGMENTATION_SCHEMA = cv.Schema( ).extend(cv.COMPONENT_SCHEMA) +def _check_lower(config: ConfigType, size: int, capacity: int | None) -> None: + """The largest packet must fit in the fragments that the layer below can take at once.""" + fragment_size = size - HEADER_SIZE + if fragment_size <= 0: + raise cv.Invalid("The layers below leave no room for fragments") + fragments = max(1, math.ceil(config[CONF_MAX_PACKET_SIZE] / fragment_size)) + limit = MAX_FRAGMENTS if capacity is None else min(MAX_FRAGMENTS, capacity) + if fragments > limit: + raise cv.Invalid( + f"A packet of {config[CONF_MAX_PACKET_SIZE]} bytes needs {fragments} fragments of {fragment_size} bytes, " + f"but at most {limit} can be sent at once", + path=[CONF_MAX_PACKET_SIZE], + ) + + # Fragments from different devices are told apart by the addressing layer's source address, so of two devices sending # fragmented packets at once, one still gets through; without addressing, the receiver can drop both -@FRAGMENTATION.register("basic", schema=BASIC_FRAGMENTATION_SCHEMA) +@FRAGMENTATION.register( + "basic", + schema=BASIC_FRAGMENTATION_SCHEMA, + packet_size=lambda config, size: config[CONF_MAX_PACKET_SIZE], + check_lower=_check_lower, +) async def build_basic_fragmentation(config: ConfigType, lower: MockObj) -> MockObj: cg.add_define("USE_PACKET_INTERFACE_BASIC_FRAGMENTATION") var = cg.new_Pvariable(config[CONF_ID], lower, config[CONF_MAX_PACKET_SIZE]) diff --git a/esphome/components/packet_interface/layers/hdlc_reliability.py b/esphome/components/packet_interface/layers/hdlc_reliability.py index 9f7c31179d1..054aaddf49f 100644 --- a/esphome/components/packet_interface/layers/hdlc_reliability.py +++ b/esphome/components/packet_interface/layers/hdlc_reliability.py @@ -6,12 +6,15 @@ from esphome.const import CONF_ID, CONF_TIMEOUT from esphome.cpp_generator import MockObj from esphome.types import ConfigType -from . import RELIABILITY, PacketInterface, packet_interface_ns +from . import RELIABILITY, PacketInterface, packet_interface_ns, remaining_packet_size CONF_ACK_DELAY = "ack_delay" CONF_MAX_QUEUE_LENGTH = "max_queue_length" CONF_MAX_RETRIES = "max_retries" +# The control byte and the CRC-16 +FRAME_OVERHEAD = 3 + HdlcReliability = packet_interface_ns.class_("HdlcReliability", PacketInterface) @@ -47,7 +50,13 @@ HDLC_SCHEMA = cv.All( ) -@RELIABILITY.register("hdlc", schema=HDLC_SCHEMA, single_peer=lambda config: True) +@RELIABILITY.register( + "hdlc", + schema=HDLC_SCHEMA, + single_peer=lambda config: True, + packet_size=lambda config, size: remaining_packet_size(size, FRAME_OVERHEAD), + send_capacity=lambda config: config[CONF_MAX_QUEUE_LENGTH], +) async def build_hdlc(config: ConfigType, lower: MockObj) -> MockObj: cg.add_define("USE_PACKET_INTERFACE_HDLC") var = cg.new_Pvariable(config[CONF_ID], lower) diff --git a/esphome/components/packet_interface/layers/xxtea_encryption.py b/esphome/components/packet_interface/layers/xxtea_encryption.py index 73ccb89967b..8c54847e331 100644 --- a/esphome/components/packet_interface/layers/xxtea_encryption.py +++ b/esphome/components/packet_interface/layers/xxtea_encryption.py @@ -8,10 +8,14 @@ from esphome.const import CONF_ID, CONF_KEY from esphome.cpp_generator import MockObj from esphome.types import ConfigType -from . import ENCRYPTION, PacketInterface, packet_interface_ns +from . import ENCRYPTION, PacketInterface, packet_interface_ns, remaining_packet_size CONF_ROLLING_CODE_ENABLE = "rolling_code_enable" +# The check word, and with the rolling code the boot numbers and counter +CHECK_SIZE = 4 +ROLLING_SIZE = 12 + XxteaEncryption = packet_interface_ns.class_("XxteaEncryption", PacketInterface) XXTEA_SCHEMA = cv.Schema( @@ -31,6 +35,11 @@ XXTEA_SCHEMA = cv.Schema( single_peer=lambda config: config[CONF_ROLLING_CODE_ENABLE], needs_session=lambda config: config[CONF_ROLLING_CODE_ENABLE], auto_load=lambda config: ["xxtea"], + # XXTEA works on whole 32-bit words + packet_size=lambda config, size: remaining_packet_size( + size & ~3, + CHECK_SIZE + (ROLLING_SIZE if config[CONF_ROLLING_CODE_ENABLE] else 0), + ), ) async def build_xxtea(config: ConfigType, lower: MockObj) -> MockObj: cg.add_define("USE_PACKET_INTERFACE_XXTEA") diff --git a/tests/components/packet_interface/common.yaml b/tests/components/packet_interface/common.yaml index 12d43bc6d1d..5f34a6d437a 100644 --- a/tests/components/packet_interface/common.yaml +++ b/tests/components/packet_interface/common.yaml @@ -29,8 +29,9 @@ packet_interface: - id: packet_interface_peer_b destination: 0x0003 reliability: + max_queue_length: 5 fragmentation: - max_packet_size: 2048 + max_packet_size: 1024 session: - id: packet_interface_bus destination: broadcast diff --git a/tests/unit_tests/test_packet_interface.py b/tests/unit_tests/test_packet_interface.py index 895bceab29b..3209346c89c 100644 --- a/tests/unit_tests/test_packet_interface.py +++ b/tests/unit_tests/test_packet_interface.py @@ -1,4 +1,4 @@ -"""Unit tests for the packet_interface component and its platforms.""" +"""Unit tests for the packet_interface component.""" from typing import Any @@ -72,6 +72,62 @@ def test_reliability_queue_length_limit(setup_core) -> None: ) +def test_reliability_timeout_minimum(setup_core) -> None: + """A timeout shorter than a few loop passes cannot work.""" + uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "50ms"}}) + with pytest.raises(cv.Invalid): + uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "40ms"}}) + + +def test_ack_delay_shorter_than_timeout(setup_core) -> None: + """The acknowledgement must not wait longer than the peer's timeout.""" + reliability = uart_packet_interface.CONFIG_SCHEMA( + {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "100ms"}} + )[CONF_RELIABILITY] + assert reliability["ack_delay"].total_milliseconds == 100 + with pytest.raises(cv.Invalid, match="shorter"): + uart_packet_interface.CONFIG_SCHEMA( + {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "500ms"}} + ) + + +def test_layer_types_registered() -> None: + """Importing packet_interface registers the built-in layer types.""" + assert {kind.key: set(kind.types) for kind in STACK_ORDER} == { + CONF_ENCRYPTION: {"xxtea"}, + CONF_RELIABILITY: {"hdlc"}, + CONF_FRAGMENTATION: {"basic"}, + CONF_SESSION: {"basic"}, + } + + +async def _build(config: dict, lower: Any) -> Any: + return lower + + +def test_registered_type_selectable(setup_core) -> None: + """A newly registered layer type can be selected with type.""" + kind = LayerKind("test_kind", "test") + kind.register("test", schema=cv.Schema({}))(_build) + config = kind.schema()({"type": "test"}) + assert config["type"] == "test" + assert kind.get_type(config).build is _build + + +def test_duplicate_layer_type_rejected(setup_core) -> None: + """Registering a type under a name already in use is an error, not a silent replacement.""" + kind = LayerKind("test_kind", "test") + kind.register("test", schema=cv.Schema({}))(_build) + with pytest.raises(ValueError, match="already registered"): + kind.register("test", schema=cv.Schema({}))(_build) + + +def test_late_registration_rejected(setup_core) -> None: + """A type registered after the schemas were built would be missing from them, so it is an error.""" + with pytest.raises(ValueError, match="after the schemas were built"): + RELIABILITY.register("late", schema=cv.Schema({}))(_build) + + def test_session_defaults(setup_core) -> None: """An empty session key selects the basic session, timing out after three keep-alives.""" session = uart_packet_interface.CONFIG_SCHEMA({CONF_SESSION: None})[CONF_SESSION] @@ -154,7 +210,7 @@ def test_two_device_layers_need_destination(setup_core, layers: dict) -> None: ) -def _peer(peer_id: str, destination: Any, **layers: Any) -> dict: +def _peer(peer_id: str | ID, destination: Any, **layers: Any) -> dict: return {"id": peer_id, "destination": destination, **layers} @@ -196,62 +252,6 @@ def test_addressing_peers_invalid( ) -def test_layer_types_registered() -> None: - """Importing packet_interface registers the built-in layer types.""" - assert {kind.key: set(kind.types) for kind in STACK_ORDER} == { - CONF_ENCRYPTION: {"xxtea"}, - CONF_RELIABILITY: {"hdlc"}, - CONF_FRAGMENTATION: {"basic"}, - CONF_SESSION: {"basic"}, - } - - -async def _build(config: dict, lower: Any) -> Any: - return lower - - -def test_registered_type_selectable(setup_core) -> None: - """A newly registered layer type can be selected with type.""" - kind = LayerKind("test_kind", "test") - kind.register("test", schema=cv.Schema({}))(_build) - config = kind.schema()({"type": "test"}) - assert config["type"] == "test" - assert kind.get_type(config).build is _build - - -def test_duplicate_layer_type_rejected(setup_core) -> None: - """Registering a type under a name already in use is an error, not a silent replacement.""" - kind = LayerKind("test_kind", "test") - kind.register("test", schema=cv.Schema({}))(_build) - with pytest.raises(ValueError, match="already registered"): - kind.register("test", schema=cv.Schema({}))(_build) - - -def test_late_registration_rejected(setup_core) -> None: - """A type registered after the schemas were built would be missing from them, so it is an error.""" - with pytest.raises(ValueError, match="after the schemas were built"): - RELIABILITY.register("late", schema=cv.Schema({}))(_build) - - -def test_reliability_timeout_minimum(setup_core) -> None: - """A timeout shorter than a few loop passes cannot work.""" - uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "50ms"}}) - with pytest.raises(cv.Invalid): - uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "40ms"}}) - - -def test_ack_delay_shorter_than_timeout(setup_core) -> None: - """The acknowledgement must not wait longer than the peer's timeout.""" - reliability = uart_packet_interface.CONFIG_SCHEMA( - {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "100ms"}} - )[CONF_RELIABILITY] - assert reliability["ack_delay"].total_milliseconds == 100 - with pytest.raises(cv.Invalid, match="shorter"): - uart_packet_interface.CONFIG_SCHEMA( - {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "500ms"}} - ) - - def test_fragmentation_defaults(setup_core) -> None: """An empty fragmentation key joins packets of up to 1024 bytes.""" fragmentation = uart_packet_interface.CONFIG_SCHEMA({CONF_FRAGMENTATION: None})[ @@ -265,12 +265,71 @@ def test_fragmentation_defaults(setup_core) -> None: ) +@pytest.mark.parametrize( + ("layers", "valid"), + [ + # 508 - 3 leaves fragments of 503 bytes: 1024 bytes take 3, 2048 bytes take 5 + ({CONF_RELIABILITY: None, CONF_FRAGMENTATION: None}, True), + ( + {CONF_RELIABILITY: None, CONF_FRAGMENTATION: {"max_packet_size": 2048}}, + False, + ), + ( + { + CONF_RELIABILITY: {CONF_MAX_QUEUE_LENGTH: 5}, + CONF_FRAGMENTATION: {"max_packet_size": 2048}, + }, + True, + ), + ({CONF_FRAGMENTATION: {"max_packet_size": 16384}}, True), + ], +) +def test_fragments_fit_reliability_queue(setup_core, layers: dict, valid: bool) -> None: + """Every fragment of the largest packet must fit in the reliability queue at once.""" + if valid: + uart_packet_interface.CONFIG_SCHEMA(layers) + return + with pytest.raises(cv.Invalid, match="5 fragments") as err: + uart_packet_interface.CONFIG_SCHEMA(layers) + assert err.value.path == [CONF_FRAGMENTATION, CONF_MAX_PACKET_SIZE] + + +@pytest.mark.parametrize( + ("link", "match"), + [ + # 256 - 4 for addressing - 16 for the rolling code - 3 for reliability leaves fragments of 231 bytes + ({}, "5 fragments of 231 bytes"), + # 24 - 4 - 16 - 3 leaves 1 byte, too few for the fragment header + ({CONF_MAX_PACKET_SIZE: 24}, "no room"), + ], +) +def test_fragments_checked_in_peers(setup_core, link: dict, match: str) -> None: + """Fragments are checked against the layers of each peer, below the addressing layer.""" + peer = _peer( + "a", + 2, + encryption={"key": "secret", "rolling_code_enable": True}, + reliability=None, + fragmentation=None, + session=None, + ) + with pytest.raises(cv.Invalid, match=match) as err: + uart_packet_interface.CONFIG_SCHEMA( + { + CONF_MAX_PACKET_SIZE: 256, + **link, + CONF_ADDRESSING: {"address": 1, "peers": [peer]}, + } + ) + assert err.value.path[:4] == [CONF_ADDRESSING, "peers", 0, CONF_FRAGMENTATION] + + @pytest.mark.parametrize( ("config", "loads"), [ ({}, []), - ({CONF_RELIABILITY: None, "session": None}, []), - ({"session": {"binary_sensor": {"name": "Link"}}}, ["binary_sensor"]), + ({CONF_RELIABILITY: None, CONF_SESSION: None}, []), + ({CONF_SESSION: {"binary_sensor": {"name": "Link"}}}, ["binary_sensor"]), ({CONF_ENCRYPTION: "secret"}, ["xxtea"]), ( { @@ -315,7 +374,7 @@ def test_application_needs_peer_id_with_several_peers(setup_core) -> None: token = fv.full_config.set( { "packet_interface": [ - {"id": ID("bus"), CONF_ADDRESSING: {"peers": [_peer("a", 2)]}}, + {"id": ID("bus"), CONF_ADDRESSING: {"peers": [_peer(ID("a"), 2)]}}, {"id": ID("single"), CONF_ADDRESSING: {"destination": 2}}, ] }