diff --git a/bumble/gatt.py b/bumble/gatt.py index 371255e72..66e92f32b 100644 --- a/bumble/gatt.py +++ b/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') diff --git a/bumble/profiles/csip.py b/bumble/profiles/csip.py index 6cfe9a62e..de7cce0a4 100644 --- a/bumble/profiles/csip.py +++ b/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() diff --git a/tests/csip_test.py b/tests/csip_test.py index 865ad71e6..4d5fe7578 100644 --- a/tests/csip_test.py +++ b/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()