Merge pull request #158 from benquike/main

Fix HCI_PIN_Code_Reply_Command

bumble/device.py

@@ -2843,19 +2843,22 @@ class Device(CompositeEventEmitter):
 
             async def get_pin_code():
                 pin_code = await connection.abort_on(
-                    'disconnection', pairing_config.delegate.get_number()
+                    'disconnection', pairing_config.delegate.get_string(16)
                 )
 
                 if pin_code is not None:
-                    pin_code = bytes(str(pin_code).zfill(6))
+                    pin_code = bytes(pin_code, encoding='utf-8')
+                    pin_code_len = len(pin_code)
+                    assert 0 < pin_code_len <= 16, "pin_code should be 1-16 bytes"
                     await self.host.send_command(
                         HCI_PIN_Code_Request_Reply_Command(
                             bd_addr=connection.peer_address,
-                            pin_code_length=len(pin_code),
+                            pin_code_length=pin_code_len,
                             pin_code=pin_code,
                         )
                     )
                 else:
+                    logger.debug("delegate.get_string() returned None")
                     await self.host.send_command(
                         HCI_PIN_Code_Request_Negative_Reply_Command(
                             bd_addr=connection.peer_address

bumble/hci.py

@@ -2102,7 +2102,7 @@ class HCI_Link_Key_Request_Negative_Reply_Command(HCI_Command):
     fields=[
         ('bd_addr', Address.parse_address),
         ('pin_code_length', 1),
-        ('pin_code', '*'),
+        ('pin_code', 16),
     ],
     return_parameters_fields=[
         ('status', STATUS_SPEC),

bumble/smp.py

@@ -525,6 +525,12 @@ class PairingDelegate:
     async def get_number(self) -> int:
         return 0
 
+    async def get_string(self, max_length) -> Optional[str]:
+        '''
+        Returns a string whose utf-8 encoding is up to max_length bytes.
+        '''
+        return None
+
     # pylint: disable-next=unused-argument
     async def display_number(self, number: int, digits: int) -> None:
         pass

tests/hci_test.py

@@ -215,12 +215,18 @@ def test_HCI_Command():
 
 # -----------------------------------------------------------------------------
 def test_HCI_PIN_Code_Request_Reply_Command():
+    pin_code = b'1234'
+    pin_code_length = len(pin_code)
+    # here to make the test pass, we need to
+    # pad pin_code, as HCI_Object.format_fields
+    # does not do it for us
+    padded_pin_code = pin_code + bytes(16 - pin_code_length)
     command = HCI_PIN_Code_Request_Reply_Command(
         bd_addr=Address(
             '00:11:22:33:44:55', address_type=Address.PUBLIC_DEVICE_ADDRESS
         ),
-        pin_code_length=4,
+        pin_code_length=pin_code_length,
-        pin_code=b'1234',
+        pin_code=padded_pin_code,
     )
     basic_check(command)