mirror of
https://github.com/home-assistant/core.git
synced 2026-09-03 02:34:57 -05:00
Fix Victron BLE false reauth on unrecognised advertisement mode bytes (#168209)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
@@ -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 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
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
|
||||
|
||||
|
||||
@@ -201,6 +201,22 @@ VICTRON_DC_DC_CONVERTER_UNKNOWN_OFF_REASON_SERVICE_INFO = BluetoothServiceInfo(
|
||||
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",
|
||||
|
||||
@@ -41,6 +41,7 @@ from .fixtures import (
|
||||
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"
|
||||
|
||||
Reference in New Issue
Block a user