diff --git a/esphome/components/packet_interface/__init__.py b/esphome/components/packet_interface/__init__.py index 60a7f281080..362deb51126 100644 --- a/esphome/components/packet_interface/__init__.py +++ b/esphome/components/packet_interface/__init__.py @@ -13,6 +13,7 @@ from . import layers from .layers import ( PacketInterface, build_stack, + check_session, packet_interface_ns, stack_auto_load, stack_schema, @@ -53,6 +54,18 @@ def AUTO_LOAD(config: list[ConfigType] | None = None) -> list[str]: return sorted(loads) +def _final_validate(config: list[ConfigType]) -> None: + for index, interface in enumerate(config): + try: + check_session(interface) + except cv.Invalid as err: + err.prepend([index]) + raise + + +FINAL_VALIDATE_SCHEMA = _final_validate + + def packet_interface_schema( class_: MockObjClass, default_max_packet_size: int, diff --git a/esphome/components/packet_interface/layers/__init__.py b/esphome/components/packet_interface/layers/__init__.py index fa9ab8792b6..b0a3667f313 100644 --- a/esphome/components/packet_interface/layers/__init__.py +++ b/esphome/components/packet_interface/layers/__init__.py @@ -30,6 +30,10 @@ CONF_SESSION = "session" LayerBuilder = Callable[[ConfigType, MockObj], Awaitable[Any]] +def _never(config: ConfigType) -> bool: + return False + + def _nothing(config: ConfigType) -> list[str]: return [] @@ -40,6 +44,8 @@ class LayerType: schema: Any build: LayerBuilder + # True if this configuration works only when the peer sends packets back, which a session layer makes sure of + needs_session: Callable[[ConfigType], bool] # The components this configuration needs loaded auto_load: Callable[[ConfigType], list[str]] @@ -63,6 +69,7 @@ class LayerKind: name: str, *, schema: Any, + needs_session: Callable[[ConfigType], bool] = _never, auto_load: Callable[[ConfigType], list[str]] = _nothing, ) -> Callable[[LayerBuilder], LayerBuilder]: """Register a type with its schema.""" @@ -77,7 +84,7 @@ class LayerKind: raise ValueError( f"{self.key} layer type '{name}' is already registered" ) - self.types[name] = LayerType(schema, build, auto_load) + self.types[name] = LayerType(schema, build, needs_session, auto_load) return build return decorator @@ -117,6 +124,21 @@ def stack_schema() -> dict: return {cv.Optional(kind.key): kind.schema() for kind in STACK_ORDER} +def check_session(config: ConfigType) -> None: + """Layers that need packets back from the peer need a session layer, whose keep-alives send them.""" + if CONF_SESSION in config: + return + for kind in STACK_ORDER: + if (layer := config.get(kind.key)) is not None and kind.get_type( + layer + ).needs_session(layer): + raise cv.Invalid( + f"This '{kind.key}' needs packets from the peer, so it needs a '{CONF_SESSION}' layer, whose " + "keep-alives make sure they are sent", + path=[kind.key], + ) + + 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/xxtea_encryption.py b/esphome/components/packet_interface/layers/xxtea_encryption.py index 7c69935c0ee..57ffc49baf7 100644 --- a/esphome/components/packet_interface/layers/xxtea_encryption.py +++ b/esphome/components/packet_interface/layers/xxtea_encryption.py @@ -23,7 +23,13 @@ XXTEA_SCHEMA = cv.Schema( ).extend(cv.COMPONENT_SCHEMA) -@ENCRYPTION.register("xxtea", schema=XXTEA_SCHEMA, auto_load=lambda config: ["xxtea"]) +@ENCRYPTION.register( + "xxtea", + schema=XXTEA_SCHEMA, + # The rolling code accepts a packet only once the peer has sent back this device's boot number + needs_session=lambda config: config[CONF_ROLLING_CODE_ENABLE], + auto_load=lambda config: ["xxtea"], +) async def build_xxtea(config: ConfigType, lower: MockObj) -> MockObj: cg.add_define("USE_PACKET_INTERFACE_XXTEA") var = cg.new_Pvariable(config[CONF_ID], lower, config[CONF_ROLLING_CODE_ENABLE]) diff --git a/tests/unit_tests/test_packet_interface.py b/tests/unit_tests/test_packet_interface.py index e22481d11c7..16a203c38a5 100644 --- a/tests/unit_tests/test_packet_interface.py +++ b/tests/unit_tests/test_packet_interface.py @@ -162,6 +162,21 @@ def test_encryption_requires_key(setup_core) -> None: uart_packet_interface.CONFIG_SCHEMA({CONF_ENCRYPTION: ""}) +def test_rolling_code_requires_session(setup_core) -> None: + """The rolling code needs the session's keep-alives, so the peer sends this device's boot number back.""" + rolling = {CONF_ENCRYPTION: {"key": "secret", "rolling_code_enable": True}} + with pytest.raises(cv.Invalid, match="session"): + packet_interface.FINAL_VALIDATE_SCHEMA( + [uart_packet_interface.CONFIG_SCHEMA(rolling)] + ) + packet_interface.FINAL_VALIDATE_SCHEMA( + [ + uart_packet_interface.CONFIG_SCHEMA({**rolling, CONF_SESSION: None}), + uart_packet_interface.CONFIG_SCHEMA({CONF_ENCRYPTION: "secret"}), + ] + ) + + @pytest.mark.parametrize( ("config", "loads"), [