auracaster/bumble_mirror
3
0
Fork 1
mirror of https://github.com/google/bumble.git synced 2026-05-09 04:08:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

classic: complete authentication when being the .authenticate acceptor

Browse Source
This commit is contained in:
Abel Lucas
2022-11-22 18:55:26 +00:00
parent 02a474c44e
commit 662704e551
2 changed files with 19 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
bumble/device.py
View File

@@ -415,6 +415,7 @@ class Connection(CompositeEventEmitter):
self.authenticated = False self.authenticated = False
self.sc = False self.sc = False
self.link_key_type = None self.link_key_type = None
self.authenticating = False
self.phy = phy self.phy = phy
self.att_mtu = ATT_DEFAULT_MTU self.att_mtu = ATT_DEFAULT_MTU
self.data_length = DEVICE_DEFAULT_DATA_LENGTH self.data_length = DEVICE_DEFAULT_DATA_LENGTH
@@ -1722,9 +1723,13 @@ class Device(CompositeEventEmitter):
logger.warn(f'HCI_Authentication_Requested_Command failed: {HCI_Constant.error_name(result.status)}') logger.warn(f'HCI_Authentication_Requested_Command failed: {HCI_Constant.error_name(result.status)}')
raise HCI_StatusError(result) raise HCI_StatusError(result)
# Save in connection we are trying to authenticate
connection.authenticating = True
# Wait for the authentication to complete # Wait for the authentication to complete
await pending_authentication await pending_authentication
finally: finally:
connection.authenticating = False
connection.remove_listener('connection_authentication', on_authentication) connection.remove_listener('connection_authentication', on_authentication)
connection.remove_listener('connection_authentication_failure', on_authentication_failure) connection.remove_listener('connection_authentication_failure', on_authentication_failure)
@@ -2062,6 +2067,17 @@ class Device(CompositeEventEmitter):
logger.debug(f'*** Connection Authentication Failure: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}, error={error}') logger.debug(f'*** Connection Authentication Failure: [0x{connection.handle:04X}] {connection.peer_address} as {connection.role_name}, error={error}')
connection.emit('connection_authentication_failure', error) connection.emit('connection_authentication_failure', error)
@host_event_handler
@with_connection_from_address
def on_ssp_complete(self, connection):
# On Secure Simple Pairing complete, in case:
# - Connection isn't already authenticated
# - AND We are not the initiator of the authentication
# We must trigger authentication to known if we are truly authenticated
if not connection.authenticating and not connection.authenticated:
logger.debug(f'*** Trigger Connection Authentication: [0x{connection.handle:04X}] {connection.peer_address}')
asyncio.create_task(connection.authenticate())
# [Classic only] # [Classic only]
@host_event_handler @host_event_handler
@with_connection_from_address @with_connection_from_address

3
bumble/host.py
View File

@@ -595,6 +595,9 @@ class Host(EventEmitter):
def on_hci_simple_pairing_complete_event(self, event): def on_hci_simple_pairing_complete_event(self, event):
logger.debug(f'simple pairing complete for {event.bd_addr}: status={HCI_Constant.status_name(event.status)}') logger.debug(f'simple pairing complete for {event.bd_addr}: status={HCI_Constant.status_name(event.status)}')
# Notify the client
if event.status == HCI_SUCCESS:
self.emit('ssp_complete', event.bd_addr)
def on_hci_pin_code_request_event(self, event): def on_hci_pin_code_request_event(self, event):
# For now, just refuse all requests # For now, just refuse all requests