pstruebi/bumble_fork
1
0
Fork 0
forked from auracaster/bumble_mirror
Code Pull Requests Activity

Merge pull request #679 from google/gbg/pairing-ios

Browse Source
This commit is contained in:
Gilles Boccon-Gibod
2025-05-05 09:50:49 -07:00
committed by GitHub
parent 6926d5cb70 8b59b4f515
commit 8e84b528ce
7 changed files with 316 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
apps/bench.py
View File

@@ -1256,6 +1256,7 @@ class Central(Connection.Listener):
self.device.classic_enabled = self.classic
# Set up a pairing config factory with minimal requirements.
self.device.config.keystore = "JsonKeyStore"
self.device.pairing_config_factory = lambda _: PairingConfig(
sc=False, mitm=False, bonding=False
)
@@ -1408,6 +1409,7 @@ class Peripheral(Device.Listener, Connection.Listener):
self.device.classic_enabled = self.classic
# Set up a pairing config factory with minimal requirements.
self.device.config.keystore = "JsonKeyStore"
self.device.pairing_config_factory = lambda _: PairingConfig(
sc=False, mitm=False, bonding=False
)

217
apps/pair.py
View File

@@ -18,9 +18,12 @@
import asyncio
import os
import logging
import struct
import click
from prompt_toolkit.shortcuts import PromptSession
from bumble.a2dp import make_audio_sink_service_sdp_records
from bumble.colors import color
from bumble.device import Device, Peer
from bumble.transport import open_transport_or_link
@@ -30,16 +33,20 @@ from bumble.smp import error_name as smp_error_name
from bumble.keys import JsonKeyStore
from bumble.core import (
AdvertisingData,
Appearance,
ProtocolError,
PhysicalTransport,
UUID,
)
from bumble.gatt import (
GATT_DEVICE_NAME_CHARACTERISTIC,
GATT_GENERIC_ACCESS_SERVICE,
GATT_HEART_RATE_SERVICE,
GATT_HEART_RATE_MEASUREMENT_CHARACTERISTIC,
Service,
Characteristic,
CharacteristicValue,
)
from bumble.hci import OwnAddressType
from bumble.att import (
ATT_Error,
ATT_INSUFFICIENT_AUTHENTICATION_ERROR,
@@ -62,7 +69,7 @@ class Waiter:
self.linger = linger
def terminate(self):
if not self.linger:
if not self.linger and not self.done.done:
self.done.set_result(None)
async def wait_until_terminated(self):
@@ -193,7 +200,7 @@ class Delegate(PairingDelegate):
# -----------------------------------------------------------------------------
async def get_peer_name(peer, mode):
if mode == 'classic':
if peer.connection.transport == PhysicalTransport.BR_EDR:
return await peer.request_name()
# Try to get the peer name from GATT
@@ -225,13 +232,14 @@ def read_with_error(connection):
raise ATT_Error(ATT_INSUFFICIENT_AUTHENTICATION_ERROR)
def write_with_error(connection, _value):
if not connection.is_encrypted:
raise ATT_Error(ATT_INSUFFICIENT_ENCRYPTION_ERROR)
if not AUTHENTICATION_ERROR_RETURNED[1]:
AUTHENTICATION_ERROR_RETURNED[1] = True
raise ATT_Error(ATT_INSUFFICIENT_AUTHENTICATION_ERROR)
# -----------------------------------------------------------------------------
def sdp_records():
service_record_handle = 0x00010001
return {
service_record_handle: make_audio_sink_service_sdp_records(
service_record_handle
)
}
# -----------------------------------------------------------------------------
@@ -239,15 +247,19 @@ def on_connection(connection, request):
print(color(f'<<< Connection: {connection}', 'green'))
# Listen for pairing events
connection.on('pairing_start', on_pairing_start)
connection.on('pairing', lambda keys: on_pairing(connection, keys))
connection.on(connection.EVENT_PAIRING_START, on_pairing_start)
connection.on(connection.EVENT_PAIRING, lambda keys: on_pairing(connection, keys))
connection.on(
'pairing_failure', lambda reason: on_pairing_failure(connection, reason)
connection.EVENT_CLASSIC_PAIRING, lambda: on_classic_pairing(connection)
)
connection.on(
connection.EVENT_PAIRING_FAILURE,
lambda reason: on_pairing_failure(connection, reason),
)
# Listen for encryption changes
connection.on(
'connection_encryption_change',
connection.EVENT_CONNECTION_ENCRYPTION_CHANGE,
lambda: on_connection_encryption_change(connection),
)
@@ -288,6 +300,20 @@ async def on_pairing(connection, keys):
Waiter.instance.terminate()
# -----------------------------------------------------------------------------
@AsyncRunner.run_in_task()
async def on_classic_pairing(connection):
print(color('***-----------------------------------', 'cyan'))
print(
color(
f'*** Paired [Classic]! (peer identity={connection.peer_address})', 'cyan'
)
)
print(color('***-----------------------------------', 'cyan'))
await asyncio.sleep(POST_PAIRING_DELAY)
Waiter.instance.terminate()
# -----------------------------------------------------------------------------
@AsyncRunner.run_in_task()
async def on_pairing_failure(connection, reason):
@@ -305,6 +331,7 @@ async def pair(
mitm,
bond,
ctkd,
advertising_address,
identity_address,
linger,
io,
@@ -313,6 +340,8 @@ async def pair(
request,
print_keys,
keystore_file,
advertise_service_uuids,
advertise_appearance,
device_config,
hci_transport,
address_or_name,
@@ -328,29 +357,33 @@ async def pair(
# Expose a GATT characteristic that can be used to trigger pairing by
# responding with an authentication error when read
if mode == 'le':
device.le_enabled = True
if mode in ('le', 'dual'):
device.add_service(
Service(
'50DB505C-8AC4-4738-8448-3B1D9CC09CC5',
GATT_HEART_RATE_SERVICE,
[
Characteristic(
'552957FB-CF1F-4A31-9535-E78847E1A714',
Characteristic.Properties.READ
| Characteristic.Properties.WRITE,
Characteristic.READABLE | Characteristic.WRITEABLE,
CharacteristicValue(
read=read_with_error, write=write_with_error
),
GATT_HEART_RATE_MEASUREMENT_CHARACTERISTIC,
Characteristic.Properties.READ,
Characteristic.READ_REQUIRES_AUTHENTICATION,
bytes(1),
)
],
)
)
# Select LE or Classic
if mode == 'classic':
# LE and Classic support
if mode in ('classic', 'dual'):
device.classic_enabled = True
device.classic_smp_enabled = ctkd
if mode in ('le', 'dual'):
device.le_enabled = True
if mode == 'dual':
device.le_simultaneous_enabled = True
# Setup SDP
if mode in ('classic', 'dual'):
device.sdp_service_records = sdp_records()
# Get things going
await device.power_on()
@@ -436,13 +469,109 @@ async def pair(
print(color(f'Pairing failed: {error}', 'red'))
else:
if mode == 'le':
# Advertise so that peers can find us and connect
await device.start_advertising(auto_restart=True)
else:
if mode in ('le', 'dual'):
# Advertise so that peers can find us and connect.
# Include the heart rate service UUID in the advertisement data
# so that devices like iPhones can show this device in their
# Bluetooth selector.
service_uuids_16 = []
service_uuids_32 = []
service_uuids_128 = []
if advertise_service_uuids:
for uuid in advertise_service_uuids:
uuid = uuid.replace("-", "")
if len(uuid) == 4:
service_uuids_16.append(UUID(uuid))
elif len(uuid) == 8:
service_uuids_32.append(UUID(uuid))
elif len(uuid) == 32:
service_uuids_128.append(UUID(uuid))
else:
print(color('Invalid UUID format', 'red'))
return
else:
service_uuids_16.append(GATT_HEART_RATE_SERVICE)
flags = AdvertisingData.Flags.LE_LIMITED_DISCOVERABLE_MODE
if mode == 'le':
flags |= AdvertisingData.Flags.BR_EDR_NOT_SUPPORTED
if mode == 'dual':
flags |= AdvertisingData.Flags.SIMULTANEOUS_LE_BR_EDR_CAPABLE
ad_structs = [
(
AdvertisingData.FLAGS,
bytes([flags]),
),
(AdvertisingData.COMPLETE_LOCAL_NAME, 'Bumble'.encode()),
]
if service_uuids_16:
ad_structs.append(
(
AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
b"".join(bytes(uuid) for uuid in service_uuids_16),
)
)
if service_uuids_32:
ad_structs.append(
(
AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
b"".join(bytes(uuid) for uuid in service_uuids_32),
)
)
if service_uuids_128:
ad_structs.append(
(
AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
b"".join(bytes(uuid) for uuid in service_uuids_128),
)
)
if advertise_appearance:
advertise_appearance = advertise_appearance.upper()
try:
advertise_appearance_int = int(advertise_appearance)
except ValueError:
category, subcategory = advertise_appearance.split('/')
try:
category_enum = Appearance.Category[category]
except ValueError:
print(
color(f'Invalid appearance category {category}', 'red')
)
return
subcategory_class = Appearance.SUBCATEGORY_CLASSES[
category_enum
]
try:
subcategory_enum = subcategory_class[subcategory]
except ValueError:
print(color(f'Invalid subcategory {subcategory}', 'red'))
return
advertise_appearance_int = int(
Appearance(category_enum, subcategory_enum)
)
ad_structs.append(
(
AdvertisingData.APPEARANCE,
struct.pack('<H', advertise_appearance_int),
)
)
device.advertising_data = bytes(AdvertisingData(ad_structs))
await device.start_advertising(
auto_restart=True,
own_address_type=(
OwnAddressType.PUBLIC
if advertising_address == 'public'
else OwnAddressType.RANDOM
),
)
if mode in ('classic', 'dual'):
# Become discoverable and connectable
await device.set_discoverable(True)
await device.set_connectable(True)
print(color('Ready for connections on', 'blue'), device.public_address)
# Run until the user asks to exit
await Waiter.instance.wait_until_terminated()
@@ -462,7 +591,10 @@ class LogHandler(logging.Handler):
# -----------------------------------------------------------------------------
@click.command()
@click.option(
'--mode', type=click.Choice(['le', 'classic']), default='le', show_default=True
'--mode',
type=click.Choice(['le', 'classic', 'dual']),
default='le',
show_default=True,
)
@click.option(
'--sc',
@@ -484,6 +616,10 @@ class LogHandler(logging.Handler):
help='Enable CTKD',
show_default=True,
)
@click.option(
'--advertising-address',
type=click.Choice(['random', 'public']),
)
@click.option(
'--identity-address',
type=click.Choice(['random', 'public']),
@@ -512,9 +648,20 @@ class LogHandler(logging.Handler):
@click.option('--print-keys', is_flag=True, help='Print the bond keys before pairing')
@click.option(
'--keystore-file',
metavar='<filename>',
metavar='FILENAME',
help='File in which to store the pairing keys',
)
@click.option(
'--advertise-service-uuid',
metavar="UUID",
multiple=True,
help="Advertise a GATT service UUID (may be specified more than once)",
)
@click.option(
'--advertise-appearance',
metavar='APPEARANCE',
help='Advertise an Appearance ID (int value or string)',
)
@click.argument('device-config')
@click.argument('hci_transport')
@click.argument('address-or-name', required=False)
@@ -524,6 +671,7 @@ def main(
mitm,
bond,
ctkd,
advertising_address,
identity_address,
linger,
io,
@@ -532,6 +680,8 @@ def main(
request,
print_keys,
keystore_file,
advertise_service_uuid,
advertise_appearance,
device_config,
hci_transport,
address_or_name,
@@ -550,6 +700,7 @@ def main(
mitm,
bond,
ctkd,
advertising_address,
identity_address,
linger,
io,
@@ -558,6 +709,8 @@ def main(
request,
print_keys,
keystore_file,
advertise_service_uuid,
advertise_appearance,
device_config,
hci_transport,
address_or_name,

20
bumble/core.py
View File

@@ -809,7 +809,7 @@ class Appearance:
STICK_PC = 0x0F
class WatchSubcategory(utils.OpenIntEnum):
GENENERIC_WATCH = 0x00
GENERIC_WATCH = 0x00
SPORTS_WATCH = 0x01
SMARTWATCH = 0x02
@@ -1127,7 +1127,7 @@ class Appearance:
TURNTABLE = 0x05
CD_PLAYER = 0x06
DVD_PLAYER = 0x07
BLUERAY_PLAYER = 0x08
BLURAY_PLAYER = 0x08
OPTICAL_DISC_PLAYER = 0x09
SET_TOP_BOX = 0x0A
@@ -1351,6 +1351,12 @@ class AdvertisingData:
THREE_D_INFORMATION_DATA = 0x3D
MANUFACTURER_SPECIFIC_DATA = 0xFF
class Flags(enum.IntFlag):
LE_LIMITED_DISCOVERABLE_MODE = 1 << 0
LE_GENERAL_DISCOVERABLE_MODE = 1 << 1
BR_EDR_NOT_SUPPORTED = 1 << 2
SIMULTANEOUS_LE_BR_EDR_CAPABLE = 1 << 3
# For backward-compatibility
FLAGS = Type.FLAGS
INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS = Type.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS
@@ -1407,11 +1413,11 @@ class AdvertisingData:
THREE_D_INFORMATION_DATA = Type.THREE_D_INFORMATION_DATA
MANUFACTURER_SPECIFIC_DATA = Type.MANUFACTURER_SPECIFIC_DATA
LE_LIMITED_DISCOVERABLE_MODE_FLAG = 0x01
LE_GENERAL_DISCOVERABLE_MODE_FLAG = 0x02
BR_EDR_NOT_SUPPORTED_FLAG = 0x04
BR_EDR_CONTROLLER_FLAG = 0x08
BR_EDR_HOST_FLAG = 0x10
LE_LIMITED_DISCOVERABLE_MODE_FLAG = Flags.LE_LIMITED_DISCOVERABLE_MODE
LE_GENERAL_DISCOVERABLE_MODE_FLAG = Flags.LE_GENERAL_DISCOVERABLE_MODE
BR_EDR_NOT_SUPPORTED_FLAG = Flags.BR_EDR_NOT_SUPPORTED
BR_EDR_CONTROLLER_FLAG = Flags.SIMULTANEOUS_LE_BR_EDR_CAPABLE
BR_EDR_HOST_FLAG = 0x10 # Deprecated
ad_structures: list[tuple[int, bytes]]

77
bumble/device.py
View File

@@ -1586,9 +1586,10 @@ class Connection(utils.CompositeEventEmitter):
peer_le_features: Optional[hci.LeFeatureMask]
role: hci.Role
encryption: int
encryption_key_size: int
authenticated: bool
sc: bool
link_key_type: Optional[int]
link_key_type: Optional[int] # [Classic only]
gatt_client: gatt_client.Client
pairing_peer_io_capability: Optional[int]
pairing_peer_authentication_requirements: Optional[int]
@@ -1688,6 +1689,7 @@ class Connection(utils.CompositeEventEmitter):
self.role = role
self.parameters = parameters
self.encryption = 0
self.encryption_key_size = 0
self.authenticated = False
self.sc = False
self.link_key_type = None
@@ -1809,7 +1811,7 @@ class Connection(utils.CompositeEventEmitter):
try:
await asyncio.wait_for(
utils.cancel_on_event(self.device, 'flush', abort), timeout
utils.cancel_on_event(self.device, Device.EVENT_FLUSH, abort), timeout
)
finally:
self.remove_listener(self.EVENT_DISCONNECTION, abort.set_result)
@@ -3756,7 +3758,9 @@ class Device(utils.CompositeEventEmitter):
self.le_connecting = True
if timeout is None:
return await utils.cancel_on_event(self, 'flush', pending_connection)
return await utils.cancel_on_event(
self, Device.EVENT_FLUSH, pending_connection
)
try:
return await asyncio.wait_for(
@@ -3774,7 +3778,7 @@ class Device(utils.CompositeEventEmitter):
try:
return await utils.cancel_on_event(
self, 'flush', pending_connection
self, Device.EVENT_FLUSH, pending_connection
)
except core.ConnectionError as error:
raise core.TimeoutError() from error
@@ -3831,7 +3835,9 @@ class Device(utils.CompositeEventEmitter):
try:
# Wait for a request or a completed connection
pending_request = utils.cancel_on_event(self, 'flush', pending_request_fut)
pending_request = utils.cancel_on_event(
self, Device.EVENT_FLUSH, pending_request_fut
)
result = await (
asyncio.wait_for(pending_request, timeout)
if timeout
@@ -3893,7 +3899,9 @@ class Device(utils.CompositeEventEmitter):
)
# Wait for connection complete
return await utils.cancel_on_event(self, 'flush', pending_connection)
return await utils.cancel_on_event(
self, Device.EVENT_FLUSH, pending_connection
)
finally:
self.remove_listener(self.EVENT_CONNECTION, on_connection)
@@ -3969,7 +3977,9 @@ class Device(utils.CompositeEventEmitter):
# Wait for the disconnection process to complete
self.disconnecting = True
return await utils.cancel_on_event(self, 'flush', pending_disconnection)
return await utils.cancel_on_event(
self, Device.EVENT_FLUSH, pending_disconnection
)
finally:
connection.remove_listener(
connection.EVENT_DISCONNECTION, pending_disconnection.set_result
@@ -4193,7 +4203,7 @@ class Device(utils.CompositeEventEmitter):
else:
return None
return await utils.cancel_on_event(self, 'flush', peer_address)
return await utils.cancel_on_event(self, Device.EVENT_FLUSH, peer_address)
finally:
if listener is not None:
self.remove_listener(event_name, listener)
@@ -4243,7 +4253,7 @@ class Device(utils.CompositeEventEmitter):
if not self.scanning:
await self.start_scanning(filter_duplicates=True)
return await utils.cancel_on_event(self, 'flush', peer_address)
return await utils.cancel_on_event(self, Device.EVENT_FLUSH, peer_address)
finally:
if listener is not None:
self.remove_listener(event_name, listener)
@@ -4351,7 +4361,7 @@ class Device(utils.CompositeEventEmitter):
# Wait for the authentication to complete
await utils.cancel_on_event(
connection, 'disconnection', pending_authentication
connection, Connection.EVENT_DISCONNECTION, pending_authentication
)
finally:
connection.remove_listener(
@@ -4439,7 +4449,9 @@ class Device(utils.CompositeEventEmitter):
raise hci.HCI_StatusError(result)
# Wait for the result
await utils.cancel_on_event(connection, 'disconnection', pending_encryption)
await utils.cancel_on_event(
connection, Connection.EVENT_DISCONNECTION, pending_encryption
)
finally:
connection.remove_listener(
connection.EVENT_CONNECTION_ENCRYPTION_CHANGE, on_encryption_change
@@ -4484,7 +4496,7 @@ class Device(utils.CompositeEventEmitter):
)
raise hci.HCI_StatusError(result)
await utils.cancel_on_event(
connection, 'disconnection', pending_role_change
connection, Connection.EVENT_DISCONNECTION, pending_role_change
)
finally:
connection.remove_listener(connection.EVENT_ROLE_CHANGE, on_role_change)
@@ -4536,7 +4548,7 @@ class Device(utils.CompositeEventEmitter):
raise hci.HCI_StatusError(result)
# Wait for the result
return await utils.cancel_on_event(self, 'flush', pending_name)
return await utils.cancel_on_event(self, Device.EVENT_FLUSH, pending_name)
finally:
self.remove_listener(self.EVENT_REMOTE_NAME, handler)
self.remove_listener(self.EVENT_REMOTE_NAME_FAILURE, failure_handler)
@@ -5069,13 +5081,14 @@ class Device(utils.CompositeEventEmitter):
)
utils.cancel_on_event(
self, 'flush', self.update_keys(str(bd_addr), pairing_keys)
self, Device.EVENT_FLUSH, self.update_keys(str(bd_addr), pairing_keys)
)
if connection := self.find_connection_by_bd_addr(
bd_addr, transport=PhysicalTransport.BR_EDR
):
connection.link_key_type = key_type
connection.emit(connection.EVENT_LINK_KEY)
def add_service(self, service):
self.gatt_server.add_service(service)
@@ -5338,8 +5351,10 @@ class Device(utils.CompositeEventEmitter):
# Setup auto-restart of the advertising set if needed.
if advertising_set.auto_restart:
connection.once(
'disconnection',
lambda _: utils.cancel_on_event(self, 'flush', advertising_set.start()),
Connection.EVENT_DISCONNECTION,
lambda _: utils.cancel_on_event(
self, Device.EVENT_FLUSH, advertising_set.start()
),
)
self.emit(self.EVENT_CONNECTION, connection)
@@ -5453,8 +5468,10 @@ class Device(utils.CompositeEventEmitter):
if self.legacy_advertiser.auto_restart:
advertiser = self.legacy_advertiser
connection.once(
'disconnection',
lambda _: utils.cancel_on_event(self, 'flush', advertiser.start()),
Connection.EVENT_DISCONNECTION,
lambda _: utils.cancel_on_event(
self, Device.EVENT_FLUSH, advertiser.start()
),
)
else:
self.legacy_advertiser = None
@@ -5713,7 +5730,9 @@ class Device(utils.CompositeEventEmitter):
async def reply() -> None:
try:
if await utils.cancel_on_event(connection, 'disconnection', method()):
if await utils.cancel_on_event(
connection, Connection.EVENT_DISCONNECTION, method()
):
await self.host.send_command(
hci.HCI_User_Confirmation_Request_Reply_Command(
bd_addr=connection.peer_address
@@ -5741,7 +5760,9 @@ class Device(utils.CompositeEventEmitter):
async def reply() -> None:
try:
number = await utils.cancel_on_event(
connection, 'disconnection', pairing_config.delegate.get_number()
connection,
Connection.EVENT_DISCONNECTION,
pairing_config.delegate.get_number(),
)
if number is not None:
await self.host.send_command(
@@ -5775,7 +5796,9 @@ class Device(utils.CompositeEventEmitter):
# Ask the user to enter a string
async def get_pin_code():
pin_code = await utils.cancel_on_event(
connection, 'disconnection', pairing_config.delegate.get_string(16)
connection,
Connection.EVENT_DISCONNECTION,
pairing_config.delegate.get_string(16),
)
if pin_code is not None:
@@ -5814,7 +5837,9 @@ class Device(utils.CompositeEventEmitter):
# Show the passkey to the user
utils.cancel_on_event(
connection, 'disconnection', pairing_config.delegate.display_number(passkey)
connection,
Connection.EVENT_DISCONNECTION,
pairing_config.delegate.display_number(passkey, digits=6),
)
# [Classic only]
@@ -5950,13 +5975,17 @@ class Device(utils.CompositeEventEmitter):
@host_event_handler
@with_connection_from_handle
def on_connection_encryption_change(self, connection, encryption):
def on_connection_encryption_change(
self, connection, encryption, encryption_key_size
):
logger.debug(
f'*** Connection Encryption Change: [0x{connection.handle:04X}] '
f'{connection.peer_address} as {connection.role_name}, '
f'encryption={encryption}'
f'encryption={encryption}, '
f'key_size={encryption_key_size}'
)
connection.encryption = encryption
connection.encryption_key_size = encryption_key_size
if (
not connection.authenticated
and connection.transport == PhysicalTransport.BR_EDR

42
bumble/hci.py
View File

@@ -29,13 +29,12 @@ from typing_extensions import Self
from bumble import crypto
from bumble.colors import color
from bumble.core import (
PhysicalTransport,
AdvertisingData,
DeviceClass,
InvalidArgumentError,
InvalidPacketError,
ProtocolError,
PhysicalTransport,
ProtocolError,
bit_flags_to_strings,
name_or_number,
padded_bytes,
@@ -225,6 +224,7 @@ HCI_CONNECTIONLESS_PERIPHERAL_BROADCAST_CHANNEL_MAP_CHANGE_EVENT = 0X55
HCI_INQUIRY_RESPONSE_NOTIFICATION_EVENT = 0X56
HCI_AUTHENTICATED_PAYLOAD_TIMEOUT_EXPIRED_EVENT = 0X57
HCI_SAM_STATUS_CHANGE_EVENT = 0X58
HCI_ENCRYPTION_CHANGE_V2_EVENT = 0x59
HCI_VENDOR_EVENT = 0xFF
@@ -3364,6 +3364,20 @@ class HCI_Set_Event_Mask_Page_2_Command(HCI_Command):
See Bluetooth spec @ 7.3.69 Set Event Mask Page 2 Command
'''
@staticmethod
def mask(event_codes: Iterable[int]) -> bytes:
'''
Compute the event mask value for a list of events.
'''
# NOTE: this implementation takes advantage of the fact that as of version 6.0
# of the core specification, the bit number for each event code is equal to 64
# less than the event code.
# If future versions of the specification deviate from that, a different
# implementation would be needed.
return sum((1 << event_code - 64) for event_code in event_codes).to_bytes(
8, 'little'
)
# -----------------------------------------------------------------------------
@HCI_Command.command(
@@ -6977,6 +6991,30 @@ class HCI_Encryption_Change_Event(HCI_Event):
)
# -----------------------------------------------------------------------------
@HCI_Event.event(
[
('status', STATUS_SPEC),
('connection_handle', 2),
(
'encryption_enabled',
{
'size': 1,
# pylint: disable-next=unnecessary-lambda
'mapper': lambda x: HCI_Encryption_Change_Event.encryption_enabled_name(
x
),
},
),
('encryption_key_size', 1),
]
)
class HCI_Encryption_Change_V2_Event(HCI_Event):
'''
See Bluetooth spec @ 7.7.8 Encryption Change Event
'''
# -----------------------------------------------------------------------------
@HCI_Event.event(
[('status', STATUS_SPEC), ('connection_handle', 2), ('lmp_features', 8)]

23
bumble/host.py
View File

@@ -435,6 +435,14 @@ class Host(utils.EventEmitter):
)
)
)
if self.supports_command(hci.HCI_SET_EVENT_MASK_PAGE_2_COMMAND):
await self.send_command(
hci.HCI_Set_Event_Mask_Page_2_Command(
event_mask_page_2=hci.HCI_Set_Event_Mask_Page_2_Command.mask(
[hci.HCI_ENCRYPTION_CHANGE_V2_EVENT]
)
)
)
if (
self.local_version is not None
@@ -1384,6 +1392,21 @@ class Host(utils.EventEmitter):
'connection_encryption_change',
event.connection_handle,
event.encryption_enabled,
0,
)
else:
self.emit(
'connection_encryption_failure', event.connection_handle, event.status
)
def on_hci_encryption_change_v2_event(self, event):
# Notify the client
if event.status == hci.HCI_SUCCESS:
self.emit(
'connection_encryption_change',
event.connection_handle,
event.encryption_enabled,
event.encryption_key_size,
)
else:
self.emit(

6
examples/run_a2dp_sink.py
View File

@@ -33,12 +33,6 @@ from bumble.avdtp import (
from bumble.a2dp import (
make_audio_sink_service_sdp_records,
A2DP_SBC_CODEC_TYPE,
SBC_MONO_CHANNEL_MODE,
SBC_DUAL_CHANNEL_MODE,
SBC_SNR_ALLOCATION_METHOD,
SBC_LOUDNESS_ALLOCATION_METHOD,
SBC_STEREO_CHANNEL_MODE,
SBC_JOINT_STEREO_CHANNEL_MODE,
SbcMediaCodecInformation,
)