fix Windows implementation

The handling of the `SP_DEVICE_INTERFACE_DETAIL_DATA_A` argument
to `SetupDiGetDeviceInterfaceDetail` is wrong: the hard-coded size
only works for 32bits.

Fix call to `CloseHandle` with an invalid (0) handle: use
`INVALID_HANDLE_VALUE` for `StenographMachine._usb_device`
initial value.
This commit is contained in:
Benoit Pierre 2021-05-26 21:42:26 +02:00
commit d3a13abba6

View file

@ -234,135 +234,245 @@ if sys.platform.startswith('win32'):
import ctypes import ctypes
import uuid import uuid
# Class GUID for Stenograph USB Writer GUID = wintypes.BYTE * 16
USB_WRITER_GUID = uuid.UUID('{c5682e20-8059-604a-b761-77c4de9d5dbf}') HDEVINFO = wintypes.HANDLE
class DeviceInterfaceData(ctypes.Structure): # Stubs.
LPOVERLAPPED = wintypes.LPVOID
LPSECURITY_ATTRIBUTES = wintypes.LPVOID
PSP_DEVINFO_DATA = wintypes.LPVOID
# Class GUID for Stenograph USB Writer.
USB_WRITER_GUID = GUID(*uuid.UUID('{c5682e20-8059-604a-b761-77c4de9d5dbf}').bytes)
class SP_DEVICE_INTERFACE_DATA(ctypes.Structure):
_fields_ = [ _fields_ = [
('cbSize', wintypes.DWORD), ('cbSize', wintypes.DWORD),
('InterfaceClassGuid', wintypes.BYTE * 16), ('InterfaceClassGuid', GUID),
('Flags', wintypes.DWORD), ('Flags', wintypes.DWORD),
('Reserved', ctypes.POINTER(ctypes.c_ulonglong)) ('Reserved', wintypes.PULONG),
] ]
PSP_DEVICE_INTERFACE_DATA = ctypes.POINTER(SP_DEVICE_INTERFACE_DATA)
class SP_DEVICE_INTERFACE_DETAIL_DATA_A(ctypes.Structure):
_fields_ = [
('cbSize', wintypes.DWORD),
('_DevicePath', wintypes.CHAR * 1),
]
@property
def DevicePath(self):
return ctypes.string_at(ctypes.byref(self, ctypes.sizeof(wintypes.DWORD)))
PSP_DEVICE_INTERFACE_DETAIL_DATA_A = ctypes.POINTER(SP_DEVICE_INTERFACE_DETAIL_DATA_A)
SetupDiGetClassDevs = windll.setupapi.SetupDiGetClassDevsA SetupDiGetClassDevs = windll.setupapi.SetupDiGetClassDevsA
SetupDiGetClassDevs.argtypes = [
ctypes.POINTER(GUID), # ClassGuid
wintypes.LPCWSTR, # Enumerator
wintypes.HWND, # hwndParent
wintypes.DWORD, # Flags
]
SetupDiGetClassDevs.restype = HDEVINFO
SetupDiDestroyDeviceInfoList = windll.setupapi.SetupDiDestroyDeviceInfoList
SetupDiDestroyDeviceInfoList.argtypes = [
HDEVINFO, # DeviceInfoSet
]
SetupDiDestroyDeviceInfoList.restype = wintypes.BOOL
SetupDiEnumDeviceInterfaces = windll.setupapi.SetupDiEnumDeviceInterfaces SetupDiEnumDeviceInterfaces = windll.setupapi.SetupDiEnumDeviceInterfaces
SetupDiGetInterfaceDeviceDetail = ( SetupDiEnumDeviceInterfaces.argtypes = [
windll.setupapi.SetupDiGetDeviceInterfaceDetailA) HDEVINFO, # DeviceInfoSet
PSP_DEVINFO_DATA, # DeviceInfoData
ctypes.POINTER(GUID), # InterfaceClassGuid
wintypes.DWORD, # MemberIndex
PSP_DEVICE_INTERFACE_DATA, # DeviceInterfaceData
]
SetupDiEnumDeviceInterfaces.restype = wintypes.BOOL
SetupDiGetDeviceInterfaceDetail = windll.setupapi.SetupDiGetDeviceInterfaceDetailA
SetupDiGetDeviceInterfaceDetail.argtypes = [
HDEVINFO, # DeviceInfoSet
PSP_DEVICE_INTERFACE_DATA, # DeviceInterfaceData
PSP_DEVICE_INTERFACE_DETAIL_DATA_A, # DeviceInterfaceDetailData
wintypes.DWORD, # DeviceInterfaceDetailDataSize
wintypes.PDWORD, # RequiredSize
PSP_DEVINFO_DATA, # DeviceInfoData
]
SetupDiGetDeviceInterfaceDetail.restype = wintypes.BOOL
CreateFile = windll.kernel32.CreateFileA CreateFile = windll.kernel32.CreateFileA
CreateFile.argtypes = [
wintypes.LPCSTR, # lpFileName
wintypes.DWORD, # dwDesiredAccess
wintypes.DWORD, # dwShareMode
LPSECURITY_ATTRIBUTES, # lpSecurityAttributes
wintypes.DWORD, # dwCreationDisposition
wintypes.DWORD, # dwFlagsAndAttributes
wintypes.HANDLE, # hTemplateFile
]
CreateFile.restype = wintypes.HANDLE
ReadFile = windll.kernel32.ReadFile ReadFile = windll.kernel32.ReadFile
ReadFile.argtypes = [
wintypes.HANDLE, # hFile
wintypes.LPVOID, # lpBuffer
wintypes.DWORD, # nNumberOfBytesToRead
wintypes.LPDWORD, # lpNumberOfBytesRead
LPOVERLAPPED, # lpOverlapped
]
ReadFile.restype = wintypes.BOOL
WriteFile = windll.kernel32.WriteFile WriteFile = windll.kernel32.WriteFile
WriteFile.argtypes = [
wintypes.HANDLE, # hFile
wintypes.LPCVOID, # lpBuffer
wintypes.DWORD, # nNumberOfBytesToWrite
wintypes.LPDWORD, # lpNumberOfBytesWritten
LPOVERLAPPED, # lpOverlapped
]
WriteFile.restype = wintypes.BOOL
CloseHandle = windll.kernel32.CloseHandle CloseHandle = windll.kernel32.CloseHandle
GetLastError = windll.kernel32.GetLastError CloseHandle.argtypes = [
wintypes.HANDLE, # hObject
]
CloseHandle.restype = wintypes.BOOL
# Defines.
CREATE_ALWAYS = 2
CREATE_NEW = 1
DIGCF_DEVICEINTERFACE = 0x00000010
DIGCF_PRESENT = 0x00000002
ERROR_INSUFFICIENT_BUFFER = 0x0000007A
ERROR_NO_MORE_ITEMS = 0x00000103
FILE_ATTRIBUTE_NORMAL = 0x80
FILE_SHARE_READ = 0x00000001
FILE_SHARE_WRITE = 0x00000002
GENERIC_READ = 0x80000000
GENERIC_WRITE = 0x40000000
INVALID_HANDLE_VALUE = -1 INVALID_HANDLE_VALUE = -1
ERROR_INSUFFICIENT_BUFFER = 122
class StenographMachine: class StenographMachine:
def __init__(self): def __init__(self):
self._usb_device = wintypes.HANDLE(0) self._usb_device = INVALID_HANDLE_VALUE
self._read_buffer = ctypes.create_string_buffer(MAX_READ + StenoPacket.HEADER_SIZE) self._read_buffer = ctypes.create_string_buffer(MAX_READ + StenoPacket.HEADER_SIZE)
@staticmethod @staticmethod
def _open_device_instance(device_info, guid): def _open_device_instance(device_info, guid):
dev_interface_data = DeviceInterfaceData() dev_interface_data = SP_DEVICE_INTERFACE_DATA()
dev_interface_data.cbSize = ctypes.sizeof(dev_interface_data) dev_interface_data.cbSize = ctypes.sizeof(SP_DEVICE_INTERFACE_DATA)
status = SetupDiEnumDeviceInterfaces( if not SetupDiEnumDeviceInterfaces(
device_info, None, guid.bytes, 0, ctypes.byref(dev_interface_data)) device_info, None, ctypes.byref(guid),
if status == 0: 0, ctypes.byref(dev_interface_data)
log.debug('status is zero') ):
if ctypes.GetLastError() != ERROR_NO_MORE_ITEMS:
log.error('SetupDiEnumDeviceInterfaces: %s', ctypes.WinError())
return INVALID_HANDLE_VALUE return INVALID_HANDLE_VALUE
request_length = wintypes.DWORD(0) request_length = wintypes.DWORD(0)
# Call with None to see how big a buffer we need for detail data. status = SetupDiGetDeviceInterfaceDetail(
SetupDiGetInterfaceDeviceDetail(
device_info, device_info,
ctypes.byref(dev_interface_data), ctypes.byref(dev_interface_data),
None, # Call with (None, 0) to see how big a buffer is needed.
0, None, 0,
ctypes.pointer(request_length), ctypes.pointer(request_length),
None None,
) )
err = GetLastError() if status or ctypes.GetLastError() != ERROR_INSUFFICIENT_BUFFER:
if err != ERROR_INSUFFICIENT_BUFFER: log.debug('last error not insufficient buffer: %s', ctypes.WinError())
log.debug('last error not insufficient buffer')
return INVALID_HANDLE_VALUE return INVALID_HANDLE_VALUE
characters = request_length.value dev_detail_data_buffer = ctypes.create_string_buffer(request_length.value)
dev_detail_data_ptr = ctypes.cast(dev_detail_data_buffer, PSP_DEVICE_INTERFACE_DETAIL_DATA_A)
class DeviceDetailData(ctypes.Structure): dev_detail_data_ptr[0].cbSize = ctypes.sizeof(SP_DEVICE_INTERFACE_DETAIL_DATA_A)
_fields_ = [('cbSize', wintypes.DWORD),
('DevicePath', ctypes.c_char * characters)]
dev_detail_data = DeviceDetailData()
dev_detail_data.cbSize = 5
# Now put the actual detail data into the buffer # Now put the actual detail data into the buffer
status = SetupDiGetInterfaceDeviceDetail( if not SetupDiGetDeviceInterfaceDetail(
device_info, ctypes.byref(dev_interface_data), ctypes.byref(dev_detail_data), device_info,
characters, ctypes.pointer(request_length), None ctypes.byref(dev_interface_data),
) dev_detail_data_ptr,
if not status: ctypes.sizeof(dev_detail_data_buffer),
log.debug('not status') None,
None,
):
log.error('SetupDiGetDeviceInterfaceDetail: %s', ctypes.WinError())
return INVALID_HANDLE_VALUE return INVALID_HANDLE_VALUE
log.debug('okay, creating file')
return CreateFile( device_path = dev_detail_data_ptr[0].DevicePath
dev_detail_data.DevicePath,
0xC0000000, 0x3, 0, 0x3, 0x80, 0 log.debug('okay, creating file, device path: %s', device_path)
)
handle = CreateFile(device_path,
GENERIC_READ | GENERIC_WRITE,
FILE_SHARE_READ | FILE_SHARE_WRITE,
None,
CREATE_ALWAYS | CREATE_NEW,
FILE_ATTRIBUTE_NORMAL,
None)
if handle == INVALID_HANDLE_VALUE:
log.error('CreateFile: %s', ctypes.WinError())
return handle
@staticmethod @staticmethod
def _open_device_by_class_interface_and_instance(class_guid): def _open_device_by_class_interface_and_instance(class_guid):
device_info = SetupDiGetClassDevs(class_guid.bytes, 0, 0, 0x12) device_info = SetupDiGetClassDevs(ctypes.byref(class_guid), None, None,
DIGCF_DEVICEINTERFACE | DIGCF_PRESENT)
if device_info == INVALID_HANDLE_VALUE: if device_info == INVALID_HANDLE_VALUE:
log.debug('dev info is invalid handle') log.error('SetupDiGetClassDevs: %s', ctypes.WinError())
return INVALID_HANDLE_VALUE return INVALID_HANDLE_VALUE
usb_device = StenographMachine._open_device_instance(device_info, class_guid)
usb_device = StenographMachine._open_device_instance( if not SetupDiDestroyDeviceInfoList(device_info):
device_info, class_guid) log.error('SetupDiDestroyDeviceInfoList: %s', ctypes.WinError())
return usb_device return usb_device
def _usb_write_packet(self, request): def _usb_write_packet(self, request):
bytes_written = wintypes.DWORD(0) bytes_written = wintypes.DWORD(0)
request_packet = request.pack() request_packet = request.pack()
WriteFile( if not WriteFile(self._usb_device,
self._usb_device, request_packet,
request_packet, StenoPacket.HEADER_SIZE + request.data_length,
StenoPacket.HEADER_SIZE + request.data_length, ctypes.byref(bytes_written),
ctypes.byref(bytes_written), None):
None log.error('WriteFile: %s', ctypes.WinError())
) return 0
return bytes_written.value return bytes_written.value
def _usb_read_packet(self): def _usb_read_packet(self):
bytes_read = wintypes.DWORD(0) bytes_read = wintypes.DWORD(0)
ReadFile( if not ReadFile(self._usb_device,
self._usb_device, self._read_buffer,
ctypes.byref(self._read_buffer), MAX_READ + StenoPacket.HEADER_SIZE,
MAX_READ + StenoPacket.HEADER_SIZE, ctypes.byref(bytes_read),
ctypes.byref(bytes_read), None):
None log.error('ReadFile: %s', ctypes.WinError())
) return None
# Return None if not enough data was read. # Return None if not enough data was read.
if bytes_read.value < StenoPacket.HEADER_SIZE: if bytes_read.value < StenoPacket.HEADER_SIZE:
log.error('ReadFile: short read, %u < %u',
bytes_read.value, StenoPacket.HEADER_SIZE)
return None return None
writer_packet = StenoPacket.unpack(self._read_buffer) writer_packet = StenoPacket.unpack(self._read_buffer)
return writer_packet return writer_packet
def disconnect(self): def disconnect(self):
CloseHandle(self._usb_device) if not CloseHandle(self._usb_device):
log.error('CloseHandle: %s', ctypes.WinError())
self._usb_device = INVALID_HANDLE_VALUE self._usb_device = INVALID_HANDLE_VALUE
def connect(self): def connect(self):
# If already connected, disconnect first. # If already connected, disconnect first.
if self._usb_device != INVALID_HANDLE_VALUE: if self._usb_device != INVALID_HANDLE_VALUE:
self.disconnect() self.disconnect()
self._usb_device = ( self._usb_device = self._open_device_by_class_interface_and_instance(USB_WRITER_GUID)
self._open_device_by_class_interface_and_instance(
USB_WRITER_GUID))
return self._usb_device != INVALID_HANDLE_VALUE return self._usb_device != INVALID_HANDLE_VALUE
def send_receive(self, request): def send_receive(self, request):