From 26b1222faedd72be7c6f0f53209976ca7a34f978 Mon Sep 17 00:00:00 2001 From: Jc2k Date: Sat, 2 Sep 2023 15:21:05 +0100 Subject: [PATCH] Support tracking private bluetooth devices (#99465) Co-authored-by: J. Nick Koston --- .strict-typing | 1 + CODEOWNERS | 2 + .../components/private_ble_device/__init__.py | 19 ++ .../private_ble_device/config_flow.py | 60 +++++ .../components/private_ble_device/const.py | 2 + .../private_ble_device/coordinator.py | 236 ++++++++++++++++++ .../private_ble_device/device_tracker.py | 75 ++++++ .../components/private_ble_device/entity.py | 71 ++++++ .../private_ble_device/manifest.json | 10 + .../private_ble_device/strings.json | 20 ++ homeassistant/generated/config_flows.py | 1 + homeassistant/generated/integrations.json | 6 + mypy.ini | 10 + requirements_all.txt | 1 + requirements_test_all.txt | 1 + .../components/private_ble_device/__init__.py | 78 ++++++ .../components/private_ble_device/conftest.py | 1 + .../private_ble_device/test_config_flow.py | 132 ++++++++++ .../private_ble_device/test_device_tracker.py | 183 ++++++++++++++ 19 files changed, 909 insertions(+) create mode 100644 homeassistant/components/private_ble_device/__init__.py create mode 100644 homeassistant/components/private_ble_device/config_flow.py create mode 100644 homeassistant/components/private_ble_device/const.py create mode 100644 homeassistant/components/private_ble_device/coordinator.py create mode 100644 homeassistant/components/private_ble_device/device_tracker.py create mode 100644 homeassistant/components/private_ble_device/entity.py create mode 100644 homeassistant/components/private_ble_device/manifest.json create mode 100644 homeassistant/components/private_ble_device/strings.json create mode 100644 tests/components/private_ble_device/__init__.py create mode 100644 tests/components/private_ble_device/conftest.py create mode 100644 tests/components/private_ble_device/test_config_flow.py create mode 100644 tests/components/private_ble_device/test_device_tracker.py diff --git a/.strict-typing b/.strict-typing index 3059c42f33f..2a6e9b04cbe 100644 --- a/.strict-typing +++ b/.strict-typing @@ -255,6 +255,7 @@ homeassistant.components.persistent_notification.* homeassistant.components.pi_hole.* homeassistant.components.ping.* homeassistant.components.powerwall.* +homeassistant.components.private_ble_device.* homeassistant.components.proximity.* homeassistant.components.prusalink.* homeassistant.components.pure_energie.* diff --git a/CODEOWNERS b/CODEOWNERS index bf6fdaf9fc5..b937c2769fc 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -951,6 +951,8 @@ build.json @home-assistant/supervisor /tests/components/poolsense/ @haemishkyd /homeassistant/components/powerwall/ @bdraco @jrester @daniel-simpson /tests/components/powerwall/ @bdraco @jrester @daniel-simpson +/homeassistant/components/private_ble_device/ @Jc2k +/tests/components/private_ble_device/ @Jc2k /homeassistant/components/profiler/ @bdraco /tests/components/profiler/ @bdraco /homeassistant/components/progettihwsw/ @ardaseremet diff --git a/homeassistant/components/private_ble_device/__init__.py b/homeassistant/components/private_ble_device/__init__.py new file mode 100644 index 00000000000..c4666ccc02f --- /dev/null +++ b/homeassistant/components/private_ble_device/__init__.py @@ -0,0 +1,19 @@ +"""Private BLE Device integration.""" +from __future__ import annotations + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant + +PLATFORMS = [Platform.DEVICE_TRACKER] + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up tracking of a private bluetooth device from a config entry.""" + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Unload entities for a config entry.""" + return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) diff --git a/homeassistant/components/private_ble_device/config_flow.py b/homeassistant/components/private_ble_device/config_flow.py new file mode 100644 index 00000000000..5bf130a0396 --- /dev/null +++ b/homeassistant/components/private_ble_device/config_flow.py @@ -0,0 +1,60 @@ +"""Config flow for the BLE Tracker.""" +from __future__ import annotations + +import base64 +import binascii +import logging + +import voluptuous as vol + +from homeassistant.components import bluetooth +from homeassistant.config_entries import ConfigFlow +from homeassistant.data_entry_flow import FlowResult + +from .const import DOMAIN +from .coordinator import async_last_service_info + +_LOGGER = logging.getLogger(__name__) + +CONF_IRK = "irk" + + +class BLEDeviceTrackerConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for BLE Device Tracker.""" + + VERSION = 1 + + async def async_step_user( + self, user_input: dict[str, str] | None = None + ) -> FlowResult: + """Set up by user.""" + errors: dict[str, str] = {} + + if not bluetooth.async_scanner_count(self.hass, connectable=False): + return self.async_abort(reason="bluetooth_not_available") + + if user_input is not None: + irk = user_input[CONF_IRK] + if irk.startswith("irk:"): + irk = irk[4:] + + if irk.endswith("="): + irk_bytes = bytes(reversed(base64.b64decode(irk))) + else: + irk_bytes = binascii.unhexlify(irk) + + if len(irk_bytes) != 16: + errors[CONF_IRK] = "irk_not_valid" + elif not (service_info := async_last_service_info(self.hass, irk_bytes)): + errors[CONF_IRK] = "irk_not_found" + else: + await self.async_set_unique_id(irk_bytes.hex()) + return self.async_create_entry( + title=service_info.name or "BLE Device Tracker", + data={CONF_IRK: irk_bytes.hex()}, + ) + + data_schema = vol.Schema({CONF_IRK: str}) + return self.async_show_form( + step_id="user", data_schema=data_schema, errors=errors + ) diff --git a/homeassistant/components/private_ble_device/const.py b/homeassistant/components/private_ble_device/const.py new file mode 100644 index 00000000000..086fd06bfd5 --- /dev/null +++ b/homeassistant/components/private_ble_device/const.py @@ -0,0 +1,2 @@ +"""Constants for Private BLE Device.""" +DOMAIN = "private_ble_device" diff --git a/homeassistant/components/private_ble_device/coordinator.py b/homeassistant/components/private_ble_device/coordinator.py new file mode 100644 index 00000000000..863b2833851 --- /dev/null +++ b/homeassistant/components/private_ble_device/coordinator.py @@ -0,0 +1,236 @@ +"""Central manager for tracking devices with random but resolvable MAC addresses.""" +from __future__ import annotations + +from collections.abc import Callable +import logging +from typing import cast + +from bluetooth_data_tools import get_cipher_for_irk, resolve_private_address +from cryptography.hazmat.primitives.ciphers import Cipher + +from homeassistant.components import bluetooth +from homeassistant.components.bluetooth.match import BluetoothCallbackMatcher +from homeassistant.core import HomeAssistant + +from .const import DOMAIN + +_LOGGER = logging.getLogger(__name__) + +UnavailableCallback = Callable[[bluetooth.BluetoothServiceInfoBleak], None] +Cancellable = Callable[[], None] + + +def async_last_service_info( + hass: HomeAssistant, irk: bytes +) -> bluetooth.BluetoothServiceInfoBleak | None: + """Find a BluetoothServiceInfoBleak for the irk. + + This iterates over all currently visible mac addresses and checks them against `irk`. + It returns the newest. + """ + + # This can't use existing data collected by the coordinator - its called when + # the coordinator doesn't know about the IRK, so doesn't optimise this lookup. + + cur: bluetooth.BluetoothServiceInfoBleak | None = None + cipher = get_cipher_for_irk(irk) + + for service_info in bluetooth.async_discovered_service_info(hass, False): + if resolve_private_address(cipher, service_info.address): + if not cur or cur.time < service_info.time: + cur = service_info + + return cur + + +class PrivateDevicesCoordinator: + """Monitor private bluetooth devices and correlate them with known IRK. + + This class should not be instanced directly - use `async_get_coordinator` to get an instance. + + There is a single shared coordinator for all instances of this integration. This is to avoid + unnecessary hashing (AES) operations as much as possible. + """ + + def __init__(self, hass: HomeAssistant) -> None: + """Initialize the manager.""" + self.hass = hass + + self._irks: dict[bytes, Cipher] = {} + self._unavailable_callbacks: dict[bytes, list[UnavailableCallback]] = {} + self._service_info_callbacks: dict[ + bytes, list[bluetooth.BluetoothCallback] + ] = {} + + self._mac_to_irk: dict[str, bytes] = {} + self._irk_to_mac: dict[bytes, str] = {} + + # These MAC addresses have been compared to the IRK list + # They are unknown, so we can ignore them. + self._ignored: dict[str, Cancellable] = {} + + self._unavailability_trackers: dict[bytes, Cancellable] = {} + self._listener_cancel: Cancellable | None = None + + def _async_ensure_started(self) -> None: + if not self._listener_cancel: + self._listener_cancel = bluetooth.async_register_callback( + self.hass, + self._async_track_service_info, + BluetoothCallbackMatcher(connectable=False), + bluetooth.BluetoothScanningMode.ACTIVE, + ) + + def _async_ensure_stopped(self) -> None: + if self._listener_cancel: + self._listener_cancel() + self._listener_cancel = None + + for cancel in self._ignored.values(): + cancel() + self._ignored.clear() + + def _async_track_unavailable( + self, service_info: bluetooth.BluetoothServiceInfoBleak + ) -> None: + # This should be called when the current MAC address associated with an IRK goes away. + if resolved := self._mac_to_irk.get(service_info.address): + if callbacks := self._unavailable_callbacks.get(resolved): + for cb in callbacks: + cb(service_info) + return + + def _async_irk_resolved_to_mac(self, irk: bytes, mac: str) -> None: + if previous_mac := self._irk_to_mac.get(irk): + self._mac_to_irk.pop(previous_mac, None) + + self._mac_to_irk[mac] = irk + self._irk_to_mac[irk] = mac + + # Stop ignoring this MAC + self._ignored.pop(mac, None) + + # Ignore availability events for the previous address + if cancel := self._unavailability_trackers.pop(irk, None): + cancel() + + # Track available for new address + self._unavailability_trackers[irk] = bluetooth.async_track_unavailable( + self.hass, self._async_track_unavailable, mac, False + ) + + def _async_track_service_info( + self, + service_info: bluetooth.BluetoothServiceInfoBleak, + change: bluetooth.BluetoothChange, + ) -> None: + mac = service_info.address + + if mac in self._ignored: + return + + if resolved := self._mac_to_irk.get(mac): + if callbacks := self._service_info_callbacks.get(resolved): + for cb in callbacks: + cb(service_info, change) + return + + for irk, cipher in self._irks.items(): + if resolve_private_address(cipher, service_info.address): + self._async_irk_resolved_to_mac(irk, mac) + if callbacks := self._service_info_callbacks.get(irk): + for cb in callbacks: + cb(service_info, change) + return + + def _unignore(service_info: bluetooth.BluetoothServiceInfoBleak) -> None: + self._ignored.pop(service_info.address, None) + + self._ignored[mac] = bluetooth.async_track_unavailable( + self.hass, _unignore, mac, False + ) + + def _async_maybe_learn_irk(self, irk: bytes) -> None: + """Add irk to list of irks that we can use to resolve RPAs.""" + if irk not in self._irks: + if service_info := async_last_service_info(self.hass, irk): + self._async_irk_resolved_to_mac(irk, service_info.address) + self._irks[irk] = get_cipher_for_irk(irk) + + def _async_maybe_forget_irk(self, irk: bytes) -> None: + """If no downstream caller is tracking this irk, lets forget it.""" + if irk in self._service_info_callbacks or irk in self._unavailable_callbacks: + return + + # Ignore availability events for this irk as no + # one is listening. + if cancel := self._unavailability_trackers.pop(irk, None): + cancel() + + del self._irks[irk] + + if mac := self._irk_to_mac.pop(irk, None): + self._mac_to_irk.pop(mac, None) + + if not self._mac_to_irk: + self._async_ensure_stopped() + + def async_track_service_info( + self, callback: bluetooth.BluetoothCallback, irk: bytes + ) -> Cancellable: + """Receive a callback when a new advertisement is received for an irk. + + Returns a callback that can be used to cancel the registration. + """ + self._async_ensure_started() + self._async_maybe_learn_irk(irk) + + callbacks = self._service_info_callbacks.setdefault(irk, []) + callbacks.append(callback) + + def _unsubscribe() -> None: + callbacks.remove(callback) + if not callbacks: + self._service_info_callbacks.pop(irk, None) + self._async_maybe_forget_irk(irk) + + return _unsubscribe + + def async_track_unavailable( + self, + callback: UnavailableCallback, + irk: bytes, + ) -> Cancellable: + """Register to receive a callback when an irk is unavailable. + + Returns a callback that can be used to cancel the registration. + """ + self._async_ensure_started() + self._async_maybe_learn_irk(irk) + + callbacks = self._unavailable_callbacks.setdefault(irk, []) + callbacks.append(callback) + + def _unsubscribe() -> None: + callbacks.remove(callback) + if not callbacks: + self._unavailable_callbacks.pop(irk, None) + + self._async_maybe_forget_irk(irk) + + return _unsubscribe + + +def async_get_coordinator(hass: HomeAssistant) -> PrivateDevicesCoordinator: + """Create or return an existing PrivateDeviceManager. + + There should only be one per HomeAssistant instance. Associating private + mac addresses with an IRK involves AES operations. We don't want to + duplicate that work. + """ + if existing := hass.data.get(DOMAIN): + return cast(PrivateDevicesCoordinator, existing) + + pdm = hass.data[DOMAIN] = PrivateDevicesCoordinator(hass) + + return pdm diff --git a/homeassistant/components/private_ble_device/device_tracker.py b/homeassistant/components/private_ble_device/device_tracker.py new file mode 100644 index 00000000000..64e23b25ebe --- /dev/null +++ b/homeassistant/components/private_ble_device/device_tracker.py @@ -0,0 +1,75 @@ +"""Tracking for bluetooth low energy devices.""" +from __future__ import annotations + +from collections.abc import Mapping +import logging + +from homeassistant.components import bluetooth +from homeassistant.components.device_tracker import SourceType +from homeassistant.components.device_tracker.config_entry import BaseTrackerEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import STATE_HOME, STATE_NOT_HOME +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from .entity import BasePrivateDeviceEntity + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Load Device Tracker entities for a config entry.""" + async_add_entities([BasePrivateDeviceTracker(config_entry)]) + + +class BasePrivateDeviceTracker(BasePrivateDeviceEntity, BaseTrackerEntity): + """A trackable Private Bluetooth Device.""" + + _attr_should_poll = False + _attr_has_entity_name = True + _attr_name = None + + @property + def extra_state_attributes(self) -> Mapping[str, str]: + """Return extra state attributes for this device.""" + if last_info := self._last_info: + return { + "current_address": last_info.address, + "source": last_info.source, + } + return {} + + @callback + def _async_track_unavailable( + self, service_info: bluetooth.BluetoothServiceInfoBleak + ) -> None: + self._last_info = None + self.async_write_ha_state() + + @callback + def _async_track_service_info( + self, + service_info: bluetooth.BluetoothServiceInfoBleak, + change: bluetooth.BluetoothChange, + ) -> None: + self._last_info = service_info + self.async_write_ha_state() + + @property + def state(self) -> str: + """Return the state of the device.""" + return STATE_HOME if self._last_info else STATE_NOT_HOME + + @property + def source_type(self) -> SourceType: + """Return the source type, eg gps or router, of the device.""" + return SourceType.BLUETOOTH_LE + + @property + def icon(self) -> str: + """Return device icon.""" + return "mdi:bluetooth-connect" if self._last_info else "mdi:bluetooth-off" diff --git a/homeassistant/components/private_ble_device/entity.py b/homeassistant/components/private_ble_device/entity.py new file mode 100644 index 00000000000..ae632213506 --- /dev/null +++ b/homeassistant/components/private_ble_device/entity.py @@ -0,0 +1,71 @@ +"""Tracking for bluetooth low energy devices.""" +from __future__ import annotations + +from abc import abstractmethod +import binascii + +from homeassistant.components import bluetooth +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import callback +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity import Entity + +from .const import DOMAIN +from .coordinator import async_get_coordinator, async_last_service_info + + +class BasePrivateDeviceEntity(Entity): + """Base Private Bluetooth Entity.""" + + _attr_should_poll = False + _attr_has_entity_name = True + + def __init__(self, config_entry: ConfigEntry) -> None: + """Set up a new BleScanner entity.""" + irk = config_entry.data["irk"] + + self._attr_unique_id = irk + + self._attr_device_info = DeviceInfo( + name=f"Private BLE Device {irk[:6]}", + identifiers={(DOMAIN, irk)}, + ) + + self._entry = config_entry + self._irk = binascii.unhexlify(irk) + self._last_info: bluetooth.BluetoothServiceInfoBleak | None = None + + async def async_added_to_hass(self) -> None: + """Configure entity when it is added to Home Assistant.""" + coordinator = async_get_coordinator(self.hass) + self.async_on_remove( + coordinator.async_track_service_info( + self._async_track_service_info, self._irk + ) + ) + self.async_on_remove( + coordinator.async_track_unavailable( + self._async_track_unavailable, self._irk + ) + ) + + if service_info := async_last_service_info(self.hass, self._irk): + self._async_track_service_info( + service_info, bluetooth.BluetoothChange.ADVERTISEMENT + ) + + @abstractmethod + @callback + def _async_track_unavailable( + self, service_info: bluetooth.BluetoothServiceInfoBleak + ) -> None: + """Respond when the bluetooth device being tracked is no longer visible.""" + + @abstractmethod + @callback + def _async_track_service_info( + self, + service_info: bluetooth.BluetoothServiceInfoBleak, + change: bluetooth.BluetoothChange, + ) -> None: + """Respond when the bluetooth device being tracked broadcasted updated information.""" diff --git a/homeassistant/components/private_ble_device/manifest.json b/homeassistant/components/private_ble_device/manifest.json new file mode 100644 index 00000000000..3497138178c --- /dev/null +++ b/homeassistant/components/private_ble_device/manifest.json @@ -0,0 +1,10 @@ +{ + "domain": "private_ble_device", + "name": "Private BLE Device", + "codeowners": ["@Jc2k"], + "config_flow": true, + "dependencies": ["bluetooth_adapters"], + "documentation": "https://www.home-assistant.io/integrations/private_ble_device", + "iot_class": "local_push", + "requirements": ["bluetooth-data-tools==1.11.0"] +} diff --git a/homeassistant/components/private_ble_device/strings.json b/homeassistant/components/private_ble_device/strings.json new file mode 100644 index 00000000000..c62ea5c4d50 --- /dev/null +++ b/homeassistant/components/private_ble_device/strings.json @@ -0,0 +1,20 @@ +{ + "config": { + "flow_title": "[%key:component::bluetooth::config::flow_title%]", + "step": { + "user": { + "description": "What is the IRK (Identity Resolving Key) of the BLE device you want to track?", + "data": { + "irk": "IRK" + } + } + }, + "error": { + "irk_not_found": "The provided IRK does not match any BLE devices that Home Assistant can see.", + "irk_not_valid": "The key does not look like a valid IRK." + }, + "abort": { + "bluetooth_not_available": "At least one Bluetooth adapter or remote bluetooth proxy must be configured to track Private BLE Devices." + } + } +} diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 7d84dc87cbe..6c992fd4b5e 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -351,6 +351,7 @@ FLOWS = { "point", "poolsense", "powerwall", + "private_ble_device", "profiler", "progettihwsw", "prosegur", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index c357b5aed4c..a9e19441693 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -4320,6 +4320,12 @@ "config_flow": true, "iot_class": "cloud_polling" }, + "private_ble_device": { + "name": "Private BLE Device", + "integration_type": "hub", + "config_flow": true, + "iot_class": "local_push" + }, "profiler": { "name": "Profiler", "integration_type": "hub", diff --git a/mypy.ini b/mypy.ini index 9802c26c3c6..178b82fd359 100644 --- a/mypy.ini +++ b/mypy.ini @@ -2312,6 +2312,16 @@ disallow_untyped_defs = true warn_return_any = true warn_unreachable = true +[mypy-homeassistant.components.private_ble_device.*] +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +warn_return_any = true +warn_unreachable = true + [mypy-homeassistant.components.proximity.*] check_untyped_defs = true disallow_incomplete_defs = true diff --git a/requirements_all.txt b/requirements_all.txt index d72b41f442e..be7a06399d2 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -550,6 +550,7 @@ bluetooth-auto-recovery==1.2.1 # homeassistant.components.esphome # homeassistant.components.ld2410_ble # homeassistant.components.led_ble +# homeassistant.components.private_ble_device bluetooth-data-tools==1.11.0 # homeassistant.components.bond diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 3049cb33268..5362d5ac2b5 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -461,6 +461,7 @@ bluetooth-auto-recovery==1.2.1 # homeassistant.components.esphome # homeassistant.components.ld2410_ble # homeassistant.components.led_ble +# homeassistant.components.private_ble_device bluetooth-data-tools==1.11.0 # homeassistant.components.bond diff --git a/tests/components/private_ble_device/__init__.py b/tests/components/private_ble_device/__init__.py new file mode 100644 index 00000000000..df9929293a1 --- /dev/null +++ b/tests/components/private_ble_device/__init__.py @@ -0,0 +1,78 @@ +"""Tests for private_ble_device.""" + +from datetime import timedelta +import time +from unittest.mock import patch + +from home_assistant_bluetooth import BluetoothServiceInfoBleak + +from homeassistant import config_entries +from homeassistant.components.private_ble_device.const import DOMAIN +from homeassistant.core import HomeAssistant +from homeassistant.util import dt as dt_util + +from tests.common import MockConfigEntry, async_fire_time_changed +from tests.components.bluetooth import ( + generate_advertisement_data, + generate_ble_device, + inject_bluetooth_service_info_bleak, +) + +MAC_RPA_VALID_1 = "40:01:02:0a:c4:a6" +MAC_RPA_VALID_2 = "40:02:03:d2:74:ce" +MAC_RPA_INVALID = "40:00:00:d2:74:ce" +MAC_STATIC = "00:01:ff:a0:3a:76" + +DUMMY_IRK = "00000000000000000000000000000000" + + +async def async_mock_config_entry(hass: HomeAssistant, irk: str = DUMMY_IRK) -> None: + """Create a test device for a dummy IRK.""" + entry = MockConfigEntry( + version=1, + domain=DOMAIN, + entry_id=irk, + data={"irk": irk}, + title="Private BLE Device 000000", + ) + entry.add_to_hass(hass) + + assert await hass.config_entries.async_setup(entry.entry_id) + assert entry.state is config_entries.ConfigEntryState.LOADED + await hass.async_block_till_done() + + +async def async_inject_broadcast( + hass: HomeAssistant, + mac: str = MAC_RPA_VALID_1, + mfr_data: bytes = b"", + broadcast_time: float | None = None, +) -> None: + """Inject an advertisement.""" + inject_bluetooth_service_info_bleak( + hass, + BluetoothServiceInfoBleak( + name="Test Test Test", + address=mac, + rssi=-63, + service_data={}, + manufacturer_data={1: mfr_data}, + service_uuids=[], + source="local", + device=generate_ble_device(mac, "Test Test Test"), + advertisement=generate_advertisement_data(local_name="Not it"), + time=broadcast_time or time.monotonic(), + connectable=False, + ), + ) + await hass.async_block_till_done() + + +async def async_move_time_forwards(hass: HomeAssistant, offset: float): + """Mock time advancing from now to now+offset.""" + with patch( + "homeassistant.components.bluetooth.manager.MONOTONIC_TIME", + return_value=time.monotonic() + offset, + ): + async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=offset)) + await hass.async_block_till_done() diff --git a/tests/components/private_ble_device/conftest.py b/tests/components/private_ble_device/conftest.py new file mode 100644 index 00000000000..b33dc1d4ea2 --- /dev/null +++ b/tests/components/private_ble_device/conftest.py @@ -0,0 +1 @@ +"""private_ble_device fixtures.""" diff --git a/tests/components/private_ble_device/test_config_flow.py b/tests/components/private_ble_device/test_config_flow.py new file mode 100644 index 00000000000..aa8ea0d905c --- /dev/null +++ b/tests/components/private_ble_device/test_config_flow.py @@ -0,0 +1,132 @@ +"""Tests for private bluetooth device config flow.""" +from unittest.mock import patch + +from homeassistant import config_entries +from homeassistant.components.private_ble_device import const +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResult, FlowResultType +from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo + +from tests.components.bluetooth import inject_bluetooth_service_info + + +def assert_form_error(result: FlowResult, key: str, value: str) -> None: + """Assert that a flow returned a form error.""" + assert result["type"] == "form" + assert result["errors"] + assert result["errors"][key] == value + + +async def test_setup_user_no_bluetooth( + hass: HomeAssistant, mock_bluetooth_adapters: None +) -> None: + """Test setting up via user interaction when bluetooth is not enabled.""" + result = await hass.config_entries.flow.async_init( + const.DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "bluetooth_not_available" + + +async def test_invalid_irk(hass: HomeAssistant, enable_bluetooth: None) -> None: + """Test invalid irk.""" + result = await hass.config_entries.flow.async_init( + const.DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] == "form" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={"irk": "irk:000000"} + ) + assert_form_error(result, "irk", "irk_not_valid") + + +async def test_irk_not_found(hass: HomeAssistant, enable_bluetooth: None) -> None: + """Test irk not found.""" + result = await hass.config_entries.flow.async_init( + const.DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] == "form" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"irk": "irk:00000000000000000000000000000000"}, + ) + assert_form_error(result, "irk", "irk_not_found") + + +async def test_flow_works(hass: HomeAssistant, enable_bluetooth: None) -> None: + """Test config flow works.""" + + inject_bluetooth_service_info( + hass, + BluetoothServiceInfo( + name="Test Test Test", + address="40:01:02:0a:c4:a6", + rssi=-63, + service_data={}, + manufacturer_data={}, + service_uuids=[], + source="local", + ), + ) + + result = await hass.config_entries.flow.async_init( + const.DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] == "form" + + # Check you can finish the flow + with patch( + "homeassistant.components.private_ble_device.async_setup_entry", + return_value=True, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"irk": "irk:00000000000000000000000000000000"}, + ) + + assert result["type"] == FlowResultType.CREATE_ENTRY + assert result["title"] == "Test Test Test" + assert result["data"] == {"irk": "00000000000000000000000000000000"} + assert result["result"].unique_id == "00000000000000000000000000000000" + + +async def test_flow_works_by_base64( + hass: HomeAssistant, enable_bluetooth: None +) -> None: + """Test config flow works.""" + + inject_bluetooth_service_info( + hass, + BluetoothServiceInfo( + name="Test Test Test", + address="40:01:02:0a:c4:a6", + rssi=-63, + service_data={}, + manufacturer_data={}, + service_uuids=[], + source="local", + ), + ) + + result = await hass.config_entries.flow.async_init( + const.DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] == "form" + + # Check you can finish the flow + with patch( + "homeassistant.components.private_ble_device.async_setup_entry", + return_value=True, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"irk": "AAAAAAAAAAAAAAAAAAAAAA=="}, + ) + + assert result["type"] == FlowResultType.CREATE_ENTRY + assert result["title"] == "Test Test Test" + assert result["data"] == {"irk": "00000000000000000000000000000000"} + assert result["result"].unique_id == "00000000000000000000000000000000" diff --git a/tests/components/private_ble_device/test_device_tracker.py b/tests/components/private_ble_device/test_device_tracker.py new file mode 100644 index 00000000000..776ba503983 --- /dev/null +++ b/tests/components/private_ble_device/test_device_tracker.py @@ -0,0 +1,183 @@ +"""Tests for polling measures.""" + + +import time + +from homeassistant.components.bluetooth.advertisement_tracker import ( + ADVERTISING_TIMES_NEEDED, +) +from homeassistant.core import HomeAssistant + +from . import ( + MAC_RPA_VALID_1, + MAC_RPA_VALID_2, + MAC_STATIC, + async_inject_broadcast, + async_mock_config_entry, + async_move_time_forwards, +) + +from tests.components.bluetooth.test_advertisement_tracker import ONE_HOUR_SECONDS + + +async def test_tracker_created(hass: HomeAssistant, enable_bluetooth: None) -> None: + """Test creating a tracker entity when no devices have been seen.""" + await async_mock_config_entry(hass) + + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "not_home" + + +async def test_tracker_ignore_other_rpa( + hass: HomeAssistant, enable_bluetooth: None +) -> None: + """Test that tracker ignores RPA's that don't match us.""" + await async_mock_config_entry(hass) + await async_inject_broadcast(hass, MAC_STATIC) + + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "not_home" + + +async def test_tracker_already_home( + hass: HomeAssistant, enable_bluetooth: None +) -> None: + """Test creating a tracker and the device was already discovered by HA.""" + await async_inject_broadcast(hass, MAC_RPA_VALID_1) + await async_mock_config_entry(hass) + + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + + +async def test_tracker_arrive_home(hass: HomeAssistant, enable_bluetooth: None) -> None: + """Test transition from not_home to home.""" + await async_mock_config_entry(hass) + await async_inject_broadcast(hass, MAC_RPA_VALID_1, b"1") + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + assert state.attributes["current_address"] == "40:01:02:0a:c4:a6" + assert state.attributes["source"] == "local" + + await async_inject_broadcast(hass, MAC_STATIC, b"1") + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + + # Test same wrong mac address again to exercise some caching + await async_inject_broadcast(hass, MAC_STATIC, b"2") + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + + # And test original mac address again. + # Use different mfr data so that event bubbles up + await async_inject_broadcast(hass, MAC_RPA_VALID_1, b"2") + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + assert state.attributes["current_address"] == "40:01:02:0a:c4:a6" + + +async def test_tracker_isolation(hass: HomeAssistant, enable_bluetooth: None) -> None: + """Test creating 2 tracker entities doesn't confuse anything.""" + await async_mock_config_entry(hass) + await async_mock_config_entry(hass, irk="1" * 32) + + # This broadcast should only impact the first entity + await async_inject_broadcast(hass, MAC_RPA_VALID_1, b"1") + + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + + state = hass.states.get("device_tracker.private_ble_device_111111") + assert state + assert state.state == "not_home" + + +async def test_tracker_mac_rotate(hass: HomeAssistant, enable_bluetooth: None) -> None: + """Test MAC address rotation.""" + await async_inject_broadcast(hass, MAC_RPA_VALID_1) + await async_mock_config_entry(hass) + + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + assert state.attributes["current_address"] == MAC_RPA_VALID_1 + + await async_inject_broadcast(hass, MAC_RPA_VALID_2) + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + assert state.attributes["current_address"] == MAC_RPA_VALID_2 + + +async def test_tracker_start_stale(hass: HomeAssistant, enable_bluetooth: None) -> None: + """Test edge case where we find an existing stale record, and it expires before we see any more.""" + time.monotonic() + + await async_inject_broadcast(hass, MAC_RPA_VALID_1) + await async_mock_config_entry(hass) + + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + + await async_move_time_forwards( + hass, ((ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS) + ) + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "not_home" + + +async def test_tracker_leave_home(hass: HomeAssistant, enable_bluetooth: None) -> None: + """Test tracker notices we have left.""" + time.monotonic() + + await async_mock_config_entry(hass) + await async_inject_broadcast(hass, MAC_RPA_VALID_1) + + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + + await async_move_time_forwards( + hass, ((ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS) + ) + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "not_home" + + +async def test_old_tracker_leave_home( + hass: HomeAssistant, enable_bluetooth: None +) -> None: + """Test tracker ignores an old stale mac address timing out.""" + start_time = time.monotonic() + + await async_mock_config_entry(hass) + + await async_inject_broadcast(hass, MAC_RPA_VALID_2, broadcast_time=start_time) + await async_inject_broadcast(hass, MAC_RPA_VALID_2, broadcast_time=start_time + 15) + + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + + # First address has timed out - still home + await async_move_time_forwards(hass, 910) + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "home" + + # Second address has time out - now away + await async_move_time_forwards(hass, 920) + state = hass.states.get("device_tracker.private_ble_device_000000") + assert state + assert state.state == "not_home"