| # Copyright 2021-2022 Google LLC |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # https://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| # ----------------------------------------------------------------------------- |
| # Imports |
| # ----------------------------------------------------------------------------- |
| import asyncio |
| import os |
| import logging |
| import click |
| import aioconsole |
| from colors import color |
| |
| from bumble.device import Device, Peer |
| from bumble.transport import open_transport_or_link |
| from bumble.smp import PairingDelegate, PairingConfig |
| from bumble.smp import error_name as smp_error_name |
| from bumble.keys import JsonKeyStore |
| from bumble.core import ProtocolError |
| from bumble.gatt import ( |
| GATT_DEVICE_NAME_CHARACTERISTIC, |
| GATT_GENERIC_ACCESS_SERVICE, |
| Service, |
| Characteristic, |
| CharacteristicValue, |
| ) |
| from bumble.att import ( |
| ATT_Error, |
| ATT_INSUFFICIENT_AUTHENTICATION_ERROR, |
| ATT_INSUFFICIENT_ENCRYPTION_ERROR, |
| ) |
| |
| |
| # ----------------------------------------------------------------------------- |
| class Delegate(PairingDelegate): |
| def __init__(self, mode, connection, capability_string, prompt): |
| super().__init__( |
| { |
| 'keyboard': PairingDelegate.KEYBOARD_INPUT_ONLY, |
| 'display': PairingDelegate.DISPLAY_OUTPUT_ONLY, |
| 'display+keyboard': PairingDelegate.DISPLAY_OUTPUT_AND_KEYBOARD_INPUT, |
| 'display+yes/no': PairingDelegate.DISPLAY_OUTPUT_AND_YES_NO_INPUT, |
| 'none': PairingDelegate.NO_OUTPUT_NO_INPUT, |
| }[capability_string.lower()] |
| ) |
| |
| self.mode = mode |
| self.peer = Peer(connection) |
| self.peer_name = None |
| self.prompt = prompt |
| |
| async def update_peer_name(self): |
| if self.peer_name is not None: |
| # We already asked the peer |
| return |
| |
| # Try to get the peer's name |
| if self.peer: |
| peer_name = await get_peer_name(self.peer, self.mode) |
| self.peer_name = f'{peer_name or ""} [{self.peer.connection.peer_address}]' |
| else: |
| self.peer_name = '[?]' |
| |
| async def accept(self): |
| if self.prompt: |
| await self.update_peer_name() |
| |
| # Wait a bit to allow some of the log lines to print before we prompt |
| await asyncio.sleep(1) |
| |
| # Prompt for acceptance |
| print(color('###-----------------------------------', 'yellow')) |
| print(color(f'### Pairing request from {self.peer_name}', 'yellow')) |
| print(color('###-----------------------------------', 'yellow')) |
| while True: |
| response = await aioconsole.ainput(color('>>> Accept? ', 'yellow')) |
| response = response.lower().strip() |
| |
| if response == 'yes': |
| return True |
| |
| if response == 'no': |
| return False |
| |
| # Accept silently |
| return True |
| |
| async def compare_numbers(self, number, digits=6): |
| await self.update_peer_name() |
| |
| # Wait a bit to allow some of the log lines to print before we prompt |
| await asyncio.sleep(1) |
| |
| # Prompt for a numeric comparison |
| print(color('###-----------------------------------', 'yellow')) |
| print(color(f'### Pairing with {self.peer_name}', 'yellow')) |
| print(color('###-----------------------------------', 'yellow')) |
| while True: |
| response = await aioconsole.ainput( |
| color( |
| f'>>> Does the other device display {number:0{digits}}? ', 'yellow' |
| ) |
| ) |
| response = response.lower().strip() |
| |
| if response == 'yes': |
| return True |
| |
| if response == 'no': |
| return False |
| |
| async def get_number(self): |
| await self.update_peer_name() |
| |
| # Wait a bit to allow some of the log lines to print before we prompt |
| await asyncio.sleep(1) |
| |
| # Prompt for a PIN |
| while True: |
| try: |
| print(color('###-----------------------------------', 'yellow')) |
| print(color(f'### Pairing with {self.peer_name}', 'yellow')) |
| print(color('###-----------------------------------', 'yellow')) |
| return int(await aioconsole.ainput(color('>>> Enter PIN: ', 'yellow'))) |
| except ValueError: |
| pass |
| |
| async def display_number(self, number, digits=6): |
| await self.update_peer_name() |
| |
| # Wait a bit to allow some of the log lines to print before we prompt |
| await asyncio.sleep(1) |
| |
| # Display a PIN code |
| print(color('###-----------------------------------', 'yellow')) |
| print(color(f'### Pairing with {self.peer_name}', 'yellow')) |
| print(color(f'### PIN: {number:0{digits}}', 'yellow')) |
| print(color('###-----------------------------------', 'yellow')) |
| |
| |
| # ----------------------------------------------------------------------------- |
| async def get_peer_name(peer, mode): |
| if mode == 'classic': |
| return await peer.request_name() |
| |
| # Try to get the peer name from GATT |
| services = await peer.discover_service(GATT_GENERIC_ACCESS_SERVICE) |
| if not services: |
| return None |
| |
| values = await peer.read_characteristics_by_uuid( |
| GATT_DEVICE_NAME_CHARACTERISTIC, services[0] |
| ) |
| if values: |
| return values[0].decode('utf-8') |
| |
| return None |
| |
| |
| # ----------------------------------------------------------------------------- |
| AUTHENTICATION_ERROR_RETURNED = [False, False] |
| |
| |
| def read_with_error(connection): |
| if not connection.is_encrypted: |
| raise ATT_Error(ATT_INSUFFICIENT_ENCRYPTION_ERROR) |
| |
| if AUTHENTICATION_ERROR_RETURNED[0]: |
| return bytes([1]) |
| |
| AUTHENTICATION_ERROR_RETURNED[0] = True |
| raise ATT_Error(ATT_INSUFFICIENT_AUTHENTICATION_ERROR) |
| |
| |
| def write_with_error(connection, _value): |
| if not connection.is_encrypted: |
| raise ATT_Error(ATT_INSUFFICIENT_ENCRYPTION_ERROR) |
| |
| if not AUTHENTICATION_ERROR_RETURNED[1]: |
| AUTHENTICATION_ERROR_RETURNED[1] = True |
| raise ATT_Error(ATT_INSUFFICIENT_AUTHENTICATION_ERROR) |
| |
| |
| # ----------------------------------------------------------------------------- |
| def on_connection(connection, request): |
| print(color(f'<<< Connection: {connection}', 'green')) |
| |
| # Listen for pairing events |
| connection.on('pairing_start', on_pairing_start) |
| connection.on('pairing', on_pairing) |
| connection.on('pairing_failure', on_pairing_failure) |
| |
| # Listen for encryption changes |
| connection.on( |
| 'connection_encryption_change', |
| lambda: on_connection_encryption_change(connection), |
| ) |
| |
| # Request pairing if needed |
| if request: |
| print(color('>>> Requesting pairing', 'green')) |
| connection.request_pairing() |
| |
| |
| # ----------------------------------------------------------------------------- |
| def on_connection_encryption_change(connection): |
| print(color('@@@-----------------------------------', 'blue')) |
| print( |
| color( |
| f'@@@ Connection is {"" if connection.is_encrypted else "not"}encrypted', |
| 'blue', |
| ) |
| ) |
| print(color('@@@-----------------------------------', 'blue')) |
| |
| |
| # ----------------------------------------------------------------------------- |
| def on_pairing_start(): |
| print(color('***-----------------------------------', 'magenta')) |
| print(color('*** Pairing starting', 'magenta')) |
| print(color('***-----------------------------------', 'magenta')) |
| |
| |
| # ----------------------------------------------------------------------------- |
| def on_pairing(keys): |
| print(color('***-----------------------------------', 'cyan')) |
| print(color('*** Paired!', 'cyan')) |
| keys.print(prefix=color('*** ', 'cyan')) |
| print(color('***-----------------------------------', 'cyan')) |
| |
| |
| # ----------------------------------------------------------------------------- |
| def on_pairing_failure(reason): |
| print(color('***-----------------------------------', 'red')) |
| print(color(f'*** Pairing failed: {smp_error_name(reason)}', 'red')) |
| print(color('***-----------------------------------', 'red')) |
| |
| |
| # ----------------------------------------------------------------------------- |
| async def pair( |
| mode, |
| sc, |
| mitm, |
| bond, |
| io, |
| prompt, |
| request, |
| print_keys, |
| keystore_file, |
| device_config, |
| hci_transport, |
| address_or_name, |
| ): |
| print('<<< connecting to HCI...') |
| async with await open_transport_or_link(hci_transport) as (hci_source, hci_sink): |
| print('<<< connected') |
| |
| # Create a device to manage the host |
| device = Device.from_config_file_with_hci(device_config, hci_source, hci_sink) |
| |
| # Set a custom keystore if specified on the command line |
| if keystore_file: |
| device.keystore = JsonKeyStore(namespace=None, filename=keystore_file) |
| |
| # Print the existing keys before pairing |
| if print_keys and device.keystore: |
| print(color('@@@-----------------------------------', 'blue')) |
| print(color('@@@ Pairing Keys:', 'blue')) |
| await device.keystore.print(prefix=color('@@@ ', 'blue')) |
| print(color('@@@-----------------------------------', 'blue')) |
| |
| # Expose a GATT characteristic that can be used to trigger pairing by |
| # responding with an authentication error when read |
| if mode == 'le': |
| device.add_service( |
| Service( |
| '50DB505C-8AC4-4738-8448-3B1D9CC09CC5', |
| [ |
| Characteristic( |
| '552957FB-CF1F-4A31-9535-E78847E1A714', |
| Characteristic.READ | Characteristic.WRITE, |
| Characteristic.READABLE | Characteristic.WRITEABLE, |
| CharacteristicValue( |
| read=read_with_error, write=write_with_error |
| ), |
| ) |
| ], |
| ) |
| ) |
| |
| # Select LE or Classic |
| if mode == 'classic': |
| device.classic_enabled = True |
| device.le_enabled = False |
| |
| # Get things going |
| await device.power_on() |
| |
| # Set up a pairing config factory |
| device.pairing_config_factory = lambda connection: PairingConfig( |
| sc, mitm, bond, Delegate(mode, connection, io, prompt) |
| ) |
| |
| # Connect to a peer or wait for a connection |
| device.on('connection', lambda connection: on_connection(connection, request)) |
| if address_or_name is not None: |
| print(color(f'=== Connecting to {address_or_name}...', 'green')) |
| connection = await device.connect(address_or_name) |
| |
| if not request: |
| try: |
| if mode == 'le': |
| await connection.pair() |
| else: |
| await connection.authenticate() |
| return |
| except ProtocolError as error: |
| print(color(f'Pairing failed: {error}', 'red')) |
| return |
| else: |
| # Advertise so that peers can find us and connect |
| await device.start_advertising(auto_restart=True) |
| |
| await hci_source.wait_for_termination() |
| |
| |
| # ----------------------------------------------------------------------------- |
| @click.command() |
| @click.option( |
| '--mode', type=click.Choice(['le', 'classic']), default='le', show_default=True |
| ) |
| @click.option( |
| '--sc', |
| type=bool, |
| default=True, |
| help='Use the Secure Connections protocol', |
| show_default=True, |
| ) |
| @click.option( |
| '--mitm', type=bool, default=True, help='Request MITM protection', show_default=True |
| ) |
| @click.option( |
| '--bond', type=bool, default=True, help='Enable bonding', show_default=True |
| ) |
| @click.option( |
| '--io', |
| type=click.Choice( |
| ['keyboard', 'display', 'display+keyboard', 'display+yes/no', 'none'] |
| ), |
| default='display+keyboard', |
| show_default=True, |
| ) |
| @click.option('--prompt', is_flag=True, help='Prompt to accept/reject pairing request') |
| @click.option( |
| '--request', is_flag=True, help='Request that the connecting peer initiate pairing' |
| ) |
| @click.option('--print-keys', is_flag=True, help='Print the bond keys before pairing') |
| @click.option('--keystore-file', help='File in which to store the pairing keys') |
| @click.argument('device-config') |
| @click.argument('hci_transport') |
| @click.argument('address-or-name', required=False) |
| def main( |
| mode, |
| sc, |
| mitm, |
| bond, |
| io, |
| prompt, |
| request, |
| print_keys, |
| keystore_file, |
| device_config, |
| hci_transport, |
| address_or_name, |
| ): |
| logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper()) |
| asyncio.run( |
| pair( |
| mode, |
| sc, |
| mitm, |
| bond, |
| io, |
| prompt, |
| request, |
| print_keys, |
| keystore_file, |
| device_config, |
| hci_transport, |
| address_or_name, |
| ) |
| ) |
| |
| |
| # ----------------------------------------------------------------------------- |
| if __name__ == '__main__': |
| main() |
