gatt: Add ATT long-write (Prepare/Execute Write) support

bumble/gatt_client.py

@@ -1096,15 +1096,21 @@ async def write_value(
         with_response: bool = False,
     ) -> None:
         '''
-        See Vol 3, Part G - 4.9.1 Write Without Response & 4.9.3 Write Characteristic
-        Value
+        See Vol 3, Part G - 4.9.1 Write Without Response, 4.9.3 Write Characteristic
+        Value & 4.9.4 Write Long Characteristic Values
 
         `attribute` can be an Attribute object, or a handle value
         '''
 
         # Send a request or command to write
         attribute_handle = attribute if isinstance(attribute, int) else attribute.handle
         if with_response:
+            # Use the Write Long procedure when the value does not fit in a single
+            # Write Request
+            if len(value) > self.mtu - 3:
+                await self.write_long_value(attribute_handle, value)
+                return
+
             response = await self.send_request(
                 att.ATT_Write_Request(
                     attribute_handle=attribute_handle, attribute_value=value
@@ -1119,6 +1125,36 @@ async def write_value(
                 )
             )
 
+    async def write_long_value(self, attribute_handle: int, value: bytes) -> None:
+        '''
+        See Vol 3, Part G - 4.9.4 Write Long Characteristic Values
+
+        Writes a value that does not fit in a single Write Request by queuing it
+        with Prepare Write Requests and committing it with an Execute Write Request.
+        '''
+
+        # Each Prepare Write Request carries opcode(1) + handle(2) + offset(2), so
+        # the part value is limited to ATT_MTU - 5 bytes.
+        max_part_size = self.mtu - 5
+        for offset in range(0, len(value), max_part_size):
+            part = value[offset : offset + max_part_size]
+            response = await self.send_request(
+                att.ATT_Prepare_Write_Request(
+                    attribute_handle=attribute_handle,
+                    value_offset=offset,
+                    part_attribute_value=part,
+                )
+            )
+            if response.op_code == att.Opcode.ATT_ERROR_RESPONSE:
+                # Ask the server to drop whatever it has already queued
+                await self.send_request(att.ATT_Execute_Write_Request(flags=0x00))
+                raise att.ATT_Error(error_code=response.error_code, message=response)
+
+        # Commit the queued values (flags=0x01 = write all pending prepared values)
+        response = await self.send_request(att.ATT_Execute_Write_Request(flags=0x01))
+        if response.op_code == att.Opcode.ATT_ERROR_RESPONSE:
+            raise att.ATT_Error(error_code=response.error_code, message=response)
+
     def on_disconnection(self, *args) -> None:
         del args  # unused.
         if self.pending_response and not self.pending_response.done():

bumble/gatt_server.py

@@ -104,6 +104,8 @@ def __init__(self, device: Device) -> None:
         )  # Map of subscriber states by connection handle and attribute handle
         self.indication_semaphores = defaultdict(lambda: asyncio.Semaphore(1))
         self.pending_confirmations = defaultdict(lambda: None)
+        # Queued prepared writes, per bearer
+        self.prepared_writes: dict[att.Bearer, list[tuple[int, int, bytes]]] = {}
 
     def __str__(self) -> str:
         return "\n".join(map(str, self.attributes))
@@ -559,6 +561,7 @@ def on_disconnection(self, bearer: att.Bearer) -> None:
         self.subscribers.pop(bearer, None)
         self.indication_semaphores.pop(bearer, None)
         self.pending_confirmations.pop(bearer, None)
+        self.prepared_writes.pop(bearer, None)
 
     def on_gatt_pdu(self, bearer: att.Bearer, att_pdu: att.ATT_PDU) -> None:
         logger.debug(f'GATT Request to server: {_bearer_id(bearer)} {att_pdu}')
@@ -1153,6 +1156,88 @@ async def on_att_write_command(
         except Exception:
             logger.exception('!!! ignoring exception')
 
+    def on_att_prepare_write_request(
+        self, bearer: att.Bearer, request: att.ATT_Prepare_Write_Request
+    ):
+        '''
+        See Bluetooth spec Vol 3, Part F - 3.4.6.1 Prepare Write Request
+        '''
+
+        # Check that the attribute exists
+        if self.get_attribute(request.attribute_handle) is None:
+            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_INVALID_HANDLE_ERROR,
+                ),
+            )
+            return
+
+        # Queue the partial value, to be committed on Execute Write Request
+        self.prepared_writes.setdefault(bearer, []).append(
+            (
+                request.attribute_handle,
+                request.value_offset,
+                request.part_attribute_value,
+            )
+        )
+
+        # Acknowledge by echoing back the received part
+        self.send_response(
+            bearer,
+            att.ATT_Prepare_Write_Response(
+                attribute_handle=request.attribute_handle,
+                value_offset=request.value_offset,
+                part_attribute_value=request.part_attribute_value,
+            ),
+        )
+
+    @utils.AsyncRunner.run_in_task()
+    async def on_att_execute_write_request(
+        self, bearer: att.Bearer, request: att.ATT_Execute_Write_Request
+    ):
+        '''
+        See Bluetooth spec Vol 3, Part F - 3.4.6.3 Execute Write Request
+        '''
+
+        queue = self.prepared_writes.pop(bearer, [])
+
+        # flags == 0x00 means cancel all prepared writes
+        if request.flags == 0:
+            self.send_response(bearer, att.ATT_Execute_Write_Response())
+            return
+
+        try:
+            # Reassemble the queued parts; reject a gap with INVALID_OFFSET
+            values: dict[int, bytearray] = {}
+            for handle, offset, part in queue:
+                buffer = values.setdefault(handle, bytearray())
+                if offset > len(buffer):
+                    raise att.ATT_Error(
+                        error_code=att.ATT_INVALID_OFFSET_ERROR, att_handle=handle
+                    )
+                buffer[offset : offset + len(part)] = part
+
+            # Commit the reassembled values through the normal write path
+            for handle, value in values.items():
+                attribute = self.get_attribute(handle)
+                assert attribute is not None
+                await attribute.write_value(bearer, bytes(value))
+        except att.ATT_Error as error:
+            self.send_response(
+                bearer,
+                att.ATT_Error_Response(
+                    request_opcode_in_error=request.op_code,
+                    attribute_handle_in_error=error.att_handle,
+                    error_code=error.error_code,
+                ),
+            )
+            return
+
+        self.send_response(bearer, att.ATT_Execute_Write_Response())
+
     def on_att_handle_value_confirmation(
         self,
         bearer: att.Bearer,

tests/gatt_test.py

@@ -1736,6 +1736,136 @@ async def test_read_multiple_variable() -> None:
     assert response.error_code == att.ATT_ATTRIBUTE_NOT_FOUND_ERROR
 
 
+# -----------------------------------------------------------------------------
+async def _connect_with_writeable_characteristic():
+    devices = await TwoDevices.create_with_connection()
+    [_, server] = devices
+
+    characteristic = Characteristic(
+        'FDB159DB-036C-49E3-B3DB-6325AC750806',
+        Characteristic.Properties.READ | Characteristic.Properties.WRITE,
+        Characteristic.READABLE | Characteristic.WRITEABLE,
+    )
+    server.add_service(
+        Service('3A657F47-D34F-46B3-B1EC-698E29B6B829', [characteristic])
+    )
+
+    peer = Peer(devices.connections[0])
+    await peer.discover_services()
+    await peer.discover_characteristics()
+    [proxy] = peer.get_characteristics_by_uuid(characteristic.uuid)
+    return peer, proxy, characteristic
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_write_long_value():
+    peer, proxy, characteristic = await _connect_with_writeable_characteristic()
+
+    # A value that does not fit in a single Write Request (ATT_MTU - 3)
+    long_value = bytes((i * 7) % 256 for i in range(149))
+    assert len(long_value) > peer.gatt_client.mtu - 3
+
+    # Record the request opcodes the client emits
+    sent_opcodes = []
+    inner_send_request = peer.gatt_client.send_request
+
+    async def recording_send_request(request):
+        sent_opcodes.append(request.op_code)
+        return await inner_send_request(request)
+
+    peer.gatt_client.send_request = recording_send_request
+
+    await proxy.write_value(long_value, with_response=True)
+    await async_barrier()
+
+    # The client used the Write Long procedure, not a single Write Request
+    assert att.Opcode.ATT_PREPARE_WRITE_REQUEST in sent_opcodes
+    assert att.Opcode.ATT_EXECUTE_WRITE_REQUEST in sent_opcodes
+    assert att.Opcode.ATT_WRITE_REQUEST not in sent_opcodes
+
+    # The server reassembled all the queued parts into the complete value
+    assert characteristic.value == long_value
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_write_long_value_manual():
+    peer, _, characteristic = await _connect_with_writeable_characteristic()
+    client = peer.gatt_client
+
+    long_value = bytes((i * 3) % 256 for i in range(60))
+
+    # Queue the value in three parts at increasing offsets
+    for offset in range(0, len(long_value), 20):
+        part = long_value[offset : offset + 20]
+        response = await client.send_request(
+            att.ATT_Prepare_Write_Request(
+                attribute_handle=characteristic.handle,
+                value_offset=offset,
+                part_attribute_value=part,
+            )
+        )
+        assert isinstance(response, att.ATT_Prepare_Write_Response)
+        assert response.value_offset == offset
+        assert response.part_attribute_value == part
+
+    # Commit the queued values (flags=0x01)
+    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 == long_value
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_write_long_value_cancel():
+    peer, _, characteristic = await _connect_with_writeable_characteristic()
+    client = peer.gatt_client
+
+    # Queue a value, then cancel instead of committing (flags=0x00)
+    response = await client.send_request(
+        att.ATT_Prepare_Write_Request(
+            attribute_handle=characteristic.handle,
+            value_offset=0,
+            part_attribute_value=bytes([1, 2, 3, 4]),
+        )
+    )
+    assert isinstance(response, att.ATT_Prepare_Write_Response)
+
+    response = await client.send_request(att.ATT_Execute_Write_Request(flags=0x00))
+    assert isinstance(response, att.ATT_Execute_Write_Response)
+    await async_barrier()
+
+    # The cancelled value was not written (the attribute keeps its initial value)
+    assert characteristic.value is None
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_write_long_value_gap_rejected():
+    peer, _, characteristic = await _connect_with_writeable_characteristic()
+    client = peer.gatt_client
+
+    # Queue parts leaving a gap between them (offset 0 then offset 10)
+    for offset in (0, 10):
+        response = await client.send_request(
+            att.ATT_Prepare_Write_Request(
+                attribute_handle=characteristic.handle,
+                value_offset=offset,
+                part_attribute_value=bytes([1, 2, 3, 4]),
+            )
+        )
+        assert isinstance(response, att.ATT_Prepare_Write_Response)
+
+    # Committing a non-contiguous queue is rejected with INVALID_OFFSET
+    response = await client.send_request(att.ATT_Execute_Write_Request(flags=0x01))
+    assert isinstance(response, att.ATT_Error_Response)
+    assert response.error_code == att.ATT_INVALID_OFFSET_ERROR
+    await async_barrier()
+    assert characteristic.value is None
+
+
 # -----------------------------------------------------------------------------
 if __name__ == '__main__':
     logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper())