fix: guard device get_basic_info parsers against short responses

switchbot/devices/air_purifier.py

@@ -97,6 +97,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
             return None
 
         _data, led_settings, led_status = res[0], res[1], res[2]
+        if len(_data) < 16 or len(led_settings) < 6 or len(led_status) < 2:
+            return None
 
         _LOGGER.debug(
             "%s %s basic info %s", self._model, self._device.address, _data.hex()

switchbot/devices/art_frame.py

@@ -25,6 +25,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         """Get device basic settings."""
         if not (_data := await self._get_basic_info()):
             return None
+        if len(_data) < 7:
+            return None
         _LOGGER.debug("basic info data: %s", _data.hex())
 
         battery_charging = bool(_data[1] & 0x80)
@@ -36,6 +38,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         last_network_status = (_data[4] >> 2) & 0x01
         current_image_index = _data[5]
         total_num_of_images = _data[6]
+        if len(_data) < 7 + total_num_of_images:
+            return None
         all_images_index = [_data[x] for x in range(7, 7 + total_num_of_images)]
 
         basic_info = {

switchbot/devices/base_cover.py

@@ -78,7 +78,7 @@ async def get_extended_info_adv(self) -> dict[str, Any] | None:
             _LOGGER.error("%s: Unsuccessful, no result from device", self.name)
             return None
 
-        if _data in (b"\x07", b"\x00"):
+        if len(_data) < 4:
             _LOGGER.error("%s: Unsuccessful, please try again", self.name)
             return None
 
@@ -94,15 +94,21 @@ async def get_extended_info_adv(self) -> dict[str, Any] | None:
         self.ext_info_adv["device0"] = {
             "battery": _data[1],
             "firmware": _data[2] / 10.0,
-            "stateOfCharge": _state_of_charge[_data[3]],
+            "stateOfCharge": (
+                _state_of_charge[_data[3]] if _data[3] < len(_state_of_charge) else None
+            ),
         }
 
         # If grouped curtain device present.
-        if _data[4]:
+        if len(_data) >= 7 and _data[4]:
             self.ext_info_adv["device1"] = {
                 "battery": _data[4],
                 "firmware": _data[5] / 10.0,
-                "stateOfCharge": _state_of_charge[_data[6]],
+                "stateOfCharge": (
+                    _state_of_charge[_data[6]]
+                    if _data[6] < len(_state_of_charge)
+                    else None
+                ),
             }
 
         return self.ext_info_adv

switchbot/devices/blind_tilt.py

@@ -111,6 +111,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         """Get device basic settings."""
         if not (_data := await self._get_basic_info()):
             return None
+        if len(_data) < 8:
+            return None
 
         _tilt = max(min(_data[6], 100), 0)
         _moving = bool(_data[5] & 0b00000011)
@@ -150,7 +152,7 @@ async def get_extended_info_summary(self) -> dict[str, Any] | None:
             _LOGGER.error("%s: Unsuccessful, no result from device", self.name)
             return None
 
-        if _data in (b"\x07", b"\x00"):
+        if len(_data) < 2:
             _LOGGER.error("%s: Unsuccessful, please try again", self.name)
             return None
 

switchbot/devices/bot.py

@@ -101,6 +101,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         """Get device basic settings."""
         if not (_data := await self._get_basic_info()):
             return None
+        if len(_data) < 11:
+            return None
         return {
             "battery": _data[1],
             "firmware": _data[2] / 10.0,

switchbot/devices/bulb.py

@@ -48,6 +48,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         ):
             return None
         _version_info, _data = res
+        if len(_data) < 11 or len(_version_info) < 3:
+            return None
 
         self._state["r"] = _data[3]
         self._state["g"] = _data[4]

switchbot/devices/ceiling_light.py

@@ -59,6 +59,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         ):
             return None
         _version_info, _data = res
+        if len(_data) < 5 or len(_version_info) < 3:
+            return None
 
         self._state["cw"] = int.from_bytes(_data[3:5], "big")
 

switchbot/devices/curtain.py

@@ -104,6 +104,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         """Get device basic settings."""
         if not (_data := await self._get_basic_info()):
             return None
+        if len(_data) < 8:
+            return None
 
         _position = max(min(_data[6], 100), 0)
         _direction_adjusted_position = (100 - _position) if self._reverse else _position
@@ -153,7 +155,7 @@ async def get_extended_info_summary(self) -> dict[str, Any] | None:
             _LOGGER.error("%s: Unsuccessful, no result from device", self.name)
             return None
 
-        if _data in (b"\x07", b"\x00"):
+        if len(_data) < 3:
             _LOGGER.error("%s: Unsuccessful, please try again", self.name)
             return None
 

switchbot/devices/evaporative_humidifier.py

@@ -54,6 +54,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         """Get device basic settings."""
         if not (_data := await self._get_basic_info(DEVICE_GET_BASIC_SETTINGS_KEY)):
             return None
+        if len(_data) < 11:
+            return None
 
         _LOGGER.debug("basic info data: %s", _data.hex())
         isOn = bool(_data[1] & 0b10000000)

switchbot/devices/fan.py

@@ -72,6 +72,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
             return None
         if not (_data1 := await self._get_basic_info(DEVICE_GET_BASIC_SETTINGS_KEY)):
             return None
+        if len(_data) < 10 or len(_data1) < 3:
+            return None
 
         _LOGGER.debug("data: %s", _data)
         return self._parse_basic_info(_data, _data1)
@@ -111,7 +113,7 @@ async def _get_basic_info(self, cmd: str) -> bytes | None:
         """Return basic info of device."""
         _data = await self._send_command(key=cmd, retry=self._retry_count)
 
-        if _data in (b"\x07", b"\x00"):
+        if _data is None or len(_data) <= 1:
             _LOGGER.error("Unsuccessful, please try again")
             return None
 

switchbot/devices/keypad_vision.py

@@ -46,6 +46,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         """Get device basic settings."""
         if not (_data := await self._get_basic_info()):
             return None
+        if len(_data) < 15:
+            return None
         _LOGGER.debug("Raw model %s basic info data: %s", self._model, _data.hex())
 
         battery = _data[1] & 0x7F
@@ -137,6 +139,9 @@ async def get_password_count(self) -> dict[str, int] | None:
         """Get the number of passwords stored in the Keypad Vision (Pro)."""
         if not (_data := await self._send_command(COMMAND_GET_PASSWORD_COUNT)):
             return None
+        min_len = 8 if self._model == SwitchbotModel.KEYPAD_VISION_PRO else 6
+        if len(_data) < min_len:
+            return None
         _LOGGER.debug("Raw model %s password count data: %s", self._model, _data.hex())
 
         pin = _data[1]

switchbot/devices/light_strip.py

@@ -268,6 +268,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
             return None
 
         _version_info, _data = res
+        if len(_data) < 11 or len(_version_info) < 3:
+            return None
         self._state["r"] = _data[3]
         self._state["g"] = _data[4]
         self._state["b"] = _data[5]
@@ -321,6 +323,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         ):
             return None
         _version_info, _data = res
+        if len(_data) < 3 or len(_version_info) < 3:
+            return None
         return {
             "isOn": bool(_data[1] & 0b10000000),
             "brightness": _data[2] & 0b01111111,

switchbot/devices/roller_shade.py

@@ -127,6 +127,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         """Get device basic settings."""
         if not (_data := await self._get_basic_info()):
             return None
+        if len(_data) < 7:
+            return None
 
         _position = max(min(_data[5], 100), 0)
         _direction_adjusted_position = (100 - _position) if self._reverse else _position

switchbot/devices/smart_thermostat_radiator.py

@@ -133,6 +133,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         """Get device basic settings."""
         if not (_data := await self._get_basic_info()):
             return None
+        if len(_data) < 15:
+            return None
         _LOGGER.debug("data: %s", _data)
 
         battery = _data[1]

switchbot/devices/vacuum.py

@@ -33,6 +33,8 @@ async def get_basic_info(self) -> dict[str, Any] | None:
         """Only support get the ble version through the command."""
         if not (_data := await self._get_basic_info()):
             return None
+        if len(_data) < 3:
+            return None
         return {
             "firmware": _data[2],
         }

tests/test_art_frame.py

@@ -133,6 +133,7 @@ async def test_next_image(
     current_index: int, all_images_index: list[int], expected_cmd: str
 ) -> None:
     device = create_device_for_command_testing(ART_FRAME_INFO)
+    device._get_current_image_index = AsyncMock()
 
     with (
         patch.object(device, "get_current_image_index", return_value=current_index),
@@ -157,6 +158,7 @@ async def test_prev_image(
     current_index: int, all_images_index: list[int], expected_cmd: str
 ) -> None:
     device = create_device_for_command_testing(ART_FRAME_INFO)
+    device._get_current_image_index = AsyncMock()
 
     with (
         patch.object(device, "get_current_image_index", return_value=current_index),
@@ -171,6 +173,7 @@ async def test_prev_image(
 @pytest.mark.asyncio
 async def test_set_image_with_invalid_index() -> None:
     device = create_device_for_command_testing(ART_FRAME_INFO)
+    device._get_current_image_index = AsyncMock()
 
     with (
         patch.object(device, "get_total_images", return_value=3),
@@ -185,6 +188,7 @@ async def test_set_image_with_invalid_index() -> None:
 @pytest.mark.asyncio
 async def test_set_image_with_valid_index() -> None:
     device = create_device_for_command_testing(ART_FRAME_INFO)
+    device._get_current_image_index = AsyncMock()
 
     with (
         patch.object(device, "get_total_images", return_value=3),