wip

Support CTKD over BR/EDR

bumble/device.py

@@ -18,7 +18,8 @@
 import json
 import asyncio
 import logging
-from  contextlib import asynccontextmanager, AsyncExitStack
+import secrets
+from contextlib import asynccontextmanager, AsyncExitStack
 
 from .hci import *
 from .host import Host
@@ -32,6 +33,8 @@ from . import smp
 from . import sdp
 from . import l2cap
 from . import keys
+from . import crypto
+
 
 # -----------------------------------------------------------------------------
 # Logging
@@ -51,6 +54,7 @@ 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_LE_RPA_TIMEOUT       = 15 * 60  # 15 minutes (in seconds)
 DEVICE_MIN_SCAN_INTERVAL            = 25
 DEVICE_MAX_SCAN_INTERVAL            = 10240
 DEVICE_MIN_SCAN_WINDOW              = 25
@@ -169,7 +173,6 @@ class Peer:
     async def __aexit__(self, exc_type, exc_value, traceback):
         pass
 
-
     def __str__(self):
         return f'{self.connection.peer_address} as {self.connection.role_name}'
 
@@ -202,11 +205,22 @@ class Connection(CompositeEventEmitter):
         def on_connection_encryption_key_refresh(self):
             pass
 
-    def __init__(self, device, handle, transport, peer_address, peer_resolvable_address, role, parameters):
+    def __init__(
+        self,
+        device,
+        handle,
+        transport,
+        local_address,
+        peer_address,
+        peer_resolvable_address,
+        role,
+        parameters
+    ):
         super().__init__()
         self.device                  = device
         self.handle                  = handle
         self.transport               = transport
+        self.local_address           = local_address
         self.peer_address            = peer_address
         self.peer_resolvable_address = peer_resolvable_address
         self.peer_name               = None  # Classic only
@@ -297,7 +311,12 @@ class Connection(CompositeEventEmitter):
                     raise
 
     def __str__(self):
-        return f'Connection(handle=0x{self.handle:04X}, role={self.role_name}, address={self.peer_address})'
+        return (
+            f'Connection(handle=0x{self.handle:04X}, '
+            f'role={self.role_name}, '
+            f'local_address={self.local_address}, '
+            f'peer_address={self.peer_address})'
+        )
 
 
 # -----------------------------------------------------------------------------
@@ -311,8 +330,10 @@ class DeviceConfiguration:
         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.le_privacy_enabled       = False
+        self.le_rpa_timeout           = DEVICE_DEFAULT_LE_RPA_TIMEOUT
+        self.classic_enabled          = False
         self.classic_sc_enabled       = True
         self.classic_ssp_enabled      = True
         self.connectable              = True
@@ -320,19 +341,22 @@ class DeviceConfiguration:
         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.irk      = bytes([0xFF] * 16)  # This really must be changed for any level of security
         self.keystore = None
 
     def load_from_dict(self, config):
         # Load simple properties
-        self.name = config.get('name', self.name)
-        self.address = Address(config.get('address', self.address))
-        self.class_of_device = config.get('class_of_device', self.class_of_device)
+        self.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.le_privacy_enabled       = config.get('le_privacy_enabled', self.le_privacy_enabled)
+        self.le_rpa_timeout           = config.get('le_rpa_timeout', self.le_rpa_timeout)
+        self.classic_enabled          = config.get('classic_enabled', self.classic_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.connectable              = config.get('connectable', self.connectable)
@@ -352,6 +376,10 @@ class DeviceConfiguration:
         advertising_data = config.get('advertising_data')
         if advertising_data:
             self.advertising_data = bytes.fromhex(advertising_data)
+        else:
+            self.advertising_data = bytes(
+                AdvertisingData([(AdvertisingData.COMPLETE_LOCAL_NAME, bytes(self.name, 'utf-8'))])
+            )
 
     def load_from_file(self, filename):
         with open(filename, 'r') as file:
@@ -458,9 +486,9 @@ class Device(CompositeEventEmitter):
         self.connecting               = False
         self.disconnecting            = False
         self.connections              = {}  # Connections, by connection handle
-        self.classic_enabled          = False
         self.inquiry_response         = None
         self.address_resolver         = None
+        self.le_rpa_task              = None
 
         # Use the initial config or a default
         self.public_address = Address('00:00:00:00:00:00')
@@ -468,6 +496,7 @@ class Device(CompositeEventEmitter):
             config = DeviceConfiguration()
         self.name                     = config.name
         self.random_address           = config.address
+        self.identity_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
@@ -477,6 +506,9 @@ class Device(CompositeEventEmitter):
         self.irk                      = config.irk
         self.le_enabled               = config.le_enabled
         self.le_simultaneous_enabled  = config.le_simultaneous_enabled
+        self.le_privacy_enabled       = config.le_privacy_enabled
+        self.le_rpa_timeout           = config.le_rpa_timeout
+        self.classic_enabled          = config.classic_enabled
         self.classic_ssp_enabled      = config.classic_ssp_enabled
         self.classic_sc_enabled       = config.classic_sc_enabled
         self.discoverable             = config.discoverable
@@ -490,13 +522,16 @@ class Device(CompositeEventEmitter):
         if address:
             if type(address) is str:
                 address = Address(address)
-            self.random_address = address
+            self.random_address   = address
+            self.identity_address = address
 
         # Setup SMP
         # TODO: allow using a public address
-        self.smp_manager = smp.Manager(self, self.random_address)
+        self.smp_manager = smp.Manager(self, self.random_address, self.identity_address)
         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)
@@ -589,6 +624,14 @@ class Device(CompositeEventEmitter):
             ))
 
         if self.le_enabled:
+            # If LE Privacy is enabled, generate an RPA
+            if self.le_privacy_enabled:
+                self.random_address = self.generate_le_rpa()
+                logger.info(f'Initial RPA: {self.random_address}')
+                if self.le_rpa_timeout > 0:
+                    # Start a task to periodically generate a new RPA
+                    self.le_rpa_task = asyncio.create_task(self.run_le_rpa_generation())
+
             # Set the controller address
             await self.send_command(HCI_LE_Set_Random_Address_Command(
                 random_address = self.random_address
@@ -635,13 +678,48 @@ class Device(CompositeEventEmitter):
             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
 
+    async def run_le_rpa_generation(self):
+        while self.le_rpa_timeout != 0:
+            await asyncio.sleep(self.le_rpa_timeout)
+
+            # Check if this is a good time to rotate the address
+            if self.advertising or self.scanning or self.connecting:
+                logger.debug('skipping RPA rotation')
+                continue
+
+            random_address = self.generate_le_rpa()
+            response = await self.send_command(HCI_LE_Set_Random_Address_Command(
+                random_address = self.random_address
+            ))
+            if response.return_parameters == HCI_SUCCESS:
+                logger.info(f'New RPA: {random_address}')
+                self.random_address = random_address
+            else:
+                logger.warning(f'failed to set RPA: {response.return_parameters}')
+
+    def generate_le_rpa(self):
+        # See 1.3.2.2 Private device address generation
+
+        # Generate `prand`
+        while True:
+            # Generate a 22-bit random number for the random part of `prand`
+            prand_random = secrets.randbelow(0x400000)
+
+            # As least on bit shall be 0 and one bit shall be 1
+            if prand_random != 0 and prand_random != 0x3FFFFF:
+                break
+
+        prand = prand_random | 0x400000  # The two MSBs are |1|0|
+
+        # Generate `hash`
+        hash = crypto.ah(self.irk, struct.pack('<I', prand)[:3])
+
+        # Generate the address from `prand` and `hash`
+        return Address(hash + struct.pack('<I', prand)[:3], Address.RANDOM_IDENTITY_ADDRESS)
+
     async def start_advertising(self, auto_restart=False):
         self.auto_restart_advertising = auto_restart
 
@@ -673,18 +751,24 @@ class Device(CompositeEventEmitter):
         ))
 
         # Enable advertising
-        await self.send_command(HCI_LE_Set_Advertising_Enable_Command(
+        response = await self.send_command(HCI_LE_Set_Advertising_Enable_Command(
             advertising_enable = 1
         ))
+        if response.return_parameters != HCI_SUCCESS:
+            logger.warning(f'HCI_LE_Set_Advertising_Enable_Command failed ({response.return_parameters})')
+            raise HCI_Error(response.return_parameters)
 
         self.advertising = True
 
     async def stop_advertising(self):
         # Disable advertising
         if self.advertising:
-            await self.send_command(HCI_LE_Set_Advertising_Enable_Command(
+            response = await self.send_command(HCI_LE_Set_Advertising_Enable_Command(
                 advertising_enable = 0
             ))
+            if response.return_parameters != HCI_SUCCESS:
+                logger.warning(f'HCI_LE_Set_Advertising_Enable_Command failed ({response.return_parameters})')
+                raise HCI_Error(response.return_parameters)
 
             self.advertising = False
 
@@ -719,17 +803,23 @@ class Device(CompositeEventEmitter):
         ))
 
         # Enable scanning
-        await self.send_command(HCI_LE_Set_Scan_Enable_Command(
+        response = await self.send_command(HCI_LE_Set_Scan_Enable_Command(
             le_scan_enable    = 1,
             filter_duplicates = 1 if filter_duplicates else 0
         ))
+        if response.return_parameters != HCI_SUCCESS:
+            raise HCI_Error(response.return_parameters)
+
         self.scanning = True
 
     async def stop_scanning(self):
-        await self.send_command(HCI_LE_Set_Scan_Enable_Command(
+        response = await self.send_command(HCI_LE_Set_Scan_Enable_Command(
             le_scan_enable    = 0,
             filter_duplicates = 0
         ))
+        if response.return_parameters != HCI_SUCCESS:
+            raise HCI_Error(response.return_parameters)
+
         self.scanning = False
 
     @property
@@ -1240,6 +1330,7 @@ class Device(CompositeEventEmitter):
             self,
             connection_handle,
             transport,
+            self.public_address if transport == BT_BR_EDR_TRANSPORT else self.random_address,
             peer_address,
             peer_resolvable_address,
             role,

bumble/hci.py

@@ -1375,9 +1375,11 @@ class HCI_Error(ProtocolError):
 
 class HCI_StatusError(ProtocolError):
     def __init__(self, response):
-        super().__init__(response.status,
-                error_namespace=HCI_Command.command_name(response.command_opcode),
-                error_name=HCI_Constant.status_name(response.status))
+        super().__init__(
+            response.status,
+            error_namespace=HCI_Command.command_name(response.command_opcode),
+            error_name=HCI_Constant.status_name(response.status)
+        )
 
 
 # -----------------------------------------------------------------------------

bumble/host.py

@@ -44,12 +44,13 @@ HOST_HC_TOTAL_NUM_ACL_DATA_PACKETS        = 1
 
 # -----------------------------------------------------------------------------
 class Connection:
-    def __init__(self, host, handle, role, peer_address):
+    def __init__(self, host, handle, role, peer_address, transport):
         self.host                  = host
         self.handle                = handle
         self.role                  = role
         self.peer_address          = peer_address
         self.assembler             = HCI_AclDataPacketAssembler(self.on_acl_pdu)
+        self.transport             = transport
 
     def on_hci_acl_data_packet(self, packet):
         self.assembler.feed_packet(packet)
@@ -364,7 +365,7 @@ class Host(EventEmitter):
 
             connection = self.connections.get(event.connection_handle)
             if connection is None:
-                connection = Connection(self, event.connection_handle, event.role, event.peer_address)
+                connection = Connection(self, event.connection_handle, event.role, event.peer_address, BT_LE_TRANSPORT)
                 self.connections[event.connection_handle] = connection
 
             # Notify the client
@@ -399,7 +400,7 @@ class Host(EventEmitter):
 
             connection = self.connections.get(event.connection_handle)
             if connection is None:
-                connection = Connection(self, event.connection_handle, BT_CENTRAL_ROLE, event.bd_addr)
+                connection = Connection(self, event.connection_handle, BT_CENTRAL_ROLE, event.bd_addr, BT_BR_EDR_TRANSPORT)
                 self.connections[event.connection_handle] = connection
 
             # Notify the client

bumble/smp.py

@@ -44,6 +44,7 @@ logger = logging.getLogger(__name__)
 # Constants
 # -----------------------------------------------------------------------------
 SMP_CID = 0x06
+SMP_BR_CID = 0x07
 
 SMP_PAIRING_REQUEST_COMMAND               = 0x01
 SMP_PAIRING_RESPONSE_COMMAND              = 0x02
@@ -152,6 +153,7 @@ SMP_CT2_AUTHREQ      = 0b00100000
 
 # Crypto salt
 SMP_CTKD_H7_LEBR_SALT = bytes.fromhex('00000000000000000000000000000000746D7031')
+SMP_CTKD_H7_BRLE_SALT = bytes.fromhex('00000000000000000000000000000000746D7032')
 
 # -----------------------------------------------------------------------------
 # Utils
@@ -598,6 +600,7 @@ class Session:
         self.pairing_config              = pairing_config
         self.wait_before_continuing      = None
         self.completed                   = False
+        self.ctkd_task                   = None
 
         # Decide if we're the initiator or the responder
         self.is_initiator = (connection.role == BT_CENTRAL_ROLE)
@@ -635,13 +638,13 @@ class Session:
         # Set up addresses
         peer_address = connection.peer_resolvable_address or connection.peer_address
         if self.is_initiator:
-            self.ia  = bytes(manager.address)
-            self.iat = 1 if manager.address.is_random else 0
+            self.ia  = bytes(connection.local_address)
+            self.iat = 1 if connection.local_address.is_random else 0
             self.ra  = bytes(peer_address)
             self.rat = 1 if peer_address.is_random else 0
         else:
-            self.ra  = bytes(manager.address)
-            self.rat = 1 if manager.address.is_random else 0
+            self.ra  = bytes(connection.local_address)
+            self.rat = 1 if connection.local_address.is_random else 0
             self.ia  = bytes(peer_address)
             self.iat = 1 if peer_address.is_random else 0
 
@@ -877,10 +880,21 @@ class Session:
             )
         )
 
+    async def derive_ltk(self):
+        link_key = await self.manager.device.get_link_key(self.connection.peer_address)
+        assert link_key is not None
+        ilk = crypto.h7(
+            salt=SMP_CTKD_H7_BRLE_SALT,
+            w=link_key) if self.ct2 else crypto.h6(link_key, b'tmp2')
+        self.ltk = crypto.h6(ilk, b'brle')
+
     def distribute_keys(self):
         # Distribute the keys as required
         if self.is_initiator:
-            if not self.sc:
+            # CTKD: Derive LTK from LinkKey
+            if self.connection.transport == BT_BR_EDR_TRANSPORT and self.initiator_key_distribution & SMP_ENC_KEY_DISTRIBUTION_FLAG:
+                self.ctkd_task = asyncio.create_task(self.derive_ltk())
+            elif not self.sc:
                 # Distribute the LTK, EDIV and RAND
                 if self.initiator_key_distribution & SMP_ENC_KEY_DISTRIBUTION_FLAG:
                     self.send_command(SMP_Encryption_Information_Command(long_term_key=self.ltk))
@@ -892,15 +906,15 @@ class Session:
                     SMP_Identity_Information_Command(identity_resolving_key=self.manager.device.irk)
                 )
                 self.send_command(SMP_Identity_Address_Information_Command(
-                    addr_type = self.manager.address.address_type,
-                    bd_addr   = self.manager.address
+                    addr_type = self.manager.identity_address.address_type,
+                    bd_addr   = self.manager.identity_address
                 ))
 
             # Distribute CSRK
             csrk = bytes(16)  # FIXME: testing
             if self.initiator_key_distribution & SMP_SIGN_KEY_DISTRIBUTION_FLAG:
                 self.send_command(SMP_Signing_Information_Command(signature_key=csrk))
-            
+
             # CTKD, calculate BR/EDR link key
             if self.initiator_key_distribution & SMP_LINK_KEY_DISTRIBUTION_FLAG:
                 ilk = crypto.h7(
@@ -909,8 +923,11 @@ class Session:
                 self.link_key = crypto.h6(ilk, b'lebr')
 
         else:
+            # CTKD: Derive LTK from LinkKey
+            if self.connection.transport == BT_BR_EDR_TRANSPORT and self.responder_key_distribution & SMP_ENC_KEY_DISTRIBUTION_FLAG:
+                self.ctkd_task = asyncio.create_task(self.derive_ltk())
             # Distribute the LTK, EDIV and RAND
-            if not self.sc:
+            elif not self.sc:
                 if self.responder_key_distribution & SMP_ENC_KEY_DISTRIBUTION_FLAG:
                     self.send_command(SMP_Encryption_Information_Command(long_term_key=self.ltk))
                     self.send_command(SMP_Master_Identification_Command(ediv=self.ltk_ediv, rand=self.ltk_rand))
@@ -921,15 +938,15 @@ class Session:
                     SMP_Identity_Information_Command(identity_resolving_key=self.manager.device.irk)
                 )
                 self.send_command(SMP_Identity_Address_Information_Command(
-                    addr_type = self.manager.address.address_type,
-                    bd_addr   = self.manager.address
+                    addr_type = self.manager.identity_address.address_type,
+                    bd_addr   = self.manager.identity_address
                 ))
 
             # Distribute CSRK
             csrk = bytes(16)  # FIXME: testing
             if self.responder_key_distribution & SMP_SIGN_KEY_DISTRIBUTION_FLAG:
                 self.send_command(SMP_Signing_Information_Command(signature_key=csrk))
-            
+
             # CTKD, calculate BR/EDR link key
             if self.responder_key_distribution & SMP_LINK_KEY_DISTRIBUTION_FLAG:
                 ilk = crypto.h7(
@@ -940,7 +957,7 @@ class Session:
     def compute_peer_expected_distributions(self, key_distribution_flags):
         # Set our expectations for what to wait for in the key distribution phase
         self.peer_expected_distributions = []
-        if not self.sc:
+        if not self.sc and self.connection.transport == BT_LE_TRANSPORT:
             if (key_distribution_flags & SMP_ENC_KEY_DISTRIBUTION_FLAG != 0):
                 self.peer_expected_distributions.append(SMP_Encryption_Information_Command)
                 self.peer_expected_distributions.append(SMP_Master_Identification_Command)
@@ -968,7 +985,7 @@ class Session:
                     self.distribute_keys()
 
                 # Nothing left to expect, we're done
-                self.on_pairing()
+                asyncio.create_task(self.on_pairing())
         else:
             logger.warn(color(f'!!! unexpected key distribution command: {command_class.__name__}', 'red'))
             self.send_pairing_failed(SMP_UNSPECIFIED_REASON_ERROR)
@@ -999,7 +1016,7 @@ class Session:
         # Do as if the connection had just been encrypted
         self.on_connection_encryption_change()
 
-    def on_pairing(self):
+    async def on_pairing(self):
         logger.debug('pairing complete')
 
         if self.completed:
@@ -1016,11 +1033,16 @@ class Session:
         else:
             peer_address = self.connection.peer_address
 
+        # Wait for link key fetch and key derivation
+        if self.ctkd_task is not None:
+            await self.ctkd_task
+            self.ctkd_task = None
+
         # Create an object to hold the keys
         keys = PairingKeys()
         keys.address_type = peer_address.address_type
         authenticated = self.pairing_method != self.JUST_WORKS
-        if self.sc:
+        if self.sc or self.connection.transport == BT_BR_EDR_TRANSPORT:
             keys.ltk = PairingKeys.Key(
                 value         = self.ltk,
                 authenticated = authenticated
@@ -1059,7 +1081,6 @@ class Session:
                 value         = self.link_key,
                 authenticated = authenticated
             )
-
         self.manager.on_pairing(self, peer_address, keys)
 
     def on_pairing_failure(self, reason):
@@ -1137,6 +1158,12 @@ class Session:
         # Respond
         self.send_pairing_response_command()
 
+        # Vol 3, Part C, 5.2.2.1.3
+        # CTKD over BR/EDR should happen after the connection has been encrypted,
+        # so when receiving pairing requests, responder should start distributing keys
+        if self.connection.transport == BT_BR_EDR_TRANSPORT and self.connection.is_encrypted and self.is_responder and accepted:
+            self.distribute_keys()
+
     def on_smp_pairing_response_command(self, command):
         if self.is_responder:
             logger.warn(color('received pairing response as a responder', 'red'))
@@ -1452,17 +1479,18 @@ class Manager(EventEmitter):
     Implements the Initiator and Responder roles of the Security Manager Protocol
     '''
 
-    def __init__(self, device, address):
+    def __init__(self, device, address, identity_address):
         super().__init__()
         self.device                 = device
-        self.address                = address
+        self.identity_address       = identity_address
         self.sessions               = {}
         self._ecc_key               = None
         self.pairing_config_factory = lambda connection: PairingConfig()
 
     def send_command(self, connection, command):
         logger.debug(f'>>> Sending SMP Command on connection [0x{connection.handle:04X}] {connection.peer_address}: {command}')
-        connection.send_l2cap_pdu(SMP_CID, command.to_bytes())
+        cid = SMP_BR_CID if connection.transport == BT_BR_EDR_TRANSPORT else SMP_CID
+        connection.send_l2cap_pdu(cid, command.to_bytes())
 
     def on_smp_pdu(self, connection, pdu):
         # Look for a session with this connection, and create one if none exists

examples/battery_client.py

@@ -20,7 +20,7 @@ import sys
 import os
 import logging
 from colors import color
-from bumble.device import Device, Peer
+from bumble.device import Device
 from bumble.transport import open_transport
 from bumble.profiles.battery_service import BatteryServiceProxy