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

Merge pull request #35 from zxzxwu/ctkd

Browse Source 
Support CTKD over BR/EDR
This commit is contained in:
Gilles Boccon-Gibod
2022-08-30 06:19:57 -07:00
committed by GitHub
parent 84d70ad4f3 99cba19d7c
commit d188041694
3 changed files with 42 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bumble/device.py
View File

@@ -497,6 +497,8 @@ class Device(CompositeEventEmitter):
self.smp_manager = smp.Manager(self, self.random_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)

7
bumble/host.py
View File

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

44
bumble/smp.py
View File

@@ -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)
@@ -876,11 +879,22 @@ 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))
@@ -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))
@@ -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'))
@@ -1462,7 +1489,8 @@ class Manager(EventEmitter):
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