| # Copyright 2021-2022 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. |
| |
| # ----------------------------------------------------------------------------- |
| # GATT - Generic Attribute Profile |
| # Server |
| # |
| # See Bluetooth spec @ Vol 3, Part G |
| # |
| # ----------------------------------------------------------------------------- |
| |
| # ----------------------------------------------------------------------------- |
| # Imports |
| # ----------------------------------------------------------------------------- |
| import asyncio |
| import logging |
| from collections import defaultdict |
| import struct |
| from typing import List, Tuple, Optional |
| from pyee import EventEmitter |
| |
| from .colors import color |
| from .core import UUID |
| from .att import ( |
| ATT_ATTRIBUTE_NOT_FOUND_ERROR, |
| ATT_ATTRIBUTE_NOT_LONG_ERROR, |
| ATT_CID, |
| ATT_DEFAULT_MTU, |
| ATT_INVALID_ATTRIBUTE_LENGTH_ERROR, |
| ATT_INVALID_HANDLE_ERROR, |
| ATT_INVALID_OFFSET_ERROR, |
| ATT_REQUEST_NOT_SUPPORTED_ERROR, |
| ATT_REQUESTS, |
| ATT_UNLIKELY_ERROR_ERROR, |
| ATT_UNSUPPORTED_GROUP_TYPE_ERROR, |
| ATT_Error, |
| ATT_Error_Response, |
| ATT_Exchange_MTU_Response, |
| ATT_Find_By_Type_Value_Response, |
| ATT_Find_Information_Response, |
| ATT_Handle_Value_Indication, |
| ATT_Handle_Value_Notification, |
| ATT_Read_Blob_Response, |
| ATT_Read_By_Group_Type_Response, |
| ATT_Read_By_Type_Response, |
| ATT_Read_Response, |
| ATT_Write_Response, |
| Attribute, |
| ) |
| from .gatt import ( |
| GATT_CHARACTERISTIC_ATTRIBUTE_TYPE, |
| GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR, |
| GATT_MAX_ATTRIBUTE_VALUE_SIZE, |
| GATT_PRIMARY_SERVICE_ATTRIBUTE_TYPE, |
| GATT_REQUEST_TIMEOUT, |
| GATT_SECONDARY_SERVICE_ATTRIBUTE_TYPE, |
| Characteristic, |
| CharacteristicDeclaration, |
| CharacteristicValue, |
| Descriptor, |
| Service, |
| ) |
| |
| |
| # ----------------------------------------------------------------------------- |
| # Logging |
| # ----------------------------------------------------------------------------- |
| logger = logging.getLogger(__name__) |
| |
| |
| # ----------------------------------------------------------------------------- |
| # Constants |
| # ----------------------------------------------------------------------------- |
| GATT_SERVER_DEFAULT_MAX_MTU = 517 |
| |
| |
| # ----------------------------------------------------------------------------- |
| # GATT Server |
| # ----------------------------------------------------------------------------- |
| class Server(EventEmitter): |
| attributes: List[Attribute] |
| |
| def __init__(self, device): |
| super().__init__() |
| self.device = device |
| self.attributes = [] # Attributes, ordered by increasing handle values |
| self.attributes_by_handle = {} # Map for fast attribute access by handle |
| self.max_mtu = ( |
| GATT_SERVER_DEFAULT_MAX_MTU # The max MTU we're willing to negotiate |
| ) |
| self.subscribers = ( |
| {} |
| ) # Map of subscriber states by connection handle and attribute handle |
| self.indication_semaphores = defaultdict(lambda: asyncio.Semaphore(1)) |
| self.pending_confirmations = defaultdict(lambda: None) |
| |
| def __str__(self): |
| return "\n".join(map(str, self.attributes)) |
| |
| def send_gatt_pdu(self, connection_handle, pdu): |
| self.device.send_l2cap_pdu(connection_handle, ATT_CID, pdu) |
| |
| def next_handle(self): |
| return 1 + len(self.attributes) |
| |
| def get_advertising_service_data(self): |
| return { |
| attribute: data |
| for attribute in self.attributes |
| if isinstance(attribute, Service) |
| and (data := attribute.get_advertising_data()) |
| } |
| |
| def get_attribute(self, handle): |
| attribute = self.attributes_by_handle.get(handle) |
| if attribute: |
| return attribute |
| |
| # Not in the cached map, perform a linear lookup |
| for attribute in self.attributes: |
| if attribute.handle == handle: |
| # Store in cached map |
| self.attributes_by_handle[handle] = attribute |
| return attribute |
| return None |
| |
| def get_service_attribute(self, service_uuid: UUID) -> Optional[Service]: |
| return next( |
| ( |
| attribute |
| for attribute in self.attributes |
| if attribute.type == GATT_PRIMARY_SERVICE_ATTRIBUTE_TYPE |
| and isinstance(attribute, Service) |
| and attribute.uuid == service_uuid |
| ), |
| None, |
| ) |
| |
| def get_characteristic_attributes( |
| self, service_uuid: UUID, characteristic_uuid: UUID |
| ) -> Optional[Tuple[CharacteristicDeclaration, Characteristic]]: |
| service_handle = self.get_service_attribute(service_uuid) |
| if not service_handle: |
| return None |
| |
| return next( |
| ( |
| (attribute, self.get_attribute(attribute.characteristic.handle)) |
| for attribute in map( |
| self.get_attribute, |
| range(service_handle.handle, service_handle.end_group_handle + 1), |
| ) |
| if attribute.type == GATT_CHARACTERISTIC_ATTRIBUTE_TYPE |
| and attribute.characteristic.uuid == characteristic_uuid |
| ), |
| None, |
| ) |
| |
| def get_descriptor_attribute( |
| self, service_uuid: UUID, characteristic_uuid: UUID, descriptor_uuid: UUID |
| ) -> Optional[Descriptor]: |
| characteristics = self.get_characteristic_attributes( |
| service_uuid, characteristic_uuid |
| ) |
| if not characteristics: |
| return None |
| |
| (_, characteristic_value) = characteristics |
| |
| return next( |
| ( |
| attribute |
| for attribute in map( |
| self.get_attribute, |
| range( |
| characteristic_value.handle + 1, |
| characteristic_value.end_group_handle + 1, |
| ), |
| ) |
| if attribute.type == descriptor_uuid |
| ), |
| None, |
| ) |
| |
| def add_attribute(self, attribute): |
| # Assign a handle to this attribute |
| attribute.handle = self.next_handle() |
| attribute.end_group_handle = ( |
| attribute.handle |
| ) # TODO: keep track of descriptors in the group |
| |
| # Add this attribute to the list |
| self.attributes.append(attribute) |
| |
| def add_service(self, service: Service): |
| # Add the service attribute to the DB |
| self.add_attribute(service) |
| |
| # TODO: add included services |
| |
| # Add all characteristics |
| for characteristic in service.characteristics: |
| # Add a Characteristic Declaration |
| characteristic_declaration = CharacteristicDeclaration( |
| characteristic, self.next_handle() + 1 |
| ) |
| self.add_attribute(characteristic_declaration) |
| |
| # Add the characteristic value |
| self.add_attribute(characteristic) |
| |
| # Add the descriptors |
| for descriptor in characteristic.descriptors: |
| self.add_attribute(descriptor) |
| |
| # If the characteristic supports subscriptions, add a CCCD descriptor |
| # unless there is one already |
| if ( |
| characteristic.properties |
| & (Characteristic.NOTIFY | Characteristic.INDICATE) |
| and characteristic.get_descriptor( |
| GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR |
| ) |
| is None |
| ): |
| self.add_attribute( |
| # pylint: disable=line-too-long |
| Descriptor( |
| GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR, |
| Attribute.READABLE | Attribute.WRITEABLE, |
| CharacteristicValue( |
| read=lambda connection, characteristic=characteristic: self.read_cccd( |
| connection, characteristic |
| ), |
| write=lambda connection, value, characteristic=characteristic: self.write_cccd( |
| connection, characteristic, value |
| ), |
| ), |
| ) |
| ) |
| |
| # Update the service and characteristic group ends |
| characteristic_declaration.end_group_handle = self.attributes[-1].handle |
| characteristic.end_group_handle = self.attributes[-1].handle |
| |
| # Update the service group end |
| service.end_group_handle = self.attributes[-1].handle |
| |
| def add_services(self, services): |
| for service in services: |
| self.add_service(service) |
| |
| def read_cccd(self, connection, characteristic): |
| if connection is None: |
| return bytes([0, 0]) |
| |
| subscribers = self.subscribers.get(connection.handle) |
| cccd = None |
| if subscribers: |
| cccd = subscribers.get(characteristic.handle) |
| |
| return cccd or bytes([0, 0]) |
| |
| def write_cccd(self, connection, characteristic, value): |
| logger.debug( |
| f'Subscription update for connection=0x{connection.handle:04X}, ' |
| f'handle=0x{characteristic.handle:04X}: {value.hex()}' |
| ) |
| |
| # Sanity check |
| if len(value) != 2: |
| logger.warning('CCCD value not 2 bytes long') |
| return |
| |
| cccds = self.subscribers.setdefault(connection.handle, {}) |
| cccds[characteristic.handle] = value |
| logger.debug(f'CCCDs: {cccds}') |
| notify_enabled = value[0] & 0x01 != 0 |
| indicate_enabled = value[0] & 0x02 != 0 |
| characteristic.emit( |
| 'subscription', connection, notify_enabled, indicate_enabled |
| ) |
| self.emit( |
| 'characteristic_subscription', |
| connection, |
| characteristic, |
| notify_enabled, |
| indicate_enabled, |
| ) |
| |
| def send_response(self, connection, response): |
| logger.debug( |
| f'GATT Response from server: [0x{connection.handle:04X}] {response}' |
| ) |
| self.send_gatt_pdu(connection.handle, response.to_bytes()) |
| |
| async def notify_subscriber(self, connection, attribute, value=None, force=False): |
| # Check if there's a subscriber |
| if not force: |
| subscribers = self.subscribers.get(connection.handle) |
| if not subscribers: |
| logger.debug('not notifying, no subscribers') |
| return |
| cccd = subscribers.get(attribute.handle) |
| if not cccd: |
| logger.debug( |
| f'not notifying, no subscribers for handle {attribute.handle:04X}' |
| ) |
| return |
| if len(cccd) != 2 or (cccd[0] & 0x01 == 0): |
| logger.debug(f'not notifying, cccd={cccd.hex()}') |
| return |
| |
| # Get or encode the value |
| value = ( |
| attribute.read_value(connection) |
| if value is None |
| else attribute.encode_value(value) |
| ) |
| |
| # Truncate if needed |
| if len(value) > connection.att_mtu - 3: |
| value = value[: connection.att_mtu - 3] |
| |
| # Notify |
| notification = ATT_Handle_Value_Notification( |
| attribute_handle=attribute.handle, attribute_value=value |
| ) |
| logger.debug( |
| f'GATT Notify from server: [0x{connection.handle:04X}] {notification}' |
| ) |
| self.send_gatt_pdu(connection.handle, bytes(notification)) |
| |
| async def indicate_subscriber(self, connection, attribute, value=None, force=False): |
| # Check if there's a subscriber |
| if not force: |
| subscribers = self.subscribers.get(connection.handle) |
| if not subscribers: |
| logger.debug('not indicating, no subscribers') |
| return |
| cccd = subscribers.get(attribute.handle) |
| if not cccd: |
| logger.debug( |
| f'not indicating, no subscribers for handle {attribute.handle:04X}' |
| ) |
| return |
| if len(cccd) != 2 or (cccd[0] & 0x02 == 0): |
| logger.debug(f'not indicating, cccd={cccd.hex()}') |
| return |
| |
| # Get or encode the value |
| value = ( |
| attribute.read_value(connection) |
| if value is None |
| else attribute.encode_value(value) |
| ) |
| |
| # Truncate if needed |
| if len(value) > connection.att_mtu - 3: |
| value = value[: connection.att_mtu - 3] |
| |
| # Indicate |
| indication = ATT_Handle_Value_Indication( |
| attribute_handle=attribute.handle, attribute_value=value |
| ) |
| logger.debug( |
| f'GATT Indicate from server: [0x{connection.handle:04X}] {indication}' |
| ) |
| |
| # Wait until we can send (only one pending indication at a time per connection) |
| async with self.indication_semaphores[connection.handle]: |
| assert self.pending_confirmations[connection.handle] is None |
| |
| # Create a future value to hold the eventual response |
| self.pending_confirmations[ |
| connection.handle |
| ] = asyncio.get_running_loop().create_future() |
| |
| try: |
| self.send_gatt_pdu(connection.handle, indication.to_bytes()) |
| await asyncio.wait_for( |
| self.pending_confirmations[connection.handle], GATT_REQUEST_TIMEOUT |
| ) |
| except asyncio.TimeoutError as error: |
| logger.warning(color('!!! GATT Indicate timeout', 'red')) |
| raise TimeoutError(f'GATT timeout for {indication.name}') from error |
| finally: |
| self.pending_confirmations[connection.handle] = None |
| |
| async def notify_or_indicate_subscribers( |
| self, indicate, attribute, value=None, force=False |
| ): |
| # Get all the connections for which there's at least one subscription |
| connections = [ |
| connection |
| for connection in [ |
| self.device.lookup_connection(connection_handle) |
| for (connection_handle, subscribers) in self.subscribers.items() |
| if force or subscribers.get(attribute.handle) |
| ] |
| if connection is not None |
| ] |
| |
| # Indicate or notify for each connection |
| if connections: |
| coroutine = self.indicate_subscriber if indicate else self.notify_subscriber |
| await asyncio.wait( |
| [ |
| asyncio.create_task(coroutine(connection, attribute, value, force)) |
| for connection in connections |
| ] |
| ) |
| |
| async def notify_subscribers(self, attribute, value=None, force=False): |
| return await self.notify_or_indicate_subscribers(False, attribute, value, force) |
| |
| async def indicate_subscribers(self, attribute, value=None, force=False): |
| return await self.notify_or_indicate_subscribers(True, attribute, value, force) |
| |
| def on_disconnection(self, connection): |
| if connection.handle in self.subscribers: |
| del self.subscribers[connection.handle] |
| if connection.handle in self.indication_semaphores: |
| del self.indication_semaphores[connection.handle] |
| if connection.handle in self.pending_confirmations: |
| del self.pending_confirmations[connection.handle] |
| |
| def on_gatt_pdu(self, connection, att_pdu): |
| logger.debug(f'GATT Request to server: [0x{connection.handle:04X}] {att_pdu}') |
| handler_name = f'on_{att_pdu.name.lower()}' |
| handler = getattr(self, handler_name, None) |
| if handler is not None: |
| try: |
| handler(connection, att_pdu) |
| except ATT_Error as error: |
| logger.debug(f'normal exception returned by handler: {error}') |
| response = ATT_Error_Response( |
| request_opcode_in_error=att_pdu.op_code, |
| attribute_handle_in_error=error.att_handle, |
| error_code=error.error_code, |
| ) |
| self.send_response(connection, response) |
| except Exception as error: |
| logger.warning(f'{color("!!! Exception in handler:", "red")} {error}') |
| response = ATT_Error_Response( |
| request_opcode_in_error=att_pdu.op_code, |
| attribute_handle_in_error=0x0000, |
| error_code=ATT_UNLIKELY_ERROR_ERROR, |
| ) |
| self.send_response(connection, response) |
| raise error |
| else: |
| # No specific handler registered |
| if att_pdu.op_code in ATT_REQUESTS: |
| # Invoke the generic handler |
| self.on_att_request(connection, att_pdu) |
| else: |
| # Just ignore |
| logger.warning( |
| color( |
| f'--- Ignoring GATT Request from [0x{connection.handle:04X}]: ', |
| 'red', |
| ) |
| + str(att_pdu) |
| ) |
| |
| ####################################################### |
| # ATT handlers |
| ####################################################### |
| def on_att_request(self, connection, pdu): |
| ''' |
| Handler for requests without a more specific handler |
| ''' |
| logger.warning( |
| color( |
| f'--- Unsupported ATT Request from [0x{connection.handle:04X}]: ', 'red' |
| ) |
| + str(pdu) |
| ) |
| response = ATT_Error_Response( |
| request_opcode_in_error=pdu.op_code, |
| attribute_handle_in_error=0x0000, |
| error_code=ATT_REQUEST_NOT_SUPPORTED_ERROR, |
| ) |
| self.send_response(connection, response) |
| |
| def on_att_exchange_mtu_request(self, connection, request): |
| ''' |
| See Bluetooth spec Vol 3, Part F - 3.4.2.1 Exchange MTU Request |
| ''' |
| self.send_response( |
| connection, ATT_Exchange_MTU_Response(server_rx_mtu=self.max_mtu) |
| ) |
| |
| # Compute the final MTU |
| if request.client_rx_mtu >= ATT_DEFAULT_MTU: |
| mtu = min(self.max_mtu, request.client_rx_mtu) |
| |
| # Notify the device |
| self.device.on_connection_att_mtu_update(connection.handle, mtu) |
| else: |
| logger.warning('invalid client_rx_mtu received, MTU not changed') |
| |
| def on_att_find_information_request(self, connection, request): |
| ''' |
| See Bluetooth spec Vol 3, Part F - 3.4.3.1 Find Information Request |
| ''' |
| |
| # Check the request parameters |
| if ( |
| request.starting_handle == 0 |
| or request.starting_handle > request.ending_handle |
| ): |
| self.send_response( |
| connection, |
| ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.starting_handle, |
| error_code=ATT_INVALID_HANDLE_ERROR, |
| ), |
| ) |
| return |
| |
| # Build list of returned attributes |
| pdu_space_available = connection.att_mtu - 2 |
| attributes = [] |
| uuid_size = 0 |
| for attribute in ( |
| attribute |
| for attribute in self.attributes |
| if attribute.handle >= request.starting_handle |
| and attribute.handle <= request.ending_handle |
| ): |
| this_uuid_size = len(attribute.type.to_pdu_bytes()) |
| |
| if attributes: |
| # Check if this attribute has the same type size as the previous one |
| if this_uuid_size != uuid_size: |
| break |
| |
| # Check if there's enough space for one more entry |
| uuid_size = this_uuid_size |
| if pdu_space_available < 2 + uuid_size: |
| break |
| |
| # Add the attribute to the list |
| attributes.append(attribute) |
| pdu_space_available -= 2 + uuid_size |
| |
| # Return the list of attributes |
| if attributes: |
| information_data_list = [ |
| struct.pack('<H', attribute.handle) + attribute.type.to_pdu_bytes() |
| for attribute in attributes |
| ] |
| response = ATT_Find_Information_Response( |
| format=1 if len(attributes[0].type.to_pdu_bytes()) == 2 else 2, |
| information_data=b''.join(information_data_list), |
| ) |
| else: |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.starting_handle, |
| error_code=ATT_ATTRIBUTE_NOT_FOUND_ERROR, |
| ) |
| |
| self.send_response(connection, response) |
| |
| def on_att_find_by_type_value_request(self, connection, request): |
| ''' |
| See Bluetooth spec Vol 3, Part F - 3.4.3.3 Find By Type Value Request |
| ''' |
| |
| # Build list of returned attributes |
| pdu_space_available = connection.att_mtu - 2 |
| attributes = [] |
| 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 attribute.read_value(connection) == request.attribute_value |
| and pdu_space_available >= 4 |
| ): |
| # TODO: check permissions |
| |
| # Add the attribute to the list |
| attributes.append(attribute) |
| pdu_space_available -= 4 |
| |
| # Return the list of attributes |
| if attributes: |
| handles_information_list = [] |
| for attribute in attributes: |
| if attribute.type in ( |
| GATT_PRIMARY_SERVICE_ATTRIBUTE_TYPE, |
| GATT_SECONDARY_SERVICE_ATTRIBUTE_TYPE, |
| GATT_CHARACTERISTIC_ATTRIBUTE_TYPE, |
| ): |
| # Part of a group |
| group_end_handle = attribute.end_group_handle |
| else: |
| # Not part of a group |
| group_end_handle = attribute.handle |
| handles_information_list.append( |
| struct.pack('<HH', attribute.handle, group_end_handle) |
| ) |
| response = ATT_Find_By_Type_Value_Response( |
| handles_information_list=b''.join(handles_information_list) |
| ) |
| else: |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.starting_handle, |
| error_code=ATT_ATTRIBUTE_NOT_FOUND_ERROR, |
| ) |
| |
| self.send_response(connection, response) |
| |
| def on_att_read_by_type_request(self, connection, request): |
| ''' |
| See Bluetooth spec Vol 3, Part F - 3.4.4.1 Read By Type Request |
| ''' |
| |
| pdu_space_available = connection.att_mtu - 2 |
| |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.starting_handle, |
| error_code=ATT_ATTRIBUTE_NOT_FOUND_ERROR, |
| ) |
| |
| attributes = [] |
| for attribute in ( |
| attribute |
| for attribute in self.attributes |
| if attribute.type == request.attribute_type |
| and attribute.handle >= request.starting_handle |
| and attribute.handle <= request.ending_handle |
| and pdu_space_available |
| ): |
| |
| try: |
| attribute_value = attribute.read_value(connection) |
| except ATT_Error as error: |
| # If the first attribute is unreadable, return an error |
| # Otherwise return attributes up to this point |
| if not attributes: |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=attribute.handle, |
| error_code=error.error_code, |
| ) |
| break |
| |
| # Check the attribute value size |
| max_attribute_size = min(connection.att_mtu - 4, 253) |
| if len(attribute_value) > max_attribute_size: |
| # We need to truncate |
| attribute_value = attribute_value[:max_attribute_size] |
| if attributes and len(attributes[0][1]) != len(attribute_value): |
| # Not the same size as previous attribute, stop here |
| break |
| |
| # Check if there is enough space |
| entry_size = 2 + len(attribute_value) |
| if pdu_space_available < entry_size: |
| break |
| |
| # Add the attribute to the list |
| attributes.append((attribute.handle, attribute_value)) |
| pdu_space_available -= entry_size |
| |
| if attributes: |
| attribute_data_list = [ |
| struct.pack('<H', handle) + value for handle, value in attributes |
| ] |
| response = ATT_Read_By_Type_Response( |
| length=entry_size, attribute_data_list=b''.join(attribute_data_list) |
| ) |
| else: |
| logging.debug(f"not found {request}") |
| |
| self.send_response(connection, response) |
| |
| def on_att_read_request(self, connection, request): |
| ''' |
| See Bluetooth spec Vol 3, Part F - 3.4.4.3 Read Request |
| ''' |
| |
| if attribute := self.get_attribute(request.attribute_handle): |
| try: |
| value = attribute.read_value(connection) |
| except ATT_Error as error: |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.attribute_handle, |
| error_code=error.error_code, |
| ) |
| else: |
| value_size = min(connection.att_mtu - 1, len(value)) |
| response = ATT_Read_Response(attribute_value=value[:value_size]) |
| else: |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.attribute_handle, |
| error_code=ATT_INVALID_HANDLE_ERROR, |
| ) |
| self.send_response(connection, response) |
| |
| def on_att_read_blob_request(self, connection, request): |
| ''' |
| See Bluetooth spec Vol 3, Part F - 3.4.4.5 Read Blob Request |
| ''' |
| |
| if attribute := self.get_attribute(request.attribute_handle): |
| try: |
| value = attribute.read_value(connection) |
| except ATT_Error as error: |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.attribute_handle, |
| error_code=error.error_code, |
| ) |
| else: |
| if request.value_offset > len(value): |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.attribute_handle, |
| error_code=ATT_INVALID_OFFSET_ERROR, |
| ) |
| elif len(value) <= connection.att_mtu - 1: |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.attribute_handle, |
| error_code=ATT_ATTRIBUTE_NOT_LONG_ERROR, |
| ) |
| else: |
| part_size = min( |
| connection.att_mtu - 1, len(value) - request.value_offset |
| ) |
| response = ATT_Read_Blob_Response( |
| part_attribute_value=value[ |
| request.value_offset : request.value_offset + part_size |
| ] |
| ) |
| else: |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.attribute_handle, |
| error_code=ATT_INVALID_HANDLE_ERROR, |
| ) |
| self.send_response(connection, response) |
| |
| def on_att_read_by_group_type_request(self, connection, request): |
| ''' |
| See Bluetooth spec Vol 3, Part F - 3.4.4.9 Read by Group Type Request |
| ''' |
| if request.attribute_group_type not in ( |
| GATT_PRIMARY_SERVICE_ATTRIBUTE_TYPE, |
| GATT_SECONDARY_SERVICE_ATTRIBUTE_TYPE, |
| ): |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.starting_handle, |
| error_code=ATT_UNSUPPORTED_GROUP_TYPE_ERROR, |
| ) |
| self.send_response(connection, response) |
| return |
| |
| pdu_space_available = connection.att_mtu - 2 |
| attributes = [] |
| for attribute in ( |
| attribute |
| for attribute in self.attributes |
| if attribute.type == request.attribute_group_type |
| and attribute.handle >= request.starting_handle |
| and attribute.handle <= request.ending_handle |
| and pdu_space_available |
| ): |
| # No need to catch permission errors here, since these attributes |
| # must all be world-readable |
| attribute_value = attribute.read_value(connection) |
| # Check the attribute value size |
| max_attribute_size = min(connection.att_mtu - 6, 251) |
| if len(attribute_value) > max_attribute_size: |
| # We need to truncate |
| attribute_value = attribute_value[:max_attribute_size] |
| if attributes and len(attributes[0][2]) != len(attribute_value): |
| # Not the same size as previous attributes, stop here |
| break |
| |
| # Check if there is enough space |
| entry_size = 4 + len(attribute_value) |
| if pdu_space_available < entry_size: |
| break |
| |
| # Add the attribute to the list |
| attributes.append( |
| (attribute.handle, attribute.end_group_handle, attribute_value) |
| ) |
| pdu_space_available -= entry_size |
| |
| if attributes: |
| attribute_data_list = [ |
| struct.pack('<HH', handle, end_group_handle) + value |
| for handle, end_group_handle, value in attributes |
| ] |
| response = ATT_Read_By_Group_Type_Response( |
| length=len(attribute_data_list[0]), |
| attribute_data_list=b''.join(attribute_data_list), |
| ) |
| else: |
| response = ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.starting_handle, |
| error_code=ATT_ATTRIBUTE_NOT_FOUND_ERROR, |
| ) |
| |
| self.send_response(connection, response) |
| |
| def on_att_write_request(self, connection, request): |
| ''' |
| See Bluetooth spec Vol 3, Part F - 3.4.5.1 Write Request |
| ''' |
| |
| # Check that the attribute exists |
| attribute = self.get_attribute(request.attribute_handle) |
| if attribute is None: |
| self.send_response( |
| connection, |
| ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.attribute_handle, |
| error_code=ATT_INVALID_HANDLE_ERROR, |
| ), |
| ) |
| return |
| |
| # TODO: check permissions |
| |
| # Check the request parameters |
| if len(request.attribute_value) > GATT_MAX_ATTRIBUTE_VALUE_SIZE: |
| self.send_response( |
| connection, |
| ATT_Error_Response( |
| request_opcode_in_error=request.op_code, |
| attribute_handle_in_error=request.attribute_handle, |
| error_code=ATT_INVALID_ATTRIBUTE_LENGTH_ERROR, |
| ), |
| ) |
| return |
| |
| # Accept the value |
| attribute.write_value(connection, request.attribute_value) |
| |
| # Done |
| self.send_response(connection, ATT_Write_Response()) |
| |
| def on_att_write_command(self, connection, request): |
| ''' |
| See Bluetooth spec Vol 3, Part F - 3.4.5.3 Write Command |
| ''' |
| |
| # Check that the attribute exists |
| attribute = self.get_attribute(request.attribute_handle) |
| if attribute is None: |
| return |
| |
| # TODO: check permissions |
| |
| # Check the request parameters |
| if len(request.attribute_value) > GATT_MAX_ATTRIBUTE_VALUE_SIZE: |
| return |
| |
| # Accept the value |
| try: |
| attribute.write_value(connection, request.attribute_value) |
| except Exception as error: |
| logger.warning(f'!!! ignoring exception: {error}') |
| |
| def on_att_handle_value_confirmation(self, connection, _confirmation): |
| ''' |
| See Bluetooth spec Vol 3, Part F - 3.4.7.3 Handle Value Confirmation |
| ''' |
| if self.pending_confirmations[connection.handle] is None: |
| # Not expected! |
| logger.warning( |
| '!!! unexpected confirmation, there is no pending indication' |
| ) |
| return |
| |
| self.pending_confirmations[connection.handle].set_result(None) |