SMP: Delegate distributed address type

bumble/pairing.py

@@ -19,6 +19,7 @@ import enum
 from typing import Optional, Tuple
 
 from .hci import (
+    Address,
     HCI_NO_INPUT_NO_OUTPUT_IO_CAPABILITY,
     HCI_DISPLAY_ONLY_IO_CAPABILITY,
     HCI_DISPLAY_YES_NO_IO_CAPABILITY,
@@ -168,21 +169,28 @@ class PairingDelegate:
 class PairingConfig:
     """Configuration for the Pairing protocol."""
 
+    class AddressType(enum.IntEnum):
+        PUBLIC = Address.PUBLIC_DEVICE_ADDRESS
+        RANDOM = Address.RANDOM_DEVICE_ADDRESS
+
     def __init__(
         self,
         sc: bool = True,
         mitm: bool = True,
         bonding: bool = True,
         delegate: Optional[PairingDelegate] = None,
+        identity_address_type: Optional[AddressType] = None,
     ) -> None:
         self.sc = sc
         self.mitm = mitm
         self.bonding = bonding
         self.delegate = delegate or PairingDelegate()
+        self.identity_address_type = identity_address_type
 
     def __str__(self) -> str:
         return (
             f'PairingConfig(sc={self.sc}, '
             f'mitm={self.mitm}, bonding={self.bonding}, '
+            f'identity_address_type={self.identity_address_type}, '
             f'delegate[{self.delegate.io_capability}])'
         )

bumble/smp.py

@@ -993,6 +993,19 @@ class Session:
             )
         )
 
+    def send_identity_address_command(self) -> None:
+        identity_address = {
+            None: self.connection.self_address,
+            Address.PUBLIC_DEVICE_ADDRESS: self.manager.device.public_address,
+            Address.RANDOM_DEVICE_ADDRESS: self.manager.device.random_address,
+        }[self.pairing_config.identity_address_type]
+        self.send_command(
+            SMP_Identity_Address_Information_Command(
+                addr_type=identity_address.address_type,
+                bd_addr=identity_address,
+            )
+        )
+
     def start_encryption(self, key: bytes) -> None:
         # We can now encrypt the connection with the short term key, so that we can
         # distribute the long term and/or other keys over an encrypted connection
@@ -1016,6 +1029,7 @@ class Session:
         self.ltk = crypto.h6(ilk, b'brle')
 
     def distribute_keys(self) -> None:
+
         # Distribute the keys as required
         if self.is_initiator:
             # CTKD: Derive LTK from LinkKey
@@ -1045,12 +1059,7 @@ class Session:
                         identity_resolving_key=self.manager.device.irk
                     )
                 )
-                self.send_command(
-                    SMP_Identity_Address_Information_Command(
-                        addr_type=self.connection.self_address.address_type,
-                        bd_addr=self.connection.self_address,
-                    )
-                )
+                self.send_identity_address_command()
 
             # Distribute CSRK
             csrk = bytes(16)  # FIXME: testing
@@ -1094,12 +1103,7 @@ class Session:
                         identity_resolving_key=self.manager.device.irk
                     )
                 )
-                self.send_command(
-                    SMP_Identity_Address_Information_Command(
-                        addr_type=self.connection.self_address.address_type,
-                        bd_addr=self.connection.self_address,
-                    )
-                )
+                self.send_identity_address_command()
 
             # Distribute CSRK
             csrk = bytes(16)  # FIXME: testing

tests/self_test.py

@@ -556,6 +556,26 @@ async def test_self_smp_over_classic():
         MockSmpSession.send_pairing_random_command.assert_not_called()
 
 
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_self_smp_public_address():
+    pairing_config = PairingConfig(
+        mitm=True,
+        sc=True,
+        bonding=True,
+        identity_address_type=PairingConfig.AddressType.PUBLIC,
+        delegate=PairingDelegate(
+            PairingDelegate.IoCapability.DISPLAY_OUTPUT_AND_YES_NO_INPUT,
+            PairingDelegate.KeyDistribution.DISTRIBUTE_ENCRYPTION_KEY
+            | PairingDelegate.KeyDistribution.DISTRIBUTE_IDENTITY_KEY
+            | PairingDelegate.KeyDistribution.DISTRIBUTE_SIGNING_KEY
+            | PairingDelegate.KeyDistribution.DISTRIBUTE_LINK_KEY,
+        ),
+    )
+
+    await _test_self_smp_with_configs(pairing_config, pairing_config)
+
+
 # -----------------------------------------------------------------------------
 async def run_test_self():
     await test_self_connection()
@@ -565,6 +585,7 @@ async def run_test_self():
     await test_self_smp_reject()
     await test_self_smp_wrong_pin()
     await test_self_smp_over_classic()
+    await test_self_smp_public_address()
 
 
 # -----------------------------------------------------------------------------