pstruebi/bumble_fork
1
0
Fork 0
forked from auracaster/bumble_mirror
Code Pull Requests Activity

Return ATT_Error_Response on rejected write

Browse Source
This commit is contained in:
Josh Wu
2024-09-07 16:46:16 +08:00
parent 7b34bb4050
commit 8051c23375
4 changed files with 100 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
bumble/att.py
View File

@@ -41,6 +41,7 @@ from typing import (
from pyee import EventEmitter from pyee import EventEmitter
from bumble import utils
from bumble.core import UUID, name_or_number, ProtocolError from bumble.core import UUID, name_or_number, ProtocolError
from bumble.hci import HCI_Object, key_with_value from bumble.hci import HCI_Object, key_with_value
from bumble.colors import color from bumble.colors import color
@@ -145,43 +146,57 @@ ATT_RESPONSES = [
ATT_EXECUTE_WRITE_RESPONSE ATT_EXECUTE_WRITE_RESPONSE
] ]
ATT_INVALID_HANDLE_ERROR = 0x01 class ErrorCode(utils.OpenIntEnum):
ATT_READ_NOT_PERMITTED_ERROR = 0x02 '''
ATT_WRITE_NOT_PERMITTED_ERROR = 0x03 See
ATT_INVALID_PDU_ERROR = 0x04
ATT_INSUFFICIENT_AUTHENTICATION_ERROR = 0x05
ATT_REQUEST_NOT_SUPPORTED_ERROR = 0x06
ATT_INVALID_OFFSET_ERROR = 0x07
ATT_INSUFFICIENT_AUTHORIZATION_ERROR = 0x08
ATT_PREPARE_QUEUE_FULL_ERROR = 0x09
ATT_ATTRIBUTE_NOT_FOUND_ERROR = 0x0A
ATT_ATTRIBUTE_NOT_LONG_ERROR = 0x0B
ATT_INSUFFICIENT_ENCRYPTION_KEY_SIZE_ERROR = 0x0C
ATT_INVALID_ATTRIBUTE_LENGTH_ERROR = 0x0D
ATT_UNLIKELY_ERROR_ERROR = 0x0E
ATT_INSUFFICIENT_ENCRYPTION_ERROR = 0x0F
ATT_UNSUPPORTED_GROUP_TYPE_ERROR = 0x10
ATT_INSUFFICIENT_RESOURCES_ERROR = 0x11
ATT_ERROR_NAMES = { * Bluetooth spec @ Vol 3, Part F - 3.4.1.1 Error Response
ATT_INVALID_HANDLE_ERROR: 'ATT_INVALID_HANDLE_ERROR', * Core Specification Supplement: Common Profile And Service Error Codes
ATT_READ_NOT_PERMITTED_ERROR: 'ATT_READ_NOT_PERMITTED_ERROR', '''
ATT_WRITE_NOT_PERMITTED_ERROR: 'ATT_WRITE_NOT_PERMITTED_ERROR', INVALID_HANDLE = 0x01
ATT_INVALID_PDU_ERROR: 'ATT_INVALID_PDU_ERROR', READ_NOT_PERMITTED = 0x02
ATT_INSUFFICIENT_AUTHENTICATION_ERROR: 'ATT_INSUFFICIENT_AUTHENTICATION_ERROR', WRITE_NOT_PERMITTED = 0x03
ATT_REQUEST_NOT_SUPPORTED_ERROR: 'ATT_REQUEST_NOT_SUPPORTED_ERROR', INVALID_PDU = 0x04
ATT_INVALID_OFFSET_ERROR: 'ATT_INVALID_OFFSET_ERROR', INSUFFICIENT_AUTHENTICATION = 0x05
ATT_INSUFFICIENT_AUTHORIZATION_ERROR: 'ATT_INSUFFICIENT_AUTHORIZATION_ERROR', REQUEST_NOT_SUPPORTED = 0x06
ATT_PREPARE_QUEUE_FULL_ERROR: 'ATT_PREPARE_QUEUE_FULL_ERROR', INVALID_OFFSET = 0x07
ATT_ATTRIBUTE_NOT_FOUND_ERROR: 'ATT_ATTRIBUTE_NOT_FOUND_ERROR', INSUFFICIENT_AUTHORIZATION = 0x08
ATT_ATTRIBUTE_NOT_LONG_ERROR: 'ATT_ATTRIBUTE_NOT_LONG_ERROR', PREPARE_QUEUE_FULL = 0x09
ATT_INSUFFICIENT_ENCRYPTION_KEY_SIZE_ERROR: 'ATT_INSUFFICIENT_ENCRYPTION_KEY_SIZE_ERROR', ATTRIBUTE_NOT_FOUND = 0x0A
ATT_INVALID_ATTRIBUTE_LENGTH_ERROR: 'ATT_INVALID_ATTRIBUTE_LENGTH_ERROR', ATTRIBUTE_NOT_LONG = 0x0B
ATT_UNLIKELY_ERROR_ERROR: 'ATT_UNLIKELY_ERROR_ERROR', INSUFFICIENT_ENCRYPTION_KEY_SIZE = 0x0C
ATT_INSUFFICIENT_ENCRYPTION_ERROR: 'ATT_INSUFFICIENT_ENCRYPTION_ERROR', INVALID_ATTRIBUTE_LENGTH = 0x0D
ATT_UNSUPPORTED_GROUP_TYPE_ERROR: 'ATT_UNSUPPORTED_GROUP_TYPE_ERROR', UNLIKELY_ERROR = 0x0E
ATT_INSUFFICIENT_RESOURCES_ERROR: 'ATT_INSUFFICIENT_RESOURCES_ERROR' INSUFFICIENT_ENCRYPTION = 0x0F
} UNSUPPORTED_GROUP_TYPE = 0x10
INSUFFICIENT_RESOURCES = 0x11
DATABASE_OUT_OF_SYNC = 0x12
VALUE_NOT_ALLOWED = 0x13
# 0x80 – 0x9F: Application Error
# 0xE0 – 0xFF: Common Profile and Service Error Codes
WRITE_REQUEST_REJECTED = 0xFC
CCCD_IMPROPERLY_CONFIGURED = 0xFD
PROCEDURE_ALREADY_IN_PROGRESS = 0xFE
OUT_OF_RANGE = 0xFF
# Backward Compatible Constants
ATT_INVALID_HANDLE_ERROR = ErrorCode.INVALID_HANDLE
ATT_READ_NOT_PERMITTED_ERROR = ErrorCode.READ_NOT_PERMITTED
ATT_WRITE_NOT_PERMITTED_ERROR = ErrorCode.WRITE_NOT_PERMITTED
ATT_INVALID_PDU_ERROR = ErrorCode.INVALID_PDU
ATT_INSUFFICIENT_AUTHENTICATION_ERROR = ErrorCode.INSUFFICIENT_AUTHENTICATION
ATT_REQUEST_NOT_SUPPORTED_ERROR = ErrorCode.REQUEST_NOT_SUPPORTED
ATT_INVALID_OFFSET_ERROR = ErrorCode.INVALID_OFFSET
ATT_INSUFFICIENT_AUTHORIZATION_ERROR = ErrorCode.INSUFFICIENT_AUTHORIZATION
ATT_PREPARE_QUEUE_FULL_ERROR = ErrorCode.PREPARE_QUEUE_FULL
ATT_ATTRIBUTE_NOT_FOUND_ERROR = ErrorCode.ATTRIBUTE_NOT_FOUND
ATT_ATTRIBUTE_NOT_LONG_ERROR = ErrorCode.ATTRIBUTE_NOT_LONG
ATT_INSUFFICIENT_ENCRYPTION_KEY_SIZE_ERROR = ErrorCode.INSUFFICIENT_ENCRYPTION_KEY_SIZE
ATT_INVALID_ATTRIBUTE_LENGTH_ERROR = ErrorCode.INVALID_ATTRIBUTE_LENGTH
ATT_UNLIKELY_ERROR_ERROR = ErrorCode.UNLIKELY_ERROR
ATT_INSUFFICIENT_ENCRYPTION_ERROR = ErrorCode.INSUFFICIENT_ENCRYPTION
ATT_UNSUPPORTED_GROUP_TYPE_ERROR = ErrorCode.UNSUPPORTED_GROUP_TYPE
ATT_INSUFFICIENT_RESOURCES_ERROR = ErrorCode.INSUFFICIENT_RESOURCES
ATT_DEFAULT_MTU = 23 ATT_DEFAULT_MTU = 23
@@ -245,9 +260,9 @@ class ATT_PDU:
def pdu_name(op_code): def pdu_name(op_code):
return name_or_number(ATT_PDU_NAMES, op_code, 2) return name_or_number(ATT_PDU_NAMES, op_code, 2)
@staticmethod @classmethod
def error_name(error_code): def error_name(cls, error_code: int) -> str:
return name_or_number(ATT_ERROR_NAMES, error_code, 2) return ErrorCode(error_code).name
@staticmethod @staticmethod
def subclass(fields): def subclass(fields):

30
bumble/gatt_client.py
View File

@@ -68,7 +68,7 @@ from .att import (
ATT_Error, ATT_Error,
) )
from . import core from . import core
from .core import UUID, InvalidStateError, ProtocolError from .core import UUID, InvalidStateError
from .gatt import ( from .gatt import (
GATT_CHARACTERISTIC_ATTRIBUTE_TYPE, GATT_CHARACTERISTIC_ATTRIBUTE_TYPE,
GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR, GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR,
@@ -345,12 +345,7 @@ class Client:
self.mtu_exchange_done = True self.mtu_exchange_done = True
response = await self.send_request(ATT_Exchange_MTU_Request(client_rx_mtu=mtu)) response = await self.send_request(ATT_Exchange_MTU_Request(client_rx_mtu=mtu))
if response.op_code == ATT_ERROR_RESPONSE: if response.op_code == ATT_ERROR_RESPONSE:
raise ProtocolError( raise ATT_Error(error_code=response.error_code, message=response)
response.error_code,
'att',
ATT_PDU.error_name(response.error_code),
response,
)
# Compute the final MTU # Compute the final MTU
self.connection.att_mtu = min(mtu, response.server_rx_mtu) self.connection.att_mtu = min(mtu, response.server_rx_mtu)
@@ -936,12 +931,7 @@ class Client:
if response is None: if response is None:
raise TimeoutError('read timeout') raise TimeoutError('read timeout')
if response.op_code == ATT_ERROR_RESPONSE: if response.op_code == ATT_ERROR_RESPONSE:
raise ProtocolError( raise ATT_Error(error_code=response.error_code, message=response)
response.error_code,
'att',
ATT_PDU.error_name(response.error_code),
response,
)
# If the value is the max size for the MTU, try to read more unless the caller # If the value is the max size for the MTU, try to read more unless the caller
# specifically asked not to do that # specifically asked not to do that
@@ -963,12 +953,7 @@ class Client:
ATT_INVALID_OFFSET_ERROR, ATT_INVALID_OFFSET_ERROR,
): ):
break break
raise ProtocolError( raise ATT_Error(error_code=response.error_code, message=response)
response.error_code,
'att',
ATT_PDU.error_name(response.error_code),
response,
)
part = response.part_attribute_value part = response.part_attribute_value
attribute_value += part attribute_value += part
@@ -1061,12 +1046,7 @@ class Client:
) )
) )
if response.op_code == ATT_ERROR_RESPONSE: if response.op_code == ATT_ERROR_RESPONSE:
raise ProtocolError( raise ATT_Error(error_code=response.error_code, message=response)
response.error_code,
'att',
ATT_PDU.error_name(response.error_code),
response,
)
else: else:
await self.send_command( await self.send_command(
ATT_Write_Command( ATT_Write_Command(

18
bumble/gatt_server.py
View File

@@ -942,11 +942,19 @@ class Server(EventEmitter):
) )
return return
# Accept the value try:
await attribute.write_value(connection, request.attribute_value) # Accept the value
await attribute.write_value(connection, request.attribute_value)
# Done except ATT_Error as error:
self.send_response(connection, ATT_Write_Response()) response = ATT_Error_Response(
request_opcode_in_error=request.op_code,
attribute_handle_in_error=request.attribute_handle,
error_code=error.error_code,
)
else:
# Done
response = ATT_Write_Response()
self.send_response(connection, response)
@AsyncRunner.run_in_task() @AsyncRunner.run_in_task()
async def on_att_write_command(self, connection, request): async def on_att_write_command(self, connection, request):

28
tests/gatt_test.py
View File

@@ -47,8 +47,10 @@ from bumble.att import (
ATT_EXCHANGE_MTU_REQUEST, ATT_EXCHANGE_MTU_REQUEST,
ATT_ATTRIBUTE_NOT_FOUND_ERROR, ATT_ATTRIBUTE_NOT_FOUND_ERROR,
ATT_PDU, ATT_PDU,
ATT_Error,
ATT_Error_Response, ATT_Error_Response,
ATT_Read_By_Group_Type_Request, ATT_Read_By_Group_Type_Request,
ErrorCode,
) )
from .test_utils import async_barrier from .test_utils import async_barrier
@@ -1247,6 +1249,32 @@ async def test_get_characteristics_by_uuid():
assert len(s) == 1 assert len(s) == 1
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_write_return_error():
[client, server] = LinkedDevices().devices[:2]
on_write = Mock(side_effect=ATT_Error(error_code=ErrorCode.VALUE_NOT_ALLOWED))
characteristic = Characteristic(
'1234',
Characteristic.Properties.WRITE,
Characteristic.Permissions.WRITEABLE,
CharacteristicValue(write=on_write),
)
service = Service('ABCD', [characteristic])
server.add_service(service)
await client.power_on()
await server.power_on()
connection = await client.connect(server.random_address)
async with Peer(connection) as peer:
c = peer.get_characteristics_by_uuid(uuid=UUID('1234'))[0]
with pytest.raises(ATT_Error) as e:
await c.write_value(b'', with_response=True)
assert e.value.error_code == ErrorCode.VALUE_NOT_ALLOWED
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
if __name__ == '__main__': if __name__ == '__main__':
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper()) logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper())