Fix Victron BLE false reauth on unrecognised advertisement mode bytes

homeassistant/components/victron_ble/__init__.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 import logging
+from struct import error as struct_error
 
 from sensor_state_data import SensorUpdate
 from victron_ble_ha_parser import VictronBluetoothDeviceData
+from victron_ble_ha_parser.parser import detect_device_type
 
 from homeassistant.components.bluetooth import (
     BluetoothScanningMode,
@@ -38,25 +40,33 @@ def _update(
         nonlocal consecutive_failures
         update = data.update(service_info)
 
-        # Only consider a reauth when the device type is recognised (devices
-        # populated) but the advertisement key fails the quick-check built into
-        # validate_advertisement_key.  Using the key check instead of counting
-        # entity values avoids false positives: some devices legitimately return
-        # few (or zero) sensor values when in certain error or alarm states.
+        # Only assess key validity for instant-readout advertisements
+        # (0x10 prefix) whose device type the parser actually recognizes.
+        # Unrecognized mode bytes or non-instant-readout packets are neutral:
+        # they say nothing about whether the encryption key is correct, so
+        # they must not increment or reset the failure counter.
         raw_data = service_info.manufacturer_data.get(VICTRON_IDENTIFIER)
         if update.devices and raw_data is not None:
-            if not data.validate_advertisement_key(raw_data):
-                consecutive_failures += 1
-                if consecutive_failures >= REAUTH_AFTER_FAILURES:
-                    _LOGGER.debug(
-                        "Triggering reauth for %s after %d consecutive failures",
-                        address,
-                        consecutive_failures,
-                    )
-                    entry.async_start_reauth(hass)
+            try:
+                is_recognizable = (
+                    raw_data[:1] == b"\x10" and detect_device_type(raw_data) is not None
+                )
+            except struct_error, IndexError:
+                is_recognizable = False
+
+            if is_recognizable:
+                if not data.validate_advertisement_key(raw_data):
+                    consecutive_failures += 1
+                    if consecutive_failures >= REAUTH_AFTER_FAILURES:
+                        _LOGGER.debug(
+                            "Triggering reauth for %s after %d consecutive failures",
+                            address,
+                            consecutive_failures,
+                        )
+                        entry.async_start_reauth(hass)
+                        consecutive_failures = 0
+                else:
                     consecutive_failures = 0
-            else:
-                consecutive_failures = 0
         else:
             consecutive_failures = 0
 

tests/components/victron_ble/fixtures.py

@@ -201,6 +201,22 @@
     source="local",
 )
 
+# Same Victron manufacturer data prefix but with an unrecognized mode byte
+# (0xEE at offset 4).  detect_device_type returns None for this payload,
+# so validate_advertisement_key would also return False.  The reauth logic
+# must treat this as neutral (not a key failure).
+VICTRON_VEBUS_UNRECOGNIZED_MODE_SERVICE_INFO = BluetoothServiceInfo(
+    name="Inverter Charger",
+    address="01:02:03:04:05:06",
+    rssi=-60,
+    manufacturer_data={
+        0x02E1: bytes.fromhex("10038027ee1252dad26f0b8eb39162074d140df410")
+    },
+    service_data={},
+    service_uuids=[],
+    source="local",
+)
+
 VICTRON_VEBUS_SENSORS = {
     "inverter_charger_device_state": "float",
     "inverter_charger_battery_voltage": "14.45",

tests/components/victron_ble/test_sensor.py

@@ -41,6 +41,7 @@
     VICTRON_VEBUS_BAD_KEY_SERVICE_INFO,
     VICTRON_VEBUS_SERVICE_INFO,
     VICTRON_VEBUS_TOKEN,
+    VICTRON_VEBUS_UNRECOGNIZED_MODE_SERVICE_INFO,
 )
 
 from tests.common import MockConfigEntry, snapshot_platform
@@ -165,6 +166,28 @@ def _inject_bad_advertisement(hass: HomeAssistant, seq: int = 0) -> None:
     )
 
 
+def _inject_unrecognized_mode_advertisement(hass: HomeAssistant, seq: int = 0) -> None:
+    """Inject a Victron advertisement with an unrecognized mode byte.
+
+    detect_device_type returns None for this payload so the reauth guard
+    must treat it as neutral (neither increment nor reset the failure counter).
+    """
+    info = VICTRON_VEBUS_UNRECOGNIZED_MODE_SERVICE_INFO
+    raw = bytearray(info.manufacturer_data[VICTRON_IDENTIFIER])
+    raw[-1] = seq & 0xFF
+    device = generate_ble_device(address=info.address, name=info.name, details={})
+    adv = generate_advertisement_data(
+        local_name=info.name,
+        manufacturer_data={VICTRON_IDENTIFIER: bytes(raw)},
+        service_data=info.service_data,
+        service_uuids=info.service_uuids,
+        rssi=-60,
+    )
+    inject_advertisement_with_time_and_source_connectable(
+        hass, device, adv, time.monotonic(), "local", True
+    )
+
+
 @pytest.mark.usefixtures("enable_bluetooth")
 async def test_reauth_triggered_after_consecutive_failures(
     hass: HomeAssistant,
@@ -323,3 +346,76 @@ async def test_charger_error_state(
     state = hass.states.get("sensor.solar_charger_charger_error")
     assert state is not None
     assert state.state == expected_state
+
+
+@pytest.mark.usefixtures("enable_bluetooth")
+async def test_reauth_not_triggered_on_unrecognized_mode(
+    hass: HomeAssistant,
+    mock_config_entry_added_to_hass: MockConfigEntry,
+) -> None:
+    """Test reauth is NOT triggered by advertisements with unrecognized mode bytes.
+
+    Some Victron devices broadcast advertisements with mode bytes that
+    detect_device_type does not recognize (returns None).
+    validate_advertisement_key also returns False for these, but that does
+    not mean the encryption key is wrong.
+
+    Regression test for https://github.com/home-assistant/core/issues/168019
+    """
+    entry = mock_config_entry_added_to_hass
+
+    assert await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    # First inject a valid advertisement so update.devices is populated
+    inject_bluetooth_service_info(hass, VICTRON_VEBUS_SERVICE_INFO)
+    await hass.async_block_till_done()
+
+    # Now send many unrecognized-mode advertisements
+    for i in range(REAUTH_AFTER_FAILURES + 5):
+        _inject_unrecognized_mode_advertisement(hass, seq=i)
+        await hass.async_block_till_done()
+
+    flows = hass.config_entries.flow.async_progress_by_handler(DOMAIN)
+    assert len(flows) == 0
+
+
+@pytest.mark.usefixtures("enable_bluetooth")
+async def test_reauth_still_triggers_across_unrecognized_mode(
+    hass: HomeAssistant,
+    mock_config_entry_added_to_hass: MockConfigEntry,
+) -> None:
+    """Test that unrecognized-mode advertisements are neutral for the failure counter.
+
+    The sequence bad → bad → unrecognized → bad must still trigger reauth
+    because unrecognized advertisements should neither increment nor reset the
+    consecutive failure counter.
+
+    Regression test for https://github.com/home-assistant/core/issues/168019
+    """
+    entry = mock_config_entry_added_to_hass
+
+    assert await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    # First inject a valid advertisement so update.devices is populated
+    inject_bluetooth_service_info(hass, VICTRON_VEBUS_SERVICE_INFO)
+    await hass.async_block_till_done()
+
+    # bad, bad (2 failures)
+    _inject_bad_advertisement(hass, seq=100)
+    await hass.async_block_till_done()
+    _inject_bad_advertisement(hass, seq=101)
+    await hass.async_block_till_done()
+
+    # unrecognized mode — should be neutral
+    _inject_unrecognized_mode_advertisement(hass, seq=50)
+    await hass.async_block_till_done()
+
+    # one more bad → 3 consecutive failures → reauth
+    _inject_bad_advertisement(hass, seq=102)
+    await hass.async_block_till_done()
+
+    flows = hass.config_entries.flow.async_progress_by_handler(DOMAIN)
+    assert len(flows) == 1
+    assert flows[0]["context"]["source"] == "reauth"