diff --git a/bumble/hid.py b/bumble/hid.py index 3a89e49d3..4b12eb4fb 100644 --- a/bumble/hid.py +++ b/bumble/hid.py @@ -1,4 +1,4 @@ -# Copyright 2021-2022 Google LLC +# 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. @@ -17,45 +17,111 @@ # ----------------------------------------------------------------------------- from __future__ import annotations +import abc +import asyncio +import dataclasses import enum import logging import struct -from abc import ABC, abstractmethod -from collections.abc import Callable -from dataclasses import dataclass +from collections.abc import Sequence +from typing import Any, ClassVar, Self, TypeVar from typing_extensions import override -from bumble import device, l2cap, utils -from bumble.core import InvalidStateError, ProtocolError -from bumble.hci import Address +from bumble import core, hci, l2cap, sdp, utils +from bumble import device as device_lib # ----------------------------------------------------------------------------- # Logging # ----------------------------------------------------------------------------- logger = logging.getLogger(__name__) - # ----------------------------------------------------------------------------- # Constants # ----------------------------------------------------------------------------- -# fmt: on HID_CONTROL_PSM = 0x0011 HID_INTERRUPT_PSM = 0x0013 +class HidProtocolError(core.ProtocolError): + result_code: HandshakeMessage.ResultCode + + def __init__(self, result_code: HandshakeMessage.ResultCode): + self.result_code = result_code + super().__init__( + result_code.value, error_namespace="HID", error_name=result_code.name + ) + + +# Report types +class ReportType(utils.OpenIntEnum): + OTHER_REPORT = 0x00 + INPUT_REPORT = 0x01 + OUTPUT_REPORT = 0x02 + FEATURE_REPORT = 0x03 + + +# Protocol modes +class ProtocolMode(utils.OpenIntEnum): + BOOT_PROTOCOL = 0x00 + REPORT_PROTOCOL = 0x01 + + +# Messages class Message: - message_type: MessageType + """Base class for HID messages.""" - # Report types - class ReportType(enum.IntEnum): - OTHER_REPORT = 0x00 - INPUT_REPORT = 0x01 - OUTPUT_REPORT = 0x02 - FEATURE_REPORT = 0x03 + class Type(utils.OpenIntEnum): + HANDSHAKE = 0x00 + CONTROL = 0x01 + GET_REPORT = 0x04 + SET_REPORT = 0x05 + GET_PROTOCOL = 0x06 + SET_PROTOCOL = 0x07 + GET_IDLE = 0x08 + SET_IDLE = 0x09 + DATA = 0x0A + + message_type: ClassVar[Type] + + subclasses: ClassVar[dict[Type, type[Message]]] = {} - # Handshake parameters - class Handshake(enum.IntEnum): + _MESSAGE_TYPE = TypeVar("_MESSAGE_TYPE", bound="Message") + + @classmethod + def message(cls, subclass: type[_MESSAGE_TYPE]) -> type[_MESSAGE_TYPE]: + cls.subclasses[subclass.message_type] = subclass + return subclass + + # Class Method to derive header + @classmethod + def header(cls, lower_bits: int = 0x00) -> bytes: + return bytes([(cls.message_type << 4) | lower_bits]) + + @classmethod + def from_bytes(cls, data: bytes) -> Message: + if not data: + raise core.InvalidPacketError("Empty HID message") + message_type = Message.Type(data[0] >> 4) + if subclass := cls.subclasses.get(message_type): + return subclass.from_bytes(data) + raise core.InvalidPacketError(f"Unknown message type {message_type.name}") + + def __bytes__(self) -> bytes: + raise NotImplementedError + + +@Message.message +@dataclasses.dataclass +class HandshakeMessage(Message): + """HID Handshake message. + + This message is used to acknowledge or reject control channel requests. + """ + + message_type = Message.Type.HANDSHAKE + + class ResultCode(utils.OpenIntEnum): SUCCESSFUL = 0x00 NOT_READY = 0x01 ERR_INVALID_REPORT_ID = 0x02 @@ -64,140 +130,241 @@ class Handshake(enum.IntEnum): ERR_UNKNOWN = 0x0E ERR_FATAL = 0x0F - # Message Type - class MessageType(enum.IntEnum): - HANDSHAKE = 0x00 - CONTROL = 0x01 - GET_REPORT = 0x04 - SET_REPORT = 0x05 - GET_PROTOCOL = 0x06 - SET_PROTOCOL = 0x07 - DATA = 0x0A + result_code: ResultCode - # Protocol modes - class ProtocolMode(enum.IntEnum): - BOOT_PROTOCOL = 0x00 - REPORT_PROTOCOL = 0x01 + def __bytes__(self) -> bytes: + return self.header(self.result_code) + + @classmethod + def from_bytes(cls, data: bytes) -> HandshakeMessage: + if len(data) != 1: + raise HidProtocolError(cls.ResultCode.ERR_INVALID_PARAMETER) + return cls(result_code=cls.ResultCode(data[0] & 0x0F)) - # Control Operations - class ControlCommand(enum.IntEnum): + +@Message.message +@dataclasses.dataclass +class ControlMessage(Message): + """HID Control message. + + This message is used to send control commands between the host and device. + """ + + message_type = Message.Type.CONTROL + + class Command(utils.OpenIntEnum): SUSPEND = 0x03 EXIT_SUSPEND = 0x04 VIRTUAL_CABLE_UNPLUG = 0x05 - # Class Method to derive header + command: Command + + def __bytes__(self) -> bytes: + return self.header(self.command) + @classmethod - def header(cls, lower_bits: int = 0x00) -> bytes: - return bytes([(cls.message_type << 4) | lower_bits]) + def from_bytes(cls, data: bytes) -> ControlMessage: + if len(data) != 1: + raise HidProtocolError(HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER) + return cls(command=ControlMessage.Command(data[0] & 0x0F)) -# HIDP messages -@dataclass +@Message.message +@dataclasses.dataclass class GetReportMessage(Message): - report_type: int - report_id: int - buffer_size: int - message_type = Message.MessageType.GET_REPORT + """HID Get Report message. - def __bytes__(self) -> bytes: - packet_bytes = bytearray() - packet_bytes.append(self.report_id) - if self.buffer_size == 0: - return self.header(self.report_type) + packet_bytes - else: - return ( - self.header(0x08 | self.report_type) - + packet_bytes - + struct.pack(" bytes: - return self.header(self.report_type) + self.data + data = self.header( + self.report_type + | (self.FLAG_HAS_SIZE if self.buffer_size is not None else 0) + ) + if self.report_id is not None: + data += bytes([self.report_id]) + if self.buffer_size is not None: + data += struct.pack(" GetReportMessage: + report_type = ReportType(data[0] & 0x03) + has_size = bool(data[0] & cls.FLAG_HAS_SIZE) + match (has_size, len(data)): + case (False, 1): + return cls(report_type=report_type) + case (False, 2): + return cls(report_type=report_type, report_id=data[1]) + case (True, 3): + return cls( + report_type=report_type, + buffer_size=struct.unpack(" bytes: return self.header(self.report_type) + self.data + @classmethod + def from_bytes(cls, data: bytes) -> SetReportMessage: + if len(data) < 2: + raise HidProtocolError(HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER) + return cls(report_type=ReportType(data[0] & 0x03), data=data[1:]) + -@dataclass +@Message.message +@dataclasses.dataclass class GetProtocolMessage(Message): - message_type = Message.MessageType.GET_PROTOCOL + """HID Get Protocol message. + + This message is used by the host to get the protocol mode for the device. + """ + + message_type = Message.Type.GET_PROTOCOL def __bytes__(self) -> bytes: return self.header() + @classmethod + def from_bytes(cls, data: bytes) -> GetProtocolMessage: + if len(data) != 1: + raise HidProtocolError(HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER) + return cls() + -@dataclass +@Message.message +@dataclasses.dataclass class SetProtocolMessage(Message): - protocol_mode: int - message_type = Message.MessageType.SET_PROTOCOL + """HID Set Protocol message. + + This message is used by the host to set the protocol mode for the device. + """ + + message_type = Message.Type.SET_PROTOCOL + + protocol_mode: ProtocolMode def __bytes__(self) -> bytes: return self.header(self.protocol_mode) + @classmethod + def from_bytes(cls, data: bytes) -> SetProtocolMessage: + if len(data) != 1: + raise HidProtocolError(HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER) + return cls(protocol_mode=ProtocolMode(data[0] & 0x01)) -@dataclass -class Suspend(Message): - message_type = Message.MessageType.CONTROL - def __bytes__(self) -> bytes: - return self.header(Message.ControlCommand.SUSPEND) +@Message.message +@dataclasses.dataclass +class GetIdleMessage(Message): + """HID Get Idle message. + This message is used by the host to get the idle time for the device. + """ -@dataclass -class ExitSuspend(Message): - message_type = Message.MessageType.CONTROL + message_type = Message.Type.GET_IDLE def __bytes__(self) -> bytes: - return self.header(Message.ControlCommand.EXIT_SUSPEND) + return self.header() + @classmethod + def from_bytes(cls, data: bytes) -> GetIdleMessage: + if len(data) != 1: + raise HidProtocolError(HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER) + return cls() + + +@Message.message +@dataclasses.dataclass +class SetIdleMessage(Message): + """HID Set Idle message. -@dataclass -class VirtualCableUnplug(Message): - message_type = Message.MessageType.CONTROL + This message is used by the host to set the idle time for the device. + """ + + message_type = Message.Type.SET_IDLE + + idle_time: int def __bytes__(self) -> bytes: - return self.header(Message.ControlCommand.VIRTUAL_CABLE_UNPLUG) + return self.header() + bytes([self.idle_time]) + + @classmethod + def from_bytes(cls, data: bytes) -> SetIdleMessage: + if len(data) != 2: + raise HidProtocolError(HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER) + return cls(idle_time=data[1]) # Device sends input report, host sends output report. -@dataclass -class SendData(Message): +@Message.message +@dataclasses.dataclass +class DataMessage(Message): + """HID Data message. + + This message is used to send report data. The direction of the report depends + on the role: + - Device sends input report. + - Host sends output report. + """ + + message_type = Message.Type.DATA + data: bytes - report_type: int - message_type = Message.MessageType.DATA + report_type: ReportType def __bytes__(self) -> bytes: return self.header(self.report_type) + self.data + @classmethod + def from_bytes(cls, data: bytes) -> DataMessage: + if not data: + raise core.InvalidPacketError("Empty DATA message") + return cls(data=data[1:], report_type=ReportType(data[0] & 0x03)) -@dataclass -class SendHandshakeMessage(Message): - result_code: int - message_type = Message.MessageType.HANDSHAKE - def __bytes__(self) -> bytes: - return self.header(self.result_code) +# ----------------------------------------------------------------------------- +class HID(abc.ABC, utils.EventEmitter): + """Base class for Bluetooth Human Interface Device (HID) profiles. + This class provides the fundamental structure for both HID Host and HID Device + roles, handling L2CAP channel management for control and interrupt channels. + """ -# ----------------------------------------------------------------------------- -class HID(ABC, utils.EventEmitter): - l2cap_ctrl_channel: l2cap.ClassicChannel | None = None - l2cap_intr_channel: l2cap.ClassicChannel | None = None - connection: device.Connection | None = None + control_channel: l2cap.ClassicChannel | None = None + interrupt_channel: l2cap.ClassicChannel | None = None EVENT_INTERRUPT_DATA = "interrupt_data" EVENT_CONTROL_DATA = "control_data" @@ -205,366 +372,859 @@ class HID(ABC, utils.EventEmitter): EVENT_EXIT_SUSPEND = "exit_suspend" EVENT_VIRTUAL_CABLE_UNPLUG = "virtual_cable_unplug" EVENT_HANDSHAKE = "handshake" + EVENT_CONNECTION = "connection" + EVENT_DISCONNECTION = "disconnection" - class Role(enum.IntEnum): + class Role(utils.OpenIntEnum): HOST = 0x00 DEVICE = 0x01 - def __init__(self, device: device.Device, role: Role) -> None: - super().__init__() - self.remote_device_bd_address: Address | None = None + role: ClassVar[Role] + + def __init__(self, device: device_lib.Device) -> None: + utils.EventEmitter.__init__(self) + self.remote_device_bd_address: hci.Address | None = None self.device = device - self.role = role # Register ourselves with the L2CAP channel manager device.create_l2cap_server( - l2cap.ClassicChannelSpec(HID_CONTROL_PSM), self.on_l2cap_connection + l2cap.ClassicChannelSpec(HID_CONTROL_PSM), self._on_l2cap_connection ) device.create_l2cap_server( - l2cap.ClassicChannelSpec(HID_INTERRUPT_PSM), self.on_l2cap_connection + l2cap.ClassicChannelSpec(HID_INTERRUPT_PSM), self._on_l2cap_connection ) - device.on(device.EVENT_CONNECTION, self.on_device_connection) - - async def connect_control_channel(self) -> None: - if not self.connection: - raise InvalidStateError("Connection is not established!") - # Create a new L2CAP connection - control channel - try: - channel = await self.connection.create_l2cap_channel( - l2cap.ClassicChannelSpec(HID_CONTROL_PSM) - ) - channel.sink = self.on_ctrl_pdu - self.l2cap_ctrl_channel = channel - except ProtocolError: - logging.exception('L2CAP connection failed.') - raise - - async def connect_interrupt_channel(self) -> None: - if not self.connection: - raise InvalidStateError("Connection is not established!") - # Create a new L2CAP connection - interrupt channel + async def connect(self, connection: device_lib.Connection) -> None: + self.remote_device_bd_address = connection.peer_address + self.control_channel = await connection.create_l2cap_channel( + l2cap.ClassicChannelSpec(HID_CONTROL_PSM) + ) + self.control_channel.sink = self._on_control_pdu try: - channel = await self.connection.create_l2cap_channel( + self.interrupt_channel = await connection.create_l2cap_channel( l2cap.ClassicChannelSpec(HID_INTERRUPT_PSM) ) - channel.sink = self.on_intr_pdu - self.l2cap_intr_channel = channel - except ProtocolError: - logging.exception('L2CAP connection failed.') + self.interrupt_channel.sink = self._on_interrupt_pdu + except Exception: + await self.disconnect() raise - async def disconnect_interrupt_channel(self) -> None: - if self.l2cap_intr_channel is None: - raise InvalidStateError('invalid state') - channel = self.l2cap_intr_channel - self.l2cap_intr_channel = None - await channel.disconnect() - - async def disconnect_control_channel(self) -> None: - if self.l2cap_ctrl_channel is None: - raise InvalidStateError('invalid state') - channel = self.l2cap_ctrl_channel - self.l2cap_ctrl_channel = None - await channel.disconnect() - - def on_device_connection(self, connection: device.Connection) -> None: - self.connection = connection - self.remote_device_bd_address = connection.peer_address - connection.on(connection.EVENT_DISCONNECTION, self.on_device_disconnection) - - def on_device_disconnection(self, reason: int) -> None: - self.connection = None + async def disconnect(self) -> None: + if self.interrupt_channel: + await self.interrupt_channel.disconnect() + self.interrupt_channel = None + if self.control_channel: + await self.control_channel.disconnect() + self.control_channel = None - def on_l2cap_connection(self, l2cap_channel: l2cap.ClassicChannel) -> None: - logger.debug(f'+++ New L2CAP connection: {l2cap_channel}') + def _on_l2cap_connection(self, l2cap_channel: l2cap.ClassicChannel) -> None: + logger.debug("+++ New L2CAP connection: %s", l2cap_channel) l2cap_channel.on( - l2cap_channel.EVENT_OPEN, lambda: self.on_l2cap_channel_open(l2cap_channel) + l2cap_channel.EVENT_OPEN, + lambda: self._on_l2cap_channel_open(l2cap_channel), ) l2cap_channel.on( l2cap_channel.EVENT_CLOSE, - lambda: self.on_l2cap_channel_close(l2cap_channel), + lambda: self._on_l2cap_channel_close(l2cap_channel), ) - def on_l2cap_channel_open(self, l2cap_channel: l2cap.ClassicChannel) -> None: + def _on_l2cap_channel_open(self, l2cap_channel: l2cap.ClassicChannel) -> None: + self.remote_device_bd_address = l2cap_channel.connection.peer_address if l2cap_channel.psm == HID_CONTROL_PSM: - self.l2cap_ctrl_channel = l2cap_channel - self.l2cap_ctrl_channel.sink = self.on_ctrl_pdu + self.control_channel = l2cap_channel + self.control_channel.sink = self._on_control_pdu else: - self.l2cap_intr_channel = l2cap_channel - self.l2cap_intr_channel.sink = self.on_intr_pdu - logger.debug(f'$$$ L2CAP channel open: {l2cap_channel}') + self.interrupt_channel = l2cap_channel + self.interrupt_channel.sink = self._on_interrupt_pdu + if not self.control_channel: + logger.warning("Interrupt channel established before control channel!") + logger.debug("$$$ L2CAP channel open: %s", l2cap_channel) + + if self.control_channel and self.interrupt_channel: + self.emit(self.EVENT_CONNECTION) - def on_l2cap_channel_close(self, l2cap_channel: l2cap.ClassicChannel) -> None: + def _on_l2cap_channel_close(self, l2cap_channel: l2cap.ClassicChannel) -> None: if l2cap_channel.psm == HID_CONTROL_PSM: - self.l2cap_ctrl_channel = None + self.control_channel = None else: - self.l2cap_intr_channel = None - logger.debug(f'$$$ L2CAP channel close: {l2cap_channel}') + self.interrupt_channel = None + logger.debug("$$$ L2CAP channel close: %s", l2cap_channel) - @abstractmethod - def on_ctrl_pdu(self, pdu: bytes) -> None: + if not self.control_channel and not self.interrupt_channel: + self.emit(self.EVENT_DISCONNECTION) + + @abc.abstractmethod + def _on_control_pdu(self, pdu: bytes) -> None: pass - def on_intr_pdu(self, pdu: bytes) -> None: - logger.debug(f'<<< HID INTERRUPT PDU: {pdu.hex()}') - self.emit(self.EVENT_INTERRUPT_DATA, pdu) + def _on_interrupt_pdu(self, pdu: bytes) -> None: + if not pdu: + logger.warning("Ignoring empty HID interrupt PDU") + return + try: + message = DataMessage.from_bytes(pdu) + except (core.InvalidPacketError, HidProtocolError) as e: + logger.warning("Ignoring invalid HID interrupt PDU: %s", e) + return + logger.debug("<<< [Interrupt] %s", message) + self.emit( + self.EVENT_INTERRUPT_DATA, + message.report_type, + message.data, + ) - def send_pdu_on_ctrl(self, msg: bytes) -> None: - assert self.l2cap_ctrl_channel - self.l2cap_ctrl_channel.write(msg) + def _send_control_pdu(self, message: Message) -> None: + if not self.control_channel: + raise core.InvalidStateError("Control channel is not connected") + logger.debug(">>> [Control] %s", message) + self.control_channel.write(bytes(message)) - def send_pdu_on_intr(self, msg: bytes) -> None: - assert self.l2cap_intr_channel - self.l2cap_intr_channel.write(msg) + def _send_interrupt_pdu(self, message: Message) -> None: + if not self.interrupt_channel: + raise core.InvalidStateError("Interrupt channel is not connected") + logger.debug(">>> [Interrupt] %s", message) + self.interrupt_channel.write(bytes(message)) - def send_data(self, data: bytes) -> None: + def send_interrupt_data(self, data: bytes) -> None: if self.role == HID.Role.HOST: - report_type = Message.ReportType.OUTPUT_REPORT + report_type = ReportType.OUTPUT_REPORT else: - report_type = Message.ReportType.INPUT_REPORT - msg = SendData(data, report_type) - hid_message = bytes(msg) - if self.l2cap_intr_channel is not None: - logger.debug(f'>>> HID INTERRUPT SEND DATA, PDU: {hid_message.hex()}') - self.send_pdu_on_intr(hid_message) + report_type = ReportType.INPUT_REPORT + if self.interrupt_channel is not None: + self._send_interrupt_pdu(DataMessage(data, report_type)) def virtual_cable_unplug(self) -> None: - msg = VirtualCableUnplug() - hid_message = bytes(msg) - logger.debug(f'>>> HID CONTROL VIRTUAL CABLE UNPLUG, PDU: {hid_message.hex()}') - self.send_pdu_on_ctrl(hid_message) + self._send_control_pdu( + ControlMessage(ControlMessage.Command.VIRTUAL_CABLE_UNPLUG) + ) # ----------------------------------------------------------------------------- +class Device(HID): + """HID Device role implementation. + This class represents a Bluetooth HID Device, handling incoming control and + interrupt channel PDUs and providing an interface for sending data. + """ -class Device(HID): - class GetSetReturn(enum.IntEnum): - FAILURE = 0x00 - REPORT_ID_NOT_FOUND = 0x01 - ERR_UNSUPPORTED_REQUEST = 0x02 - ERR_UNKNOWN = 0x03 - ERR_INVALID_PARAMETER = 0x04 - SUCCESS = 0xFF + EVENT_PROTOCOL_CHANGE = "protocol_change" + EVENT_IDLE_TIME_CHANGE = "idle_time_change" + _idle_time: int | None = 0 - @dataclass - class GetSetStatus: - data: bytes = b'' - status: int = 0 + class Delegate: + """Delegate class for handling HID device requests. - get_report_cb: Callable[[int, int, int], GetSetStatus] | None = None - set_report_cb: Callable[[int, int, int, bytes], GetSetStatus] | None = None - get_protocol_cb: Callable[[], GetSetStatus] | None = None - set_protocol_cb: Callable[[int], GetSetStatus] | None = None + This class defines the interface for handling `set_report` and `get_report` + requests from the HID Host. + """ - def __init__(self, device: device.Device) -> None: - super().__init__(device, HID.Role.DEVICE) + def set_report(self, report_type: ReportType, data: bytes) -> None: + del report_type, data # unused. + raise HidProtocolError(HandshakeMessage.ResultCode.ERR_UNSUPPORTED_REQUEST) - @override - def on_ctrl_pdu(self, pdu: bytes) -> None: - logger.debug(f'<<< HID CONTROL PDU: {pdu.hex()}') - param = pdu[0] & 0x0F - message_type = pdu[0] >> 4 - - if message_type == Message.MessageType.GET_REPORT: - logger.debug('<<< HID GET REPORT') - self.handle_get_report(pdu) - elif message_type == Message.MessageType.SET_REPORT: - logger.debug('<<< HID SET REPORT') - self.handle_set_report(pdu) - elif message_type == Message.MessageType.GET_PROTOCOL: - logger.debug('<<< HID GET PROTOCOL') - self.handle_get_protocol(pdu) - elif message_type == Message.MessageType.SET_PROTOCOL: - logger.debug('<<< HID SET PROTOCOL') - self.handle_set_protocol(pdu) - elif message_type == Message.MessageType.DATA: - logger.debug('<<< HID CONTROL DATA') - self.emit(self.EVENT_CONTROL_DATA, pdu) - elif message_type == Message.MessageType.CONTROL: - if param == Message.ControlCommand.SUSPEND: - logger.debug('<<< HID SUSPEND') - self.emit(self.EVENT_SUSPEND) - elif param == Message.ControlCommand.EXIT_SUSPEND: - logger.debug('<<< HID EXIT SUSPEND') - self.emit(self.EVENT_EXIT_SUSPEND) - elif param == Message.ControlCommand.VIRTUAL_CABLE_UNPLUG: - logger.debug('<<< HID VIRTUAL CABLE UNPLUG') - self.emit(self.EVENT_VIRTUAL_CABLE_UNPLUG) - else: - logger.debug('<<< HID CONTROL OPERATION UNSUPPORTED') - else: - logger.debug('<<< HID MESSAGE TYPE UNSUPPORTED') - self.send_handshake_message(Message.Handshake.ERR_UNSUPPORTED_REQUEST) - - def send_handshake_message(self, result_code: int) -> None: - msg = SendHandshakeMessage(result_code) - hid_message = bytes(msg) - logger.debug(f'>>> HID HANDSHAKE MESSAGE, PDU: {hid_message.hex()}') - self.send_pdu_on_ctrl(hid_message) - - def send_control_data(self, report_type: int, data: bytes): - msg = SendControlData(report_type=report_type, data=data) - hid_message = bytes(msg) - logger.debug(f'>>> HID CONTROL DATA: {hid_message.hex()}') - self.send_pdu_on_ctrl(hid_message) - - def handle_get_report(self, pdu: bytes): - if self.get_report_cb is None: - logger.debug("GetReport callback not registered !!") - self.send_handshake_message(Message.Handshake.ERR_UNSUPPORTED_REQUEST) - return - report_type = pdu[0] & 0x03 - buffer_flag = (pdu[0] & 0x08) >> 3 - report_id = pdu[1] - logger.debug(f"buffer_flag: {buffer_flag}") - if buffer_flag == 1: - buffer_size = (pdu[3] << 8) | pdu[2] - else: - buffer_size = 0 - - ret = self.get_report_cb(report_id, report_type, buffer_size) - if ret.status == self.GetSetReturn.FAILURE: - self.send_handshake_message(Message.Handshake.ERR_UNKNOWN) - elif ret.status == self.GetSetReturn.SUCCESS: - data = bytearray() - data.append(report_id) - data.extend(ret.data) - if len(data) < self.l2cap_ctrl_channel.peer_mtu: # type: ignore[union-attr] - self.send_control_data(report_type=report_type, data=data) - else: - self.send_handshake_message(Message.Handshake.ERR_INVALID_PARAMETER) - elif ret.status == self.GetSetReturn.REPORT_ID_NOT_FOUND: - self.send_handshake_message(Message.Handshake.ERR_INVALID_REPORT_ID) - elif ret.status == self.GetSetReturn.ERR_INVALID_PARAMETER: - self.send_handshake_message(Message.Handshake.ERR_INVALID_PARAMETER) - elif ret.status == self.GetSetReturn.ERR_UNSUPPORTED_REQUEST: - self.send_handshake_message(Message.Handshake.ERR_UNSUPPORTED_REQUEST) - - def register_get_report_cb( - self, cb: Callable[[int, int, int], Device.GetSetStatus] + def get_report(self, report_type: ReportType, report_id: int | None) -> bytes: + del report_type, report_id # unused. + raise HidProtocolError(HandshakeMessage.ResultCode.ERR_UNSUPPORTED_REQUEST) + + role = HID.Role.DEVICE + + def __init__( + self, + device: device_lib.Device, + delegate: Delegate | None = None, + protocol: ProtocolMode | None = ProtocolMode.REPORT_PROTOCOL, + idle_time: int | None = 0, ) -> None: - self.get_report_cb = cb - logger.debug("GetReport callback registered successfully") + super().__init__(device) + self.delegate = delegate + self.protocol = protocol + self._idle_time = idle_time - def handle_set_report(self, pdu: bytes): - if self.set_report_cb is None: - logger.debug("SetReport callback not registered !!") - self.send_handshake_message(Message.Handshake.ERR_UNSUPPORTED_REQUEST) + @override + def _on_control_pdu(self, pdu: bytes) -> None: + if not pdu: + logger.warning("Ignoring empty HID control PDU") return - report_type = pdu[0] & 0x03 - report_id = pdu[1] - report_data = pdu[2:] - report_size = len(report_data) + 1 - ret = self.set_report_cb(report_id, report_type, report_size, report_data) - if ret.status == self.GetSetReturn.SUCCESS: - self.send_handshake_message(Message.Handshake.SUCCESSFUL) - elif ret.status == self.GetSetReturn.ERR_INVALID_PARAMETER: - self.send_handshake_message(Message.Handshake.ERR_INVALID_PARAMETER) - elif ret.status == self.GetSetReturn.REPORT_ID_NOT_FOUND: - self.send_handshake_message(Message.Handshake.ERR_INVALID_REPORT_ID) - else: - self.send_handshake_message(Message.Handshake.ERR_UNSUPPORTED_REQUEST) - def register_set_report_cb( - self, cb: Callable[[int, int, int, bytes], Device.GetSetStatus] - ) -> None: - self.set_report_cb = cb - logger.debug("SetReport callback registered successfully") + try: + message = Message.from_bytes(pdu) + logger.debug("<<< [Control] %s", message) + + match message: + case GetReportMessage(): + self._handle_get_report(message) + case SetReportMessage(): + self._handle_set_report(message) + case GetProtocolMessage(): + self._handle_get_protocol() + case SetProtocolMessage(): + self._handle_set_protocol(message) + case GetIdleMessage(): + self._handle_get_idle() + case SetIdleMessage(): + self._handle_set_idle(message) + case DataMessage(data=data, report_type=report_type): + self.emit(self.EVENT_CONTROL_DATA, report_type, data) + case ControlMessage(command=ControlMessage.Command.SUSPEND): + self.emit(self.EVENT_SUSPEND) + case ControlMessage(command=ControlMessage.Command.EXIT_SUSPEND): + self.emit(self.EVENT_EXIT_SUSPEND) + case ControlMessage( + command=ControlMessage.Command.VIRTUAL_CABLE_UNPLUG + ): + self.emit(self.EVENT_VIRTUAL_CABLE_UNPLUG) + case ControlMessage(command=command): + logger.error("Unsupported command %s", command.name) + case _: + logger.error( + "Unsupported command type %s", message.message_type.name + ) + self._send_handshake_message( + HandshakeMessage.ResultCode.ERR_UNSUPPORTED_REQUEST + ) + except (NotImplementedError, core.InvalidPacketError): + self._send_handshake_message( + HandshakeMessage.ResultCode.ERR_UNSUPPORTED_REQUEST + ) + except HidProtocolError as e: + self._send_handshake_message(e.result_code) + + def _send_handshake_message(self, result_code: HandshakeMessage.ResultCode) -> None: + self._send_control_pdu(HandshakeMessage(result_code)) - def handle_get_protocol(self, pdu: bytes): - if self.get_protocol_cb is None: - logger.debug("GetProtocol callback not registered !!") - self.send_handshake_message(Message.Handshake.ERR_UNSUPPORTED_REQUEST) + def _send_control_data(self, report_type: ReportType, data: bytes) -> None: + self._send_control_pdu(DataMessage(report_type=report_type, data=data)) + + def _handle_get_report(self, message: GetReportMessage) -> None: + if not self.delegate: + self._send_handshake_message( + HandshakeMessage.ResultCode.ERR_UNSUPPORTED_REQUEST + ) return - ret = self.get_protocol_cb() - if ret.status == self.GetSetReturn.SUCCESS: - self.send_control_data(Message.ReportType.OTHER_REPORT, ret.data) - else: - self.send_handshake_message(Message.Handshake.ERR_UNSUPPORTED_REQUEST) + result = self.delegate.get_report(message.report_type, message.report_id) + data = ( + bytes([message.report_id] if message.report_id is not None else []) + result + ) + if message.buffer_size is not None: + data = data[: message.buffer_size] - def register_get_protocol_cb(self, cb: Callable[[], Device.GetSetStatus]) -> None: - self.get_protocol_cb = cb - logger.debug("GetProtocol callback registered successfully") + assert self.control_channel + if len(data) < self.control_channel.peer_mtu: + self._send_control_data(report_type=message.report_type, data=data) + else: + self._send_handshake_message( + HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER + ) - def handle_set_protocol(self, pdu: bytes): - if self.set_protocol_cb is None: - logger.debug("SetProtocol callback not registered !!") - self.send_handshake_message(Message.Handshake.ERR_UNSUPPORTED_REQUEST) + def _handle_set_report(self, message: SetReportMessage) -> None: + if not self.delegate: + self._send_handshake_message( + HandshakeMessage.ResultCode.ERR_UNSUPPORTED_REQUEST + ) return - ret = self.set_protocol_cb(pdu[0] & 0x01) - if ret.status == self.GetSetReturn.SUCCESS: - self.send_handshake_message(Message.Handshake.SUCCESSFUL) + self.delegate.set_report(message.report_type, message.data) + self._send_handshake_message(HandshakeMessage.ResultCode.SUCCESSFUL) + + def _handle_get_protocol(self) -> None: + if self.protocol is None: + self._send_handshake_message( + HandshakeMessage.ResultCode.ERR_UNSUPPORTED_REQUEST + ) else: - self.send_handshake_message(Message.Handshake.ERR_UNSUPPORTED_REQUEST) + self._send_control_data(ReportType.OTHER_REPORT, bytes([self.protocol])) - def register_set_protocol_cb( - self, cb: Callable[[int], Device.GetSetStatus] - ) -> None: - self.set_protocol_cb = cb - logger.debug("SetProtocol callback registered successfully") + def _handle_set_protocol(self, message: SetProtocolMessage) -> None: + if self.protocol is None: + self._send_handshake_message( + HandshakeMessage.ResultCode.ERR_UNSUPPORTED_REQUEST + ) + else: + self.protocol = message.protocol_mode + self._send_handshake_message(HandshakeMessage.ResultCode.SUCCESSFUL) + self.emit(self.EVENT_PROTOCOL_CHANGE) + + def _handle_get_idle(self) -> None: + if self._idle_time is None: + self._send_handshake_message( + HandshakeMessage.ResultCode.ERR_UNSUPPORTED_REQUEST + ) + else: + self._send_control_data(ReportType.OTHER_REPORT, bytes([self._idle_time])) + + def _handle_set_idle(self, message: SetIdleMessage) -> None: + if self._idle_time is None: + self._send_handshake_message( + HandshakeMessage.ResultCode.ERR_UNSUPPORTED_REQUEST + ) + else: + self._idle_time = message.idle_time + self._send_handshake_message(HandshakeMessage.ResultCode.SUCCESSFUL) + self.emit(self.EVENT_IDLE_TIME_CHANGE) # ----------------------------------------------------------------------------- class Host(HID): - def __init__(self, device: device.Device) -> None: - super().__init__(device, HID.Role.HOST) + """HID Host role implementation. + + This class represents a Bluetooth HID Host, handling the connection and + communication with a HID Device. It provides methods to send and receive + HID reports and control messages. + """ + + role = HID.Role.HOST + + def __init__(self, device: device_lib.Device) -> None: + super().__init__(device) + self._command_lock = asyncio.Lock() + self._pending_command_future: asyncio.Future[DataMessage | None] | None = None + + def _abort_pending_command(self, error: Exception) -> None: + if self._pending_command_future and not self._pending_command_future.done(): + self._pending_command_future.set_exception(error) + self._pending_command_future = None + + @override + async def disconnect(self) -> None: + self._abort_pending_command(core.InvalidStateError("HID host disconnected")) + await super().disconnect() + + @override + def _on_l2cap_channel_close(self, l2cap_channel: l2cap.ClassicChannel) -> None: + if l2cap_channel.psm == HID_CONTROL_PSM: + self._abort_pending_command( + core.InvalidStateError("HID control channel closed") + ) + super()._on_l2cap_channel_close(l2cap_channel) + + async def _send_control_message(self, message: Message) -> DataMessage | None: + async with self._command_lock: + loop = asyncio.get_running_loop() + future: asyncio.Future[DataMessage | None] = loop.create_future() + self._pending_command_future = future + try: + self._send_control_pdu(message) + return await future + finally: + if self._pending_command_future is future: + self._pending_command_future = None + + async def get_report( + self, + report_type: ReportType, + report_id: int | None = None, + buffer_size: int | None = None, + ) -> bytes: + result = await self._send_control_message( + GetReportMessage( + report_type=report_type, + report_id=report_id, + buffer_size=buffer_size, + ) + ) + if result: + return result.data + raise core.UnreachableError() + + async def set_report(self, report_type: ReportType, data: bytes) -> None: + await self._send_control_message( + SetReportMessage(report_type=report_type, data=data) + ) + + async def get_protocol(self) -> ProtocolMode: + result = await self._send_control_message(GetProtocolMessage()) + if result: + return ProtocolMode(result.data[0]) + raise core.UnreachableError() - def get_report(self, report_type: int, report_id: int, buffer_size: int) -> None: - msg = GetReportMessage( - report_type=report_type, report_id=report_id, buffer_size=buffer_size + async def set_protocol(self, protocol_mode: ProtocolMode) -> None: + await self._send_control_message( + SetProtocolMessage(protocol_mode=protocol_mode) ) - hid_message = bytes(msg) - logger.debug(f'>>> HID CONTROL GET REPORT, PDU: {hid_message.hex()}') - self.send_pdu_on_ctrl(hid_message) - - def set_report(self, report_type: int, data: bytes) -> None: - msg = SetReportMessage(report_type=report_type, data=data) - hid_message = bytes(msg) - logger.debug(f'>>> HID CONTROL SET REPORT, PDU:{hid_message.hex()}') - self.send_pdu_on_ctrl(hid_message) - - def get_protocol(self) -> None: - msg = GetProtocolMessage() - hid_message = bytes(msg) - logger.debug(f'>>> HID CONTROL GET PROTOCOL, PDU: {hid_message.hex()}') - self.send_pdu_on_ctrl(hid_message) - - def set_protocol(self, protocol_mode: int) -> None: - msg = SetProtocolMessage(protocol_mode=protocol_mode) - hid_message = bytes(msg) - logger.debug(f'>>> HID CONTROL SET PROTOCOL, PDU: {hid_message.hex()}') - self.send_pdu_on_ctrl(hid_message) + + async def get_idle(self) -> int: + result = await self._send_control_message(GetIdleMessage()) + if result: + return result.data[0] + raise core.UnreachableError() + + async def set_idle(self, idle_time: int) -> None: + await self._send_control_message(SetIdleMessage(idle_time=idle_time)) def suspend(self) -> None: - msg = Suspend() - hid_message = bytes(msg) - logger.debug(f'>>> HID CONTROL SUSPEND, PDU:{hid_message.hex()}') - self.send_pdu_on_ctrl(hid_message) + self._send_control_pdu(ControlMessage(ControlMessage.Command.SUSPEND)) def exit_suspend(self) -> None: - msg = ExitSuspend() - hid_message = bytes(msg) - logger.debug(f'>>> HID CONTROL EXIT SUSPEND, PDU:{hid_message.hex()}') - self.send_pdu_on_ctrl(hid_message) + self._send_control_pdu(ControlMessage(ControlMessage.Command.EXIT_SUSPEND)) @override - def on_ctrl_pdu(self, pdu: bytes) -> None: - logger.debug(f'<<< HID CONTROL PDU: {pdu.hex()}') - param = pdu[0] & 0x0F - message_type = pdu[0] >> 4 - if message_type == Message.MessageType.HANDSHAKE: - logger.debug(f'<<< HID HANDSHAKE: {Message.Handshake(param).name}') - self.emit(self.EVENT_HANDSHAKE, Message.Handshake(param)) - elif message_type == Message.MessageType.DATA: - logger.debug('<<< HID CONTROL DATA') - self.emit(self.EVENT_CONTROL_DATA, pdu) - elif message_type == Message.MessageType.CONTROL: - if param == Message.ControlCommand.VIRTUAL_CABLE_UNPLUG: - logger.debug('<<< HID VIRTUAL CABLE UNPLUG') + def _on_control_pdu(self, pdu: bytes) -> None: + if not pdu: + logger.warning("Ignoring empty HID control PDU") + return + try: + message = Message.from_bytes(pdu) + except (core.InvalidPacketError, HidProtocolError) as e: + logger.warning("Ignoring invalid HID control PDU: %s", e) + self._abort_pending_command(e) + return + logger.debug("<<< [Control] %s", message) + match message: + case DataMessage(data=data, report_type=report_type): + if ( + self._pending_command_future + and not self._pending_command_future.done() + ): + self._pending_command_future.set_result(message) + self._pending_command_future = None + else: + self.emit( + self.EVENT_CONTROL_DATA, + report_type, + data, + ) + case HandshakeMessage(result_code=result_code): + self.emit(self.EVENT_HANDSHAKE, result_code) + if ( + self._pending_command_future + and not self._pending_command_future.done() + ): + if result_code == HandshakeMessage.ResultCode.SUCCESSFUL: + self._pending_command_future.set_result(None) + else: + self._pending_command_future.set_exception( + HidProtocolError(result_code) + ) + self._pending_command_future = None + else: + logger.debug("Unhandled handshake message %s", message) + case ControlMessage(command=ControlMessage.Command.VIRTUAL_CABLE_UNPLUG): self.emit(self.EVENT_VIRTUAL_CABLE_UNPLUG) - else: - logger.debug('<<< HID CONTROL OPERATION UNSUPPORTED') - else: - logger.debug('<<< HID MESSAGE TYPE UNSUPPORTED') + case ControlMessage(command=command): + logger.debug("Unsupported command %s", command.name) + case _: + logger.debug("Unsupported message %s", message.message_type.name) + + +DEFAULT_REPORT_MAP = bytes( + [ + # fmt: off + # pylint: disable=line-too-long + 0x05, 0x01, # Usage Page (Generic Desktop Ctrls) + 0x09, 0x06, # Usage (Keyboard) + 0xA1, 0x01, # Collection (Application) + 0x85, 0x01, # Report ID (1) + 0x05, 0x07, # Usage Page (Kbrd/Keypad) + 0x19, 0xE0, # Usage Minimum (0xE0) + 0x29, 0xE7, # Usage Maximum (0xE7) + 0x15, 0x00, # Logical Minimum (0) + 0x25, 0x01, # Logical Maximum (1) + 0x75, 0x01, # Report Size (1) + 0x95, 0x08, # Report Count (8) + 0x81, 0x02, # Input (Data,Var,Abs,No Wrap,Linear,Preferred State,No Null Position) + 0x95, 0x01, # Report Count (1) + 0x75, 0x08, # Report Size (8) + 0x81, 0x01, # Input (Const,Array,Abs,No Wrap,Linear,Preferred State,No Null Position) + 0x95, 0x06, # Report Count (6) + 0x75, 0x08, # Report Size (8) + 0x15, 0x00, # Logical Minimum (0) + 0x25, 0x94, # Logical Maximum (-108) + 0x05, 0x07, # Usage Page (Kbrd/Keypad) + 0x19, 0x00, # Usage Minimum (0x00) + 0x29, 0x94, # Usage Maximum (0x94) + 0x81, 0x00, # Input (Data,Array,Abs,No Wrap,Linear,Preferred State,No Null Position) + 0x95, 0x05, # Report Count (5) + 0x75, 0x01, # Report Size (1) + 0x05, 0x08, # Usage Page (LEDs) + 0x19, 0x01, # Usage Minimum (Num Lock) + 0x29, 0x05, # Usage Maximum (Kana) + 0x91, 0x02, # Output (Data,Var,Abs,No Wrap,Linear,Preferred State,No Null Position,Non-volatile) + 0x95, 0x01, # Report Count (1) + 0x75, 0x03, # Report Size (3) + 0x91, 0x01, # Output (Const,Array,Abs,No Wrap,Linear,Preferred State,No Null Position,Non-volatile) + 0xC0, # End Collection + 0x05, 0x01, # Usage Page (Generic Desktop Ctrls) + 0x09, 0x02, # Usage (Mouse) + 0xA1, 0x01, # Collection (Application) + 0x85, 0x02, # Report ID (2) + 0x09, 0x01, # Usage (Pointer) + 0xA1, 0x00, # Collection (Physical) + 0x05, 0x09, # Usage Page (Button) + 0x19, 0x01, # Usage Minimum (0x01) + 0x29, 0x05, # Usage Maximum (0x05) + 0x15, 0x00, # Logical Minimum (0) + 0x25, 0x01, # Logical Maximum (1) + 0x95, 0x05, # Report Count (5) + 0x75, 0x01, # Report Size (1) + 0x81, 0x02, # Input (Data,Var,Abs,No Wrap,Linear,Preferred State,No Null Position) + 0x95, 0x01, # Report Count (1) + 0x75, 0x03, # Report Size (3) + 0x81, 0x01, # Input (Const,Array,Abs,No Wrap,Linear,Preferred State,No Null Position) + 0x05, 0x01, # Usage Page (Generic Desktop Ctrls) + 0x09, 0x30, # Usage (X) + 0x09, 0x31, # Usage (Y) + 0x16, 0x00, 0x80, # Logical Minimum (-32768) + 0x26, 0xFF, 0x7F, # Logical Maximum (32767) + 0x75, 0x10, # Report Size (16) + 0x95, 0x02, # Report Count (2) + 0x81, 0x06, # Input (Data,Var,Rel,No Wrap,Linear,Preferred State,No Null Position) + 0x09, 0x38, # Usage (Wheel) + 0x15, 0x81, # Logical Minimum (-127) + 0x25, 0x7F, # Logical Maximum (127) + 0x75, 0x08, # Report Size (8) + 0x95, 0x01, # Report Count (1) + 0x81, 0x06, # Input (Data,Var,Rel,No Wrap,Linear,Preferred State,No Null Position) + 0xC0, # End Collection + 0xC0, # End Collection + ] +) + + +class AttributeId(enum.IntEnum): + """SDP attribute IDs.""" + + SERVICE_NAME = 0x0100 + SERVICE_DESCRIPTION = 0x0101 + PROVIDER_NAME = 0x0102 + DEVICE_RELEASE_NUMBER = 0x0200 # [DEPRECATED] + PARSER_VERSION = 0x0201 + DEVICE_SUBCLASS = 0x0202 + COUNTRY_CODE = 0x0203 + VIRTUAL_CABLE = 0x0204 + RECONNECT_INITIATE = 0x0205 + DESCRIPTOR_LIST = 0x0206 + LANGID_BASE_LIST = 0x0207 + SDP_DISABLE = 0x0208 # [DEPRECATED] + BATTERY_POWER = 0x0209 + REMOTE_WAKE = 0x020A + PROFILE_VERSION = 0x020B # [DEPRECATED] + SUPERVISION_TIMEOUT = 0x020C + NORMALLY_CONNECTABLE = 0x020D + BOOT_DEVICE = 0x020E + SSR_HOST_MAX_LATENCY = 0x020F + SSR_HOST_MIN_TIMEOUT = 0x0210 + + +# pylint: disable=too-many-instance-attributes +@dataclasses.dataclass +class DeviceSdpRecord: + """HID Device SDP record.""" + + service_record_handle: int + report_map: bytes + version_number: int = 0x0101 # 0x0101 uint16 version number (v1.1) + parser_version: int = 0x0111 # uint16 0x0111 (v1.1.1) + device_subclass: int = 0xC0 # Combo keyboard/pointing device + country_code: int = 0x21 # 0x21 Uint8, USA + virtual_cable: bool = True # Virtual cable enabled + reconnect_initiate: bool = True # Reconnect initiate enabled + report_descriptor_type: int = 0x22 # 0x22 Type = Report Descriptor + langid_base_language: int = 0x0409 # 0x0409 Language = en_US + langid_base_bluetooth_string_offset: int = 0x0100 # 0x0100 Default + boot_device: bool = True # Boot device support enabled + battery_power: bool | None = True # Battery power enabled + remote_wake: bool | None = True # Remote wake enabled + supervision_timeout: int | None = 0xC80 # uint16 0xC80 (2s) + normally_connectable: bool | None = True # Normally connectable enabled + service_name: bytes | None = b"Bumble HID" + service_description: bytes | None = b"Bumble" + provider_name: bytes | None = b"Bumble" + ssr_host_max_latency: int | None = 0x640 # uint16 0x640 (1s) + ssr_host_min_timeout: int | None = 0xC80 # uint16 0xC80 (2s) + + def to_service_attributes(self) -> list[sdp.ServiceAttribute]: + """Converts the SDP record to a list of SDP service attributes.""" + attributes = [ + sdp.ServiceAttribute( + sdp.SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID, + sdp.DataElement.unsigned_integer_32(self.service_record_handle), + ), + sdp.ServiceAttribute( + sdp.SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID, + sdp.DataElement.sequence( + [sdp.DataElement.uuid(sdp.SDP_PUBLIC_BROWSE_ROOT)] + ), + ), + sdp.ServiceAttribute( + sdp.SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID, + sdp.DataElement.sequence( + [ + sdp.DataElement.uuid(core.BT_HUMAN_INTERFACE_DEVICE_SERVICE), + ] + ), + ), + sdp.ServiceAttribute( + sdp.SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, + sdp.DataElement.sequence( + [ + sdp.DataElement.sequence( + [ + sdp.DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID), + sdp.DataElement.unsigned_integer_16(HID_CONTROL_PSM), + ] + ), + sdp.DataElement.sequence( + [sdp.DataElement.uuid(core.BT_HIDP_PROTOCOL_ID)] + ), + ] + ), + ), + sdp.ServiceAttribute( + sdp.SDP_LANGUAGE_BASE_ATTRIBUTE_ID_LIST_ATTRIBUTE_ID, + sdp.DataElement.sequence( + [ + sdp.DataElement.unsigned_integer_16(0x656E), # "en" + sdp.DataElement.unsigned_integer_16(0x6A), + sdp.DataElement.unsigned_integer_16(0x0100), + ] + ), + ), + sdp.ServiceAttribute( + sdp.SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID, + sdp.DataElement.sequence( + [ + sdp.DataElement.sequence( + [ + sdp.DataElement.uuid( + core.BT_HUMAN_INTERFACE_DEVICE_SERVICE + ), + sdp.DataElement.unsigned_integer_16( + self.version_number + ), + ] + ), + ] + ), + ), + sdp.ServiceAttribute( + sdp.SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, + sdp.DataElement.sequence( + [ + sdp.DataElement.sequence( + [ + sdp.DataElement.sequence( + [ + sdp.DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID), + sdp.DataElement.unsigned_integer_16( + HID_INTERRUPT_PSM + ), + ] + ), + sdp.DataElement.sequence( + [ + sdp.DataElement.uuid(core.BT_HIDP_PROTOCOL_ID), + ] + ), + ] + ), + ] + ), + ), + sdp.ServiceAttribute( + AttributeId.PARSER_VERSION, + sdp.DataElement.unsigned_integer_32(self.parser_version), + ), + sdp.ServiceAttribute( + AttributeId.DEVICE_SUBCLASS, + sdp.DataElement.unsigned_integer_32(self.device_subclass), + ), + sdp.ServiceAttribute( + AttributeId.COUNTRY_CODE, + sdp.DataElement.unsigned_integer_32(self.country_code), + ), + sdp.ServiceAttribute( + AttributeId.VIRTUAL_CABLE, + sdp.DataElement.boolean(self.virtual_cable), + ), + sdp.ServiceAttribute( + AttributeId.RECONNECT_INITIATE, + sdp.DataElement.boolean(self.reconnect_initiate), + ), + sdp.ServiceAttribute( + AttributeId.DESCRIPTOR_LIST, + sdp.DataElement.sequence( + [ + sdp.DataElement.sequence( + [ + sdp.DataElement.unsigned_integer_16( + self.report_descriptor_type + ), + sdp.DataElement( + sdp.DataElement.TEXT_STRING, self.report_map + ), + ] + ), + ] + ), + ), + sdp.ServiceAttribute( + AttributeId.LANGID_BASE_LIST, + sdp.DataElement.sequence( + [ + sdp.DataElement.sequence( + [ + sdp.DataElement.unsigned_integer_16( + self.langid_base_language + ), + sdp.DataElement.unsigned_integer_16( + self.langid_base_bluetooth_string_offset + ), + ] + ), + ] + ), + ), + sdp.ServiceAttribute( + AttributeId.BOOT_DEVICE, + sdp.DataElement.boolean(self.boot_device), + ), + ] + if self.service_name is not None: + attributes.append( + sdp.ServiceAttribute( + AttributeId.SERVICE_NAME, + sdp.DataElement(sdp.DataElement.TEXT_STRING, self.service_name), + ) + ) + if self.service_description is not None: + attributes.append( + sdp.ServiceAttribute( + AttributeId.SERVICE_DESCRIPTION, + sdp.DataElement( + sdp.DataElement.TEXT_STRING, self.service_description + ), + ) + ) + if self.provider_name is not None: + attributes.append( + sdp.ServiceAttribute( + AttributeId.PROVIDER_NAME, + sdp.DataElement(sdp.DataElement.TEXT_STRING, self.provider_name), + ) + ) + if self.battery_power is not None: + attributes.append( + sdp.ServiceAttribute( + AttributeId.BATTERY_POWER, + sdp.DataElement.boolean(self.battery_power), + ) + ) + if self.remote_wake is not None: + attributes.append( + sdp.ServiceAttribute( + AttributeId.REMOTE_WAKE, + sdp.DataElement.boolean(self.remote_wake), + ) + ) + if self.supervision_timeout is not None: + attributes.append( + sdp.ServiceAttribute( + AttributeId.SUPERVISION_TIMEOUT, + sdp.DataElement.unsigned_integer_16(self.supervision_timeout), + ) + ) + if self.normally_connectable is not None: + attributes.append( + sdp.ServiceAttribute( + AttributeId.NORMALLY_CONNECTABLE, + sdp.DataElement.boolean(self.normally_connectable), + ) + ) + if self.ssr_host_max_latency is not None: + attributes.append( + sdp.ServiceAttribute( + AttributeId.SSR_HOST_MAX_LATENCY, + sdp.DataElement.unsigned_integer_16(self.ssr_host_max_latency), + ) + ) + if self.ssr_host_min_timeout is not None: + attributes.append( + sdp.ServiceAttribute( + AttributeId.SSR_HOST_MIN_TIMEOUT, + sdp.DataElement.unsigned_integer_16(self.ssr_host_min_timeout), + ) + ) + return attributes + + @classmethod + def from_service_attributes( + cls, attributes: Sequence[sdp.ServiceAttribute] + ) -> Self: + """Creates a DeviceSdpRecord from a sequence of SDP service attributes.""" + attr: dict[str, Any] = { + "battery_power": None, + "remote_wake": None, + "supervision_timeout": None, + "normally_connectable": None, + "service_name": None, + "service_description": None, + "provider_name": None, + "ssr_host_max_latency": None, + "ssr_host_min_timeout": None, + } + for attribute in attributes: + match attribute.id: + case sdp.SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID: + attr["service_record_handle"] = attribute.value.value + case sdp.SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID: + attr["version_number"] = attribute.value.value[0].value[1].value + case AttributeId.SERVICE_NAME: + attr["service_name"] = attribute.value.value + case AttributeId.SERVICE_DESCRIPTION: + attr["service_description"] = attribute.value.value + case AttributeId.PROVIDER_NAME: + attr["provider_name"] = attribute.value.value + case AttributeId.PARSER_VERSION: + attr["parser_version"] = attribute.value.value + case AttributeId.DEVICE_SUBCLASS: + attr["device_subclass"] = attribute.value.value + case AttributeId.COUNTRY_CODE: + attr["country_code"] = attribute.value.value + case AttributeId.VIRTUAL_CABLE: + attr["virtual_cable"] = attribute.value.value + case AttributeId.RECONNECT_INITIATE: + attr["reconnect_initiate"] = attribute.value.value + case AttributeId.DESCRIPTOR_LIST: + attr["report_descriptor_type"] = ( + attribute.value.value[0].value[0].value + ) + attr["report_map"] = attribute.value.value[0].value[1].value + case AttributeId.BATTERY_POWER: + attr["battery_power"] = attribute.value.value + case AttributeId.REMOTE_WAKE: + attr["remote_wake"] = attribute.value.value + case AttributeId.SUPERVISION_TIMEOUT: + attr["supervision_timeout"] = attribute.value.value + case AttributeId.NORMALLY_CONNECTABLE: + attr["normally_connectable"] = attribute.value.value + case AttributeId.LANGID_BASE_LIST: + attr["langid_base_language"] = ( + attribute.value.value[0].value[0].value + ) + attr["langid_base_bluetooth_string_offset"] = ( + attribute.value.value[0].value[1].value + ) + case AttributeId.BOOT_DEVICE: + attr["boot_device"] = attribute.value.value + case AttributeId.SSR_HOST_MAX_LATENCY: + attr["ssr_host_max_latency"] = attribute.value.value + case AttributeId.SSR_HOST_MIN_TIMEOUT: + attr["ssr_host_min_timeout"] = attribute.value.value + case _: + pass + return cls(**attr) + + @classmethod + async def find(cls, connection: device_lib.Connection) -> list[Self]: + """Searches for HID Device SDP records on a remote device.""" + records: list[Self] = [] + async with sdp.Client(connection) as sdp_client: + search_result = await sdp_client.search_attributes( + uuids=[core.BT_HUMAN_INTERFACE_DEVICE_SERVICE], + attribute_ids=[(0x0000, 0xFFFF)], + ) + for attribute_list in search_result: + try: + records.append(cls.from_service_attributes(attribute_list)) + except (TypeError, ValueError, IndexError, KeyError): + logger.exception("Cannot build HID DeviceSdpRecord") + return records diff --git a/examples/run_hid_device.py b/examples/run_hid_device.py index 696e73497..bb72763d8 100644 --- a/examples/run_hid_device.py +++ b/examples/run_hid_device.py @@ -1,4 +1,4 @@ -# Copyright 2021-2022 Google LLC +# 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. @@ -15,389 +15,23 @@ # ----------------------------------------------------------------------------- # Imports # ----------------------------------------------------------------------------- +# pylint: disable=duplicate-code +from __future__ import annotations + import asyncio import json import struct import sys import websockets.asyncio.server +from typing_extensions import override import bumble.logging -from bumble.core import ( - BT_HIDP_PROTOCOL_ID, - BT_HUMAN_INTERFACE_DEVICE_SERVICE, - BT_L2CAP_PROTOCOL_ID, - PhysicalTransport, -) -from bumble.device import Device -from bumble.hid import HID_CONTROL_PSM, HID_INTERRUPT_PSM, Message -from bumble.hid import Device as HID_Device -from bumble.sdp import ( - SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, - SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID, - SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID, - SDP_LANGUAGE_BASE_ATTRIBUTE_ID_LIST_ATTRIBUTE_ID, - SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, - SDP_PUBLIC_BROWSE_ROOT, - SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID, - SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID, - DataElement, - ServiceAttribute, -) +from bumble import hid +from bumble.core import PhysicalTransport +from bumble.device import Connection, Device from bumble.transport import open_transport -# ----------------------------------------------------------------------------- -# SDP attributes for Bluetooth HID devices -SDP_HID_SERVICE_NAME_ATTRIBUTE_ID = 0x0100 -SDP_HID_SERVICE_DESCRIPTION_ATTRIBUTE_ID = 0x0101 -SDP_HID_PROVIDER_NAME_ATTRIBUTE_ID = 0x0102 -SDP_HID_DEVICE_RELEASE_NUMBER_ATTRIBUTE_ID = 0x0200 # [DEPRECATED] -SDP_HID_PARSER_VERSION_ATTRIBUTE_ID = 0x0201 -SDP_HID_DEVICE_SUBCLASS_ATTRIBUTE_ID = 0x0202 -SDP_HID_COUNTRY_CODE_ATTRIBUTE_ID = 0x0203 -SDP_HID_VIRTUAL_CABLE_ATTRIBUTE_ID = 0x0204 -SDP_HID_RECONNECT_INITIATE_ATTRIBUTE_ID = 0x0205 -SDP_HID_DESCRIPTOR_LIST_ATTRIBUTE_ID = 0x0206 -SDP_HID_LANGID_BASE_LIST_ATTRIBUTE_ID = 0x0207 -SDP_HID_SDP_DISABLE_ATTRIBUTE_ID = 0x0208 # [DEPRECATED] -SDP_HID_BATTERY_POWER_ATTRIBUTE_ID = 0x0209 -SDP_HID_REMOTE_WAKE_ATTRIBUTE_ID = 0x020A -SDP_HID_PROFILE_VERSION_ATTRIBUTE_ID = 0x020B # DEPRECATED] -SDP_HID_SUPERVISION_TIMEOUT_ATTRIBUTE_ID = 0x020C -SDP_HID_NORMALLY_CONNECTABLE_ATTRIBUTE_ID = 0x020D -SDP_HID_BOOT_DEVICE_ATTRIBUTE_ID = 0x020E -SDP_HID_SSR_HOST_MAX_LATENCY_ATTRIBUTE_ID = 0x020F -SDP_HID_SSR_HOST_MIN_TIMEOUT_ATTRIBUTE_ID = 0x0210 - -# Refer to HID profile specification v1.1.1, "5.3 Service Discovery Protocol (SDP)" for details -# HID SDP attribute values -LANGUAGE = 0x656E # 0x656E uint16 “en” (English) -ENCODING = 0x6A # 0x006A uint16 UTF-8 encoding -PRIMARY_LANGUAGE_BASE_ID = 0x100 # 0x0100 uint16 PrimaryLanguageBaseID -VERSION_NUMBER = 0x0101 # 0x0101 uint16 version number (v1.1) -SERVICE_NAME = b'Bumble HID' -SERVICE_DESCRIPTION = b'Bumble' -PROVIDER_NAME = b'Bumble' -HID_PARSER_VERSION = 0x0111 # uint16 0x0111 (v1.1.1) -HID_DEVICE_SUBCLASS = 0xC0 # Combo keyboard/pointing device -HID_COUNTRY_CODE = 0x21 # 0x21 Uint8, USA -HID_VIRTUAL_CABLE = True # Virtual cable enabled -HID_RECONNECT_INITIATE = True # Reconnect initiate enabled -REPORT_DESCRIPTOR_TYPE = 0x22 # 0x22 Type = Report Descriptor -HID_LANGID_BASE_LANGUAGE = 0x0409 # 0x0409 Language = English (United States) -HID_LANGID_BASE_BLUETOOTH_STRING_OFFSET = 0x100 # 0x0100 Default -HID_BATTERY_POWER = True # Battery power enabled -HID_REMOTE_WAKE = True # Remote wake enabled -HID_SUPERVISION_TIMEOUT = 0xC80 # uint16 0xC80 (2s) -HID_NORMALLY_CONNECTABLE = True # Normally connectable enabled -HID_BOOT_DEVICE = True # Boot device support enabled -HID_SSR_HOST_MAX_LATENCY = 0x640 # uint16 0x640 (1s) -HID_SSR_HOST_MIN_TIMEOUT = 0xC80 # uint16 0xC80 (2s) -HID_REPORT_MAP = bytes( # Text String, 50 Octet Report Descriptor - # pylint: disable=line-too-long - [ - 0x05, - 0x01, # Usage Page (Generic Desktop Ctrls) - 0x09, - 0x06, # Usage (Keyboard) - 0xA1, - 0x01, # Collection (Application) - 0x85, - 0x01, # . Report ID (1) - 0x05, - 0x07, # . Usage Page (Kbrd/Keypad) - 0x19, - 0xE0, # . Usage Minimum (0xE0) - 0x29, - 0xE7, # . Usage Maximum (0xE7) - 0x15, - 0x00, # . Logical Minimum (0) - 0x25, - 0x01, # . Logical Maximum (1) - 0x75, - 0x01, # . Report Size (1) - 0x95, - 0x08, # . Report Count (8) - 0x81, - 0x02, # . Input (Data,Var,Abs,No Wrap,Linear,Preferred State,No Null Position) - 0x95, - 0x01, # . Report Count (1) - 0x75, - 0x08, # . Report Size (8) - 0x81, - 0x03, # . Input (Const,Var,Abs,No Wrap,Linear,Preferred State,No Null Position) - 0x95, - 0x05, # . Report Count (5) - 0x75, - 0x01, # . Report Size (1) - 0x05, - 0x08, # . Usage Page (LEDs) - 0x19, - 0x01, # . Usage Minimum (Num Lock) - 0x29, - 0x05, # . Usage Maximum (Kana) - 0x91, - 0x02, # . Output (Data,Var,Abs,No Wrap,Linear,Preferred State,No Null Position,Non-volatile) - 0x95, - 0x01, # . Report Count (1) - 0x75, - 0x03, # . Report Size (3) - 0x91, - 0x03, # . Output (Const,Var,Abs,No Wrap,Linear,Preferred State,No Null Position,Non-volatile) - 0x95, - 0x06, # . Report Count (6) - 0x75, - 0x08, # . Report Size (8) - 0x15, - 0x00, # . Logical Minimum (0) - 0x25, - 0x65, # . Logical Maximum (101) - 0x05, - 0x07, # . Usage Page (Kbrd/Keypad) - 0x19, - 0x00, # . Usage Minimum (0x00) - 0x29, - 0x65, # . Usage Maximum (0x65) - 0x81, - 0x00, # . Input (Data,Array,Abs,No Wrap,Linear,Preferred State,No Null Position) - 0xC0, # End Collection - 0x05, - 0x01, # Usage Page (Generic Desktop Ctrls) - 0x09, - 0x02, # Usage (Mouse) - 0xA1, - 0x01, # Collection (Application) - 0x85, - 0x02, # . Report ID (2) - 0x09, - 0x01, # . Usage (Pointer) - 0xA1, - 0x00, # . Collection (Physical) - 0x05, - 0x09, # . Usage Page (Button) - 0x19, - 0x01, # . Usage Minimum (0x01) - 0x29, - 0x03, # . Usage Maximum (0x03) - 0x15, - 0x00, # . Logical Minimum (0) - 0x25, - 0x01, # . Logical Maximum (1) - 0x95, - 0x03, # . Report Count (3) - 0x75, - 0x01, # . Report Size (1) - 0x81, - 0x02, # . Input (Data,Var,Abs,No Wrap,Linear,Preferred State,No Null Position) - 0x95, - 0x01, # . Report Count (1) - 0x75, - 0x05, # . Report Size (5) - 0x81, - 0x03, # . Input (Const,Var,Abs,No Wrap,Linear,Preferred State,No Null Position) - 0x05, - 0x01, # . Usage Page (Generic Desktop Ctrls) - 0x09, - 0x30, # . Usage (X) - 0x09, - 0x31, # . Usage (Y) - 0x15, - 0x81, # . Logical Minimum (-127) - 0x25, - 0x7F, # . Logical Maximum (127) - 0x75, - 0x08, # . Report Size (8) - 0x95, - 0x02, # . Report Count (2) - 0x81, - 0x06, # . Input (Data,Var,Rel,No Wrap,Linear,Preferred State,No Null Position) - 0xC0, # . End Collection - 0xC0, # End Collection - ] -) - - -# Default protocol mode set to report protocol -protocol_mode = Message.ProtocolMode.REPORT_PROTOCOL - - -# ----------------------------------------------------------------------------- -def sdp_records(): - service_record_handle = 0x00010002 - return { - service_record_handle: [ - ServiceAttribute( - SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID, - DataElement.unsigned_integer_32(service_record_handle), - ), - ServiceAttribute( - SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID, - DataElement.sequence([DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)]), - ), - ServiceAttribute( - SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID, - DataElement.sequence( - [DataElement.uuid(BT_HUMAN_INTERFACE_DEVICE_SERVICE)] - ), - ), - ServiceAttribute( - SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, - DataElement.sequence( - [ - DataElement.sequence( - [ - DataElement.uuid(BT_L2CAP_PROTOCOL_ID), - DataElement.unsigned_integer_16(HID_CONTROL_PSM), - ] - ), - DataElement.sequence( - [ - DataElement.uuid(BT_HIDP_PROTOCOL_ID), - ] - ), - ] - ), - ), - ServiceAttribute( - SDP_LANGUAGE_BASE_ATTRIBUTE_ID_LIST_ATTRIBUTE_ID, - DataElement.sequence( - [ - DataElement.unsigned_integer_16(LANGUAGE), - DataElement.unsigned_integer_16(ENCODING), - DataElement.unsigned_integer_16(PRIMARY_LANGUAGE_BASE_ID), - ] - ), - ), - ServiceAttribute( - SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID, - DataElement.sequence( - [ - DataElement.sequence( - [ - DataElement.uuid(BT_HUMAN_INTERFACE_DEVICE_SERVICE), - DataElement.unsigned_integer_16(VERSION_NUMBER), - ] - ), - ] - ), - ), - ServiceAttribute( - SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, - DataElement.sequence( - [ - DataElement.sequence( - [ - DataElement.sequence( - [ - DataElement.uuid(BT_L2CAP_PROTOCOL_ID), - DataElement.unsigned_integer_16( - HID_INTERRUPT_PSM - ), - ] - ), - DataElement.sequence( - [ - DataElement.uuid(BT_HIDP_PROTOCOL_ID), - ] - ), - ] - ), - ] - ), - ), - ServiceAttribute( - SDP_HID_SERVICE_NAME_ATTRIBUTE_ID, - DataElement(DataElement.TEXT_STRING, SERVICE_NAME), - ), - ServiceAttribute( - SDP_HID_SERVICE_DESCRIPTION_ATTRIBUTE_ID, - DataElement(DataElement.TEXT_STRING, SERVICE_DESCRIPTION), - ), - ServiceAttribute( - SDP_HID_PROVIDER_NAME_ATTRIBUTE_ID, - DataElement(DataElement.TEXT_STRING, PROVIDER_NAME), - ), - ServiceAttribute( - SDP_HID_PARSER_VERSION_ATTRIBUTE_ID, - DataElement.unsigned_integer_32(HID_PARSER_VERSION), - ), - ServiceAttribute( - SDP_HID_DEVICE_SUBCLASS_ATTRIBUTE_ID, - DataElement.unsigned_integer_32(HID_DEVICE_SUBCLASS), - ), - ServiceAttribute( - SDP_HID_COUNTRY_CODE_ATTRIBUTE_ID, - DataElement.unsigned_integer_32(HID_COUNTRY_CODE), - ), - ServiceAttribute( - SDP_HID_VIRTUAL_CABLE_ATTRIBUTE_ID, - DataElement.boolean(HID_VIRTUAL_CABLE), - ), - ServiceAttribute( - SDP_HID_RECONNECT_INITIATE_ATTRIBUTE_ID, - DataElement.boolean(HID_RECONNECT_INITIATE), - ), - ServiceAttribute( - SDP_HID_DESCRIPTOR_LIST_ATTRIBUTE_ID, - DataElement.sequence( - [ - DataElement.sequence( - [ - DataElement.unsigned_integer_16(REPORT_DESCRIPTOR_TYPE), - DataElement(DataElement.TEXT_STRING, HID_REPORT_MAP), - ] - ), - ] - ), - ), - ServiceAttribute( - SDP_HID_LANGID_BASE_LIST_ATTRIBUTE_ID, - DataElement.sequence( - [ - DataElement.sequence( - [ - DataElement.unsigned_integer_16( - HID_LANGID_BASE_LANGUAGE - ), - DataElement.unsigned_integer_16( - HID_LANGID_BASE_BLUETOOTH_STRING_OFFSET - ), - ] - ), - ] - ), - ), - ServiceAttribute( - SDP_HID_BATTERY_POWER_ATTRIBUTE_ID, - DataElement.boolean(HID_BATTERY_POWER), - ), - ServiceAttribute( - SDP_HID_REMOTE_WAKE_ATTRIBUTE_ID, - DataElement.boolean(HID_REMOTE_WAKE), - ), - ServiceAttribute( - SDP_HID_SUPERVISION_TIMEOUT_ATTRIBUTE_ID, - DataElement.unsigned_integer_16(HID_SUPERVISION_TIMEOUT), - ), - ServiceAttribute( - SDP_HID_NORMALLY_CONNECTABLE_ATTRIBUTE_ID, - DataElement.boolean(HID_NORMALLY_CONNECTABLE), - ), - ServiceAttribute( - SDP_HID_BOOT_DEVICE_ATTRIBUTE_ID, - DataElement.boolean(HID_BOOT_DEVICE), - ), - ServiceAttribute( - SDP_HID_SSR_HOST_MAX_LATENCY_ATTRIBUTE_ID, - DataElement.unsigned_integer_16(HID_SSR_HOST_MAX_LATENCY), - ), - ServiceAttribute( - SDP_HID_SSR_HOST_MIN_TIMEOUT_ATTRIBUTE_ID, - DataElement.unsigned_integer_16(HID_SSR_HOST_MIN_TIMEOUT), - ), - ] - } - # ----------------------------------------------------------------------------- async def get_stream_reader(pipe) -> asyncio.StreamReader: @@ -410,71 +44,127 @@ async def get_stream_reader(pipe) -> asyncio.StreamReader: class DeviceData: def __init__(self) -> None: - self.keyboardData = bytearray( + self.keyboard_data = bytearray( [0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00] ) - self.mouseData = bytearray([0x02, 0x00, 0x00, 0x00]) + self.mouse_data = bytearray([0x02, 0x00, 0x00, 0x00]) # Device's live data - Mouse and Keyboard will be stored in this -deviceData = DeviceData() +device_data = DeviceData() + + +class HidDeviceDelegate(hid.Device.Delegate): + @override + def get_report(self, report_type: hid.ReportType, report_id: int | None) -> bytes: + print(f"GET_REPORT report_id: {report_id}, report_type: {report_type.name}") + match (report_type, report_id): + case (hid.ReportType.INPUT_REPORT, 1): + return bytes(device_data.keyboard_data[1:]) + case (hid.ReportType.INPUT_REPORT, 2): + return bytes(device_data.mouse_data[1:]) + case (hid.ReportType.INPUT_REPORT, _) | ( + hid.ReportType.OTHER_REPORT, + 3, + ): + raise hid.HidProtocolError( + hid.HandshakeMessage.ResultCode.ERR_INVALID_REPORT_ID + ) + case (hid.ReportType.OUTPUT_REPORT, _): + # Return single byte sample data for testing + return bytes([0x11]) + case (hid.ReportType.FEATURE_REPORT, _): + raise hid.HidProtocolError( + hid.HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER + ) + case _: + raise hid.HidProtocolError( + hid.HandshakeMessage.ResultCode.ERR_UNSUPPORTED_REQUEST + ) + + @override + def set_report(self, report_type: hid.ReportType, data: bytes) -> None: + report_id = data[0] if data else 0 + print( + f"SET_REPORT report_id: {report_id}, report_type: {report_type.name}, " + f"report_size: {len(data)}, data: {data.hex()}" + ) + match (report_type, report_id): + case (hid.ReportType.FEATURE_REPORT, _): + raise hid.HidProtocolError( + hid.HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER + ) + case (hid.ReportType.INPUT_REPORT, 1) if len(data) != len( + device_data.keyboard_data + ): + raise hid.HidProtocolError( + hid.HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER + ) + case (hid.ReportType.INPUT_REPORT, 2) if len(data) != len( + device_data.mouse_data + ): + raise hid.HidProtocolError( + hid.HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER + ) + case (hid.ReportType.INPUT_REPORT, 3): + raise hid.HidProtocolError( + hid.HandshakeMessage.ResultCode.ERR_INVALID_REPORT_ID + ) # ----------------------------------------------------------------------------- -async def keyboard_device(hid_device: HID_Device): +async def keyboard_device(hid_device: hid.Device) -> None: # Start a Websocket server to receive events from a web page - async def serve(websocket: websockets.asyncio.server.ServerConnection): - global deviceData + async def serve(websocket: websockets.asyncio.server.ServerConnection) -> None: while True: try: message = await websocket.recv() - print('Received: ', str(message)) + print("Received: ", str(message)) parsed = json.loads(message) - message_type = parsed['type'] - if message_type == 'keydown': - # Only deal with keys a to z for now - key = parsed['key'] - if len(key) == 1: - code = ord(key) - if ord('a') <= code <= ord('z'): - hid_code = 0x04 + code - ord('a') - deviceData.keyboardData = bytearray( - [ - 0x01, - 0x00, - 0x00, - hid_code, - 0x00, - 0x00, - 0x00, - 0x00, - 0x00, - ] - ) - hid_device.send_data(deviceData.keyboardData) - elif message_type == 'keyup': - deviceData.keyboardData = bytearray( - [0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00] - ) - hid_device.send_data(deviceData.keyboardData) - elif message_type == "mousemove": - # logical min and max values - log_min = -127 - log_max = 127 - x = parsed['x'] - y = parsed['y'] - # limiting x and y values within logical max and min range - x = max(log_min, min(log_max, x)) - y = max(log_min, min(log_max, y)) - deviceData.mouseData = bytearray([0x02, 0x00]) + struct.pack( - ">bb", x, y - ) - hid_device.send_data(deviceData.mouseData) + match parsed["type"]: + case "keydown": + # Only deal with keys a to z for now + key = parsed["key"] + if len(key) == 1: + code = ord(key) + if ord("a") <= code <= ord("z"): + hid_code = 0x04 + code - ord("a") + device_data.keyboard_data = bytearray( + [ + 0x01, + 0x00, + 0x00, + hid_code, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + ] + ) + hid_device.send_interrupt_data( + bytes(device_data.keyboard_data) + ) + case "keyup": + device_data.keyboard_data = bytearray( + [0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00] + ) + hid_device.send_interrupt_data(bytes(device_data.keyboard_data)) + case "mousemove": + # logical min and max values + log_min = -127 + log_max = 127 + x = max(log_min, min(log_max, parsed["x"])) + y = max(log_min, min(log_max, parsed["y"])) + device_data.mouse_data = bytearray([0x02, 0x00]) + struct.pack( + ">bb", x, y + ) + hid_device.send_interrupt_data(bytes(device_data.mouse_data)) except websockets.exceptions.ConnectionClosedOK: pass # pylint: disable-next=no-member - await websockets.asyncio.server.serve(serve, 'localhost', 8989) + await websockets.asyncio.server.serve(serve, "localhost", 8989) await asyncio.get_event_loop().create_future() @@ -482,118 +172,20 @@ async def serve(websocket: websockets.asyncio.server.ServerConnection): async def main() -> None: if len(sys.argv) < 3: print( - 'Usage: python run_hid_device.py ' - ' where is one of:\n' - ' test-mode (run with menu enabled for testing)\n' - ' web (run a keyboard with keypress input from a web page, ' - 'see keyboard.html' + "Usage: python run_hid_device.py " + "\n" + " where is one of:\n" + " test-mode (run with menu enabled for testing)\n" + " web (run a keyboard with keypress input from a web page, " + "see keyboard.html" ) - print('example: python run_hid_device.py hid_keyboard.json usb:0 web') - print('example: python run_hid_device.py hid_keyboard.json usb:0 test-mode') - + print("example: python run_hid_device.py hid_keyboard.json usb:0 web") + print("example: python run_hid_device.py hid_keyboard.json usb:0 test-mode") return - async def handle_virtual_cable_unplug(): - hid_host_bd_addr = str(hid_device.remote_device_bd_address) - await hid_device.disconnect_interrupt_channel() - await hid_device.disconnect_control_channel() - await device.keystore.delete(hid_host_bd_addr) # type: ignore - connection = hid_device.connection - if connection is not None: - await connection.disconnect() - - def on_hid_data_cb(pdu: bytes): - print(f'Received Data, PDU: {pdu.hex()}') - - def on_get_report_cb( - report_id: int, report_type: int, buffer_size: int - ) -> HID_Device.GetSetStatus: - retValue = hid_device.GetSetStatus() - print( - "GET_REPORT report_id: " - + str(report_id) - + "report_type: " - + str(report_type) - + "buffer_size:" - + str(buffer_size) - ) - if report_type == Message.ReportType.INPUT_REPORT: - if report_id == 1: - retValue.data = deviceData.keyboardData[1:] - retValue.status = hid_device.GetSetReturn.SUCCESS - elif report_id == 2: - retValue.data = deviceData.mouseData[1:] - retValue.status = hid_device.GetSetReturn.SUCCESS - else: - retValue.status = hid_device.GetSetReturn.REPORT_ID_NOT_FOUND - - if buffer_size: - data_len = buffer_size - 1 - retValue.data = retValue.data[:data_len] - elif report_type == Message.ReportType.OUTPUT_REPORT: - # This sample app has nothing to do with the report received, to enable PTS - # testing, we will return single byte random data. - retValue.data = bytearray([0x11]) - retValue.status = hid_device.GetSetReturn.SUCCESS - elif report_type == Message.ReportType.FEATURE_REPORT: - retValue.status = hid_device.GetSetReturn.ERR_INVALID_PARAMETER - elif report_type == Message.ReportType.OTHER_REPORT: - if report_id == 3: - retValue.status = hid_device.GetSetReturn.REPORT_ID_NOT_FOUND - else: - retValue.status = hid_device.GetSetReturn.FAILURE - - return retValue - - def on_set_report_cb( - report_id: int, report_type: int, report_size: int, data: bytes - ) -> HID_Device.GetSetStatus: - print( - "SET_REPORT report_id: " - + str(report_id) - + "report_type: " - + str(report_type) - + "report_size " - + str(report_size) - + "data:" - + str(data) - ) - if report_type == Message.ReportType.FEATURE_REPORT: - status = HID_Device.GetSetReturn.ERR_INVALID_PARAMETER - elif report_type == Message.ReportType.INPUT_REPORT: - if report_id == 1 and report_size != len(deviceData.keyboardData): - status = HID_Device.GetSetReturn.ERR_INVALID_PARAMETER - elif report_id == 2 and report_size != len(deviceData.mouseData): - status = HID_Device.GetSetReturn.ERR_INVALID_PARAMETER - elif report_id == 3: - status = HID_Device.GetSetReturn.REPORT_ID_NOT_FOUND - else: - status = HID_Device.GetSetReturn.SUCCESS - else: - status = HID_Device.GetSetReturn.SUCCESS - - return HID_Device.GetSetStatus(status=status) - - def on_get_protocol_cb() -> HID_Device.GetSetStatus: - return HID_Device.GetSetStatus( - data=bytes([protocol_mode]), - status=hid_device.GetSetReturn.SUCCESS, - ) - - def on_set_protocol_cb(protocol: int) -> HID_Device.GetSetStatus: - # We do not support SET_PROTOCOL. - print(f"SET_PROTOCOL report_id: {protocol}") - return HID_Device.GetSetStatus( - status=hid_device.GetSetReturn.ERR_UNSUPPORTED_REQUEST - ) - - def on_virtual_cable_unplug_cb(): - print('Received Virtual Cable Unplug') - asyncio.create_task(handle_virtual_cable_unplug()) - - print('<<< connecting to HCI...') + print("<<< connecting to HCI...") async with await open_transport(sys.argv[2]) as hci_transport: - print('<<< connected') + print("<<< connected") # Create a device device = Device.from_config_file_with_hci( @@ -601,22 +193,56 @@ def on_virtual_cable_unplug_cb(): ) device.classic_enabled = True + active_connection: Connection | None = None + + def on_connection(connection: Connection) -> None: + nonlocal active_connection + active_connection = connection + + def on_disconnection(_reason: int) -> None: + nonlocal active_connection + active_connection = None + + connection.on( + connection.EVENT_DISCONNECTION, + on_disconnection, + ) + + device.on(device.EVENT_CONNECTION, on_connection) + # Create and register HID device - hid_device = HID_Device(device) + delegate = HidDeviceDelegate() + hid_device = hid.Device(device, delegate=delegate) + + async def handle_virtual_cable_unplug() -> None: + await hid_device.disconnect() + if hid_device.remote_device_bd_address and device.keystore: + try: + await device.keystore.delete( + str(hid_device.remote_device_bd_address) + ) + except KeyError: + pass + if active_connection is not None: + await active_connection.disconnect() - # Register for call backs - hid_device.on('interrupt_data', on_hid_data_cb) + def on_hid_data_cb(report_type: hid.ReportType, data: bytes) -> None: + print(f"Received Data, report_type: {report_type.name}, PDU: {data.hex()}") - hid_device.register_get_report_cb(on_get_report_cb) - hid_device.register_set_report_cb(on_set_report_cb) - hid_device.register_get_protocol_cb(on_get_protocol_cb) - hid_device.register_set_protocol_cb(on_set_protocol_cb) + def on_virtual_cable_unplug_cb() -> None: + print("Received Virtual Cable Unplug") + asyncio.create_task(handle_virtual_cable_unplug()) - # Register for virtual cable unplug call back - hid_device.on('virtual_cable_unplug', on_virtual_cable_unplug_cb) + hid_device.on(hid_device.EVENT_INTERRUPT_DATA, on_hid_data_cb) + hid_device.on(hid_device.EVENT_VIRTUAL_CABLE_UNPLUG, on_virtual_cable_unplug_cb) # Setup the SDP to advertise HID Device service - device.sdp_service_records = sdp_records() + device.sdp_service_records = { + 0x00010002: hid.DeviceSdpRecord( + service_record_handle=0x00010002, + report_map=hid.DEFAULT_REPORT_MAP, + ).to_service_attributes() + } # Start the controller await device.power_on() @@ -625,109 +251,135 @@ def on_virtual_cable_unplug_cb(): await device.set_discoverable(True) await device.set_connectable(True) - async def menu(): + async def menu() -> None: + nonlocal active_connection reader = await get_stream_reader(sys.stdin) while True: - print( - "\n************************ HID Device Menu *****************************\n" - ) - print(" 1. Connect Control Channel") - print(" 2. Connect Interrupt Channel") - print(" 3. Disconnect Control Channel") - print(" 4. Disconnect Interrupt Channel") - print(" 5. Send Report on Interrupt Channel") - print(" 6. Virtual Cable Unplug") - print(" 7. Disconnect device") - print(" 8. Delete Bonding") - print(" 9. Re-connect to device") - print("10. Exit ") + print("\n" + "*" * 20 + " HID Device Menu " + "*" * 20 + "\n") + print(" 1. Connect HID Channels") + print(" 2. Disconnect HID Channels") + print(" 3. Send Report on Interrupt Channel") + print(" 4. Virtual Cable Unplug") + print(" 5. Disconnect device") + print(" 6. Delete Bonding") + print(" 7. Re-connect to device") + print(" 8. Exit ") print("\nEnter your choice : \n") - choice = await reader.readline() - choice = choice.decode('utf-8').strip() - - if choice == '1': - await hid_device.connect_control_channel() - - elif choice == '2': - await hid_device.connect_interrupt_channel() - - elif choice == '3': - await hid_device.disconnect_control_channel() - - elif choice == '4': - await hid_device.disconnect_interrupt_channel() - - elif choice == '5': - print(" 1. Report ID 0x01") - print(" 2. Report ID 0x02") - print(" 3. Invalid Report ID") - - choice1 = await reader.readline() - choice1 = choice1.decode('utf-8').strip() - - if choice1 == '1': - data = bytearray( - [0x01, 0x00, 0x00, 0x04, 0x00, 0x00, 0x00, 0x00, 0x00] - ) - hid_device.send_data(data) - data = bytearray( - [0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00] - ) - hid_device.send_data(data) - - elif choice1 == '2': - data = bytearray([0x02, 0x00, 0x00, 0xF6]) - hid_device.send_data(data) - data = bytearray([0x02, 0x00, 0x00, 0x00]) - hid_device.send_data(data) - - elif choice1 == '3': - data = bytearray([0x00, 0x00, 0x00, 0x00]) - hid_device.send_data(data) - data = bytearray([0x00, 0x00, 0x00, 0x00]) - hid_device.send_data(data) - - else: - print('Incorrect option selected') - - elif choice == '6': - hid_device.virtual_cable_unplug() - try: - hid_host_bd_addr = str(hid_device.remote_device_bd_address) - await device.keystore.delete(hid_host_bd_addr) - except KeyError: - print('Device not found or Device already unpaired.') - - elif choice == '7': - connection = hid_device.connection - if connection is not None: - await connection.disconnect() - else: - print("Already disconnected from device") - - elif choice == '8': - try: - hid_host_bd_addr = str(hid_device.remote_device_bd_address) - await device.keystore.delete(hid_host_bd_addr) - except KeyError: - print('Device NOT found or Device already unpaired.') - - elif choice == '9': - hid_host_bd_addr = str(hid_device.remote_device_bd_address) - connection = await device.connect( - hid_host_bd_addr, transport=PhysicalTransport.BR_EDR - ) - await connection.authenticate() - await connection.encrypt() + choice_line = await reader.readline() + choice = choice_line.decode("utf-8").strip() + + match choice: + case "1": + if active_connection is not None: + await hid_device.connect(active_connection) + else: + print("No active connection") + + case "2": + await hid_device.disconnect() + + case "3": + print(" 1. Report ID 0x01") + print(" 2. Report ID 0x02") + print(" 3. Invalid Report ID") + + choice1_line = await reader.readline() + choice1 = choice1_line.decode("utf-8").strip() + + match choice1: + case "1": + hid_device.send_interrupt_data( + bytes( + [ + 0x01, + 0x00, + 0x00, + 0x04, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + ] + ) + ) + hid_device.send_interrupt_data( + bytes( + [ + 0x01, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + ] + ) + ) + case "2": + hid_device.send_interrupt_data( + bytes([0x02, 0x00, 0x00, 0xF6]) + ) + hid_device.send_interrupt_data( + bytes([0x02, 0x00, 0x00, 0x00]) + ) + case "3": + hid_device.send_interrupt_data( + bytes([0x00, 0x00, 0x00, 0x00]) + ) + hid_device.send_interrupt_data( + bytes([0x00, 0x00, 0x00, 0x00]) + ) + case _: + print("Incorrect option selected") + + case "4": + hid_device.virtual_cable_unplug() + if hid_device.remote_device_bd_address and device.keystore: + try: + await device.keystore.delete( + str(hid_device.remote_device_bd_address) + ) + except KeyError: + print("Device not found or Device already unpaired.") + + case "5": + if active_connection is not None: + await active_connection.disconnect() + active_connection = None + else: + print("Already disconnected from device") + + case "6": + if hid_device.remote_device_bd_address and device.keystore: + try: + await device.keystore.delete( + str(hid_device.remote_device_bd_address) + ) + except KeyError: + print("Device NOT found or Device already unpaired.") + + case "7": + if hid_device.remote_device_bd_address: + active_connection = await device.connect( + hid_device.remote_device_bd_address, + transport=PhysicalTransport.BR_EDR, + ) + await active_connection.authenticate() + await active_connection.encrypt() + else: + print("Remote device address unknown.") - elif choice == '10': - sys.exit("Exit successful") + case "8": + sys.exit("Exit successful") - else: - print("Invalid option selected.") + case _: + print("Invalid option selected.") - if (len(sys.argv) > 3) and (sys.argv[3] == 'test-mode'): + if (len(sys.argv) > 3) and (sys.argv[3] == "test-mode"): # Test mode for PTS/Unit testing await menu() else: @@ -739,5 +391,6 @@ async def menu(): # ----------------------------------------------------------------------------- -bumble.logging.setup_basic_logging('DEBUG') -asyncio.run(main()) +if __name__ == "__main__": + bumble.logging.setup_basic_logging("DEBUG") + asyncio.run(main()) diff --git a/examples/run_hid_host.py b/examples/run_hid_host.py index a3d4feff2..e403dc206 100644 --- a/examples/run_hid_host.py +++ b/examples/run_hid_host.py @@ -1,4 +1,4 @@ -# Copyright 2021-2022 Google LLC +# 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. @@ -15,249 +15,40 @@ # ----------------------------------------------------------------------------- # Imports # ----------------------------------------------------------------------------- +# pylint: disable=duplicate-code +from __future__ import annotations + import asyncio +import dataclasses import sys from hid_report_parser import ReportParser import bumble.logging +from bumble import hid from bumble.colors import color -from bumble.core import BT_HUMAN_INTERFACE_DEVICE_SERVICE, PhysicalTransport -from bumble.device import Device +from bumble.core import PhysicalTransport +from bumble.device import Connection, Device from bumble.hci import Address -from bumble.hid import Host, Message -from bumble.sdp import ( - SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, - SDP_ALL_ATTRIBUTES_RANGE, - SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID, - SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID, - SDP_LANGUAGE_BASE_ATTRIBUTE_ID_LIST_ATTRIBUTE_ID, - SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, - SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID, - SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID, -) -from bumble.sdp import Client as SDP_Client from bumble.transport import open_transport -# ----------------------------------------------------------------------------- -# SDP attributes for Bluetooth HID devices -SDP_HID_SERVICE_NAME_ATTRIBUTE_ID = 0x0100 -SDP_HID_SERVICE_DESCRIPTION_ATTRIBUTE_ID = 0x0101 -SDP_HID_PROVIDER_NAME_ATTRIBUTE_ID = 0x0102 -SDP_HID_DEVICE_RELEASE_NUMBER_ATTRIBUTE_ID = 0x0200 # [DEPRECATED] -SDP_HID_PARSER_VERSION_ATTRIBUTE_ID = 0x0201 -SDP_HID_DEVICE_SUBCLASS_ATTRIBUTE_ID = 0x0202 -SDP_HID_COUNTRY_CODE_ATTRIBUTE_ID = 0x0203 -SDP_HID_VIRTUAL_CABLE_ATTRIBUTE_ID = 0x0204 -SDP_HID_RECONNECT_INITIATE_ATTRIBUTE_ID = 0x0205 -SDP_HID_DESCRIPTOR_LIST_ATTRIBUTE_ID = 0x0206 -SDP_HID_LANGID_BASE_LIST_ATTRIBUTE_ID = 0x0207 -SDP_HID_SDP_DISABLE_ATTRIBUTE_ID = 0x0208 # [DEPRECATED] -SDP_HID_BATTERY_POWER_ATTRIBUTE_ID = 0x0209 -SDP_HID_REMOTE_WAKE_ATTRIBUTE_ID = 0x020A -SDP_HID_PROFILE_VERSION_ATTRIBUTE_ID = 0x020B # DEPRECATED] -SDP_HID_SUPERVISION_TIMEOUT_ATTRIBUTE_ID = 0x020C -SDP_HID_NORMALLY_CONNECTABLE_ATTRIBUTE_ID = 0x020D -SDP_HID_BOOT_DEVICE_ATTRIBUTE_ID = 0x020E -SDP_HID_SSR_HOST_MAX_LATENCY_ATTRIBUTE_ID = 0x020F -SDP_HID_SSR_HOST_MIN_TIMEOUT_ATTRIBUTE_ID = 0x0210 - # ----------------------------------------------------------------------------- - - -async def get_hid_device_sdp_record(connection): - # Connect to the SDP Server - sdp_client = SDP_Client(connection) - await sdp_client.connect() - if sdp_client: - print(color('Connected to SDP Server', 'blue')) - else: - print(color('Failed to connect to SDP Server', 'red')) - - # List BT HID Device service in the root browse group - service_record_handles = await sdp_client.search_services( - [BT_HUMAN_INTERFACE_DEVICE_SERVICE] - ) - - if len(service_record_handles) < 1: - await sdp_client.disconnect() - raise Exception( - color('BT HID Device service not found on peer device!!!!', 'red') +async def get_hid_device_sdp_record(connection: Connection) -> None: + sdp_records = await hid.DeviceSdpRecord.find(connection) + if not sdp_records: + raise RuntimeError( + color("BT HID Device service not found on peer device!", "red") ) + sdp_info = sdp_records[0] - # For BT_HUMAN_INTERFACE_DEVICE_SERVICE service, get all its attributes - for service_record_handle in service_record_handles: - attributes = await sdp_client.get_attributes( - service_record_handle, [SDP_ALL_ATTRIBUTES_RANGE] - ) - print(color('SERVICE {service_record_handle:04X} attributes:', 'yellow')) - print(color('SDP attributes for HID device', 'magenta')) - for attribute in attributes: - if attribute.id == SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID: - print( - color(' Service Record Handle : ', 'cyan'), - hex(attribute.value.value), - ) - - elif attribute.id == SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID: - print( - color(' Service Class : ', 'cyan'), attribute.value.value[0].value - ) - - elif attribute.id == SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID: - print( - color(' SDP Browse Group List : ', 'cyan'), - attribute.value.value[0].value, - ) - - elif attribute.id == SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID: - print( - color(' BT_L2CAP_PROTOCOL_ID : ', 'cyan'), - attribute.value.value[0].value[0].value, - ) - print( - color(' PSM for Bluetooth HID Control channel : ', 'cyan'), - hex(attribute.value.value[0].value[1].value), - ) - print( - color(' BT_HIDP_PROTOCOL_ID : ', 'cyan'), - attribute.value.value[1].value[0].value, - ) - - elif attribute.id == SDP_LANGUAGE_BASE_ATTRIBUTE_ID_LIST_ATTRIBUTE_ID: - print( - color(' Lanugage : ', 'cyan'), hex(attribute.value.value[0].value) - ) - print( - color(' Encoding : ', 'cyan'), hex(attribute.value.value[1].value) - ) - print( - color(' PrimaryLanguageBaseID : ', 'cyan'), - hex(attribute.value.value[2].value), - ) - - elif attribute.id == SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID: - print( - color(' BT_HUMAN_INTERFACE_DEVICE_SERVICE ', 'cyan'), - attribute.value.value[0].value[0].value, - ) - print( - color(' HID Profileversion number : ', 'cyan'), - hex(attribute.value.value[0].value[1].value), - ) - - elif attribute.id == SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID: - print( - color(' BT_L2CAP_PROTOCOL_ID : ', 'cyan'), - attribute.value.value[0].value[0].value[0].value, - ) - print( - color(' PSM for Bluetooth HID Interrupt channel : ', 'cyan'), - hex(attribute.value.value[0].value[0].value[1].value), - ) - print( - color(' BT_HIDP_PROTOCOL_ID : ', 'cyan'), - attribute.value.value[0].value[1].value[0].value, - ) - - elif attribute.id == SDP_HID_SERVICE_NAME_ATTRIBUTE_ID: - print(color(' Service Name: ', 'cyan'), attribute.value.value) - - elif attribute.id == SDP_HID_SERVICE_DESCRIPTION_ATTRIBUTE_ID: - print(color(' Service Description: ', 'cyan'), attribute.value.value) - - elif attribute.id == SDP_HID_PROVIDER_NAME_ATTRIBUTE_ID: - print(color(' Provider Name: ', 'cyan'), attribute.value.value) - - elif attribute.id == SDP_HID_DEVICE_RELEASE_NUMBER_ATTRIBUTE_ID: - print(color(' Release Number: ', 'cyan'), hex(attribute.value.value)) - - elif attribute.id == SDP_HID_PARSER_VERSION_ATTRIBUTE_ID: - print( - color(' HID Parser Version: ', 'cyan'), hex(attribute.value.value) - ) - - elif attribute.id == SDP_HID_DEVICE_SUBCLASS_ATTRIBUTE_ID: - print( - color(' HIDDeviceSubclass: ', 'cyan'), hex(attribute.value.value) - ) - - elif attribute.id == SDP_HID_COUNTRY_CODE_ATTRIBUTE_ID: - print(color(' HIDCountryCode: ', 'cyan'), hex(attribute.value.value)) - - elif attribute.id == SDP_HID_VIRTUAL_CABLE_ATTRIBUTE_ID: - print(color(' HIDVirtualCable: ', 'cyan'), attribute.value.value) - - elif attribute.id == SDP_HID_RECONNECT_INITIATE_ATTRIBUTE_ID: - print(color(' HIDReconnectInitiate: ', 'cyan'), attribute.value.value) - - elif attribute.id == SDP_HID_DESCRIPTOR_LIST_ATTRIBUTE_ID: - print( - color(' HID Report Descriptor type: ', 'cyan'), - hex(attribute.value.value[0].value[0].value), - ) - print( - color(' HID Report DescriptorList: ', 'cyan'), - attribute.value.value[0].value[1].value, - ) - - elif attribute.id == SDP_HID_LANGID_BASE_LIST_ATTRIBUTE_ID: - print( - color(' HID LANGID Base Language: ', 'cyan'), - hex(attribute.value.value[0].value[0].value), - ) - print( - color(' HID LANGID Base Bluetooth String Offset: ', 'cyan'), - hex(attribute.value.value[0].value[1].value), - ) - - elif attribute.id == SDP_HID_BATTERY_POWER_ATTRIBUTE_ID: - print(color(' HIDBatteryPower: ', 'cyan'), attribute.value.value) - - elif attribute.id == SDP_HID_REMOTE_WAKE_ATTRIBUTE_ID: - print(color(' HIDRemoteWake: ', 'cyan'), attribute.value.value) - - elif attribute.id == SDP_HID_PROFILE_VERSION_ATTRIBUTE_ID: - print( - color(' HIDProfileVersion : ', 'cyan'), hex(attribute.value.value) - ) - - elif attribute.id == SDP_HID_SUPERVISION_TIMEOUT_ATTRIBUTE_ID: - print( - color(' HIDSupervisionTimeout: ', 'cyan'), - hex(attribute.value.value), - ) - - elif attribute.id == SDP_HID_NORMALLY_CONNECTABLE_ATTRIBUTE_ID: - print( - color(' HIDNormallyConnectable: ', 'cyan'), attribute.value.value - ) - - elif attribute.id == SDP_HID_BOOT_DEVICE_ATTRIBUTE_ID: - print(color(' HIDBootDevice: ', 'cyan'), attribute.value.value) - - elif attribute.id == SDP_HID_SSR_HOST_MAX_LATENCY_ATTRIBUTE_ID: - print( - color(' HIDSSRHostMaxLatency: ', 'cyan'), - hex(attribute.value.value), - ) - - elif attribute.id == SDP_HID_SSR_HOST_MIN_TIMEOUT_ATTRIBUTE_ID: - print( - color(' HIDSSRHostMinTimeout: ', 'cyan'), - hex(attribute.value.value), - ) - - else: - print( - color( - f' Warning: Attribute ID: {attribute.id} match not found.\n Attribute Info: {attribute}', - 'yellow', - ) - ) - - await sdp_client.disconnect() + print(color("SDP attributes for HID device:", "magenta")) + for field in dataclasses.fields(sdp_info): + val = getattr(sdp_info, field.name) + if isinstance(val, int) and not isinstance(val, bool): + print(color(f" {field.name}: ", "cyan"), hex(val)) + else: + print(color(f" {field.name}: ", "cyan"), val) # ----------------------------------------------------------------------------- @@ -273,54 +64,42 @@ async def get_stream_reader(pipe) -> asyncio.StreamReader: async def main() -> None: if len(sys.argv) < 4: print( - 'Usage: run_hid_host.py ' - ' [test-mode]' + "Usage: run_hid_host.py " + " [test-mode]" ) - - print('example: run_hid_host.py classic1.json usb:0 E1:CA:72:48:C4:E8/P') + print("example: run_hid_host.py classic1.json usb:0 E1:CA:72:48:C4:E8/P") return - def on_hid_control_data_cb(pdu: bytes): - print(f'Received Control Data, PDU: {pdu.hex()}') + def on_hid_control_data_cb(report_type: hid.ReportType, data: bytes) -> None: + print( + f"Received Control Data, report_type: {report_type.name}, " + f"PDU: {data.hex()}" + ) - def on_hid_interrupt_data_cb(pdu: bytes): - report_type = pdu[0] & 0x0F - if len(pdu) == 1: - print(color('Warning: No report received', 'yellow')) + def on_hid_interrupt_data_cb(report_type: hid.ReportType, data: bytes) -> None: + if not data: + print(color("Warning: No report received", "yellow")) return - report_length = len(pdu[1:]) - report_id = pdu[1] - - if report_type != Message.ReportType.OTHER_REPORT: - print( - color( - f' Report type = {report_type}, Report length = {report_length}, Report id = {report_id}', - 'blue', - None, - 'bold', - ) + report_id = data[0] + report_length = len(data) + + if report_type != hid.ReportType.OTHER_REPORT: + msg = ( + f" Report type = {report_type.name}, " + f"Report length = {report_length}, " + f"Report id = {report_id}" ) + print(color(msg, "blue", None, "bold")) if (report_length <= 1) or (report_id == 0): return # Parse report over interrupt channel - if report_type == Message.ReportType.INPUT_REPORT: - ReportParser.parse_input_report(pdu[1:]) # type: ignore - - async def handle_virtual_cable_unplug(): - await hid_host.disconnect_interrupt_channel() - await hid_host.disconnect_control_channel() - await device.keystore.delete(target_address) # type: ignore - connection = hid_host.connection - if connection is not None: - await connection.disconnect() + if report_type == hid.ReportType.INPUT_REPORT: + ReportParser.parse_input_report(data) - def on_hid_virtual_cable_unplug_cb(): - asyncio.create_task(handle_virtual_cable_unplug()) - - print('<<< connecting to HCI...') + print("<<< connecting to HCI...") async with await open_transport(sys.argv[2]) as hci_transport: - print('<<< CONNECTED') + print("<<< CONNECTED") # Create a device device = Device.from_config_file_with_hci( @@ -329,236 +108,302 @@ def on_hid_virtual_cable_unplug_cb(): device.classic_enabled = True # Create HID host and start it - print('@@@ Starting HID Host...') - hid_host = Host(device) - - # Register for HID data call back - hid_host.on('interrupt_data', on_hid_interrupt_data_cb) - hid_host.on('control_data', on_hid_control_data_cb) + print("@@@ Starting HID Host...") + hid_host = hid.Host(device) - # Register for virtual cable unplug call back - hid_host.on('virtual_cable_unplug', on_hid_virtual_cable_unplug_cb) + target_address = sys.argv[3] + active_connection: Connection | None = None + + async def handle_virtual_cable_unplug() -> None: + nonlocal active_connection + await hid_host.disconnect() + if device.keystore: + try: + await device.keystore.delete(target_address) + except KeyError: + pass + if active_connection is not None: + await active_connection.disconnect() + active_connection = None + + def on_hid_virtual_cable_unplug_cb() -> None: + asyncio.create_task(handle_virtual_cable_unplug()) + + # Register for HID data callbacks + hid_host.on(hid_host.EVENT_INTERRUPT_DATA, on_hid_interrupt_data_cb) + hid_host.on(hid_host.EVENT_CONTROL_DATA, on_hid_control_data_cb) + hid_host.on(hid_host.EVENT_VIRTUAL_CABLE_UNPLUG, on_hid_virtual_cable_unplug_cb) await device.power_on() # Connect to a peer - target_address = sys.argv[3] - print(f'=== Connecting to {target_address}...') - connection = await device.connect( + print(f"=== Connecting to {target_address}...") + active_connection = await device.connect( target_address, transport=PhysicalTransport.BR_EDR ) - print(f'=== Connected to {connection.peer_address}!') + print(f"=== Connected to {active_connection.peer_address}!") # Request authentication - print('*** Authenticating...') - await connection.authenticate() - print('*** Authenticated...') + print("*** Authenticating...") + await active_connection.authenticate() + print("*** Authenticated...") # Enable encryption - print('*** Enabling encryption...') - await connection.encrypt() - print('*** Encryption on') + print("*** Enabling encryption...") + await active_connection.encrypt() + print("*** Encryption on") - await get_hid_device_sdp_record(connection) + await get_hid_device_sdp_record(active_connection) - async def menu(): + async def menu() -> None: + nonlocal active_connection reader = await get_stream_reader(sys.stdin) while True: - print( - "\n************************ HID Host Menu *****************************\n" - ) - print(" 1. Connect Control Channel") - print(" 2. Connect Interrupt Channel") - print(" 3. Disconnect Control Channel") - print(" 4. Disconnect Interrupt Channel") - print(" 5. Get Report") - print(" 6. Set Report") - print(" 7. Set Protocol Mode") - print(" 8. Get Protocol Mode") - print(" 9. Send Report on Interrupt Channel") - print("10. Suspend") - print("11. Exit Suspend") - print("12. Virtual Cable Unplug") - print("13. Disconnect device") - print("14. Delete Bonding") - print("15. Re-connect to device") - print("16. Exit") + print("\n" + "*" * 20 + " HID Host Menu " + "*" * 20 + "\n") + print(" 1. Connect HID Channels") + print(" 2. Disconnect HID Channels") + print(" 3. Get Report") + print(" 4. Set Report") + print(" 5. Set Protocol Mode") + print(" 6. Get Protocol Mode") + print(" 7. Send Report on Interrupt Channel") + print(" 8. Suspend") + print(" 9. Exit Suspend") + print("10. Virtual Cable Unplug") + print("11. Disconnect device") + print("12. Delete Bonding") + print("13. Re-connect to device") + print("14. Exit") print("\nEnter your choice : \n") - choice = await reader.readline() - choice = choice.decode('utf-8').strip() - - if choice == '1': - await hid_host.connect_control_channel() - - elif choice == '2': - await hid_host.connect_interrupt_channel() - - elif choice == '3': - await hid_host.disconnect_control_channel() - - elif choice == '4': - await hid_host.disconnect_interrupt_channel() - - elif choice == '5': - print(" 1. Input Report with ID 0x01") - print(" 2. Input Report with ID 0x02") - print(" 3. Input Report with ID 0x0F - Invalid ReportId") - print(" 4. Output Report with ID 0x02") - print(" 5. Feature Report with ID 0x05 - Unsupported Request") - print(" 6. Input Report with ID 0x02, BufferSize 3") - print(" 7. Output Report with ID 0x03, BufferSize 2") - print(" 8. Feature Report with ID 0x05, BufferSize 3") - choice1 = await reader.readline() - choice1 = choice1.decode('utf-8').strip() - - if choice1 == '1': - hid_host.get_report(1, 1, 0) - - elif choice1 == '2': - hid_host.get_report(1, 2, 0) - - elif choice1 == '3': - hid_host.get_report(1, 5, 0) - - elif choice1 == '4': - hid_host.get_report(2, 2, 0) - - elif choice1 == '5': - hid_host.get_report(3, 15, 0) - - elif choice1 == '6': - hid_host.get_report(1, 2, 3) - - elif choice1 == '7': - hid_host.get_report(2, 3, 2) - - elif choice1 == '8': - hid_host.get_report(3, 5, 3) - else: - print('Incorrect option selected') - - elif choice == '6': - print(" 1. Report type 1 and Report id 0x01") - print(" 2. Report type 2 and Report id 0x03") - print(" 3. Report type 3 and Report id 0x05") - choice1 = await reader.readline() - choice1 = choice1.decode('utf-8').strip() - - if choice1 == '1': - # data includes first octet as report id - data = bytearray( - [0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01] - ) - hid_host.set_report(1, data) - - elif choice1 == '2': - data = bytearray([0x03, 0x01, 0x01]) - hid_host.set_report(2, data) - - elif choice1 == '3': - data = bytearray([0x05, 0x01, 0x01, 0x01]) - hid_host.set_report(3, data) - - else: - print('Incorrect option selected') - - elif choice == '7': - print(" 0. Boot") - print(" 1. Report") - choice1 = await reader.readline() - choice1 = choice1.decode('utf-8').strip() - - if choice1 == '0': - hid_host.set_protocol(Message.ProtocolMode.BOOT_PROTOCOL) - - elif choice1 == '1': - hid_host.set_protocol(Message.ProtocolMode.REPORT_PROTOCOL) - - else: - print('Incorrect option selected') - - elif choice == '8': - hid_host.get_protocol() - - elif choice == '9': - print(" 1. Report ID 0x01") - print(" 2. Report ID 0x03") - choice1 = await reader.readline() - choice1 = choice1.decode('utf-8').strip() - - if choice1 == '1': - data = bytearray( - [0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00] - ) - hid_host.send_data(data) - - elif choice1 == '2': - data = bytearray([0x03, 0x00, 0x0D, 0xFD, 0x00, 0x00]) - hid_host.send_data(data) - - else: - print('Incorrect option selected') - - elif choice == '10': - hid_host.suspend() - - elif choice == '11': - hid_host.exit_suspend() - - elif choice == '12': - hid_host.virtual_cable_unplug() - try: - await device.keystore.delete(target_address) - print("Unpair successful") - except KeyError: - print('Device not found or Device already unpaired.') - - elif choice == '13': - peer_address = Address.from_string_for_transport( - target_address, transport=PhysicalTransport.BR_EDR - ) - connection = device.find_connection_by_bd_addr( - peer_address, transport=PhysicalTransport.BR_EDR - ) - if connection is not None: - await connection.disconnect() - else: - print("Already disconnected from device") - - elif choice == '14': - try: - await device.keystore.delete(target_address) - print("Unpair successful") - except KeyError: - print('Device not found or Device already unpaired.') - - elif choice == '15': - connection = await device.connect( - target_address, transport=PhysicalTransport.BR_EDR - ) - await connection.authenticate() - await connection.encrypt() - - elif choice == '16': - sys.exit("Exit successful") - - else: - print("Invalid option selected.") - - if (len(sys.argv) > 4) and (sys.argv[4] == 'test-mode'): + choice_line = await reader.readline() + choice = choice_line.decode("utf-8").strip() + + try: + match choice: + case "1": + if active_connection is not None: + await hid_host.connect(active_connection) + else: + print("No active connection") + + case "2": + await hid_host.disconnect() + + case "3": + print(" 1. Input Report with ID 0x01") + print(" 2. Input Report with ID 0x02") + print(" 3. Input Report with ID 0x05 - Invalid ReportId") + print(" 4. Output Report with ID 0x02") + print( + " 5. Feature Report with ID 0x0F - Unsupported Request" + ) + print(" 6. Input Report with ID 0x02, BufferSize 3") + print(" 7. Output Report with ID 0x03, BufferSize 2") + print(" 8. Feature Report with ID 0x05, BufferSize 3") + choice1_line = await reader.readline() + choice1 = choice1_line.decode("utf-8").strip() + + match choice1: + case "1": + res = await hid_host.get_report( + hid.ReportType.INPUT_REPORT, 1 + ) + print(f"Report received: {res.hex()}") + case "2": + res = await hid_host.get_report( + hid.ReportType.INPUT_REPORT, 2 + ) + print(f"Report received: {res.hex()}") + case "3": + res = await hid_host.get_report( + hid.ReportType.INPUT_REPORT, 5 + ) + print(f"Report received: {res.hex()}") + case "4": + res = await hid_host.get_report( + hid.ReportType.OUTPUT_REPORT, 2 + ) + print(f"Report received: {res.hex()}") + case "5": + res = await hid_host.get_report( + hid.ReportType.FEATURE_REPORT, 15 + ) + print(f"Report received: {res.hex()}") + case "6": + res = await hid_host.get_report( + hid.ReportType.INPUT_REPORT, 2, 3 + ) + print(f"Report received: {res.hex()}") + case "7": + res = await hid_host.get_report( + hid.ReportType.OUTPUT_REPORT, 3, 2 + ) + print(f"Report received: {res.hex()}") + case "8": + res = await hid_host.get_report( + hid.ReportType.FEATURE_REPORT, 5, 3 + ) + print(f"Report received: {res.hex()}") + case _: + print("Incorrect option selected") + + case "4": + print(" 1. Report type 1 (INPUT) and Report id 0x01") + print(" 2. Report type 2 (OUTPUT) and Report id 0x03") + print(" 3. Report type 3 (FEATURE) and Report id 0x05") + choice1_line = await reader.readline() + choice1 = choice1_line.decode("utf-8").strip() + + match choice1: + case "1": + await hid_host.set_report( + hid.ReportType.INPUT_REPORT, + bytes( + [ + 0x01, + 0x01, + 0x01, + 0x01, + 0x01, + 0x01, + 0x01, + 0x01, + 0x01, + ] + ), + ) + case "2": + await hid_host.set_report( + hid.ReportType.OUTPUT_REPORT, + bytes([0x03, 0x01, 0x01]), + ) + case "3": + await hid_host.set_report( + hid.ReportType.FEATURE_REPORT, + bytes([0x05, 0x01, 0x01, 0x01]), + ) + case _: + print("Incorrect option selected") + + case "5": + print(" 0. Boot") + print(" 1. Report") + choice1_line = await reader.readline() + choice1 = choice1_line.decode("utf-8").strip() + + match choice1: + case "0": + await hid_host.set_protocol( + hid.ProtocolMode.BOOT_PROTOCOL + ) + case "1": + await hid_host.set_protocol( + hid.ProtocolMode.REPORT_PROTOCOL + ) + case _: + print("Incorrect option selected") + + case "6": + proto = await hid_host.get_protocol() + print(f"Protocol mode: {proto.name}") + + case "7": + print(" 1. Report ID 0x01") + print(" 2. Report ID 0x03") + choice1_line = await reader.readline() + choice1 = choice1_line.decode("utf-8").strip() + + match choice1: + case "1": + hid_host.send_interrupt_data( + bytes( + [ + 0x01, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + 0x00, + ] + ) + ) + case "2": + hid_host.send_interrupt_data( + bytes([0x03, 0x00, 0x0D, 0xFD, 0x00, 0x00]) + ) + case _: + print("Incorrect option selected") + + case "8": + hid_host.suspend() + + case "9": + hid_host.exit_suspend() + + case "10": + hid_host.virtual_cable_unplug() + if device.keystore: + try: + await device.keystore.delete(target_address) + print("Unpair successful") + except KeyError: + print( + "Device not found or Device already unpaired." + ) + + case "11": + peer_address = Address.from_string_for_transport( + target_address, transport=PhysicalTransport.BR_EDR + ) + conn = device.find_connection_by_bd_addr( + peer_address, transport=PhysicalTransport.BR_EDR + ) + if conn is not None: + await conn.disconnect() + active_connection = None + else: + print("Already disconnected from device") + + case "12": + if device.keystore: + try: + await device.keystore.delete(target_address) + print("Unpair successful") + except KeyError: + print( + "Device not found or Device already unpaired." + ) + + case "13": + active_connection = await device.connect( + target_address, transport=PhysicalTransport.BR_EDR + ) + await active_connection.authenticate() + await active_connection.encrypt() + + case "14": + sys.exit("Exit successful") + + case _: + print("Invalid option selected.") + except hid.HidProtocolError as err: + print(color(f"HID Protocol Error: {err}", "red")) + + if (len(sys.argv) > 4) and (sys.argv[4] == "test-mode"): # Enabling menu for testing await menu() else: - # HID Connection - # Control channel - await hid_host.connect_control_channel() - # Interrupt Channel - await hid_host.connect_interrupt_channel() + # Connect HID Control & Interrupt Channels + await hid_host.connect(active_connection) await hci_transport.source.terminated # ----------------------------------------------------------------------------- -bumble.logging.setup_basic_logging('DEBUG') -asyncio.run(main()) +if __name__ == "__main__": + bumble.logging.setup_basic_logging("DEBUG") + asyncio.run(main()) diff --git a/tests/hid_test.py b/tests/hid_test.py new file mode 100644 index 000000000..724afc65e --- /dev/null +++ b/tests/hid_test.py @@ -0,0 +1,476 @@ +# 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. + +# ----------------------------------------------------------------------------- +# Imports +# ----------------------------------------------------------------------------- +from __future__ import annotations + +import asyncio + +import pytest +from typing_extensions import override + +from bumble import core, hid, l2cap +from bumble.testing import test_utils + +_STEP_TIMEOUT_SECONDS = 0.5 + +# Handshake result codes 0x05..0x0D are reserved by the HID Profile 1.1 +# specification (section 3.1.2.1, Table 3.2), so a peer may legitimately +# send a value that Bumble does not name. +RESERVED_HANDSHAKE_CODES = tuple(range(0x05, 0x0E)) + + +class MockDelegate(hid.Device.Delegate): + def __init__(self) -> None: + self.get_report_response = b"\x00" + self.last_set_report_data: bytes | None = None + self.last_set_report_type: hid.ReportType | None = None + + @override + def get_report( + self, report_type: hid.ReportType, report_id: int | None = None + ) -> bytes: + if report_id is None: + raise hid.HidProtocolError( + hid.HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER + ) + return self.get_report_response + + @override + def set_report(self, report_type: hid.ReportType, data: bytes) -> None: + self.last_set_report_type = report_type + self.last_set_report_data = data + + +async def open_control_channel( + devices: test_utils.TwoDevices, +) -> tuple[l2cap.ClassicChannel, asyncio.Queue[bytes]]: + """Open an L2CAP channel to the HID control PSM and capture responses.""" + channel = await devices.connections[0].create_l2cap_channel( + spec=l2cap.ClassicChannelSpec(hid.HID_CONTROL_PSM) + ) + received: asyncio.Queue[bytes] = asyncio.Queue() + channel.sink = received.put_nowait + return channel, received + + +async def _setup_hid_pair() -> ( + tuple[test_utils.TwoDevices, MockDelegate, hid.Device, hid.Host] +): + devices = test_utils.TwoDevices() + await devices.setup_connection() + delegate = MockDelegate() + device = hid.Device(devices[0], delegate=delegate) + host = hid.Host(devices[1]) + await host.connect(devices.connections[1]) + return devices, delegate, device, host + + +@pytest.mark.asyncio +async def test_find_device_sdp_record() -> None: + devices = test_utils.TwoDevices() + await devices.setup_connection() + + expected_record = hid.DeviceSdpRecord( + service_record_handle=1, + report_map=b"123", + version_number=2, + service_name=b"456", + service_description=b"abc", + provider_name=b"def", + parser_version=3, + device_subclass=4, + country_code=5, + virtual_cable=False, + reconnect_initiate=True, + report_descriptor_type=6, + langid_base_language=7, + langid_base_bluetooth_string_offset=8, + battery_power=True, + remote_wake=False, + supervision_timeout=9, + normally_connectable=True, + boot_device=False, + ssr_host_max_latency=10, + ssr_host_min_timeout=11, + ) + devices[0].sdp_service_records = {1: expected_record.to_service_attributes()} + found_records = await asyncio.wait_for( + hid.DeviceSdpRecord.find(devices.connections[1]), + timeout=_STEP_TIMEOUT_SECONDS, + ) + + assert found_records == [expected_record] + + +@pytest.mark.asyncio +async def test_device_send_data() -> None: + _, _, device, host = await _setup_hid_pair() + event_queue: asyncio.Queue[tuple[hid.ReportType, bytes]] = asyncio.Queue() + + host.on( + host.EVENT_INTERRUPT_DATA, + lambda r_type, data: event_queue.put_nowait((r_type, data)), + ) + + device.send_interrupt_data(b"123") + + report_type, data = await asyncio.wait_for( + event_queue.get(), timeout=_STEP_TIMEOUT_SECONDS + ) + + assert report_type == hid.ReportType.INPUT_REPORT + assert data == b"123" + + +@pytest.mark.asyncio +async def test_device_virtual_cable_unplug() -> None: + _, _, device, host = await _setup_hid_pair() + event_queue: asyncio.Queue[None] = asyncio.Queue() + + host.on( + host.EVENT_VIRTUAL_CABLE_UNPLUG, + lambda: event_queue.put_nowait(None), + ) + device.virtual_cable_unplug() + await asyncio.wait_for(event_queue.get(), timeout=_STEP_TIMEOUT_SECONDS) + + +@pytest.mark.asyncio +async def test_device_send_handshake_message() -> None: + _, _, device, host = await _setup_hid_pair() + event_queue: asyncio.Queue[hid.HandshakeMessage.ResultCode] = asyncio.Queue() + + host.on(host.EVENT_HANDSHAKE, event_queue.put_nowait) + + # pylint: disable=protected-access + device._send_handshake_message(hid.HandshakeMessage.ResultCode.SUCCESSFUL) + + result = await asyncio.wait_for(event_queue.get(), timeout=_STEP_TIMEOUT_SECONDS) + assert result == hid.HandshakeMessage.ResultCode.SUCCESSFUL + + +@pytest.mark.asyncio +async def test_host_get_report() -> None: + _, delegate, _, host = await _setup_hid_pair() + expected_data = b"\x01\x02\x03" + delegate.get_report_response = expected_data + + data = await asyncio.wait_for( + host.get_report(report_type=hid.ReportType.INPUT_REPORT, report_id=1), + timeout=_STEP_TIMEOUT_SECONDS, + ) + + assert data == b"\x01" + expected_data + + +@pytest.mark.asyncio +async def test_host_get_report_buffer_size_truncation() -> None: + _, delegate, _, host = await _setup_hid_pair() + delegate.get_report_response = b"\x01\x02\x03" + + # buffer_size=3 limits total DATA payload (1-byte Report ID + 2 bytes of data) + data = await asyncio.wait_for( + host.get_report( + report_type=hid.ReportType.INPUT_REPORT, report_id=1, buffer_size=3 + ), + timeout=_STEP_TIMEOUT_SECONDS, + ) + + assert data == b"\x01\x01\x02" + + +@pytest.mark.asyncio +async def test_host_set_report() -> None: + _, delegate, _, host = await _setup_hid_pair() + data_to_send = b"\xab\xcd" + + await asyncio.wait_for( + host.set_report(report_type=hid.ReportType.OUTPUT_REPORT, data=data_to_send), + timeout=_STEP_TIMEOUT_SECONDS, + ) + + assert delegate.last_set_report_data == data_to_send + assert delegate.last_set_report_type == hid.ReportType.OUTPUT_REPORT + + +@pytest.mark.asyncio +async def test_host_get_protocol() -> None: + _, _, device, host = await _setup_hid_pair() + device.protocol = hid.ProtocolMode.REPORT_PROTOCOL + + protocol = await asyncio.wait_for( + host.get_protocol(), timeout=_STEP_TIMEOUT_SECONDS + ) + + assert protocol == hid.ProtocolMode.REPORT_PROTOCOL + + +@pytest.mark.asyncio +async def test_host_set_protocol() -> None: + _, _, device, host = await _setup_hid_pair() + event_queue: asyncio.Queue[None] = asyncio.Queue() + device.on(device.EVENT_PROTOCOL_CHANGE, lambda: event_queue.put_nowait(None)) + + await asyncio.wait_for( + host.set_protocol(protocol_mode=hid.ProtocolMode.BOOT_PROTOCOL), + timeout=_STEP_TIMEOUT_SECONDS, + ) + + await asyncio.wait_for(event_queue.get(), timeout=_STEP_TIMEOUT_SECONDS) + assert device.protocol == hid.ProtocolMode.BOOT_PROTOCOL + + +@pytest.mark.asyncio +async def test_host_get_and_set_idle() -> None: + _, _, device, host = await _setup_hid_pair() + event_queue: asyncio.Queue[None] = asyncio.Queue() + device.on(device.EVENT_IDLE_TIME_CHANGE, lambda: event_queue.put_nowait(None)) + + await asyncio.wait_for( + host.set_idle(idle_time=42), + timeout=_STEP_TIMEOUT_SECONDS, + ) + await asyncio.wait_for(event_queue.get(), timeout=_STEP_TIMEOUT_SECONDS) + assert device._idle_time == 42 # pylint: disable=protected-access + + idle_time = await asyncio.wait_for( + host.get_idle(), + timeout=_STEP_TIMEOUT_SECONDS, + ) + assert idle_time == 42 + + +@pytest.mark.asyncio +async def test_host_suspend() -> None: + _, _, device, host = await _setup_hid_pair() + event_queue: asyncio.Queue[None] = asyncio.Queue() + + device.on(device.EVENT_SUSPEND, lambda: event_queue.put_nowait(None)) + + host.suspend() + await asyncio.wait_for(event_queue.get(), timeout=_STEP_TIMEOUT_SECONDS) + + +@pytest.mark.asyncio +async def test_host_exit_suspend() -> None: + _, _, device, host = await _setup_hid_pair() + event_queue: asyncio.Queue[None] = asyncio.Queue() + + device.on( + device.EVENT_EXIT_SUSPEND, + lambda: event_queue.put_nowait(None), + ) + + host.exit_suspend() + await asyncio.wait_for(event_queue.get(), timeout=_STEP_TIMEOUT_SECONDS) + + +@pytest.mark.asyncio +async def test_host_send_data() -> None: + _, _, device, host = await _setup_hid_pair() + event_queue: asyncio.Queue[tuple[hid.ReportType, bytes]] = asyncio.Queue() + + device.on( + device.EVENT_INTERRUPT_DATA, + lambda r_type, data: event_queue.put_nowait((r_type, data)), + ) + + host.send_interrupt_data(b"123") + + report_type, data = await asyncio.wait_for( + event_queue.get(), timeout=_STEP_TIMEOUT_SECONDS + ) + + assert report_type == hid.ReportType.OUTPUT_REPORT + assert data == b"123" + + +@pytest.mark.asyncio +async def test_host_virtual_cable_unplug() -> None: + _, _, device, host = await _setup_hid_pair() + event_queue: asyncio.Queue[None] = asyncio.Queue() + + device.on( + device.EVENT_VIRTUAL_CABLE_UNPLUG, + lambda: event_queue.put_nowait(None), + ) + host.virtual_cable_unplug() + await asyncio.wait_for(event_queue.get(), timeout=_STEP_TIMEOUT_SECONDS) + + +@pytest.mark.parametrize("result_code", RESERVED_HANDSHAKE_CODES) +def test_reserved_handshake_result_code_does_not_raise(result_code: int) -> None: + """A reserved HANDSHAKE result code must be parsed without raising ValueError.""" + assert hid.HandshakeMessage.ResultCode(result_code) == result_code + + +@pytest.mark.asyncio +async def test_host_handles_reserved_handshake_result_code() -> None: + devices = test_utils.TwoDevices() + await devices.setup_connection() + hid_host = hid.Host(devices[1]) + handshakes: list[int] = [] + hid_host.on(hid_host.EVENT_HANDSHAKE, handshakes.append) + + channel, _ = await open_control_channel(devices) + for result_code in RESERVED_HANDSHAKE_CODES: + channel.write(bytes([result_code])) + await test_utils.async_barrier() + + assert handshakes == list(RESERVED_HANDSHAKE_CODES) + + +@pytest.mark.asyncio +async def test_host_handles_defined_handshake_result_code() -> None: + devices = test_utils.TwoDevices() + await devices.setup_connection() + hid_host = hid.Host(devices[1]) + handshakes: list[int] = [] + hid_host.on(hid_host.EVENT_HANDSHAKE, handshakes.append) + + channel, _ = await open_control_channel(devices) + channel.write(bytes([hid.HandshakeMessage.ResultCode.ERR_FATAL])) + await test_utils.async_barrier() + + assert handshakes == [hid.HandshakeMessage.ResultCode.ERR_FATAL] + + +@pytest.mark.parametrize("role", ("host", "device")) +@pytest.mark.asyncio +async def test_empty_control_pdu_is_ignored(role: str) -> None: + """An empty control PDU must be dropped rather than raise IndexError.""" + devices = test_utils.TwoDevices() + await devices.setup_connection() + if role == "host": + hid.Host(devices[1]) + else: + hid.Device(devices[1], delegate=MockDelegate()) + + loop = asyncio.get_running_loop() + escaped: list[BaseException] = [] + previous_handler = loop.get_exception_handler() + loop.set_exception_handler( + lambda _loop, context: escaped.append(context["exception"]) + ) + try: + channel, received = await open_control_channel(devices) + channel.write(b"") + await test_utils.async_barrier() + finally: + loop.set_exception_handler(previous_handler) + + assert not escaped + assert received.empty() + + +@pytest.mark.parametrize( + "pdu", + ( + bytes([0x48]), # GET_REPORT, buffer flag set, no buffer size + bytes([0x48, 0x01]), # GET_REPORT, buffer flag set, 1-byte buffer size + bytes([0x40, 0x01, 0x02]), # GET_REPORT, buffer flag clear, extra trailing byte + bytes([0x50]), # SET_REPORT, no report ID / payload + bytes([0x90]), # SET_IDLE, no idle time byte + ), + ids=( + "get_report_no_size", + "get_report_short_size", + "get_report_extra_bytes", + "set_report_empty", + "set_idle_empty", + ), +) +@pytest.mark.asyncio +async def test_truncated_report_pdu_is_rejected(pdu: bytes) -> None: + """A malformed/truncated request must return ERR_INVALID_PARAMETER.""" + devices = test_utils.TwoDevices() + await devices.setup_connection() + hid.Device(devices[1], delegate=MockDelegate()) + + channel, received = await open_control_channel(devices) + channel.write(pdu) + await test_utils.async_barrier() + + assert received.get_nowait() == bytes( + [hid.HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER] + ) + + +@pytest.mark.parametrize( + "pdu", + ( + bytes([0x40]), # GET_REPORT without report ID + bytes( + [0x48, 0x01, 0x20] + ), # GET_REPORT with buffer_size=0x2001, without report ID + ), + ids=("get_report_no_id", "get_report_with_size_no_id"), +) +@pytest.mark.asyncio +async def test_get_report_without_report_id_rejected_by_delegate( + pdu: bytes, +) -> None: + """Report-ID-less GET_REPORT rejected by delegate returns ERR_INVALID_PARAMETER.""" + devices = test_utils.TwoDevices() + await devices.setup_connection() + hid.Device(devices[1], delegate=MockDelegate()) + + channel, received = await open_control_channel(devices) + channel.write(pdu) + await test_utils.async_barrier() + + assert received.get_nowait() == bytes( + [hid.HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER] + ) + + +@pytest.mark.asyncio +async def test_host_command_error_handshake_raises() -> None: + _, _, _, host = await _setup_hid_pair() + + with pytest.raises(hid.HidProtocolError) as exc_info: + await asyncio.wait_for( + host.get_report( + report_type=hid.ReportType.INPUT_REPORT, + report_id=None, # Rejected by MockDelegate -> ERR_INVALID_PARAMETER + ), + timeout=_STEP_TIMEOUT_SECONDS, + ) + + assert ( + exc_info.value.result_code + == hid.HandshakeMessage.ResultCode.ERR_INVALID_PARAMETER + ) + + +@pytest.mark.asyncio +async def test_host_command_aborted_on_disconnect() -> None: + _, _, device, host = await _setup_hid_pair() + + # Suppress device response so the command remains in flight when disconnect happens + device._handle_get_report = ( # type: ignore[method-assign,assignment] # pylint: disable=protected-access + lambda message: None + ) + + task = asyncio.create_task( + host.get_report(report_type=hid.ReportType.INPUT_REPORT, report_id=1) + ) + await asyncio.sleep(0.05) + await host.disconnect() + + with pytest.raises(core.InvalidStateError): + await asyncio.wait_for(task, timeout=_STEP_TIMEOUT_SECONDS)