Support GATT Service

bumble/device.py

@@ -93,6 +93,7 @@
 from bumble import sdp
 from bumble import l2cap
 from bumble import core
+from bumble.profiles import gatt_service
 
 if TYPE_CHECKING:
     from .transport.common import TransportSource, TransportSink
@@ -1756,6 +1757,8 @@ class DeviceConfiguration:
     cis_enabled: bool = False
     identity_address_type: Optional[int] = None
     io_capability: int = pairing.PairingDelegate.IoCapability.NO_OUTPUT_NO_INPUT
+    gap_service_enabled: bool = True
+    gatt_service_enabled: bool = True
 
     def __post_init__(self) -> None:
         self.gatt_services: list[Dict[str, Any]] = []
@@ -1938,6 +1941,7 @@ class Device(CompositeEventEmitter):
     bis_links = dict[int, BisLink]()
     big_syncs = dict[int, BigSync]()
     _pending_cis: Dict[int, tuple[int, int]]
+    gatt_service: gatt_service.GenericAttributeProfileService | None = None
 
     @composite_listener
     class Listener:
@@ -2004,7 +2008,6 @@ def __init__(
         address: Optional[hci.Address] = None,
         config: Optional[DeviceConfiguration] = None,
         host: Optional[Host] = None,
-        generic_access_service: bool = True,
     ) -> None:
         super().__init__()
 
@@ -2151,7 +2154,10 @@ def __init__(
         # Register the SDP server with the L2CAP Channel Manager
         self.sdp_server.register(self.l2cap_channel_manager)
 
-        self.add_default_services(generic_access_service)
+        self.add_default_services(
+            add_gap_service=config.gap_service_enabled,
+            add_gatt_service=config.gatt_service_enabled,
+        )
         self.l2cap_channel_manager.register_fixed_channel(ATT_CID, self.on_gatt_pdu)
 
         # Forward some events
@@ -4515,10 +4521,15 @@ def add_service(self, service):
     def add_services(self, services):
         self.gatt_server.add_services(services)
 
-    def add_default_services(self, generic_access_service=True):
+    def add_default_services(
+        self, add_gap_service: bool = True, add_gatt_service: bool = True
+    ) -> None:
         # Add a GAP Service if requested
-        if generic_access_service:
+        if add_gap_service:
             self.gatt_server.add_service(GenericAccessService(self.name))
+        if add_gatt_service:
+            self.gatt_service = gatt_service.GenericAttributeProfileService()
+            self.gatt_server.add_service(self.gatt_service)
 
     async def notify_subscriber(self, connection, attribute, value=None, force=False):
         await self.gatt_server.notify_subscriber(connection, attribute, value, force)

bumble/gatt.py

@@ -48,7 +48,6 @@
 
 if TYPE_CHECKING:
     from bumble.gatt_client import AttributeProxy
-    from bumble.device import Connection
 
 
 # -----------------------------------------------------------------------------
@@ -802,3 +801,23 @@ class ClientCharacteristicConfigurationBits(enum.IntFlag):
     DEFAULT = 0x0000
     NOTIFICATION = 0x0001
     INDICATION = 0x0002
+
+
+# -----------------------------------------------------------------------------
+class ClientSupportedFeatures(enum.IntFlag):
+    '''
+    See Vol 3, Part G - 7.2 - Table 7.6: Client Supported Features bit assignments.
+    '''
+
+    ROBUST_CACHING = 0x01
+    ENHANCED_ATT_BEARER = 0x02
+    MULTIPLE_HANDLE_VALUE_NOTIFICATIONS = 0x04
+
+
+# -----------------------------------------------------------------------------
+class ServerSupportedFeatures(enum.IntFlag):
+    '''
+    See Vol 3, Part G - 7.4 - Table 7.11: Server Supported Features bit assignments.
+    '''
+
+    EATT_SUPPORTED = 0x01

bumble/profiles/gatt_service.py

@@ -0,0 +1,166 @@
+# Copyright 2021-2025 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 struct
+from typing import TYPE_CHECKING
+
+from bumble import att
+from bumble import gatt
+from bumble import gatt_client
+from bumble import crypto
+
+if TYPE_CHECKING:
+    from bumble import device
+
+
+# -----------------------------------------------------------------------------
+class GenericAttributeProfileService(gatt.TemplateService):
+    '''See Vol 3, Part G - 7 - DEFINED GENERIC ATTRIBUTE PROFILE SERVICE.'''
+
+    UUID = gatt.GATT_GENERIC_ATTRIBUTE_SERVICE
+
+    client_supported_features_characteristic: gatt.Characteristic | None = None
+    server_supported_features_characteristic: gatt.Characteristic | None = None
+    database_hash_characteristic: gatt.Characteristic | None = None
+    service_changed_characteristic: gatt.Characteristic | None = None
+
+    def __init__(
+        self,
+        server_supported_features: gatt.ServerSupportedFeatures | None = None,
+        database_hash_enabled: bool = True,
+        service_change_enabled: bool = True,
+    ) -> None:
+
+        if server_supported_features is not None:
+            self.server_supported_features_characteristic = gatt.Characteristic(
+                uuid=gatt.GATT_SERVER_SUPPORTED_FEATURES_CHARACTERISTIC,
+                properties=gatt.Characteristic.Properties.READ,
+                permissions=gatt.Characteristic.Permissions.READABLE,
+                value=bytes([server_supported_features]),
+            )
+
+        if database_hash_enabled:
+            self.database_hash_characteristic = gatt.Characteristic(
+                uuid=gatt.GATT_DATABASE_HASH_CHARACTERISTIC,
+                properties=gatt.Characteristic.Properties.READ,
+                permissions=gatt.Characteristic.Permissions.READABLE,
+                value=gatt.CharacteristicValue(read=self.get_database_hash),
+            )
+
+        if service_change_enabled:
+            self.service_changed_characteristic = gatt.Characteristic(
+                uuid=gatt.GATT_SERVICE_CHANGED_CHARACTERISTIC,
+                properties=gatt.Characteristic.Properties.INDICATE,
+                permissions=gatt.Characteristic.Permissions(0),
+                value=b'',
+            )
+
+        if (database_hash_enabled and service_change_enabled) or (
+            server_supported_features
+            and (
+                server_supported_features & gatt.ServerSupportedFeatures.EATT_SUPPORTED
+            )
+        ):  # TODO: Support Multiple Handle Value Notifications
+            self.client_supported_features_characteristic = gatt.Characteristic(
+                uuid=gatt.GATT_CLIENT_SUPPORTED_FEATURES_CHARACTERISTIC,
+                properties=(
+                    gatt.Characteristic.Properties.READ
+                    | gatt.Characteristic.Properties.WRITE
+                ),
+                permissions=(
+                    gatt.Characteristic.Permissions.READABLE
+                    | gatt.Characteristic.Permissions.WRITEABLE
+                ),
+                value=bytes(1),
+            )
+
+        super().__init__(
+            characteristics=[
+                c
+                for c in (
+                    self.service_changed_characteristic,
+                    self.client_supported_features_characteristic,
+                    self.database_hash_characteristic,
+                    self.server_supported_features_characteristic,
+                )
+                if c is not None
+            ],
+            primary=True,
+        )
+
+    @classmethod
+    def get_attribute_data(cls, attribute: att.Attribute) -> bytes:
+        if attribute.type in (
+            gatt.GATT_PRIMARY_SERVICE_ATTRIBUTE_TYPE,
+            gatt.GATT_SECONDARY_SERVICE_ATTRIBUTE_TYPE,
+            gatt.GATT_INCLUDE_ATTRIBUTE_TYPE,
+            gatt.GATT_CHARACTERISTIC_ATTRIBUTE_TYPE,
+            gatt.GATT_CHARACTERISTIC_EXTENDED_PROPERTIES_DESCRIPTOR,
+        ):
+            return (
+                struct.pack("<H", attribute.handle)
+                + attribute.type.to_bytes()
+                + attribute.value
+            )
+        elif attribute.type in (
+            gatt.GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR,
+            gatt.GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR,
+            gatt.GATT_SERVER_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR,
+            gatt.GATT_CHARACTERISTIC_PRESENTATION_FORMAT_DESCRIPTOR,
+            gatt.GATT_CHARACTERISTIC_AGGREGATE_FORMAT_DESCRIPTOR,
+        ):
+            return struct.pack("<H", attribute.handle) + attribute.type.to_bytes()
+
+        return b''
+
+    def get_database_hash(self, connection: device.Connection | None) -> bytes:
+        assert connection
+
+        m = b''.join(
+            [
+                self.get_attribute_data(attribute)
+                for attribute in connection.device.gatt_server.attributes
+            ]
+        )
+
+        return crypto.aes_cmac(m=m, k=bytes(16))
+
+
+class GenericAttributeProfileServiceProxy(gatt_client.ProfileServiceProxy):
+    SERVICE_CLASS = GenericAttributeProfileService
+
+    client_supported_features_characteristic: gatt_client.CharacteristicProxy | None = (
+        None
+    )
+    server_supported_features_characteristic: gatt_client.CharacteristicProxy | None = (
+        None
+    )
+    database_hash_characteristic: gatt_client.CharacteristicProxy | None = None
+    service_changed_characteristic: gatt_client.CharacteristicProxy | None = None
+
+    _CHARACTERISTICS = {
+        gatt.GATT_CLIENT_SUPPORTED_FEATURES_CHARACTERISTIC: 'client_supported_features_characteristic',
+        gatt.GATT_SERVER_SUPPORTED_FEATURES_CHARACTERISTIC: 'server_supported_features_characteristic',
+        gatt.GATT_DATABASE_HASH_CHARACTERISTIC: 'database_hash_characteristic',
+        gatt.GATT_SERVICE_CHANGED_CHARACTERISTIC: 'service_changed_characteristic',
+    }
+
+    def __init__(self, service_proxy: gatt_client.ServiceProxy) -> None:
+        self.service_proxy = service_proxy
+
+        for uuid, attribute_name in self._CHARACTERISTICS.items():
+            if characteristics := self.service_proxy.get_characteristics_by_uuid(uuid):
+                setattr(self, attribute_name, characteristics[0])

tests/device_test.py

@@ -50,12 +50,7 @@
     HCI_Error,
     HCI_Packet,
 )
-from bumble.gatt import (
-    GATT_GENERIC_ACCESS_SERVICE,
-    GATT_CHARACTERISTIC_ATTRIBUTE_TYPE,
-    GATT_DEVICE_NAME_CHARACTERISTIC,
-    GATT_APPEARANCE_CHARACTERISTIC,
-)
+from bumble import gatt
 
 from .test_utils import TwoDevices, async_barrier
 
@@ -592,32 +587,54 @@ async def test_power_on_default_static_address_should_not_be_any():
 
 
 # -----------------------------------------------------------------------------
-def test_gatt_services_with_gas():
+def test_gatt_services_with_gas_and_gatt():
     device = Device(host=Host(None, None))
 
-    # there should be one service and two chars, therefore 5 attributes
-    assert len(device.gatt_server.attributes) == 5
-    assert device.gatt_server.attributes[0].uuid == GATT_GENERIC_ACCESS_SERVICE
-    assert device.gatt_server.attributes[1].type == GATT_CHARACTERISTIC_ATTRIBUTE_TYPE
-    assert device.gatt_server.attributes[2].uuid == GATT_DEVICE_NAME_CHARACTERISTIC
-    assert device.gatt_server.attributes[3].type == GATT_CHARACTERISTIC_ATTRIBUTE_TYPE
-    assert device.gatt_server.attributes[4].uuid == GATT_APPEARANCE_CHARACTERISTIC
-
-
-# -----------------------------------------------------------------------------
-def test_gatt_services_without_gas():
-    device = Device(host=Host(None, None), generic_access_service=False)
+    # there should be 2 service, 5 chars, and 1 descriptors, therefore 13 attributes
+    assert len(device.gatt_server.attributes) == 13
+    assert device.gatt_server.attributes[0].uuid == gatt.GATT_GENERIC_ACCESS_SERVICE
+    assert (
+        device.gatt_server.attributes[1].type == gatt.GATT_CHARACTERISTIC_ATTRIBUTE_TYPE
+    )
+    assert device.gatt_server.attributes[2].uuid == gatt.GATT_DEVICE_NAME_CHARACTERISTIC
+    assert (
+        device.gatt_server.attributes[3].type == gatt.GATT_CHARACTERISTIC_ATTRIBUTE_TYPE
+    )
+    assert device.gatt_server.attributes[4].uuid == gatt.GATT_APPEARANCE_CHARACTERISTIC
 
-    # there should be no services
-    assert len(device.gatt_server.attributes) == 0
+    assert device.gatt_server.attributes[5].uuid == gatt.GATT_GENERIC_ATTRIBUTE_SERVICE
+    assert (
+        device.gatt_server.attributes[6].type == gatt.GATT_CHARACTERISTIC_ATTRIBUTE_TYPE
+    )
+    assert (
+        device.gatt_server.attributes[7].uuid
+        == gatt.GATT_CLIENT_SUPPORTED_FEATURES_CHARACTERISTIC
+    )
+    assert (
+        device.gatt_server.attributes[8].type == gatt.GATT_CHARACTERISTIC_ATTRIBUTE_TYPE
+    )
+    assert (
+        device.gatt_server.attributes[9].uuid == gatt.GATT_DATABASE_HASH_CHARACTERISTIC
+    )
+    assert (
+        device.gatt_server.attributes[10].type
+        == gatt.GATT_CHARACTERISTIC_ATTRIBUTE_TYPE
+    )
+    assert (
+        device.gatt_server.attributes[11].uuid
+        == gatt.GATT_SERVICE_CHANGED_CHARACTERISTIC
+    )
+    assert (
+        device.gatt_server.attributes[12].type
+        == gatt.GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR
+    )
 
 
 # -----------------------------------------------------------------------------
 async def run_test_device():
     await test_device_connect_parallel()
     await test_flush()
-    await test_gatt_services_with_gas()
-    await test_gatt_services_without_gas()
+    await test_gatt_services_with_gas_and_gatt()
 
 
 # -----------------------------------------------------------------------------