A2DP example: Codec selection based on file type

Currently support SBC and AAC

.github/workflows/code-check.yml

@@ -33,7 +33,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        python -m pip install ".[build,test,development]"
+        python -m pip install ".[build,examples,test,development]"
     - name: Check
       run: |
         invoke project.pre-commit

apps/bench.py

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

apps/pair.py

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

bumble/core.py

@@ -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]]
 

bumble/device.py

@@ -1586,9 +1586,9 @@ 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]
     gatt_client: gatt_client.Client
     pairing_peer_io_capability: Optional[int]
     pairing_peer_authentication_requirements: Optional[int]
@@ -1688,9 +1688,9 @@ 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
         self.att_mtu = ATT_DEFAULT_MTU
         self.data_length = DEVICE_DEFAULT_DATA_LENGTH
         self.gatt_client = gatt_client.Client(self)  # Per-connection client
@@ -1809,7 +1809,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 +3756,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 +3776,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 +3833,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 +3897,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 +3975,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 +4201,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 +4251,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 +4359,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 +4447,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 +4494,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 +4546,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)
@@ -5063,19 +5073,19 @@ class Device(utils.CompositeEventEmitter):
                 hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_192_TYPE,
                 hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_256_TYPE,
             )
-            pairing_keys = PairingKeys()
-            pairing_keys.link_key = PairingKeys.Key(
-                value=link_key, authenticated=authenticated
+            pairing_keys = PairingKeys(
+                link_key=PairingKeys.Key(value=link_key, authenticated=authenticated),
+                link_key_type=key_type,
             )
 
             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 +5348,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 +5465,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 +5727,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 +5757,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 +5793,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 +5834,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 +5972,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

bumble/hci.py

@@ -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)]

bumble/host.py

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

bumble/keys.py

@@ -22,14 +22,15 @@
 # -----------------------------------------------------------------------------
 from __future__ import annotations
 import asyncio
+import dataclasses
 import logging
 import os
 import json
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Type
+from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Type, Any
 from typing_extensions import Self
 
 from bumble.colors import color
-from bumble.hci import Address
+from bumble import hci
 
 if TYPE_CHECKING:
     from bumble.device import Device
@@ -42,16 +43,17 @@ logger = logging.getLogger(__name__)
 
 
 # -----------------------------------------------------------------------------
+@dataclasses.dataclass
 class PairingKeys:
+    @dataclasses.dataclass
     class Key:
-        def __init__(self, value, authenticated=False, ediv=None, rand=None):
-            self.value = value
-            self.authenticated = authenticated
-            self.ediv = ediv
-            self.rand = rand
+        value: bytes
+        authenticated: bool = False
+        ediv: Optional[int] = None
+        rand: Optional[bytes] = None
 
         @classmethod
-        def from_dict(cls, key_dict):
+        def from_dict(cls, key_dict: dict[str, Any]) -> PairingKeys.Key:
             value = bytes.fromhex(key_dict['value'])
             authenticated = key_dict.get('authenticated', False)
             ediv = key_dict.get('ediv')
@@ -61,7 +63,7 @@ class PairingKeys:
 
             return cls(value, authenticated, ediv, rand)
 
-        def to_dict(self):
+        def to_dict(self) -> dict[str, Any]:
             key_dict = {'value': self.value.hex(), 'authenticated': self.authenticated}
             if self.ediv is not None:
                 key_dict['ediv'] = self.ediv
@@ -70,39 +72,42 @@ class PairingKeys:
 
             return key_dict
 
-    def __init__(self):
-        self.address_type = None
-        self.ltk = None
-        self.ltk_central = None
-        self.ltk_peripheral = None
-        self.irk = None
-        self.csrk = None
-        self.link_key = None  # Classic
+    address_type: Optional[hci.AddressType] = None
+    ltk: Optional[Key] = None
+    ltk_central: Optional[Key] = None
+    ltk_peripheral: Optional[Key] = None
+    irk: Optional[Key] = None
+    csrk: Optional[Key] = None
+    link_key: Optional[Key] = None  # Classic
+    link_key_type: Optional[int] = None  # Classic
 
-    @staticmethod
-    def key_from_dict(keys_dict, key_name):
+    @classmethod
+    def key_from_dict(cls, keys_dict: dict[str, Any], key_name: str) -> Optional[Key]:
         key_dict = keys_dict.get(key_name)
         if key_dict is None:
             return None
 
         return PairingKeys.Key.from_dict(key_dict)
 
-    @staticmethod
-    def from_dict(keys_dict):
-        keys = PairingKeys()
+    @classmethod
+    def from_dict(cls, keys_dict: dict[str, Any]) -> PairingKeys:
+        return PairingKeys(
+            address_type=(
+                hci.AddressType(t)
+                if (t := keys_dict.get('address_type')) is not None
+                else None
+            ),
+            ltk=PairingKeys.key_from_dict(keys_dict, 'ltk'),
+            ltk_central=PairingKeys.key_from_dict(keys_dict, 'ltk_central'),
+            ltk_peripheral=PairingKeys.key_from_dict(keys_dict, 'ltk_peripheral'),
+            irk=PairingKeys.key_from_dict(keys_dict, 'irk'),
+            csrk=PairingKeys.key_from_dict(keys_dict, 'csrk'),
+            link_key=PairingKeys.key_from_dict(keys_dict, 'link_key'),
+            link_key_type=keys_dict.get('link_key_type'),
+        )
 
-        keys.address_type = keys_dict.get('address_type')
-        keys.ltk = PairingKeys.key_from_dict(keys_dict, 'ltk')
-        keys.ltk_central = PairingKeys.key_from_dict(keys_dict, 'ltk_central')
-        keys.ltk_peripheral = PairingKeys.key_from_dict(keys_dict, 'ltk_peripheral')
-        keys.irk = PairingKeys.key_from_dict(keys_dict, 'irk')
-        keys.csrk = PairingKeys.key_from_dict(keys_dict, 'csrk')
-        keys.link_key = PairingKeys.key_from_dict(keys_dict, 'link_key')
-
-        return keys
-
-    def to_dict(self):
-        keys = {}
+    def to_dict(self) -> dict[str, Any]:
+        keys: dict[str, Any] = {}
 
         if self.address_type is not None:
             keys['address_type'] = self.address_type
@@ -125,9 +130,12 @@ class PairingKeys:
         if self.link_key is not None:
             keys['link_key'] = self.link_key.to_dict()
 
+        if self.link_key_type is not None:
+            keys['link_key_type'] = self.link_key_type
+
         return keys
 
-    def print(self, prefix=''):
+    def print(self, prefix: str = '') -> None:
         keys_dict = self.to_dict()
         for container_property, value in keys_dict.items():
             if isinstance(value, dict):
@@ -156,20 +164,28 @@ class KeyStore:
         all_keys = await self.get_all()
         await asyncio.gather(*(self.delete(name) for (name, _) in all_keys))
 
-    async def get_resolving_keys(self):
+    async def get_resolving_keys(self) -> list[tuple[bytes, hci.Address]]:
         all_keys = await self.get_all()
         resolving_keys = []
         for name, keys in all_keys:
             if keys.irk is not None:
-                if keys.address_type is None:
-                    address_type = Address.RANDOM_DEVICE_ADDRESS
-                else:
-                    address_type = keys.address_type
-                resolving_keys.append((keys.irk.value, Address(name, address_type)))
+                resolving_keys.append(
+                    (
+                        keys.irk.value,
+                        hci.Address(
+                            name,
+                            (
+                                keys.address_type
+                                if keys.address_type is not None
+                                else hci.Address.RANDOM_DEVICE_ADDRESS
+                            ),
+                        ),
+                    )
+                )
 
         return resolving_keys
 
-    async def print(self, prefix=''):
+    async def print(self, prefix: str = '') -> None:
         entries = await self.get_all()
         separator = ''
         for name, keys in entries:
@@ -177,8 +193,8 @@ class KeyStore:
             keys.print(prefix=prefix + '  ')
             separator = '\n'
 
-    @staticmethod
-    def create_for_device(device: Device) -> KeyStore:
+    @classmethod
+    def create_for_device(cls, device: Device) -> KeyStore:
         if device.config.keystore is None:
             return MemoryKeyStore()
 
@@ -266,9 +282,9 @@ class JsonKeyStore(KeyStore):
                     filename = params[0]
 
         # Use a namespace based on the device address
-        if device.public_address not in (Address.ANY, Address.ANY_RANDOM):
+        if device.public_address not in (hci.Address.ANY, hci.Address.ANY_RANDOM):
             namespace = str(device.public_address)
-        elif device.random_address != Address.ANY_RANDOM:
+        elif device.random_address != hci.Address.ANY_RANDOM:
             namespace = str(device.random_address)
         else:
             namespace = JsonKeyStore.DEFAULT_NAMESPACE

bumble/pandora/security.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 import asyncio
 import contextlib
+from collections.abc import Awaitable
 import grpc
 import logging
 
@@ -24,6 +25,7 @@ from bumble import hci
 from bumble.core import (
     PhysicalTransport,
     ProtocolError,
+    InvalidArgumentError,
 )
 import bumble.utils
 from bumble.device import Connection as BumbleConnection, Device
@@ -188,35 +190,6 @@ class PairingDelegate(BasePairingDelegate):
         self.service.event_queue.put_nowait(event)
 
 
-BR_LEVEL_REACHED: Dict[SecurityLevel, Callable[[BumbleConnection], bool]] = {
-    LEVEL0: lambda connection: True,
-    LEVEL1: lambda connection: connection.encryption == 0 or connection.authenticated,
-    LEVEL2: lambda connection: connection.encryption != 0 and connection.authenticated,
-    LEVEL3: lambda connection: connection.encryption != 0
-    and connection.authenticated
-    and connection.link_key_type
-    in (
-        hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_192_TYPE,
-        hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_256_TYPE,
-    ),
-    LEVEL4: lambda connection: connection.encryption
-    == hci.HCI_Encryption_Change_Event.AES_CCM
-    and connection.authenticated
-    and connection.link_key_type
-    == hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_256_TYPE,
-}
-
-LE_LEVEL_REACHED: Dict[LESecurityLevel, Callable[[BumbleConnection], bool]] = {
-    LE_LEVEL1: lambda connection: True,
-    LE_LEVEL2: lambda connection: connection.encryption != 0,
-    LE_LEVEL3: lambda connection: connection.encryption != 0
-    and connection.authenticated,
-    LE_LEVEL4: lambda connection: connection.encryption != 0
-    and connection.authenticated
-    and connection.sc,
-}
-
-
 class SecurityService(SecurityServicer):
     def __init__(self, device: Device, config: Config) -> None:
         self.log = utils.BumbleServerLoggerAdapter(
@@ -248,6 +221,59 @@ class SecurityService(SecurityServicer):
 
         self.device.pairing_config_factory = pairing_config_factory
 
+    async def _classic_level_reached(
+        self, level: SecurityLevel, connection: BumbleConnection
+    ) -> bool:
+        if level == LEVEL0:
+            return True
+        if level == LEVEL1:
+            return connection.encryption == 0 or connection.authenticated
+        if level == LEVEL2:
+            return connection.encryption != 0 and connection.authenticated
+
+        link_key_type: Optional[int] = None
+        if (keystore := connection.device.keystore) and (
+            keys := await keystore.get(str(connection.peer_address))
+        ):
+            link_key_type = keys.link_key_type
+        self.log.debug("link_key_type: %d", link_key_type)
+
+        if level == LEVEL3:
+            return (
+                connection.encryption != 0
+                and connection.authenticated
+                and link_key_type
+                in (
+                    hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_192_TYPE,
+                    hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_256_TYPE,
+                )
+            )
+        if level == LEVEL4:
+            return (
+                connection.encryption == hci.HCI_Encryption_Change_Event.AES_CCM
+                and connection.authenticated
+                and link_key_type
+                == hci.HCI_AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_256_TYPE
+            )
+        raise InvalidArgumentError(f"Unexpected level {level}")
+
+    def _le_level_reached(
+        self, level: LESecurityLevel, connection: BumbleConnection
+    ) -> bool:
+        if level == LE_LEVEL1:
+            return True
+        if level == LE_LEVEL2:
+            return connection.encryption != 0
+        if level == LE_LEVEL3:
+            return connection.encryption != 0 and connection.authenticated
+        if level == LE_LEVEL4:
+            return (
+                connection.encryption != 0
+                and connection.authenticated
+                and connection.sc
+            )
+        raise InvalidArgumentError(f"Unexpected level {level}")
+
     @utils.rpc
     async def OnPairing(
         self, request: AsyncIterator[PairingEventAnswer], context: grpc.ServicerContext
@@ -290,7 +316,7 @@ class SecurityService(SecurityServicer):
         ] == oneof
 
         # security level already reached
-        if self.reached_security_level(connection, level):
+        if await self.reached_security_level(connection, level):
             return SecureResponse(success=empty_pb2.Empty())
 
         # trigger pairing if needed
@@ -361,7 +387,7 @@ class SecurityService(SecurityServicer):
                 return SecureResponse(encryption_failure=empty_pb2.Empty())
 
         # security level has been reached ?
-        if self.reached_security_level(connection, level):
+        if await self.reached_security_level(connection, level):
             return SecureResponse(success=empty_pb2.Empty())
         return SecureResponse(not_reached=empty_pb2.Empty())
 
@@ -388,13 +414,10 @@ class SecurityService(SecurityServicer):
         pair_task: Optional[asyncio.Future[None]] = None
 
         async def authenticate() -> None:
-            assert connection
             if (encryption := connection.encryption) != 0:
                 self.log.debug('Disable encryption...')
-                try:
+                with contextlib.suppress(Exception):
                     await connection.encrypt(enable=False)
-                except:
-                    pass
                 self.log.debug('Disable encryption: done')
 
             self.log.debug('Authenticate...')
@@ -413,15 +436,13 @@ class SecurityService(SecurityServicer):
 
             return wrapper
 
-        def try_set_success(*_: Any) -> None:
-            assert connection
-            if self.reached_security_level(connection, level):
+        async def try_set_success(*_: Any) -> None:
+            if await self.reached_security_level(connection, level):
                 self.log.debug('Wait for security: done')
                 wait_for_security.set_result('success')
 
-        def on_encryption_change(*_: Any) -> None:
-            assert connection
-            if self.reached_security_level(connection, level):
+        async def on_encryption_change(*_: Any) -> None:
+            if await self.reached_security_level(connection, level):
                 self.log.debug('Wait for security: done')
                 wait_for_security.set_result('success')
             elif (
@@ -436,7 +457,7 @@ class SecurityService(SecurityServicer):
             if self.need_pairing(connection, level):
                 pair_task = asyncio.create_task(connection.pair())
 
-        listeners: Dict[str, Callable[..., None]] = {
+        listeners: Dict[str, Callable[..., Union[None, Awaitable[None]]]] = {
             'disconnection': set_failure('connection_died'),
             'pairing_failure': set_failure('pairing_failure'),
             'connection_authentication_failure': set_failure('authentication_failure'),
@@ -455,7 +476,7 @@ class SecurityService(SecurityServicer):
                 watcher.on(connection, event, listener)
 
             # security level already reached
-            if self.reached_security_level(connection, level):
+            if await self.reached_security_level(connection, level):
                 return WaitSecurityResponse(success=empty_pb2.Empty())
 
             self.log.debug('Wait for security...')
@@ -465,24 +486,20 @@ class SecurityService(SecurityServicer):
         # wait for `authenticate` to finish if any
         if authenticate_task is not None:
             self.log.debug('Wait for authentication...')
-            try:
+            with contextlib.suppress(Exception):
                 await authenticate_task  # type: ignore
-            except:
-                pass
             self.log.debug('Authenticated')
 
         # wait for `pair` to finish if any
         if pair_task is not None:
             self.log.debug('Wait for authentication...')
-            try:
+            with contextlib.suppress(Exception):
                 await pair_task  # type: ignore
-            except:
-                pass
             self.log.debug('paired')
 
         return WaitSecurityResponse(**kwargs)
 
-    def reached_security_level(
+    async def reached_security_level(
         self, connection: BumbleConnection, level: Union[SecurityLevel, LESecurityLevel]
     ) -> bool:
         self.log.debug(
@@ -492,15 +509,14 @@ class SecurityService(SecurityServicer):
                     'encryption': connection.encryption,
                     'authenticated': connection.authenticated,
                     'sc': connection.sc,
-                    'link_key_type': connection.link_key_type,
                 }
             )
         )
 
         if isinstance(level, LESecurityLevel):
-            return LE_LEVEL_REACHED[level](connection)
+            return self._le_level_reached(level, connection)
 
-        return BR_LEVEL_REACHED[level](connection)
+        return await self._classic_level_reached(level, connection)
 
     def need_pairing(self, connection: BumbleConnection, level: int) -> bool:
         if connection.transport == PhysicalTransport.LE:

bumble/smp.py

@@ -1380,8 +1380,10 @@ class Session:
                 ediv=self.ltk_ediv,
                 rand=self.ltk_rand,
             )
+            if not self.peer_ltk:
+                logger.error("peer_ltk is None")
             peer_ltk_key = PairingKeys.Key(
-                value=self.peer_ltk,
+                value=self.peer_ltk or b'',
                 authenticated=authenticated,
                 ediv=self.peer_ediv,
                 rand=self.peer_rand,

examples/run_a2dp_sink.py

@@ -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,
 )
 

examples/run_a2dp_source.py

@@ -16,28 +16,43 @@
 # Imports
 # -----------------------------------------------------------------------------
 import asyncio
-import sys
-import os
 import logging
+import os
+import sys
+from dataclasses import dataclass
 
-from bumble.colors import color
-from bumble.device import Device
-from bumble.transport import open_transport_or_link
-from bumble.core import PhysicalTransport
+import ffmpeg
+
+from bumble.a2dp import (
+    A2DP_MPEG_2_4_AAC_CODEC_TYPE,
+    A2DP_SBC_CODEC_TYPE,
+    AacMediaCodecInformation,
+    AacPacketSource,
+    SbcMediaCodecInformation,
+    SbcPacketSource,
+    make_audio_source_service_sdp_records,
+)
 from bumble.avdtp import (
-    find_avdtp_service_with_connection,
     AVDTP_AUDIO_MEDIA_TYPE,
+    Listener,
     MediaCodecCapabilities,
     MediaPacketPump,
     Protocol,
-    Listener,
-)
-from bumble.a2dp import (
-    make_audio_source_service_sdp_records,
-    A2DP_SBC_CODEC_TYPE,
-    SbcMediaCodecInformation,
-    SbcPacketSource,
+    find_avdtp_service_with_connection,
 )
+from bumble.colors import color
+from bumble.core import PhysicalTransport
+from bumble.device import Device
+from bumble.transport import open_transport_or_link
+
+from typing import Dict, Union
+
+
+@dataclass
+class CodecCapabilities:
+    name: str
+    sample_rate: str
+    number_of_channels: str
 
 
 # -----------------------------------------------------------------------------
@@ -51,67 +66,147 @@ def sdp_records():
 
 
 # -----------------------------------------------------------------------------
-def codec_capabilities():
-    # NOTE: this shouldn't be hardcoded, but should be inferred from the input file
-    # instead
-    return MediaCodecCapabilities(
-        media_type=AVDTP_AUDIO_MEDIA_TYPE,
-        media_codec_type=A2DP_SBC_CODEC_TYPE,
-        media_codec_information=SbcMediaCodecInformation(
-            sampling_frequency=SbcMediaCodecInformation.SamplingFrequency.SF_44100,
-            channel_mode=SbcMediaCodecInformation.ChannelMode.JOINT_STEREO,
-            block_length=SbcMediaCodecInformation.BlockLength.BL_16,
-            subbands=SbcMediaCodecInformation.Subbands.S_8,
-            allocation_method=SbcMediaCodecInformation.AllocationMethod.LOUDNESS,
-            minimum_bitpool_value=2,
-            maximum_bitpool_value=53,
-        ),
-    )
-
-
-# -----------------------------------------------------------------------------
-def on_avdtp_connection(read_function, protocol):
+def on_avdtp_connection(
+    read_function, protocol, codec_capabilities: MediaCodecCapabilities
+):
     packet_source = SbcPacketSource(read_function, protocol.l2cap_channel.peer_mtu)
     packet_pump = MediaPacketPump(packet_source.packets)
-    protocol.add_source(codec_capabilities(), packet_pump)
+    protocol.add_source(codec_capabilities, packet_pump)
 
 
 # -----------------------------------------------------------------------------
-async def stream_packets(read_function, protocol):
+async def stream_packets(
+    read_function, protocol, codec_capabilities: MediaCodecCapabilities
+):
     # Discover all endpoints on the remote device
     endpoints = await protocol.discover_remote_endpoints()
     for endpoint in endpoints:
         print('@@@', endpoint)
 
     # Select a sink
+    assert codec_capabilities.media_codec_type in [
+        A2DP_SBC_CODEC_TYPE,
+        A2DP_MPEG_2_4_AAC_CODEC_TYPE,
+    ]
     sink = protocol.find_remote_sink_by_codec(
-        AVDTP_AUDIO_MEDIA_TYPE, A2DP_SBC_CODEC_TYPE
+        AVDTP_AUDIO_MEDIA_TYPE, codec_capabilities.media_codec_type
     )
     if sink is None:
-        print(color('!!! no SBC sink found', 'red'))
+        print(color('!!! no Sink found', 'red'))
         return
     print(f'### Selected sink: {sink.seid}')
 
     # Stream the packets
-    packet_source = SbcPacketSource(read_function, protocol.l2cap_channel.peer_mtu)
-    packet_pump = MediaPacketPump(packet_source.packets)
-    source = protocol.add_source(codec_capabilities(), packet_pump)
+    packet_sources = {
+        A2DP_SBC_CODEC_TYPE: SbcPacketSource(
+            read_function, protocol.l2cap_channel.peer_mtu
+        ),
+        A2DP_MPEG_2_4_AAC_CODEC_TYPE: AacPacketSource(
+            read_function, protocol.l2cap_channel.peer_mtu
+        ),
+    }
+    packet_source = packet_sources[codec_capabilities.media_codec_type]
+    packet_pump = MediaPacketPump(packet_source.packets)  # type: ignore
+    source = protocol.add_source(codec_capabilities, packet_pump)
     stream = await protocol.create_stream(source, sink)
     await stream.start()
-    await asyncio.sleep(5)
-    await stream.stop()
-    await asyncio.sleep(5)
-    await stream.start()
-    await asyncio.sleep(5)
+    await asyncio.sleep(60)
     await stream.stop()
     await stream.close()
 
 
+# -----------------------------------------------------------------------------
+def fetch_codec_informations(filepath) -> MediaCodecCapabilities:
+    probe = ffmpeg.probe(filepath)
+    assert 'streams' in probe
+    streams = probe['streams']
+
+    if not streams or len(streams) > 1:
+        print(streams)
+        print(color('!!! file not supported', 'red'))
+        exit()
+    audio_stream = streams[0]
+
+    media_codec_type = None
+    media_codec_information: Union[
+        SbcMediaCodecInformation, AacMediaCodecInformation, None
+    ] = None
+
+    assert 'codec_name' in audio_stream
+    codec_name: str = audio_stream['codec_name']
+    if codec_name == "sbc":
+        media_codec_type = A2DP_SBC_CODEC_TYPE
+        sbc_sampling_frequency: Dict[
+            str, SbcMediaCodecInformation.SamplingFrequency
+        ] = {
+            '16000': SbcMediaCodecInformation.SamplingFrequency.SF_16000,
+            '32000': SbcMediaCodecInformation.SamplingFrequency.SF_32000,
+            '44100': SbcMediaCodecInformation.SamplingFrequency.SF_44100,
+            '48000': SbcMediaCodecInformation.SamplingFrequency.SF_48000,
+        }
+        sbc_channel_mode: Dict[int, SbcMediaCodecInformation.ChannelMode] = {
+            1: SbcMediaCodecInformation.ChannelMode.MONO,
+            2: SbcMediaCodecInformation.ChannelMode.JOINT_STEREO,
+        }
+
+        assert 'sample_rate' in audio_stream
+        assert 'channels' in audio_stream
+        media_codec_information = SbcMediaCodecInformation(
+            sampling_frequency=sbc_sampling_frequency[audio_stream['sample_rate']],
+            channel_mode=sbc_channel_mode[audio_stream['channels']],
+            block_length=SbcMediaCodecInformation.BlockLength.BL_16,
+            subbands=SbcMediaCodecInformation.Subbands.S_8,
+            allocation_method=SbcMediaCodecInformation.AllocationMethod.LOUDNESS,
+            minimum_bitpool_value=2,
+            maximum_bitpool_value=53,
+        )
+    elif codec_name == "aac":
+        media_codec_type = A2DP_MPEG_2_4_AAC_CODEC_TYPE
+        object_type: Dict[str, AacMediaCodecInformation.ObjectType] = {
+            'LC': AacMediaCodecInformation.ObjectType.MPEG_2_AAC_LC,
+            'LTP': AacMediaCodecInformation.ObjectType.MPEG_4_AAC_LTP,
+            'SSR': AacMediaCodecInformation.ObjectType.MPEG_4_AAC_SCALABLE,
+        }
+        aac_sampling_frequency: Dict[
+            str, AacMediaCodecInformation.SamplingFrequency
+        ] = {
+            '44100': AacMediaCodecInformation.SamplingFrequency.SF_44100,
+            '48000': AacMediaCodecInformation.SamplingFrequency.SF_48000,
+        }
+        aac_channel_mode: Dict[int, AacMediaCodecInformation.Channels] = {
+            1: AacMediaCodecInformation.Channels.MONO,
+            2: AacMediaCodecInformation.Channels.STEREO,
+        }
+
+        assert 'profile' in audio_stream
+        assert 'sample_rate' in audio_stream
+        assert 'channels' in audio_stream
+        media_codec_information = AacMediaCodecInformation(
+            object_type=object_type[audio_stream['profile']],
+            sampling_frequency=aac_sampling_frequency[audio_stream['sample_rate']],
+            channels=aac_channel_mode[audio_stream['channels']],
+            vbr=1,
+            bitrate=128000,
+        )
+    else:
+        print(color('!!! codec not supported, only aac & sbc are supported', 'red'))
+        exit()
+
+    assert media_codec_type is not None
+    assert media_codec_information is not None
+
+    return MediaCodecCapabilities(
+        media_type=AVDTP_AUDIO_MEDIA_TYPE,
+        media_codec_type=media_codec_type,
+        media_codec_information=media_codec_information,
+    )
+
+
 # -----------------------------------------------------------------------------
 async def main() -> None:
     if len(sys.argv) < 4:
         print(
-            'Usage: run_a2dp_source.py <device-config> <transport-spec> <sbc-file> '
+            'Usage: run_a2dp_source.py <device-config> <transport-spec> <audio-file> '
             '[<bluetooth-address>]'
         )
         print(
@@ -135,11 +230,13 @@ async def main() -> None:
         # Start
         await device.power_on()
 
-        with open(sys.argv[3], 'rb') as sbc_file:
+        with open(sys.argv[3], 'rb') as audio_file:
             # NOTE: this should be using asyncio file reading, but blocking reads are
             # good enough for testing
             async def read(byte_count):
-                return sbc_file.read(byte_count)
+                return audio_file.read(byte_count)
+
+            codec_capabilities = fetch_codec_informations(sys.argv[3])
 
             if len(sys.argv) > 4:
                 # Connect to a peer
@@ -170,12 +267,15 @@ async def main() -> None:
                 protocol = await Protocol.connect(connection, avdtp_version)
 
                 # Start streaming
-                await stream_packets(read, protocol)
+                await stream_packets(read, protocol, codec_capabilities)
             else:
                 # Create a listener to wait for AVDTP connections
                 listener = Listener.for_device(device=device, version=(1, 2))
                 listener.on(
-                    'connection', lambda protocol: on_avdtp_connection(read, protocol)
+                    'connection',
+                    lambda protocol: on_avdtp_connection(
+                        read, protocol, codec_capabilities
+                    ),
                 )
 
                 # Become connectable and wait for a connection

pyproject.toml

@@ -13,11 +13,11 @@ dependencies = [
     "aiohttp ~= 3.8; platform_system!='Emscripten'",
     "appdirs >= 1.4; platform_system!='Emscripten'",
     "click >= 8.1.3; platform_system!='Emscripten'",
-    "cryptography >= 39; platform_system!='Emscripten'",
+    "cryptography >= 44.0.3; platform_system!='Emscripten'",
     # Pyodide bundles a version of cryptography that is built for wasm, which may not match the
     # versions available on PyPI. Relax the version requirement since it's better than being
     # completely unable to import the package in case of version mismatch.
-    "cryptography >= 39.0; platform_system=='Emscripten'",
+    "cryptography >= 44.0.3; platform_system=='Emscripten'",
     "grpcio >= 1.62.1; platform_system!='Emscripten'",
     "humanize >= 4.6.0; platform_system!='Emscripten'",
     "libusb1 >= 2.0.1; platform_system!='Emscripten'",
@@ -55,6 +55,9 @@ development = [
     "types-invoke >= 1.7.3",
     "types-protobuf >= 4.21.0",
 ]
+examples = [
+    "ffmpeg-python == 0.2.0",
+]
 avatar = [
     "pandora-avatar == 0.0.10",
     "rootcanal == 1.11.1 ; python_version>='3.10'",
@@ -184,6 +187,10 @@ ignore_missing_imports = true
 module = "construct.*"
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = "ffmpeg.*"
+ignore_missing_imports = true
+
 [[tool.mypy.overrides]]
 module = "grpc.*"
 ignore_missing_imports = true