From 8a1cdef15208d7a41ce71a8cf5fcafe955e2028d Mon Sep 17 00:00:00 2001 From: Gilles Boccon-Gibod Date: Tue, 8 Nov 2022 17:33:29 -0800 Subject: [PATCH 1/3] fix classic connection event filtering --- bumble/device.py | 5 +++++ bumble/hci.py | 3 +++ 2 files changed, 8 insertions(+) diff --git a/bumble/device.py b/bumble/device.py index ce49fb9..07cff34 100644 --- a/bumble/device.py +++ b/bumble/device.py @@ -1217,6 +1217,11 @@ class Device(CompositeEventEmitter): logger.debug('looking for peer by name') peer_address = await self.find_peer_by_name(peer_address, transport) # TODO: timeout + # All BR/EDR addresses should be public addresses + if transport == BT_BR_EDR_TRANSPORT and peer_address.address_type != Address.PUBLIC_DEVICE_ADDRESS: + peer_address = peer_address.clone() + peer_address.address_type = Address.PUBLIC_DEVICE_ADDRESS + def on_connection(connection): if transport == BT_LE_TRANSPORT or ( # match BR/EDR connection event against peer address diff --git a/bumble/hci.py b/bumble/hci.py index cf9f682..387f3ff 100644 --- a/bumble/hci.py +++ b/bumble/hci.py @@ -1706,6 +1706,9 @@ class Address: self.address_type = address_type + def clone(self): + return Address(self.address_bytes, self.address_type) + @property def is_public(self): return self.address_type == self.PUBLIC_DEVICE_ADDRESS or self.address_type == self.PUBLIC_IDENTITY_ADDRESS From 4a1345cf9525648891bf86ed50bfccd4eeb93f36 Mon Sep 17 00:00:00 2001 From: Gilles Boccon-Gibod Date: Tue, 8 Nov 2022 19:10:13 -0800 Subject: [PATCH 2/3] only force the type if the address is passed as a string --- bumble/device.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/bumble/device.py b/bumble/device.py index 07cff34..6ff5a97 100644 --- a/bumble/device.py +++ b/bumble/device.py @@ -1212,15 +1212,16 @@ class Device(CompositeEventEmitter): if type(peer_address) is str: try: peer_address = Address(peer_address) + if transport == BT_BR_EDR_TRANSPORT: + peer_address.address_type = Address.PUBLIC_DEVICE_ADDRESS except ValueError: # If the address is not parsable, assume it is a name instead logger.debug('looking for peer by name') peer_address = await self.find_peer_by_name(peer_address, transport) # TODO: timeout - - # All BR/EDR addresses should be public addresses - if transport == BT_BR_EDR_TRANSPORT and peer_address.address_type != Address.PUBLIC_DEVICE_ADDRESS: - peer_address = peer_address.clone() - peer_address.address_type = Address.PUBLIC_DEVICE_ADDRESS + else: + # All BR/EDR addresses should be public addresses + if transport == BT_BR_EDR_TRANSPORT and peer_address.address_type != Address.PUBLIC_DEVICE_ADDRESS: + raise ValueError('BR/EDR addresses must be PUBLIC') def on_connection(connection): if transport == BT_LE_TRANSPORT or ( From 0f219eff1242562790cae9c19c27acb4a3837cc6 Mon Sep 17 00:00:00 2001 From: Gilles Boccon-Gibod Date: Wed, 9 Nov 2022 13:18:30 -0800 Subject: [PATCH 3/3] address PR comments --- bumble/device.py | 4 +--- bumble/hci.py | 8 ++++++++ 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/bumble/device.py b/bumble/device.py index 6ff5a97..4b3fa43 100644 --- a/bumble/device.py +++ b/bumble/device.py @@ -1211,9 +1211,7 @@ class Device(CompositeEventEmitter): if type(peer_address) is str: try: - peer_address = Address(peer_address) - if transport == BT_BR_EDR_TRANSPORT: - peer_address.address_type = Address.PUBLIC_DEVICE_ADDRESS + peer_address = Address.from_string_for_transport(peer_address, transport) except ValueError: # If the address is not parsable, assume it is a name instead logger.debug('looking for peer by name') diff --git a/bumble/hci.py b/bumble/hci.py index 387f3ff..1527c92 100644 --- a/bumble/hci.py +++ b/bumble/hci.py @@ -1666,6 +1666,14 @@ class Address: def address_type_name(address_type): return name_or_number(Address.ADDRESS_TYPE_NAMES, address_type) + @staticmethod + def from_string_for_transport(string, transport): + if transport == BT_BR_EDR_TRANSPORT: + address_type = Address.PUBLIC_DEVICE_ADDRESS + else: + address_type = Address.RANDOM_DEVICE_ADDRESS + return Address(string, address_type) + @staticmethod def parse_address(data, offset): # Fix the type to a default value. This is used for parsing type-less Classic addresses