Feature - add <Coordinated Set Name> characteristic to csip.py

bumble/gatt.py

@@ -205,6 +205,7 @@
 GATT_COORDINATED_SET_SIZE_CHARACTERISTIC        = UUID.from_16_bits(0x2B85, 'Coordinated Set Size')
 GATT_SET_MEMBER_LOCK_CHARACTERISTIC             = UUID.from_16_bits(0x2B86, 'Set Member Lock')
 GATT_SET_MEMBER_RANK_CHARACTERISTIC             = UUID.from_16_bits(0x2B87, 'Set Member Rank')
+GATT_COORDINATED_SET_NAME_CHARACTERISTIC        = UUID.from_16_bits(0x2C1A, 'Coordinated Set Name')
 
 # Media Control Service (MCS)
 GATT_MEDIA_PLAYER_NAME_CHARACTERISTIC                     = UUID.from_16_bits(0x2B93, 'Media Player Name')

bumble/profiles/csip.py

@@ -22,11 +22,13 @@
 import struct
 
 from bumble import core, crypto, device, gatt, gatt_client
+from bumble.gatt_adapters import UTF8CharacteristicProxyAdapter
 
 # -----------------------------------------------------------------------------
 # Constants
 # -----------------------------------------------------------------------------
 SET_IDENTITY_RESOLVING_KEY_LENGTH = 16
+COORDINATED_SET_NAME_MAX_LENGTH = 128
 
 
 class SirkType(enum.IntEnum):
@@ -98,6 +100,7 @@ class CoordinatedSetIdentificationService(gatt.TemplateService):
     coordinated_set_size_characteristic: gatt.Characteristic[bytes] | None = None
     set_member_lock_characteristic: gatt.Characteristic[bytes] | None = None
     set_member_rank_characteristic: gatt.Characteristic[bytes] | None = None
+    coordinated_set_name_characteristic: gatt.Characteristic[bytes] | None = None
 
     def __init__(
         self,
@@ -106,6 +109,7 @@ def __init__(
         coordinated_set_size: int | None = None,
         set_member_lock: MemberLock | None = None,
         set_member_rank: int | None = None,
+        coordinated_set_name: str | None = None,
     ) -> None:
         if len(set_identity_resolving_key) != SET_IDENTITY_RESOLVING_KEY_LENGTH:
             raise core.InvalidArgumentError(
@@ -157,6 +161,22 @@ def __init__(
             )
             characteristics.append(self.set_member_rank_characteristic)
 
+        if coordinated_set_name is not None:
+            name_bytes = coordinated_set_name.encode('utf-8')
+            if len(name_bytes) > COORDINATED_SET_NAME_MAX_LENGTH:
+                raise core.InvalidArgumentError(
+                    f'Coordinated Set Name is {len(name_bytes)} octets, '
+                    f'maximum is {COORDINATED_SET_NAME_MAX_LENGTH}'
+                )
+            self.coordinated_set_name_characteristic = gatt.Characteristic(
+                uuid=gatt.GATT_COORDINATED_SET_NAME_CHARACTERISTIC,
+                properties=gatt.Characteristic.Properties.READ
+                | gatt.Characteristic.Properties.NOTIFY,
+                permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
+                value=name_bytes,
+            )
+            characteristics.append(self.coordinated_set_name_characteristic)
+
         super().__init__(characteristics)
 
     async def on_sirk_read(self, connection: device.Connection) -> bytes:
@@ -200,6 +220,7 @@ class CoordinatedSetIdentificationProxy(gatt_client.ProfileServiceProxy):
     coordinated_set_size: gatt_client.CharacteristicProxy[bytes] | None = None
     set_member_lock: gatt_client.CharacteristicProxy[bytes] | None = None
     set_member_rank: gatt_client.CharacteristicProxy[bytes] | None = None
+    coordinated_set_name: UTF8CharacteristicProxyAdapter | None = None
 
     def __init__(self, service_proxy: gatt_client.ServiceProxy) -> None:
         self.service_proxy = service_proxy
@@ -223,6 +244,13 @@ def __init__(self, service_proxy: gatt_client.ServiceProxy) -> None:
         ):
             self.set_member_rank = characteristics[0]
 
+        if characteristics := service_proxy.get_characteristics_by_uuid(
+            gatt.GATT_COORDINATED_SET_NAME_CHARACTERISTIC
+        ):
+            self.coordinated_set_name = UTF8CharacteristicProxyAdapter(
+                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()

tests/csip_test.py

@@ -23,7 +23,7 @@
 
 import pytest
 
-from bumble import device
+from bumble import core, device
 from bumble.profiles import csip
 from bumble.testing.test_utils import TwoDevices
 
@@ -109,6 +109,96 @@ async def test_csis(sirk_type):
     assert await csis_client.set_member_rank.read_value() == struct.pack('B', 0)
 
 
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_coordinated_set_name():
+    SIRK = bytes.fromhex('2f62c8ae41867d1bb619e788a2605faa')
+    LTK = bytes.fromhex('2f62c8ae41867d1bb619e788a2605faa')
+    SET_NAME = 'My Earbuds'
+
+    devices = TwoDevices()
+    devices[0].add_service(
+        csip.CoordinatedSetIdentificationService(
+            set_identity_resolving_key=SIRK,
+            set_identity_resolving_key_type=csip.SirkType.PLAINTEXT,
+            coordinated_set_name=SET_NAME,
+        )
+    )
+
+    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
+    )
+
+    # Verify the optional Coordinated Set Name characteristic is present and readable.
+    assert csis_client.coordinated_set_name is not None
+    name = await csis_client.coordinated_set_name.read_value()
+    assert name == SET_NAME
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_coordinated_set_name_optional():
+    '''Coordinated Set Name is optional: omitting it should leave the proxy attribute as None.'''
+    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,
+        )
+    )
+
+    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
+    )
+
+    # Coordinated Set Name was not provided, so the proxy attribute should be None.
+    assert csis_client.coordinated_set_name is None
+
+
+# -----------------------------------------------------------------------------
+def test_coordinated_set_name_max_length():
+    '''Coordinated Set Name is limited to 128 octets as UTF-8.'''
+    SIRK = bytes.fromhex('2f62c8ae41867d1bb619e788a2605faa')
+
+    # A 128-character ASCII string encodes to exactly 128 octets (within limit).
+    valid_name = 'a' * 128
+    service = csip.CoordinatedSetIdentificationService(
+        set_identity_resolving_key=SIRK,
+        set_identity_resolving_key_type=csip.SirkType.PLAINTEXT,
+        coordinated_set_name=valid_name,
+    )
+    assert service.coordinated_set_name_characteristic is not None
+
+    # A 129-character ASCII string encodes to 129 octets (over the limit).
+    with pytest.raises(core.InvalidArgumentError):
+        csip.CoordinatedSetIdentificationService(
+            set_identity_resolving_key=SIRK,
+            set_identity_resolving_key_type=csip.SirkType.PLAINTEXT,
+            coordinated_set_name='a' * 129,
+        )
+
+
 # -----------------------------------------------------------------------------
 async def run():
     test_sih()