From ad0a7e171bc4af1efae07a56b28d53eb57466adf Mon Sep 17 00:00:00 2001 From: deadcafe Date: Mon, 5 Oct 2026 03:22:38 +0530 Subject: [PATCH 1/3] Check attribute read and write permissions in the GATT server The server never returned Read Not Permitted or Write Not Permitted: an attribute could be written without any write permission, and read without any read permission. Only the encryption, authentication and authorization requirements were enforced. Requests from a peer are now checked before the attribute is accessed: - Read, Read Blob, Read By Type, Read Multiple and Read Multiple Variable return Read Not Permitted. - Find By Type Value skips attributes that cannot be read. - Write and Prepare Write return Write Not Permitted. - Write Command is ignored. A security requirement implies the access that it protects, so that permissions like READ_REQUIRES_ENCRYPTION alone keep working. Notifications and indications are not affected, since the value is then read locally. Read Multiple and Read Multiple Variable now also return an error response when reading an attribute fails, instead of not responding. The heart rate test was reading the measurement characteristic, which can only be notified. It now checks a notification. --- bumble/gatt_server.py | 98 +++++++++++++--- tests/gatt_test.py | 133 ++++++++++++++++++++++ tests/profiles/heart_rate_service_test.py | 12 +- 3 files changed, 226 insertions(+), 17 deletions(-) diff --git a/bumble/gatt_server.py b/bumble/gatt_server.py index ee7ac7ba..03d0fcf2 100644 --- a/bumble/gatt_server.py +++ b/bumble/gatt_server.py @@ -62,6 +62,20 @@ # ----------------------------------------------------------------------------- GATT_SERVER_DEFAULT_MAX_MTU = 517 +# A security requirement implies the access that it protects +_READ_PERMISSIONS = ( + att.Attribute.READABLE + | att.Attribute.READ_REQUIRES_ENCRYPTION + | att.Attribute.READ_REQUIRES_AUTHENTICATION + | att.Attribute.READ_REQUIRES_AUTHORIZATION +) +_WRITE_PERMISSIONS = ( + att.Attribute.WRITEABLE + | att.Attribute.WRITE_REQUIRES_ENCRYPTION + | att.Attribute.WRITE_REQUIRES_AUTHENTICATION + | att.Attribute.WRITE_REQUIRES_AUTHORIZATION +) + # ----------------------------------------------------------------------------- # Helpers @@ -370,6 +384,16 @@ def send_response(self, bearer: att.Bearer, response: att.ATT_PDU) -> None: logger.debug(f'GATT Response from server: {_bearer_id(bearer)} {response}') self.send_gatt_pdu(bearer, bytes(response)) + async def _read_attribute_value( + self, bearer: att.Bearer, attribute: att.Attribute + ) -> bytes: + # Read a value on behalf of a peer + if not attribute.permissions & _READ_PERMISSIONS: + raise att.ATT_Error( + error_code=att.ATT_READ_NOT_PERMITTED_ERROR, att_handle=attribute.handle + ) + return await attribute.read_value(bearer) + async def notify_subscriber( self, bearer: att.Bearer, @@ -721,16 +745,21 @@ async def on_att_find_by_type_value_request( pdu_space_available = bearer.att_mtu - 2 attributes = [] response: att.ATT_PDU - async for attribute in ( + 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 (await attribute.read_value(bearer)) == request.attribute_value and pdu_space_available >= 4 ): - # TODO: check permissions + # Only attributes that can be read are returned + try: + attribute_value = await self._read_attribute_value(bearer, attribute) + except att.ATT_Error: + continue + if attribute_value != request.attribute_value: + continue # Add the attribute to the list attributes.append(attribute) @@ -803,7 +832,7 @@ async def on_att_read_by_type_request( and pdu_space_available ): try: - attribute_value = await attribute.read_value(bearer) + attribute_value = await self._read_attribute_value(bearer, attribute) except att.ATT_Error as error: # If the first attribute is unreadable, return an error # Otherwise return attributes up to this point @@ -856,7 +885,7 @@ async def on_att_read_request( response: att.ATT_PDU if attribute := self.get_attribute(request.attribute_handle): try: - value = await attribute.read_value(bearer) + value = await self._read_attribute_value(bearer, attribute) except att.ATT_Error as error: response = att.ATT_Error_Response( request_opcode_in_error=request.op_code, @@ -885,7 +914,7 @@ async def on_att_read_blob_request( response: att.ATT_PDU if attribute := self.get_attribute(request.attribute_handle): try: - value = await attribute.read_value(bearer) + value = await self._read_attribute_value(bearer, attribute) except att.ATT_Error as error: response = att.ATT_Error_Response( request_opcode_in_error=request.op_code, @@ -1014,9 +1043,16 @@ async def on_att_read_multiple_request( ) self.send_response(bearer, response) return - # No need to catch permission errors here, since these attributes - # must all be world-readable - attribute_value = await attribute.read_value(bearer) + try: + attribute_value = await self._read_attribute_value(bearer, attribute) + except att.ATT_Error as error: + response = att.ATT_Error_Response( + request_opcode_in_error=request.op_code, + attribute_handle_in_error=handle, + error_code=error.error_code, + ) + self.send_response(bearer, response) + return # Check the attribute value size max_attribute_size = min(bearer.att_mtu - 1, 251) if len(attribute_value) > max_attribute_size: @@ -1056,9 +1092,16 @@ async def on_att_read_multiple_variable_request( ) self.send_response(bearer, response) return - # No need to catch permission errors here, since these attributes - # must all be world-readable - attribute_value = await attribute.read_value(bearer) + try: + attribute_value = await self._read_attribute_value(bearer, attribute) + except att.ATT_Error as error: + response = att.ATT_Error_Response( + request_opcode_in_error=request.op_code, + attribute_handle_in_error=handle, + error_code=error.error_code, + ) + self.send_response(bearer, response) + return length = len(attribute_value) # Check the attribute value size max_attribute_size = min(bearer.att_mtu - 3, 251) @@ -1102,7 +1145,17 @@ async def on_att_write_request( ) return - # TODO: check permissions + # Check that the attribute can be written + if not attribute.permissions & _WRITE_PERMISSIONS: + self.send_response( + bearer, + att.ATT_Error_Response( + request_opcode_in_error=request.op_code, + attribute_handle_in_error=request.attribute_handle, + error_code=att.ATT_WRITE_NOT_PERMITTED_ERROR, + ), + ) + return # Check the request parameters if len(request.attribute_value) > GATT_MAX_ATTRIBUTE_VALUE_SIZE: @@ -1144,7 +1197,9 @@ async def on_att_write_command( if attribute is None: return - # TODO: check permissions + # Check that the attribute can be written + if not attribute.permissions & _WRITE_PERMISSIONS: + return # Check the request parameters if len(request.attribute_value) > GATT_MAX_ATTRIBUTE_VALUE_SIZE: @@ -1164,7 +1219,8 @@ def on_att_prepare_write_request( ''' # Check that the attribute exists - if self.get_attribute(request.attribute_handle) is None: + attribute = self.get_attribute(request.attribute_handle) + if attribute is None: self.send_response( bearer, att.ATT_Error_Response( @@ -1175,6 +1231,18 @@ def on_att_prepare_write_request( ) return + # Check that the attribute can be written + if not attribute.permissions & _WRITE_PERMISSIONS: + self.send_response( + bearer, + att.ATT_Error_Response( + request_opcode_in_error=request.op_code, + attribute_handle_in_error=request.attribute_handle, + error_code=att.ATT_WRITE_NOT_PERMITTED_ERROR, + ), + ) + return + # Queue the partial value, to be committed on Execute Write Request self.prepared_writes.setdefault(bearer, []).append( ( diff --git a/tests/gatt_test.py b/tests/gatt_test.py index 64cfa7c4..21561dd8 100644 --- a/tests/gatt_test.py +++ b/tests/gatt_test.py @@ -1865,6 +1865,139 @@ async def test_write_long_value_gap_rejected(): assert characteristic.value is None +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_write_not_permitted(): + devices = await TwoDevices.create_with_connection() + + characteristic = Characteristic( + '1234', Characteristic.Properties.READ, Characteristic.READABLE, b'1234' + ) + devices[1].add_service(Service('ABCD', [characteristic])) + client = devices.connections[0].gatt_client + + response = await client.send_request( + att.ATT_Write_Request( + attribute_handle=characteristic.handle, attribute_value=b'5678' + ) + ) + assert isinstance(response, att.ATT_Error_Response) + assert response.error_code == att.ATT_WRITE_NOT_PERMITTED_ERROR + + await client.send_command( + att.ATT_Write_Command( + attribute_handle=characteristic.handle, attribute_value=b'5678' + ) + ) + + response = await client.send_request( + att.ATT_Prepare_Write_Request( + attribute_handle=characteristic.handle, + value_offset=0, + part_attribute_value=b'5678', + ) + ) + assert isinstance(response, att.ATT_Error_Response) + assert response.error_code == att.ATT_WRITE_NOT_PERMITTED_ERROR + + # Nothing was queued, so there is nothing to commit + response = await client.send_request(att.ATT_Execute_Write_Request(flags=0x01)) + assert isinstance(response, att.ATT_Execute_Write_Response) + await async_barrier() + + assert characteristic.value == b'1234' + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_read_not_permitted(): + devices = await TwoDevices.create_with_connection() + + characteristic = Characteristic( + '1234', Characteristic.Properties.WRITE, Characteristic.WRITEABLE, b'1234' + ) + devices[1].add_service(Service('ABCD', [characteristic])) + client = devices.connections[0].gatt_client + + for request in ( + att.ATT_Read_Request(attribute_handle=characteristic.handle), + att.ATT_Read_Blob_Request( + attribute_handle=characteristic.handle, value_offset=0 + ), + att.ATT_Read_By_Type_Request( + starting_handle=0x0001, + ending_handle=0xFFFF, + attribute_type=characteristic.uuid, + ), + att.ATT_Read_Multiple_Request(set_of_handles=[characteristic.handle]), + att.ATT_Read_Multiple_Variable_Request(set_of_handles=[characteristic.handle]), + ): + response = await client.send_request(request) + assert isinstance(response, att.ATT_Error_Response) + assert response.error_code == att.ATT_READ_NOT_PERMITTED_ERROR + assert response.attribute_handle_in_error == characteristic.handle + + # An attribute that cannot be read is not matched by value + response = await client.send_request( + att.ATT_Find_By_Type_Value_Request( + starting_handle=0x0001, + ending_handle=0xFFFF, + attribute_type=characteristic.uuid, + attribute_value=b'1234', + ) + ) + assert isinstance(response, att.ATT_Error_Response) + assert response.error_code == att.ATT_ATTRIBUTE_NOT_FOUND_ERROR + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_requires_permissions_imply_access(): + devices = await TwoDevices.create_with_connection() + + # No READABLE or WRITEABLE bit, only the security requirements + characteristic = Characteristic( + '1234', + Characteristic.Properties.READ | Characteristic.Properties.WRITE, + Characteristic.READ_REQUIRES_ENCRYPTION + | Characteristic.WRITE_REQUIRES_ENCRYPTION, + b'1234', + ) + devices[1].add_service(Service('ABCD', [characteristic])) + client = devices.connections[0].gatt_client + + # Not encrypted: the security requirement is reported, not NOT_PERMITTED + response = await client.send_request( + att.ATT_Read_Request(attribute_handle=characteristic.handle) + ) + assert isinstance(response, att.ATT_Error_Response) + assert response.error_code == att.ATT_INSUFFICIENT_ENCRYPTION_ERROR + + response = await client.send_request( + att.ATT_Write_Request( + attribute_handle=characteristic.handle, attribute_value=b'5678' + ) + ) + assert isinstance(response, att.ATT_Error_Response) + assert response.error_code == att.ATT_INSUFFICIENT_ENCRYPTION_ERROR + + # Encrypted: both are allowed + devices.connections[1].encryption = 1 + response = await client.send_request( + att.ATT_Read_Request(attribute_handle=characteristic.handle) + ) + assert isinstance(response, att.ATT_Read_Response) + assert response.attribute_value == b'1234' + + response = await client.send_request( + att.ATT_Write_Request( + attribute_handle=characteristic.handle, attribute_value=b'5678' + ) + ) + assert isinstance(response, att.ATT_Write_Response) + assert characteristic.value == b'5678' + + # ----------------------------------------------------------------------------- if __name__ == '__main__': logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper()) diff --git a/tests/profiles/heart_rate_service_test.py b/tests/profiles/heart_rate_service_test.py index 70119c1f..3bfdff91 100644 --- a/tests/profiles/heart_rate_service_test.py +++ b/tests/profiles/heart_rate_service_test.py @@ -31,7 +31,7 @@ (1, 1000), (True, False, None), (2, None), ((3.0, 4.0, 5.0), None) ), ) -async def test_read_measurement( +async def test_notify_measurement( heart_rate: int, sensor_contact_detected: bool | None, energy_expanded: int | None, @@ -47,7 +47,15 @@ async def test_read_measurement( async with device_module.Peer(devices.connections[1]) as peer: client = peer.create_service_proxy(heart_rate_service.HeartRateServiceProxy) assert client - assert await client.heart_rate_measurement.read_value() == measurement + # The measurement can only be notified, not read + notifications = asyncio.Queue[ + heart_rate_service.HeartRateService.HeartRateMeasurement + ]() + await client.heart_rate_measurement.subscribe(notifications.put_nowait) + await devices[0].notify_subscribers( + service.heart_rate_measurement_characteristic + ) + assert await notifications.get() == measurement # ----------------------------------------------------------------------------- From 6ea8ae7b140144f689c7f607532f138c8911f399 Mon Sep 17 00:00:00 2001 From: deadcafe Date: Mon, 5 Oct 2026 15:58:35 +0530 Subject: [PATCH 2/3] Make the GATT server permission checks optional Bumble is also used to simulate devices that do not follow the specification, and existing tests may read or write attributes that do not have the matching permission, or rely on a value callback to return its own error code. Server.strict_permissions, True by default, controls whether READABLE and WRITEABLE are required for reads and writes from a peer. When it is False, those checks are skipped, as before, and the value callbacks are always reached. --- bumble/gatt_server.py | 10 ++++++---- tests/gatt_test.py | 26 ++++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/bumble/gatt_server.py b/bumble/gatt_server.py index 03d0fcf2..0a9e4297 100644 --- a/bumble/gatt_server.py +++ b/bumble/gatt_server.py @@ -113,6 +113,8 @@ def __init__(self, device: Device) -> None: self.max_mtu = ( GATT_SERVER_DEFAULT_MAX_MTU # The max MTU we're willing to negotiate ) + # When False, READABLE and WRITEABLE are not required for peer reads and writes + self.strict_permissions = True self.subscribers = ( {} ) # Map of subscriber states by connection handle and attribute handle @@ -388,7 +390,7 @@ async def _read_attribute_value( self, bearer: att.Bearer, attribute: att.Attribute ) -> bytes: # Read a value on behalf of a peer - if not attribute.permissions & _READ_PERMISSIONS: + if self.strict_permissions and not attribute.permissions & _READ_PERMISSIONS: raise att.ATT_Error( error_code=att.ATT_READ_NOT_PERMITTED_ERROR, att_handle=attribute.handle ) @@ -1146,7 +1148,7 @@ async def on_att_write_request( return # Check that the attribute can be written - if not attribute.permissions & _WRITE_PERMISSIONS: + if self.strict_permissions and not attribute.permissions & _WRITE_PERMISSIONS: self.send_response( bearer, att.ATT_Error_Response( @@ -1198,7 +1200,7 @@ async def on_att_write_command( return # Check that the attribute can be written - if not attribute.permissions & _WRITE_PERMISSIONS: + if self.strict_permissions and not attribute.permissions & _WRITE_PERMISSIONS: return # Check the request parameters @@ -1232,7 +1234,7 @@ def on_att_prepare_write_request( return # Check that the attribute can be written - if not attribute.permissions & _WRITE_PERMISSIONS: + if self.strict_permissions and not attribute.permissions & _WRITE_PERMISSIONS: self.send_response( bearer, att.ATT_Error_Response( diff --git a/tests/gatt_test.py b/tests/gatt_test.py index 21561dd8..dfc36060 100644 --- a/tests/gatt_test.py +++ b/tests/gatt_test.py @@ -1998,6 +1998,32 @@ async def test_requires_permissions_imply_access(): assert characteristic.value == b'5678' +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_permissions_not_strict(): + devices = await TwoDevices.create_with_connection() + devices[1].gatt_server.strict_permissions = False + + characteristic = Characteristic( + '1234', Characteristic.Properties.NOTIFY, Characteristic.Permissions(0), b'1234' + ) + devices[1].add_service(Service('ABCD', [characteristic])) + client = devices.connections[0].gatt_client + + response = await client.send_request( + att.ATT_Write_Request( + attribute_handle=characteristic.handle, attribute_value=b'5678' + ) + ) + assert isinstance(response, att.ATT_Write_Response) + + response = await client.send_request( + att.ATT_Read_Request(attribute_handle=characteristic.handle) + ) + assert isinstance(response, att.ATT_Read_Response) + assert response.attribute_value == b'5678' + + # ----------------------------------------------------------------------------- if __name__ == '__main__': logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper()) From 097eb3e8b9d9028cfda0d6296a049363118535da Mon Sep 17 00:00:00 2001 From: deadcafe Date: Sat, 17 Oct 2026 00:30:37 +0530 Subject: [PATCH 3/3] Move the GATT permission check switch to DeviceConfiguration, off by default Server.strict_permissions was a loose attribute on the server, unlike the other behaviour switches, which live in DeviceConfiguration. It is now DeviceConfiguration.gatt_strict_permissions, so it can also be set from a config file. It is off by default: existing tests that read or write attributes without the matching permission bit keep working unchanged, and the Read Not Permitted and Write Not Permitted errors are opted into. --- bumble/device.py | 1 + bumble/gatt_server.py | 22 ++++++++++++++++------ tests/gatt_test.py | 4 +++- 3 files changed, 20 insertions(+), 7 deletions(-) diff --git a/bumble/device.py b/bumble/device.py index 9240fe93..509ba7ec 100644 --- a/bumble/device.py +++ b/bumble/device.py @@ -2260,6 +2260,7 @@ class DeviceConfiguration: l2cap.L2CAP_Information_Request.ExtendedFeatures.ENHANCED_RETRANSMISSION_MODE, ) eatt_enabled: bool = False + gatt_strict_permissions: bool = False gatt_services: list[dict[str, Any]] = field(init=False) smp_debug_mode: bool = False diff --git a/bumble/gatt_server.py b/bumble/gatt_server.py index 0a9e4297..ddaf2a6c 100644 --- a/bumble/gatt_server.py +++ b/bumble/gatt_server.py @@ -113,8 +113,6 @@ def __init__(self, device: Device) -> None: self.max_mtu = ( GATT_SERVER_DEFAULT_MAX_MTU # The max MTU we're willing to negotiate ) - # When False, READABLE and WRITEABLE are not required for peer reads and writes - self.strict_permissions = True self.subscribers = ( {} ) # Map of subscriber states by connection handle and attribute handle @@ -390,7 +388,10 @@ async def _read_attribute_value( self, bearer: att.Bearer, attribute: att.Attribute ) -> bytes: # Read a value on behalf of a peer - if self.strict_permissions and not attribute.permissions & _READ_PERMISSIONS: + if ( + self.device.config.gatt_strict_permissions + and not attribute.permissions & _READ_PERMISSIONS + ): raise att.ATT_Error( error_code=att.ATT_READ_NOT_PERMITTED_ERROR, att_handle=attribute.handle ) @@ -1148,7 +1149,10 @@ async def on_att_write_request( return # Check that the attribute can be written - if self.strict_permissions and not attribute.permissions & _WRITE_PERMISSIONS: + if ( + self.device.config.gatt_strict_permissions + and not attribute.permissions & _WRITE_PERMISSIONS + ): self.send_response( bearer, att.ATT_Error_Response( @@ -1200,7 +1204,10 @@ async def on_att_write_command( return # Check that the attribute can be written - if self.strict_permissions and not attribute.permissions & _WRITE_PERMISSIONS: + if ( + self.device.config.gatt_strict_permissions + and not attribute.permissions & _WRITE_PERMISSIONS + ): return # Check the request parameters @@ -1234,7 +1241,10 @@ def on_att_prepare_write_request( return # Check that the attribute can be written - if self.strict_permissions and not attribute.permissions & _WRITE_PERMISSIONS: + if ( + self.device.config.gatt_strict_permissions + and not attribute.permissions & _WRITE_PERMISSIONS + ): self.send_response( bearer, att.ATT_Error_Response( diff --git a/tests/gatt_test.py b/tests/gatt_test.py index dfc36060..02d6d1bc 100644 --- a/tests/gatt_test.py +++ b/tests/gatt_test.py @@ -1869,6 +1869,7 @@ async def test_write_long_value_gap_rejected(): @pytest.mark.asyncio async def test_write_not_permitted(): devices = await TwoDevices.create_with_connection() + devices[1].config.gatt_strict_permissions = True characteristic = Characteristic( '1234', Characteristic.Properties.READ, Characteristic.READABLE, b'1234' @@ -1912,6 +1913,7 @@ async def test_write_not_permitted(): @pytest.mark.asyncio async def test_read_not_permitted(): devices = await TwoDevices.create_with_connection() + devices[1].config.gatt_strict_permissions = True characteristic = Characteristic( '1234', Characteristic.Properties.WRITE, Characteristic.WRITEABLE, b'1234' @@ -1954,6 +1956,7 @@ async def test_read_not_permitted(): @pytest.mark.asyncio async def test_requires_permissions_imply_access(): devices = await TwoDevices.create_with_connection() + devices[1].config.gatt_strict_permissions = True # No READABLE or WRITEABLE bit, only the security requirements characteristic = Characteristic( @@ -2002,7 +2005,6 @@ async def test_requires_permissions_imply_access(): @pytest.mark.asyncio async def test_permissions_not_strict(): devices = await TwoDevices.create_with_connection() - devices[1].gatt_server.strict_permissions = False characteristic = Characteristic( '1234', Characteristic.Properties.NOTIFY, Characteristic.Permissions(0), b'1234'