Do not mark an LE link as authenticated after Just Works pairing

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
 
@@ -5012,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')
@@ -5025,16 +5035,15 @@ 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')
 
-                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,
@@ -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
@@ -6742,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
@@ -7036,6 +7049,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 +7065,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:

tests/self_test.py

@@ -23,10 +23,12 @@
 
 import pytest
 
-from bumble.core import PhysicalTransport, ProtocolError
+from bumble.att import ATT_INSUFFICIENT_AUTHENTICATION_ERROR, ATT_Error
+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,
@@ -448,6 +450,164 @@ 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_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():