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

le: save own_address_type on ACL connection for SMP to be able to use the right self address

Browse Source
This commit is contained in:
Abel Lucas
2022-11-10 02:06:37 +00:00
parent 8bbba7c84c
commit eb46f60c87
3 changed files with 74 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
bumble/device.py
View File

@@ -394,6 +394,7 @@ class Connection(CompositeEventEmitter):
device,
handle,
transport,
self_address,
peer_address,
peer_resolvable_address,
role,
@@ -404,6 +405,7 @@ class Connection(CompositeEventEmitter):
self.device = device
self.handle = handle
self.transport = transport
self.self_address = self_address
self.peer_address = peer_address
self.peer_resolvable_address = peer_resolvable_address
self.peer_name = None # Classic only
@@ -699,6 +701,10 @@ class Device(CompositeEventEmitter):
self.address_resolver = None
self.classic_pending_accepts = {Address.ANY: []} # Futures, by BD address OR [Futures] for Address.ANY
# Own address type cache
self.advertising_own_address_type = None
self.connect_own_address_type = None
# Use the initial config or a default
self.public_address = Address('00:00:00:00:00:00')
if config is None:
@@ -731,8 +737,7 @@ class Device(CompositeEventEmitter):
self.random_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.l2cap_channel_manager.register_fixed_channel(
smp.SMP_CID, self.on_smp_pdu)
self.l2cap_channel_manager.register_fixed_channel(
@@ -928,7 +933,7 @@ class Device(CompositeEventEmitter):
self,
advertising_type=AdvertisingType.UNDIRECTED_CONNECTABLE_SCANNABLE,
target=None,
own_address_type=Address.RANDOM_DEVICE_ADDRESS,
own_address_type=OwnAddressType.RANDOM,
auto_restart=False
):
# If we're advertising, stop first
@@ -975,9 +980,10 @@ class Device(CompositeEventEmitter):
advertising_enable = 1
), check_result=True)
self.auto_restart_advertising = auto_restart
self.advertising_type = advertising_type
self.advertising = True
self.advertising_own_address_type = own_address_type
self.auto_restart_advertising = auto_restart
self.advertising_type = advertising_type
self.advertising = True
async def stop_advertising(self):
# Disable advertising
@@ -986,9 +992,10 @@ class Device(CompositeEventEmitter):
advertising_enable = 0
), check_result=True)
self.advertising = False
self.advertising_type = None
self.auto_restart_advertising = False
self.advertising_own_address_type = None
self.advertising = False
self.advertising_type = None
self.auto_restart_advertising = False
@property
def is_advertising(self):
@@ -1000,7 +1007,7 @@ class Device(CompositeEventEmitter):
active=True,
scan_interval=DEVICE_DEFAULT_SCAN_INTERVAL, # Scan interval in ms
scan_window=DEVICE_DEFAULT_SCAN_WINDOW, # Scan window in ms
own_address_type=Address.RANDOM_DEVICE_ADDRESS,
own_address_type=OwnAddressType.RANDOM,
filter_duplicates=False,
scanning_phys=(HCI_LE_1M_PHY, HCI_LE_CODED_PHY)
):
@@ -1181,7 +1188,7 @@ class Device(CompositeEventEmitter):
peer_address,
transport=BT_LE_TRANSPORT,
connection_parameters_preferences=None,
own_address_type=Address.RANDOM_DEVICE_ADDRESS,
own_address_type=OwnAddressType.RANDOM,
timeout=DEVICE_DEFAULT_CONNECT_TIMEOUT
):
'''
@@ -1251,6 +1258,8 @@ class Device(CompositeEventEmitter):
HCI_LE_CODED_PHY: ConnectionParametersPreferences.default
}
self.connect_own_address_type = own_address_type
if self.host.supports_command(HCI_LE_EXTENDED_CREATE_CONNECTION_COMMAND):
# Only keep supported PHYs
phys = sorted(list(set(filter(self.supports_le_phy, connection_parameters_preferences.keys()))))
@@ -1350,6 +1359,7 @@ class Device(CompositeEventEmitter):
self.remove_listener('connection_failure', on_connection_failure)
if transport == BT_LE_TRANSPORT:
self.le_connecting = False
self.connect_own_address_type = None
async def accept(
self,
@@ -1847,6 +1857,7 @@ class Device(CompositeEventEmitter):
self,
connection_handle,
transport,
self.public_address,
peer_address,
peer_resolvable_address,
role,
@@ -1875,8 +1886,17 @@ class Device(CompositeEventEmitter):
peer_resolvable_address = peer_address
peer_address = resolved_address
# Guess which own address type is used for this connection.
# This logic is somewhat correct but may need to be improved
# when multiple advertising are run simultaneously.
if self.connect_own_address_type is not None:
own_address_type = self.connect_own_address_type
else:
own_address_type = self.advertising_own_address_type
# We are no longer advertising
self.advertising = False
self.advertising_own_address_type = None
self.advertising = False
# Create and notify of the new connection asynchronously
async def new_connection():
@@ -1890,11 +1910,16 @@ class Device(CompositeEventEmitter):
else:
phy = ConnectionPHY(HCI_LE_1M_PHY, HCI_LE_1M_PHY)
self_address = self.random_address
if own_address_type in (OwnAddressType.PUBLIC, OwnAddressType.RESOLVABLE_OR_PUBLIC):
self_address = self.public_address
# Create a new connection
connection = Connection(
self,
connection_handle,
transport,
self_address,
peer_address,
peer_resolvable_address,
role,
@@ -1914,7 +1939,8 @@ class Device(CompositeEventEmitter):
# For directed advertising, this means a timeout
if transport == BT_LE_TRANSPORT and self.advertising and self.advertising_type.is_directed:
self.advertising = False
self.advertising_own_address_type = None
self.advertising = False
# Notify listeners
error = ConnectionError(

30
bumble/hci.py
View File

@@ -1756,6 +1756,26 @@ class Address:
return ':'.join([f'{x:02X}' for x in reversed(self.address_bytes)])
# -----------------------------------------------------------------------------
class OwnAddressType:
PUBLIC = 0
RANDOM = 1
RESOLVABLE_OR_PUBLIC = 2
RESOLVABLE_OR_RANDOM = 3
TYPE_NAMES = {
PUBLIC: 'PUBLIC',
RANDOM: 'RANDOM',
RESOLVABLE_OR_PUBLIC: 'RESOLVABLE_OR_PUBLIC',
RESOLVABLE_OR_RANDOM: 'RESOLVABLE_OR_RANDOM'
}
@staticmethod
def type_name(type):
return name_or_number(OwnAddressType.TYPE_NAMES, type)
TYPE_SPEC = {'size': 1, 'mapper': lambda x: OwnAddressType.type_name(x)}
# -----------------------------------------------------------------------------
class HCI_Packet:
'''
@@ -2848,7 +2868,7 @@ class HCI_LE_Set_Random_Address_Command(HCI_Command):
('advertising_interval_min', 2),
('advertising_interval_max', 2),
('advertising_type', {'size': 1, 'mapper': lambda x: HCI_LE_Set_Advertising_Parameters_Command.advertising_type_name(x)}),
('own_address_type', Address.ADDRESS_TYPE_SPEC),
('own_address_type', OwnAddressType.TYPE_SPEC),
('peer_address_type', Address.ADDRESS_TYPE_SPEC),
('peer_address', Address.parse_address_preceded_by_type),
('advertising_channel_map', 1),
@@ -2927,7 +2947,7 @@ class HCI_LE_Set_Advertising_Enable_Command(HCI_Command):
('le_scan_type', 1),
('le_scan_interval', 2),
('le_scan_window', 2),
('own_address_type', Address.ADDRESS_TYPE_SPEC),
('own_address_type', OwnAddressType.TYPE_SPEC),
('scanning_filter_policy', 1)
])
class HCI_LE_Set_Scan_Parameters_Command(HCI_Command):
@@ -2961,7 +2981,7 @@ class HCI_LE_Set_Scan_Enable_Command(HCI_Command):
('initiator_filter_policy', 1),
('peer_address_type', Address.ADDRESS_TYPE_SPEC),
('peer_address', Address.parse_address_preceded_by_type),
('own_address_type', Address.ADDRESS_TYPE_SPEC),
('own_address_type', OwnAddressType.TYPE_SPEC),
('connection_interval_min', 2),
('connection_interval_max', 2),
('max_latency', 2),
@@ -3283,7 +3303,7 @@ class HCI_LE_Set_Advertising_Set_Random_Address_Command(HCI_Command):
('primary_advertising_interval_min', 3),
('primary_advertising_interval_max', 3),
('primary_advertising_channel_map', {'size': 1, 'mapper': lambda x: HCI_LE_Set_Extended_Advertising_Parameters_Command.channel_map_string(x)}),
('own_address_type', Address.ADDRESS_TYPE_SPEC),
('own_address_type', OwnAddressType.TYPE_SPEC),
('peer_address_type', Address.ADDRESS_TYPE_SPEC),
('peer_address', Address.parse_address_preceded_by_type),
('advertising_filter_policy', 1),
@@ -3687,7 +3707,7 @@ class HCI_LE_Extended_Create_Connection_Command(HCI_Command):
initiating_phys_strs = bit_flags_to_strings(self.initiating_phys, HCI_LE_PHY_BIT_NAMES)
fields = [
('initiator_filter_policy:', self.initiator_filter_policy),
('own_address_type: ', Address.address_type_name(self.own_address_type)),
('own_address_type: ', OwnAddressType.type_name(self.own_address_type)),
('peer_address_type: ', Address.address_type_name(self.peer_address_type)),
('peer_address: ', str(self.peer_address)),
('initiating_phys: ', ','.join(initiating_phys_strs)),

20
bumble/smp.py
View File

@@ -640,15 +640,16 @@ class Session:
self.oob = False
# Set up addresses
self_address = connection.self_address
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(self_address)
self.iat = 1 if self_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(self_address)
self.rat = 1 if self_address.is_random else 0
self.ia = bytes(peer_address)
self.iat = 1 if peer_address.is_random else 0
@@ -925,8 +926,8 @@ 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.connection.self_address.address_type,
bd_addr = self.connection.self_address
))
# Distribute CSRK
@@ -957,8 +958,8 @@ 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.connection.self_address.address_type,
bd_addr = self.connection.self_address
))
# Distribute CSRK
@@ -1504,10 +1505,9 @@ class Manager(EventEmitter):
Implements the Initiator and Responder roles of the Security Manager Protocol
'''
def __init__(self, device, address):
def __init__(self, device):
super().__init__()
self.device = device
self.address = address
self.sessions = {}
self._ecc_key = None
self.pairing_config_factory = lambda connection: PairingConfig()