compatibility with recent host ACL property changes

apps: Add a controller loopback throughput test app

.gitignore

@@ -10,3 +10,5 @@ __pycache__
 bumble/_version.py
 .vscode/launch.json
 /.idea
+venv/
+.venv/

.vscode/settings.json

@@ -22,6 +22,7 @@
         "cmac",
         "CONNECTIONLESS",
         "csip",
+        "csis",
         "csrcs",
         "CVSD",
         "datagram",

apps/console.py

@@ -777,7 +777,7 @@ class ConsoleApp:
                 if not service:
                     continue
                 values = [
-                    attribute.read_value(connection)
+                    await attribute.read_value(connection)
                     for connection in self.device.connections.values()
                 ]
                 if not values:
@@ -796,11 +796,11 @@ class ConsoleApp:
                 if not characteristic:
                     continue
                 values = [
-                    attribute.read_value(connection)
+                    await attribute.read_value(connection)
                     for connection in self.device.connections.values()
                 ]
                 if not values:
-                    values = [attribute.read_value(None)]
+                    values = [await attribute.read_value(None)]
 
                 # TODO: future optimization: convert CCCD value to human readable string
 
@@ -944,7 +944,7 @@ class ConsoleApp:
 
         # send data to any subscribers
         if isinstance(attribute, Characteristic):
-            attribute.write_value(None, value)
+            await attribute.write_value(None, value)
             if attribute.has_properties(Characteristic.NOTIFY):
                 await self.device.gatt_server.notify_subscribers(attribute)
             if attribute.has_properties(Characteristic.INDICATE):

apps/controller_loopback.py

@@ -0,0 +1,200 @@
+# 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.
+
+# -----------------------------------------------------------------------------
+# Imports
+# -----------------------------------------------------------------------------
+import asyncio
+import logging
+import os
+import time
+from typing import Optional
+from bumble.colors import color
+from bumble.hci import (
+    HCI_READ_LOOPBACK_MODE_COMMAND,
+    HCI_Read_Loopback_Mode_Command,
+    HCI_WRITE_LOOPBACK_MODE_COMMAND,
+    HCI_Write_Loopback_Mode_Command,
+    LoopbackMode,
+)
+from bumble.host import Host
+from bumble.transport import open_transport_or_link
+import click
+
+
+class Loopback:
+    """Send and receive ACL data packets in local loopback mode"""
+
+    def __init__(self, packet_size: int, packet_count: int, transport: str):
+        self.transport = transport
+        self.packet_size = packet_size
+        self.packet_count = packet_count
+        self.connection_handle: Optional[int] = None
+        self.connection_event = asyncio.Event()
+        self.done = asyncio.Event()
+        self.expected_cid = 0
+        self.bytes_received = 0
+        self.start_timestamp = 0.0
+        self.last_timestamp = 0.0
+
+    def on_connection(self, connection_handle: int, *args):
+        """Retrieve connection handle from new connection event"""
+        if not self.connection_event.is_set():
+            # save first connection handle for ACL
+            # subsequent connections are SCO
+            self.connection_handle = connection_handle
+            self.connection_event.set()
+
+    def on_l2cap_pdu(self, connection_handle: int, cid: int, pdu: bytes):
+        """Calculate packet receive speed"""
+        now = time.time()
+        print(f'<<< Received packet {cid}: {len(pdu)} bytes')
+        assert connection_handle == self.connection_handle
+        assert cid == self.expected_cid
+        self.expected_cid += 1
+        if cid == 0:
+            self.start_timestamp = now
+        else:
+            elapsed_since_start = now - self.start_timestamp
+            elapsed_since_last = now - self.last_timestamp
+            self.bytes_received += len(pdu)
+            instant_rx_speed = len(pdu) / elapsed_since_last
+            average_rx_speed = self.bytes_received / elapsed_since_start
+            print(
+                color(
+                    f'@@@ RX speed: instant={instant_rx_speed:.4f},'
+                    f' average={average_rx_speed:.4f}',
+                    'cyan',
+                )
+            )
+
+        self.last_timestamp = now
+
+        if self.expected_cid == self.packet_count:
+            print(color('@@@ Received last packet', 'green'))
+            self.done.set()
+
+    async def run(self):
+        """Run a loopback throughput test"""
+        print(color('>>> Connecting to HCI...', 'green'))
+        async with await open_transport_or_link(self.transport) as (
+            hci_source,
+            hci_sink,
+        ):
+            print(color('>>> Connected', 'green'))
+
+            host = Host(hci_source, hci_sink)
+            await host.reset()
+
+            # make sure data can fit in one l2cap pdu
+            l2cap_header_size = 4
+            max_packet_size = host.acl_packet_queue.max_packet_size - l2cap_header_size
+            if self.packet_size > max_packet_size:
+                print(
+                    color(
+                        f'!!! Packet size ({self.packet_size}) larger than max supported'
+                        f' size ({max_packet_size})',
+                        'red',
+                    )
+                )
+                return
+
+            if not host.supports_command(
+                HCI_WRITE_LOOPBACK_MODE_COMMAND
+            ) or not host.supports_command(HCI_READ_LOOPBACK_MODE_COMMAND):
+                print(color('!!! Loopback mode not supported', 'red'))
+                return
+
+            # set event callbacks
+            host.on('connection', self.on_connection)
+            host.on('l2cap_pdu', self.on_l2cap_pdu)
+
+            loopback_mode = LoopbackMode.LOCAL
+
+            print(color('### Setting loopback mode', 'blue'))
+            await host.send_command(
+                HCI_Write_Loopback_Mode_Command(loopback_mode=LoopbackMode.LOCAL),
+                check_result=True,
+            )
+
+            print(color('### Checking loopback mode', 'blue'))
+            response = await host.send_command(
+                HCI_Read_Loopback_Mode_Command(), check_result=True
+            )
+            if response.return_parameters.loopback_mode != loopback_mode:
+                print(color('!!! Loopback mode mismatch', 'red'))
+                return
+
+            await self.connection_event.wait()
+            print(color('### Connected', 'cyan'))
+
+            print(color('=== Start sending', 'magenta'))
+            start_time = time.time()
+            bytes_sent = 0
+            for cid in range(0, self.packet_count):
+                # using the cid as an incremental index
+                host.send_l2cap_pdu(
+                    self.connection_handle, cid, bytes(self.packet_size)
+                )
+                print(
+                    color(
+                        f'>>> Sending packet {cid}: {self.packet_size} bytes', 'yellow'
+                    )
+                )
+                bytes_sent += self.packet_size  # don't count L2CAP or HCI header sizes
+                await asyncio.sleep(0)  # yield to allow packet receive
+
+            await self.done.wait()
+            print(color('=== Done!', 'magenta'))
+
+            elapsed = time.time() - start_time
+            average_tx_speed = bytes_sent / elapsed
+            print(
+                color(
+                    f'@@@ TX speed: average={average_tx_speed:.4f} ({bytes_sent} bytes'
+                    f' in {elapsed:.2f} seconds)',
+                    'green',
+                )
+            )
+
+
+# -----------------------------------------------------------------------------
+@click.command()
+@click.option(
+    '--packet-size',
+    '-s',
+    metavar='SIZE',
+    type=click.IntRange(8, 4096),
+    default=500,
+    help='Packet size',
+)
+@click.option(
+    '--packet-count',
+    '-c',
+    metavar='COUNT',
+    type=int,
+    default=10,
+    help='Packet count',
+)
+@click.argument('transport')
+def main(packet_size, packet_count, transport):
+    logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'WARNING').upper())
+
+    loopback = Loopback(packet_size, packet_count, transport)
+    asyncio.run(loopback.run())
+
+
+# -----------------------------------------------------------------------------
+if __name__ == '__main__':
+    main()

bumble/att.py

@@ -25,9 +25,21 @@
 from __future__ import annotations
 import enum
 import functools
+import inspect
 import struct
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Type,
+    Union,
+    TYPE_CHECKING,
+)
+
 from pyee import EventEmitter
-from typing import Dict, Type, List, Protocol, Union, Optional, Any, TYPE_CHECKING
 
 from bumble.core import UUID, name_or_number, ProtocolError
 from bumble.hci import HCI_Object, key_with_value
@@ -722,12 +734,38 @@ class ATT_Handle_Value_Confirmation(ATT_PDU):
 
 
 # -----------------------------------------------------------------------------
-class ConnectionValue(Protocol):
-    def read(self, connection) -> bytes:
-        ...
+class AttributeValue:
+    '''
+    Attribute value where reading and/or writing is delegated to functions
+    passed as arguments to the constructor.
+    '''
 
-    def write(self, connection, value: bytes) -> None:
-        ...
+    def __init__(
+        self,
+        read: Union[
+            Callable[[Optional[Connection]], bytes],
+            Callable[[Optional[Connection]], Awaitable[bytes]],
+            None,
+        ] = None,
+        write: Union[
+            Callable[[Optional[Connection], bytes], None],
+            Callable[[Optional[Connection], bytes], Awaitable[None]],
+            None,
+        ] = None,
+    ):
+        self._read = read
+        self._write = write
+
+    def read(self, connection: Optional[Connection]) -> Union[bytes, Awaitable[bytes]]:
+        return self._read(connection) if self._read else b''
+
+    def write(
+        self, connection: Optional[Connection], value: bytes
+    ) -> Union[Awaitable[None], None]:
+        if self._write:
+            return self._write(connection, value)
+
+        return None
 
 
 # -----------------------------------------------------------------------------
@@ -770,13 +808,13 @@ class Attribute(EventEmitter):
     READ_REQUIRES_AUTHORIZATION = Permissions.READ_REQUIRES_AUTHORIZATION
     WRITE_REQUIRES_AUTHORIZATION = Permissions.WRITE_REQUIRES_AUTHORIZATION
 
-    value: Union[str, bytes, ConnectionValue]
+    value: Union[bytes, AttributeValue]
 
     def __init__(
         self,
         attribute_type: Union[str, bytes, UUID],
         permissions: Union[str, Attribute.Permissions],
-        value: Union[str, bytes, ConnectionValue] = b'',
+        value: Union[str, bytes, AttributeValue] = b'',
     ) -> None:
         EventEmitter.__init__(self)
         self.handle = 0
@@ -806,7 +844,7 @@ class Attribute(EventEmitter):
     def decode_value(self, value_bytes: bytes) -> Any:
         return value_bytes
 
-    def read_value(self, connection: Optional[Connection]) -> bytes:
+    async def read_value(self, connection: Optional[Connection]) -> bytes:
         if (
             (self.permissions & self.READ_REQUIRES_ENCRYPTION)
             and connection is not None
@@ -832,6 +870,8 @@ class Attribute(EventEmitter):
         if hasattr(self.value, 'read'):
             try:
                 value = self.value.read(connection)
+                if inspect.isawaitable(value):
+                    value = await value
             except ATT_Error as error:
                 raise ATT_Error(
                     error_code=error.error_code, att_handle=self.handle
@@ -841,7 +881,7 @@ class Attribute(EventEmitter):
 
         return self.encode_value(value)
 
-    def write_value(self, connection: Connection, value_bytes: bytes) -> None:
+    async def write_value(self, connection: Connection, value_bytes: bytes) -> None:
         if (
             self.permissions & self.WRITE_REQUIRES_ENCRYPTION
         ) and not connection.encryption:
@@ -864,7 +904,9 @@ class Attribute(EventEmitter):
 
         if hasattr(self.value, 'write'):
             try:
-                self.value.write(connection, value)  # pylint: disable=not-callable
+                result = self.value.write(connection, value)
+                if inspect.isawaitable(result):
+                    await result
             except ATT_Error as error:
                 raise ATT_Error(
                     error_code=error.error_code, att_handle=self.handle

bumble/gatt.py

@@ -23,16 +23,28 @@
 # Imports
 # -----------------------------------------------------------------------------
 from __future__ import annotations
-import asyncio
 import enum
 import functools
 import logging
 import struct
-from typing import Optional, Sequence, Iterable, List, Union
+from typing import (
+    Callable,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Sequence,
+    Union,
+    TYPE_CHECKING,
+)
 
-from .colors import color
-from .core import UUID, get_dict_key_by_value
-from .att import Attribute
+from bumble.colors import color
+from bumble.core import UUID
+from bumble.att import Attribute, AttributeValue
+
+if TYPE_CHECKING:
+    from bumble.gatt_client import AttributeProxy
+    from bumble.device import Connection
 
 
 # -----------------------------------------------------------------------------
@@ -522,56 +534,43 @@ class CharacteristicDeclaration(Attribute):
 
 
 # -----------------------------------------------------------------------------
-class CharacteristicValue:
-    '''
-    Characteristic value where reading and/or writing is delegated to functions
-    passed as arguments to the constructor.
-    '''
-
-    def __init__(self, read=None, write=None):
-        self._read = read
-        self._write = write
-
-    def read(self, connection):
-        return self._read(connection) if self._read else b''
-
-    def write(self, connection, value):
-        if self._write:
-            self._write(connection, value)
+class CharacteristicValue(AttributeValue):
+    """Same as AttributeValue, for backward compatibility"""
 
 
 # -----------------------------------------------------------------------------
 class CharacteristicAdapter:
     '''
-    An adapter that can adapt any object with `read_value` and `write_value`
-    methods (like Characteristic and CharacteristicProxy objects) by wrapping
-    those methods with ones that return/accept encoded/decoded values.
-    Objects with async methods are considered proxies, so the adaptation is one
-    where the return value of `read_value` is decoded and the value passed to
-    `write_value` is encoded. Other objects are considered local characteristics
-    so the adaptation is one where the return value of `read_value` is encoded
-    and the value passed to `write_value` is decoded.
-    If the characteristic has a `subscribe` method, it is wrapped with one where
-    the values are decoded before being passed to the subscriber.
+    An adapter that can adapt Characteristic and AttributeProxy objects
+    by wrapping their `read_value()` and `write_value()` methods with ones that
+    return/accept encoded/decoded values.
+
+    For proxies (i.e used by a GATT client), the adaptation is one where the return
+    value of `read_value()` is decoded and the value passed to `write_value()` is
+    encoded. The `subscribe()` method, is wrapped with one where the values are decoded
+    before being passed to the subscriber.
+
+    For local values (i.e hosted by a GATT server) the adaptation is one where the
+    return value of `read_value()` is encoded and the value passed to `write_value()`
+    is decoded.
     '''
 
-    def __init__(self, characteristic):
-        self.wrapped_characteristic = characteristic
-        self.subscribers = {}  # Map from subscriber to proxy subscriber
+    read_value: Callable
+    write_value: Callable
 
-        if asyncio.iscoroutinefunction(
-            characteristic.read_value
-        ) and asyncio.iscoroutinefunction(characteristic.write_value):
-            self.read_value = self.read_decoded_value
-            self.write_value = self.write_decoded_value
-        else:
+    def __init__(self, characteristic: Union[Characteristic, AttributeProxy]):
+        self.wrapped_characteristic = characteristic
+        self.subscribers: Dict[
+            Callable, Callable
+        ] = {}  # Map from subscriber to proxy subscriber
+
+        if isinstance(characteristic, Characteristic):
             self.read_value = self.read_encoded_value
             self.write_value = self.write_encoded_value
-
-        if hasattr(self.wrapped_characteristic, 'subscribe'):
+        else:
+            self.read_value = self.read_decoded_value
+            self.write_value = self.write_decoded_value
             self.subscribe = self.wrapped_subscribe
-
-        if hasattr(self.wrapped_characteristic, 'unsubscribe'):
             self.unsubscribe = self.wrapped_unsubscribe
 
     def __getattr__(self, name):
@@ -590,11 +589,13 @@ class CharacteristicAdapter:
         else:
             setattr(self.wrapped_characteristic, name, value)
 
-    def read_encoded_value(self, connection):
-        return self.encode_value(self.wrapped_characteristic.read_value(connection))
+    async def read_encoded_value(self, connection):
+        return self.encode_value(
+            await self.wrapped_characteristic.read_value(connection)
+        )
 
-    def write_encoded_value(self, connection, value):
-        return self.wrapped_characteristic.write_value(
+    async def write_encoded_value(self, connection, value):
+        return await self.wrapped_characteristic.write_value(
             connection, self.decode_value(value)
         )
 
@@ -729,13 +730,24 @@ class Descriptor(Attribute):
     '''
 
     def __str__(self) -> str:
+        if isinstance(self.value, bytes):
+            value_str = self.value.hex()
+        elif isinstance(self.value, CharacteristicValue):
+            value = self.value.read(None)
+            if isinstance(value, bytes):
+                value_str = value.hex()
+            else:
+                value_str = '<async>'
+        else:
+            value_str = '<...>'
         return (
             f'Descriptor(handle=0x{self.handle:04X}, '
             f'type={self.type}, '
-            f'value={self.read_value(None).hex()})'
+            f'value={value_str})'
         )
 
 
+# -----------------------------------------------------------------------------
 class ClientCharacteristicConfigurationBits(enum.IntFlag):
     '''
     See Vol 3, Part G - 3.3.3.3 - Table 3.11 Client Characteristic Configuration bit

bumble/gatt_server.py

@@ -31,9 +31,9 @@ import struct
 from typing import List, Tuple, Optional, TypeVar, Type, Dict, Iterable, TYPE_CHECKING
 from pyee import EventEmitter
 
-from .colors import color
-from .core import UUID
-from .att import (
+from bumble.colors import color
+from bumble.core import UUID
+from bumble.att import (
     ATT_ATTRIBUTE_NOT_FOUND_ERROR,
     ATT_ATTRIBUTE_NOT_LONG_ERROR,
     ATT_CID,
@@ -60,7 +60,7 @@ from .att import (
     ATT_Write_Response,
     Attribute,
 )
-from .gatt import (
+from bumble.gatt import (
     GATT_CHARACTERISTIC_ATTRIBUTE_TYPE,
     GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR,
     GATT_MAX_ATTRIBUTE_VALUE_SIZE,
@@ -74,6 +74,7 @@ from .gatt import (
     Descriptor,
     Service,
 )
+from bumble.utils import AsyncRunner
 
 if TYPE_CHECKING:
     from bumble.device import Device, Connection
@@ -379,7 +380,7 @@ class Server(EventEmitter):
 
         # Get or encode the value
         value = (
-            attribute.read_value(connection)
+            await attribute.read_value(connection)
             if value is None
             else attribute.encode_value(value)
         )
@@ -422,7 +423,7 @@ class Server(EventEmitter):
 
         # Get or encode the value
         value = (
-            attribute.read_value(connection)
+            await attribute.read_value(connection)
             if value is None
             else attribute.encode_value(value)
         )
@@ -650,7 +651,8 @@ class Server(EventEmitter):
 
         self.send_response(connection, response)
 
-    def on_att_find_by_type_value_request(self, connection, request):
+    @AsyncRunner.run_in_task()
+    async def on_att_find_by_type_value_request(self, connection, request):
         '''
         See Bluetooth spec Vol 3, Part F - 3.4.3.3 Find By Type Value Request
         '''
@@ -658,13 +660,13 @@ class Server(EventEmitter):
         # Build list of returned attributes
         pdu_space_available = connection.att_mtu - 2
         attributes = []
-        for attribute in (
+        async for attribute in (
             attribute
             for attribute in self.attributes
             if attribute.handle >= request.starting_handle
             and attribute.handle <= request.ending_handle
             and attribute.type == request.attribute_type
-            and attribute.read_value(connection) == request.attribute_value
+            and (await attribute.read_value(connection)) == request.attribute_value
             and pdu_space_available >= 4
         ):
             # TODO: check permissions
@@ -702,7 +704,8 @@ class Server(EventEmitter):
 
         self.send_response(connection, response)
 
-    def on_att_read_by_type_request(self, connection, request):
+    @AsyncRunner.run_in_task()
+    async def on_att_read_by_type_request(self, connection, request):
         '''
         See Bluetooth spec Vol 3, Part F - 3.4.4.1 Read By Type Request
         '''
@@ -725,7 +728,7 @@ class Server(EventEmitter):
             and pdu_space_available
         ):
             try:
-                attribute_value = attribute.read_value(connection)
+                attribute_value = await attribute.read_value(connection)
             except ATT_Error as error:
                 # If the first attribute is unreadable, return an error
                 # Otherwise return attributes up to this point
@@ -767,14 +770,15 @@ class Server(EventEmitter):
 
         self.send_response(connection, response)
 
-    def on_att_read_request(self, connection, request):
+    @AsyncRunner.run_in_task()
+    async def on_att_read_request(self, connection, request):
         '''
         See Bluetooth spec Vol 3, Part F - 3.4.4.3 Read Request
         '''
 
         if attribute := self.get_attribute(request.attribute_handle):
             try:
-                value = attribute.read_value(connection)
+                value = await attribute.read_value(connection)
             except ATT_Error as error:
                 response = ATT_Error_Response(
                     request_opcode_in_error=request.op_code,
@@ -792,14 +796,15 @@ class Server(EventEmitter):
             )
         self.send_response(connection, response)
 
-    def on_att_read_blob_request(self, connection, request):
+    @AsyncRunner.run_in_task()
+    async def on_att_read_blob_request(self, connection, request):
         '''
         See Bluetooth spec Vol 3, Part F - 3.4.4.5 Read Blob Request
         '''
 
         if attribute := self.get_attribute(request.attribute_handle):
             try:
-                value = attribute.read_value(connection)
+                value = await attribute.read_value(connection)
             except ATT_Error as error:
                 response = ATT_Error_Response(
                     request_opcode_in_error=request.op_code,
@@ -836,7 +841,8 @@ class Server(EventEmitter):
             )
         self.send_response(connection, response)
 
-    def on_att_read_by_group_type_request(self, connection, request):
+    @AsyncRunner.run_in_task()
+    async def on_att_read_by_group_type_request(self, connection, request):
         '''
         See Bluetooth spec Vol 3, Part F - 3.4.4.9 Read by Group Type Request
         '''
@@ -864,7 +870,7 @@ class Server(EventEmitter):
         ):
             # No need to catch permission errors here, since these attributes
             # must all be world-readable
-            attribute_value = attribute.read_value(connection)
+            attribute_value = await attribute.read_value(connection)
             # Check the attribute value size
             max_attribute_size = min(connection.att_mtu - 6, 251)
             if len(attribute_value) > max_attribute_size:
@@ -903,7 +909,8 @@ class Server(EventEmitter):
 
         self.send_response(connection, response)
 
-    def on_att_write_request(self, connection, request):
+    @AsyncRunner.run_in_task()
+    async def on_att_write_request(self, connection, request):
         '''
         See Bluetooth spec Vol 3, Part F - 3.4.5.1 Write Request
         '''
@@ -936,12 +943,13 @@ class Server(EventEmitter):
             return
 
         # Accept the value
-        attribute.write_value(connection, request.attribute_value)
+        await attribute.write_value(connection, request.attribute_value)
 
         # Done
         self.send_response(connection, ATT_Write_Response())
 
-    def on_att_write_command(self, connection, request):
+    @AsyncRunner.run_in_task()
+    async def on_att_write_command(self, connection, request):
         '''
         See Bluetooth spec Vol 3, Part F - 3.4.5.3 Write Command
         '''
@@ -959,7 +967,7 @@ class Server(EventEmitter):
 
         # Accept the value
         try:
-            attribute.write_value(connection, request.attribute_value)
+            await attribute.write_value(connection, request.attribute_value)
         except Exception as error:
             logger.exception(f'!!! ignoring exception: {error}')
 

bumble/hci.py

@@ -2026,6 +2026,17 @@ class OwnAddressType(enum.IntEnum):
         return {'size': 1, 'mapper': lambda x: OwnAddressType(x).name}
 
 
+# -----------------------------------------------------------------------------
+class LoopbackMode(enum.IntEnum):
+    DISABLED = 0
+    LOCAL = 1
+    REMOTE = 2
+
+    @classmethod
+    def type_spec(cls):
+        return {'size': 1, 'mapper': lambda x: LoopbackMode(x).name}
+
+
 # -----------------------------------------------------------------------------
 class HCI_Packet:
     '''
@@ -3352,6 +3363,27 @@ class HCI_Read_Encryption_Key_Size_Command(HCI_Command):
     '''
 
 
+# -----------------------------------------------------------------------------
+@HCI_Command.command(
+    return_parameters_fields=[
+        ('status', STATUS_SPEC),
+        ('loopback_mode', LoopbackMode.type_spec()),
+    ],
+)
+class HCI_Read_Loopback_Mode_Command(HCI_Command):
+    '''
+    See Bluetooth spec @ 7.6.1 Read Loopback Mode Command
+    '''
+
+
+# -----------------------------------------------------------------------------
+@HCI_Command.command([('loopback_mode', 1)])
+class HCI_Write_Loopback_Mode_Command(HCI_Command):
+    '''
+    See Bluetooth spec @ 7.6.2 Write Loopback Mode Command
+    '''
+
+
 # -----------------------------------------------------------------------------
 @HCI_Command.command([('le_event_mask', 8)])
 class HCI_LE_Set_Event_Mask_Command(HCI_Command):

bumble/profiles/asha_service.py

@@ -18,7 +18,7 @@
 # -----------------------------------------------------------------------------
 import struct
 import logging
-from typing import List
+from typing import List, Optional
 
 from bumble import l2cap
 from ..core import AdvertisingData
@@ -67,7 +67,7 @@ class AshaService(TemplateService):
             self.emit('volume', connection, value[0])
 
         # Handler for audio control commands
-        def on_audio_control_point_write(connection: Connection, value):
+        def on_audio_control_point_write(connection: Optional[Connection], value):
             logger.info(f'--- AUDIO CONTROL POINT Write:{value.hex()}')
             opcode = value[0]
             if opcode == AshaService.OPCODE_START:

bumble/profiles/bap.py

@@ -114,7 +114,7 @@ class SamplingFrequency(enum.IntEnum):
     '''Bluetooth Assigned Numbers, Section 6.12.5.1 - Sampling Frequency'''
 
     # fmt: off
-    FREQ_8000    = 0x01 
+    FREQ_8000    = 0x01
     FREQ_11025   = 0x02
     FREQ_16000   = 0x03
     FREQ_22050   = 0x04
@@ -430,7 +430,7 @@ class AseResponseCode(enum.IntEnum):
     REJECTED_METADATA                           = 0x0B
     INVALID_METADATA                            = 0x0C
     INSUFFICIENT_RESOURCES                      = 0x0D
-    UNSPECIFIED_ERROR                           = 0x0E    
+    UNSPECIFIED_ERROR                           = 0x0E
 
 
 class AseReasonCode(enum.IntEnum):
@@ -1066,7 +1066,7 @@ class AseStateMachine(gatt.Characteristic):
         # Readonly. Do nothing in the setter.
         pass
 
-    def on_read(self, _: device.Connection) -> bytes:
+    def on_read(self, _: Optional[device.Connection]) -> bytes:
         return self.value
 
     def __str__(self) -> str:

bumble/profiles/csip.py

@@ -19,7 +19,7 @@
 from __future__ import annotations
 import enum
 import struct
-from typing import Optional
+from typing import Optional, Tuple
 
 from bumble import core
 from bumble import crypto
@@ -31,6 +31,9 @@ from bumble import gatt_client
 # -----------------------------------------------------------------------------
 # Constants
 # -----------------------------------------------------------------------------
+SET_IDENTITY_RESOLVING_KEY_LENGTH = 16
+
+
 class SirkType(enum.IntEnum):
     '''Coordinated Set Identification Service - 5.1 Set Identity Resolving Key.'''
 
@@ -66,6 +69,10 @@ def k1(n: bytes, salt: bytes, p: bytes) -> bytes:
 def sef(k: bytes, r: bytes) -> bytes:
     '''
     Coordinated Set Identification Service - 4.5 SIRK encryption function sef.
+
+    SIRK decryption function sdf shares the same algorithm. The only difference is that argument r is:
+      * Plaintext in encryption
+      * Cipher in decryption
     '''
     return crypto.xor(k1(k, s1(b'SIRKenc'[::-1]), b'csis'[::-1]), r)
 
@@ -105,6 +112,11 @@ class CoordinatedSetIdentificationService(gatt.TemplateService):
         set_member_lock: Optional[MemberLock] = None,
         set_member_rank: Optional[int] = None,
     ) -> None:
+        if len(set_identity_resolving_key) != SET_IDENTITY_RESOLVING_KEY_LENGTH:
+            raise ValueError(
+                f'Invalid SIRK length {len(set_identity_resolving_key)}, expected {SET_IDENTITY_RESOLVING_KEY_LENGTH}'
+            )
+
         characteristics = []
 
         self.set_identity_resolving_key = set_identity_resolving_key
@@ -113,7 +125,7 @@ class CoordinatedSetIdentificationService(gatt.TemplateService):
             uuid=gatt.GATT_SET_IDENTITY_RESOLVING_KEY_CHARACTERISTIC,
             properties=gatt.Characteristic.Properties.READ
             | gatt.Characteristic.Properties.NOTIFY,
-            permissions=gatt.Characteristic.Permissions.READABLE,
+            permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
             value=gatt.CharacteristicValue(read=self.on_sirk_read),
         )
         characteristics.append(self.set_identity_resolving_key_characteristic)
@@ -123,7 +135,7 @@ class CoordinatedSetIdentificationService(gatt.TemplateService):
                 uuid=gatt.GATT_COORDINATED_SET_SIZE_CHARACTERISTIC,
                 properties=gatt.Characteristic.Properties.READ
                 | gatt.Characteristic.Properties.NOTIFY,
-                permissions=gatt.Characteristic.Permissions.READABLE,
+                permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
                 value=struct.pack('B', coordinated_set_size),
             )
             characteristics.append(self.coordinated_set_size_characteristic)
@@ -134,7 +146,7 @@ class CoordinatedSetIdentificationService(gatt.TemplateService):
                 properties=gatt.Characteristic.Properties.READ
                 | gatt.Characteristic.Properties.NOTIFY
                 | gatt.Characteristic.Properties.WRITE,
-                permissions=gatt.Characteristic.Permissions.READABLE
+                permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION
                 | gatt.Characteristic.Permissions.WRITEABLE,
                 value=struct.pack('B', set_member_lock),
             )
@@ -145,18 +157,32 @@ class CoordinatedSetIdentificationService(gatt.TemplateService):
                 uuid=gatt.GATT_SET_MEMBER_RANK_CHARACTERISTIC,
                 properties=gatt.Characteristic.Properties.READ
                 | gatt.Characteristic.Properties.NOTIFY,
-                permissions=gatt.Characteristic.Permissions.READABLE,
+                permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
                 value=struct.pack('B', set_member_rank),
             )
             characteristics.append(self.set_member_rank_characteristic)
 
         super().__init__(characteristics)
 
-    def on_sirk_read(self, _connection: device.Connection) -> bytes:
+    async def on_sirk_read(self, connection: Optional[device.Connection]) -> bytes:
         if self.set_identity_resolving_key_type == SirkType.PLAINTEXT:
-            return bytes([SirkType.PLAINTEXT]) + self.set_identity_resolving_key
+            sirk_bytes = self.set_identity_resolving_key
         else:
-            raise NotImplementedError('TODO: Pending async Characteristic read.')
+            assert connection
+
+            if connection.transport == core.BT_LE_TRANSPORT:
+                key = await connection.device.get_long_term_key(
+                    connection_handle=connection.handle, rand=b'', ediv=0
+                )
+            else:
+                key = await connection.device.get_link_key(connection.peer_address)
+
+            if not key:
+                raise RuntimeError('LTK or LinkKey is not present')
+
+            sirk_bytes = sef(key, self.set_identity_resolving_key)
+
+        return bytes([self.set_identity_resolving_key_type]) + sirk_bytes
 
     def get_advertising_data(self) -> bytes:
         return bytes(
@@ -203,3 +229,29 @@ class CoordinatedSetIdentificationProxy(gatt_client.ProfileServiceProxy):
             gatt.GATT_SET_MEMBER_RANK_CHARACTERISTIC
         ):
             self.set_member_rank = characteristics[0]
+
+    async def read_set_identity_resolving_key(self) -> Tuple[SirkType, bytes]:
+        '''Reads SIRK and decrypts if encrypted.'''
+        response = await self.set_identity_resolving_key.read_value()
+        if len(response) != SET_IDENTITY_RESOLVING_KEY_LENGTH + 1:
+            raise RuntimeError('Invalid SIRK value')
+
+        sirk_type = SirkType(response[0])
+        if sirk_type == SirkType.PLAINTEXT:
+            sirk = response[1:]
+        else:
+            connection = self.service_proxy.client.connection
+            device = connection.device
+            if connection.transport == core.BT_LE_TRANSPORT:
+                key = await device.get_long_term_key(
+                    connection_handle=connection.handle, rand=b'', ediv=0
+                )
+            else:
+                key = await device.get_link_key(connection.peer_address)
+
+            if not key:
+                raise RuntimeError('LTK or LinkKey is not present')
+
+            sirk = sef(key, response[1:])
+
+        return (sirk_type, sirk)

bumble/utils.py

@@ -280,17 +280,14 @@ class AsyncRunner:
             def wrapper(*args, **kwargs):
                 coroutine = func(*args, **kwargs)
                 if queue is None:
-                    # Create a task to run the coroutine
+                    # Spawn the coroutine as a task
                     async def run():
                         try:
                             await coroutine
                         except Exception:
-                            logger.warning(
-                                f'{color("!!! Exception in wrapper:", "red")} '
-                                f'{traceback.format_exc()}'
-                            )
+                            logger.exception(color("!!! Exception in wrapper:", "red"))
 
-                    asyncio.create_task(run())
+                    AsyncRunner.spawn(run())
                 else:
                     # Queue the coroutine to be awaited by the work queue
                     queue.enqueue(coroutine)

tests/bap_test.py

@@ -48,7 +48,8 @@ from bumble.profiles.bap import (
     PublishedAudioCapabilitiesService,
     PublishedAudioCapabilitiesServiceProxy,
 )
-from .test_utils import TwoDevices
+from tests.test_utils import TwoDevices
+
 
 # -----------------------------------------------------------------------------
 # Logging

tests/csip_test.py

@@ -20,6 +20,7 @@ import os
 import pytest
 import struct
 import logging
+from unittest import mock
 
 from bumble import device
 from bumble.profiles import csip
@@ -68,14 +69,18 @@ def test_sef():
 
 # -----------------------------------------------------------------------------
 @pytest.mark.asyncio
-async def test_csis():
+@pytest.mark.parametrize(
+    'sirk_type,', [(csip.SirkType.ENCRYPTED), (csip.SirkType.PLAINTEXT)]
+)
+async def test_csis(sirk_type):
     SIRK = bytes.fromhex('2f62c8ae41867d1bb619e788a2605faa')
+    LTK = bytes.fromhex('2f62c8ae41867d1bb619e788a2605faa')
 
     devices = TwoDevices()
     devices[0].add_service(
         csip.CoordinatedSetIdentificationService(
             set_identity_resolving_key=SIRK,
-            set_identity_resolving_key_type=csip.SirkType.PLAINTEXT,
+            set_identity_resolving_key_type=sirk_type,
             coordinated_set_size=2,
             set_member_lock=csip.MemberLock.UNLOCKED,
             set_member_rank=0,
@@ -83,15 +88,19 @@ async def test_csis():
     )
 
     await devices.setup_connection()
+
+    # Mock encryption.
+    devices.connections[0].encryption = 1
+    devices.connections[1].encryption = 1
+    devices[0].get_long_term_key = mock.AsyncMock(return_value=LTK)
+    devices[1].get_long_term_key = mock.AsyncMock(return_value=LTK)
+
     peer = device.Peer(devices.connections[1])
     csis_client = await peer.discover_service_and_create_proxy(
         csip.CoordinatedSetIdentificationProxy
     )
 
-    assert (
-        await csis_client.set_identity_resolving_key.read_value()
-        == bytes([csip.SirkType.PLAINTEXT]) + SIRK
-    )
+    assert await csis_client.read_set_identity_resolving_key() == (sirk_type, SIRK)
     assert await csis_client.coordinated_set_size.read_value() == struct.pack('B', 2)
     assert await csis_client.set_member_lock.read_value() == struct.pack(
         'B', csip.MemberLock.UNLOCKED

tests/gatt_test.py

@@ -20,11 +20,10 @@ import logging
 import os
 import struct
 import pytest
-from unittest.mock import Mock, ANY
+from unittest.mock import AsyncMock, Mock, ANY
 
 from bumble.controller import Controller
 from bumble.gatt_client import CharacteristicProxy
-from bumble.gatt_server import Server
 from bumble.link import LocalLink
 from bumble.device import Device, Peer
 from bumble.host import Host
@@ -120,9 +119,9 @@ async def test_characteristic_encoding():
         Characteristic.READABLE,
         123,
     )
-    x = c.read_value(None)
+    x = await c.read_value(None)
     assert x == bytes([123])
-    c.write_value(None, bytes([122]))
+    await c.write_value(None, bytes([122]))
     assert c.value == 122
 
     class FooProxy(CharacteristicProxy):
@@ -152,7 +151,22 @@ async def test_characteristic_encoding():
         bytes([123]),
     )
 
-    service = Service('3A657F47-D34F-46B3-B1EC-698E29B6B829', [characteristic])
+    async def async_read(connection):
+        return 0x05060708
+
+    async_characteristic = PackedCharacteristicAdapter(
+        Characteristic(
+            '2AB7E91B-43E8-4F73-AC3B-80C1683B47F9',
+            Characteristic.Properties.READ,
+            Characteristic.READABLE,
+            CharacteristicValue(read=async_read),
+        ),
+        '>I',
+    )
+
+    service = Service(
+        '3A657F47-D34F-46B3-B1EC-698E29B6B829', [characteristic, async_characteristic]
+    )
     server.add_service(service)
 
     await client.power_on()
@@ -184,6 +198,13 @@ async def test_characteristic_encoding():
     await async_barrier()
     assert characteristic.value == bytes([50])
 
+    c2 = peer.get_characteristics_by_uuid(async_characteristic.uuid)
+    assert len(c2) == 1
+    c2 = c2[0]
+    cd2 = PackedCharacteristicAdapter(c2, ">I")
+    cd2v = await cd2.read_value()
+    assert cd2v == 0x05060708
+
     last_change = None
 
     def on_change(value):
@@ -285,7 +306,8 @@ async def test_attribute_getters():
 
 
 # -----------------------------------------------------------------------------
-def test_CharacteristicAdapter():
+@pytest.mark.asyncio
+async def test_CharacteristicAdapter():
     # Check that the CharacteristicAdapter base class is transparent
     v = bytes([1, 2, 3])
     c = Characteristic(
@@ -296,11 +318,11 @@ def test_CharacteristicAdapter():
     )
     a = CharacteristicAdapter(c)
 
-    value = a.read_value(None)
+    value = await a.read_value(None)
     assert value == v
 
     v = bytes([3, 4, 5])
-    a.write_value(None, v)
+    await a.write_value(None, v)
     assert c.value == v
 
     # Simple delegated adapter
@@ -308,11 +330,11 @@ def test_CharacteristicAdapter():
         c, lambda x: bytes(reversed(x)), lambda x: bytes(reversed(x))
     )
 
-    value = a.read_value(None)
+    value = await a.read_value(None)
     assert value == bytes(reversed(v))
 
     v = bytes([3, 4, 5])
-    a.write_value(None, v)
+    await a.write_value(None, v)
     assert a.value == bytes(reversed(v))
 
     # Packed adapter with single element format
@@ -321,10 +343,10 @@ def test_CharacteristicAdapter():
     c.value = v
     a = PackedCharacteristicAdapter(c, '>H')
 
-    value = a.read_value(None)
+    value = await a.read_value(None)
     assert value == pv
     c.value = None
-    a.write_value(None, pv)
+    await a.write_value(None, pv)
     assert a.value == v
 
     # Packed adapter with multi-element format
@@ -334,10 +356,10 @@ def test_CharacteristicAdapter():
     c.value = (v1, v2)
     a = PackedCharacteristicAdapter(c, '>HH')
 
-    value = a.read_value(None)
+    value = await a.read_value(None)
     assert value == pv
     c.value = None
-    a.write_value(None, pv)
+    await a.write_value(None, pv)
     assert a.value == (v1, v2)
 
     # Mapped adapter
@@ -348,10 +370,10 @@ def test_CharacteristicAdapter():
     c.value = mapped
     a = MappedCharacteristicAdapter(c, '>HH', ('v1', 'v2'))
 
-    value = a.read_value(None)
+    value = await a.read_value(None)
     assert value == pv
     c.value = None
-    a.write_value(None, pv)
+    await a.write_value(None, pv)
     assert a.value == mapped
 
     # UTF-8 adapter
@@ -360,27 +382,49 @@ def test_CharacteristicAdapter():
     c.value = v
     a = UTF8CharacteristicAdapter(c)
 
-    value = a.read_value(None)
+    value = await a.read_value(None)
     assert value == ev
     c.value = None
-    a.write_value(None, ev)
+    await a.write_value(None, ev)
     assert a.value == v
 
 
 # -----------------------------------------------------------------------------
-def test_CharacteristicValue():
+@pytest.mark.asyncio
+async def test_CharacteristicValue():
     b = bytes([1, 2, 3])
-    c = CharacteristicValue(read=lambda _: b)
-    x = c.read(None)
+
+    async def read_value(connection):
+        return b
+
+    c = CharacteristicValue(read=read_value)
+    x = await c.read(None)
     assert x == b
 
-    result = []
-    c = CharacteristicValue(
-        write=lambda connection, value: result.append((connection, value))
-    )
+    m = Mock()
+    c = CharacteristicValue(write=m)
     z = object()
     c.write(z, b)
-    assert result == [(z, b)]
+    m.assert_called_once_with(z, b)
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_CharacteristicValue_async():
+    b = bytes([1, 2, 3])
+
+    async def read_value(connection):
+        return b
+
+    c = CharacteristicValue(read=read_value)
+    x = await c.read(None)
+    assert x == b
+
+    m = AsyncMock()
+    c = CharacteristicValue(write=m)
+    z = object()
+    await c.write(z, b)
+    m.assert_called_once_with(z, b)
 
 
 # -----------------------------------------------------------------------------
@@ -961,12 +1005,18 @@ Descriptor(handle=0x0009, type=UUID-16:2902 (Client Characteristic Configuration
 
 # -----------------------------------------------------------------------------
 async def async_main():
+    test_UUID()
+    test_ATT_Error_Response()
+    test_ATT_Read_By_Group_Type_Request()
     await test_read_write()
     await test_read_write2()
     await test_subscribe_notify()
     await test_unsubscribe()
     await test_characteristic_encoding()
     await test_mtu_exchange()
+    await test_CharacteristicValue()
+    await test_CharacteristicValue_async()
+    await test_CharacteristicAdapter()
 
 
 # -----------------------------------------------------------------------------
@@ -1105,9 +1155,4 @@ def test_get_attribute_group():
 # -----------------------------------------------------------------------------
 if __name__ == '__main__':
     logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper())
-    test_UUID()
-    test_ATT_Error_Response()
-    test_ATT_Read_By_Group_Type_Request()
-    test_CharacteristicValue()
-    test_CharacteristicAdapter()
     asyncio.run(async_main())