Merge pull request #167 from dzervas/ble_multimple_conns

BLE Multiple Connections

kmk/ble.py

@@ -1,69 +1,101 @@
 from adafruit_ble import BLERadio
 from adafruit_ble.advertising.standard import ProvideServicesAdvertisement
 from adafruit_ble.services.standard.hid import HIDService
-from kmk.hid import (
-    HID_REPORT_SIZES,
-    AbstractHID,
-    HIDReportTypes,
-    HIDUsage,
-    HIDUsagePage,
-)
+from kmk.hid import HID_REPORT_SIZES, AbstractHID
 
 BLE_APPEARANCE_HID_KEYBOARD = 961
+# Hardcoded in CPy
+MAX_CONNECTIONS = 2
 
 
 class BLEHID(AbstractHID):
     def post_init(self, ble_name='KMK Keyboard', **kwargs):
-        self.devices = {}
+        self.conn_id = -1
 
-        hid = HIDService()
+        self.ble = BLERadio()
+        self.ble.name = ble_name
+        self.hid = HIDService()
+        self.hid.protocol_mode = 0  # Boot protocol
 
-        advertisement = ProvideServicesAdvertisement(hid)
-        advertisement.appearance = BLE_APPEARANCE_HID_KEYBOARD
+        # Security-wise this is not right. While you're away someone turns
+        # on your keyboard and they can pair with it nice and clean and then
+        # listen to keystrokes.
+        # On the other hand we don't have LESC so it's like shouting your
+        # keystrokes in the air
+        if not self.ble.connected or not self.hid.devices:
+            self.start_advertising()
 
-        ble = BLERadio()
-        ble.name = ble_name
-        # ble.tx_power = 2
+        self.conn_id = 0
 
-        if not ble.connected:
-            ble.start_advertising(advertisement)
-            while not ble.connected:
-                pass
+    @property
+    def devices(self):
+        """Search through the provided list of devices to find the ones with the
+        send_report attribute."""
+        if not self.ble.connected:
+            return []
 
-        for device in hid.devices:
-            us = device.usage
-            up = device.usage_page
+        result = []
+        # Security issue:
+        # This introduces a race condition. Let's say you have 2 active
+        # connections: Alice and Bob - Alice is connection 1 and Bob 2.
+        # Now Chuck who has already paired with the device in the past
+        # (this assumption is needed only in the case of LESC)
+        # wants to gather the keystrokes you send to Alice. You have
+        # selected right now to talk to Alice (1) and you're typing a secret.
+        # If Chuck kicks Alice off and is quick enough to connect to you,
+        # which means quicker than the running interval of this function,
+        # he'll be earlier in the `self.hid.devices` so will take over the
+        # selected 1 position in the resulted array.
+        # If no LESC is in place, Chuck can sniff the keystrokes anyway
+        for device in self.hid.devices:
+            if hasattr(device, "send_report"):
+                result.append(device)
 
-            if up == HIDUsagePage.CONSUMER and us == HIDUsage.CONSUMER:
-                self.devices[HIDReportTypes.CONSUMER] = device
-                continue
+        return result
 
-            if up == HIDUsagePage.KEYBOARD and us == HIDUsage.KEYBOARD:
-                self.devices[HIDReportTypes.KEYBOARD] = device
-                continue
+    def _check_connection(self):
+        devices = self.devices
+        if not devices:
+            return False
 
-            if up == HIDUsagePage.MOUSE and us == HIDUsage.MOUSE:
-                self.devices[HIDReportTypes.MOUSE] = device
-                continue
+        if self.conn_id >= len(devices):
+            self.conn_id = len(devices) - 1
 
-            if up == HIDUsagePage.SYSCONTROL and us == HIDUsage.SYSCONTROL:
-                self.devices[HIDReportTypes.SYSCONTROL] = device
-                continue
+        if self.conn_id < 0:
+            return False
+
+        if not devices[self.conn_id]:
+            return False
+
+        return True
 
     def hid_send(self, evt):
-        # int, can be looked up in HIDReportTypes
-        reporting_device_const = self.report_device[0]
+        if not self._check_connection():
+            return
 
-        report_size = HID_REPORT_SIZES[reporting_device_const]
+        device = self.devices[self.conn_id]
 
-        while len(evt) < report_size + 1:
+        while len(evt) < len(device._characteristic.value) + 1:
             evt.append(0)
 
-        return self.devices[reporting_device_const].send_report(
-            evt[1 : report_size + 1]
-        )
+        return device.send_report(evt[1:])
 
     def clear_bonds(self):
         import _bleio
 
         _bleio.adapter.erase_bonding()
+
+    def next_connection(self):
+        self.conn_id = (self.conn_id + 1) % len(self.devices)
+
+    def previous_connection(self):
+        self.conn_id = (self.conn_id - 1) % len(self.devices)
+
+    def start_advertising(self):
+        advertisement = ProvideServicesAdvertisement(self.hid)
+        advertisement.appearance = BLE_APPEARANCE_HID_KEYBOARD
+
+        self.ble.start_advertising(advertisement)
+
+    def stop_advertising(self):
+        self.ble.stop_advertising()

kmk/handlers/stock.py

@@ -259,3 +259,13 @@ def led_mode_breathe(key, state, *args, **kwargs):
 def bt_clear_bonds(key, state, *args, **kwargs):
     state.config._hid_helper_inst.clear_bonds()
     return state
+
+
+def bt_next_conn(key, state, *args, **kwargs):
+    state.config._hid_helper_inst.next_connection()
+    return state
+
+
+def bt_prev_conn(key, state, *args, **kwargs):
+    state.config._hid_helper_inst.previous_connection()
+    return state

kmk/keys.py

@@ -638,6 +638,8 @@ make_key(names=('LED_AND',), on_press=handlers.led_and)
 make_key(names=('LED_MODE_PLAIN', 'LED_M_P'), on_press=handlers.led_mode_static)
 make_key(names=('LED_MODE_BREATHE', 'LED_M_B'), on_press=handlers.led_mode_breathe)
 make_key(names=('BT_CLEAR_BONDS', 'BT_CLR'), on_press=handlers.bt_clear_bonds)
+make_key(names=('BT_NEXT_CONN', 'BT_NXT'), on_press=handlers.bt_next_conn)
+make_key(names=('BT_PREV_CONN', 'BT_PRV'), on_press=handlers.bt_prev_conn)
 
 
 make_key(