| # Copyright 2021-2022 Google LLC |
| # |
| # 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 |
| # |
| # https://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. |
| |
| # ----------------------------------------------------------------------------- |
| # Imports |
| # ----------------------------------------------------------------------------- |
| import json |
| import asyncio |
| import logging |
| |
| from .hci import * |
| from .host import Host |
| from .gatt import * |
| from .gap import GenericAccessService |
| from .core import AdvertisingData, BT_CENTRAL_ROLE, BT_PERIPHERAL_ROLE |
| from .utils import AsyncRunner, CompositeEventEmitter, setup_event_forwarding, composite_listener |
| from . import gatt_client |
| from . import gatt_server |
| from . import smp |
| from . import sdp |
| from . import l2cap |
| from . import keys |
| |
| # ----------------------------------------------------------------------------- |
| # Logging |
| # ----------------------------------------------------------------------------- |
| logger = logging.getLogger(__name__) |
| |
| # ----------------------------------------------------------------------------- |
| # Constants |
| # ----------------------------------------------------------------------------- |
| DEVICE_DEFAULT_ADDRESS = '00:00:00:00:00:00' |
| DEVICE_DEFAULT_ADVERTISING_INTERVAL = 1000 # ms |
| DEVICE_DEFAULT_ADVERTISING_DATA = '' |
| DEVICE_DEFAULT_NAME = 'Bumble' |
| DEVICE_DEFAULT_INQUIRY_LENGTH = 8 # 10.24 seconds |
| DEVICE_DEFAULT_CLASS_OF_DEVICE = 0 |
| DEVICE_DEFAULT_SCAN_RESPONSE_DATA = b'' |
| DEVICE_DEFAULT_DATA_LENGTH = (27, 328, 27, 328) |
| DEVICE_DEFAULT_SCAN_INTERVAL = 60 # ms |
| DEVICE_DEFAULT_SCAN_WINDOW = 60 # ms |
| DEVICE_MIN_SCAN_INTERVAL = 25 |
| DEVICE_MAX_SCAN_INTERVAL = 10240 |
| DEVICE_MIN_SCAN_WINDOW = 25 |
| DEVICE_MAX_SCAN_WINDOW = 10240 |
| |
| # ----------------------------------------------------------------------------- |
| # Classes |
| # ----------------------------------------------------------------------------- |
| |
| |
| # ----------------------------------------------------------------------------- |
| class AdvertisementDataAccumulator: |
| def __init__(self): |
| self.advertising_data = AdvertisingData() |
| self.last_advertisement_type = None |
| self.connectable = False |
| self.flushable = False |
| |
| def update(self, data, advertisement_type): |
| if advertisement_type == HCI_LE_Advertising_Report_Event.SCAN_RSP: |
| if self.last_advertisement_type != HCI_LE_Advertising_Report_Event.SCAN_RSP: |
| self.advertising_data.append(data) |
| self.flushable = True |
| else: |
| self.advertising_data = AdvertisingData.from_bytes(data) |
| self.flushable = self.last_advertisement_type != HCI_LE_Advertising_Report_Event.SCAN_RSP |
| |
| if advertisement_type == HCI_LE_Advertising_Report_Event.ADV_IND or advertisement_type == HCI_LE_Advertising_Report_Event.ADV_DIRECT_IND: |
| self.connectable = True |
| elif advertisement_type == HCI_LE_Advertising_Report_Event.ADV_SCAN_IND or advertisement_type == HCI_LE_Advertising_Report_Event.ADV_NONCONN_IND: |
| self.connectable = False |
| |
| self.last_advertisement_type = advertisement_type |
| |
| |
| # ----------------------------------------------------------------------------- |
| class Peer: |
| def __init__(self, connection): |
| self.connection = connection |
| |
| # Create a GATT client for the connection |
| self.gatt_client = gatt_client.Client(connection) |
| connection.gatt_client = self.gatt_client |
| |
| @property |
| def services(self): |
| return self.gatt_client.services |
| |
| async def request_mtu(self, mtu): |
| return await self.gatt_client.request_mtu(mtu) |
| |
| async def discover_service(self, uuid): |
| return await self.gatt_client.discover_service(uuid) |
| |
| async def discover_services(self, uuids = []): |
| return await self.gatt_client.discover_services(uuids) |
| |
| async def discover_included_services(self, service): |
| return await self.gatt_client.discover_included_services(service) |
| |
| async def discover_characteristics(self, uuids = [], service = None): |
| return await self.gatt_client.discover_characteristics(uuids = uuids, service = service) |
| |
| async def discover_descriptors(self, characteristic = None, start_handle = None, end_handle = None): |
| return await self.gatt_client.discover_descriptors(characteristic, start_handle, end_handle) |
| |
| async def discover_attributes(self): |
| return await self.gatt_client.discover_attributes() |
| |
| async def subscribe(self, characteristic, subscriber=None): |
| return await self.gatt_client.subscribe(characteristic, subscriber) |
| |
| async def read_value(self, attribute): |
| return await self.gatt_client.read_value(attribute) |
| |
| async def write_value(self, attribute, value, with_response=False): |
| return await self.gatt_client.write_value(attribute, value, with_response) |
| |
| async def read_characteristics_by_uuid(self, uuid, service=None): |
| return await self.gatt_client.read_characteristics_by_uuid(uuid, service) |
| |
| def get_services_by_uuid(self, uuid): |
| return self.gatt_client.get_services_by_uuid(uuid) |
| |
| def get_characteristics_by_uuid(self, uuid, service = None): |
| return self.gatt_client.get_characteristics_by_uuid(uuid, service) |
| |
| # [Classic only] |
| async def request_name(self): |
| return await self.connection.request_remote_name() |
| |
| def __str__(self): |
| return f'{self.connection.peer_address} as {self.connection.role_name}' |
| |
| |
| # ----------------------------------------------------------------------------- |
| class Connection(CompositeEventEmitter): |
| @composite_listener |
| class Listener: |
| def on_disconnection(self, reason): |
| pass |
| |
| def on_connection_parameters_update(self): |
| pass |
| |
| def on_connection_parameters_update_failure(self, error): |
| pass |
| |
| def on_connection_phy_update(self): |
| pass |
| |
| def on_connection_phy_update_failure(self, error): |
| pass |
| |
| def on_connection_att_mtu_update(self): |
| pass |
| |
| def on_connection_encryption_change(self): |
| pass |
| |
| def on_connection_encryption_key_refresh(self): |
| pass |
| |
| def __init__(self, device, handle, transport, peer_address, peer_resolvable_address, role, parameters): |
| super().__init__() |
| self.device = device |
| self.handle = handle |
| self.transport = transport |
| self.peer_address = peer_address |
| self.peer_resolvable_address = peer_resolvable_address |
| self.peer_name = None # Classic only |
| self.role = role |
| self.parameters = parameters |
| self.encryption = 0 |
| self.authenticated = False |
| self.phy = ConnectionPHY(HCI_LE_1M_PHY, HCI_LE_1M_PHY) |
| self.att_mtu = ATT_DEFAULT_MTU |
| self.data_length = DEVICE_DEFAULT_DATA_LENGTH |
| self.gatt_client = None # Per-connection client |
| self.gatt_server = device.gatt_server # By default, use the device's shared server |
| |
| @property |
| def role_name(self): |
| return 'CENTRAL' if self.role == BT_CENTRAL_ROLE else 'PERIPHERAL' |
| |
| @property |
| def is_encrypted(self): |
| return self.encryption != 0 |
| |
| def send_l2cap_pdu(self, cid, pdu): |
| self.device.send_l2cap_pdu(self.handle, cid, pdu) |
| |
| def create_l2cap_connector(self, psm): |
| return self.device.create_l2cap_connector(self, psm) |
| |
| async def disconnect(self, reason = HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR): |
| return await self.device.disconnect(self, reason) |
| |
| async def pair(self): |
| return await self.device.pair(self) |
| |
| def request_pairing(self): |
| return self.device.request_pairing(self) |
| |
| # [Classic only] |
| async def authenticate(self): |
| return await self.device.authenticate(self) |
| |
| async def encrypt(self): |
| return await self.device.encrypt(self) |
| |
| async def update_parameters( |
| self, |
| conn_interval_min, |
| conn_interval_max, |
| conn_latency, |
| supervision_timeout |
| ): |
| return await self.device.update_connection_parameters( |
| self, |
| conn_interval_min, |
| conn_interval_max, |
| conn_latency, |
| supervision_timeout |
| ) |
| |
| # [Classic only] |
| async def request_remote_name(self): |
| return await self.device.request_remote_name(self) |
| |
| def __str__(self): |
| return f'Connection(handle=0x{self.handle:04X}, role={self.role_name}, address={self.peer_address})' |
| |
| |
| # ----------------------------------------------------------------------------- |
| class DeviceConfiguration: |
| def __init__(self): |
| # Setup defaults |
| self.name = DEVICE_DEFAULT_NAME |
| self.address = DEVICE_DEFAULT_ADDRESS |
| self.class_of_device = DEVICE_DEFAULT_CLASS_OF_DEVICE |
| self.scan_response_data = DEVICE_DEFAULT_SCAN_RESPONSE_DATA |
| self.advertising_interval_min = DEVICE_DEFAULT_ADVERTISING_INTERVAL |
| self.advertising_interval_max = DEVICE_DEFAULT_ADVERTISING_INTERVAL |
| self.le_enabled = True |
| # LE host enable 2nd parameter |
| self.le_simultaneous_enabled = True |
| self.classic_sc_enabled = True |
| self.classic_ssp_enabled = True |
| self.advertising_data = bytes( |
| AdvertisingData([(AdvertisingData.COMPLETE_LOCAL_NAME, bytes(self.name, 'utf-8'))]) |
| ) |
| self.irk = bytes(16) # This really must be changed for any level of security |
| self.keystore = None |
| |
| def load_from_dict(self, config): |
| # Load simple properties |
| self.name = config.get('name', self.name) |
| self.address = Address(config.get('address', self.address)) |
| self.class_of_device = config.get('class_of_device', self.class_of_device) |
| self.advertising_interval_min = config.get('advertising_interval', self.advertising_interval_min) |
| self.advertising_interval_max = self.advertising_interval_min |
| self.keystore = config.get('keystore') |
| self.le_enabled = config.get('le_enabled', self.le_enabled) |
| self.le_simultaneous_enabled = config.get('le_simultaneous_enabled', self.le_simultaneous_enabled) |
| self.classic_sc_enabled = config.get('classic_sc_enabled', self.classic_sc_enabled) |
| self.classic_ssp_enabled = config.get('classic_ssp_enabled', self.classic_ssp_enabled) |
| |
| # Load or synthesize an IRK |
| irk = config.get('irk') |
| if irk: |
| self.irk = bytes.fromhex(irk) |
| else: |
| # Construct an IRK from the address bytes |
| # NOTE: this is not secure, but will always give the same IRK for the same address |
| address_bytes = bytes(self.address) |
| self.irk = (address_bytes * 3)[:16] |
| |
| # Load advertising data |
| advertising_data = config.get('advertising_data') |
| if advertising_data: |
| self.advertising_data = bytes.fromhex(advertising_data) |
| |
| def load_from_file(self, filename): |
| with open(filename, 'r') as file: |
| self.load_from_dict(json.load(file)) |
| |
| # ----------------------------------------------------------------------------- |
| # Decorators used with the following Device class |
| # (we define them outside of the Device class, because defining decorators |
| # within a class requires unnecessarily complicated acrobatics) |
| # ----------------------------------------------------------------------------- |
| |
| # Decorator that converts the first argument from a connection handle to a connection |
| def with_connection_from_handle(function): |
| @functools.wraps(function) |
| def wrapper(self, connection_handle, *args, **kwargs): |
| if (connection := self.lookup_connection(connection_handle)) is None: |
| raise ValueError('no connection for handle') |
| return function(self, connection, *args, **kwargs) |
| return wrapper |
| |
| |
| # Decorator that converts the first argument from a bluetooth address to a connection |
| def with_connection_from_address(function): |
| @functools.wraps(function) |
| def wrapper(self, address, *args, **kwargs): |
| for connection in self.connections.values(): |
| if connection.peer_address == address: |
| return function(self, connection, *args, **kwargs) |
| raise ValueError('no connection for address') |
| return wrapper |
| |
| |
| # Decorator that adds a method to the list of event handlers for host events. |
| # This assumes that the method name starts with `on_` |
| def host_event_handler(function): |
| device_host_event_handlers.append(function.__name__[3:]) |
| return function |
| |
| |
| # List of host event handlers for the Device class. |
| # (we define this list outside the class, because referencing a class in method |
| # decorators is not straightforward) |
| device_host_event_handlers = [] |
| |
| |
| # ----------------------------------------------------------------------------- |
| class Device(CompositeEventEmitter): |
| |
| @composite_listener |
| class Listener: |
| def on_advertisement(self, address, data, rssi, advertisement_type): |
| pass |
| |
| def on_inquiry_result(self, address, class_of_device, data, rssi): |
| pass |
| |
| def on_connection(self, connection): |
| pass |
| |
| def on_connection_failure(self, error): |
| pass |
| |
| def on_characteristic_subscription(self, connection, characteristic, notify_enabled, indicate_enabled): |
| pass |
| |
| @classmethod |
| def with_hci(cls, name, address, hci_source, hci_sink): |
| ''' |
| Create a Device instance with a Host configured to communicate with a controller |
| through an HCI source/sink |
| ''' |
| host = Host(controller_source = hci_source, controller_sink = hci_sink) |
| return cls(name = name, address = address, host = host) |
| |
| @classmethod |
| def from_config_file(cls, filename): |
| config = DeviceConfiguration() |
| config.load_from_file(filename) |
| return cls(config=config) |
| |
| @classmethod |
| def from_config_file_with_hci(cls, filename, hci_source, hci_sink): |
| config = DeviceConfiguration() |
| config.load_from_file(filename) |
| host = Host(controller_source = hci_source, controller_sink = hci_sink) |
| return cls(config = config, host = host) |
| |
| def __init__(self, name = None, address = None, config = None, host = None, generic_access_service = True): |
| super().__init__() |
| |
| self._host = None |
| self.powered_on = False |
| self.advertising = False |
| self.auto_restart_advertising = False |
| self.command_timeout = 10 # seconds |
| self.gatt_server = gatt_server.Server(self) |
| self.sdp_server = sdp.Server(self) |
| self.l2cap_channel_manager = l2cap.ChannelManager() |
| self.advertisement_data = {} |
| self.scanning = False |
| self.discovering = False |
| self.connecting = False |
| self.disconnecting = False |
| self.connections = {} # Connections, by connection handle |
| self.classic_enabled = False |
| self.discoverable = False |
| self.connectable = False |
| self.inquiry_response = None |
| self.address_resolver = None |
| |
| # Use the initial config or a default |
| self.public_address = Address('00:00:00:00:00:00') |
| if config is None: |
| config = DeviceConfiguration() |
| self.name = config.name |
| self.random_address = config.address |
| self.class_of_device = config.class_of_device |
| self.scan_response_data = config.scan_response_data |
| self.advertising_data = config.advertising_data |
| self.advertising_interval_min = config.advertising_interval_min |
| self.advertising_interval_max = config.advertising_interval_max |
| self.keystore = keys.KeyStore.create_for_device(config) |
| self.irk = config.irk |
| self.le_enabled = config.le_enabled |
| self.le_simultaneous_enabled = config.le_simultaneous_enabled |
| self.classic_ssp_enabled = config.classic_ssp_enabled |
| self.classic_sc_enabled = config.classic_sc_enabled |
| |
| # If a name is passed, override the name from the config |
| if name: |
| self.name = name |
| |
| # If an address is passed, override the address from the config |
| if address: |
| if type(address) is str: |
| address = Address(address) |
| self.random_address = address |
| |
| # Setup SMP |
| # TODO: allow using a public address |
| self.smp_manager = smp.Manager(self, self.random_address) |
| |
| # Register the SDP server with the L2CAP Channel Manager |
| self.sdp_server.register(self.l2cap_channel_manager) |
| |
| # Add a GAP Service if requested |
| if generic_access_service: |
| self.gatt_server.add_service(GenericAccessService(self.name)) |
| |
| # Forward some events |
| setup_event_forwarding(self.gatt_server, self, 'characteristic_subscription') |
| |
| # Set the initial host |
| self.host = host |
| |
| @property |
| def host(self): |
| return self._host |
| |
| @host.setter |
| def host(self, host): |
| # Unsubscribe from events from the current host |
| if self._host: |
| for event_name in device_host_event_handlers: |
| self._host.remove_listener(event_name, getattr(self, f'on_{event_name}')) |
| |
| # Subscribe to events from the new host |
| if host: |
| for event_name in device_host_event_handlers: |
| host.on(event_name, getattr(self, f'on_{event_name}')) |
| |
| # Update the references to the new host |
| self._host = host |
| self.l2cap_channel_manager.host = host |
| |
| # Set providers for the new host |
| if host: |
| host.long_term_key_provider = self.get_long_term_key |
| host.link_key_provider = self.get_link_key |
| |
| @property |
| def sdp_service_records(self): |
| return self.sdp_server.service_records |
| |
| @sdp_service_records.setter |
| def sdp_service_records(self, service_records): |
| self.sdp_server.service_records = service_records |
| |
| def lookup_connection(self, connection_handle): |
| if connection := self.connections.get(connection_handle): |
| return connection |
| |
| def find_connection_by_bd_addr(self, bd_addr): |
| for connection in self.connections.values(): |
| if connection.peer_address == bd_addr: |
| return connection |
| |
| def register_l2cap_server(self, psm, server): |
| self.l2cap_channel_manager.register_server(psm, server) |
| |
| def create_l2cap_connector(self, connection, psm): |
| return lambda: self.l2cap_channel_manager.connect(connection, psm) |
| |
| def create_l2cap_registrar(self, psm): |
| return lambda handler: self.register_l2cap_server(psm, handler) |
| |
| def send_l2cap_pdu(self, connection_handle, cid, pdu): |
| self.host.send_l2cap_pdu(connection_handle, cid, pdu) |
| |
| async def send_command(self, command): |
| try: |
| return await asyncio.wait_for(self.host.send_command(command), self.command_timeout) |
| except asyncio.TimeoutError: |
| logger.warning('!!! Command timed out') |
| |
| async def power_on(self): |
| # Reset the controller |
| await self.host.reset() |
| |
| response = await self.send_command(HCI_Read_BD_ADDR_Command()) |
| if response.return_parameters.status == HCI_SUCCESS: |
| logger.debug(color(f'BD_ADDR: {response.return_parameters.bd_addr}', 'yellow')) |
| self.public_address = response.return_parameters.bd_addr |
| |
| |
| await self.send_command(HCI_Write_LE_Host_Support_Command( |
| le_supported_host = int(self.le_enabled), |
| simultaneous_le_host = int(self.le_simultaneous_enabled), |
| )) |
| if self.le_enabled: |
| # Set the controller address |
| await self.send_command(HCI_LE_Set_Random_Address_Command( |
| random_address = self.random_address |
| )) |
| |
| # Load the address resolving list |
| if self.keystore: |
| await self.send_command(HCI_LE_Clear_Resolving_List_Command()) |
| |
| resolving_keys = await self.keystore.get_resolving_keys() |
| for (irk, address) in resolving_keys: |
| await self.send_command( |
| HCI_LE_Add_Device_To_Resolving_List_Command( |
| peer_identity_address_type = address.address_type, |
| peer_identity_address = address, |
| peer_irk = irk, |
| local_irk = self.irk |
| ) |
| ) |
| |
| # Enable address resolution |
| # await self.send_command( |
| # HCI_LE_Set_Address_Resolution_Enable_Command(address_resolution_enable=1) |
| # ) |
| |
| # Create a host-side address resolver |
| self.address_resolver = smp.AddressResolver(resolving_keys) |
| |
| if self.classic_enabled: |
| await self.send_command( |
| HCI_Write_Local_Name_Command(local_name=self.name.encode('utf8')) |
| ) |
| await self.send_command( |
| HCI_Write_Class_Of_Device_Command(class_of_device = self.class_of_device) |
| ) |
| await self.send_command( |
| HCI_Write_Simple_Pairing_Mode_Command( |
| simple_pairing_mode=int(self.classic_ssp_enabled)) |
| ) |
| await self.send_command( |
| HCI_Write_Secure_Connections_Host_Support_Command( |
| secure_connections_host_support=int(self.classic_sc_enabled)) |
| ) |
| |
| # Let the SMP manager know about the address |
| # TODO: allow using a public address |
| self.smp_manager.address = self.random_address |
| |
| # Done |
| self.powered_on = True |
| |
| async def start_advertising(self, auto_restart=False): |
| self.auto_restart_advertising = auto_restart |
| |
| # If we're advertising, stop first |
| if self.advertising: |
| await self.stop_advertising() |
| |
| # Set/update the advertising data |
| await self.send_command(HCI_LE_Set_Advertising_Data_Command( |
| advertising_data = self.advertising_data |
| )) |
| |
| # Set/update the scan response data |
| await self.send_command(HCI_LE_Set_Scan_Response_Data_Command( |
| scan_response_data = self.scan_response_data |
| )) |
| |
| # Set the advertising parameters |
| await self.send_command(HCI_LE_Set_Advertising_Parameters_Command( |
| # TODO: use real values, not fixed ones |
| advertising_interval_min = self.advertising_interval_min, |
| advertising_interval_max = self.advertising_interval_max, |
| advertising_type = HCI_LE_Set_Advertising_Parameters_Command.ADV_IND, |
| own_address_type = Address.RANDOM_DEVICE_ADDRESS, # TODO: allow using the public address |
| peer_address_type = Address.PUBLIC_DEVICE_ADDRESS, |
| peer_address = Address('00:00:00:00:00:00'), |
| advertising_channel_map = 7, |
| advertising_filter_policy = 0 |
| )) |
| |
| # Enable advertising |
| await self.send_command(HCI_LE_Set_Advertising_Enable_Command( |
| advertising_enable = 1 |
| )) |
| |
| self.advertising = True |
| |
| async def stop_advertising(self): |
| # Disable advertising |
| if self.advertising: |
| await self.send_command(HCI_LE_Set_Advertising_Enable_Command( |
| advertising_enable = 0 |
| )) |
| |
| self.advertising = False |
| |
| @property |
| def is_advertising(self): |
| return self.advertising |
| |
| async def start_scanning( |
| self, |
| active=True, |
| scan_interval=DEVICE_DEFAULT_SCAN_INTERVAL, # Scan interval in ms |
| scan_window=DEVICE_DEFAULT_SCAN_WINDOW, # Scan window in ms |
| own_address_type=Address.RANDOM_DEVICE_ADDRESS, |
| filter_duplicates=False |
| ): |
| # Check that the arguments are legal |
| if scan_interval < scan_window: |
| raise ValueError('scan_interval must be >= scan_window') |
| if scan_interval < DEVICE_MIN_SCAN_INTERVAL or scan_interval > DEVICE_MAX_SCAN_INTERVAL: |
| raise ValueError('scan_interval out of range') |
| if scan_window < DEVICE_MIN_SCAN_WINDOW or scan_window > DEVICE_MAX_SCAN_WINDOW: |
| raise ValueError('scan_interval out of range') |
| |
| # Set the scanning parameters |
| scan_type = HCI_LE_Set_Scan_Parameters_Command.ACTIVE_SCANNING if active else HCI_LE_Set_Scan_Parameters_Command.PASSIVE_SCANNING |
| await self.send_command(HCI_LE_Set_Scan_Parameters_Command( |
| le_scan_type = scan_type, |
| le_scan_interval = int(scan_window / 0.625), |
| le_scan_window = int(scan_window / 0.625), |
| own_address_type = own_address_type, |
| scanning_filter_policy = HCI_LE_Set_Scan_Parameters_Command.BASIC_UNFILTERED_POLICY |
| )) |
| |
| # Enable scanning |
| await self.send_command(HCI_LE_Set_Scan_Enable_Command( |
| le_scan_enable = 1, |
| filter_duplicates = 1 if filter_duplicates else 0 |
| )) |
| self.scanning = True |
| |
| async def stop_scanning(self): |
| await self.send_command(HCI_LE_Set_Scan_Enable_Command( |
| le_scan_enable = 0, |
| filter_duplicates = 0 |
| )) |
| self.scanning = False |
| |
| @property |
| def is_scanning(self): |
| return self.scanning |
| |
| @host_event_handler |
| def on_advertising_report(self, address, data, rssi, advertisement_type): |
| if not (accumulator := self.advertisement_data.get(address)): |
| accumulator = AdvertisementDataAccumulator() |
| self.advertisement_data[address] = accumulator |
| accumulator.update(data, advertisement_type) |
| if accumulator.flushable: |
| self.emit( |
| 'advertisement', |
| address, |
| accumulator.advertising_data, |
| rssi, |
| accumulator.connectable |
| ) |
| |
| async def start_discovery(self): |
| await self.host.send_command(HCI_Write_Inquiry_Mode_Command(inquiry_mode=HCI_EXTENDED_INQUIRY_MODE)) |
| |
| response = await self.send_command(HCI_Inquiry_Command( |
| lap = HCI_GENERAL_INQUIRY_LAP, |
| inquiry_length = DEVICE_DEFAULT_INQUIRY_LENGTH, |
| num_responses = 0 # Unlimited number of responses. |
| )) |
| if response.status != HCI_Command_Status_Event.PENDING: |
| self.discovering = False |
| raise RuntimeError(f'HCI_Inquiry command failed: {HCI_Constant.status_name(response.status)} ({response.status})') |
| |
| self.discovering = True |
| |
| async def stop_discovery(self): |
| await self.send_command(HCI_Inquiry_Cancel_Command()) |
| self.discovering = False |
| |
| @host_event_handler |
| def on_inquiry_result(self, address, class_of_device, data, rssi): |
| self.emit( |
| 'inquiry_result', |
| address, |
| class_of_device, |
| AdvertisingData.from_bytes(data), |
| rssi |
| ) |
| |
| async def set_scan_enable(self, inquiry_scan_enabled, page_scan_enabled): |
| if inquiry_scan_enabled and page_scan_enabled: |
| scan_enable = 0x03 |
| elif page_scan_enabled: |
| scan_enable = 0x02 |
| elif inquiry_scan_enabled: |
| scan_enable = 0x01 |
| else: |
| scan_enable = 0x00 |
| |
| return await self.send_command(HCI_Write_Scan_Enable_Command(scan_enable = scan_enable)) |
| |
| async def set_discoverable(self, discoverable=True): |
| self.discoverable = discoverable |
| if self.classic_enabled: |
| # Synthesize an inquiry response if none is set already |
| if self.inquiry_response is None: |
| self.inquiry_response = bytes( |
| AdvertisingData([ |
| (AdvertisingData.COMPLETE_LOCAL_NAME, bytes(self.name, 'utf-8')) |
| ]) |
| ) |
| |
| # Update the controller |
| await self.host.send_command( |
| HCI_Write_Extended_Inquiry_Response_Command( |
| fec_required = 0, |
| extended_inquiry_response = self.inquiry_response |
| ) |
| ) |
| await self.set_scan_enable( |
| inquiry_scan_enabled = self.discoverable, |
| page_scan_enabled = self.connectable |
| ) |
| |
| async def set_connectable(self, connectable=True): |
| self.connectable = connectable |
| if self.classic_enabled: |
| await self.set_scan_enable( |
| inquiry_scan_enabled = self.discoverable, |
| page_scan_enabled = self.connectable |
| ) |
| |
| async def connect(self, peer_address, transport=BT_LE_TRANSPORT): |
| ''' |
| Request a connection to a peer. |
| This method cannot be called if there is already a pending connection. |
| ''' |
| |
| # Adjust the transport automatically if we need to |
| if transport == BT_LE_TRANSPORT and not self.le_enabled: |
| transport = BT_BR_EDR_TRANSPORT |
| elif transport == BT_BR_EDR_TRANSPORT and not self.classic_enabled: |
| transport = BT_LE_TRANSPORT |
| |
| # Check that there isn't already a pending connection |
| if self.is_connecting: |
| raise InvalidStateError('connection already pending') |
| |
| if type(peer_address) is str: |
| try: |
| peer_address = Address(peer_address) |
| except ValueError: |
| # If the address is not parssable, assume it is a name instead |
| logger.debug('looking for peer by name') |
| peer_address = await self.find_peer_by_name(peer_address, transport) |
| |
| # Create a future so that we can wait for the connection's result |
| pending_connection = asyncio.get_running_loop().create_future() |
| self.on('connection', pending_connection.set_result) |
| self.on('connection_failure', pending_connection.set_exception) |
| |
| # Tell the controller to connect |
| if transport == BT_LE_TRANSPORT: |
| # TODO: use real values, not fixed ones |
| result = await self.send_command(HCI_LE_Create_Connection_Command( |
| le_scan_interval = 96, |
| le_scan_window = 96, |
| initiator_filter_policy = 0, |
| peer_address_type = peer_address.address_type, |
| peer_address = peer_address, |
| own_address_type = Address.RANDOM_DEVICE_ADDRESS, |
| conn_interval_min = 12, |
| conn_interval_max = 24, |
| conn_latency = 0, |
| supervision_timeout = 72, |
| minimum_ce_length = 0, |
| maximum_ce_length = 0 |
| )) |
| else: |
| # TODO: use real values, not fixed ones |
| result = await self.send_command(HCI_Create_Connection_Command( |
| bd_addr = peer_address, |
| packet_type = 0xCC18, # FIXME: change |
| page_scan_repetition_mode = HCI_R2_PAGE_SCAN_REPETITION_MODE, |
| clock_offset = 0x0000, |
| allow_role_switch = 0x01, |
| reserved = 0 |
| )) |
| |
| try: |
| if result.status != HCI_Command_Status_Event.PENDING: |
| raise RuntimeError(f'HCI_LE_Create_Connection_Command failed: {HCI_Constant.status_name(result.status)} ({result.status})') |
| |
| # Wait for the connection process to complete |
| self.connecting = True |
| return await pending_connection |
| finally: |
| self.remove_listener('connection', pending_connection.set_result) |
| self.remove_listener('connection_failure', pending_connection.set_exception) |
| self.connecting = False |
| |
| @property |
| def is_connecting(self): |
| return self.connecting |
| |
| @property |
| def is_disconnecting(self): |
| return self.disconnecting |
| |
| async def cancel_connection(self): |
| if not self.is_connecting: |
| return |
| await self.send_command(HCI_LE_Create_Connection_Cancel_Command()) |
| |
| async def disconnect(self, connection, reason): |
| # Create a future so that we can wait for the disconnection's result |
| pending_disconnection = asyncio.get_running_loop().create_future() |
| connection.on('disconnection', pending_disconnection.set_result) |
| connection.on('disconnection_failure', pending_disconnection.set_exception) |
| |
| # Request a disconnection |
| result = await self.send_command(HCI_Disconnect_Command(connection_handle = connection.handle, reason = reason)) |
| |
| try: |
| if result.status != HCI_Command_Status_Event.PENDING: |
| raise RuntimeError(f'HCI_Disconnect_Command failed: {HCI_Constant.status_name(result.status)} ({result.status})') |
| |
| # Wait for the disconnection process to complete |
| self.disconnecting = True |
| return await pending_disconnection |
| finally: |
| connection.remove_listener('disconnection', pending_disconnection.set_result) |
| connection.remove_listener('disconnection_failure', pending_disconnection.set_exception) |
| self.disconnecting = False |
| |
| async def update_connection_parameters( |
| self, |
| connection, |
| conn_interval_min, |
| conn_interval_max, |
| conn_latency, |
| supervision_timeout, |
| minimum_ce_length = 0, |
| maximum_ce_length = 0 |
| ): |
| ''' |
| NOTE: the name of the parameters may look odd, but it just follows the names used in the Bluetooth spec. |
| ''' |
| await self.send_command(HCI_LE_Connection_Update_Command( |
| connection_handle = connection.handle, |
| conn_interval_min = conn_interval_min, |
| conn_interval_max = conn_interval_max, |
| conn_latency = conn_latency, |
| supervision_timeout = supervision_timeout, |
| minimum_ce_length = minimum_ce_length, |
| maximum_ce_length = maximum_ce_length |
| )) |
| # TODO: check result |
| |
| async def find_peer_by_name(self, name, transport=BT_LE_TRANSPORT): |
| """ |
| Scan for a peer with a give name and return its address and transport |
| """ |
| |
| # Create a future to wait for an address to be found |
| peer_address = asyncio.get_running_loop().create_future() |
| |
| # Scan/inquire with event handlers to handle scan/inquiry results |
| def on_peer_found(address, ad_data): |
| local_name = ad_data.get(AdvertisingData.COMPLETE_LOCAL_NAME) |
| if local_name is None: |
| local_name = ad_data.get(AdvertisingData.SHORTENED_LOCAL_NAME) |
| if local_name is not None: |
| if local_name.decode('utf-8') == name: |
| peer_address.set_result(address) |
| try: |
| handler = None |
| if transport == BT_LE_TRANSPORT: |
| event_name = 'advertisement' |
| handler = self.on( |
| event_name, |
| lambda address, ad_data, rssi, connectable: |
| on_peer_found(address, ad_data) |
| ) |
| |
| was_scanning = self.scanning |
| if not self.scanning: |
| await self.start_scanning(filter_duplicates=True) |
| |
| elif transport == BT_BR_EDR_TRANSPORT: |
| event_name = 'inquiry_result' |
| handler = self.on( |
| event_name, |
| lambda address, class_of_device, eir_data, rssi: |
| on_peer_found(address, eir_data) |
| ) |
| |
| was_discovering = self.discovering |
| if not self.discovering: |
| await self.start_discovery() |
| else: |
| return None |
| |
| return await peer_address |
| finally: |
| if handler is not None: |
| self.remove_listener(event_name, handler) |
| |
| if transport == BT_LE_TRANSPORT and not was_scanning: |
| await self.stop_scanning() |
| elif transport == BT_BR_EDR_TRANSPORT and not was_discovering: |
| await self.stop_discovery() |
| |
| @property |
| def pairing_config_factory(self): |
| return self.smp_manager.pairing_config_factory |
| |
| @pairing_config_factory.setter |
| def pairing_config_factory(self, pairing_config_factory): |
| self.smp_manager.pairing_config_factory = pairing_config_factory |
| |
| async def pair(self, connection): |
| return await self.smp_manager.pair(connection) |
| |
| def request_pairing(self, connection): |
| return self.smp_manager.request_pairing(connection) |
| |
| async def get_long_term_key(self, connection_handle, rand, ediv): |
| if (connection := self.lookup_connection(connection_handle)) is None: |
| return |
| |
| # Start by looking for the key in an SMP session |
| ltk = self.smp_manager.get_long_term_key(connection, rand, ediv) |
| if ltk is not None: |
| return ltk |
| |
| # Then look for the key in the keystore |
| if self.keystore is not None: |
| keys = await self.keystore.get(str(connection.peer_address)) |
| if keys is not None: |
| logger.debug('found keys in the key store') |
| if keys.ltk: |
| return keys.ltk.value |
| elif connection.role == BT_CENTRAL_ROLE and keys.ltk_central: |
| return keys.ltk_central.value |
| elif connection.role == BT_PERIPHERAL_ROLE and keys.ltk_peripheral: |
| return keys.ltk_peripheral.value |
| |
| async def get_link_key(self, address): |
| # Look for the key in the keystore |
| if self.keystore is not None: |
| keys = await self.keystore.get(str(address)) |
| if keys is not None: |
| logger.debug('found keys in the key store') |
| return keys.link_key.value |
| |
| # [Classic only] |
| async def authenticate(self, connection): |
| # Set up event handlers |
| pending_authentication = asyncio.get_running_loop().create_future() |
| |
| def on_authentication(): |
| pending_authentication.set_result(None) |
| |
| def on_authentication_failure(error_code): |
| pending_authentication.set_exception(HCI_Error(error_code)) |
| |
| connection.on('connection_authentication', on_authentication) |
| connection.on('connection_authentication_failure', on_authentication_failure) |
| |
| # Request the authentication |
| try: |
| result = await self.send_command( |
| HCI_Authentication_Requested_Command(connection_handle = connection.handle) |
| ) |
| if result.status != HCI_COMMAND_STATUS_PENDING: |
| logger.warn(f'HCI_Authentication_Requested_Command failed: {HCI_Constant.error_name(result.status)}') |
| raise HCI_Error(result.status) |
| |
| # Wait for the authentication to complete |
| await pending_authentication |
| finally: |
| connection.remove_listener('connection_authentication', on_authentication) |
| connection.remove_listener('connection_authentication_failure', on_authentication_failure) |
| |
| async def encrypt(self, connection): |
| # Set up event handlers |
| pending_encryption = asyncio.get_running_loop().create_future() |
| |
| def on_encryption_change(): |
| pending_encryption.set_result(None) |
| |
| def on_encryption_failure(error_code): |
| pending_encryption.set_exception(HCI_Error(error_code)) |
| |
| connection.on('connection_encryption_change', on_encryption_change) |
| connection.on('connection_encryption_failure', on_encryption_failure) |
| |
| # Request the encryption |
| try: |
| if connection.transport == BT_LE_TRANSPORT: |
| # Look for a key in the key store |
| if self.keystore is None: |
| raise RuntimeError('no key store') |
| |
| keys = await self.keystore.get(str(connection.peer_address)) |
| if keys is None: |
| raise RuntimeError('keys not found in key store') |
| |
| if keys.ltk is not None: |
| ltk = keys.ltk.value |
| rand = bytes(8) |
| ediv = 0 |
| elif keys.ltk_central is not None: |
| ltk = keys.ltk_central.value |
| rand = keys.ltk_central.rand |
| ediv = keys.ltk_central.ediv |
| else: |
| raise RuntimeError('no LTK found for peer') |
| |
| if connection.role != HCI_CENTRAL_ROLE: |
| raise InvalidStateError('only centrals can start encryption') |
| |
| result = await self.send_command( |
| HCI_LE_Start_Encryption_Command( |
| connection_handle = connection.handle, |
| random_number = rand, |
| encrypted_diversifier = ediv, |
| long_term_key = ltk |
| ) |
| ) |
| |
| if result.status != HCI_COMMAND_STATUS_PENDING: |
| logger.warn(f'HCI_LE_Start_Encryption_Command failed: {HCI_Constant.error_name(result.status)}') |
| raise HCI_Error(result.status) |
| else: |
| result = await self.send_command( |
| HCI_Set_Connection_Encryption_Command( |
| connection_handle = connection.handle, |
| encryption_enable = 0x01 |
| ) |
| ) |
| |
| if result.status != HCI_COMMAND_STATUS_PENDING: |
| logger.warn(f'HCI_Set_Connection_Encryption_Command failed: {HCI_Constant.error_name(result.status)}') |
| raise HCI_Error(result.status) |
| |
| # Wait for the result |
| await pending_encryption |
| finally: |
| connection.remove_listener('connection_encryption_change', on_encryption_change) |
| connection.remove_listener('connection_encryption_failure', on_encryption_failure) |
| |
| # [Classic only] |
| async def request_remote_name(self, connection): |
| # Set up event handlers |
| pending_name = asyncio.get_running_loop().create_future() |
| |
| def on_remote_name(): |
| pending_name.set_result(connection.peer_name) |
| |
| def on_remote_name_failure(error_code): |
| pending_name.set_exception(HCI_Error(error_code)) |
| |
| connection.on('remote_name', on_remote_name) |
| connection.on('remote_name_failure', on_remote_name_failure) |
| |
| try: |
| result = await self.send_command( |
| HCI_Remote_Name_Request_Command( |
| bd_addr = connection.peer_address, |
| page_scan_repetition_mode = HCI_Remote_Name_Request_Command.R0, # TODO investigate other options |
| reserved = 0, |
| clock_offset = 0 # TODO investigate non-0 values |
| ) |
| ) |
| |
| if result.status != HCI_COMMAND_STATUS_PENDING: |
| logger.warn(f'HCI_Set_Connection_Encryption_Command failed: {HCI_Constant.error_name(result.status)}') |
| raise HCI_Error(result.status) |
| |
| # Wait for the result |
| return await pending_name |
| finally: |
| connection.remove_listener('remote_name', on_remote_name) |
| connection.remove_listener('remote_name_failure', on_remote_name_failure) |
| |
| # [Classic only] |
| @host_event_handler |
| def on_link_key(self, bd_addr, link_key, key_type): |
| # Store the keys in the key store |
| if self.keystore: |
| pairing_keys = keys.PairingKeys() |
| pairing_keys.link_key = keys.PairingKeys.Key(value = link_key) |
| |
| async def store_keys(): |
| try: |
| await self.keystore.update(str(bd_addr), pairing_keys) |
| except Exception as error: |
| logger.warn(f'!!! error while storing keys: {error}') |
| |
| asyncio.create_task(store_keys()) |
| |
| def add_service(self, service): |
| self.gatt_server.add_service(service) |
| |
| def add_services(self, services): |
| self.gatt_server.add_services(services) |
| |
| async def notify_subscriber(self, connection, attribute, force=False): |
| await self.gatt_server.notify_subscriber(connection, attribute, force) |
| |
| async def notify_subscribers(self, attribute, force=False): |
| await self.gatt_server.notify_subscribers(attribute, force) |
| |
| async def indicate_subscriber(self, connection, attribute, force=False): |
| await self.gatt_server.indicate_subscriber(connection, attribute, force) |
| |
| async def indicate_subscribers(self, attribute): |
| await self.gatt_server.indicate_subscribers(attribute) |
| |
| @host_event_handler |
| def on_connection(self, connection_handle, transport, peer_address, peer_resolvable_address, role, connection_parameters): |
| logger.debug(f'*** Connection: [0x{connection_handle:04X}] {peer_address} as {HCI_Constant.role_name(role)}') |
| if connection_handle in self.connections: |
| logger.warn('new connection reuses the same handle as a previous connection') |
| |
| # Resolve the peer address if we can |
| if self.address_resolver: |
| if peer_address.is_resolvable: |
| resolved_address = self.address_resolver.resolve(peer_address) |
| if resolved_address is not None: |
| logger.debug(f'*** Address resolved as {resolved_address}') |
| peer_resolvable_address = peer_address |
| peer_address = resolved_address |
| |
| # Create a new connection |
| connection = Connection( |
| self, |
| connection_handle, |
| transport, |
| peer_address, |
| peer_resolvable_address, |
| role, |
| connection_parameters |
| ) |
| self.connections[connection_handle] = connection |
| |
| # We are no longer advertising |
| self.advertising = False |
| |
| # Emit an event to notify listeners of the new connection |
| self.emit('connection', connection) |
| |
| @host_event_handler |
| def on_connection_failure(self, error_code): |
| logger.debug(f'*** Connection failed: {error_code}') |
| error = ConnectionError( |
| error_code, |
| 'hci', |
| HCI_Constant.error_name(error_code) |
| ) |
| self.emit('connection_failure', error) |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_disconnection(self, connection, reason): |
| logger.debug(f'*** Disconnection: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}, reason={reason}') |
| connection.emit('disconnection', reason) |
| |
| # Remove the connection from the map |
| del self.connections[connection.handle] |
| |
| # Cleanup subsystems that maintain per-connection state |
| self.gatt_server.on_disconnection(connection) |
| |
| # Restart advertising if auto-restart is enabled |
| if self.auto_restart_advertising: |
| logger.debug('restarting advertising') |
| asyncio.create_task(self.start_advertising(auto_restart=self.auto_restart_advertising)) |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_disconnection_failure(self, connection, error_code): |
| logger.debug(f'*** Disconnection failed: {error_code}') |
| error = ConnectionError( |
| error_code, |
| 'hci', |
| HCI_Constant.error_name(error_code) |
| ) |
| connection.emit('disconnection_failure', error) |
| |
| @host_event_handler |
| @AsyncRunner.run_in_task() |
| async def on_inquiry_complete(self): |
| if self.discovering: |
| # Inquire again |
| await self.start_discovery() |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_connection_authentication(self, connection): |
| logger.debug(f'*** Connection Authentication: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}') |
| connection.authenticated = True |
| connection.emit('connection_authentication') |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_connection_authentication_failure(self, connection, error): |
| logger.debug(f'*** Connection Authentication Failure: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}, error={error}') |
| connection.emit('connection_authentication_failure', error) |
| |
| # [Classic only] |
| @host_event_handler |
| @with_connection_from_address |
| def on_authentication_io_capability_request(self, connection): |
| # Ask what the pairing config should be for this connection |
| pairing_config = self.pairing_config_factory(connection) |
| |
| # Map the SMP IO capability to a Classic IO capability |
| io_capability = { |
| smp.SMP_DISPLAY_ONLY_IO_CAPABILITY: HCI_DISPLAY_ONLY_IO_CAPABILITY, |
| smp.SMP_DISPLAY_YES_NO_IO_CAPABILITY: HCI_DISPLAY_YES_NO_IO_CAPABILITY, |
| smp.SMP_KEYBOARD_ONLY_IO_CAPABILITY: HCI_KEYBOARD_ONLY_IO_CAPABILITY, |
| smp.SMP_NO_INPUT_NO_OUTPUT_IO_CAPABILITY: HCI_NO_INPUT_NO_OUTPUT_IO_CAPABILITY, |
| smp.SMP_KEYBOARD_DISPLAY_IO_CAPABILITY: HCI_DISPLAY_YES_NO_IO_CAPABILITY |
| }.get(pairing_config.delegate.io_capability) |
| |
| if io_capability is None: |
| logger.warning(f'cannot map IO capability ({pairing_config.delegate.io_capability}') |
| io_capability = HCI_NO_INPUT_NO_OUTPUT_IO_CAPABILITY |
| |
| # Compute the authentication requirements |
| authentication_requirements = ( |
| # No Bonding |
| ( |
| HCI_MITM_NOT_REQUIRED_NO_BONDING_AUTHENTICATION_REQUIREMENTS, |
| HCI_MITM_REQUIRED_NO_BONDING_AUTHENTICATION_REQUIREMENTS |
| ), |
| # General Bonding |
| ( |
| HCI_MITM_NOT_REQUIRED_GENERAL_BONDING_AUTHENTICATION_REQUIREMENTS, |
| HCI_MITM_REQUIRED_GENERAL_BONDING_AUTHENTICATION_REQUIREMENTS |
| ) |
| )[1 if pairing_config.bonding else 0][1 if pairing_config.mitm else 0] |
| |
| # Respond |
| self.host.send_command_sync( |
| HCI_IO_Capability_Request_Reply_Command( |
| bd_addr = connection.peer_address, |
| io_capability = io_capability, |
| oob_data_present = 0x00, # Not present |
| authentication_requirements = authentication_requirements |
| ) |
| ) |
| |
| # [Classic only] |
| @host_event_handler |
| @with_connection_from_address |
| def on_authentication_user_confirmation_request(self, connection, code): |
| # Ask what the pairing config should be for this connection |
| pairing_config = self.pairing_config_factory(connection) |
| |
| can_confirm = pairing_config.delegate.io_capability not in { |
| smp.SMP_NO_INPUT_NO_OUTPUT_IO_CAPABILITY, |
| smp.SMP_DISPLAY_ONLY_IO_CAPABILITY |
| } |
| |
| # Respond |
| if can_confirm and pairing_config.delegate: |
| async def compare_numbers(): |
| numbers_match = await pairing_config.delegate.compare_numbers(code, digits=6) |
| if numbers_match: |
| self.host.send_command_sync( |
| HCI_User_Confirmation_Request_Reply_Command(bd_addr=connection.peer_address) |
| ) |
| else: |
| self.host.send_command_sync( |
| HCI_User_Confirmation_Request_Negative_Reply_Command(bd_addr=connection.peer_address) |
| ) |
| |
| asyncio.create_task(compare_numbers()) |
| else: |
| self.host.send_command_sync( |
| HCI_User_Confirmation_Request_Reply_Command(bd_addr=connection.peer_address) |
| ) |
| |
| # [Classic only] |
| @host_event_handler |
| @with_connection_from_address |
| def on_authentication_user_passkey_request(self, connection): |
| # Ask what the pairing config should be for this connection |
| pairing_config = self.pairing_config_factory(connection) |
| |
| can_input = pairing_config.delegate.io_capability in { |
| smp.SMP_KEYBOARD_ONLY_IO_CAPABILITY, |
| smp.SMP_KEYBOARD_DISPLAY_IO_CAPABILITY |
| } |
| |
| # Respond |
| if can_input and pairing_config.delegate: |
| async def get_number(): |
| number = await pairing_config.delegate.get_number() |
| if number is not None: |
| self.host.send_command_sync( |
| HCI_User_Passkey_Request_Reply_Command( |
| bd_addr = connection.peer_address, |
| numeric_value = number) |
| ) |
| else: |
| self.host.send_command_sync( |
| HCI_User_Passkey_Request_Negative_Reply_Command(bd_addr=connection.peer_address) |
| ) |
| |
| asyncio.create_task(get_number()) |
| else: |
| self.host.send_command_sync( |
| HCI_User_Passkey_Request_Negative_Reply_Command(bd_addr=connection.peer_address) |
| ) |
| |
| # [Classic only] |
| @host_event_handler |
| @with_connection_from_address |
| def on_remote_name(self, connection, remote_name): |
| # Try to decode the name |
| try: |
| connection.peer_name = remote_name.decode('utf-8') |
| connection.emit('remote_name') |
| except UnicodeDecodeError as error: |
| logger.warning('peer name is not valid UTF-8') |
| connection.emit('remote_name_failure', error) |
| |
| # [Classic only] |
| @host_event_handler |
| @with_connection_from_address |
| def on_remote_name_failure(self, connection, error): |
| connection.emit('remote_name_failure', error) |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_connection_encryption_change(self, connection, encryption): |
| logger.debug(f'*** Connection Encryption Change: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}, encryption={encryption}') |
| connection.encryption = encryption |
| connection.emit('connection_encryption_change') |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_connection_encryption_failure(self, connection, error): |
| logger.debug(f'*** Connection Encryption Failure: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}, error={error}') |
| connection.emit('connection_encryption_failure', error) |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_connection_encryption_key_refresh(self, connection): |
| logger.debug(f'*** Connection Key Refresh: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}') |
| connection.emit('connection_encryption_key_refresh') |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_connection_parameters_update(self, connection, connection_parameters): |
| logger.debug(f'*** Connection Parameters Update: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}, {connection_parameters}') |
| connection.parameters = connection_parameters |
| connection.emit('connection_parameters_update') |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_connection_parameters_update_failure(self, connection, error): |
| logger.debug(f'*** Connection Parameters Update Failed: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}, error={error}') |
| connection.emit('connection_parameters_update_failure', error) |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_connection_phy_update(self, connection, connection_phy): |
| logger.debug(f'*** Connection PHY Update: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}, {connection_phy}') |
| connection.phy = connection_phy |
| connection.emit('connection_phy_update') |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_connection_phy_update_failure(self, connection, error): |
| logger.debug(f'*** Connection PHY Update Failed: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}, error={error}') |
| connection.emit('connection_phy_update_failure', error) |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_connection_att_mtu_update(self, connection, att_mtu): |
| logger.debug(f'*** Connection ATT MTU Update: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}, {att_mtu}') |
| connection.att_mtu = att_mtu |
| connection.emit('connection_att_mtu_update') |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_connection_data_length_change(self, connection, max_tx_octets, max_tx_time, max_rx_octets, max_rx_time): |
| logger.debug(f'*** Connection Data Length Change: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}') |
| connection.data_length = (max_tx_octets, max_tx_time, max_rx_octets, max_rx_time) |
| connection.emit('connection_data_length_change') |
| |
| @with_connection_from_handle |
| def on_pairing_start(self, connection): |
| connection.emit('pairing_start') |
| |
| @with_connection_from_handle |
| def on_pairing(self, connection, keys): |
| connection.emit('pairing', keys) |
| |
| @with_connection_from_handle |
| def on_pairing_failure(self, connection, reason): |
| connection.emit('pairing_failure', reason) |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_gatt_pdu(self, connection, pdu): |
| # Parse the L2CAP payload into an ATT PDU object |
| att_pdu = ATT_PDU.from_bytes(pdu) |
| |
| # Conveniently, even-numbered op codes are client->server and |
| # odd-numbered ones are server->client |
| if att_pdu.op_code & 1: |
| if connection.gatt_client is None: |
| logger.warn(color('no GATT client for connection 0x{connection_handle:04X}')) |
| return |
| connection.gatt_client.on_gatt_pdu(att_pdu) |
| else: |
| if connection.gatt_server is None: |
| logger.warn(color('no GATT server for connection 0x{connection_handle:04X}')) |
| return |
| connection.gatt_server.on_gatt_pdu(connection, att_pdu) |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_smp_pdu(self, connection, pdu): |
| self.smp_manager.on_smp_pdu(connection, pdu) |
| |
| @host_event_handler |
| @with_connection_from_handle |
| def on_l2cap_pdu(self, connection, cid, pdu): |
| self.l2cap_channel_manager.on_pdu(connection, cid, pdu) |
| |
| def __str__(self): |
| return f'Device(name="{self.name}", random_address="{self.random_address}"", public_address="{self.public_address}")' |
