Review changes comment fix. Classes/Subclass/dataclass. Enum constants.

Naming conventions
This commit is contained in:
skarnataki
2023-10-05 10:55:28 +00:00
committed by Lucas Abel
parent 5ddee17411
commit fc1bf36ace
4 changed files with 76 additions and 126 deletions

View File

@@ -40,123 +40,104 @@ logger = logging.getLogger(__name__)
# Constants # Constants
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
# fmt: off # fmt: off
class Message():
class HIDPsm(enum.IntEnum):
HID_CONTROL_PSM = 0x0011 HID_CONTROL_PSM = 0x0011
HID_INTERRUPT_PSM = 0x0013 HID_INTERRUPT_PSM = 0x0013
class Message():
# Report types # Report types
class ReportType(enum.IntEnum): class ReportType(enum.IntEnum):
HID_OTHER_REPORT = 0x00 OTHER_REPORT = 0x00
HID_INPUT_REPORT = 0x01 INPUT_REPORT = 0x01
HID_OUTPUT_REPORT = 0x02 OUTPUT_REPORT = 0x02
HID_FEATURE_REPORT = 0x03 FEATURE_REPORT = 0x03
# Handshake parameters # Handshake parameters
class HandshakeState(enum.IntEnum): class Handshake(enum.IntEnum):
HANDSHAKE_SUCCESSFUL = 0x00 SUCCESSFUL = 0x00
HANDSHAKE_NOT_READY = 0x01 NOT_READY = 0x01
HANDSHAKE_ERR_INVALID_REPORT_ID = 0x02 ERR_INVALID_REPORT_ID = 0x02
HANDSHAKE_ERR_UNSUPPORTED_REQUEST = 0x03 ERR_UNSUPPORTED_REQUEST = 0x03
HANDSHAKE_ERR_UNKNOWN = 0x0E ERR_UNKNOWN = 0x0E
HANDSHAKE_ERR_FATAL = 0x0F ERR_FATAL = 0x0F
class Type(enum.IntEnum): #Message Type
HID_HANDSHAKE = 0x00 class MessageType(enum.IntEnum):
HID_CONTROL = 0x01 HANDSHAKE = 0x00
HID_GET_REPORT = 0x04 CONTROL = 0x01
HID_SET_REPORT = 0x05 GET_REPORT = 0x04
HID_GET_PROTOCOL = 0x06 SET_REPORT = 0x05
HID_SET_PROTOCOL = 0x07 GET_PROTOCOL = 0x06
HID_DATA = 0x0A SET_PROTOCOL = 0x07
DATA = 0x0A
# Protocol modes # Protocol modes
class ProtocolMode(enum.IntEnum): class ProtocolMode(enum.IntEnum):
HID_BOOT_PROTOCOL_MODE = 0x00 BOOT_PROTOCOL = 0x00
HID_REPORT_PROTOCOL_MODE = 0x01 REPORT_PROTOCOL = 0x01
# Control Operations # Control Operations
class ControlCommand(enum.IntEnum): class ControlCommand(enum.IntEnum):
HID_SUSPEND = 0x03 SUSPEND = 0x03
HID_EXIT_SUSPEND = 0x04 EXIT_SUSPEND = 0x04
HID_VIRTUAL_CABLE_UNPLUG = 0x05 VIRTUAL_CABLE_UNPLUG = 0x05
# HIDP message types # HIDP messages
@dataclass @dataclass
class GetReportMessage(Message): class GetReportMessage(Message):
report_type : int report_type : int
report_id : int report_id : int
buffer_size : int buffer_size : int
'''
def __init__(self,
report_type: Optional[int] = None,
report_id: Optional[int] = None,
buffer_size: Optional[int] = None,
):
self.report_type = report_type
self.report_id = report_id
self.buffer_size = buffer_size
'''
def __bytes__(self) -> bytes: def __bytes__(self) -> bytes:
if(self.report_type == Message.ReportType.HID_OTHER_REPORT): if(self.report_type == Message.ReportType.OTHER_REPORT):
param = self.report_type param = self.report_type
else: else:
param = 0x08 | self.report_type param = 0x08 | self.report_type
header = ((Message.Type.HID_GET_REPORT << 4) | param) header = ((Message.MessageType.GET_REPORT << 4) | param)
packet_bytes = bytearray() packet_bytes = bytearray()
packet_bytes.append(header) packet_bytes.append(header)
packet_bytes.append(self.report_id) packet_bytes.append(self.report_id)
packet_bytes.extend([(self.buffer_size & 0xff), ((self.buffer_size >> 8) & 0xff)]) packet_bytes.extend([(self.buffer_size & 0xff), ((self.buffer_size >> 8) & 0xff)])
return bytes(packet_bytes) return bytes(packet_bytes)
@dataclass
class SetReportMessage(Message): class SetReportMessage(Message):
report_type: int
def __init__(self, data : bytes
report_type: int,
data : bytes):
self.report_type = report_type
self.data = data
def __bytes__(self) -> bytes: def __bytes__(self) -> bytes:
header = ((Message.Type.HID_SET_REPORT << 4) | self.report_type) header = ((Message.MessageType.SET_REPORT << 4) | self.report_type)
packet_bytes = bytearray() packet_bytes = bytearray()
packet_bytes.append(header) packet_bytes.append(header)
packet_bytes.extend(self.data) packet_bytes.extend(self.data)
return bytes(packet_bytes) return bytes(packet_bytes)
@dataclass
class GetProtocolMessage(Message): class GetProtocolMessage(Message):
def __bytes__(self) -> bytes: def __bytes__(self) -> bytes:
header = (Message.Type.HID_GET_PROTOCOL << 4) header = (Message.MessageType.GET_PROTOCOL << 4)
packet_bytes = bytearray() packet_bytes = bytearray()
packet_bytes.append(header) packet_bytes.append(header)
return bytes(packet_bytes) return bytes(packet_bytes)
@dataclass
class SetProtocolMessage(Message): class SetProtocolMessage(Message):
protocol_mode: int
def __init__(self, protocol_mode: int):
self.protocol_mode = protocol_mode
def __bytes__(self) -> bytes: def __bytes__(self) -> bytes:
header = (Message.Type.HID_SET_PROTOCOL << 4 | self.protocol_mode) header = (Message.MessageType.SET_PROTOCOL << 4 | self.protocol_mode)
packet_bytes = bytearray() packet_bytes = bytearray()
packet_bytes.append(header) packet_bytes.append(header)
packet_bytes.append(self.protocol_mode) packet_bytes.append(self.protocol_mode)
return bytes(packet_bytes) return bytes(packet_bytes)
@dataclass
class SendData(Message): class SendData(Message):
def __init__(self, data : bytes):
self.data = data data : bytes
def __bytes__(self) -> bytes: def __bytes__(self) -> bytes:
header = ((Message.Type.HID_DATA << 4) | Message.ReportType.HID_OUTPUT_REPORT) header = ((Message.MessageType.DATA << 4) | Message.ReportType.OUTPUT_REPORT)
packet_bytes = bytearray() packet_bytes = bytearray()
packet_bytes.append(header) packet_bytes.append(header)
packet_bytes.extend(self.data) packet_bytes.extend(self.data)
@@ -173,14 +154,14 @@ class Host(EventEmitter):
self.l2cap_intr_channel = None self.l2cap_intr_channel = None
# Register ourselves with the L2CAP channel manager # Register ourselves with the L2CAP channel manager
device.register_l2cap_server(Message.HIDPsm.HID_CONTROL_PSM, self.on_connection) device.register_l2cap_server(HID_CONTROL_PSM, self.on_connection)
device.register_l2cap_server(Message.HIDPsm.HID_INTERRUPT_PSM, self.on_connection) device.register_l2cap_server(HID_INTERRUPT_PSM, self.on_connection)
async def connect_control_channel(self) -> None: async def connect_control_channel(self) -> None:
# Create a new L2CAP connection - control channel # Create a new L2CAP connection - control channel
try: try:
self.l2cap_ctrl_channel = await self.device.l2cap_channel_manager.connect( self.l2cap_ctrl_channel = await self.device.l2cap_channel_manager.connect(
self.connection, Message.HIDPsm.HID_CONTROL_PSM self.connection, HID_CONTROL_PSM
) )
except ProtocolError as error: except ProtocolError as error:
logging.exception(f'L2CAP connection failed: {error}') logging.exception(f'L2CAP connection failed: {error}')
@@ -194,7 +175,7 @@ class Host(EventEmitter):
# Create a new L2CAP connection - interrupt channel # Create a new L2CAP connection - interrupt channel
try: try:
self.l2cap_intr_channel = await self.device.l2cap_channel_manager.connect( self.l2cap_intr_channel = await self.device.l2cap_channel_manager.connect(
self.connection, Message.HIDPsm.HID_INTERRUPT_PSM self.connection, HID_INTERRUPT_PSM
) )
except ProtocolError as error: except ProtocolError as error:
logging.exception(f'L2CAP connection failed: {error}') logging.exception(f'L2CAP connection failed: {error}')
@@ -207,7 +188,6 @@ class Host(EventEmitter):
async def disconnect_interrupt_channel(self) -> None: async def disconnect_interrupt_channel(self) -> None:
if self.l2cap_intr_channel is None: if self.l2cap_intr_channel is None:
raise InvalidStateError('invalid state') raise InvalidStateError('invalid state')
await self.l2cap_intr_channel.disconnect() # type: ignore
channel = self.l2cap_intr_channel channel = self.l2cap_intr_channel
self.l2cap_intr_channel = None self.l2cap_intr_channel = None
await channel.disconnect() # type: ignore await channel.disconnect() # type: ignore
@@ -215,7 +195,6 @@ class Host(EventEmitter):
async def disconnect_control_channel(self) -> None: async def disconnect_control_channel(self) -> None:
if self.l2cap_ctrl_channel is None: if self.l2cap_ctrl_channel is None:
raise InvalidStateError('invalid state') raise InvalidStateError('invalid state')
await self.l2cap_ctrl_channel.disconnect() # type: ignore
channel = self.l2cap_ctrl_channel channel = self.l2cap_ctrl_channel
self.l2cap_ctrl_channel = None self.l2cap_ctrl_channel = None
await channel.disconnect() # type: ignore await channel.disconnect() # type: ignore
@@ -225,7 +204,7 @@ class Host(EventEmitter):
l2cap_channel.on('open', lambda: self.on_l2cap_channel_open(l2cap_channel)) l2cap_channel.on('open', lambda: self.on_l2cap_channel_open(l2cap_channel))
def on_l2cap_channel_open(self, l2cap_channel: l2cap.Channel) -> None: def on_l2cap_channel_open(self, l2cap_channel: l2cap.Channel) -> None:
if l2cap_channel.psm == Message.HIDPsm.HID_CONTROL_PSM: if l2cap_channel.psm == HID_CONTROL_PSM:
self.l2cap_ctrl_channel = l2cap_channel self.l2cap_ctrl_channel = l2cap_channel
self.l2cap_ctrl_channel.sink = self.on_ctrl_pdu self.l2cap_ctrl_channel.sink = self.on_ctrl_pdu
else: else:
@@ -239,27 +218,20 @@ class Host(EventEmitter):
message_type = pdu[0] >> 4 message_type = pdu[0] >> 4
param = pdu[0] & 0x0f param = pdu[0] & 0x0f
for command in Message.ControlCommand.__members__items(): if message_type == Message.MessageType.HANDSHAKE :
if param == command: logger.debug(f'<<< HID HANDSHAKE: {Message.Handshake(param).name}')
logger.debug(f'<<< ', command + pdu) self.emit('handshake', Message.Handshake(param))
self.handle_handshake(param) elif message_type == Message.MessageType.DATA :
self.emit(command, pdu)
'''
if message_type == Message.Type.HID_HANDSHAKE :
logger.debug('<<< HID HANDSHAKE')
self.handle_handshake(param)
self.emit('handshake', pdu)
elif message_type == Message.Type.HID_DATA :
logger.debug('<<< HID CONTROL DATA') logger.debug('<<< HID CONTROL DATA')
self.emit('data', pdu) self.emit('data', pdu)
elif message_type == Message.Type.HID_CONTROL : elif message_type == Message.MessageType.CONTROL :
if param == Message.ControlCommand.HID_SUSPEND : if param == Message.ControlCommand.SUSPEND :
logger.debug('<<< HID SUSPEND') logger.debug('<<< HID SUSPEND')
self.emit('suspend', pdu) self.emit('suspend', pdu)
elif param == HID_EXIT_SUSPEND : elif param == Message.ControlCommand.EXIT_SUSPEND :
logger.debug('<<< HID EXIT SUSPEND') logger.debug('<<< HID EXIT SUSPEND')
self.emit('exit_suspend', pdu) self.emit('exit_suspend', pdu)
elif param == HID_VIRTUAL_CABLE_UNPLUG : elif param == Message.ControlCommand.VIRTUAL_CABLE_UNPLUG :
logger.debug('<<< HID VIRTUAL CABLE UNPLUG') logger.debug('<<< HID VIRTUAL CABLE UNPLUG')
self.emit('virtual_cable_unplug') self.emit('virtual_cable_unplug')
else: else:
@@ -267,7 +239,7 @@ class Host(EventEmitter):
else: else:
logger.debug('<<< HID CONTROL DATA') logger.debug('<<< HID CONTROL DATA')
self.emit('data', pdu) self.emit('data', pdu)
'''
def on_intr_pdu(self, pdu: bytes) -> None: def on_intr_pdu(self, pdu: bytes) -> None:
logger.debug(f'<<< HID INTERRUPT PDU: {pdu.hex()}') logger.debug(f'<<< HID INTERRUPT PDU: {pdu.hex()}')
@@ -304,46 +276,26 @@ class Host(EventEmitter):
self.l2cap_intr_channel.send_pdu(msg) # type: ignore self.l2cap_intr_channel.send_pdu(msg) # type: ignore
def send_data(self, data): def send_data(self, data):
msg = Message(data= data) msg = SendData(data)
hid_message = msg.__bytes__() hid_message = msg.__bytes__()
logger.debug(f'>>> HID INTERRUPT SEND DATA, PDU: {hid_message.hex()}') logger.debug(f'>>> HID INTERRUPT SEND DATA, PDU: {hid_message.hex()}')
self.send_pdu_on_intr(hid_message) # type: ignore self.send_pdu_on_intr(hid_message) # type: ignore
def suspend(self): def suspend(self):
header = (Message.Type.HID_CONTROL << 4 | Message.ControlCommand.HID_SUSPEND) header = (Message.MessageType.CONTROL << 4 | Message.ControlCommand.SUSPEND)
msg = bytearray([header]) msg = bytearray([header])
logger.debug(f'>>> HID CONTROL SUSPEND, PDU:{msg.hex()}') logger.debug(f'>>> HID CONTROL SUSPEND, PDU:{msg.hex()}')
self.l2cap_ctrl_channel.send_pdu(msg) # type: ignore self.l2cap_ctrl_channel.send_pdu(msg) # type: ignore
def exit_suspend(self): def exit_suspend(self):
header = (Message.Type.HID_CONTROL << 4 | Message.ControlCommand.HID_EXIT_SUSPEND) header = (Message.MessageType.CONTROL << 4 | Message.ControlCommand.EXIT_SUSPEND)
msg = bytearray([header]) msg = bytearray([header])
logger.debug(f'>>> HID CONTROL EXIT SUSPEND, PDU:{msg.hex()}') logger.debug(f'>>> HID CONTROL EXIT SUSPEND, PDU:{msg.hex()}')
self.l2cap_ctrl_channel.send_pdu(msg) # type: ignore self.l2cap_ctrl_channel.send_pdu(msg) # type: ignore
def virtual_cable_unplug(self): def virtual_cable_unplug(self):
header = (Message.Type.HID_CONTROL << 4 | Message.ControlCommand.HID_VIRTUAL_CABLE_UNPLUG) header = (Message.MessageType.CONTROL << 4 | Message.ControlCommand.VIRTUAL_CABLE_UNPLUG)
msg = bytearray([header]) msg = bytearray([header])
logger.debug(f'>>> HID CONTROL VIRTUAL CABLE UNPLUG, PDU: {msg.hex()}') logger.debug(f'>>> HID CONTROL VIRTUAL CABLE UNPLUG, PDU: {msg.hex()}')
self.l2cap_ctrl_channel.send_pdu(msg) # type: ignore self.l2cap_ctrl_channel.send_pdu(msg) # type: ignore
def handle_handshake(self, param: Message.HandshakeState):
for state in Message.HandshakeState.__members__items():
if param == state:
logger.debug(f'<<< HID HANDSHAKE: ', state)
'''
if param == HANDSHAKE_SUCCESSFUL :
logger.debug(f'<<< HID HANDSHAKE: SUCCESSFUL')
elif param == HANDSHAKE_NOT_READY :
logger.warning(f'<<< HID HANDSHAKE: NOT_READY')
elif param == HANDSHAKE_ERR_INVALID_REPORT_ID :
logger.warning(f'<<< HID HANDSHAKE: ERR_INVALID_REPORT_ID')
elif param == HANDSHAKE_ERR_UNSUPPORTED_REQUEST :
logger.warning(f'<<< HID HANDSHAKE: ERR_UNSUPPORTED_REQUEST')
elif param == HANDSHAKE_ERR_UNKNOWN :
logger.warning(f'<<< HID HANDSHAKE: ERR_UNKNOWN')
elif param == HANDSHAKE_ERR_FATAL :
logger.warning(f'<<< HID HANDSHAKE: ERR_FATAL')
else: # 0x5-0xD = Reserved
logger.warning("<<< HID HANDSHAKE: RESERVED VALUE")
'''

View File

@@ -167,7 +167,7 @@ class DataElement:
UUID: lambda x: DataElement( UUID: lambda x: DataElement(
DataElement.UUID, core.UUID.from_bytes(bytes(reversed(x))) DataElement.UUID, core.UUID.from_bytes(bytes(reversed(x)))
), ),
TEXT_STRING: lambda x: DataElement(DataElement.TEXT_STRING, x.decode('latin1')), TEXT_STRING: lambda x: DataElement(DataElement.TEXT_STRING, x),
BOOLEAN: lambda x: DataElement(DataElement.BOOLEAN, x[0] == 1), BOOLEAN: lambda x: DataElement(DataElement.BOOLEAN, x[0] == 1),
SEQUENCE: lambda x: DataElement( SEQUENCE: lambda x: DataElement(
DataElement.SEQUENCE, DataElement.list_from_bytes(x) DataElement.SEQUENCE, DataElement.list_from_bytes(x)
@@ -376,8 +376,6 @@ class DataElement:
raise ValueError('invalid value_size') raise ValueError('invalid value_size')
elif self.type == DataElement.UUID: elif self.type == DataElement.UUID:
data = bytes(reversed(bytes(self.value))) data = bytes(reversed(bytes(self.value)))
elif self.type == DataElement.TEXT_STRING:
data = self.value.encode('latin1')
elif self.type == DataElement.URL: elif self.type == DataElement.URL:
data = self.value.encode('utf8') data = self.value.encode('utf8')
elif self.type == DataElement.BOOLEAN: elif self.type == DataElement.BOOLEAN:

View File

@@ -169,7 +169,7 @@ async def get_hid_device_sdp_record(device, connection):
elif attribute.id == SDP_HID_DESCRIPTOR_LIST_ATTRIBUTE_ID : 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 Descriptor type: ', 'cyan'), hex(attribute.value.value[0].value[0].value))
print(color(' HID Report DescriptorList: ', 'cyan'), (attribute.value.value[0].value[1].value).encode('latin-1')) print(color(' HID Report DescriptorList: ', 'cyan'), attribute.value.value[0].value[1].value)
HID_Descriptor_Type = attribute.value.value[0].value[0].value HID_Descriptor_Type = attribute.value.value[0].value[0].value
HID_Report_Descriptor_List = attribute.value.value[0].value[1].value HID_Report_Descriptor_List = attribute.value.value[0].value[1].value
@@ -243,13 +243,13 @@ async def main():
report_length = len(pdu[1:]) report_length = len(pdu[1:])
report_id = pdu[1] report_id = pdu[1]
if (report_type != Message.ReportType.HID_OTHER_REPORT): 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')) print(color(f' Report type = {report_type}, Report length = {report_length}, Report id = {report_id}', 'blue', None, 'bold'))
if ((report_length <= 1) or (report_id == 0)): if ((report_length <= 1) or (report_id == 0)):
return return
if report_type == Message.ReportType.HID_INPUT_REPORT: if report_type == Message.ReportType.INPUT_REPORT:
ReportParser.parse_input_report(pdu[1:]) #type: ignore ReportParser.parse_input_report(pdu[1:]) #type: ignore
async def handle_virtual_cable_unplug(): async def handle_virtual_cable_unplug():
@@ -383,10 +383,10 @@ async def main():
choice1 = choice1.decode('utf-8').strip() choice1 = choice1.decode('utf-8').strip()
if choice1 == '0': if choice1 == '0':
hid_host.set_protocol(Message.ProtocolMode.HID_BOOT_PROTOCOL_MODE) hid_host.set_protocol(Message.ProtocolMode.BOOT_PROTOCOL)
elif choice1 == '1': elif choice1 == '1':
hid_host.set_protocol(Message.ProtocolMode.HID_REPORT_PROTOCOL_MODE) hid_host.set_protocol(Message.ProtocolMode.REPORT_PROTOCOL)
else: else:
print('Incorrect option selected') print('Incorrect option selected')

View File

@@ -100,13 +100,13 @@ def test_data_elements() -> None:
e = DataElement(DataElement.UUID, UUID('61A3512C-09BE-4DDC-A6A6-0B03667AAFC6')) e = DataElement(DataElement.UUID, UUID('61A3512C-09BE-4DDC-A6A6-0B03667AAFC6'))
basic_check(e) basic_check(e)
e = DataElement(DataElement.TEXT_STRING, 'hello') e = DataElement(DataElement.TEXT_STRING, b'hello')
basic_check(e) basic_check(e)
e = DataElement(DataElement.TEXT_STRING, 'hello' * 60) e = DataElement(DataElement.TEXT_STRING, b'hello' * 60)
basic_check(e) basic_check(e)
e = DataElement(DataElement.TEXT_STRING, 'hello' * 20000) e = DataElement(DataElement.TEXT_STRING, b'hello' * 20000)
basic_check(e) basic_check(e)
e = DataElement(DataElement.BOOLEAN, True) e = DataElement(DataElement.BOOLEAN, True)
@@ -122,7 +122,7 @@ def test_data_elements() -> None:
DataElement.SEQUENCE, DataElement.SEQUENCE,
[ [
DataElement(DataElement.BOOLEAN, True), DataElement(DataElement.BOOLEAN, True),
DataElement(DataElement.TEXT_STRING, 'hello'), DataElement(DataElement.TEXT_STRING, b'hello'),
], ],
) )
basic_check(e) basic_check(e)
@@ -134,7 +134,7 @@ def test_data_elements() -> None:
DataElement.ALTERNATIVE, DataElement.ALTERNATIVE,
[ [
DataElement(DataElement.BOOLEAN, True), DataElement(DataElement.BOOLEAN, True),
DataElement(DataElement.TEXT_STRING, 'hello'), DataElement(DataElement.TEXT_STRING, b'hello'),
], ],
) )
basic_check(e) basic_check(e)
@@ -152,19 +152,19 @@ def test_data_elements() -> None:
e = DataElement.uuid(UUID.from_16_bits(1234)) e = DataElement.uuid(UUID.from_16_bits(1234))
basic_check(e) basic_check(e)
e = DataElement.text_string('hello') e = DataElement.text_string(b'hello')
basic_check(e) basic_check(e)
e = DataElement.boolean(True) e = DataElement.boolean(True)
basic_check(e) basic_check(e)
e = DataElement.sequence( e = DataElement.sequence(
[DataElement.signed_integer(0, 1), DataElement.text_string('hello')] [DataElement.signed_integer(0, 1), DataElement.text_string(b'hello')]
) )
basic_check(e) basic_check(e)
e = DataElement.alternative( e = DataElement.alternative(
[DataElement.signed_integer(0, 1), DataElement.text_string('hello')] [DataElement.signed_integer(0, 1), DataElement.text_string(b'hello')]
) )
basic_check(e) basic_check(e)