From ebef61686318510dba9a2380fabe9d8a573be468 Mon Sep 17 00:00:00 2001 From: deadcafe Date: Mon, 5 Oct 2026 04:20:38 +0530 Subject: [PATCH 1/2] Do not mark an LE link as authenticated after Just Works pairing `Connection.authenticated` was set to true after every pairing, and again whenever LE encryption was enabled, so the GATT server gave access to attributes with READ/WRITE_REQUIRES_AUTHENTICATION to any peer that paired with Just Works. Derive the flag from the key that the link is encrypted with: - after pairing, from the `authenticated` flag of the keys that the pairing generated, which is false for Just Works - when encrypting with an LTK from the key store, from the `authenticated` flag of that key, applied once encryption is enabled. `sc` is set the same way, instead of always being set to true. Classic links are not changed. --- bumble/device.py | 27 +++++++-- tests/self_test.py | 134 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 157 insertions(+), 4 deletions(-) diff --git a/bumble/device.py b/bumble/device.py index 9240fe93..2d6e83bc 100644 --- a/bumble/device.py +++ b/bumble/device.py @@ -1784,6 +1784,7 @@ class Connection(utils.CompositeEventEmitter): encryption_key_size: int authenticated: bool sc: bool + ltk_security: tuple[bool, bool] | None gatt_client: gatt_client.Client pairing_peer_io_capability: int | None pairing_peer_authentication_requirements: int | None @@ -1915,6 +1916,9 @@ def __init__( self.encryption_key_size = 0 self.authenticated = False self.sc = False + # (authenticated, sc) of the stored LTK that was last given to the + # controller, applied once encryption with that key is enabled [LE only] + self.ltk_security = None self.att_mtu = att.ATT_DEFAULT_MTU self.data_length = self.LeDataLength(*DEVICE_DEFAULT_DATA_LENGTH) self.gatt_client = gatt_client.Client(self) # Per-connection client @@ -4943,12 +4947,15 @@ async def get_long_term_key( if keys is not None: logger.debug('found keys in the key store') if keys.ltk: + connection.ltk_security = (keys.ltk.authenticated, True) return keys.ltk.value if connection.role == hci.Role.CENTRAL and keys.ltk_central: + connection.ltk_security = (keys.ltk_central.authenticated, False) return keys.ltk_central.value if connection.role == hci.Role.PERIPHERAL and keys.ltk_peripheral: + connection.ltk_security = (keys.ltk_peripheral.authenticated, False) return keys.ltk_peripheral.value return None @@ -5025,10 +5032,12 @@ def _(error_code: int): ltk = keys.ltk.value rand = bytes(8) ediv = 0 + connection.ltk_security = (keys.ltk.authenticated, True) elif keys.ltk_central is not None: ltk = keys.ltk_central.value rand = keys.ltk_central.rand or b'' ediv = keys.ltk_central.ediv or 0 + connection.ltk_security = (keys.ltk_central.authenticated, False) else: raise InvalidOperationError('no LTK found for peer') @@ -6726,12 +6735,14 @@ def on_connection_encryption_change( connection.authenticated = True connection.sc = True if ( - not connection.authenticated + connection.ltk_security is not None and connection.transport == PhysicalTransport.LE and encryption == hci.HCI_Encryption_Change_Event.Enabled.E0_OR_AES_CCM ): - connection.authenticated = True - connection.sc = True + # The link is only as secure as the stored key it is encrypted with. + # When pairing, this is set in `on_pairing` instead. + connection.authenticated, connection.sc = connection.ltk_security + connection.ltk_security = None connection.emit(connection.EVENT_CONNECTION_ENCRYPTION_CHANGE) @host_event_handler @@ -7036,6 +7047,8 @@ def on_classic_pairing_failure(self, connection: Connection, status: int) -> Non self.emit(connection.EVENT_CLASSIC_PAIRING_FAILURE, connection, status) def on_pairing_start(self, connection: Connection) -> None: + # The link will be encrypted with a new key, not with a stored one + connection.ltk_security = None connection.emit(connection.EVENT_PAIRING_START) self.emit(connection.EVENT_PAIRING_START, connection) @@ -7050,7 +7063,13 @@ def on_pairing( connection.peer_resolvable_address = connection.peer_address connection.peer_address = identity_address connection.sc = sc - connection.authenticated = True + # Only a pairing method with MITM protection (i.e. not Just Works) + # authenticates the link, which the keys that it generated keep track of. + connection.authenticated = any( + key.authenticated + for key in (keys.ltk, keys.ltk_central, keys.ltk_peripheral) + if key is not None + ) connection.emit(connection.EVENT_PAIRING, keys) def on_pairing_failure(self, connection: Connection, reason: int) -> None: diff --git a/tests/self_test.py b/tests/self_test.py index d067bdc4..d766ef1c 100644 --- a/tests/self_test.py +++ b/tests/self_test.py @@ -23,6 +23,7 @@ import pytest +from bumble.att import ATT_INSUFFICIENT_AUTHENTICATION_ERROR, ATT_Error from bumble.core import PhysicalTransport, ProtocolError from bumble.device import Peer from bumble.gatt import Characteristic, Service @@ -448,6 +449,139 @@ async def compare_numbers(self, number, digits): assert not paired +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'sc, authenticated, reconnect', + [ + (False, False, False), + (False, True, False), + (True, False, False), + (True, True, False), + (True, False, True), + (True, True, True), + ], +) +async def test_self_smp_authentication_requirement(sc, authenticated, reconnect): + # Create two devices, each with a controller, attached to the same link + two_devices = TwoDevices() + + # Add a GATT characteristic that requires authentication to device 1 + characteristic = Characteristic( + 'FDB159DB-036C-49E3-B3DB-6325AC750806', + Characteristic.Properties.READ | Characteristic.Properties.WRITE, + Characteristic.READABLE + | Characteristic.WRITEABLE + | Characteristic.READ_REQUIRES_AUTHENTICATION + | Characteristic.WRITE_REQUIRES_AUTHENTICATION, + bytes([1, 2, 3]), + ) + service = Service('3A657F47-D34F-46B3-B1EC-698E29B6B829', [characteristic]) + two_devices.devices[1].add_service(service) + + # Pair with a passkey, or with Just Works when there is no way to enter one + passkey = asyncio.get_running_loop().create_future() + + class Delegate(PairingDelegate): + async def display_number(self, number, digits): + passkey.set_result(number) + + async def get_number(self): + return await passkey + + if authenticated: + io_capabilities = [ + PairingDelegate.IoCapability.KEYBOARD_INPUT_ONLY, + PairingDelegate.IoCapability.DISPLAY_OUTPUT_ONLY, + ] + else: + io_capabilities = [PairingDelegate.IoCapability.NO_OUTPUT_NO_INPUT] * 2 + # Bond with the random addresses, which are the ones used to connect + pairing_configs = [ + PairingConfig( + sc=sc, + identity_address_type=PairingConfig.AddressType.RANDOM, + delegate=Delegate(io_capability), + ) + for io_capability in io_capabilities + ] + two_devices.devices[0].pairing_config_factory = lambda connection: pairing_configs[ + 0 + ] + two_devices.devices[1].pairing_config_factory = lambda connection: pairing_configs[ + 1 + ] + + # Connect and pair the two devices + await two_devices.setup_connection() + two_devices.connections[0].on( + 'pairing', lambda keys: two_devices.on_paired(0, keys) + ) + two_devices.connections[1].on( + 'pairing', lambda keys: two_devices.on_paired(1, keys) + ) + await two_devices.devices[0].pair(two_devices.connections[0]) + await asyncio.gather(*two_devices.paired) + + if reconnect: + # Reconnect and encrypt the link with the stored keys + await two_devices.connections[0].disconnect() + await async_barrier() + await two_devices.devices[1].start_advertising(advertising_interval_min=1.0) + await two_devices.devices[0].connect(two_devices.devices[1].random_address) + await async_barrier() + assert not two_devices.connections[1].authenticated + await two_devices.connections[0].encrypt() + await async_barrier() + + for connection in two_devices.connections.values(): + assert connection.is_encrypted + assert connection.authenticated == authenticated + assert connection.sc == sc + + # Only an authenticated link gives access to the characteristic + peer = Peer(two_devices.connections[0]) + await peer.discover_services() + await peer.discover_characteristics() + remote_characteristic = peer.get_characteristics_by_uuid(characteristic.uuid)[0] + if authenticated: + assert await remote_characteristic.read_value() == bytes([1, 2, 3]) + await remote_characteristic.write_value(bytes([4]), with_response=True) + assert characteristic.value == bytes([4]) + else: + with pytest.raises(ATT_Error) as error: + await remote_characteristic.read_value() + assert error.value.error_code == ATT_INSUFFICIENT_AUTHENTICATION_ERROR + with pytest.raises(ATT_Error) as error: + await remote_characteristic.write_value(bytes([4]), with_response=True) + assert error.value.error_code == ATT_INSUFFICIENT_AUTHENTICATION_ERROR + assert characteristic.value == bytes([1, 2, 3]) + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_self_smp_stale_stored_key_security(): + # Create two devices, each with a controller, attached to the same link + two_devices = TwoDevices() + await two_devices.setup_connection() + connection = two_devices.connections[0] + connection.on('pairing', lambda keys: two_devices.on_paired(0, keys)) + + # As left by a failed attempt to encrypt with a stored authenticated key + connection.ltk_security = (True, True) + authenticated = [] + connection.on( + 'connection_encryption_change', + lambda: authenticated.append(connection.authenticated), + ) + + # Pairing with Just Works must not make the link authenticated at any point + await two_devices.devices[0].pair(connection) + await two_devices.paired[0] + assert authenticated == [False] + assert not connection.authenticated + + # ----------------------------------------------------------------------------- @pytest.mark.asyncio async def test_self_smp_over_classic(): From e2e75c18b55fed5e95b9f3d9d89a9c4f81d93e02 Mon Sep 17 00:00:00 2001 From: deadcafe Date: Mon, 5 Oct 2026 16:07:14 +0530 Subject: [PATCH 2/2] Do not leave stored key security flags on a link the key did not encrypt Address review comments: - Device.encrypt checks the role before looking up the key, so a call on a peripheral does not leave Connection.ltk_security set. - Connection.ltk_security is cleared when encryption fails. --- bumble/device.py | 8 +++++--- tests/self_test.py | 28 +++++++++++++++++++++++++++- 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/bumble/device.py b/bumble/device.py index 2d6e83bc..541ed9bd 100644 --- a/bumble/device.py +++ b/bumble/device.py @@ -5019,6 +5019,9 @@ def _(error_code: int): pending_encryption.set_exception(hci.HCI_Error(error_code)) if connection.transport == PhysicalTransport.LE: + if connection.role != hci.Role.CENTRAL: + raise InvalidStateError('only centrals can start encryption') + # Look for a key in the key store if self.keystore is None: raise InvalidOperationError('no key store') @@ -5041,9 +5044,6 @@ def _(error_code: int): else: raise InvalidOperationError('no LTK found for peer') - if connection.role != hci.Role.CENTRAL: - raise InvalidStateError('only centrals can start encryption') - await self.send_async_command( hci.HCI_LE_Enable_Encryption_Command( connection_handle=connection.handle, @@ -6753,6 +6753,8 @@ def on_connection_encryption_failure(self, connection, error): f'{connection.peer_address} as {connection.role_name}, ' f'error={error}' ) + # The stored key, if one was used, did not encrypt the link + connection.ltk_security = None connection.emit(connection.EVENT_CONNECTION_ENCRYPTION_FAILURE, error) @host_event_handler diff --git a/tests/self_test.py b/tests/self_test.py index d766ef1c..bc199e25 100644 --- a/tests/self_test.py +++ b/tests/self_test.py @@ -24,10 +24,11 @@ import pytest from bumble.att import ATT_INSUFFICIENT_AUTHENTICATION_ERROR, ATT_Error -from bumble.core import PhysicalTransport, ProtocolError +from bumble.core import InvalidStateError, PhysicalTransport, ProtocolError from bumble.device import Peer from bumble.gatt import Characteristic, Service from bumble.hci import Role +from bumble.keys import PairingKeys from bumble.pairing import PairingConfig, PairingDelegate from bumble.smp import ( ErrorCode, @@ -582,6 +583,31 @@ async def test_self_smp_stale_stored_key_security(): assert not connection.authenticated +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_self_smp_unused_stored_key_security(): + # Create two devices, each with a controller, attached to the same link + two_devices = TwoDevices() + await two_devices.setup_connection() + device = two_devices.devices[1] + connection = two_devices.connections[1] + + # A peripheral cannot start encryption, so its stored key is not used + assert device.keystore is not None + await device.keystore.update( + str(connection.peer_address), + PairingKeys(ltk=PairingKeys.Key(value=bytes(16), authenticated=True)), + ) + with pytest.raises(InvalidStateError): + await connection.encrypt() + assert connection.ltk_security is None + + # Nor is a stored key that fails to encrypt the link + connection.ltk_security = (True, True) + device.on_connection_encryption_failure(connection.handle, 0x06) + assert connection.ltk_security is None + + # ----------------------------------------------------------------------------- @pytest.mark.asyncio async def test_self_smp_over_classic():