# # Copyright 2022 aiohomekit team # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # from __future__ import annotations import asyncio from collections.abc import Callable, Iterable from datetime import timedelta import logging import random import struct import time from typing import TYPE_CHECKING, Any, TypeVar, cast from uuid import UUID from bleak.backends.characteristic import BleakGATTCharacteristic from bleak.backends.device import BLEDevice from bleak.backends.scanner import AdvertisementData from bleak.exc import BleakError from bleak_retry_connector import ( BLEAK_RETRY_EXCEPTIONS as BLEAK_EXCEPTIONS, retry_bluetooth_connection_error, ) from aiohomekit.exceptions import ( AccessoryDisconnectedError, AccessoryNotFoundError, AuthenticationError, InvalidError, UnknownError, ) from aiohomekit.meshcop import Meshcop from aiohomekit.model import ( Accessories, AccessoriesState, Accessory, CharacteristicsTypes, Transport, ) from aiohomekit.model.characteristics import ( EVENT_CHARACTERISTICS, Characteristic, CharacteristicPermissions, ) from aiohomekit.model.services import Service, ServicesTypes from aiohomekit.pdu import OpCode, PDUStatus, decode_pdu, encode_pdu from aiohomekit.protocol import get_session_keys from aiohomekit.protocol.statuscodes import HapStatusCode from aiohomekit.protocol.tlv import TLV from aiohomekit.utils import async_create_task from aiohomekit.uuid import normalize_uuid from ..abstract import AbstractPairing, AbstractPairingData from .bleak import AIOHomeKitBleakClient from .client import ( PDUStatusError, ble_request, disconnect_on_missing_services, drive_pairing_state_machine, raise_for_pdu_status, ) from .connection import establish_connection from .key import BroadcastDecryptionKey, DecryptionKey, EncryptionKey from .manufacturer_data import HomeKitAdvertisement, HomeKitEncryptedNotification from .structs import ( HAP_BLE_CHARACTERISTIC_CONFIGURATION_REQUEST_TLV, HAP_BLE_PROTOCOL_CONFIGURATION_REQUEST_TLV, HAP_TLV, Characteristic as CharacteristicTLV, ProtocolParams, ProtocolParamsTLV, Service as ServiceTLV, ) from .values import from_bytes, to_bytes if TYPE_CHECKING: from aiohomekit.controller.ble.controller import BleController logger = logging.getLogger(__name__) # The discover timeout is how long we will wait for and advertisement to be # received. If we don't get it in this time we will try again later since # the scanner is always running anyways. DISCOVER_TIMEOUT = 10 # Battery powered devices may not broadcast once paired until # there is an event so we use a long availablity interval. AVAILABILITY_INTERVAL = 86400 * 7 # 7 days NEVER_TIME = -AVAILABILITY_INTERVAL START_NOTIFY_DEBOUNCE = 1.5 MAX_GSN = 65535 SERVICE_INSTANCE_ID = "E604E95D-A759-4817-87D3-AA005083A0D1" SERVICE_INSTANCE_ID_UUID = UUID(SERVICE_INSTANCE_ID) SERVICE_SIGNATURE_UUID = UUID(CharacteristicsTypes.SERVICE_SIGNATURE) SKIP_SYNC_SERVICES = { ServicesTypes.PAIRING, ServicesTypes.TRANSFER_TRANSPORT_MANAGEMENT, } # These characteristics are not readable unless there has been a write WRITE_FIRST_REQUIRED_CHARACTERISTICS = { CharacteristicsTypes.SETUP_DATA_STREAM_TRANSPORT, # Setup Data Stream Transport CharacteristicsTypes.SELECTED_RTP_STREAM_CONFIGURATION, # Selected RTP Stream Configuration CharacteristicsTypes.SETUP_ENDPOINTS, # Setup Endpoints "00000138-0000-1000-8000-0026BB765291", # Unknown write first characteristic "246912DC-8FA3-82ED-DEA4-9EB91D8FC2EE", # Unknown Vendor char seen on Belkin Wemo Switch } IGNORE_READ_CHARACTERISTICS = { CharacteristicsTypes.SERVICE_SIGNATURE } | WRITE_FIRST_REQUIRED_CHARACTERISTICS # IGNORE_READ_CHARACTERISTICS is used for notification IGNORE_POLL_CHARACTERISTICS = IGNORE_READ_CHARACTERISTICS | EVENT_CHARACTERISTICS # IGNORE_POLL_CHARACTERISTICS is used for polling BLE_AID = 1 # The aid for BLE devices is always 1 ENABLE_BROADCAST_PAYLOAD = TLV.encode_list( [ ( HAP_BLE_CHARACTERISTIC_CONFIGURATION_REQUEST_TLV.kTLVHAPParamCharacteristicConfigurationProperties, int.to_bytes(1, 2, "little"), ), ( HAP_BLE_CHARACTERISTIC_CONFIGURATION_REQUEST_TLV.kTLVHAPParamCharacteristicConfigurationBroadcastInterval, bytes([0x01]), ), ] ) GENERATE_BROADCAST_KEY_PAYLOAD = ( bytes([HAP_BLE_PROTOCOL_CONFIGURATION_REQUEST_TLV.GenerateBroadcastEncryptionKey]) + b"\x00" ) GET_ALL_PARAMS_PAYLOAD = ( bytes([HAP_BLE_PROTOCOL_CONFIGURATION_REQUEST_TLV.GetAllParams]) + b"\x00" ) # Higher priority means we will fetch these first CHAR_FETCH_PRIORITY = { CharacteristicsTypes.MOTION_DETECTED: 100, CharacteristicsTypes.CONTACT_STATE: 100, CharacteristicsTypes.OCCUPANCY_DETECTED: 100, CharacteristicsTypes.DOOR_STATE_CURRENT: 100, CharacteristicsTypes.DOOR_STATE_TARGET: 100, CharacteristicsTypes.INPUT_EVENT: 100, CharacteristicsTypes.ON: 100, CharacteristicsTypes.ACTIVE: 100, CharacteristicsTypes.LOCK_MECHANISM_CURRENT_STATE: 100, CharacteristicsTypes.LOCK_MECHANISM_TARGET_STATE: 100, CharacteristicsTypes.LEAK_DETECTED: 90, CharacteristicsTypes.SMOKE_DETECTED: 90, CharacteristicsTypes.CARBON_MONOXIDE_DETECTED: 90, CharacteristicsTypes.CARBON_DIOXIDE_DETECTED: 90, CharacteristicsTypes.BRIGHTNESS: 90, CharacteristicsTypes.HUE: 90, CharacteristicsTypes.SATURATION: 90, CharacteristicsTypes.COLOR_TEMPERATURE: 90, CharacteristicsTypes.VERSION: -50, CharacteristicsTypes.NAME: -50, CharacteristicsTypes.THREAD_CONTROL_POINT: -100, CharacteristicsTypes.THREAD_STATUS: -100, CharacteristicsTypes.THREAD_NODE_CAPABILITIES: -100, CharacteristicsTypes.THREAD_OPENTHREAD_VERSION: -100, } WrapFuncType = TypeVar("WrapFuncType", bound=Callable[..., Any]) def operation_lock(func: WrapFuncType) -> WrapFuncType: """Define a wrapper to only allow a single operation at a time.""" async def _async_operation_lock_wrap( self: BlePairing, *args: Any, **kwargs: Any ) -> None: async with self._operation_lock: return await func(self, *args, **kwargs) return cast(WrapFuncType, _async_operation_lock_wrap) def restore_connection_and_resume(func: WrapFuncType) -> WrapFuncType: """Define a wrapper restore connection, populate data, and then resume when the operation completes.""" async def _async_restore_and_resume( self: BlePairing, *args: Any, **kwargs: Any ) -> None: """Restore connection, populate data, and then resume when the operation completes.""" if self._shutdown: return await self._populate_accessories_and_characteristics() try: return await func(self, *args, **kwargs) finally: logger.debug( "%s: Finished %s, checking for subscription restore: %s", self.name, func.__name__, self._restore_pending, ) if not self._shutdown and self._restore_pending: await self._async_restore_subscriptions() return cast(WrapFuncType, _async_restore_and_resume) def force_fresh_connection(func: WrapFuncType) -> WrapFuncType: """Define a wrapper to force a fresh connection.""" async def _async_force_fresh_connection( self: BlePairing, *args: Any, **kwargs: Any ) -> None: """Force a fresh connection.""" if self.client: await self.client.disconnect() return await func(self, *args, **kwargs) return cast(WrapFuncType, _async_force_fresh_connection) class BlePairing(AbstractPairing): """ This represents a paired HomeKit IP accessory. """ description: HomeKitAdvertisement controller: BleController def __init__( self, controller: BleController, pairing_data: AbstractPairingData, device: BLEDevice | None = None, client: AIOHomeKitBleakClient | None = None, description: HomeKitAdvertisement | None = None, ble_advertisement: AdvertisementData | None = None, ) -> None: self.device = device self.ble_advertisement = ble_advertisement self.client = client self.description = description self.pairing_data = pairing_data super().__init__(controller, pairing_data) self._last_seen = time.monotonic() if description else NEVER_TIME if not description and self.state_num: self.description = HomeKitAdvertisement.from_cache( address=self.address, id=self.id, config_num=self.config_num, state_num=self.state_num, ) # Encryption self._derive = None self._session_id = None self._encryption_key: EncryptionKey | None = None self._decryption_key: DecryptionKey | None = None self._broadcast_decryption_key: BroadcastDecryptionKey | None = None cached_key = self.broadcast_key if cached_key: self._broadcast_decryption_key = BroadcastDecryptionKey(cached_key) # Used to keep track of which characteristics we already started # notifications for self._notifications: set[int] = set() self._broadcast_notifications: set[int] = set() # Only allow one attempt to aquire the connection at a time self._connection_lock = asyncio.Lock() # We don't want to read/write from characteristics in parallel # * If 2 coroutines read from the same char at the same time there # would be a race error - a read result could be overwritten by another. # * The enc/dec counters are global. Therefore our API's for # a read/write need to be atomic otherwise we end up having # to guess what encryption counter to use for the decrypt self._ble_request_lock = asyncio.Lock() # Only allow a single operation at at time self._operation_lock = asyncio.Lock() # Only allow a single attempt to sync config at a time self._config_lock = asyncio.Lock() # Only subscribe to characteristics one at a time self._subscription_lock = asyncio.Lock() # Only process disconnected events once self._disconnected_events_lock = asyncio.Lock() self._start_notify_timer: asyncio.TimerHandle | None = None self._tried_to_connect_once = False self._restore_pending = False self._fetched_gsn_this_session = False self._had_notify_this_session = False @property def address(self) -> str: """Return the address of the device.""" return ( self.device.address if self.device else self.pairing_data["AccessoryAddress"] ) @property def name(self) -> str: """Return the name of the pairing with the address.""" if self.description: return f"{self.description.name} [{self.address}] (id={self.id})" return f"[{self.address}] (id={self.id})" @property def rssi(self) -> int | None: return self.ble_advertisement.rssi if self.ble_advertisement else None @property def is_connected(self) -> bool: return bool(self.client and self.client.is_connected and self._encryption_key) @property def is_available(self) -> bool: """Returns true if the device is currently available.""" return self._is_available_at_time(time.monotonic()) @property def poll_interval(self) -> timedelta: """Returns how often the device should be polled.""" if any(a.needs_polling for a in self.accessories): # Currently only used for devices that have energy data return timedelta(minutes=5) return timedelta(hours=24) def _is_available_at_time(self, monotonic: float) -> bool: """Check if we are considered available at the given time.""" return self.is_connected or monotonic - self._last_seen < AVAILABILITY_INTERVAL @property def transport(self) -> Transport: """The transport used for the connection.""" return Transport.BLE def _update_state_num(self, state_num: int) -> None: """Update the state number.""" self.description.state_num = state_num self._update_cached_state_num(state_num) def _update_cached_state_num(self, state_num: int) -> None: """Update the cached state number which is restored between restarts.""" old_state_num = self._accessories_state.state_num self._accessories_state.state_num = state_num if old_state_num != state_num: self._update_accessories_state_cache() def _async_ble_update( self, device: BLEDevice, ble_advertisement: AdvertisementData ) -> None: """Update the BLE device and ble_advertisement.""" self.device = device self.ble_advertisement = ble_advertisement def _async_description_update( self, description: HomeKitAdvertisement | None ) -> None: """Update the description of the accessory.""" now = time.monotonic() was_available = self._is_available_at_time(now) self._last_seen = now if not was_available: self._callback_availability_changed(True) if self.description != description: logger.debug( "%s: Description updated: old=%s new=%s", self.name, self.description, description, ) super()._async_description_update(description) self._update_cached_state_num(description.state_num) async def _async_request( self, opcode: OpCode, char: Characteristic, data: bytes | None = None, iid: int | None = None, ) -> bytes: async with self._ble_request_lock: return await self._async_request_under_lock(opcode, char, data, iid) async def _async_request_under_lock( self, opcode: OpCode, char: Characteristic, data: bytes | None = None, iid: int | None = None, ) -> bytes: assert self._ble_request_lock.locked(), "_ble_request_lock Should be locked" if not self.client or not self.client.is_connected: logger.debug("%s: Client not connected; rssi=%s", self.name, self.rssi) raise AccessoryDisconnectedError(f"{self.name} is not connected") endpoint_iid = iid if iid is not None else char.iid endpoint = await self.client.get_characteristic( char.service.type, char.type, endpoint_iid ) try: pdu_status, result_data = await ble_request( self.client, self._encryption_key, self._decryption_key, opcode, endpoint, endpoint_iid, data, ) except BaseException as ex: # If the request fails for any reason (especially # cancellation), we need to close the connection # otherwise the encryption counters will be out of sync try: await self._close_while_locked() except Exception as exc: logger.debug( "%s: Failed to close connection after exception: %s", self.name, exc, ) # Make sure we propagate the exception to the caller # especially if it was a cancellation raise ex if not self.client or not self.client.is_connected: logger.debug("%s: Client not connected; rssi=%s", self.name, self.rssi) raise AccessoryDisconnectedError(f"{self.name} is not connected") raise_for_pdu_status(self.client, pdu_status) return result_data def _async_disconnected(self, client: AIOHomeKitBleakClient) -> None: """Called when bleak disconnects from the accessory closed the connection.""" logger.debug("%s: Session closed callback: rssi=%s", self.name, self.rssi) self._async_reset_connection_state() def _async_reset_connection_state(self) -> None: """Reset the connection state after a disconnect.""" self._encryption_key = None self._decryption_key = None self._notifications = set() self._broadcast_notifications = set() self._restore_pending = False self._fetched_gsn_this_session = False self._had_notify_this_session = False async def _ensure_connected(self, attempts: int | None = None) -> bool | None: """Ensure that we are connected to the accessory. Returns True if we had to make the connection, returns False if we were already connected or shutdown. """ assert self._config_lock.locked(), "_config_lock Should be locked" if self._shutdown or (self.client and self.client.is_connected): return False async with self._connection_lock: # Check again while holding the lock if self._shutdown or (self.client and self.client.is_connected): return False if not self.device and ( discovery := await self.controller.async_find(self.id, DISCOVER_TIMEOUT) ): self.device = discovery.device self.ble_advertisement = discovery.ble_advertisement self.description = discovery.description elif not self.device: raise AccessoryNotFoundError(f"{self.name}: Could not find {self.id}") self.client = await establish_connection( self.device, self.name, self._async_disconnected, use_services_cache=True, ble_device_callback=lambda: self.device, max_attempts=attempts, ) return True async def _async_start_notify(self, iid: int) -> None: assert self._operation_lock.locked(), "_operation_lock should be locked" char = self.accessories.aid(BLE_AID).characteristics.iid(iid) # Find the GATT Characteristic object for this iid endpoint = await self.client.get_characteristic( char.service.type, char.type, iid ) # We only want to allow one in flight read # and one pending read at a time since there # may be a notify storm and the read it always # going to give us the latest value anyways max_callback_enforcer = asyncio.Semaphore(2) async def _async_callback() -> None: if max_callback_enforcer.locked(): # Already one being read now, and one pending return async with max_callback_enforcer: if not self.client or not self.client.is_connected: # Client disconnected return async with self._operation_lock: logger.debug("%s: Retrieving event for iid: %s", self.name, iid) await self._get_characteristics_while_connected( [char], notify_listeners=True, skip_characteristics=IGNORE_READ_CHARACTERISTICS, ) # The GSN value is always the same per session. # It will never increment until the session is closed. # # If we haven't fetched the GSN this session, do it now. if not self._fetched_gsn_this_session and ( protocol_param := await self._get_all_protocol_params() ): self._update_state_num(protocol_param.state_number) if not self._had_notify_this_session: self._had_notify_this_session = True # 7.4.1.8 Global State Number (GSN) # The GSN is only increased once per session # so we only want to increase it if we have # not already done so this session. # # The fetched GSN from _get_all_protocol_params # is NOT increased when we get a notify so we need # to do it here. # new_state_num = self.description.state_num + 1 if new_state_num >= MAX_GSN: new_state_num = 1 # GSN rolled over which invalidates the broadcast # encryption key. We need to re-fetch it. await self._async_set_broadcast_encryption_key() self._update_state_num(new_state_num) def _callback(id: int, data: bytes) -> None: logger.debug("%s: Received event for iid=%s: %s", self.name, iid, data) if data != b"": # We should only poll on empty messages, otherwise we may poll # the device every second on DBUS systems. return if max_callback_enforcer.locked(): # Already one being read now, and one pending return async_create_task(_async_callback()) logger.debug("%s: Subscribing to gatt notify for iid: %s", self.name, iid) await self.client.start_notify(endpoint, _callback) self._notifications.add(iid) async def _async_pair_verify(self) -> None: async with self._ble_request_lock: session_id, derive = await drive_pairing_state_machine( self.client, CharacteristicsTypes.PAIR_VERIFY, get_session_keys(self.pairing_data, self._session_id, self._derive), ) self._encryption_key = EncryptionKey( derive(b"Control-Salt", b"Control-Write-Encryption-Key") ) self._decryption_key = DecryptionKey( derive(b"Control-Salt", b"Control-Read-Encryption-Key") ) # Used for session resume self._session_id = session_id self._derive = derive def _process_disconnected_events(self) -> None: async_create_task(self._async_process_disconnected_events()) async def _async_process_disconnected_events(self) -> None: """Handle disconnected events seen from the advertisement.""" if not self._tried_to_connect_once: # We never tried connected to the accessory, so we don't need to # process the disconnected events logger.debug( "%s: Skipping disconnected events because we have not yet connected.", self.name, ) return if self._disconnected_events_lock.locked(): # Already processing disconnected events return async with self._disconnected_events_lock: logger.debug( "%s: Polling subscriptions for changes during disconnection; rssi=%s", self.name, self.rssi, ) try: protocol_param = await self._process_disconnected_events_with_retry() except ( AccessoryDisconnectedError, *BLEAK_EXCEPTIONS, AccessoryNotFoundError, ) as exc: logger.exception( "%s: Failed to fetch disconnected events: %s; rssi=%s", self.name, exc, self.rssi, ) return if protocol_param: self._update_state_num(protocol_param.state_number) @operation_lock @retry_bluetooth_connection_error() @disconnect_on_missing_services @restore_connection_and_resume async def _process_disconnected_events_with_retry( self, ) -> ProtocolParams | None: accessory_chars = self.accessories.aid(BLE_AID).characteristics protocol_param = await self._get_all_protocol_params() chars_to_update = [] for _, iid in self.subscriptions: char = accessory_chars.iid(iid) if char.broadcast_events or char.disconnected_events: chars_to_update.append(char) if chars_to_update: await self._get_characteristics_while_connected( chars_to_update, notify_listeners=True, ) return protocol_param def _async_notification(self, data: HomeKitEncryptedNotification) -> None: """Receive a notification from the accessory.""" if not self._broadcast_decryption_key: logger.debug( "%s: Received notification before session is setup, " "falling back processing as disconnected event: %s", self.name, data, ) self._process_disconnected_events() return if not self.description: logger.error( "%s: Received encrypted notification before advertisement.", self.name, ) return start_state_num = self.description.state_num # Usually we increment by one, but sometimes we get multiple with the same # state number so the first pass is optimistic to reduce the number of # of decrypts we do. for state_num in ( start_state_num + 1, # This is the state number we are expecting (next one) start_state_num, # This is the old state number (already used) *range( start_state_num + 2, start_state_num + 100 ), # If bluetooth was offline for a while we try a few more ): logger.debug( "%s: Trying state_num %s for encrypted notification: %s", self.name, state_num, data, ) decrypted = self._broadcast_decryption_key.decrypt( data.encrypted_payload, state_num, data.advertising_identifier, ) if decrypted is None: continue if state_num == start_state_num: logger.debug( "%s: Encrypted notification with stale state_num %s ignored: %s", self.name, state_num, data, ) return gsn = int.from_bytes(decrypted[0:2], "little") if gsn != state_num: logger.debug( "%s: GSN mismatch, expected: %s, got: %s", self.name, state_num, gsn, ) return iid = int.from_bytes(decrypted[2:4], "little") value = decrypted[4:12] logger.debug( "%s: Received notification: encrypted = %s - decrypted = %s - gsn=%s - iid=%s - value=%s", self.name, data.encrypted_payload, decrypted, gsn, iid, value, ) # We had a successful decrypt, so we can update the state_num self.description.state_num = gsn char = self.accessories.aid(BLE_AID).characteristics.iid(iid) results = {(BLE_AID, iid): {"value": from_bytes(char, value)}} logger.debug("%s: Received notification: results = %s", self.name, results) self._callback_listeners(results) return logger.warning( "%s: Received notification but could not decrypt, falling back processing as disconnected event: %s", self.name, data, ) self._process_disconnected_events() def _async_get_service_signature_char(self) -> Characteristic | None: """Get the service signature characteristic.""" info = self.accessories.aid(BLE_AID).services.first( service_type=ServicesTypes.PROTOCOL_INFORMATION ) if not info: logger.debug("%s: No signature service found", self.name) return None if not info.has(CharacteristicsTypes.SERVICE_SIGNATURE): logger.debug( "%s: No signature characteristic found, " "accessory may not implement encrypted notifications", self.name, ) return None return info[CharacteristicsTypes.SERVICE_SIGNATURE] async def _async_set_broadcast_encryption_key(self) -> None: """Get the broadcast key for the accessory.""" assert self._operation_lock.locked(), "_operation_lock should be locked" logger.debug("%s: Setting broadcast encryption key", self.name) if self._ble_request_lock.locked(): logger.debug( "%s: Waiting ble request lock to set broadcast encryption key", self.name, ) async with self._ble_request_lock: hap_char = self._async_get_service_signature_char() if not hap_char: return service_iid = hap_char.service.iid logger.debug( "%s: Setting broadcast key for service_iid: %s", self.name, service_iid, ) try: await self._async_request_under_lock( OpCode.PROTOCOL_CONFIG, hap_char, GENERATE_BROADCAST_KEY_PAYLOAD, iid=service_iid, ) except PDUStatusError: logger.exception( "%s: Failed to set broadcast key, try un-paring and re-pairing the accessory.", self.name, ) long_term_pub_key_hex: str = self.pairing_data["iOSDeviceLTPK"] long_term_pub_key_bytes = bytes.fromhex(long_term_pub_key_hex) broadcast_key_bytes = self._derive( long_term_pub_key_bytes, b"Broadcast-Encryption-Key" ) self._broadcast_decryption_key = BroadcastDecryptionKey(broadcast_key_bytes) if self._accessories_state and self.broadcast_key != broadcast_key_bytes: self._accessories_state.broadcast_key = broadcast_key_bytes self._callback_and_save_config_changed(self.config_num) async def _read_signature( self, char: BleakGATTCharacteristic, op_code: OpCode, iid: int, tlv_struct: ServiceTLV | CharacteristicTLV, ) -> dict[str, Any]: """Read the signature for the given characteristic.""" tid = random.randint(1, 254) for data in encode_pdu(op_code, tid, iid): await self.client.write_gatt_char( char, data, "write-without-response" not in char.properties, ) payload = await self.client.read_gatt_char(char) status, _, signature = decode_pdu(tid, payload) if status != PDUStatus.SUCCESS: return {} return tlv_struct.decode(signature).to_dict() async def _async_fetch_gatt_database(self) -> Accessories: logger.debug("%s: Fetching GATT database; rssi=%s", self.name, self.rssi) accessory = Accessory(1) # Never use the cache when fetching the GATT database services_to_link: list[tuple[Service, list[int]]] = [] services = self.client.services for service in services: ble_service_char = service.get_characteristic(SERVICE_INSTANCE_ID_UUID) if not ble_service_char: logger.debug( "%s: Skipping service without service instance id: %s", self.name, service, ) continue service_iid_bytes = await self.client.read_gatt_char( ble_service_char.handle ) service_iid = int.from_bytes(service_iid_bytes, "little") logger.debug( "%s: Service %s iid: %s (decoded as %s)", self.name, service.uuid, service_iid_bytes, service_iid, ) s = accessory.add_service(normalize_uuid(service.uuid), iid=service_iid) service_signature_char = service.get_characteristic(SERVICE_SIGNATURE_UUID) if service_signature_char: decoded_service = await self._read_signature( service_signature_char, OpCode.SERV_SIG_READ, service_iid, ServiceTLV, ) if (linked_iids := decoded_service.get("linked")) and isinstance( linked_iids, list ): services_to_link.append((s, [iid for iid in linked_iids if iid])) logger.debug( "%s: service: %s decoded: %s", self.name, service, decoded_service ) for char in service.characteristics: normalized_uuid = normalize_uuid(char.uuid) if normalized_uuid == SERVICE_INSTANCE_ID: continue iid = await self.client.get_characteristic_iid(char) if not iid: # None or 0 are both invalid logger.debug("%s: No iid for %s", self.name, char.uuid) continue decoded = await self._read_signature( char, OpCode.CHAR_SIG_READ, iid, CharacteristicTLV ) if normalized_uuid == CharacteristicsTypes.IDENTIFY: # Workaround for older eve v1 devices which has a broken identify characteristic # that presents identify as data. decoded["format"] = "bool" hap_char = s.add_char(normalized_uuid, iid=iid) logger.debug("%s: char: %s decoded: %s", self.name, char, decoded) assert hap_char.iid == iid, "iid should be set" hap_char.handle = char.handle # Some vendor characteristics have no perms # See https://github.com/home-assistant/core/issues/91686 hap_char.perms = decoded.get("perms", []) # Some vendor characteristics have no format # See https://github.com/home-assistant/core/issues/76104 if "format" in decoded: hap_char.format = decoded["format"] if "minStep" in decoded: hap_char.minStep = decoded["minStep"] if "minValue" in decoded: hap_char.minValue = decoded["minValue"] if "maxValue" in decoded: hap_char.maxValue = decoded["maxValue"] if "disconnected_events" in decoded: hap_char.disconnected_events = decoded["disconnected_events"] if "broadcast_events" in decoded: hap_char.broadcast_events = decoded["broadcast_events"] # Link services after we are done since we need to have all services # in the accessory to link them. for service, linked in services_to_link: for iid in linked: service.add_linked_service(accessory.services.iid(iid)) accessories = Accessories() accessories.add_accessory(accessory) logger.debug("%s: Completed fetching GATT database", self.name) return accessories @operation_lock async def close_after_operation(self) -> None: """Close the client after an operation.""" await self.close() async def close(self) -> None: async with self._connection_lock: await self._close_while_locked() async def _close_while_locked(self) -> None: if not self.client or not self.client.is_connected: return try: await self.client.disconnect() except BleakError: logger.debug( "%s: Failed to close connection, client may have already closed it; rssi=%s", self.name, self.rssi, ) self.client = None self._async_reset_connection_state() logger.debug( "%s: Connection closed from close call; rssi=%s", self.name, self.rssi ) @operation_lock @retry_bluetooth_connection_error() @disconnect_on_missing_services @restore_connection_and_resume async def list_accessories_and_characteristics(self) -> list[dict[str, Any]]: return self.accessories.serialize() async def get_primary_name(self) -> str: """Return the primary name of the device. This overrides the default implementation to get name from the advertisement if the accessories are not available which happens frequently with the ESPHome proxies because they cannot currently connect right away again after disconnecting. We want to avoid raising here if possible since this is called right after pairing and if we raise the pairing will fail but the device will still be paired and the user will have to un-pair it manually with a factory reset. """ if not self.accessories and self.description: self._accessories_state = AccessoriesState( Accessories(), -1, self.broadcast_key, self.state_num ) return self.description.name return await super().get_primary_name() async def _populate_char_values(self, config_changed: bool) -> None: """Populate the values of all characteristics.""" chars: list[Characteristic] = [] for service in self.accessories.aid(BLE_AID).services: if service.type in SKIP_SYNC_SERVICES: continue if ( not config_changed and service.type == ServicesTypes.ACCESSORY_INFORMATION and service.value(CharacteristicsTypes.NAME) ): continue for char in service.characteristics: if char.type in IGNORE_POLL_CHARACTERISTICS: continue if CharacteristicPermissions.paired_read not in char.perms: continue chars.append(char) if not chars: return protocol_params = await self._get_all_protocol_params() results = await self._get_characteristics_while_connected(chars) logger.debug("%s: Read %s", self.name, results) for char in chars: result = results.get((BLE_AID, char.iid)) if not result or "value" not in result: logger.debug( "%s: No value for %s/%s", self.name, char.service.type, char.type ) continue char.value = result["value"] if protocol_params: self.description.state_num = protocol_params.state_number async def _get_all_protocol_params(self) -> ProtocolParams | None: """Get the current protocol params number.""" assert self._operation_lock.locked(), "_operation_lock should be locked" hap_char = self._async_get_service_signature_char() if not hap_char: return service_iid = hap_char.service.iid try: resp = await self._async_request( OpCode.PROTOCOL_CONFIG, hap_char, GET_ALL_PARAMS_PAYLOAD, iid=service_iid, ) except PDUStatusError: logger.exception( "%s: Failed to get global state number.", self.name, ) return None response = dict(TLV.decode_bytes(resp)) protocol_params = ProtocolParams( state_number=int.from_bytes( response[ProtocolParamsTLV.GlobalStateNumber], "little" ), config_number=int.from_bytes( response[ProtocolParamsTLV.ConfigurationNumber], "little" ), advertising_id=response[ProtocolParamsTLV.AdvertisingId], broadcast_key=response.get(ProtocolParamsTLV.BroadcastKey), ) logger.debug( "%s: Fetched protocol params: gsn=%s, c#=%s", self.name, protocol_params.state_number, protocol_params.config_number, ) self._fetched_gsn_this_session = True return protocol_params async def async_populate_accessories_state( self, force_update: bool = False, attempts: int | None = None ) -> None: """Populate the state of all accessories. This method should try not to fetch all the accessories unless we know the config num is out of date. Callers should not get BleakError as they expect to trap AccessoryDisconnectedError. """ try: await self._async_populate_accessories_state(force_update, attempts) except BleakError as ex: raise AccessoryDisconnectedError( f"{self.name} connection failed: {ex}; rssi={self.rssi}" ) from ex @operation_lock @retry_bluetooth_connection_error() @disconnect_on_missing_services async def _async_populate_accessories_state( self, force_update: bool = False, attempts: int | None = None ) -> None: """Populate the state of all accessories under the lock.""" await self._populate_accessories_and_characteristics(force_update, attempts) if self._restore_pending: await self._async_restore_subscriptions() async def _populate_accessories_and_characteristics( self, force_update: bool = False, attempts: int | None = None ) -> None: was_locked = self._config_lock.locked() async with self._config_lock: if self._shutdown: return try: made_connection = await self._ensure_connected(attempts) finally: # Only set _tried_to_connect_once after the connection # attempt is complete so we don't try to process disconnected # events while we are still trying to connect. self._tried_to_connect_once = True logger.debug( "%s: Populating accessories and characteristics: made_connection=%s restore_pending=%s", self.name, made_connection, self._restore_pending, ) self._restore_pending |= made_connection if was_locked and not force_update: # No need to do it twice if we already have the data # and we are not forcing an update return update_values = force_update config_changed = False if self.description: config_changed = self.config_num != self.description.config_num if not self.accessories or config_changed: logger.debug( "%s: Fetching gatt database because, cached_config_num: %s, adv config_num: %s", self.name, self.config_num, self.description.config_num, ) accessories = await self._async_fetch_gatt_database() new_config_num = self.description.config_num if self.description else 0 self._accessories_state = AccessoriesState( accessories, new_config_num, self.broadcast_key ) update_values = True if not self._encryption_key: await self._async_pair_verify() if update_values: await self._populate_char_values(config_changed) self._update_accessories_state_cache() if config_changed: self._callback_and_save_config_changed(self.config_num) async def _async_subscribe_broadcast_events( self, subscriptions: list[tuple[int, int]] ) -> None: """Subscribe to broadcast events.""" accessory_chars = self.accessories.aid(BLE_AID).characteristics to_subscribe: list[Characteristic] = [] for _, iid in subscriptions: hap_char = accessory_chars.iid(iid) if ( not hap_char or not hap_char.broadcast_events or iid in self._broadcast_notifications ): continue to_subscribe.append(hap_char) if not to_subscribe: return async with self._ble_request_lock: for hap_char in to_subscribe: iid = hap_char.iid if iid in self._broadcast_notifications: continue # check again with the lock logger.debug( "%s: Subscribing to broadcast notify for iid: %s", self.name, iid ) try: await self._async_request_under_lock( OpCode.CHAR_CONFIG, hap_char, ENABLE_BROADCAST_PAYLOAD ) except PDUStatusError: logger.debug( "%s: Failed to subscribe to broadcast events for %s", self.name, hap_char, ) continue self._broadcast_notifications.add(iid) async def _async_restore_subscriptions(self) -> None: """Restore subscriptions and setup notifications after after connecting.""" if not self._restore_pending or not self.client or not self.client.is_connected: return if not self.subscriptions: logger.debug("%s: No subscriptions to restore", self.name) self._restore_pending = False return assert self._operation_lock.locked(), "_operation_lock should be locked" await self._async_set_broadcast_encryption_key() subscriptions = list(self.subscriptions) logger.debug( "%s: Connected, resuming subscriptions: %s; rssi=%s", self.name, subscriptions, self.rssi, ) await self._async_subscribe_broadcast_events(subscriptions) self._restore_pending = False # After we have restored subscriptions, we need to read # the state number again to make sure we are in sync protocol_param = await self._get_all_protocol_params() if protocol_param: self._update_state_num(protocol_param.state_number) self._async_schedule_start_notify_subscriptions() @operation_lock async def _async_start_notify_subscriptions(self) -> None: """Start notifications for the given subscriptions. If we have an error or are disconnected we do not want to retry, as it usually means the accessory indented to disconnect us and we are now using disconnected events or encrypted broadcasts to get updates. """ if not self.client or not self.client.is_connected: return subscriptions = list(self.subscriptions) for _, iid in subscriptions: if iid in self._notifications: continue # The iid will not be in in self._notifications until # the _async_start_notify call returns. async with self._subscription_lock: if iid in self._notifications or not self.client.is_connected: continue try: await self._async_start_notify(iid) except BLEAK_EXCEPTIONS as ex: # Likely disconnected before we could start notifications # we will get disconnected events instead. logger.debug( "%s: Could not start notify for %s: %s; rssi=%s", self.name, iid, ex, self.rssi, ) @operation_lock @retry_bluetooth_connection_error() @disconnect_on_missing_services async def _process_config_changed(self, config_num: int) -> None: """Process a config change. This method is called when the config num changes. """ await self._populate_accessories_and_characteristics() if self._restore_pending: await self._async_restore_subscriptions() @operation_lock @retry_bluetooth_connection_error() @disconnect_on_missing_services @restore_connection_and_resume async def list_pairings(self): request_tlv = TLV.encode_list( [(TLV.kTLVType_State, TLV.M1), (TLV.kTLVType_Method, TLV.ListPairings)] ) request_tlv = TLV.encode_list( [ (TLV.kTLVHAPParamParamReturnResponse, bytearray(b"\x01")), (TLV.kTLVHAPParamValue, request_tlv), ] ) info = self.accessories.aid(BLE_AID).services.first( service_type=ServicesTypes.PAIRING ) char = info[CharacteristicsTypes.PAIRING_PAIRINGS] resp = await self._async_request(OpCode.CHAR_WRITE, char, request_tlv) response = dict(TLV.decode_bytes(resp)) resp = TLV.decode_bytes(response[1]) tmp = [] r = {} for d in resp[1:]: if d[0] == TLV.kTLVType_Identifier: r = {} tmp.append(r) r["pairingId"] = d[1].decode() if d[0] == TLV.kTLVType_PublicKey: r["publicKey"] = d[1].hex() if d[0] == TLV.kTLVType_Permissions: controller_type = "regular" if d[1] == b"\x01": controller_type = "admin" r["permissions"] = int.from_bytes(d[1], byteorder="little") r["controllerType"] = controller_type return tmp @retry_bluetooth_connection_error() @disconnect_on_missing_services async def get_characteristics( self, characteristics: Iterable[tuple[int, int]], ) -> dict[tuple[int, int], dict[str, Any]]: return await self._get_characteristics_without_retry(characteristics) def _sort_characteristics_by_fetch_order( self, characteristics: list[Characteristic] ) -> list[Characteristic]: """Sort characteristics by fetch order. Characteristics are sorted by what we expect to be the most useful order to fetch them in. This is based on the following assumptions: The on/off state of the accessory is the most important characteristic to fetch. Color/Brightness/Color Temperature/etc are the next most important characteristics to fetch. The remaining characteristics are less important and can be fetched in any order except for ones that are unlikely to change frequently like the name, version, and thread status so they are fetched last. """ # If we ever fetch which service is the primary we should # also sort by that. return sorted( characteristics, key=lambda char: CHAR_FETCH_PRIORITY.get( char.type, 10 if char.broadcast_events or char.disconnected_events else 0, ), reverse=True, ) @operation_lock @restore_connection_and_resume async def _get_characteristics_without_retry( self, characteristics: Iterable[tuple[int, int]], notify_listeners: bool = False, ) -> dict[tuple[int, int], dict[str, Any]]: accessory_chars = self.accessories.aid(BLE_AID).characteristics return await self._get_characteristics_while_connected( [accessory_chars.iid(iid) for _, iid in characteristics], notify_listeners ) async def _get_characteristics_while_connected( self, unordered_characteristics: list[Characteristic], notify_listeners: bool = False, skip_characteristics: set[str] | None = None, ) -> dict[tuple[int, int], dict[str, Any]]: assert self._operation_lock.locked(), "_operation_lock should be locked" characteristics = self._sort_characteristics_by_fetch_order( unordered_characteristics ) if skip_characteristics is None: skip_characteristics = IGNORE_POLL_CHARACTERISTICS debug_enabled = logger.isEnabledFor(logging.DEBUG) if debug_enabled: logger.debug( "%s: Reading characteristics with rssi: %s", self.name, self.rssi, ) results = {} async with self._ble_request_lock: for char in characteristics: if char.type in skip_characteristics: logger.debug( "%s: Ignoring characteristic %s", self.name, char.iid, ) continue if debug_enabled: logger.debug( "%s: Reading characteristic %s (%s) with iid %s", self.name, char.description, char.type, char.iid, ) try: data = await self._async_request_under_lock(OpCode.CHAR_READ, char) except PDUStatusError as ex: # For the apple defines ones we know about we can avoid triggering # this state, but we do not know all the vendor custom chars so # we need to skip in this case. if ex.status == PDUStatus.INVALID_REQUEST: logger.debug( "%s: Reading characteristic %s (%s) with iid %s resulted in an invalid request (skipped)", self.name, char.description, char.type, char.iid, ) continue logger.exception( "%s: Reading characteristic %s (%s) with iid %s resulted in an error: %s", self.name, char.description, char.type, char.iid, ex, ) continue decoded = dict(TLV.decode_bytes(data))[1] logger.debug( "%s: Read characteristic %s (%s) with iid %s got data, expected format is %s: data=%s decoded=%s", self.name, char.description, char.type, char.iid, char.format, data, decoded, ) try: value = from_bytes(char, decoded) except struct.error as ex: logger.debug( "%s: Failed to decode characteristic %s (%s) with iid %s from %s: %s", self.name, char.description, char.type, char.iid, decoded, ex, ) continue result_key = (BLE_AID, char.iid) result_value = {"value": value} results[result_key] = result_value if notify_listeners: # Since it can take a while to read all the characteristics # we want to notify the listeners as soon as we have the # value for each characteristic. single_results = {result_key: result_value} self._callback_listeners(single_results) return results @operation_lock @retry_bluetooth_connection_error() @disconnect_on_missing_services @restore_connection_and_resume async def put_characteristics( self, characteristics: Iterable[tuple[int, int, Any]] ) -> dict[tuple[int, int], dict[str, Any]]: results: dict[tuple[int, int], Any] = {} logger.debug( "%s: Writing characteristics: %s; rssi=%s", self.name, characteristics, self.rssi, ) accessory_chars = self.accessories.aid(BLE_AID).characteristics async with self._ble_request_lock: for aid, iid, value in characteristics: char = accessory_chars.iid(iid) result_key = (aid, iid) logger.debug( "%s: Writing characteristics: iid=%s value=%s", self.name, iid, value, ) result = {} if CharacteristicPermissions.timed_write in char.perms: payload_inner = TLV.encode_list( [ (HAP_TLV.kTLVHAPParamValue, to_bytes(char, value)), (HAP_TLV.kTLVHAPParamTTL, b"\x1e"), # 3.0s ] ) payload = (len(payload_inner)).to_bytes( length=2, byteorder="little" ) + payload_inner await self._async_request_under_lock( OpCode.CHAR_TIMED_WRITE, char, payload ) await self._async_request_under_lock(OpCode.CHAR_EXEC_WRITE, char) elif CharacteristicPermissions.paired_write in char.perms: payload = TLV.encode_list( [(HAP_TLV.kTLVHAPParamValue, to_bytes(char, value))] ) await self._async_request_under_lock( OpCode.CHAR_WRITE, char, payload ) else: result = { "status": HapStatusCode.CANT_WRITE_READ_ONLY, "description": HapStatusCode.CANT_WRITE_READ_ONLY.description, } # results only set on failure, no status is success if not result: if CharacteristicPermissions.paired_read in char.perms: self._callback_listeners({result_key: {"value": value}}) else: results[result_key] = result return results @operation_lock @retry_bluetooth_connection_error() @force_fresh_connection @restore_connection_and_resume async def thread_provision( self, dataset: str, ) -> None: """ Provision a device with Thread network credentials. The credentials should be provided in the meshcop/thread operational dataset format. """ thread_service = self.accessories.aid(BLE_AID).services.first( service_type=ServicesTypes.THREAD_TRANSPORT ) thread_control = thread_service[CharacteristicsTypes.THREAD_CONTROL_POINT] inner_request_tlv = TLV.encode_list( [ (1, b"\x03"), ] ) request_tlv = TLV.encode_list( [ (TLV.kTLVHAPParamParamReturnResponse, bytearray(b"\x01")), (TLV.kTLVHAPParamValue, inner_request_tlv), ] ) resp = await self._async_request(OpCode.CHAR_WRITE, thread_control, request_tlv) logger.debug("resp=%r", resp) decoded = Meshcop.decode(bytes.fromhex(dataset)) thread_tlv = TLV.encode_list( [ (1, decoded.networkname.encode("utf-8")), (2, decoded.channel.to_bytes(1, byteorder="little")), (3, decoded.panid.to_bytes(2, byteorder="little")), (4, decoded.extpanid), (5, decoded.networkkey), ] ) unknown = 1 inner_request_tlv = TLV.encode_list( [ # TLV 1 is some sort of write/provision OpCode (1, b"\x01"), # TLV 2 contains the Thread network details (2, thread_tlv), # TLV 3 seems to be a bitfield or identifier; iOS sends 0 & Android sends 1 # 1 has worked in testing (3, unknown.to_bytes(1, byteorder="little")), ] ) request_tlv = TLV.encode_list( [ (TLV.kTLVHAPParamParamReturnResponse, bytearray(b"\x01")), (TLV.kTLVHAPParamValue, inner_request_tlv), ] ) try: resp = await self._async_request( OpCode.CHAR_WRITE, thread_control, request_tlv ) # we shouldn't get a response logger.debug("Thread provision returned a success response: %r", resp) except Exception as e: # this is the expected code flow logger.debug( "Thread provision returned error (%r), this might not indicate a failure so ignoring.", e, ) await self.shutdown() async def subscribe(self, characteristics: Iterable[tuple[int, int]]) -> None: """Subscribe to characteristics.""" new_chars = await super().subscribe(characteristics) if not new_chars or not self.client or not self.client.is_connected: # Don't force a new connection if we are not already # connected as we will get disconnected events. return # We do not want to block setup of the accessory so we # do not wait for the result as any failures will be # handled by the retry logic or fallback to disconnected # events. async_create_task(self._async_subscribe(new_chars)) self._async_schedule_start_notify_subscriptions() @operation_lock @retry_bluetooth_connection_error() @disconnect_on_missing_services @restore_connection_and_resume async def _async_subscribe(self, new_chars: Iterable[tuple[int, int]]) -> None: """Subscribe to new characteristics.""" logger.debug("%s: subscribing to %s", self.name, new_chars) await self._populate_accessories_and_characteristics() if not self._broadcast_decryption_key: await self._async_set_broadcast_encryption_key() await self._async_subscribe_broadcast_events(new_chars) def _async_schedule_start_notify_subscriptions(self) -> None: """Schedule start notify subscriptions.""" if self._start_notify_timer: self._start_notify_timer.cancel() self._start_notify_timer = None def _async_start_notify_subscriptions() -> None: """Start notify subscriptions.""" self._start_notify_timer = None async_create_task(self._async_start_notify_subscriptions()) loop = asyncio.get_running_loop() self._start_notify_timer = loop.call_later( START_NOTIFY_DEBOUNCE, _async_start_notify_subscriptions ) async def unsubscribe(self, characteristics: Iterable[tuple[int, int]]) -> None: pass async def identify(self): info = self.accessories.aid(BLE_AID).services.first( service_type=ServicesTypes.ACCESSORY_INFORMATION ) char = info[CharacteristicsTypes.IDENTIFY] await self.put_characteristics( [ (BLE_AID, char.iid, True), ] ) @operation_lock @retry_bluetooth_connection_error() @disconnect_on_missing_services @restore_connection_and_resume async def add_pairing( self, additional_controller_pairing_identifier, ios_device_ltpk, permissions ): if permissions == "User": permissions = TLV.kTLVType_Permission_RegularUser elif permissions == "Admin": permissions = TLV.kTLVType_Permission_AdminUser else: raise RuntimeError(f"{self.name} Unknown permission: {permissions}") request_tlv = TLV.encode_list( [ (TLV.kTLVType_State, TLV.M1), (TLV.kTLVType_Method, TLV.AddPairing), ( TLV.kTLVType_Identifier, additional_controller_pairing_identifier.encode(), ), (TLV.kTLVType_PublicKey, bytes.fromhex(ios_device_ltpk)), (TLV.kTLVType_Permissions, permissions), ] ) request_tlv = TLV.encode_list( [ (TLV.kTLVHAPParamParamReturnResponse, bytearray(b"\x01")), (TLV.kTLVHAPParamValue, request_tlv), ] ) info = self.accessories.aid(BLE_AID).services.first( service_type=ServicesTypes.PAIRING ) char = info[CharacteristicsTypes.PAIRING_PAIRINGS] resp = await self._async_request(OpCode.CHAR_WRITE, char, request_tlv) response = dict(TLV.decode_bytes(resp)) data = dict(TLV.decode_bytes(response[1])) if data.get(TLV.kTLVType_State, TLV.M2) != TLV.M2: raise InvalidError( f"{self.name}: Unexpected state after removing pairing request" ) if TLV.kTLVType_Error in data: if data[TLV.kTLVType_Error] == TLV.kTLVError_Authentication: raise AuthenticationError( f"{self.name}: Add pairing failed: insufficient access" ) raise UnknownError(f"{self.name}: Add pairing failed: unknown error") @operation_lock @retry_bluetooth_connection_error(attempts=10) @disconnect_on_missing_services @restore_connection_and_resume async def remove_pairing(self, pairingId: str) -> bool: request_tlv = TLV.encode_list( [ (TLV.kTLVType_State, TLV.M1), (TLV.kTLVType_Method, TLV.RemovePairing), (TLV.kTLVType_Identifier, pairingId.encode("utf-8")), ] ) request_tlv = TLV.encode_list( [ (TLV.kTLVHAPParamParamReturnResponse, bytearray(b"\x01")), (TLV.kTLVHAPParamValue, request_tlv), ] ) info = self.accessories.aid(BLE_AID).services.first( service_type=ServicesTypes.PAIRING ) char = info[CharacteristicsTypes.PAIRING_PAIRINGS] resp = await self._async_request(OpCode.CHAR_WRITE, char, request_tlv) response = dict(TLV.decode_bytes(resp)) data = dict(TLV.decode_bytes(response[1])) if data.get(TLV.kTLVType_State, TLV.M2) != TLV.M2: raise InvalidError( f"{self.name}: Unexpected state after removing pairing request" ) if TLV.kTLVType_Error in data: if data[TLV.kTLVType_Error] == TLV.kTLVError_Authentication: raise AuthenticationError( f"{self.name}: Remove pairing failed: insufficient access" ) raise UnknownError(f"{self.name}: Remove pairing failed: unknown error") await self._shutdown_if_primary_pairing_removed(pairingId) return True async def image(self, accessory: int, width: int, height: int) -> None: """Bluetooth devices don't return images.""" return None