commit 0d75e4da5586314d01a76d63085585df3df17054 parent 26e1a0cebaa83c3ba6ef48e4f08109ebc9006e9e Author: Stéphan Kochen <git@stephank.nl> Date: Wed, 02 Sep 2026 20:46:10 +0200 Validate and reassemble mesh stream responses
diff --git a/custom_components/inlite/lib/inlite_ble/hub.py b/custom_components/inlite/lib/inlite_ble/hub.py index 351ce04d47c1f669cff6ecd62297d225f5f09a53..ffd55c7e53e773f5fd1d678f7276540362afd610 100644 --- a/custom_components/inlite/lib/inlite_ble/hub.py +++ b/custom_components/inlite/lib/inlite_ble/hub.py @@ -102,6 +102,8 @@ self._ack_event = asyncio.Event() self._stream_event = asyncio.Event() self._last_ack_data = b"" self._last_stream_data = b"" + self._stream_buffer = bytearray() + self._stream_invalid = False self._zone_states: dict[int, ZoneState] = {} self._firmware_version: int | None = None self._discovered_device_ids: set[int] = set() @@ -229,8 +231,8 @@ if pkt_type == PKT_BLOCK_ACK: self._last_ack_data = payload loop.call_soon_threadsafe(self._ack_event.set) elif pkt_type == PKT_BLOCK_STREAM: - self._last_stream_data = payload - loop.call_soon_threadsafe(self._stream_event.set) + if self._accept_stream_segment(payload): + loop.call_soon_threadsafe(self._stream_event.set) elif pkt_type == PKT_BLOCK_DATA_BLK: if len(payload) >= 3 and payload[1:3] == b"\x0c\x00": self._discovered_device_ids.add(decrypted["src_id"]) @@ -241,6 +243,20 @@ if self._notification_callback: loop.call_soon_threadsafe(self._notification_callback, decrypted) + def _accept_stream_segment(self, payload: bytes) -> bool: + """Append one contiguous STREAM segment, rejecting malformed offsets.""" + if len(payload) < 2: + self._stream_invalid = True + return False + offset = payload[0] | (payload[1] << 8) + if offset != len(self._stream_buffer): + _LOGGER.warning("Discarding STREAM segment at offset %d (expected %d)", offset, len(self._stream_buffer)) + self._stream_invalid = True + return False + self._stream_buffer.extend(payload[2:]) + self._last_stream_data = bytes(self._stream_buffer) + return True + def _parse_oob_broadcast(self, payload: bytes) -> None: """Parse OOB_ALL_OUTLETS_MODE_UPDATE broadcast to update zone states.""" if len(payload) < 3: @@ -397,19 +413,23 @@ await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(0)) # Step 3: Hub sends STREAM(data) — collect and ACK self._stream_event.clear() + self._stream_buffer.clear() + self._stream_invalid = False try: await asyncio.wait_for(self._stream_event.wait(), STREAM_TIMEOUT) except asyncio.TimeoutError: _LOGGER.warning("STREAM timeout for command 0x%04X", opcode) return None + # Consecutive stream segments can already be in flight from the BLE + # bridge. Collect them briefly; offsets are validated on arrival. + await asyncio.sleep(WRITE_DELAY) stream_data = self._last_stream_data + if self._stream_invalid: + _LOGGER.warning("Discarding malformed STREAM response") + return None - # Parse byte count from stream offset header - if len(stream_data) >= 2: - stream_bytes = len(stream_data) - else: - stream_bytes = 0 + stream_bytes = len(stream_data) # ACK the stream data await asyncio.sleep(WRITE_DELAY) @@ -429,7 +449,8 @@ build_ack_payload(flush_count + 0xEF, end=False) ) _LOGGER.info("Acknowledged command 0x%04X: got %d bytes", opcode, len(stream_data)) - return bytes(stream_data) + # Restore the stream's offset prefix expected by the response parser. + return b"\x00\x00" + bytes(stream_data) async def query_zone_states(self) -> dict[int, ZoneState]: """Query all zone states from the hub using GET_INFO_DEVICES. @@ -456,7 +477,7 @@ offset = stream[0] | (stream[1] << 8) cmd_type = stream[2] opcode = stream[3] | (stream[4] << 8) - if opcode != OPCODE_GET_INFO_DEVICES: + if offset != 0 or cmd_type != 0x02 or opcode != OPCODE_GET_INFO_DEVICES: _LOGGER.warning("Unexpected opcode in STREAM: 0x%04X", opcode) return self._zone_states @@ -465,13 +486,15 @@ # product_id = stream[6] self._firmware_version = stream[7] # status = stream[8] num_zones = stream[9] + expected_length = 10 + num_zones * 7 + if len(stream) != expected_length: + _LOGGER.warning("Malformed GET_INFO_DEVICES response: %d bytes, expected %d", len(stream), expected_length) + return self._zone_states zones: dict[int, ZoneState] = {} zone_start = 10 for i in range(num_zones): pos = zone_start + i * 7 - if pos + 7 > len(stream): - break zs = ZoneState( output_id=stream[pos], output_type=stream[pos + 1], diff --git a/tests/test_hub.py b/tests/test_hub.py index 31c8553750e75a6c38ee48b5180f1da8c2781a19..3ed4befa6afa626422b52291f89551620ffd530a 100644 --- a/tests/test_hub.py +++ b/tests/test_hub.py @@ -52,3 +52,15 @@ """Verify the notification handler references call_soon_threadsafe.""" import inspect source = inspect.getsource(InliteHub._on_notification) assert "call_soon_threadsafe" in source + + def test_stream_segments_are_reassembled_by_offset(self) -> None: + hub = InliteHub(device_id=1, passphrase="test") + assert hub._accept_stream_segment(b"\x00\x00abc") + assert hub._accept_stream_segment(b"\x03\x00de") + assert hub._last_stream_data == b"abcde" + + def test_stream_segment_with_wrong_offset_is_rejected(self) -> None: + hub = InliteHub(device_id=1, passphrase="test") + assert hub._accept_stream_segment(b"\x00\x00abc") + assert not hub._accept_stream_segment(b"\x04\x00de") + assert hub._stream_invalid is True