Support tracking private bluetooth devices (#99465)

Co-authored-by: J. Nick Koston <nick@koston.org>

.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.*

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

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)

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
+        )

homeassistant/components/private_ble_device/const.py

@@ -0,0 +1,2 @@
+"""Constants for Private BLE Device."""
+DOMAIN = "private_ble_device"

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

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"

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."""

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"]
+}

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."
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -351,6 +351,7 @@ FLOWS = {
         "point",
         "poolsense",
         "powerwall",
+        "private_ble_device",
         "profiler",
         "progettihwsw",
         "prosegur",

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",

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

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

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

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()

tests/components/private_ble_device/conftest.py

@@ -0,0 +1 @@
+"""private_ble_device fixtures."""

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"

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"