aioble/security: Add DeviceConnection.pairing_in_progress flag.

micropython/bluetooth/aioble/aioble/device.py

@@ -163,14 +163,15 @@ class DeviceConnection:
     _connected = {}
 
     def __init__(self, device):
-        self.device = device
+        self.device: Device = device
         device._connection = self
 
         self.encrypted = False
         self.authenticated = False
         self.bonded = False
         self.key_size = False
         self.mtu = None
+        self.pairing_in_progress = False
 
         self._conn_handle = None
 

micropython/bluetooth/aioble/aioble/security.py

@@ -7,7 +7,7 @@
 import json
 
 from .core import log_info, log_warn, ble, register_irq_handler
-from .device import DeviceConnection
+from .device import DeviceConnection, Device
 
 _IRQ_ENCRYPTION_UPDATE = const(28)
 _IRQ_GET_SECRET = const(29)
@@ -51,6 +51,15 @@
         log_warn("No secrets available")
 
 
+def _get_connection(key) -> DeviceConnection:
+    if not key:
+        return None
+    addr = bytes(reversed(key[-6:]))
+    for connection in DeviceConnection._connected.values():
+        if connection.device.addr == addr:
+            return connection
+
+
 # Call this whenever the secrets dict changes.
 def _save_secrets(arg=None):
     global _modified, _path
@@ -84,6 +93,7 @@
             connection.authenticated = authenticated
             connection.bonded = bonded
             connection.key_size = key_size
+            connection.pairing_in_progress = False
             # TODO: Handle failure.
             if encrypted and connection._pair_event:
                 connection._pair_event.set()
@@ -126,6 +136,11 @@
                     i += 1
             return None
         else:
+            if sec_type in SEC_TYPES_PEER:
+                if conn := _get_connection(key):
+                    log_info("encryption / pairing started", conn)
+                    conn.pairing_in_progress = True
+
             # Return the secret for this key (or None).
             key = sec_type, bytes(key)
             return _secrets.get(key, None)