Google Git
Sign in
android / platform / external / python / bumble / c2959da / . / bumble / device.py
blob: 5c7bf5ada3bd276c79d726a42f8e3f60ed0f3fdc [file] [log] [blame]
# 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
# -----------------------------------------------------------------------------
from enum import IntEnum
import functools
import json
import asyncio
import logging
from contextlib import asynccontextmanager, AsyncExitStack
from dataclasses import dataclass
from colors import color
from .att import ATT_CID, ATT_DEFAULT_MTU, ATT_PDU
from .gatt import Characteristic, Descriptor, Service
from .hci import (
HCI_CENTRAL_ROLE,
HCI_COMMAND_STATUS_PENDING,
HCI_CONNECTION_REJECTED_DUE_TO_LIMITED_RESOURCES_ERROR,
HCI_DISPLAY_ONLY_IO_CAPABILITY,
HCI_DISPLAY_YES_NO_IO_CAPABILITY,
HCI_EXTENDED_INQUIRY_MODE,
HCI_GENERAL_INQUIRY_LAP,
HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR,
HCI_KEYBOARD_ONLY_IO_CAPABILITY,
HCI_LE_1M_PHY,
HCI_LE_1M_PHY_BIT,
HCI_LE_2M_PHY,
HCI_LE_2M_PHY_LE_SUPPORTED_FEATURE,
HCI_LE_CLEAR_RESOLVING_LIST_COMMAND,
HCI_LE_CODED_PHY,
HCI_LE_CODED_PHY_BIT,
HCI_LE_CODED_PHY_LE_SUPPORTED_FEATURE,
HCI_LE_EXTENDED_ADVERTISING_LE_SUPPORTED_FEATURE,
HCI_LE_EXTENDED_CREATE_CONNECTION_COMMAND,
HCI_LE_READ_PHY_COMMAND,
HCI_MITM_NOT_REQUIRED_GENERAL_BONDING_AUTHENTICATION_REQUIREMENTS,
HCI_MITM_NOT_REQUIRED_NO_BONDING_AUTHENTICATION_REQUIREMENTS,
HCI_MITM_REQUIRED_GENERAL_BONDING_AUTHENTICATION_REQUIREMENTS,
HCI_MITM_REQUIRED_NO_BONDING_AUTHENTICATION_REQUIREMENTS,
HCI_NO_INPUT_NO_OUTPUT_IO_CAPABILITY,
HCI_R2_PAGE_SCAN_REPETITION_MODE,
HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR,
HCI_SUCCESS,
HCI_WRITE_LE_HOST_SUPPORT_COMMAND,
Address,
HCI_Accept_Connection_Request_Command,
HCI_Authentication_Requested_Command,
HCI_Command_Status_Event,
HCI_Constant,
HCI_Create_Connection_Cancel_Command,
HCI_Create_Connection_Command,
HCI_Disconnect_Command,
HCI_Encryption_Change_Event,
HCI_Error,
HCI_IO_Capability_Request_Reply_Command,
HCI_Inquiry_Cancel_Command,
HCI_Inquiry_Command,
HCI_LE_Add_Device_To_Resolving_List_Command,
HCI_LE_Advertising_Report_Event,
HCI_LE_Clear_Resolving_List_Command,
HCI_LE_Connection_Update_Command,
HCI_LE_Create_Connection_Cancel_Command,
HCI_LE_Create_Connection_Command,
HCI_LE_Enable_Encryption_Command,
HCI_LE_Extended_Advertising_Report_Event,
HCI_LE_Extended_Create_Connection_Command,
HCI_LE_Read_PHY_Command,
HCI_LE_Set_Advertising_Data_Command,
HCI_LE_Set_Advertising_Enable_Command,
HCI_LE_Set_Advertising_Parameters_Command,
HCI_LE_Set_Default_PHY_Command,
HCI_LE_Set_Extended_Scan_Enable_Command,
HCI_LE_Set_Extended_Scan_Parameters_Command,
HCI_LE_Set_PHY_Command,
HCI_LE_Set_Random_Address_Command,
HCI_LE_Set_Scan_Enable_Command,
HCI_LE_Set_Scan_Parameters_Command,
HCI_LE_Set_Scan_Response_Data_Command,
HCI_Read_BD_ADDR_Command,
HCI_Read_RSSI_Command,
HCI_Reject_Connection_Request_Command,
HCI_Remote_Name_Request_Command,
HCI_Set_Connection_Encryption_Command,
HCI_StatusError,
HCI_User_Confirmation_Request_Negative_Reply_Command,
HCI_User_Confirmation_Request_Reply_Command,
HCI_User_Passkey_Request_Negative_Reply_Command,
HCI_User_Passkey_Request_Reply_Command,
HCI_Write_Class_Of_Device_Command,
HCI_Write_Extended_Inquiry_Response_Command,
HCI_Write_Inquiry_Mode_Command,
HCI_Write_LE_Host_Support_Command,
HCI_Write_Local_Name_Command,
HCI_Write_Scan_Enable_Command,
HCI_Write_Secure_Connections_Host_Support_Command,
HCI_Write_Simple_Pairing_Mode_Command,
OwnAddressType,
phy_list_to_bits,
)
from .host import Host
from .gap import GenericAccessService
from .core import (
BT_BR_EDR_TRANSPORT,
BT_CENTRAL_ROLE,
BT_LE_TRANSPORT,
BT_PERIPHERAL_ROLE,
AdvertisingData,
CommandTimeoutError,
ConnectionPHY,
InvalidStateError,
)
from .utils import (
AsyncRunner,
CompositeEventEmitter,
setup_event_forwarding,
composite_listener,
)
from .keys import (
KeyStore,
PairingKeys,
)
from . import gatt_client
from . import gatt_server
from . import smp
from . import sdp
from . import l2cap
from . import core
# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
logger = logging.getLogger(__name__)
# -----------------------------------------------------------------------------
# Constants
# -----------------------------------------------------------------------------
# fmt: off
# pylint: disable=line-too-long
DEVICE_MIN_SCAN_INTERVAL = 25
DEVICE_MAX_SCAN_INTERVAL = 10240
DEVICE_MIN_SCAN_WINDOW = 25
DEVICE_MAX_SCAN_WINDOW = 10240
DEVICE_MIN_LE_RSSI = -127
DEVICE_MAX_LE_RSSI = 20
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_DEFAULT_CONNECT_TIMEOUT = None # No timeout
DEVICE_DEFAULT_CONNECT_SCAN_INTERVAL = 60 # ms
DEVICE_DEFAULT_CONNECT_SCAN_WINDOW = 60 # ms
DEVICE_DEFAULT_CONNECTION_INTERVAL_MIN = 15 # ms
DEVICE_DEFAULT_CONNECTION_INTERVAL_MAX = 30 # ms
DEVICE_DEFAULT_CONNECTION_MAX_LATENCY = 0
DEVICE_DEFAULT_CONNECTION_SUPERVISION_TIMEOUT = 720 # ms
DEVICE_DEFAULT_CONNECTION_MIN_CE_LENGTH = 0 # ms
DEVICE_DEFAULT_CONNECTION_MAX_CE_LENGTH = 0 # ms
DEVICE_DEFAULT_L2CAP_COC_MTU = l2cap.L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MTU
DEVICE_DEFAULT_L2CAP_COC_MPS = l2cap.L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MPS
DEVICE_DEFAULT_L2CAP_COC_MAX_CREDITS = l2cap.L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_INITIAL_CREDITS
# fmt: on
# pylint: enable=line-too-long
# -----------------------------------------------------------------------------
# Classes
# -----------------------------------------------------------------------------
# -----------------------------------------------------------------------------
class Advertisement:
TX_POWER_NOT_AVAILABLE = (
HCI_LE_Extended_Advertising_Report_Event.TX_POWER_INFORMATION_NOT_AVAILABLE
)
RSSI_NOT_AVAILABLE = HCI_LE_Extended_Advertising_Report_Event.RSSI_NOT_AVAILABLE
@classmethod
def from_advertising_report(cls, report):
if isinstance(report, HCI_LE_Advertising_Report_Event.Report):
return LegacyAdvertisement.from_advertising_report(report)
if isinstance(report, HCI_LE_Extended_Advertising_Report_Event.Report):
return ExtendedAdvertisement.from_advertising_report(report)
return None
# pylint: disable=line-too-long
def __init__(
self,
address,
rssi=HCI_LE_Extended_Advertising_Report_Event.RSSI_NOT_AVAILABLE,
is_legacy=False,
is_anonymous=False,
is_connectable=False,
is_directed=False,
is_scannable=False,
is_scan_response=False,
is_complete=True,
is_truncated=False,
primary_phy=0,
secondary_phy=0,
tx_power=HCI_LE_Extended_Advertising_Report_Event.TX_POWER_INFORMATION_NOT_AVAILABLE,
sid=0,
data=b'',
):
self.address = address
self.rssi = rssi
self.is_legacy = is_legacy
self.is_anonymous = is_anonymous
self.is_connectable = is_connectable
self.is_directed = is_directed
self.is_scannable = is_scannable
self.is_scan_response = is_scan_response
self.is_complete = is_complete
self.is_truncated = is_truncated
self.primary_phy = primary_phy
self.secondary_phy = secondary_phy
self.tx_power = tx_power
self.sid = sid
self.data = AdvertisingData.from_bytes(data)
# -----------------------------------------------------------------------------
class LegacyAdvertisement(Advertisement):
@classmethod
def from_advertising_report(cls, report):
return cls(
address=report.address,
rssi=report.rssi,
is_legacy=True,
is_connectable=report.event_type
in (
HCI_LE_Advertising_Report_Event.ADV_IND,
HCI_LE_Advertising_Report_Event.ADV_DIRECT_IND,
),
is_directed=report.event_type
== HCI_LE_Advertising_Report_Event.ADV_DIRECT_IND,
is_scannable=report.event_type
in (
HCI_LE_Advertising_Report_Event.ADV_IND,
HCI_LE_Advertising_Report_Event.ADV_SCAN_IND,
),
is_scan_response=report.event_type
== HCI_LE_Advertising_Report_Event.SCAN_RSP,
data=report.data,
)
# -----------------------------------------------------------------------------
class ExtendedAdvertisement(Advertisement):
@classmethod
def from_advertising_report(cls, report):
# fmt: off
# pylint: disable=line-too-long
return cls(
address = report.address,
rssi = report.rssi,
is_legacy = report.event_type & (1 << HCI_LE_Extended_Advertising_Report_Event.LEGACY_ADVERTISING_PDU_USED) != 0,
is_anonymous = report.address.address_type == HCI_LE_Extended_Advertising_Report_Event.ANONYMOUS_ADDRESS_TYPE,
is_connectable = report.event_type & (1 << HCI_LE_Extended_Advertising_Report_Event.CONNECTABLE_ADVERTISING) != 0,
is_directed = report.event_type & (1 << HCI_LE_Extended_Advertising_Report_Event.DIRECTED_ADVERTISING) != 0,
is_scannable = report.event_type & (1 << HCI_LE_Extended_Advertising_Report_Event.SCANNABLE_ADVERTISING) != 0,
is_scan_response = report.event_type & (1 << HCI_LE_Extended_Advertising_Report_Event.SCAN_RESPONSE) != 0,
is_complete = (report.event_type >> 5 & 3) == HCI_LE_Extended_Advertising_Report_Event.DATA_COMPLETE,
is_truncated = (report.event_type >> 5 & 3) == HCI_LE_Extended_Advertising_Report_Event.DATA_INCOMPLETE_TRUNCATED_NO_MORE_TO_COME,
primary_phy = report.primary_phy,
secondary_phy = report.secondary_phy,
tx_power = report.tx_power,
sid = report.advertising_sid,
data = report.data
)
# fmt: on
# -----------------------------------------------------------------------------
class AdvertisementDataAccumulator:
def __init__(self, passive=False):
self.passive = passive
self.last_advertisement = None
self.last_data = b''
def update(self, report):
advertisement = Advertisement.from_advertising_report(report)
result = None
if advertisement.is_scan_response:
if (
self.last_advertisement is not None
and not self.last_advertisement.is_scan_response
):
# This is the response to a scannable advertisement
result = Advertisement.from_advertising_report(report)
result.is_connectable = self.last_advertisement.is_connectable
result.is_scannable = True
result.data = AdvertisingData.from_bytes(self.last_data + report.data)
self.last_data = b''
else:
if (
self.passive
or (not advertisement.is_scannable)
or (
self.last_advertisement is not None
and not self.last_advertisement.is_scan_response
)
):
# Don't wait for a scan response
result = Advertisement.from_advertising_report(report)
self.last_data = report.data
self.last_advertisement = advertisement
return result
# -----------------------------------------------------------------------------
class AdvertisingType(IntEnum):
# fmt: off
# pylint: disable=line-too-long
UNDIRECTED_CONNECTABLE_SCANNABLE = 0x00 # Undirected, connectable, scannable
DIRECTED_CONNECTABLE_HIGH_DUTY = 0x01 # Directed, connectable, non-scannable
UNDIRECTED_SCANNABLE = 0x02 # Undirected, non-connectable, scannable
UNDIRECTED = 0x03 # Undirected, non-connectable, non-scannable
DIRECTED_CONNECTABLE_LOW_DUTY = 0x04 # Directed, connectable, non-scannable
# fmt: on
@property
def has_data(self):
return self in (
AdvertisingType.UNDIRECTED_CONNECTABLE_SCANNABLE,
AdvertisingType.UNDIRECTED_SCANNABLE,
AdvertisingType.UNDIRECTED,
)
@property
def is_connectable(self):
return self in (
AdvertisingType.UNDIRECTED_CONNECTABLE_SCANNABLE,
AdvertisingType.DIRECTED_CONNECTABLE_HIGH_DUTY,
AdvertisingType.DIRECTED_CONNECTABLE_LOW_DUTY,
)
@property
def is_scannable(self):
return self in (
AdvertisingType.UNDIRECTED_CONNECTABLE_SCANNABLE,
AdvertisingType.UNDIRECTED_SCANNABLE,
)
@property
def is_directed(self):
return self in (
AdvertisingType.DIRECTED_CONNECTABLE_HIGH_DUTY,
AdvertisingType.DIRECTED_CONNECTABLE_LOW_DUTY,
)
# -----------------------------------------------------------------------------
class LePhyOptions:
# Coded PHY preference
ANY_CODED_PHY = 0
PREFER_S_2_CODED_PHY = 1
PREFER_S_8_CODED_PHY = 2
def __init__(self, coded_phy_preference=0):
self.coded_phy_preference = coded_phy_preference
def __int__(self):
return self.coded_phy_preference & 3
# -----------------------------------------------------------------------------
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):
mtu = await self.gatt_client.request_mtu(mtu)
self.connection.emit('connection_att_mtu_update')
return 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, prefer_notify=True):
return await self.gatt_client.subscribe(
characteristic, subscriber, prefer_notify
)
async def unsubscribe(self, characteristic, subscriber=None):
return await self.gatt_client.unsubscribe(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)
def create_service_proxy(self, proxy_class):
return proxy_class.from_client(self.gatt_client)
async def discover_service_and_create_proxy(self, proxy_class):
# Discover the first matching service and its characteristics
services = await self.discover_service(proxy_class.SERVICE_CLASS.UUID)
if services:
service = services[0]
await service.discover_characteristics()
return self.create_service_proxy(proxy_class)
async def sustain(self, timeout=None):
await self.connection.sustain(timeout)
# [Classic only]
async def request_name(self):
return await self.connection.request_remote_name()
async def __aenter__(self):
await self.discover_services()
for service in self.services:
await service.discover_characteristics()
return self
async def __aexit__(self, exc_type, exc_value, traceback):
pass
def __str__(self):
return f'{self.connection.peer_address} as {self.connection.role_name}'
# -----------------------------------------------------------------------------
@dataclass
class ConnectionParametersPreferences:
connection_interval_min: int = DEVICE_DEFAULT_CONNECTION_INTERVAL_MIN
connection_interval_max: int = DEVICE_DEFAULT_CONNECTION_INTERVAL_MAX
max_latency: int = DEVICE_DEFAULT_CONNECTION_MAX_LATENCY
supervision_timeout: int = DEVICE_DEFAULT_CONNECTION_SUPERVISION_TIMEOUT
min_ce_length: int = DEVICE_DEFAULT_CONNECTION_MIN_CE_LENGTH
max_ce_length: int = DEVICE_DEFAULT_CONNECTION_MAX_CE_LENGTH
ConnectionParametersPreferences.default = ConnectionParametersPreferences()
# -----------------------------------------------------------------------------
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,
self_address,
peer_address,
peer_resolvable_address,
role,
parameters,
phy,
):
super().__init__()
self.device = device
self.handle = handle
self.transport = transport
self.self_address = self_address
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.sc = False
self.link_key_type = None
self.phy = 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
# [Classic only]
@classmethod
def incomplete(cls, device, peer_address):
"""
Instantiate an incomplete connection (ie. one waiting for a HCI Connection
Complete event).
Once received it shall be completed using the `.complete` method.
"""
return cls(
device,
None,
BT_BR_EDR_TRANSPORT,
device.public_address,
peer_address,
None,
None,
None,
None,
)
# [Classic only]
def complete(self, handle, peer_resolvable_address, role, parameters):
"""
Finish an incomplete connection upon completion.
"""
assert self.handle is None
assert self.transport == BT_BR_EDR_TRANSPORT
self.handle = handle
self.peer_resolvable_address = peer_resolvable_address
self.role = role
self.parameters = parameters
@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 open_l2cap_channel(
self,
psm,
max_credits=DEVICE_DEFAULT_L2CAP_COC_MAX_CREDITS,
mtu=DEVICE_DEFAULT_L2CAP_COC_MTU,
mps=DEVICE_DEFAULT_L2CAP_COC_MPS,
):
return await self.device.open_l2cap_channel(self, psm, max_credits, mtu, mps)
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, enable=True):
return await self.device.encrypt(self, enable)
async def sustain(self, timeout=None):
"""Idles the current task waiting for a disconnect or timeout"""
abort = asyncio.get_running_loop().create_future()
self.on('disconnection', abort.set_result)
self.on('disconnection_failure', abort.set_exception)
try:
await asyncio.wait_for(self.device.abort_on('flush', abort), timeout)
except asyncio.TimeoutError:
pass
self.remove_listener('disconnection', abort.set_result)
self.remove_listener('disconnection_failure', abort.set_exception)
async def update_parameters(
self,
connection_interval_min,
connection_interval_max,
max_latency,
supervision_timeout,
):
return await self.device.update_connection_parameters(
self,
connection_interval_min,
connection_interval_max,
max_latency,
supervision_timeout,
)
async def set_phy(self, tx_phys=None, rx_phys=None, phy_options=None):
return await self.device.set_connection_phy(self, tx_phys, rx_phys, phy_options)
async def get_rssi(self):
return await self.device.get_connection_rssi(self)
async def get_phy(self):
return await self.device.get_connection_phy(self)
# [Classic only]
async def request_remote_name(self):
return await self.device.request_remote_name(self)
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_value, traceback):
if exc_type is None:
try:
await self.disconnect()
except HCI_StatusError as error:
# Invalid parameter means the connection is no longer valid
if error.error_code != HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR:
raise
def __str__(self):
return (
f'Connection(handle=0x{self.handle:04X}, '
f'role={self.role_name}, '
f'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.classic_accept_any = True
self.connectable = True
self.discoverable = 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
self.gatt_services = []
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
)
self.classic_accept_any = config.get(
'classic_accept_any', self.classic_accept_any
)
self.connectable = config.get('connectable', self.connectable)
self.discoverable = config.get('discoverable', self.discoverable)
self.gatt_services = config.get('gatt_services', self.gatt_services)
# 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', encoding='utf-8') 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(f"no connection for handle: 0x{connection_handle:04x}")
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):
if connection := self.pending_connections.get(address, False):
return function(self, connection, *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 tries to convert the first argument from a bluetooth address to a
# connection
def try_with_connection_from_address(function):
@functools.wraps(function)
def wrapper(self, address, *args, **kwargs):
if connection := self.pending_connections.get(address, False):
return function(self, connection, address, *args, **kwargs)
for connection in self.connections.values():
if connection.peer_address == address:
return function(self, connection, address, *args, **kwargs)
return function(self, None, address, *args, **kwargs)
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, advertisement):
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_connection_request(self, bd_addr, class_of_device, link_type):
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.advertising_type = None
self.auto_restart_inquiry = True
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(
[l2cap.L2CAP_Information_Request.EXTENDED_FEATURE_FIXED_CHANNELS]
)
self.advertisement_accumulators = {} # Accumulators, by address
self.scanning = False
self.scanning_is_passive = False
self.discovering = False
self.le_connecting = False
self.disconnecting = False
self.connections = {} # Connections, by connection handle
self.pending_connections = {} # Connections, by BD address (BR/EDR only)
self.classic_enabled = False
self.inquiry_response = None
self.address_resolver = None
self.classic_pending_accepts = {
Address.ANY: []
} # Futures, by BD address OR [Futures] for Address.ANY
# Own address type cache
self.advertising_own_address_type = None
self.connect_own_address_type = 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 = 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
self.discoverable = config.discoverable
self.connectable = config.connectable
self.classic_accept_any = config.classic_accept_any
for service in config.gatt_services:
characteristics = []
for characteristic in service.get("characteristics", []):
descriptors = []
for descriptor in characteristic.get("descriptors", []):
new_descriptor = Descriptor(
attribute_type=descriptor["descriptor_type"],
permissions=descriptor["permission"],
)
descriptors.append(new_descriptor)
new_characteristic = Characteristic(
uuid=characteristic["uuid"],
properties=characteristic["properties"],
permissions=int(characteristic["permissions"], 0),
descriptors=descriptors,
)
characteristics.append(new_characteristic)
new_service = Service(uuid=service["uuid"], characteristics=characteristics)
self.gatt_server.add_service(new_service)
# 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 isinstance(address, str):
address = Address(address)
self.random_address = address
# Setup SMP
self.smp_manager = smp.Manager(self)
self.l2cap_channel_manager.register_fixed_channel(smp.SMP_CID, self.on_smp_pdu)
self.l2cap_channel_manager.register_fixed_channel(
smp.SMP_BR_CID, self.on_smp_pdu
)
# Register the SDP server with the L2CAP Channel Manager
self.sdp_server.register(self.l2cap_channel_manager)
self.add_default_services(generic_access_service)
self.l2cap_channel_manager.register_fixed_channel(ATT_CID, self.on_gatt_pdu)
# 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
return None
def find_connection_by_bd_addr(
self, bd_addr, transport=None, check_address_type=False
):
for connection in self.connections.values():
if connection.peer_address.to_bytes() == bd_addr.to_bytes():
if (
check_address_type
and connection.peer_address.address_type != bd_addr.address_type
):
continue
if transport is None or connection.transport == transport:
return connection
return None
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 register_l2cap_server(self, psm, server):
self.l2cap_channel_manager.register_server(psm, server)
def register_l2cap_channel_server(
self,
psm,
server,
max_credits=DEVICE_DEFAULT_L2CAP_COC_MAX_CREDITS,
mtu=DEVICE_DEFAULT_L2CAP_COC_MTU,
mps=DEVICE_DEFAULT_L2CAP_COC_MPS,
):
return self.l2cap_channel_manager.register_le_coc_server(
psm, server, max_credits, mtu, mps
)
async def open_l2cap_channel(
self,
connection,
psm,
max_credits=DEVICE_DEFAULT_L2CAP_COC_MAX_CREDITS,
mtu=DEVICE_DEFAULT_L2CAP_COC_MTU,
mps=DEVICE_DEFAULT_L2CAP_COC_MPS,
):
return await self.l2cap_channel_manager.open_le_coc(
connection, psm, max_credits, mtu, mps
)
def send_l2cap_pdu(self, connection_handle, cid, pdu):
self.host.send_l2cap_pdu(connection_handle, cid, pdu)
async def send_command(self, command, check_result=False):
try:
return await asyncio.wait_for(
self.host.send_command(command, check_result), self.command_timeout
)
except asyncio.TimeoutError as error:
logger.warning('!!! Command timed out')
raise CommandTimeoutError() from error
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
if self.host.supports_command(HCI_WRITE_LE_HOST_SUPPORT_COMMAND):
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),
check_result=True,
)
# Load the address resolving list
if self.keystore and self.host.supports_command(
HCI_LE_CLEAR_RESOLVING_LIST_COMMAND
):
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)
)
)
await self.set_connectable(self.connectable)
await self.set_discoverable(self.discoverable)
# 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
def supports_le_feature(self, feature):
return self.host.supports_le_feature(feature)
def supports_le_phy(self, phy):
if phy == HCI_LE_1M_PHY:
return True
feature_map = {
HCI_LE_2M_PHY: HCI_LE_2M_PHY_LE_SUPPORTED_FEATURE,
HCI_LE_CODED_PHY: HCI_LE_CODED_PHY_LE_SUPPORTED_FEATURE,
}
if phy not in feature_map:
raise ValueError('invalid PHY')
return self.host.supports_le_feature(feature_map[phy])
async def start_advertising(
self,
advertising_type=AdvertisingType.UNDIRECTED_CONNECTABLE_SCANNABLE,
target=None,
own_address_type=OwnAddressType.RANDOM,
auto_restart=False,
):
# If we're advertising, stop first
if self.advertising:
await self.stop_advertising()
# Set/update the advertising data if the advertising type allows it
if advertising_type.has_data:
await self.send_command(
HCI_LE_Set_Advertising_Data_Command(
advertising_data=self.advertising_data
),
check_result=True,
)
# Set/update the scan response data if the advertising is scannable
if advertising_type.is_scannable:
await self.send_command(
HCI_LE_Set_Scan_Response_Data_Command(
scan_response_data=self.scan_response_data
),
check_result=True,
)
# Decide what peer address to use
if advertising_type.is_directed:
if target is None:
raise ValueError('directed advertising requires a target address')
peer_address = target
peer_address_type = target.address_type
else:
peer_address = Address('00:00:00:00:00:00')
peer_address_type = Address.PUBLIC_DEVICE_ADDRESS
# Set the advertising parameters
await self.send_command(
HCI_LE_Set_Advertising_Parameters_Command(
advertising_interval_min=self.advertising_interval_min,
advertising_interval_max=self.advertising_interval_max,
advertising_type=int(advertising_type),
own_address_type=own_address_type,
peer_address_type=peer_address_type,
peer_address=peer_address,
advertising_channel_map=7,
advertising_filter_policy=0,
),
check_result=True,
)
# Enable advertising
await self.send_command(
HCI_LE_Set_Advertising_Enable_Command(advertising_enable=1),
check_result=True,
)
self.advertising_own_address_type = own_address_type
self.auto_restart_advertising = auto_restart
self.advertising_type = advertising_type
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),
check_result=True,
)
self.advertising_own_address_type = None
self.advertising = False
self.advertising_type = None
self.auto_restart_advertising = False
@property
def is_advertising(self):
return self.advertising
async def start_scanning(
self,
legacy=False,
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=OwnAddressType.RANDOM,
filter_duplicates=False,
scanning_phys=(HCI_LE_1M_PHY, HCI_LE_CODED_PHY),
):
# 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')
# Reset the accumulators
self.advertisement_accumulators = {}
# Enable scanning
if not legacy and self.supports_le_feature(
HCI_LE_EXTENDED_ADVERTISING_LE_SUPPORTED_FEATURE
):
# Set the scanning parameters
scan_type = (
HCI_LE_Set_Extended_Scan_Parameters_Command.ACTIVE_SCANNING
if active
else HCI_LE_Set_Extended_Scan_Parameters_Command.PASSIVE_SCANNING
)
scanning_filter_policy = (
HCI_LE_Set_Extended_Scan_Parameters_Command.BASIC_UNFILTERED_POLICY
) # TODO: support other types
scanning_phy_count = 0
scanning_phys_bits = 0
if HCI_LE_1M_PHY in scanning_phys:
scanning_phys_bits |= 1 << HCI_LE_1M_PHY_BIT
scanning_phy_count += 1
if HCI_LE_CODED_PHY in scanning_phys:
if self.supports_le_feature(HCI_LE_CODED_PHY_LE_SUPPORTED_FEATURE):
scanning_phys_bits |= 1 << HCI_LE_CODED_PHY_BIT
scanning_phy_count += 1
if scanning_phy_count == 0:
raise ValueError('at least one scanning PHY must be enabled')
await self.send_command(
HCI_LE_Set_Extended_Scan_Parameters_Command(
own_address_type=own_address_type,
scanning_filter_policy=scanning_filter_policy,
scanning_phys=scanning_phys_bits,
scan_types=[scan_type] * scanning_phy_count,
scan_intervals=[int(scan_window / 0.625)] * scanning_phy_count,
scan_windows=[int(scan_window / 0.625)] * scanning_phy_count,
),
check_result=True,
)
# Enable scanning
await self.send_command(
HCI_LE_Set_Extended_Scan_Enable_Command(
enable=1,
filter_duplicates=1 if filter_duplicates else 0,
duration=0, # TODO allow other values
period=0, # TODO allow other values
),
check_result=True,
)
else:
# 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(
# pylint: disable=line-too-long
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,
),
check_result=True,
)
# Enable scanning
await self.send_command(
HCI_LE_Set_Scan_Enable_Command(
le_scan_enable=1, filter_duplicates=1 if filter_duplicates else 0
),
check_result=True,
)
self.scanning_is_passive = not active
self.scanning = True
async def stop_scanning(self):
# Disable scanning
if self.supports_le_feature(HCI_LE_EXTENDED_ADVERTISING_LE_SUPPORTED_FEATURE):
await self.send_command(
HCI_LE_Set_Extended_Scan_Enable_Command(
enable=0, filter_duplicates=0, duration=0, period=0
),
check_result=True,
)
else:
await self.send_command(
HCI_LE_Set_Scan_Enable_Command(le_scan_enable=0, filter_duplicates=0),
check_result=True,
)
self.scanning = False
@property
def is_scanning(self):
return self.scanning
@host_event_handler
def on_advertising_report(self, report):
if not (accumulator := self.advertisement_accumulators.get(report.address)):
accumulator = AdvertisementDataAccumulator(passive=self.scanning_is_passive)
self.advertisement_accumulators[report.address] = accumulator
if advertisement := accumulator.update(report):
self.emit('advertisement', advertisement)
async def start_discovery(self, auto_restart=True):
await self.send_command(
HCI_Write_Inquiry_Mode_Command(inquiry_mode=HCI_EXTENDED_INQUIRY_MODE),
check_result=True,
)
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 HCI_StatusError(response)
self.auto_restart_inquiry = auto_restart
self.discovering = True
async def stop_discovery(self):
if self.discovering:
await self.send_command(HCI_Inquiry_Cancel_Command(), check_result=True)
self.auto_restart_inquiry = True
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.send_command(
HCI_Write_Extended_Inquiry_Response_Command(
fec_required=0, extended_inquiry_response=self.inquiry_response
),
check_result=True,
)
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,
connection_parameters_preferences=None,
own_address_type=OwnAddressType.RANDOM,
timeout=DEVICE_DEFAULT_CONNECT_TIMEOUT,
):
'''
Request a connection to a peer.
When transport is BLE, this method cannot be called if there is already a
pending connection.
connection_parameters_preferences: (BLE only, ignored for BR/EDR)
* None: use all PHYs with default parameters
* map: each entry has a PHY as key and a ConnectionParametersPreferences
object as value
own_address_type: (BLE only)
'''
# Check parameters
if transport not in (BT_LE_TRANSPORT, BT_BR_EDR_TRANSPORT):
raise ValueError('invalid transport')
# 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 transport == BT_LE_TRANSPORT and self.is_le_connecting:
raise InvalidStateError('connection already pending')
if isinstance(peer_address, str):
try:
peer_address = Address.from_string_for_transport(
peer_address, transport
)
except ValueError:
# If the address is not parsable, assume it is a name instead
logger.debug('looking for peer by name')
peer_address = await self.find_peer_by_name(
peer_address, transport
) # TODO: timeout
else:
# All BR/EDR addresses should be public addresses
if (
transport == BT_BR_EDR_TRANSPORT
and peer_address.address_type != Address.PUBLIC_DEVICE_ADDRESS
):
raise ValueError('BR/EDR addresses must be PUBLIC')
def on_connection(connection):
if transport == BT_LE_TRANSPORT or (
# match BR/EDR connection event against peer address
connection.transport == transport
and connection.peer_address == peer_address
):
pending_connection.set_result(connection)
def on_connection_failure(error):
if transport == BT_LE_TRANSPORT or (
# match BR/EDR connection failure event against peer address
error.transport == transport
and error.peer_address == peer_address
):
pending_connection.set_exception(error)
# Create a future so that we can wait for the connection's result
pending_connection = asyncio.get_running_loop().create_future()
self.on('connection', on_connection)
self.on('connection_failure', on_connection_failure)
try:
# Tell the controller to connect
if transport == BT_LE_TRANSPORT:
if connection_parameters_preferences is None:
if connection_parameters_preferences is None:
connection_parameters_preferences = {
HCI_LE_1M_PHY: ConnectionParametersPreferences.default,
HCI_LE_2M_PHY: ConnectionParametersPreferences.default,
HCI_LE_CODED_PHY: ConnectionParametersPreferences.default,
}
self.connect_own_address_type = own_address_type
if self.host.supports_command(
HCI_LE_EXTENDED_CREATE_CONNECTION_COMMAND
):
# Only keep supported PHYs
phys = sorted(
list(
set(
filter(
self.supports_le_phy,
connection_parameters_preferences.keys(),
)
)
)
)
if not phys:
raise ValueError('least one supported PHY needed')
phy_count = len(phys)
initiating_phys = phy_list_to_bits(phys)
connection_interval_mins = [
int(
connection_parameters_preferences[
phy
].connection_interval_min
/ 1.25
)
for phy in phys
]
connection_interval_maxs = [
int(
connection_parameters_preferences[
phy
].connection_interval_max
/ 1.25
)
for phy in phys
]
max_latencies = [
connection_parameters_preferences[phy].max_latency
for phy in phys
]
supervision_timeouts = [
int(
connection_parameters_preferences[phy].supervision_timeout
/ 10
)
for phy in phys
]
min_ce_lengths = [
int(
connection_parameters_preferences[phy].min_ce_length / 0.625
)
for phy in phys
]
max_ce_lengths = [
int(
connection_parameters_preferences[phy].max_ce_length / 0.625
)
for phy in phys
]
result = await self.send_command(
HCI_LE_Extended_Create_Connection_Command(
initiator_filter_policy=0,
own_address_type=own_address_type,
peer_address_type=peer_address.address_type,
peer_address=peer_address,
initiating_phys=initiating_phys,
scan_intervals=(
int(DEVICE_DEFAULT_CONNECT_SCAN_INTERVAL / 0.625),
)
* phy_count,
scan_windows=(
int(DEVICE_DEFAULT_CONNECT_SCAN_WINDOW / 0.625),
)
* phy_count,
connection_interval_mins=connection_interval_mins,
connection_interval_maxs=connection_interval_maxs,
max_latencies=max_latencies,
supervision_timeouts=supervision_timeouts,
min_ce_lengths=min_ce_lengths,
max_ce_lengths=max_ce_lengths,
)
)
else:
if HCI_LE_1M_PHY not in connection_parameters_preferences:
raise ValueError('1M PHY preferences required')
prefs = connection_parameters_preferences[HCI_LE_1M_PHY]
result = await self.send_command(
HCI_LE_Create_Connection_Command(
le_scan_interval=int(
DEVICE_DEFAULT_CONNECT_SCAN_INTERVAL / 0.625
),
le_scan_window=int(
DEVICE_DEFAULT_CONNECT_SCAN_WINDOW / 0.625
),
initiator_filter_policy=0,
peer_address_type=peer_address.address_type,
peer_address=peer_address,
own_address_type=own_address_type,
connection_interval_min=int(
prefs.connection_interval_min / 1.25
),
connection_interval_max=int(
prefs.connection_interval_max / 1.25
),
max_latency=prefs.max_latency,
supervision_timeout=int(prefs.supervision_timeout / 10),
min_ce_length=int(prefs.min_ce_length / 0.625),
max_ce_length=int(prefs.max_ce_length / 0.625),
)
)
else:
# Save pending connection
self.pending_connections[peer_address] = Connection.incomplete(
self, peer_address
)
# TODO: allow passing other settings
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,
)
)
if result.status != HCI_Command_Status_Event.PENDING:
raise HCI_StatusError(result)
# Wait for the connection process to complete
if transport == BT_LE_TRANSPORT:
self.le_connecting = True
if timeout is None:
return await self.abort_on('flush', pending_connection)
try:
return await asyncio.wait_for(
asyncio.shield(pending_connection), timeout
)
except asyncio.TimeoutError:
if transport == BT_LE_TRANSPORT:
await self.send_command(HCI_LE_Create_Connection_Cancel_Command())
else:
await self.send_command(
HCI_Create_Connection_Cancel_Command(bd_addr=peer_address)
)
try:
return await self.abort_on('flush', pending_connection)
except ConnectionError as error:
raise core.TimeoutError() from error
finally:
self.remove_listener('connection', on_connection)
self.remove_listener('connection_failure', on_connection_failure)
if transport == BT_LE_TRANSPORT:
self.le_connecting = False
self.connect_own_address_type = None
else:
self.pending_connections.pop(peer_address, None)
async def accept(
self,
peer_address=Address.ANY,
role=BT_PERIPHERAL_ROLE,
timeout=DEVICE_DEFAULT_CONNECT_TIMEOUT,
):
'''
Wait and accept any incoming connection or a connection from `peer_address` when
set.
Notes:
* A `connect` to the same peer will also complete this call.
* The `timeout` parameter is only handled while waiting for the connection
request, once received and accepted, the controller shall issue a connection
complete event.
'''
if isinstance(peer_address, str):
try:
peer_address = Address(peer_address)
except ValueError:
# If the address is not parsable, assume it is a name instead
logger.debug('looking for peer by name')
peer_address = await self.find_peer_by_name(
peer_address, BT_BR_EDR_TRANSPORT
) # TODO: timeout
if peer_address == Address.NIL:
raise ValueError('accept on nil address')
# Create a future so that we can wait for the request
pending_request = asyncio.get_running_loop().create_future()
if peer_address == Address.ANY:
self.classic_pending_accepts[Address.ANY].append(pending_request)
elif peer_address in self.classic_pending_accepts:
raise InvalidStateError('accept connection already pending')
else:
self.classic_pending_accepts[peer_address] = pending_request
try:
# Wait for a request or a completed connection
pending_request = self.abort_on('flush', pending_request)
result = await (
asyncio.wait_for(pending_request, timeout)
if timeout
else pending_request
)
except Exception:
# Remove future from device context
if peer_address == Address.ANY:
self.classic_pending_accepts[Address.ANY].remove(pending_request)
else:
self.classic_pending_accepts.pop(peer_address)
raise
# Result may already be a completed connection,
# see `on_connection` for details
if isinstance(result, Connection):
return result
# Otherwise, result came from `on_connection_request`
peer_address, _class_of_device, _link_type = result
# Create a future so that we can wait for the connection's result
pending_connection = asyncio.get_running_loop().create_future()
def on_connection(connection):
if (
connection.transport == BT_BR_EDR_TRANSPORT
and connection.peer_address == peer_address
):
pending_connection.set_result(connection)
def on_connection_failure(error):
if (
error.transport == BT_BR_EDR_TRANSPORT
and error.peer_address == peer_address
):
pending_connection.set_exception(error)
self.on('connection', on_connection)
self.on('connection_failure', on_connection_failure)
# Save pending connection
self.pending_connections[peer_address] = Connection.incomplete(
self, peer_address
)
try:
# Accept connection request
await self.send_command(
HCI_Accept_Connection_Request_Command(bd_addr=peer_address, role=role)
)
# Wait for connection complete
return await self.abort_on('flush', pending_connection)
finally:
self.remove_listener('connection', on_connection)
self.remove_listener('connection_failure', on_connection_failure)
self.pending_connections.pop(peer_address, None)
@asynccontextmanager
async def connect_as_gatt(self, peer_address):
async with AsyncExitStack() as stack:
connection = await stack.enter_async_context(
await self.connect(peer_address)
)
peer = await stack.enter_async_context(Peer(connection))
yield peer
@property
def is_le_connecting(self):
return self.le_connecting
@property
def is_disconnecting(self):
return self.disconnecting
async def cancel_connection(self, peer_address=None):
# Low-energy: cancel ongoing connection
if peer_address is None:
if not self.is_le_connecting:
return
await self.send_command(
HCI_LE_Create_Connection_Cancel_Command(), check_result=True
)
# BR/EDR: try to cancel to ongoing connection
# NOTE: This API does not prevent from trying to cancel a connection which is
# not currently being created
else:
if isinstance(peer_address, str):
try:
peer_address = Address(peer_address)
except ValueError:
# If the address is not parsable, assume it is a name instead
logger.debug('looking for peer by name')
peer_address = await self.find_peer_by_name(
peer_address, BT_BR_EDR_TRANSPORT
) # TODO: timeout
await self.send_command(
HCI_Create_Connection_Cancel_Command(bd_addr=peer_address),
check_result=True,
)
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 HCI_StatusError(result)
# Wait for the disconnection process to complete
self.disconnecting = True
return await self.abort_on('flush', 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,
connection_interval_min,
connection_interval_max,
max_latency,
supervision_timeout,
min_ce_length=0,
max_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,
connection_interval_min=connection_interval_min,
connection_interval_max=connection_interval_max,
max_latency=max_latency,
supervision_timeout=supervision_timeout,
min_ce_length=min_ce_length,
max_ce_length=max_ce_length,
),
check_result=True,
)
async def get_connection_rssi(self, connection):
result = await self.send_command(
HCI_Read_RSSI_Command(handle=connection.handle), check_result=True
)
return result.return_parameters.rssi
async def get_connection_phy(self, connection):
result = await self.send_command(
HCI_LE_Read_PHY_Command(connection_handle=connection.handle),
check_result=True,
)
return (result.return_parameters.tx_phy, result.return_parameters.rx_phy)
async def set_connection_phy(
self, connection, tx_phys=None, rx_phys=None, phy_options=None
):
all_phys_bits = (1 if tx_phys is None else 0) | (
(1 if rx_phys is None else 0) << 1
)
return await self.send_command(
HCI_LE_Set_PHY_Command(
connection_handle=connection.handle,
all_phys=all_phys_bits,
tx_phys=phy_list_to_bits(tx_phys),
rx_phys=phy_list_to_bits(rx_phys),
phy_options=0 if phy_options is None else int(phy_options),
),
check_result=True,
)
async def set_default_phy(self, tx_phys=None, rx_phys=None):
all_phys_bits = (1 if tx_phys is None else 0) | (
(1 if rx_phys is None else 0) << 1
)
return await self.send_command(
HCI_LE_Set_Default_PHY_Command(
all_phys=all_phys_bits,
tx_phys=phy_list_to_bits(tx_phys),
rx_phys=phy_list_to_bits(rx_phys),
),
check_result=True,
)
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, raw=True)
if local_name is None:
local_name = ad_data.get(AdvertisingData.SHORTENED_LOCAL_NAME, raw=True)
if local_name is not None:
if local_name.decode('utf-8') == name:
peer_address.set_result(address)
handler = None
was_scanning = self.scanning
was_discovering = self.discovering
try:
if transport == BT_LE_TRANSPORT:
event_name = 'advertisement'
handler = self.on(
event_name,
lambda advertisement: on_peer_found(
advertisement.address, advertisement.data
),
)
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
),
)
if not self.discovering:
await self.start_discovery()
else:
return None
return await self.abort_on('flush', 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
if connection.role == BT_CENTRAL_ROLE and keys.ltk_central:
return keys.ltk_central.value
if 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.warning(
'HCI_Authentication_Requested_Command failed: '
f'{HCI_Constant.error_name(result.status)}'
)
raise HCI_StatusError(result)
# Wait for the authentication to complete
await connection.abort_on('disconnection', pending_authentication)
finally:
connection.remove_listener('connection_authentication', on_authentication)
connection.remove_listener(
'connection_authentication_failure', on_authentication_failure
)
async def encrypt(self, connection, enable=True):
if not enable and connection.transport == BT_LE_TRANSPORT:
raise ValueError('`enable` parameter is classic only.')
# 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_Enable_Encryption_Command(
connection_handle=connection.handle,
random_number=rand,
encrypted_diversifier=ediv,
long_term_key=ltk,
)
)
if result.status != HCI_COMMAND_STATUS_PENDING:
logger.warning(
'HCI_LE_Enable_Encryption_Command failed: '
f'{HCI_Constant.error_name(result.status)}'
)
raise HCI_StatusError(result)
else:
result = await self.send_command(
HCI_Set_Connection_Encryption_Command(
connection_handle=connection.handle,
encryption_enable=0x01 if enable else 0x00,
)
)
if result.status != HCI_COMMAND_STATUS_PENDING:
logger.warning(
'HCI_Set_Connection_Encryption_Command failed: '
f'{HCI_Constant.error_name(result.status)}'
)
raise HCI_StatusError(result)
# Wait for the result
await connection.abort_on('disconnection', 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, remote): # remote: Connection | Address
# Set up event handlers
pending_name = asyncio.get_running_loop().create_future()
peer_address = remote if isinstance(remote, Address) else remote.peer_address
handler = self.on(
'remote_name',
lambda address, remote_name: pending_name.set_result(remote_name)
if address == peer_address
else None,
)
failure_handler = self.on(
'remote_name_failure',
lambda address, error_code: pending_name.set_exception(
HCI_Error(error_code)
)
if address == peer_address
else None,
)
try:
result = await self.send_command(
HCI_Remote_Name_Request_Command(
bd_addr=peer_address,
# TODO investigate other options
page_scan_repetition_mode=HCI_Remote_Name_Request_Command.R0,
reserved=0,
clock_offset=0, # TODO investigate non-0 values
)
)
if result.status != HCI_COMMAND_STATUS_PENDING:
logger.warning(
'HCI_Set_Connection_Encryption_Command failed: '
f'{HCI_Constant.error_name(result.status)}'
)
raise HCI_StatusError(result)
# Wait for the result
return await self.abort_on('flush', pending_name)
finally:
self.remove_listener('remote_name', handler)
self.remove_listener('remote_name_failure', failure_handler)
@host_event_handler
def on_flush(self):
self.emit('flush')
for _, connection in self.connections.items():
connection.emit('disconnection', 0)
self.connections = {}
# [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 = PairingKeys()
pairing_keys.link_key = PairingKeys.Key(value=link_key)
async def store_keys():
try:
await self.keystore.update(str(bd_addr), pairing_keys)
except Exception as error:
logger.warning(f'!!! error while storing keys: {error}')
self.abort_on('flush', store_keys())
if connection := self.find_connection_by_bd_addr(
bd_addr, transport=BT_BR_EDR_TRANSPORT
):
connection.link_key_type = key_type
def add_service(self, service):
self.gatt_server.add_service(service)
def add_services(self, services):
self.gatt_server.add_services(services)
def add_default_services(self, generic_access_service=True):
# Add a GAP Service if requested
if generic_access_service:
self.gatt_server.add_service(GenericAccessService(self.name))
async def notify_subscriber(self, connection, attribute, value=None, force=False):
await self.gatt_server.notify_subscriber(connection, attribute, value, force)
async def notify_subscribers(self, attribute, value=None, force=False):
await self.gatt_server.notify_subscribers(attribute, value, force)
async def indicate_subscriber(self, connection, attribute, value=None, force=False):
await self.gatt_server.indicate_subscriber(connection, attribute, value, force)
async def indicate_subscribers(self, attribute, value=None, force=False):
await self.gatt_server.indicate_subscribers(attribute, value, force)
@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}] '
f'{peer_address} as {HCI_Constant.role_name(role)}'
)
if connection_handle in self.connections:
logger.warning(
'new connection reuses the same handle as a previous connection'
)
if transport == BT_BR_EDR_TRANSPORT:
# Create a new connection
connection: Connection = self.pending_connections.pop(peer_address)
connection.complete(
connection_handle, peer_resolvable_address, role, connection_parameters
)
self.connections[connection_handle] = connection
# We may have an accept ongoing waiting for a connection request for
# `peer_address`.
# Typically happen when using `connect` to the same `peer_address` we are
# waiting for with an `accept`.
# In this case, set the completed `connection` to the `accept` future
# result.
if peer_address in self.classic_pending_accepts:
future = self.classic_pending_accepts.pop(peer_address)
future.set_result(connection)
# Emit an event to notify listeners of the new connection
self.emit('connection', connection)
else:
# 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
# Guess which own address type is used for this connection.
# This logic is somewhat correct but may need to be improved
# when multiple advertising are run simultaneously.
if self.connect_own_address_type is not None:
own_address_type = self.connect_own_address_type
else:
own_address_type = self.advertising_own_address_type
# We are no longer advertising
self.advertising_own_address_type = None
self.advertising = False
# Create and notify of the new connection asynchronously
async def new_connection():
# Figure out which PHY we're connected with
if self.host.supports_command(HCI_LE_READ_PHY_COMMAND):
result = await asyncio.shield(
self.send_command(
HCI_LE_Read_PHY_Command(
connection_handle=connection_handle
),
check_result=True,
)
)
phy = ConnectionPHY(
result.return_parameters.tx_phy, result.return_parameters.rx_phy
)
else:
phy = ConnectionPHY(HCI_LE_1M_PHY, HCI_LE_1M_PHY)
self_address = self.random_address
if own_address_type in (
OwnAddressType.PUBLIC,
OwnAddressType.RESOLVABLE_OR_PUBLIC,
):
self_address = self.public_address
# Create a new connection
connection = Connection(
self,
connection_handle,
transport,
self_address,
peer_address,
peer_resolvable_address,
role,
connection_parameters,
phy,
)
self.connections[connection_handle] = connection
# Emit an event to notify listeners of the new connection
self.emit('connection', connection)
self.abort_on('flush', new_connection())
@host_event_handler
def on_connection_failure(self, transport, peer_address, error_code):
logger.debug(f'*** Connection failed: {HCI_Constant.error_name(error_code)}')
# For directed advertising, this means a timeout
if (
transport == BT_LE_TRANSPORT
and self.advertising
and self.advertising_type.is_directed
):
self.advertising_own_address_type = None
self.advertising = False
# Notify listeners
error = ConnectionError(
error_code,
transport,
peer_address,
'hci',
HCI_Constant.error_name(error_code),
)
self.emit('connection_failure', error)
# FIXME: Explore a delegate-model for BR/EDR wait connection #56.
@host_event_handler
def on_connection_request(self, bd_addr, class_of_device, link_type):
logger.debug(f'*** Connection request: {bd_addr}')
# match a pending future using `bd_addr`
if bd_addr in self.classic_pending_accepts:
future = self.classic_pending_accepts.pop(bd_addr)
future.set_result((bd_addr, class_of_device, link_type))
# match first pending future for ANY address
elif len(self.classic_pending_accepts[Address.ANY]) > 0:
future = self.classic_pending_accepts[Address.ANY].pop(0)
future.set_result((bd_addr, class_of_device, link_type))
# device configuration is set to accept any incoming connection
elif self.classic_accept_any:
# Save pending connection
self.pending_connections[bd_addr] = Connection.incomplete(self, bd_addr)
self.host.send_command_sync(
HCI_Accept_Connection_Request_Command(
bd_addr=bd_addr, role=0x01 # Remain the peripheral
)
)
# reject incoming connection
else:
self.host.send_command_sync(
HCI_Reject_Connection_Request_Command(
bd_addr=bd_addr,
reason=HCI_CONNECTION_REJECTED_DUE_TO_LIMITED_RESOURCES_ERROR,
)
)
@host_event_handler
@with_connection_from_handle
def on_disconnection(self, connection, reason):
logger.debug(
f'*** Disconnection: [0x{connection.handle:04X}] '
f'{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')
self.abort_on(
'flush',
self.start_advertising(
advertising_type=self.advertising_type, auto_restart=True
),
)
@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,
connection.transport,
connection.peer_address,
'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.auto_restart_inquiry:
# Inquire again
await self.start_discovery(auto_restart=True)
else:
self.auto_restart_inquiry = True
self.discovering = False
self.emit('inquiry_complete')
@host_event_handler
@with_connection_from_handle
def on_connection_authentication(self, connection):
logger.debug(
f'*** Connection Authentication: [0x{connection.handle:04X}] '
f'{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}] '
f'{connection.peer_address} as {connection.role_name}, error={error}'
)
connection.emit('connection_authentication_failure', error)
@host_event_handler
@with_connection_from_address
def on_ssp_complete(self, connection):
# On Secure Simple Pairing complete, in case:
# - Connection isn't already authenticated
# - AND we are not the initiator of the authentication
# We must trigger authentication to known if we are truly authenticated
if not connection.authenticating and not connection.authenticated:
logger.debug(
f'*** Trigger Connection Authentication: [0x{connection.handle:04X}] '
f'{connection.peer_address}'
)
asyncio.create_task(connection.authenticate())
# [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
# pylint: disable=line-too-long
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_compare = pairing_config.delegate.io_capability not in (
smp.SMP_NO_INPUT_NO_OUTPUT_IO_CAPABILITY,
smp.SMP_DISPLAY_ONLY_IO_CAPABILITY,
)
# Respond
if can_compare:
async def compare_numbers():
numbers_match = await connection.abort_on(
'disconnection',
pairing_config.delegate.compare_numbers(code, digits=6),
)
if numbers_match:
await self.host.send_command(
HCI_User_Confirmation_Request_Reply_Command(
bd_addr=connection.peer_address
)
)
else:
await self.host.send_command(
HCI_User_Confirmation_Request_Negative_Reply_Command(
bd_addr=connection.peer_address
)
)
asyncio.create_task(compare_numbers())
else:
async def confirm():
confirm = await connection.abort_on(
'disconnection', pairing_config.delegate.confirm()
)
if confirm:
await self.host.send_command(
HCI_User_Confirmation_Request_Reply_Command(
bd_addr=connection.peer_address
)
)
else:
await self.host.send_command(
HCI_User_Confirmation_Request_Negative_Reply_Command(
bd_addr=connection.peer_address
)
)
asyncio.create_task(confirm())
# [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:
async def get_number():
number = await connection.abort_on(
'disconnection', pairing_config.delegate.get_number()
)
if number is not None:
await self.host.send_command(
HCI_User_Passkey_Request_Reply_Command(
bd_addr=connection.peer_address, numeric_value=number
)
)
else:
await self.host.send_command(
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_authentication_user_passkey_notification(self, connection, passkey):
# Ask what the pairing config should be for this connection
pairing_config = self.pairing_config_factory(connection)
connection.abort_on(
'disconnection', pairing_config.delegate.display_number(passkey)
)
# [Classic only]
@host_event_handler
@try_with_connection_from_address
def on_remote_name(self, connection, address, remote_name):
# Try to decode the name
try:
remote_name = remote_name.decode('utf-8')
if connection:
connection.peer_name = remote_name
connection.emit('remote_name')
self.emit('remote_name', address, remote_name)
except UnicodeDecodeError as error:
logger.warning('peer name is not valid UTF-8')
if connection:
connection.emit('remote_name_failure', error)
else:
self.emit('remote_name_failure', address, error)
# [Classic only]
@host_event_handler
@try_with_connection_from_address
def on_remote_name_failure(self, connection, address, error):
if connection:
connection.emit('remote_name_failure', error)
self.emit('remote_name_failure', address, 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}] '
f'{connection.peer_address} as {connection.role_name}, '
f'encryption={encryption}'
)
connection.encryption = encryption
if (
not connection.authenticated
and encryption == HCI_Encryption_Change_Event.AES_CCM
):
connection.authenticated = True
connection.sc = True
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}] '
f'{connection.peer_address} as {connection.role_name}, '
f'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}] '
f'{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}] '
f'{connection.peer_address} as {connection.role_name}, '
f'{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}] '
f'{connection.peer_address} as {connection.role_name}, '
f'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}] '
f'{connection.peer_address} as {connection.role_name}, '
f'{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}] '
f'{connection.peer_address} as {connection.role_name}, '
f'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}] '
f'{connection.peer_address} as {connection.role_name}, '
f'{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}] '
f'{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, sc):
connection.sc = sc
connection.authenticated = True
connection.emit('pairing', keys)
@with_connection_from_handle
def on_pairing_failure(self, connection, reason):
connection.emit('pairing_failure', reason)
@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.warning(
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.warning(
color('no GATT server for connection 0x{connection_handle:04X}')
)
return
connection.gatt_server.on_gatt_pdu(connection, att_pdu)
@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}", '
f'random_address="{self.random_address}", '
f'public_address="{self.public_address}")'
)