le oob

apps/pair.py

@@ -24,10 +24,16 @@
 from bumble.colors import color
 from bumble.device import Device, Peer
 from bumble.transport import open_transport_or_link
-from bumble.pairing import PairingDelegate, PairingConfig
+from bumble.pairing import OobData, PairingDelegate, PairingConfig
+from bumble.smp import OobContext, OobLegacyContext
 from bumble.smp import error_name as smp_error_name
 from bumble.keys import JsonKeyStore
-from bumble.core import ProtocolError
+from bumble.core import (
+    AdvertisingData,
+    ProtocolError,
+    BT_LE_TRANSPORT,
+    BT_BR_EDR_TRANSPORT,
+)
 from bumble.gatt import (
     GATT_DEVICE_NAME_CHARACTERISTIC,
     GATT_GENERIC_ACCESS_SERVICE,
@@ -60,7 +66,7 @@ async def wait_until_terminated(self):
 class Delegate(PairingDelegate):
     def __init__(self, mode, connection, capability_string, do_prompt):
         super().__init__(
-            {
+            io_capability={
                 'keyboard': PairingDelegate.KEYBOARD_INPUT_ONLY,
                 'display': PairingDelegate.DISPLAY_OUTPUT_ONLY,
                 'display+keyboard': PairingDelegate.DISPLAY_OUTPUT_AND_KEYBOARD_INPUT,
@@ -286,6 +292,7 @@ async def pair(
     bond,
     ctkd,
     io,
+    oob,
     prompt,
     request,
     print_keys,
@@ -343,16 +350,51 @@ async def pair(
             await device.keystore.print(prefix=color('@@@ ', 'blue'))
             print(color('@@@-----------------------------------', 'blue'))
 
+        # Create an OOB context if needed
+        if oob:
+            our_oob_context = OobContext()
+            shared_data = (
+                None
+                if oob == '-'
+                else OobData.from_ad(AdvertisingData.from_bytes(bytes.fromhex(oob)))
+            )
+            legacy_context = OobLegacyContext()
+            oob_contexts = PairingConfig.OobConfig(
+                our_context=our_oob_context,
+                peer_data=shared_data,
+                legacy_context=legacy_context,
+            )
+            oob_data = OobData(
+                address=device.random_address,
+                shared_data=shared_data,
+                legacy_context=legacy_context,
+            )
+            print(color('@@@-----------------------------------', 'yellow'))
+            print(color('@@@ OOB Data:', 'yellow'))
+            print(color(f'@@@   {our_oob_context.share()}', 'yellow'))
+            print(color(f'@@@   TK={legacy_context.tk.hex()}', 'yellow'))
+            print(color(f'@@@   HEX: ({bytes(oob_data.to_ad()).hex()})', 'yellow'))
+            print(color('@@@-----------------------------------', 'yellow'))
+        else:
+            oob_contexts = None
+
         # Set up a pairing config factory
         device.pairing_config_factory = lambda connection: PairingConfig(
-            sc, mitm, bond, Delegate(mode, connection, io, prompt)
+            sc=sc,
+            mitm=mitm,
+            bonding=bond,
+            oob=oob_contexts,
+            delegate=Delegate(mode, connection, io, prompt),
         )
 
         # Connect to a peer or wait for a connection
         device.on('connection', lambda connection: on_connection(connection, request))
         if address_or_name is not None:
             print(color(f'=== Connecting to {address_or_name}...', 'green'))
-            connection = await device.connect(address_or_name)
+            connection = await device.connect(
+                address_or_name,
+                transport=BT_LE_TRANSPORT if mode == 'le' else BT_BR_EDR_TRANSPORT,
+            )
 
             if not request:
                 try:
@@ -421,6 +463,14 @@ def emit(self, record):
     default='display+keyboard',
     show_default=True,
 )
+@click.option(
+    '--oob',
+    metavar='<oob-data-hex>',
+    help=(
+        'Use OOB pairing with this data from the peer '
+        '(use "-" to enable OOB without peer data)'
+    ),
+)
 @click.option('--prompt', is_flag=True, help='Prompt to accept/reject pairing request')
 @click.option(
     '--request', is_flag=True, help='Request that the connecting peer initiate pairing'
@@ -441,6 +491,7 @@ def main(
     bond,
     ctkd,
     io,
+    oob,
     prompt,
     request,
     print_keys,
@@ -464,6 +515,7 @@ def main(
             bond,
             ctkd,
             io,
+            oob,
             prompt,
             request,
             print_keys,

bumble/core.py

@@ -16,6 +16,7 @@
 # Imports
 # -----------------------------------------------------------------------------
 from __future__ import annotations
+import enum
 import struct
 from typing import List, Optional, Tuple, Union, cast, Dict
 
@@ -1051,3 +1052,13 @@ def __init__(self, tx_phy, rx_phy):
 
     def __str__(self):
         return f'ConnectionPHY(tx_phy={self.tx_phy}, rx_phy={self.rx_phy})'
+
+
+# -----------------------------------------------------------------------------
+# LE Role
+# -----------------------------------------------------------------------------
+class LeRole(enum.IntEnum):
+    PERIPHERAL_ONLY = 0x00
+    CENTRAL_ONLY = 0x01
+    BOTH_PERIPHERAL_PREFERRED = 0x02
+    BOTH_CENTRAL_PREFERRED = 0x03

bumble/pairing.py

@@ -15,7 +15,9 @@
 # -----------------------------------------------------------------------------
 # Imports
 # -----------------------------------------------------------------------------
+from __future__ import annotations
 import enum
+from dataclasses import dataclass
 from typing import Optional, Tuple
 
 from .hci import (
@@ -35,7 +37,60 @@
     SMP_ID_KEY_DISTRIBUTION_FLAG,
     SMP_SIGN_KEY_DISTRIBUTION_FLAG,
     SMP_LINK_KEY_DISTRIBUTION_FLAG,
+    OobContext,
+    OobLegacyContext,
+    OobSharedData,
 )
+from .core import AdvertisingData, LeRole
+
+
+# -----------------------------------------------------------------------------
+@dataclass
+class OobData:
+    """OOB data that can be sent from one device to another."""
+
+    address: Optional[Address] = None
+    role: Optional[LeRole] = None
+    shared_data: Optional[OobSharedData] = None
+    legacy_context: Optional[OobLegacyContext] = None
+
+    @classmethod
+    def from_ad(cls, ad: AdvertisingData) -> OobData:
+        instance = cls()
+        shared_data_c: Optional[bytes] = None
+        shared_data_r: Optional[bytes] = None
+        for ad_type, ad_data in ad.ad_structures:
+            if ad_type == AdvertisingData.LE_BLUETOOTH_DEVICE_ADDRESS:
+                instance.address = Address(ad_data)
+            elif ad_type == AdvertisingData.LE_ROLE:
+                instance.role = LeRole(ad_data[0])
+            elif ad_type == AdvertisingData.LE_SECURE_CONNECTIONS_CONFIRMATION_VALUE:
+                shared_data_c = ad_data
+            elif ad_type == AdvertisingData.LE_SECURE_CONNECTIONS_RANDOM_VALUE:
+                shared_data_r = ad_data
+            elif ad_type == AdvertisingData.SECURITY_MANAGER_TK_VALUE:
+                instance.legacy_context = OobLegacyContext(tk=ad_data)
+        if shared_data_c and shared_data_r:
+            instance.shared_data = OobSharedData(c=shared_data_c, r=shared_data_r)
+
+        return instance
+
+    def to_ad(self) -> AdvertisingData:
+        ad_structures = []
+        if self.address is not None:
+            ad_structures.append(
+                (AdvertisingData.LE_BLUETOOTH_DEVICE_ADDRESS, bytes(self.address))
+            )
+        if self.role is not None:
+            ad_structures.append((AdvertisingData.LE_ROLE, bytes([self.role])))
+        if self.shared_data is not None:
+            ad_structures.extend(self.shared_data.to_ad().ad_structures)
+        if self.legacy_context is not None:
+            ad_structures.append(
+                (AdvertisingData.SECURITY_MANAGER_TK_VALUE, self.legacy_context.tk)
+            )
+
+        return AdvertisingData(ad_structures)
 
 
 # -----------------------------------------------------------------------------
@@ -173,24 +228,35 @@ class AddressType(enum.IntEnum):
         PUBLIC = Address.PUBLIC_DEVICE_ADDRESS
         RANDOM = Address.RANDOM_DEVICE_ADDRESS
 
+    @dataclass
+    class OobConfig:
+        """Config for OOB pairing."""
+
+        our_context: Optional[OobContext]
+        peer_data: Optional[OobSharedData]
+        legacy_context: Optional[OobLegacyContext]
+
     def __init__(
         self,
         sc: bool = True,
         mitm: bool = True,
         bonding: bool = True,
         delegate: Optional[PairingDelegate] = None,
         identity_address_type: Optional[AddressType] = None,
+        oob: Optional[OobConfig] = None,
     ) -> None:
         self.sc = sc
         self.mitm = mitm
         self.bonding = bonding
         self.delegate = delegate or PairingDelegate()
         self.identity_address_type = identity_address_type
+        self.oob = oob
 
     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}])'
+            f'delegate[{self.delegate.io_capability}]), '
+            f'oob[{self.oob}])'
         )