diff --git a/.github/workflows/python-avatar.yml b/.github/workflows/python-avatar.yml deleted file mode 100644 index 46b5bc3ab..000000000 --- a/.github/workflows/python-avatar.yml +++ /dev/null @@ -1,52 +0,0 @@ -name: Python Avatar - -on: - push: - branches: [ main ] - pull_request: - branches: [ main ] - workflow_dispatch: - branches: [main] - -permissions: - contents: read - -jobs: - test: - name: Avatar [${{ matrix.shard }}] - runs-on: ubuntu-latest - strategy: - matrix: - shard: [ - 1/24, 2/24, 3/24, 4/24, - 5/24, 6/24, 7/24, 8/24, - 9/24, 10/24, 11/24, 12/24, - 13/24, 14/24, 15/24, 16/24, - 17/24, 18/24, 19/24, 20/24, - 21/24, 22/24, 23/24, 24/24, - ] - steps: - - uses: actions/checkout@v6 - - name: Set Up Python 3.11 - uses: actions/setup-python@v6 - with: - python-version: 3.11 - - name: Install - run: | - python -m pip install --upgrade pip - python -m pip install .[all,avatar] - - name: Rootcanal - run: nohup python -m rootcanal > rootcanal.log & - - name: Test - run: | - avatar --list | grep -Ev '^=' > test-names.txt - timeout 5m avatar --test-beds bumble.bumbles --tests $(split test-names.txt -n l/${{ matrix.shard }}) - - name: Rootcanal Logs - if: always() - run: cat rootcanal.log - - name: Upload Mobly logs - if: always() - uses: actions/upload-artifact@v6 - with: - name: mobly-logs-${{ strategy.job-index }} - path: /tmp/logs/mobly/bumble.bumbles/ diff --git a/apps/pandora_server.py b/apps/pandora_server.py deleted file mode 100644 index eed72d3f1..000000000 --- a/apps/pandora_server.py +++ /dev/null @@ -1,52 +0,0 @@ -import asyncio -import json -import logging -from typing import Any - -import click - -from bumble.pandora import Config, PandoraDevice, serve - -BUMBLE_SERVER_GRPC_PORT = 7999 -ROOTCANAL_PORT_CUTTLEFISH = 7300 - - -@click.command() -@click.option('--grpc-port', help='gRPC port to serve', default=BUMBLE_SERVER_GRPC_PORT) -@click.option( - '--rootcanal-port', help='Rootcanal TCP port', default=ROOTCANAL_PORT_CUTTLEFISH -) -@click.option( - '--transport', - help='HCI transport', - default='tcp-client:127.0.0.1:', -) -@click.option( - '--config', - help='Bumble json configuration file', -) -def main(grpc_port: int, rootcanal_port: int, transport: str, config: str) -> None: - if '' in transport: - transport = transport.replace('', str(rootcanal_port)) - - bumble_config = retrieve_config(config) - bumble_config.setdefault('transport', transport) - device = PandoraDevice(bumble_config) - - server_config = Config() - server_config.load_from_dict(bumble_config.get('server', {})) - - logging.basicConfig(level=logging.DEBUG) - asyncio.run(serve(device, config=server_config, port=grpc_port)) - - -def retrieve_config(config: str) -> dict[str, Any]: - if not config: - return {} - - with open(config) as f: - return json.load(f) - - -if __name__ == '__main__': - main() # pylint: disable=no-value-for-parameter diff --git a/bumble/pandora/__init__.py b/bumble/pandora/__init__.py deleted file mode 100644 index 29f382d5c..000000000 --- a/bumble/pandora/__init__.py +++ /dev/null @@ -1,109 +0,0 @@ -# Copyright 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. - -""" -Bumble Pandora server. -This module implement the Pandora Bluetooth test APIs for the Bumble stack. -""" - -__version__ = "0.0.1" - -from collections.abc import Callable - -import grpc -import grpc.aio -from pandora.host_grpc_aio import add_HostServicer_to_server -from pandora.l2cap_grpc_aio import add_L2CAPServicer_to_server -from pandora.security_grpc_aio import ( - add_SecurityServicer_to_server, - add_SecurityStorageServicer_to_server, -) - -from bumble.pandora.config import Config -from bumble.pandora.device import PandoraDevice -from bumble.pandora.host import HostService -from bumble.pandora.l2cap import L2CAPService -from bumble.pandora.security import SecurityService, SecurityStorageService - -# public symbols -__all__ = [ - 'register_servicer_hook', - 'serve', - 'Config', - 'PandoraDevice', -] - - -# Add servicers hooks. -_SERVICERS_HOOKS: list[Callable[[PandoraDevice, Config, grpc.aio.Server], None]] = [] - - -def register_servicer_hook( - hook: Callable[[PandoraDevice, Config, grpc.aio.Server], None], -) -> None: - _SERVICERS_HOOKS.append(hook) - - -async def serve( - bumble: PandoraDevice, - config: Config = Config(), - grpc_server: grpc.aio.Server | None = None, - port: int = 0, -) -> None: - # initialize a gRPC server if not provided. - server = grpc_server if grpc_server is not None else grpc.aio.server() - port = server.add_insecure_port(f'localhost:{port}') - - try: - while True: - # load server config from dict. - config.load_from_dict(bumble.config.get('server', {})) - - # add Pandora services to the gRPC server. - add_HostServicer_to_server( - HostService(server, bumble.device, config), server - ) - add_SecurityServicer_to_server( - SecurityService(bumble.device, config), server - ) - add_SecurityStorageServicer_to_server( - SecurityStorageService(bumble.device, config), server - ) - add_L2CAPServicer_to_server(L2CAPService(bumble.device, config), server) - - # call hooks if any. - for hook in _SERVICERS_HOOKS: - hook(bumble, config, server) - - # open device. - await bumble.open() - try: - # Pandora require classic devices to be discoverable & connectable. - if bumble.device.classic_enabled: - await bumble.device.set_discoverable(True) - await bumble.device.set_connectable(True) - - # start & serve gRPC server. - await server.start() - await server.wait_for_termination() - finally: - # close device. - await bumble.close() - - # re-initialize the gRPC server. - server = grpc.aio.server() - server.add_insecure_port(f'localhost:{port}') - finally: - # stop server. - await server.stop(None) diff --git a/bumble/pandora/config.py b/bumble/pandora/config.py deleted file mode 100644 index 74a9fb45f..000000000 --- a/bumble/pandora/config.py +++ /dev/null @@ -1,58 +0,0 @@ -# Copyright 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. - -from __future__ import annotations - -from dataclasses import dataclass -from typing import Any - -from bumble.pairing import PairingConfig, PairingDelegate - - -@dataclass -class Config: - io_capability: PairingDelegate.IoCapability = PairingDelegate.NO_OUTPUT_NO_INPUT - identity_address_type: PairingConfig.AddressType = PairingConfig.AddressType.RANDOM - pairing_sc_enable: bool = True - pairing_mitm_enable: bool = True - pairing_bonding_enable: bool = True - smp_local_initiator_key_distribution: PairingDelegate.KeyDistribution = ( - PairingDelegate.DEFAULT_KEY_DISTRIBUTION - ) - smp_local_responder_key_distribution: PairingDelegate.KeyDistribution = ( - PairingDelegate.DEFAULT_KEY_DISTRIBUTION - ) - - def load_from_dict(self, config: dict[str, Any]) -> None: - io_capability_name: str = config.get( - 'io_capability', 'no_output_no_input' - ).upper() - self.io_capability = getattr(PairingDelegate, io_capability_name) - identity_address_type_name: str = config.get( - 'identity_address_type', 'random' - ).upper() - self.identity_address_type = getattr( - PairingConfig.AddressType, identity_address_type_name - ) - self.pairing_sc_enable = config.get('pairing_sc_enable', True) - self.pairing_mitm_enable = config.get('pairing_mitm_enable', True) - self.pairing_bonding_enable = config.get('pairing_bonding_enable', True) - self.smp_local_initiator_key_distribution = config.get( - 'smp_local_initiator_key_distribution', - PairingDelegate.DEFAULT_KEY_DISTRIBUTION, - ) - self.smp_local_responder_key_distribution = config.get( - 'smp_local_responder_key_distribution', - PairingDelegate.DEFAULT_KEY_DISTRIBUTION, - ) diff --git a/bumble/pandora/device.py b/bumble/pandora/device.py deleted file mode 100644 index 571b64949..000000000 --- a/bumble/pandora/device.py +++ /dev/null @@ -1,167 +0,0 @@ -# Copyright 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. - -"""Generic & dependency free Bumble (reference) device.""" - -from __future__ import annotations - -from typing import Any - -from bumble import transport -from bumble.core import ( - BT_GENERIC_AUDIO_SERVICE, - BT_HANDSFREE_SERVICE, - BT_L2CAP_PROTOCOL_ID, - BT_RFCOMM_PROTOCOL_ID, -) -from bumble.device import Device, DeviceConfiguration -from bumble.host import Host -from bumble.sdp import ( - SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID, - SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, - SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID, - SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID, - DataElement, - ServiceAttribute, -) - -# Default rootcanal HCI TCP address -ROOTCANAL_HCI_ADDRESS = "localhost:6402" - - -class PandoraDevice: - """ - Small wrapper around a Bumble device and it's HCI transport. - Notes: - - The Bumble device is idle by default. - - Repetitive calls to `open`/`close` will result on new Bumble device instances. - """ - - # Bumble device instance & configuration. - device: Device - config: dict[str, Any] - - # HCI transport name & instance. - _hci_name: str - _hci: transport.Transport | None # type: ignore[name-defined] - - def __init__(self, config: dict[str, Any]) -> None: - self.config = config - self.device = _make_device(config) - self._hci_name = config.get( - 'transport', f"tcp-client:{config.get('tcp', ROOTCANAL_HCI_ADDRESS)}" - ) - self._hci = None - - @property - def idle(self) -> bool: - return self._hci is None - - async def open(self) -> None: - if self._hci is not None: - return - - # open HCI transport & set device host. - self._hci = await transport.open_transport(self._hci_name) - self.device.host = Host( - controller_source=self._hci.source, controller_sink=self._hci.sink - ) # type: ignore[no-untyped-call] - - # power-on. - await self.device.power_on() - - async def close(self) -> None: - if self._hci is None: - return - - # flush & re-initialize device. - await self.device.host.flush() - self.device.host = None # type: ignore[assignment] - self.device = _make_device(self.config) - - # close HCI transport. - await self._hci.close() - self._hci = None - - async def reset(self) -> None: - await self.close() - await self.open() - - def info(self) -> dict[str, str] | None: - return { - 'public_bd_address': str(self.device.public_address), - 'random_address': str(self.device.random_address), - } - - -def _make_device(config: dict[str, Any]) -> Device: - """Initialize an idle Bumble device instance.""" - - # initialize bumble device. - device_config = DeviceConfiguration() - device_config.load_from_dict(config) - device = Device(config=device_config, host=None) - - # Add fake a2dp service to avoid Android disconnect - device.sdp_service_records = _make_sdp_records(1) - - return device - - -# TODO(b/267540823): remove when Pandora A2dp is supported -def _make_sdp_records(rfcomm_channel: int) -> dict[int, list[ServiceAttribute]]: - return { - 0x00010001: [ - ServiceAttribute( - SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID, - DataElement.unsigned_integer_32(0x00010001), - ), - ServiceAttribute( - SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID, - DataElement.sequence( - [ - DataElement.uuid(BT_HANDSFREE_SERVICE), - DataElement.uuid(BT_GENERIC_AUDIO_SERVICE), - ] - ), - ), - ServiceAttribute( - SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, - DataElement.sequence( - [ - DataElement.sequence([DataElement.uuid(BT_L2CAP_PROTOCOL_ID)]), - DataElement.sequence( - [ - DataElement.uuid(BT_RFCOMM_PROTOCOL_ID), - DataElement.unsigned_integer_8(rfcomm_channel), - ] - ), - ] - ), - ), - ServiceAttribute( - SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID, - DataElement.sequence( - [ - DataElement.sequence( - [ - DataElement.uuid(BT_HANDSFREE_SERVICE), - DataElement.unsigned_integer_16(0x0105), - ] - ) - ] - ), - ), - ] - } diff --git a/bumble/pandora/host.py b/bumble/pandora/host.py deleted file mode 100644 index ca686a224..000000000 --- a/bumble/pandora/host.py +++ /dev/null @@ -1,1017 +0,0 @@ -# Copyright 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. - -from __future__ import annotations - -import asyncio -import logging -import struct -from collections.abc import AsyncGenerator -from typing import cast - -import grpc -import grpc.aio -from google.protobuf import ( - any_pb2, # pytype: disable=pyi-error - empty_pb2, # pytype: disable=pyi-error -) -from pandora import host_pb2 -from pandora.host_grpc_aio import HostServicer -from pandora.host_pb2 import ( - DISCOVERABLE_GENERAL, - DISCOVERABLE_LIMITED, - NOT_CONNECTABLE, - NOT_DISCOVERABLE, - PRIMARY_1M, - PRIMARY_CODED, - SECONDARY_1M, - SECONDARY_2M, - SECONDARY_CODED, - SECONDARY_NONE, - AdvertiseRequest, - AdvertiseResponse, - Connection, - ConnectLERequest, - ConnectLEResponse, - ConnectRequest, - ConnectResponse, - DataTypes, - DisconnectRequest, - InquiryResponse, - PrimaryPhy, - ReadLocalAddressResponse, - ScanningResponse, - ScanRequest, - SecondaryPhy, - SetConnectabilityModeRequest, - SetDiscoverabilityModeRequest, - WaitConnectionRequest, - WaitConnectionResponse, - WaitDisconnectionRequest, -) - -import bumble.device -import bumble.utils -from bumble.core import ( - UUID, - AdvertisingData, - Appearance, - ConnectionError, - PhysicalTransport, -) -from bumble.device import ( - DEVICE_DEFAULT_SCAN_INTERVAL, - DEVICE_DEFAULT_SCAN_WINDOW, - Advertisement, - AdvertisingEventProperties, - AdvertisingParameters, - AdvertisingType, - Device, -) -from bumble.gatt import Service -from bumble.hci import ( - HCI_CONNECTION_ALREADY_EXISTS_ERROR, - HCI_PAGE_TIMEOUT_ERROR, - HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR, - Address, - OwnAddressType, - Phy, - Role, -) -from bumble.pandora import utils -from bumble.pandora.config import Config - -PRIMARY_PHY_MAP: dict[int, PrimaryPhy] = { - # Default value reported by Bumble for legacy Advertising reports. - # FIXME(uael): `None` might be a better value, but Bumble need to change accordingly. - 0: PRIMARY_1M, - 1: PRIMARY_1M, - 3: PRIMARY_CODED, -} - -SECONDARY_PHY_MAP: dict[int, SecondaryPhy] = { - 0: SECONDARY_NONE, - 1: SECONDARY_1M, - 2: SECONDARY_2M, - 3: SECONDARY_CODED, -} - -PRIMARY_PHY_TO_BUMBLE_PHY_MAP: dict[PrimaryPhy, Phy] = { - PRIMARY_1M: Phy.LE_1M, - PRIMARY_CODED: Phy.LE_CODED, -} - -SECONDARY_PHY_TO_BUMBLE_PHY_MAP: dict[SecondaryPhy, Phy] = { - SECONDARY_NONE: Phy.LE_1M, - SECONDARY_1M: Phy.LE_1M, - SECONDARY_2M: Phy.LE_2M, - SECONDARY_CODED: Phy.LE_CODED, -} - -OWN_ADDRESS_MAP: dict[host_pb2.OwnAddressType, OwnAddressType] = { - host_pb2.PUBLIC: OwnAddressType.PUBLIC, - host_pb2.RANDOM: OwnAddressType.RANDOM, - host_pb2.RESOLVABLE_OR_PUBLIC: OwnAddressType.RESOLVABLE_OR_PUBLIC, - host_pb2.RESOLVABLE_OR_RANDOM: OwnAddressType.RESOLVABLE_OR_RANDOM, -} - - -class HostService(HostServicer): - waited_connections: set[int] - - def __init__( - self, grpc_server: grpc.aio.Server, device: Device, config: Config - ) -> None: - self.log = utils.BumbleServerLoggerAdapter( - logging.getLogger(), {'service_name': 'Host', 'device': device} - ) - self.grpc_server = grpc_server - self.device = device - self.config = config - self.waited_connections = set() - - @utils.rpc - async def FactoryReset( - self, request: empty_pb2.Empty, context: grpc.ServicerContext - ) -> empty_pb2.Empty: - self.log.debug('FactoryReset') - - # delete all bonds - if self.device.keystore is not None: - await self.device.keystore.delete_all() - - # trigger gRCP server stop then return - asyncio.create_task(self.grpc_server.stop(None)) - return empty_pb2.Empty() - - @utils.rpc - async def Reset( - self, request: empty_pb2.Empty, context: grpc.ServicerContext - ) -> empty_pb2.Empty: - self.log.debug('Reset') - - # clear service. - self.waited_connections.clear() - - # (re) power device on - await self.device.power_on() - return empty_pb2.Empty() - - @utils.rpc - async def ReadLocalAddress( - self, request: empty_pb2.Empty, context: grpc.ServicerContext - ) -> ReadLocalAddressResponse: - self.log.debug('ReadLocalAddress') - return ReadLocalAddressResponse( - address=bytes(reversed(bytes(self.device.public_address))) - ) - - @utils.rpc - async def Connect( - self, request: ConnectRequest, context: grpc.ServicerContext - ) -> ConnectResponse: - # Need to reverse bytes order since Bumble Address is using MSB. - address = Address( - bytes(reversed(request.address)), address_type=Address.PUBLIC_DEVICE_ADDRESS - ) - self.log.debug(f"Connect to {address}") - - try: - connection = await self.device.connect( - address, transport=PhysicalTransport.BR_EDR - ) - except ConnectionError as e: - if e.error_code == HCI_PAGE_TIMEOUT_ERROR: - self.log.warning(f"Peer not found: {e}") - return ConnectResponse(peer_not_found=empty_pb2.Empty()) - if e.error_code == HCI_CONNECTION_ALREADY_EXISTS_ERROR: - self.log.warning(f"Connection already exists: {e}") - return ConnectResponse(connection_already_exists=empty_pb2.Empty()) - raise e - - self.log.debug(f"Connect to {address} done (handle={connection.handle})") - - cookie = any_pb2.Any(value=connection.handle.to_bytes(4, 'big')) - return ConnectResponse(connection=Connection(cookie=cookie)) - - @utils.rpc - async def WaitConnection( - self, request: WaitConnectionRequest, context: grpc.ServicerContext - ) -> WaitConnectionResponse: - if not request.address: - raise ValueError('Request address field must be set') - - # Need to reverse bytes order since Bumble Address is using MSB. - address = Address( - bytes(reversed(request.address)), address_type=Address.PUBLIC_DEVICE_ADDRESS - ) - if address in (Address.NIL, Address.ANY): - raise ValueError('Invalid address') - - self.log.debug(f"WaitConnection from {address}...") - - connection = self.device.find_connection_by_bd_addr( - address, transport=PhysicalTransport.BR_EDR - ) - if connection and id(connection) in self.waited_connections: - # this connection was already returned: wait for a new one. - connection = None - - if not connection: - connection = await self.device.accept(address) - - # save connection has waited and respond. - self.waited_connections.add(id(connection)) - - self.log.debug( - f"WaitConnection from {address} done (handle={connection.handle})" - ) - - cookie = any_pb2.Any(value=connection.handle.to_bytes(4, 'big')) - return WaitConnectionResponse(connection=Connection(cookie=cookie)) - - @utils.rpc - async def ConnectLE( - self, request: ConnectLERequest, context: grpc.ServicerContext - ) -> ConnectLEResponse: - address = utils.address_from_request(request, request.WhichOneof("address")) - if address in (Address.NIL, Address.ANY): - raise ValueError('Invalid address') - - self.log.debug(f"ConnectLE to {address}...") - - try: - connection = await self.device.connect( - address, - transport=PhysicalTransport.LE, - own_address_type=OwnAddressType(request.own_address_type), - ) - except ConnectionError as e: - if e.error_code == HCI_PAGE_TIMEOUT_ERROR: - self.log.warning(f"Peer not found: {e}") - return ConnectLEResponse(peer_not_found=empty_pb2.Empty()) - if e.error_code == HCI_CONNECTION_ALREADY_EXISTS_ERROR: - self.log.warning(f"Connection already exists: {e}") - return ConnectLEResponse(connection_already_exists=empty_pb2.Empty()) - raise e - - self.log.debug(f"ConnectLE to {address} done (handle={connection.handle})") - - cookie = any_pb2.Any(value=connection.handle.to_bytes(4, 'big')) - return ConnectLEResponse(connection=Connection(cookie=cookie)) - - @utils.rpc - async def Disconnect( - self, request: DisconnectRequest, context: grpc.ServicerContext - ) -> empty_pb2.Empty: - connection_handle = int.from_bytes(request.connection.cookie.value, 'big') - self.log.debug(f"Disconnect: {connection_handle}") - - self.log.debug("Disconnecting...") - if connection := self.device.lookup_connection(connection_handle): - await connection.disconnect(HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR) - self.log.debug("Disconnected") - - return empty_pb2.Empty() - - @utils.rpc - async def WaitDisconnection( - self, request: WaitDisconnectionRequest, context: grpc.ServicerContext - ) -> empty_pb2.Empty: - connection_handle = int.from_bytes(request.connection.cookie.value, 'big') - self.log.debug(f"WaitDisconnection: {connection_handle}") - - if connection := self.device.lookup_connection(connection_handle): - disconnection_future: asyncio.Future[None] = ( - asyncio.get_running_loop().create_future() - ) - - def on_disconnection(_: None) -> None: - disconnection_future.set_result(None) - - connection.on(connection.EVENT_DISCONNECTION, on_disconnection) - try: - await disconnection_future - self.log.debug("Disconnected") - finally: - connection.remove_listener( - connection.EVENT_DISCONNECTION, on_disconnection - ) # type: ignore - - return empty_pb2.Empty() - - @utils.rpc - async def Advertise( - self, request: AdvertiseRequest, context: grpc.ServicerContext - ) -> AsyncGenerator[AdvertiseResponse, None]: - try: - if request.legacy: - async for rsp in self.legacy_advertise(request, context): - yield rsp - else: - async for rsp in self.extended_advertise(request, context): - yield rsp - finally: - pass - - async def extended_advertise( - self, request: AdvertiseRequest, context: grpc.ServicerContext - ) -> AsyncGenerator[AdvertiseResponse, None]: - advertising_data = bytes(self.unpack_data_types(request.data)) - scan_response_data = bytes(self.unpack_data_types(request.scan_response_data)) - scannable = len(scan_response_data) != 0 - - advertising_event_properties = AdvertisingEventProperties( - is_connectable=request.connectable, - is_scannable=scannable, - is_directed=request.target is not None, - is_high_duty_cycle_directed_connectable=False, - is_legacy=False, - is_anonymous=False, - include_tx_power=False, - ) - - peer_address = Address.ANY - if request.target: - # Need to reverse bytes order since Bumble Address is using MSB. - target_bytes = bytes(reversed(request.target)) - if request.target_variant() == "public": - peer_address = Address(target_bytes, Address.PUBLIC_DEVICE_ADDRESS) - else: - peer_address = Address(target_bytes, Address.RANDOM_DEVICE_ADDRESS) - - advertising_parameters = AdvertisingParameters( - advertising_event_properties=advertising_event_properties, - own_address_type=OWN_ADDRESS_MAP[request.own_address_type], - peer_address=peer_address, - primary_advertising_phy=PRIMARY_PHY_TO_BUMBLE_PHY_MAP[request.primary_phy], - secondary_advertising_phy=SECONDARY_PHY_TO_BUMBLE_PHY_MAP[ - request.secondary_phy - ], - ) - if advertising_interval := request.interval: - advertising_parameters.primary_advertising_interval_min = int( - advertising_interval - ) - advertising_parameters.primary_advertising_interval_max = int( - advertising_interval - ) - if interval_range := request.interval_range: - advertising_parameters.primary_advertising_interval_max += int( - interval_range - ) - - advertising_set = await self.device.create_advertising_set( - advertising_parameters=advertising_parameters, - advertising_data=advertising_data, - scan_response_data=scan_response_data, - ) - - connections: asyncio.Queue[bumble.device.Connection] = asyncio.Queue() - - if request.connectable: - - def on_connection(connection: bumble.device.Connection) -> None: - if ( - connection.transport == PhysicalTransport.LE - and connection.role == Role.PERIPHERAL - ): - connections.put_nowait(connection) - - self.device.on(self.device.EVENT_CONNECTION, on_connection) - - try: - # Advertise until RPC is canceled - while True: - if not advertising_set.enabled: - self.log.debug('Advertise (extended)') - await advertising_set.start() - - if not request.connectable: - await asyncio.sleep(1) - continue - - connection = await connections.get() - - cookie = any_pb2.Any(value=connection.handle.to_bytes(4, 'big')) - yield AdvertiseResponse(connection=Connection(cookie=cookie)) - - await asyncio.sleep(1) - finally: - try: - self.log.debug('Stop Advertise (extended)') - await advertising_set.stop() - await advertising_set.remove() - except Exception: - pass - - async def legacy_advertise( - self, request: AdvertiseRequest, context: grpc.ServicerContext - ) -> AsyncGenerator[AdvertiseResponse, None]: - if advertising_interval := request.interval: - self.device.config.advertising_interval_min = int(advertising_interval) - self.device.config.advertising_interval_max = int(advertising_interval) - if interval_range := request.interval_range: - self.device.config.advertising_interval_max += int(interval_range) - if request.primary_phy: - raise NotImplementedError("TODO: add support for `request.primary_phy`") - if request.secondary_phy: - raise NotImplementedError("TODO: add support for `request.secondary_phy`") - - if self.device.is_advertising: - raise NotImplementedError('TODO: add support for advertising sets') - - if data := request.data: - self.device.advertising_data = bytes(self.unpack_data_types(data)) - - if scan_response_data := request.scan_response_data: - self.device.scan_response_data = bytes( - self.unpack_data_types(scan_response_data) - ) - scannable = True - else: - scannable = False - - # Retrieve services data - for service in self.device.gatt_server.attributes: - if isinstance(service, Service) and ( - service_data := service.get_advertising_data() - ): - service_uuid = service.uuid.to_hex_str('-') - if ( - service_uuid in request.data.incomplete_service_class_uuids16 - or service_uuid in request.data.complete_service_class_uuids16 - or service_uuid in request.data.incomplete_service_class_uuids32 - or service_uuid in request.data.complete_service_class_uuids32 - or service_uuid - in request.data.incomplete_service_class_uuids128 - or service_uuid in request.data.complete_service_class_uuids128 - ): - self.device.advertising_data += service_data - if ( - service_uuid - in scan_response_data.incomplete_service_class_uuids16 - or service_uuid - in scan_response_data.complete_service_class_uuids16 - or service_uuid - in scan_response_data.incomplete_service_class_uuids32 - or service_uuid - in scan_response_data.complete_service_class_uuids32 - or service_uuid - in scan_response_data.incomplete_service_class_uuids128 - or service_uuid - in scan_response_data.complete_service_class_uuids128 - ): - self.device.scan_response_data += service_data - - target = None - if request.connectable and scannable: - advertising_type = AdvertisingType.UNDIRECTED_CONNECTABLE_SCANNABLE - elif scannable: - advertising_type = AdvertisingType.UNDIRECTED_SCANNABLE - else: - advertising_type = AdvertisingType.UNDIRECTED - else: - target = None - advertising_type = AdvertisingType.UNDIRECTED - - if request.target: - # Need to reverse bytes order since Bumble Address is using MSB. - target_bytes = bytes(reversed(request.target)) - if request.target_variant() == "public": - target = Address(target_bytes, Address.PUBLIC_DEVICE_ADDRESS) - advertising_type = AdvertisingType.DIRECTED_CONNECTABLE_LOW_DUTY - else: - target = Address(target_bytes, Address.RANDOM_DEVICE_ADDRESS) - advertising_type = AdvertisingType.DIRECTED_CONNECTABLE_LOW_DUTY - - connections: asyncio.Queue[bumble.device.Connection] = asyncio.Queue() - - if request.connectable: - - def on_connection(connection: bumble.device.Connection) -> None: - if ( - connection.transport == PhysicalTransport.LE - and connection.role == Role.PERIPHERAL - ): - connections.put_nowait(connection) - - self.device.on(self.device.EVENT_CONNECTION, on_connection) - - try: - while True: - if not self.device.is_advertising: - self.log.debug('Advertise') - await self.device.start_advertising( - target=target, - advertising_type=advertising_type, - own_address_type=OwnAddressType(request.own_address_type), - ) - - if not request.connectable: - await asyncio.sleep(1) - continue - - self.log.debug('Wait for LE connection...') - connection = await connections.get() - - self.log.debug( - f"Advertise: Connected to {connection.peer_address} (handle={connection.handle})" - ) - - cookie = any_pb2.Any(value=connection.handle.to_bytes(4, 'big')) - yield AdvertiseResponse(connection=Connection(cookie=cookie)) - - # wait a small delay before restarting the advertisement. - await asyncio.sleep(1) - finally: - if request.connectable: - self.device.remove_listener(self.device.EVENT_CONNECTION, on_connection) # type: ignore - - try: - self.log.debug('Stop advertising') - await bumble.utils.cancel_on_event( - self.device, 'flush', self.device.stop_advertising() - ) - except Exception: - pass - - @utils.rpc - async def Scan( - self, request: ScanRequest, context: grpc.ServicerContext - ) -> AsyncGenerator[ScanningResponse, None]: - # TODO: modify `start_scanning` to accept floats instead of int for ms values - self.log.debug('Scan') - - scanning_phys = [] - if PRIMARY_1M in request.phys: - scanning_phys.append(int(Phy.LE_1M)) - if PRIMARY_CODED in request.phys: - scanning_phys.append(int(Phy.LE_CODED)) - if not scanning_phys: - scanning_phys = [int(Phy.LE_1M), int(Phy.LE_CODED)] - - scan_queue: asyncio.Queue[Advertisement] = asyncio.Queue() - handler = self.device.on(self.device.EVENT_ADVERTISEMENT, scan_queue.put_nowait) - await self.device.start_scanning( - legacy=request.legacy, - active=not request.passive, - own_address_type=OwnAddressType(request.own_address_type), - scan_interval=( - int(request.interval) - if request.interval - else DEVICE_DEFAULT_SCAN_INTERVAL - ), - scan_window=( - int(request.window) if request.window else DEVICE_DEFAULT_SCAN_WINDOW - ), - scanning_phys=scanning_phys, - ) - - try: - # TODO: add support for `direct_address` in Bumble - # TODO: add support for `periodic_advertising_interval` in Bumble - while adv := await scan_queue.get(): - sr = ScanningResponse( - legacy=adv.is_legacy, - connectable=adv.is_connectable, - scannable=adv.is_scannable, - truncated=adv.is_truncated, - sid=adv.sid, - primary_phy=PRIMARY_PHY_MAP[adv.primary_phy], - secondary_phy=SECONDARY_PHY_MAP[adv.secondary_phy], - tx_power=adv.tx_power, - rssi=adv.rssi, - data=self.pack_data_types(adv.data), - ) - - if adv.address.address_type == Address.PUBLIC_DEVICE_ADDRESS: - sr.public = bytes(reversed(bytes(adv.address))) - elif adv.address.address_type == Address.RANDOM_DEVICE_ADDRESS: - sr.random = bytes(reversed(bytes(adv.address))) - elif adv.address.address_type == Address.PUBLIC_IDENTITY_ADDRESS: - sr.public_identity = bytes(reversed(bytes(adv.address))) - else: - sr.random_static_identity = bytes(reversed(bytes(adv.address))) - - yield sr - - finally: - self.device.remove_listener(self.device.EVENT_ADVERTISEMENT, handler) # type: ignore - try: - self.log.debug('Stop scanning') - await bumble.utils.cancel_on_event( - self.device, 'flush', self.device.stop_scanning() - ) - except Exception: - pass - - @utils.rpc - async def Inquiry( - self, request: empty_pb2.Empty, context: grpc.ServicerContext - ) -> AsyncGenerator[InquiryResponse, None]: - self.log.debug('Inquiry') - - inquiry_queue: asyncio.Queue[ - tuple[Address, int, AdvertisingData, int] | None - ] = asyncio.Queue() - complete_handler = self.device.on( - self.device.EVENT_INQUIRY_COMPLETE, lambda: inquiry_queue.put_nowait(None) - ) - result_handler = self.device.on( # type: ignore - self.device.EVENT_INQUIRY_RESULT, - lambda address, class_of_device, eir_data, rssi: inquiry_queue.put_nowait( # type: ignore - (address, class_of_device, eir_data, rssi) # type: ignore - ), - ) - - await self.device.start_discovery(auto_restart=False) - try: - while inquiry_result := await inquiry_queue.get(): - (address, class_of_device, eir_data, rssi) = inquiry_result - # FIXME: if needed, add support for `page_scan_repetition_mode` and `clock_offset` in Bumble - yield InquiryResponse( - address=bytes(reversed(bytes(address))), - class_of_device=class_of_device, - rssi=rssi, - data=self.pack_data_types(eir_data), - ) - - finally: - self.device.remove_listener( - self.device.EVENT_INQUIRY_COMPLETE, complete_handler - ) # type: ignore - self.device.remove_listener( - self.device.EVENT_INQUIRY_RESULT, result_handler - ) # type: ignore - try: - self.log.debug('Stop inquiry') - await bumble.utils.cancel_on_event( - self.device, 'flush', self.device.stop_discovery() - ) - except Exception: - pass - - @utils.rpc - async def SetDiscoverabilityMode( - self, request: SetDiscoverabilityModeRequest, context: grpc.ServicerContext - ) -> empty_pb2.Empty: - self.log.debug("SetDiscoverabilityMode") - await self.device.set_discoverable(request.mode != NOT_DISCOVERABLE) - return empty_pb2.Empty() - - @utils.rpc - async def SetConnectabilityMode( - self, request: SetConnectabilityModeRequest, context: grpc.ServicerContext - ) -> empty_pb2.Empty: - self.log.debug("SetConnectabilityMode") - await self.device.set_connectable(request.mode != NOT_CONNECTABLE) - return empty_pb2.Empty() - - def unpack_data_types(self, dt: DataTypes) -> AdvertisingData: - ad_structures: list[tuple[int, bytes]] = [] - - uuids: list[str] - datas: dict[str, bytes] - - def uuid128_from_str(uuid: str) -> bytes: - """Decode a 128-bit uuid encoded as XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX - to byte format.""" - return bytes(reversed(bytes.fromhex(uuid.replace('-', '')))) - - def uuid32_from_str(uuid: str) -> bytes: - """Decode a 32-bit uuid encoded as XXXXXXXX to byte format.""" - return bytes(reversed(bytes.fromhex(uuid))) - - def uuid16_from_str(uuid: str) -> bytes: - """Decode a 16-bit uuid encoded as XXXX to byte format.""" - return bytes(reversed(bytes.fromhex(uuid))) - - if uuids := dt.incomplete_service_class_uuids16: - ad_structures.append( - ( - AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS, - b''.join([uuid16_from_str(uuid) for uuid in uuids]), - ) - ) - if uuids := dt.complete_service_class_uuids16: - ad_structures.append( - ( - AdvertisingData.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS, - b''.join([uuid16_from_str(uuid) for uuid in uuids]), - ) - ) - if uuids := dt.incomplete_service_class_uuids32: - ad_structures.append( - ( - AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS, - b''.join([uuid32_from_str(uuid) for uuid in uuids]), - ) - ) - if uuids := dt.complete_service_class_uuids32: - ad_structures.append( - ( - AdvertisingData.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS, - b''.join([uuid32_from_str(uuid) for uuid in uuids]), - ) - ) - if uuids := dt.incomplete_service_class_uuids128: - ad_structures.append( - ( - AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS, - b''.join([uuid128_from_str(uuid) for uuid in uuids]), - ) - ) - if uuids := dt.complete_service_class_uuids128: - ad_structures.append( - ( - AdvertisingData.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS, - b''.join([uuid128_from_str(uuid) for uuid in uuids]), - ) - ) - if dt.HasField('include_shortened_local_name'): - ad_structures.append( - ( - AdvertisingData.SHORTENED_LOCAL_NAME, - bytes(self.device.name[:8], 'utf-8'), - ) - ) - elif dt.shortened_local_name: - ad_structures.append( - ( - AdvertisingData.SHORTENED_LOCAL_NAME, - bytes(dt.shortened_local_name, 'utf-8'), - ) - ) - if dt.HasField('include_complete_local_name'): - ad_structures.append( - (AdvertisingData.COMPLETE_LOCAL_NAME, bytes(self.device.name, 'utf-8')) - ) - elif dt.complete_local_name: - ad_structures.append( - ( - AdvertisingData.COMPLETE_LOCAL_NAME, - bytes(dt.complete_local_name, 'utf-8'), - ) - ) - if dt.HasField('include_tx_power_level'): - raise ValueError('unsupported data type') - elif dt.tx_power_level: - ad_structures.append( - ( - AdvertisingData.TX_POWER_LEVEL, - bytes(struct.pack(' DataTypes: - dt = DataTypes() - uuids: list[UUID] - s: str - i: int - ij: tuple[int, int] - uuid_data: tuple[UUID, bytes] - data: bytes - - if uuids := cast( - list[UUID], - ad.get(AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS), - ): - dt.incomplete_service_class_uuids16.extend( - list(map(lambda x: x.to_hex_str('-'), uuids)) - ) - if uuids := cast( - list[UUID], - ad.get(AdvertisingData.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS), - ): - dt.complete_service_class_uuids16.extend( - list(map(lambda x: x.to_hex_str('-'), uuids)) - ) - if uuids := cast( - list[UUID], - ad.get(AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS), - ): - dt.incomplete_service_class_uuids32.extend( - list(map(lambda x: x.to_hex_str('-'), uuids)) - ) - if uuids := cast( - list[UUID], - ad.get(AdvertisingData.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS), - ): - dt.complete_service_class_uuids32.extend( - list(map(lambda x: x.to_hex_str('-'), uuids)) - ) - if uuids := cast( - list[UUID], - ad.get(AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS), - ): - dt.incomplete_service_class_uuids128.extend( - list(map(lambda x: x.to_hex_str('-'), uuids)) - ) - if uuids := cast( - list[UUID], - ad.get(AdvertisingData.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS), - ): - dt.complete_service_class_uuids128.extend( - list(map(lambda x: x.to_hex_str('-'), uuids)) - ) - if s := cast(str, ad.get(AdvertisingData.SHORTENED_LOCAL_NAME)): - dt.shortened_local_name = s - if s := cast(str, ad.get(AdvertisingData.COMPLETE_LOCAL_NAME)): - dt.complete_local_name = s - if i := cast(int, ad.get(AdvertisingData.TX_POWER_LEVEL)): - dt.tx_power_level = i - if i := cast(int, ad.get(AdvertisingData.CLASS_OF_DEVICE)): - dt.class_of_device = i - if ij := cast( - tuple[int, int], - ad.get(AdvertisingData.PERIPHERAL_CONNECTION_INTERVAL_RANGE), - ): - dt.peripheral_connection_interval_min = ij[0] - dt.peripheral_connection_interval_max = ij[1] - if uuids := cast( - list[UUID], - ad.get(AdvertisingData.LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS), - ): - dt.service_solicitation_uuids16.extend( - list(map(lambda x: x.to_hex_str('-'), uuids)) - ) - if uuids := cast( - list[UUID], - ad.get(AdvertisingData.LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS), - ): - dt.service_solicitation_uuids32.extend( - list(map(lambda x: x.to_hex_str('-'), uuids)) - ) - if uuids := cast( - list[UUID], - ad.get(AdvertisingData.LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS), - ): - dt.service_solicitation_uuids128.extend( - list(map(lambda x: x.to_hex_str('-'), uuids)) - ) - if uuid_data := cast( - tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_16_BIT_UUID) - ): - dt.service_data_uuid16[uuid_data[0].to_hex_str('-')] = uuid_data[1] - if uuid_data := cast( - tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_32_BIT_UUID) - ): - dt.service_data_uuid32[uuid_data[0].to_hex_str('-')] = uuid_data[1] - if uuid_data := cast( - tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_128_BIT_UUID) - ): - dt.service_data_uuid128[uuid_data[0].to_hex_str('-')] = uuid_data[1] - if data := cast(bytes, ad.get(AdvertisingData.PUBLIC_TARGET_ADDRESS, raw=True)): - dt.public_target_addresses.extend( - [data[i * 6 :: i * 6 + 6] for i in range(int(len(data) / 6))] - ) - if data := cast(bytes, ad.get(AdvertisingData.RANDOM_TARGET_ADDRESS, raw=True)): - dt.random_target_addresses.extend( - [data[i * 6 :: i * 6 + 6] for i in range(int(len(data) / 6))] - ) - if appearance := cast(Appearance, ad.get(AdvertisingData.APPEARANCE)): - dt.appearance = int(appearance) - if i := cast(int, ad.get(AdvertisingData.ADVERTISING_INTERVAL)): - dt.advertising_interval = i - if s := cast(str, ad.get(AdvertisingData.URI)): - dt.uri = s - if data := cast(bytes, ad.get(AdvertisingData.LE_SUPPORTED_FEATURES, raw=True)): - dt.le_supported_features = data - if data := cast( - bytes, ad.get(AdvertisingData.MANUFACTURER_SPECIFIC_DATA, raw=True) - ): - dt.manufacturer_specific_data = data - - return dt diff --git a/bumble/pandora/l2cap.py b/bumble/pandora/l2cap.py deleted file mode 100644 index 85a6642bf..000000000 --- a/bumble/pandora/l2cap.py +++ /dev/null @@ -1,310 +0,0 @@ -# Copyright 2024 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. -from __future__ import annotations - -import asyncio -import json -import logging -from asyncio import Future -from asyncio import Queue as AsyncQueue -from collections.abc import AsyncGenerator -from dataclasses import dataclass - -import grpc -from google.protobuf import any_pb2, empty_pb2 # pytype: disable=pyi-error -from pandora.l2cap_grpc_aio import L2CAPServicer # pytype: disable=pyi-error -from pandora.l2cap_pb2 import ( - COMMAND_NOT_UNDERSTOOD, - INVALID_CID_IN_REQUEST, - ConnectRequest, - ConnectResponse, - CreditBasedChannelRequest, - DisconnectRequest, - DisconnectResponse, - ReceiveRequest, - ReceiveResponse, - SendRequest, - SendResponse, - WaitConnectionRequest, - WaitConnectionResponse, - WaitDisconnectionRequest, - WaitDisconnectionResponse, -) -from pandora.l2cap_pb2 import Channel as PandoraChannel # pytype: disable=pyi-error - -from bumble.core import InvalidArgumentError, OutOfResourcesError -from bumble.device import Device -from bumble.l2cap import ( - ClassicChannel, - ClassicChannelServer, - ClassicChannelSpec, - LeCreditBasedChannel, - LeCreditBasedChannelServer, - LeCreditBasedChannelSpec, -) -from bumble.pandora import utils -from bumble.pandora.config import Config - -L2capChannel = ClassicChannel | LeCreditBasedChannel - - -@dataclass -class ChannelContext: - close_future: Future - sdu_queue: AsyncQueue - - -class L2CAPService(L2CAPServicer): - def __init__(self, device: Device, config: Config) -> None: - self.log = utils.BumbleServerLoggerAdapter( - logging.getLogger(), {'service_name': 'L2CAP', 'device': device} - ) - self.device = device - self.config = config - self.channels: dict[bytes, ChannelContext] = {} - - def register_event(self, l2cap_channel: L2capChannel) -> ChannelContext: - close_future = asyncio.get_running_loop().create_future() - sdu_queue: AsyncQueue = AsyncQueue() - - def on_channel_sdu(sdu): - sdu_queue.put_nowait(sdu) - - def on_close(): - close_future.set_result(None) - - l2cap_channel.sink = on_channel_sdu - l2cap_channel.on(l2cap_channel.EVENT_CLOSE, on_close) - - return ChannelContext(close_future, sdu_queue) - - @utils.rpc - async def WaitConnection( - self, request: WaitConnectionRequest, context: grpc.ServicerContext - ) -> WaitConnectionResponse: - self.log.debug('WaitConnection') - if not request.connection: - raise ValueError('A valid connection field must be set') - - # find connection on device based on connection cookie value - connection_handle = int.from_bytes(request.connection.cookie.value, 'big') - connection = self.device.lookup_connection(connection_handle) - - if not connection: - raise ValueError('The connection specified is invalid.') - - oneof = request.WhichOneof('type') - self.log.debug(f'WaitConnection channel request type: {oneof}.') - channel_type = getattr(request, oneof) - spec: ClassicChannelSpec | LeCreditBasedChannelSpec | None = None - l2cap_server: ClassicChannelServer | LeCreditBasedChannelServer | None = None - if isinstance(channel_type, CreditBasedChannelRequest): - spec = LeCreditBasedChannelSpec( - psm=channel_type.spsm, - max_credits=channel_type.initial_credit, - mtu=channel_type.mtu, - mps=channel_type.mps, - ) - if channel_type.spsm in self.device.l2cap_channel_manager.le_coc_servers: - l2cap_server = self.device.l2cap_channel_manager.le_coc_servers[ - channel_type.spsm - ] - else: - spec = ClassicChannelSpec( - psm=channel_type.psm, - mtu=channel_type.mtu, - ) - if channel_type.psm in self.device.l2cap_channel_manager.servers: - l2cap_server = self.device.l2cap_channel_manager.servers[ - channel_type.psm - ] - - self.log.info(f'Listening for L2CAP connection on PSM {spec.psm}') - channel_future: Future[PandoraChannel] = ( - asyncio.get_running_loop().create_future() - ) - - def on_l2cap_channel(l2cap_channel: L2capChannel): - try: - channel_context = self.register_event(l2cap_channel) - pandora_channel: PandoraChannel = self.craft_pandora_channel( - connection_handle, l2cap_channel - ) - self.channels[pandora_channel.cookie.value] = channel_context - channel_future.set_result(pandora_channel) - except Exception as e: - self.log.error(f'Failed to set channel future: {e}') - - if l2cap_server is None: - l2cap_server = self.device.create_l2cap_server( - spec=spec, handler=on_l2cap_channel - ) - else: - l2cap_server.on(l2cap_server.EVENT_CONNECTION, on_l2cap_channel) - - try: - self.log.debug('Waiting for a channel connection.') - pandora_channel: PandoraChannel = await channel_future - - return WaitConnectionResponse(channel=pandora_channel) - except Exception as e: - self.log.warning(f'Exception: {e}') - - return WaitConnectionResponse(error=COMMAND_NOT_UNDERSTOOD) - - @utils.rpc - async def WaitDisconnection( - self, request: WaitDisconnectionRequest, context: grpc.ServicerContext - ) -> WaitDisconnectionResponse: - try: - self.log.debug('WaitDisconnection') - - await self.lookup_context(request.channel).close_future - self.log.debug("return WaitDisconnectionResponse") - return WaitDisconnectionResponse(success=empty_pb2.Empty()) - except KeyError as e: - self.log.warning(f'WaitDisconnection: Unable to find the channel: {e}') - return WaitDisconnectionResponse(error=INVALID_CID_IN_REQUEST) - except Exception as e: - self.log.exception(f'WaitDisonnection failed: {e}') - return WaitDisconnectionResponse(error=COMMAND_NOT_UNDERSTOOD) - - @utils.rpc - async def Receive( - self, request: ReceiveRequest, context: grpc.ServicerContext - ) -> AsyncGenerator[ReceiveResponse, None]: - self.log.debug('Receive') - oneof = request.WhichOneof('source') - self.log.debug(f'Source: {oneof}.') - pandora_channel = getattr(request, oneof) - - sdu_queue = self.lookup_context(pandora_channel).sdu_queue - - while sdu := await sdu_queue.get(): - self.log.debug(f'Receive: Received {len(sdu)} bytes -> {sdu.decode()}') - response = ReceiveResponse(data=sdu) - yield response - - @utils.rpc - async def Connect( - self, request: ConnectRequest, context: grpc.ServicerContext - ) -> ConnectResponse: - self.log.debug('Connect') - - if not request.connection: - raise ValueError('A valid connection field must be set') - - # find connection on device based on connection cookie value - connection_handle = int.from_bytes(request.connection.cookie.value, 'big') - connection = self.device.lookup_connection(connection_handle) - - if not connection: - raise ValueError('The connection specified is invalid.') - - oneof = request.WhichOneof('type') - self.log.debug(f'Channel request type: {oneof}.') - channel_type = getattr(request, oneof) - spec: ClassicChannelSpec | LeCreditBasedChannelSpec | None = None - if isinstance(channel_type, CreditBasedChannelRequest): - spec = LeCreditBasedChannelSpec( - psm=channel_type.spsm, - max_credits=channel_type.initial_credit, - mtu=channel_type.mtu, - mps=channel_type.mps, - ) - else: - spec = ClassicChannelSpec( - psm=channel_type.psm, - mtu=channel_type.mtu, - ) - - try: - self.log.info(f'Opening L2CAP channel on PSM = {spec.psm}') - l2cap_channel = await connection.create_l2cap_channel(spec=spec) - channel_context = self.register_event(l2cap_channel) - pandora_channel = self.craft_pandora_channel( - connection_handle, l2cap_channel - ) - self.channels[pandora_channel.cookie.value] = channel_context - - return ConnectResponse(channel=pandora_channel) - - except OutOfResourcesError as e: - self.log.error(e) - return ConnectResponse(error=INVALID_CID_IN_REQUEST) - except InvalidArgumentError as e: - self.log.error(e) - return ConnectResponse(error=COMMAND_NOT_UNDERSTOOD) - - @utils.rpc - async def Disconnect( - self, request: DisconnectRequest, context: grpc.ServicerContext - ) -> DisconnectResponse: - try: - self.log.debug('Disconnect') - l2cap_channel = self.lookup_channel(request.channel) - if not l2cap_channel: - self.log.warning('Disconnect: Unable to find the channel') - return DisconnectResponse(error=INVALID_CID_IN_REQUEST) - - await l2cap_channel.disconnect() - return DisconnectResponse(success=empty_pb2.Empty()) - except Exception as e: - self.log.exception(f'Disonnect failed: {e}') - return DisconnectResponse(error=COMMAND_NOT_UNDERSTOOD) - - @utils.rpc - async def Send( - self, request: SendRequest, context: grpc.ServicerContext - ) -> SendResponse: - self.log.debug('Send') - try: - oneof = request.WhichOneof('sink') - self.log.debug(f'Sink: {oneof}.') - pandora_channel = getattr(request, oneof) - - l2cap_channel = self.lookup_channel(pandora_channel) - if not l2cap_channel: - return SendResponse(error=COMMAND_NOT_UNDERSTOOD) - if isinstance(l2cap_channel, ClassicChannel): - l2cap_channel.write(request.data) - else: - l2cap_channel.write(request.data) - return SendResponse(success=empty_pb2.Empty()) - except Exception as e: - self.log.exception(f'Disonnect failed: {e}') - return SendResponse(error=COMMAND_NOT_UNDERSTOOD) - - def craft_pandora_channel( - self, - connection_handle: int, - l2cap_channel: L2capChannel, - ) -> PandoraChannel: - parameters = { - "connection_handle": connection_handle, - "source_cid": l2cap_channel.source_cid, - } - cookie = any_pb2.Any() - cookie.value = json.dumps(parameters).encode() - return PandoraChannel(cookie=cookie) - - def lookup_channel(self, pandora_channel: PandoraChannel) -> L2capChannel: - (connection_handle, source_cid) = json.loads( - pandora_channel.cookie.value - ).values() - - return self.device.l2cap_channel_manager.channels[connection_handle][source_cid] - - def lookup_context(self, pandora_channel: PandoraChannel) -> ChannelContext: - return self.channels[pandora_channel.cookie.value] diff --git a/bumble/pandora/py.typed b/bumble/pandora/py.typed deleted file mode 100644 index e69de29bb..000000000 diff --git a/bumble/pandora/security.py b/bumble/pandora/security.py deleted file mode 100644 index 523f06eb5..000000000 --- a/bumble/pandora/security.py +++ /dev/null @@ -1,575 +0,0 @@ -# Copyright 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. - -from __future__ import annotations - -import asyncio -import contextlib -import logging -from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable -from typing import Any - -import grpc -from google.protobuf import ( - any_pb2, # pytype: disable=pyi-error - empty_pb2, # pytype: disable=pyi-error - wrappers_pb2, # pytype: disable=pyi-error -) -from pandora.host_pb2 import Connection -from pandora.security_grpc_aio import SecurityServicer, SecurityStorageServicer -from pandora.security_pb2 import ( - LE_LEVEL1, - LE_LEVEL2, - LE_LEVEL3, - LE_LEVEL4, - LEVEL0, - LEVEL1, - LEVEL2, - LEVEL3, - LEVEL4, - DeleteBondRequest, - IsBondedRequest, - LESecurityLevel, - PairingEvent, - PairingEventAnswer, - SecureRequest, - SecureResponse, - SecurityLevel, - WaitSecurityRequest, - WaitSecurityResponse, -) - -import bumble.utils -from bumble import hci -from bumble.core import InvalidArgumentError, PhysicalTransport, ProtocolError -from bumble.device import Connection as BumbleConnection -from bumble.device import Device -from bumble.hci import HCI_Error, Role -from bumble.pairing import PairingConfig -from bumble.pairing import PairingDelegate as BasePairingDelegate -from bumble.pandora import utils -from bumble.pandora.config import Config - - -class PairingDelegate(BasePairingDelegate): - def __init__( - self, - connection: BumbleConnection, - service: SecurityService, - io_capability: BasePairingDelegate.IoCapability = BasePairingDelegate.NO_OUTPUT_NO_INPUT, - local_initiator_key_distribution: BasePairingDelegate.KeyDistribution = BasePairingDelegate.DEFAULT_KEY_DISTRIBUTION, - local_responder_key_distribution: BasePairingDelegate.KeyDistribution = BasePairingDelegate.DEFAULT_KEY_DISTRIBUTION, - ) -> None: - self.log = utils.BumbleServerLoggerAdapter( - logging.getLogger(), - {'service_name': 'Security', 'device': connection.device}, - ) - self.connection = connection - self.service = service - super().__init__( - io_capability, - local_initiator_key_distribution, - local_responder_key_distribution, - ) - - async def accept(self) -> bool: - return True - - def add_origin(self, ev: PairingEvent) -> PairingEvent: - if not self.connection.is_incomplete: - assert ev.connection - ev.connection.CopyFrom( - Connection( - cookie=any_pb2.Any(value=self.connection.handle.to_bytes(4, 'big')) - ) - ) - else: - # In BR/EDR, connection may not be complete, - # use address instead - assert self.connection.transport == PhysicalTransport.BR_EDR - ev.address = bytes(reversed(bytes(self.connection.peer_address))) - - return ev - - async def confirm(self, auto: bool = False) -> bool: - self.log.debug( - f"Pairing event: `just_works` (io_capability: {self.io_capability})" - ) - - if self.service.event_queue is None or self.service.event_answer is None: - return True - - event = self.add_origin(PairingEvent(just_works=empty_pb2.Empty())) - self.service.event_queue.put_nowait(event) - answer = await anext(self.service.event_answer) # type: ignore - assert answer.event == event - assert answer.answer_variant() == 'confirm' and answer.confirm is not None - return answer.confirm - - async def compare_numbers(self, number: int, digits: int = 6) -> bool: - self.log.debug( - f"Pairing event: `numeric_comparison` (io_capability: {self.io_capability})" - ) - - if self.service.event_queue is None or self.service.event_answer is None: - raise RuntimeError('security: unhandled number comparison request') - - event = self.add_origin(PairingEvent(numeric_comparison=number)) - self.service.event_queue.put_nowait(event) - answer = await anext(self.service.event_answer) # type: ignore - assert answer.event == event - assert answer.answer_variant() == 'confirm' and answer.confirm is not None - return answer.confirm - - async def get_number(self) -> int | None: - self.log.debug( - f"Pairing event: `passkey_entry_request` (io_capability: {self.io_capability})" - ) - - if self.service.event_queue is None or self.service.event_answer is None: - raise RuntimeError('security: unhandled number request') - - event = self.add_origin(PairingEvent(passkey_entry_request=empty_pb2.Empty())) - self.service.event_queue.put_nowait(event) - answer = await anext(self.service.event_answer) # type: ignore - assert answer.event == event - if answer.answer_variant() is None: - return None - assert answer.answer_variant() == 'passkey' - return answer.passkey - - async def get_string(self, max_length: int) -> str | None: - self.log.debug( - f"Pairing event: `pin_code_request` (io_capability: {self.io_capability})" - ) - - if self.service.event_queue is None or self.service.event_answer is None: - raise RuntimeError('security: unhandled pin_code request') - - event = self.add_origin(PairingEvent(pin_code_request=empty_pb2.Empty())) - self.service.event_queue.put_nowait(event) - answer = await anext(self.service.event_answer) # type: ignore - assert answer.event == event - if answer.answer_variant() is None: - return None - assert answer.answer_variant() == 'pin' - - if answer.pin is None: - return None - - pin = answer.pin.decode('utf-8') - if not pin or len(pin) > max_length: - raise ValueError(f'Pin must be utf-8 encoded up to {max_length} bytes') - - return pin - - async def display_number(self, number: int, digits: int = 6) -> None: - if ( - self.connection.transport == PhysicalTransport.BR_EDR - and self.io_capability == BasePairingDelegate.DISPLAY_OUTPUT_ONLY - ): - return - - self.log.debug( - f"Pairing event: `passkey_entry_notification` (io_capability: {self.io_capability})" - ) - - if self.service.event_queue is None: - raise RuntimeError('security: unhandled number display request') - - event = self.add_origin(PairingEvent(passkey_entry_notification=number)) - self.service.event_queue.put_nowait(event) - - -class SecurityService(SecurityServicer): - def __init__(self, device: Device, config: Config) -> None: - self.log = utils.BumbleServerLoggerAdapter( - logging.getLogger(), {'service_name': 'Security', 'device': device} - ) - self.event_queue: asyncio.Queue[PairingEvent] | None = None - self.event_answer: AsyncIterator[PairingEventAnswer] | None = None - self.device = device - self.config = config - - def pairing_config_factory(connection: BumbleConnection) -> PairingConfig: - return PairingConfig( - sc=config.pairing_sc_enable, - mitm=config.pairing_mitm_enable, - bonding=config.pairing_bonding_enable, - identity_address_type=( - PairingConfig.AddressType.PUBLIC - if connection.self_address.is_public - else config.identity_address_type - ), - delegate=PairingDelegate( - connection, - self, - io_capability=config.io_capability, - local_initiator_key_distribution=config.smp_local_initiator_key_distribution, - local_responder_key_distribution=config.smp_local_responder_key_distribution, - ), - ) - - self.device.pairing_config_factory = pairing_config_factory - - async def _classic_level_reached( - self, level: SecurityLevel, connection: BumbleConnection - ) -> bool: - if level == LEVEL0: - return True - if level == LEVEL1: - return connection.encryption == 0 or connection.authenticated - if level == LEVEL2: - return connection.encryption != 0 and connection.authenticated - - link_key_type: int | None = None - if (keystore := connection.device.keystore) and ( - keys := await keystore.get(str(connection.peer_address)) - ): - link_key_type = keys.link_key_type - self.log.debug("link_key_type: %d", link_key_type) - - if level == LEVEL3: - return ( - connection.encryption != 0 - and connection.authenticated - and link_key_type - in ( - hci.LinkKeyType.AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_192, - hci.LinkKeyType.AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_256, - ) - ) - if level == LEVEL4: - return ( - connection.encryption == hci.HCI_Encryption_Change_Event.Enabled.AES_CCM - and connection.authenticated - and link_key_type - == hci.LinkKeyType.AUTHENTICATED_COMBINATION_KEY_GENERATED_FROM_P_256 - ) - raise InvalidArgumentError(f"Unexpected level {level}") - - def _le_level_reached( - self, level: LESecurityLevel, connection: BumbleConnection - ) -> bool: - if level == LE_LEVEL1: - return True - if level == LE_LEVEL2: - return connection.encryption != 0 - if level == LE_LEVEL3: - return connection.encryption != 0 and connection.authenticated - if level == LE_LEVEL4: - return ( - connection.encryption != 0 - and connection.authenticated - and connection.sc - ) - raise InvalidArgumentError(f"Unexpected level {level}") - - @utils.rpc - async def OnPairing( - self, request: AsyncIterator[PairingEventAnswer], context: grpc.ServicerContext - ) -> AsyncGenerator[PairingEvent, None]: - self.log.debug('OnPairing') - - if self.event_queue is not None: - raise RuntimeError('already streaming pairing events') - - if len(self.device.connections): - raise RuntimeError( - 'the `OnPairing` method shall be initiated before establishing any connections.' - ) - - self.event_queue = asyncio.Queue() - self.event_answer = request - - try: - while event := await self.event_queue.get(): - yield event - - finally: - self.event_queue = None - self.event_answer = None - - @utils.rpc - async def Secure( - self, request: SecureRequest, context: grpc.ServicerContext - ) -> SecureResponse: - connection_handle = int.from_bytes(request.connection.cookie.value, 'big') - self.log.debug(f"Secure: {connection_handle}") - - connection = self.device.lookup_connection(connection_handle) - assert connection - - oneof = request.WhichOneof('level') - level = getattr(request, oneof) - assert {PhysicalTransport.BR_EDR: 'classic', PhysicalTransport.LE: 'le'}[ - connection.transport - ] == oneof - - # security level already reached - if await self.reached_security_level(connection, level): - return SecureResponse(success=empty_pb2.Empty()) - - # trigger pairing if needed - if self.need_pairing(connection, level): - try: - self.log.debug('Pair...') - - security_result = asyncio.get_running_loop().create_future() - - with contextlib.closing(bumble.utils.EventWatcher()) as watcher: - - @watcher.on(connection, connection.EVENT_PAIRING) - def on_pairing(*_: Any) -> None: - security_result.set_result('success') - - @watcher.on(connection, connection.EVENT_PAIRING_FAILURE) - def on_pairing_failure(*_: Any) -> None: - security_result.set_result('pairing_failure') - - @watcher.on(connection, connection.EVENT_DISCONNECTION) - def on_disconnection(*_: Any) -> None: - security_result.set_result('connection_died') - - if ( - connection.transport == PhysicalTransport.LE - and connection.role == Role.PERIPHERAL - ): - connection.request_pairing() - else: - await connection.pair() - - result = await security_result - - self.log.debug(f'Pairing session complete, status={result}') - if result != 'success': - return SecureResponse(**{result: empty_pb2.Empty()}) - except asyncio.CancelledError: - self.log.warning("Connection died during encryption") - return SecureResponse(connection_died=empty_pb2.Empty()) - except (HCI_Error, ProtocolError) as e: - self.log.warning(f"Pairing failure: {e}") - return SecureResponse(pairing_failure=empty_pb2.Empty()) - - # trigger authentication if needed - if self.need_authentication(connection, level): - try: - self.log.debug('Authenticate...') - await connection.authenticate() - self.log.debug('Authenticated') - except asyncio.CancelledError: - self.log.warning("Connection died during authentication") - return SecureResponse(connection_died=empty_pb2.Empty()) - except (HCI_Error, ProtocolError) as e: - self.log.warning(f"Authentication failure: {e}") - return SecureResponse(authentication_failure=empty_pb2.Empty()) - - # trigger encryption if needed - if self.need_encryption(connection, level): - try: - self.log.debug('Encrypt...') - await connection.encrypt() - self.log.debug('Encrypted') - except asyncio.CancelledError: - self.log.warning("Connection died during encryption") - return SecureResponse(connection_died=empty_pb2.Empty()) - except (HCI_Error, ProtocolError) as e: - self.log.warning(f"Encryption failure: {e}") - return SecureResponse(encryption_failure=empty_pb2.Empty()) - - # security level has been reached ? - if await self.reached_security_level(connection, level): - return SecureResponse(success=empty_pb2.Empty()) - return SecureResponse(not_reached=empty_pb2.Empty()) - - @utils.rpc - async def WaitSecurity( - self, request: WaitSecurityRequest, context: grpc.ServicerContext - ) -> WaitSecurityResponse: - connection_handle = int.from_bytes(request.connection.cookie.value, 'big') - self.log.debug(f"WaitSecurity: {connection_handle}") - - connection = self.device.lookup_connection(connection_handle) - assert connection - - assert request.level - level = request.level - assert {PhysicalTransport.BR_EDR: 'classic', PhysicalTransport.LE: 'le'}[ - connection.transport - ] == request.level_variant() - - wait_for_security: asyncio.Future[str] = ( - asyncio.get_running_loop().create_future() - ) - authenticate_task: asyncio.Future[None] | None = None - pair_task: asyncio.Future[None] | None = None - - async def authenticate() -> None: - if (encryption := connection.encryption) != 0: - self.log.debug('Disable encryption...') - with contextlib.suppress(Exception): - await connection.encrypt(enable=False) - self.log.debug('Disable encryption: done') - - self.log.debug('Authenticate...') - await connection.authenticate() - self.log.debug('Authenticate: done') - - if encryption != 0 and connection.encryption != encryption: - self.log.debug('Re-enable encryption...') - await connection.encrypt() - self.log.debug('Re-enable encryption: done') - - def set_failure(name: str) -> Callable[..., None]: - def wrapper(*args: Any) -> None: - self.log.debug(f'Wait for security: error `{name}`: {args}') - wait_for_security.set_result(name) - - return wrapper - - async def try_set_success(*_: Any) -> None: - if await self.reached_security_level(connection, level): - self.log.debug('Wait for security: done') - wait_for_security.set_result('success') - - async def on_encryption_change(*_: Any) -> None: - if await self.reached_security_level(connection, level): - self.log.debug('Wait for security: done') - wait_for_security.set_result('success') - elif ( - connection.transport == PhysicalTransport.BR_EDR - and self.need_authentication(connection, level) - ): - nonlocal authenticate_task - if authenticate_task is None: - authenticate_task = asyncio.create_task(authenticate()) - - def pair(*_: Any) -> None: - if self.need_pairing(connection, level): - bumble.utils.AsyncRunner.spawn(connection.pair()) - - listeners: dict[str, Callable[..., None | Awaitable[None]]] = { - 'disconnection': set_failure('connection_died'), - 'pairing_failure': set_failure('pairing_failure'), - 'connection_authentication_failure': set_failure('authentication_failure'), - 'connection_encryption_failure': set_failure('encryption_failure'), - 'pairing': try_set_success, - 'connection_authentication': try_set_success, - 'connection_encryption_change': on_encryption_change, - 'classic_pairing': try_set_success, - 'classic_pairing_failure': set_failure('pairing_failure'), - 'security_request': pair, - } - - with contextlib.closing(bumble.utils.EventWatcher()) as watcher: - # register event handlers - for event, listener in listeners.items(): - watcher.on(connection, event, listener) - - # security level already reached - if await self.reached_security_level(connection, level): - return WaitSecurityResponse(success=empty_pb2.Empty()) - - self.log.debug('Wait for security...') - kwargs = {} - kwargs[await wait_for_security] = empty_pb2.Empty() - - # wait for `authenticate` to finish if any - if authenticate_task is not None: - self.log.debug('Wait for authentication...') - with contextlib.suppress(Exception): - await authenticate_task # type: ignore - self.log.debug('Authenticated') - - # wait for `pair` to finish if any - if pair_task is not None: - self.log.debug('Wait for authentication...') - with contextlib.suppress(Exception): - await pair_task # type: ignore - self.log.debug('paired') - - return WaitSecurityResponse(**kwargs) - - async def reached_security_level( - self, connection: BumbleConnection, level: SecurityLevel | LESecurityLevel - ) -> bool: - self.log.debug( - str( - { - 'level': level, - 'encryption': connection.encryption, - 'authenticated': connection.authenticated, - 'sc': connection.sc, - } - ) - ) - - if isinstance(level, LESecurityLevel): - return self._le_level_reached(level, connection) - - return await self._classic_level_reached(level, connection) - - def need_pairing(self, connection: BumbleConnection, level: int) -> bool: - if connection.transport == PhysicalTransport.LE: - return level >= LE_LEVEL3 and not connection.authenticated - return False - - def need_authentication(self, connection: BumbleConnection, level: int) -> bool: - if connection.transport == PhysicalTransport.LE: - return False - if level == LEVEL2 and connection.encryption != 0: - return not connection.authenticated - return level >= LEVEL2 and not connection.authenticated - - def need_encryption(self, connection: BumbleConnection, level: int) -> bool: - # TODO(abel): need to support MITM - if connection.transport == PhysicalTransport.LE: - return level == LE_LEVEL2 and not connection.encryption - return level >= LEVEL2 and not connection.encryption - - -class SecurityStorageService(SecurityStorageServicer): - def __init__(self, device: Device, config: Config) -> None: - self.log = utils.BumbleServerLoggerAdapter( - logging.getLogger(), {'service_name': 'SecurityStorage', 'device': device} - ) - self.device = device - self.config = config - - @utils.rpc - async def IsBonded( - self, request: IsBondedRequest, context: grpc.ServicerContext - ) -> wrappers_pb2.BoolValue: - address = utils.address_from_request(request, request.WhichOneof("address")) - self.log.debug(f"IsBonded: {address}") - - if self.device.keystore is not None: - is_bonded = await self.device.keystore.get(str(address)) is not None - else: - is_bonded = False - - return wrappers_pb2.BoolValue(value=is_bonded) - - @utils.rpc - async def DeleteBond( - self, request: DeleteBondRequest, context: grpc.ServicerContext - ) -> empty_pb2.Empty: - address = utils.address_from_request(request, request.WhichOneof("address")) - self.log.debug(f"DeleteBond: {address}") - - if self.device.keystore is not None: - with contextlib.suppress(KeyError): - await self.device.keystore.delete(str(address)) - - return empty_pb2.Empty() diff --git a/bumble/pandora/utils.py b/bumble/pandora/utils.py deleted file mode 100644 index 3faaae01a..000000000 --- a/bumble/pandora/utils.py +++ /dev/null @@ -1,115 +0,0 @@ -# Copyright 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. - -from __future__ import annotations - -import contextlib -import functools -import inspect -import logging -from collections.abc import Generator, MutableMapping -from typing import Any - -import grpc -from google.protobuf.message import Message # pytype: disable=pyi-error - -from bumble.device import Device -from bumble.hci import Address, AddressType - -ADDRESS_TYPES: dict[str, AddressType] = { - "public": Address.PUBLIC_DEVICE_ADDRESS, - "random": Address.RANDOM_DEVICE_ADDRESS, - "public_identity": Address.PUBLIC_IDENTITY_ADDRESS, - "random_static_identity": Address.RANDOM_IDENTITY_ADDRESS, -} - - -def address_from_request(request: Message, field: str | None) -> Address: - if field is None: - return Address.ANY - return Address(bytes(reversed(getattr(request, field))), ADDRESS_TYPES[field]) - - -class BumbleServerLoggerAdapter(logging.LoggerAdapter): # type: ignore - """Formats logs from the PandoraClient.""" - - def process( - self, msg: str, kwargs: MutableMapping[str, Any] - ) -> tuple[str, MutableMapping[str, Any]]: - assert self.extra - service_name = self.extra['service_name'] - assert isinstance(service_name, str) - device = self.extra['device'] - assert isinstance(device, Device) - addr_bytes = bytes( - reversed(bytes(device.public_address)) - ) # pytype: disable=attribute-error - addr = ':'.join([f'{x:02X}' for x in addr_bytes[4:]]) - return (f'[bumble.{service_name}:{addr}] {msg}', kwargs) - - -@contextlib.contextmanager -def exception_to_rpc_error( - context: grpc.ServicerContext, -) -> Generator[None, None, None]: - try: - yield None - except NotImplementedError as e: - context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore - context.set_details(str(e)) # type: ignore - except ValueError as e: - context.set_code(grpc.StatusCode.INVALID_ARGUMENT) # type: ignore - context.set_details(str(e)) # type: ignore - except RuntimeError as e: - context.set_code(grpc.StatusCode.ABORTED) # type: ignore - context.set_details(str(e)) # type: ignore - - -# Decorate an RPC servicer method with a wrapper that transform exceptions to gRPC errors. -def rpc(func: Any) -> Any: - @functools.wraps(func) - async def asyncgen_wrapper( - self: Any, request: Any, context: grpc.ServicerContext - ) -> Any: - with exception_to_rpc_error(context): - async for v in func(self, request, context): - yield v - - @functools.wraps(func) - async def async_wrapper( - self: Any, request: Any, context: grpc.ServicerContext - ) -> Any: - with exception_to_rpc_error(context): - return await func(self, request, context) - - @functools.wraps(func) - def gen_wrapper(self: Any, request: Any, context: grpc.ServicerContext) -> Any: - with exception_to_rpc_error(context): - yield from func(self, request, context) - - @functools.wraps(func) - def wrapper(self: Any, request: Any, context: grpc.ServicerContext) -> Any: - with exception_to_rpc_error(context): - return func(self, request, context) - - if inspect.isasyncgenfunction(func): - return asyncgen_wrapper - - if inspect.iscoroutinefunction(func): - return async_wrapper - - if inspect.isgenerator(func): - return gen_wrapper - - return wrapper diff --git a/pyproject.toml b/pyproject.toml index 9692286a8..cc2fb1af7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,11 +45,6 @@ android = [ "grpcio >= 1.62.1; platform_system!='Emscripten'", "protobuf >= 3.12.4; platform_system!='Emscripten'", ] -avatar = [ - "pandora-avatar == 0.0.10", - "rootcanal == 1.11.1 ; python_version>='3.10'", -] -pandora = ["bt-test-interfaces >= 0.0.6"] auracast = [ "lc3py >= 1.1.3; python_version>='3.10' and ((platform_system=='Linux' and platform_machine=='x86_64') or (platform_system=='Darwin' and platform_machine=='arm64'))", "sounddevice >= 0.5.1", @@ -77,7 +72,6 @@ dev = [ {include-group = "test"}, {include-group = "docs"}, "black ~= 25.1", - "bt-test-interfaces >= 0.0.6", "grpcio-tools >= 1.62.1", "invoke >= 1.7.3", "mobly >= 1.12.2", @@ -109,7 +103,6 @@ bumble-usb-probe = "bumble.apps.usb_probe:main" bumble-bench = "bumble.apps.bench:main" bumble-player = "bumble.apps.player.player:main" bumble-speaker = "bumble.apps.speaker.speaker:main" -bumble-pandora-server = "bumble.apps.pandora_server:main" bumble-rtk-util = "bumble.tools.rtk_util:main" bumble-rtk-fw-download = "bumble.tools.rtk_fw_download:main" bumble-intel-util = "bumble.tools.intel_util:main" @@ -127,7 +120,6 @@ packages = [ "bumble.drivers", "bumble.profiles", "bumble.apps", - "bumble.pandora", "bumble.tools", ] @@ -175,7 +167,7 @@ disable = [ ] [tool.pylint.main] -ignore=["pandora", "mobly"] # FIXME: pylint does not support stubs yet +ignore=["mobly"] # FIXME: pylint does not support stubs yet [tool.pylint.typecheck] signature-mutators = "AsyncRunner.run_in_task"