Introduced different key stores for LE secure, using json file or NVS

Author: Heerkog

hid_keystores.py

@@ -0,0 +1,114 @@
+import esp32
+import json
+import binascii
+
+# Class that represents a generic keystore
+class KeyStore(object):
+
+    def __init__(self):
+        self.secrets = {}
+
+    def add_secret(self, type, key, value):
+        _key = (type, bytes(key))
+        self.secrets[_key] = bytes(value)
+
+    def get_secret(self, type, index, key):
+        _key = (type, bytes(key) if key else None)
+        value = None
+
+        if key is None:
+            i = 0
+            for (t, _k), _val in self.secrets.items():
+                if t == type:
+                    if i == index:
+                        value = _val
+                    i += 1
+        else:
+            value = self.secrets.get(_key, None)
+
+        return value
+
+    def remove_secret(self, type, key):
+        _key = (type, bytes(key))
+        del self.secrets[_key]
+
+    def has_secret(self, type, key):
+        _key = (type, bytes(key))
+        return _key in self.secrets
+
+    def json_dump(self):
+        json_secrets = [
+            (sec_type, binascii.b2a_base64(key, newline=False), binascii.b2a_base64(value, newline=False))
+            for (sec_type, key), value in self.secrets.items()
+        ]
+        return json_secrets
+
+    def add_json_entries(self, entries):
+        for sec_type, key, value in entries:
+            self.secrets[sec_type, binascii.a2b_base64(key)] = binascii.a2b_base64(value)
+
+    def load_secrets(self):
+        return
+
+    def save_secrets(self):
+        return
+
+
+# Class that uses a JSON file as keystore
+class JSONKeyStore(KeyStore):
+
+    def __init__(self):
+        super(JSONKeyStore, self).__init__()
+
+    def load_secrets(self):
+        try:
+            with open("keys.json", "r") as file:
+                self.add_json_entries(json.load(file))
+        except:
+            print("No secrets available")
+
+    def save_secrets(self):
+        try:
+            with open("keys.json", "w") as file:
+                json.dump(self.json_dump(), file)
+        except:
+            print("Failed to save secrets")
+
+
+# Class that uses non-volatile storage as keystore
+class NVSKeyStore(KeyStore):
+
+    def __init__(self):
+        super(NVSKeyStore, self).__init__()
+        self.nvsdata = esp32.NVS("BLE")
+
+    # Load bonding keys from non-volatile storage.
+    def load_secrets(self):
+        data = bytearray()
+        num_bytes = 0
+
+        try:
+            num_bytes = self.nvsdata.get_blob("Keys", data)
+            data = bytearray(num_bytes)
+            self.nvsdata.get_blob("Keys", data)
+        except:
+            print("Failed to read NVS")
+
+        if num_bytes > 0:
+            s = str(data, 'utf-8')
+            try:
+                entries = json.loads(s)
+                self.add_json_entries(entries)
+            except:
+                print("Failed to load secrets")
+        else:
+            print("No secrets available")
+
+    # Save bonding keys to non-volatile storage.
+    def save_secrets(self):
+        try:
+            self.nvsdata.set_blob("Keys", json.dumps(self.json_dump()))
+            self.nvsdata.commit()
+        except:
+            print("Failed to save secrets")
+

hid_services.py

@@ -18,9 +18,9 @@
 from micropython import const
 import struct
 import bluetooth
-import json
-import binascii
 from bluetooth import UUID
+from hid_keystores import JSONKeyStore
+from hid_keystores import NVSKeyStore
 
 F_READ = bluetooth.FLAG_READ
 F_WRITE = bluetooth.FLAG_WRITE
@@ -180,7 +180,6 @@ def stop_advertising(self):
             self._ble.gap_advertise(0, adv_data=self._payload)
             print("Stopped advertising")
 
-
 # Class that represents a general HID device services.
 class HumanInterfaceDevice(object):
     # Define device states
@@ -205,9 +204,7 @@ def __init__(self, device_name="Generic HID Device"):
         self.key_size = 0                                                                                               # The encryption key size.
 
         self.passkey = 1234                                                                                             # The standard passkey for pairing. Only used when io capability allows so. Use the set_passkey(passkey) function to overwrite.
-        self.secrets = {}                                                                                               # The key store for bonding
-
-        self.load_secrets()                                                                                             # Call the function to load the known keys for bonding into the key store.
+        self.secrets = NVSKeyStore()                                                                                   # The key store for bonding
 
         # General characteristics.
         self.device_name = device_name                                                                                  # The device name.
@@ -350,36 +347,24 @@ def ble_irq(self, event, data):
                 print("Unknown passkey action")
         elif event == _IRQ_SET_SECRET:                                                                                  # Set secret for bonding.
             sec_type, key, value = data
-            key = (sec_type, bytes(key))
-            value = bytes(value) if value else None
             if value is None:                                                                                           # If value is empty, and
-                if key in self.secrets:                                                                                 # If key is known then
-                    del self.secrets[key]                                                                               # Forget key
-                    self.save_secrets()
-                    print("Removing secret:", key)
+                if self.secrets.has_secret(sec_type, key):                                                              # If key is known then
+                    self.secrets.remove_secret(sec_type, key)                                                           # Forget key
+                    self.secrets.save_secrets()
+                    print("Removing secret:", bytes(key))
                     return True
                 else:
-                    print("Secret not found:", key)
+                    print("Secret not found:", bytes(key))
                     return False
             else:
-                self.secrets[key] = value                                                                               # Remember key/value
-                self.save_secrets()
-                print("Saving secret:", key, value)
+                self.secrets.add_secret(sec_type, key, value)                                                           # Remember key/value
+                self.secrets.save_secrets()
+                print("Saving secret:", bytes(key), bytes(value))
             return True
         elif event == _IRQ_GET_SECRET:                                                                                  # Get secret for bonding
             sec_type, index, key = data
-            _key = (sec_type, bytes(key) if key else None)
-            value = None
-            if key is None:
-                i = 0
-                for (t, _k), _val in self.secrets.items():
-                    if t == sec_type:
-                        if i == index:
-                            value = _val
-                        i += 1
-            else:
-                value = self.secrets.get(_key, None)
-            print("Returning secret:", bytes(value) if value else None, "for", "key" if key else "index", _key if key else index, "with type", sec_type)
+            value = self.secrets.get_secret(sec_type, index, key)
+            print("Returning secret:", bytes(value) if value else None, "for", "key" if key else "index", bytes(key) if key else index, "with type", sec_type)
             return value
         else:
             print("Unhandled IRQ event:", event)
@@ -389,6 +374,8 @@ def ble_irq(self, event, data):
     # the overwritten function by using super(Subclass, self).start().
     def start(self):
         if self.device_state is HumanInterfaceDevice.DEVICE_STOPPED:
+            self.secrets.load_secrets()                                                                                 # Call the function to load the known keys for bonding into the key store.
+
             self._ble.irq(self.ble_irq)                                                                                 # Set interrupt request callback function.
             self._ble.active(1)                                                                                         # Turn on BLE radio.
 
@@ -464,28 +451,6 @@ def write_service_characteristics(self):
         for handle, (name, value) in self.characteristics.items():
             self._ble.gatts_write(handle, value)
 
-    # Load bonding keys from json file.
-    def load_secrets(self):
-        try:
-            with open("keys.json", "r") as file:
-                entries = json.load(file)
-                for sec_type, key, value in entries:
-                    self.secrets[sec_type, binascii.a2b_base64(key)] = binascii.a2b_base64(value)
-        except:
-            print("No secrets available")
-
-    # Save bonding keys to json file.
-    def save_secrets(self):
-        try:
-            with open("keys.json", "w") as file:
-                json_secrets = [
-                    (sec_type, binascii.b2a_base64(key, newline=False), binascii.b2a_base64(value, newline=False))
-                    for (sec_type, key), value in self.secrets.items()
-                ]
-                json.dump(json_secrets, file)
-        except:
-            print("Failed to save secrets")
-
     # Returns whether the device is not stopped.
     def is_running(self):
         return self.device_state is not HumanInterfaceDevice.DEVICE_STOPPED
@@ -607,6 +572,11 @@ def set_le_secure(self, le_secure=True):
     def set_io_capability(self, io_capability):
         self.io_capability = io_capability
 
+    # Set the keystore class to use.
+    # Must be called before calling Start().
+    def set_keystore(self, keystore):
+        self.secrets = keystore
+
     # Set callback function for pairing events.
     # Depending on the I/O capability used, the callback function should return either a
     # - boolean to accept or deny a connection, or a

package.json

@@ -1,7 +1,8 @@
 {
   "urls": [
-    ["hid_services.py", "github:Heerkog/MicroPythonBLEHID/hid_services.py"]
+    ["hid_services.py", "github:Heerkog/MicroPythonBLEHID/hid_services.py"],
+    ["hid_keystore.py", "github:Heerkog/MicroPythonBLEHID/hid_keystore.py"]
   ],
-  "version": "1.0.0",
+  "version": "2.1.0",
   "deps": []
 }

readme.md

@@ -91,6 +91,7 @@ The repository is structured as followed:
     * `mouse_example.py`
   * `tinypico/` directory containing TinyPICO specific examples. These are mostly personal projects.
 * `hid_services.py` the library.
+* `hid_keystores.py` different key stores to use with the library.
 * `LICENSE` the license.
 * `readme.md`
 
@@ -135,14 +136,15 @@ The library consists of five classes with the following functions:
   * `set_io_capability(io_capability)` (Set input/output capability of this device Determines the pairing procedure, e.g., accept connection/passkey entry/just works. Must be called before calling `start()`)
   * `set_passkey_callback(passkey_callback)` (Set callback function for pairing events. Callback function should return boolean to accept connection or passkey depending on I/O capability used)
   * `set_passkey(passkey)` (Set the passkey to use for pairing)
+  * `set_keystore(keystore)` (Sets the key store to use from `hid_keystores.py`. Default `JSONKeyStore`)
   * `set_battery_level(level)` (Sets the battery level internally)
   * `notify_battery_level()` (Notifies the central of the current battery level. Call after setting battery level)
   * `notify_hid_report()` (Function for subclasses to override)
 
 * `Joystick` (subclass of `HumanInterfaceDevice`, implements joystick service)
   * `__init__(name)` (Initialize the joystick)
   * `start()` (Starts the HID service using joystick characteristics. Calls `HumanInterfaceDevice.start()`)
-  * `write_service_characteristics(handles)` (Writes the joystick HID service characteristics.  Calls `HumanInterfaceDevice.write_service_characteristics(handles)`)
+  * `write_service_characteristics(handles)` (Writes the joystick HID service characteristics. Calls `HumanInterfaceDevice.write_service_characteristics(handles)`)
   * `notify_hid_report()` (Notifies the central of the internal HID joystick status)
   * `set_axes(x, y)` (Sets the joystick axes internally)
   * `set_buttons(b1, b2, b3, b4, b5, b6, b7, b8)` (Sets the joystick buttons internally)