Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-06-25 08:26
📄 __init__.py # Copyright 2016 Google Inc. All Rights Reserved. # # 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 # # http://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. """Implements interface for talking to hid devices. This module implenets an interface for talking to low level hid devices using various methods on different platforms. """ import sys def Enumerate(): return InternalPlatformSwitch('Enumerate') def Open(path): return InternalPlatformSwitch('__init__', path) def InternalPlatformSwitch(funcname, args, kwargs): """Determine, on a platform-specific basis, which module to use.""" # pylint: disable=g-import-not-at-top clz = None if sys.platform.startswith('linux'): from pyu2f.hid import linux clz = linux.LinuxHidDevice elif sys.platform.startswith('win32'): from pyu2f.hid import windows clz = windows.WindowsHidDevice elif sys.platform.startswith('darwin'): from pyu2f.hid import macos clz = macos.MacOsHidDevice if not clz: raise Exception('Unsupported platform: ' + sys.platform) if funcname == '__init__': return clz(args, *kwargs) return getattr(clz, funcname)(args, **kwargs)	1.56 KB	2025-02-24 20:22
📄 base.py # Copyright 2016 Google Inc. All Rights Reserved. # # 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 # # http://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. """Implement base classes for hid package. This module provides the base classes implemented by the platform-specific modules. It includes a base class for all implementations built on interacting with file-like objects. """ class HidDevice(object): """Base class for all HID devices in this package.""" @staticmethod def Enumerate(): """Enumerates all the hid devices. This function enumerates all the hid device and provides metadata for helping the client select one. Returns: A list of dictionaries of metadata. Each implementation is required to provide at least: vendor_id, product_id, product_string, usage, usage_page, and path. """ pass def __init__(self, path): """Initialize the device at path.""" pass def GetInReportDataLength(self): """Returns the max input report data length in bytes. Returns the max input report data length in bytes. This excludes the report id. """ pass def GetOutReportDataLength(self): """Returns the max output report data length in bytes. Returns the max output report data length in bytes. This excludes the report id. """ pass def Write(self, packet): """Writes packet to device. Writes the packet to the device. Args: packet: An array of integers to write to the device. Excludes the report ID. Must be equal to GetOutReportLength(). """ pass def Read(self): """Reads packet from device. Reads the packet from the device. Returns: An array of integers read from the device. Excludes the report ID. The length is equal to GetInReportDataLength(). """ pass class DeviceDescriptor(object): """Descriptor for basic attributes of the device.""" usage_page = None usage = None vendor_id = None product_id = None product_string = None path = None internal_max_in_report_len = 0 internal_max_out_report_len = 0 def ToPublicDict(self): out = {} for k, v in list(self.__dict__.items()): if not k.startswith('internal_'): out[k] = v return out	2.65 KB	2025-02-24 20:22
📄 linux.py # Copyright 2016 Google Inc. All Rights Reserved. # # 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 # # http://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. """Implements raw HID interface on Linux using SysFS and device files.""" from __future__ import division import os import struct from pyu2f import errors from pyu2f.hid import base REPORT_DESCRIPTOR_KEY_MASK = 0xfc LONG_ITEM_ENCODING = 0xfe OUTPUT_ITEM = 0x90 INPUT_ITEM = 0x80 COLLECTION_ITEM = 0xa0 REPORT_COUNT = 0x94 REPORT_SIZE = 0x74 USAGE_PAGE = 0x04 USAGE = 0x08 def GetValueLength(rd, pos): """Get value length for a key in rd. For a key at position pos in the Report Descriptor rd, return the length of the associated value. This supports both short and long format values. Args: rd: Report Descriptor pos: The position of the key in rd. Returns: (key_size, data_len) where key_size is the number of bytes occupied by the key and data_len is the length of the value associated by the key. """ rd = bytearray(rd) key = rd[pos] if key == LONG_ITEM_ENCODING: # If the key is tagged as a long item (0xfe), then the format is # [key (1 byte)] [data len (1 byte)] [item tag (1 byte)] [data (n # bytes)]. # Thus, the entire key record is 3 bytes long. if pos + 1 < len(rd): return (3, rd[pos + 1]) else: raise errors.HidError('Malformed report descriptor') else: # If the key is tagged as a short item, then the item tag and data len are # packed into one byte. The format is thus: # [tag (high 4 bits)] [type (2 bits)] [size code (2 bits)] [data (n bytes)]. # The size code specifies 1,2, or 4 bytes (0x03 means 4 bytes). code = key & 0x03 if code <= 0x02: return (1, code) elif code == 0x03: return (1, 4) raise errors.HidError('Cannot happen') def ReadLsbBytes(rd, offset, value_size): """Reads value_size bytes from rd at offset, least signifcant byte first.""" encoding = None if value_size == 1: encoding = '<B' elif value_size == 2: encoding = '<H' elif value_size == 4: encoding = '<L' else: raise errors.HidError('Invalid value size specified') ret, = struct.unpack(encoding, rd[offset:offset + value_size]) return ret class NoReportCountFound(Exception): pass def ParseReportDescriptor(rd, desc): """Parse the binary report descriptor. Parse the binary report descriptor into a DeviceDescriptor object. Args: rd: The binary report descriptor desc: The DeviceDescriptor object to update with the results from parsing the descriptor. Returns: None """ rd = bytearray(rd) pos = 0 report_count = None report_size = None usage_page = None usage = None while pos < len(rd): key = rd[pos] # First step, determine the value encoding (either long or short). key_size, value_length = GetValueLength(rd, pos) if key & REPORT_DESCRIPTOR_KEY_MASK == INPUT_ITEM: if report_count and report_size: byte_length = (report_count * report_size) // 8 desc.internal_max_in_report_len = max( desc.internal_max_in_report_len, byte_length) report_count = None report_size = None elif key & REPORT_DESCRIPTOR_KEY_MASK == OUTPUT_ITEM: if report_count and report_size: byte_length = (report_count * report_size) // 8 desc.internal_max_out_report_len = max( desc.internal_max_out_report_len, byte_length) report_count = None report_size = None elif key & REPORT_DESCRIPTOR_KEY_MASK == COLLECTION_ITEM: if usage_page: desc.usage_page = usage_page if usage: desc.usage = usage elif key & REPORT_DESCRIPTOR_KEY_MASK == REPORT_COUNT: if len(rd) >= pos + 1 + value_length: report_count = ReadLsbBytes(rd, pos + 1, value_length) elif key & REPORT_DESCRIPTOR_KEY_MASK == REPORT_SIZE: if len(rd) >= pos + 1 + value_length: report_size = ReadLsbBytes(rd, pos + 1, value_length) elif key & REPORT_DESCRIPTOR_KEY_MASK == USAGE_PAGE: if len(rd) >= pos + 1 + value_length: usage_page = ReadLsbBytes(rd, pos + 1, value_length) elif key & REPORT_DESCRIPTOR_KEY_MASK == USAGE: if len(rd) >= pos + 1 + value_length: usage = ReadLsbBytes(rd, pos + 1, value_length) pos += value_length + key_size return desc def ParseUevent(uevent, desc): lines = uevent.split(b'\n') for line in lines: line = line.strip() if not line: continue k, v = line.split(b'=') if k == b'HID_NAME': desc.product_string = v.decode('utf8') elif k == b'HID_ID': _, vid, pid = v.split(b':') desc.vendor_id = int(vid, 16) desc.product_id = int(pid, 16) class LinuxHidDevice(base.HidDevice): """Implementation of HID device for linux. Implementation of HID device interface for linux that uses block devices to interact with the device and sysfs to enumerate/discover device metadata. """ @staticmethod def Enumerate(): hidraw_devices = [] try: hidraw_devices = os.listdir('/sys/class/hidraw') except FileNotFoundError: raise errors.OsHidError('No hidraw device is available') for dev in hidraw_devices: rd_path = ( os.path.join( '/sys/class/hidraw', dev, 'device/report_descriptor')) uevent_path = os.path.join('/sys/class/hidraw', dev, 'device/uevent') rd_file = open(rd_path, 'rb') uevent_file = open(uevent_path, 'rb') desc = base.DeviceDescriptor() desc.path = os.path.join('/dev/', dev) ParseReportDescriptor(rd_file.read(), desc) ParseUevent(uevent_file.read(), desc) rd_file.close() uevent_file.close() yield desc.ToPublicDict() def __init__(self, path): base.HidDevice.__init__(self, path) self.dev = os.open(path, os.O_RDWR) self.desc = base.DeviceDescriptor() self.desc.path = path rd_file = open(os.path.join('/sys/class/hidraw', os.path.basename(path), 'device/report_descriptor'), 'rb') ParseReportDescriptor(rd_file.read(), self.desc) rd_file.close() def GetInReportDataLength(self): """See base class.""" return self.desc.internal_max_in_report_len def GetOutReportDataLength(self): """See base class.""" return self.desc.internal_max_out_report_len def Write(self, packet): """See base class.""" out = bytearray([0] + packet) # Prepend the zero-byte (report ID) os.write(self.dev, out) def Read(self): """See base class.""" raw_in = os.read(self.dev, self.GetInReportDataLength()) decoded_in = list(bytearray(raw_in)) return decoded_in	7.02 KB	2025-02-24 20:22
📄 macos.py # Copyright 2016 Google Inc. All Rights Reserved. # # 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 # # http://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. """Implements HID device interface on MacOS using IOKit and HIDManager.""" import ctypes import ctypes.util import logging import queue import range import sys import threading from pyu2f import errors from pyu2f.hid import base logger = logging.getLogger('pyu2f.macos') # Constants DEVICE_PATH_BUFFER_SIZE = 512 DEVICE_STRING_PROPERTY_BUFFER_SIZE = 512 HID_DEVICE_PROPERTY_VENDOR_ID = 'VendorId' HID_DEVICE_PROPERTY_PRODUCT_ID = 'ProductID' HID_DEVICE_PROPERTY_PRODUCT = 'Product' HID_DEVICE_PROPERTY_PRIMARY_USAGE = 'PrimaryUsage' HID_DEVICE_PROPERTY_PRIMARY_USAGE_PAGE = 'PrimaryUsagePage' HID_DEVICE_PROPERTY_MAX_INPUT_REPORT_SIZE = 'MaxInputReportSize' HID_DEVICE_PROPERTY_MAX_OUTPUT_REPORT_SIZE = 'MaxOutputReportSize' HID_DEVICE_PROPERTY_REPORT_ID = 'ReportID' # Declare C types class _CFType(ctypes.Structure): pass class _CFString(_CFType): pass class _CFSet(_CFType): pass class _IOHIDManager(_CFType): pass class _IOHIDDevice(_CFType): pass class _CFRunLoop(_CFType): pass class _CFAllocator(_CFType): pass # Linter isn't consistent about valid class names. Disabling some of the errors CF_SET_REF = ctypes.POINTER(_CFSet) CF_STRING_REF = ctypes.POINTER(_CFString) CF_TYPE_REF = ctypes.POINTER(_CFType) CF_RUN_LOOP_REF = ctypes.POINTER(_CFRunLoop) CF_RUN_LOOP_RUN_RESULT = ctypes.c_int32 CF_ALLOCATOR_REF = ctypes.POINTER(_CFAllocator) CF_TYPE_ID = ctypes.c_ulong # pylint: disable=invalid-name CF_INDEX = ctypes.c_long # pylint: disable=invalid-name CF_TIME_INTERVAL = ctypes.c_double # pylint: disable=invalid-name IO_RETURN = ctypes.c_uint IO_HID_REPORT_TYPE = ctypes.c_uint IO_OBJECT_T = ctypes.c_uint MACH_PORT_T = ctypes.c_uint IO_STRING_T = ctypes.c_char_p # pylint: disable=invalid-name IO_SERVICE_T = IO_OBJECT_T IO_REGISTRY_ENTRY_T = IO_OBJECT_T IO_HID_MANAGER_REF = ctypes.POINTER(_IOHIDManager) IO_HID_DEVICE_REF = ctypes.POINTER(_IOHIDDevice) IO_HID_REPORT_CALLBACK = ctypes.CFUNCTYPE(None, ctypes.py_object, IO_RETURN, ctypes.c_void_p, IO_HID_REPORT_TYPE, ctypes.c_uint32, ctypes.POINTER(ctypes.c_uint8), CF_INDEX) # Define C constants K_CF_NUMBER_SINT32_TYPE = 3 K_CF_STRING_ENCODING_UTF8 = 0x08000100 K_CF_ALLOCATOR_DEFAULT = None K_IO_SERVICE_PLANE = b'IOService' K_IO_MASTER_PORT_DEFAULT = 0 K_IO_HID_REPORT_TYPE_OUTPUT = 1 K_IO_RETURN_SUCCESS = 0 K_CF_RUN_LOOP_RUN_STOPPED = 2 K_CF_RUN_LOOP_RUN_TIMED_OUT = 3 K_CF_RUN_LOOP_RUN_HANDLED_SOURCE = 4 # Load relevant libraries iokit = ctypes.cdll.LoadLibrary(ctypes.util.find_library('IOKit')) cf = ctypes.cdll.LoadLibrary(ctypes.util.find_library('CoreFoundation')) # Only use iokit and cf if we're on macos, this way we can still run tests # on other platforms if we properly mock if sys.platform.startswith('darwin'): # Exported constants K_CF_RUNLOOP_DEFAULT_MODE = CF_STRING_REF.in_dll(cf, 'kCFRunLoopDefaultMode') # Declare C function prototypes cf.CFSetGetValues.restype = None cf.CFSetGetValues.argtypes = [CF_SET_REF, ctypes.POINTER(ctypes.c_void_p)] cf.CFStringCreateWithCString.restype = CF_STRING_REF cf.CFStringCreateWithCString.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_uint32] cf.CFStringGetCString.restype = ctypes.c_int cf.CFStringGetCString.argtypes = [CF_STRING_REF, ctypes.c_char_p, CF_INDEX, ctypes.c_uint32] cf.CFStringGetLength.restype = CF_INDEX cf.CFStringGetLength.argtypes = [CF_STRING_REF] cf.CFGetTypeID.restype = CF_TYPE_ID cf.CFGetTypeID.argtypes = [CF_TYPE_REF] cf.CFNumberGetTypeID.restype = CF_TYPE_ID cf.CFNumberGetValue.restype = ctypes.c_int cf.CFRelease.restype = None cf.CFRelease.argtypes = [CF_TYPE_REF] cf.CFRunLoopGetCurrent.restype = CF_RUN_LOOP_REF cf.CFRunLoopGetCurrent.argTypes = [] cf.CFRunLoopRunInMode.restype = CF_RUN_LOOP_RUN_RESULT cf.CFRunLoopRunInMode.argtypes = [CF_STRING_REF, CF_TIME_INTERVAL, ctypes.c_bool] iokit.IOObjectRelease.argtypes = [IO_OBJECT_T] iokit.IOHIDManagerCreate.restype = IO_HID_MANAGER_REF iokit.IOHIDManagerCopyDevices.restype = CF_SET_REF iokit.IOHIDManagerCopyDevices.argtypes = [IO_HID_MANAGER_REF] iokit.IOHIDManagerSetDeviceMatching.restype = None iokit.IOHIDManagerSetDeviceMatching.argtypes = [IO_HID_MANAGER_REF, CF_TYPE_REF] iokit.IOHIDDeviceGetProperty.restype = CF_TYPE_REF iokit.IOHIDDeviceGetProperty.argtypes = [IO_HID_DEVICE_REF, CF_STRING_REF] iokit.IOHIDDeviceRegisterInputReportCallback.restype = None iokit.IOHIDDeviceRegisterInputReportCallback.argtypes = [ IO_HID_DEVICE_REF, ctypes.POINTER(ctypes.c_uint8), CF_INDEX, IO_HID_REPORT_CALLBACK, ctypes.py_object] iokit.IORegistryEntryFromPath.restype = IO_REGISTRY_ENTRY_T iokit.IORegistryEntryFromPath.argtypes = [MACH_PORT_T, IO_STRING_T] iokit.IOHIDDeviceCreate.restype = IO_HID_DEVICE_REF iokit.IOHIDDeviceCreate.argtypes = [CF_ALLOCATOR_REF, IO_SERVICE_T] iokit.IOHIDDeviceScheduleWithRunLoop.restype = None iokit.IOHIDDeviceScheduleWithRunLoop.argtypes = [IO_HID_DEVICE_REF, CF_RUN_LOOP_REF, CF_STRING_REF] iokit.IOHIDDeviceUnscheduleFromRunLoop.restype = None iokit.IOHIDDeviceUnscheduleFromRunLoop.argtypes = [IO_HID_DEVICE_REF, CF_RUN_LOOP_REF, CF_STRING_REF] iokit.IOHIDDeviceSetReport.restype = IO_RETURN iokit.IOHIDDeviceSetReport.argtypes = [IO_HID_DEVICE_REF, IO_HID_REPORT_TYPE, CF_INDEX, ctypes.POINTER(ctypes.c_uint8), CF_INDEX] else: logger.warning('Not running on MacOS') def CFStr(s): """Builds a CFString from a python string. Args: s: source string Returns: CFStringRef representation of the source string Resulting CFString must be CFReleased when no longer needed. """ return cf.CFStringCreateWithCString(None, s.encode(), 0) def GetDeviceIntProperty(dev_ref, key): """Reads int property from the HID device.""" cf_key = CFStr(key) type_ref = iokit.IOHIDDeviceGetProperty(dev_ref, cf_key) cf.CFRelease(cf_key) if not type_ref: return None if cf.CFGetTypeID(type_ref) != cf.CFNumberGetTypeID(): raise errors.OsHidError('Expected number type, got {}'.format( cf.CFGetTypeID(type_ref))) out = ctypes.c_int32() ret = cf.CFNumberGetValue(type_ref, K_CF_NUMBER_SINT32_TYPE, ctypes.byref(out)) if not ret: return None return out.value def GetDeviceStringProperty(dev_ref, key): """Reads string property from the HID device.""" cf_key = CFStr(key) type_ref = iokit.IOHIDDeviceGetProperty(dev_ref, cf_key) cf.CFRelease(cf_key) if not type_ref: return None if cf.CFGetTypeID(type_ref) != cf.CFStringGetTypeID(): raise errors.OsHidError('Expected string type, got {}'.format( cf.CFGetTypeID(type_ref))) type_ref = ctypes.cast(type_ref, CF_STRING_REF) out = ctypes.create_string_buffer(DEVICE_STRING_PROPERTY_BUFFER_SIZE) ret = cf.CFStringGetCString(type_ref, out, DEVICE_STRING_PROPERTY_BUFFER_SIZE, K_CF_STRING_ENCODING_UTF8) if not ret: return None return out.value.decode('utf8') def GetDevicePath(device_handle): """Obtains the unique path for the device. Args: device_handle: reference to the device Returns: A unique path for the device, obtained from the IO Registry """ # Obtain device path from IO Registry io_service_obj = iokit.IOHIDDeviceGetService(device_handle) str_buffer = ctypes.create_string_buffer(DEVICE_PATH_BUFFER_SIZE) iokit.IORegistryEntryGetPath(io_service_obj, K_IO_SERVICE_PLANE, str_buffer) return str_buffer.value def HidReadCallback(read_queue, result, sender, report_type, report_id, report, report_length): """Handles incoming IN report from HID device.""" del result, sender, report_type, report_id # Unused by the callback function incoming_bytes = [report[i] for i in range(report_length)] read_queue.put(incoming_bytes) # C wrapper around ReadCallback() # Declared in this scope so it doesn't get GC-ed REGISTERED_READ_CALLBACK = IO_HID_REPORT_CALLBACK(HidReadCallback) def DeviceReadThread(hid_device): """Binds a device to the thread's run loop, then starts the run loop. Args: hid_device: The MacOsHidDevice object The HID manager requires a run loop to handle Report reads. This thread function serves that purpose. """ # Schedule device events with run loop hid_device.run_loop_ref = cf.CFRunLoopGetCurrent() if not hid_device.run_loop_ref: logger.error('Failed to get current run loop') return iokit.IOHIDDeviceScheduleWithRunLoop(hid_device.device_handle, hid_device.run_loop_ref, K_CF_RUNLOOP_DEFAULT_MODE) # Run the run loop run_loop_run_result = K_CF_RUN_LOOP_RUN_TIMED_OUT while (run_loop_run_result == K_CF_RUN_LOOP_RUN_TIMED_OUT or run_loop_run_result == K_CF_RUN_LOOP_RUN_HANDLED_SOURCE): run_loop_run_result = cf.CFRunLoopRunInMode( K_CF_RUNLOOP_DEFAULT_MODE, 1000, # Timeout in seconds False) # Return after source handled # log any unexpected run loop exit if run_loop_run_result != K_CF_RUN_LOOP_RUN_STOPPED: logger.error('Unexpected run loop exit code: %d', run_loop_run_result) # Unschedule from run loop iokit.IOHIDDeviceUnscheduleFromRunLoop(hid_device.device_handle, hid_device.run_loop_ref, K_CF_RUNLOOP_DEFAULT_MODE) class MacOsHidDevice(base.HidDevice): """Implementation of HID device for MacOS. Uses IOKit HID Manager to interact with the device. """ @staticmethod def Enumerate(): """See base class.""" # Init a HID manager hid_mgr = iokit.IOHIDManagerCreate(None, None) if not hid_mgr: raise errors.OsHidError('Unable to obtain HID manager reference') iokit.IOHIDManagerSetDeviceMatching(hid_mgr, None) # Get devices from HID manager device_set_ref = iokit.IOHIDManagerCopyDevices(hid_mgr) if not device_set_ref: raise errors.OsHidError('Failed to obtain devices from HID manager') num = iokit.CFSetGetCount(device_set_ref) devices = (IO_HID_DEVICE_REF * num)() iokit.CFSetGetValues(device_set_ref, devices) # Retrieve and build descriptor dictionaries for each device descriptors = [] for dev in devices: d = base.DeviceDescriptor() d.vendor_id = GetDeviceIntProperty(dev, HID_DEVICE_PROPERTY_VENDOR_ID) d.product_id = GetDeviceIntProperty(dev, HID_DEVICE_PROPERTY_PRODUCT_ID) d.product_string = GetDeviceStringProperty(dev, HID_DEVICE_PROPERTY_PRODUCT) d.usage = GetDeviceIntProperty(dev, HID_DEVICE_PROPERTY_PRIMARY_USAGE) d.usage_page = GetDeviceIntProperty( dev, HID_DEVICE_PROPERTY_PRIMARY_USAGE_PAGE) d.report_id = GetDeviceIntProperty(dev, HID_DEVICE_PROPERTY_REPORT_ID) d.path = GetDevicePath(dev) descriptors.append(d.ToPublicDict()) # Clean up CF objects cf.CFRelease(device_set_ref) cf.CFRelease(hid_mgr) return descriptors def __init__(self, path): # Resolve the path to device handle device_entry = iokit.IORegistryEntryFromPath(K_IO_MASTER_PORT_DEFAULT, path) if not device_entry: raise errors.OsHidError('Device path does not match any HID device on ' 'the system') self.device_handle = iokit.IOHIDDeviceCreate(K_CF_ALLOCATOR_DEFAULT, device_entry) if not self.device_handle: raise errors.OsHidError('Failed to obtain device handle from registry ' 'entry') iokit.IOObjectRelease(device_entry) self.device_path = path # Open device result = iokit.IOHIDDeviceOpen(self.device_handle, 0) if result != K_IO_RETURN_SUCCESS: raise errors.OsHidError('Failed to open device for communication: {}' .format(result)) # Create read queue self.read_queue = queue.Queue() # Create and start read thread self.run_loop_ref = None self.read_thread = threading.Thread(target=DeviceReadThread, args=(self,)) self.read_thread.daemon = True self.read_thread.start() # Read max report sizes for in/out self.internal_max_in_report_len = GetDeviceIntProperty( self.device_handle, HID_DEVICE_PROPERTY_MAX_INPUT_REPORT_SIZE) if not self.internal_max_in_report_len: raise errors.OsHidError('Unable to obtain max in report size') self.internal_max_out_report_len = GetDeviceIntProperty( self.device_handle, HID_DEVICE_PROPERTY_MAX_OUTPUT_REPORT_SIZE) if not self.internal_max_out_report_len: raise errors.OsHidError('Unable to obtain max out report size') # Register read callback self.in_report_buffer = (ctypes.c_uint8 * self.internal_max_in_report_len)() iokit.IOHIDDeviceRegisterInputReportCallback( self.device_handle, self.in_report_buffer, self.internal_max_in_report_len, REGISTERED_READ_CALLBACK, ctypes.py_object(self.read_queue)) def GetInReportDataLength(self): """See base class.""" return self.internal_max_in_report_len def GetOutReportDataLength(self): """See base class.""" return self.internal_max_out_report_len def Write(self, packet): """See base class.""" report_id = 0 out_report_buffer = (ctypes.c_uint8 * self.internal_max_out_report_len)() out_report_buffer[:] = packet[:] result = iokit.IOHIDDeviceSetReport(self.device_handle, K_IO_HID_REPORT_TYPE_OUTPUT, report_id, out_report_buffer, self.internal_max_out_report_len) # Non-zero status indicates failure if result != K_IO_RETURN_SUCCESS: raise errors.OsHidError('Failed to write report to device') def Read(self): """See base class.""" result = None while result is None: try: result = self.read_queue.get(timeout=60) except queue.Empty: continue return result def __del__(self): # Unregister the callback if hasattr(self, 'in_report_buffer'): iokit.IOHIDDeviceRegisterInputReportCallback( self.device_handle, self.in_report_buffer, self.internal_max_in_report_len, None, None) # Stop the run loop if hasattr(self, 'run_loop_ref'): cf.CFRunLoopStop(self.run_loop_ref) # Wait for the read thread to exit if hasattr(self, 'read_thread'): self.read_thread.join()	15.57 KB	2025-02-24 20:22
📄 try.py # Copyright 2016 Google Inc. All Rights Reserved. # # 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 # # http://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. """Simple test program for hid interface. This simple test program lists all of the HID devices on the local system. """ from __future__ import print_function from pyu2f import hid def main(): devs = hid.Enumerate() for dev in devs: print(dev) if __name__ == '__main__': main()	890 B	2025-02-24 20:22
📄 windows.py # Copyright 2016 Google Inc. All Rights Reserved. # # 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 # # http://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. """Implements raw HID device communication on Windows.""" import ctypes from ctypes import wintypes import platform from pyu2f import errors from pyu2f.hid import base # Load relevant DLLs hid = ctypes.windll.Hid setupapi = ctypes.windll.SetupAPI kernel32 = ctypes.windll.Kernel32 # Various structs that are used in the Windows APIs we call class GUID(ctypes.Structure): _fields_ = [("Data1", ctypes.c_ulong), ("Data2", ctypes.c_ushort), ("Data3", ctypes.c_ushort), ("Data4", ctypes.c_ubyte * 8)] # On Windows, SetupAPI.h packs structures differently in 64bit and # 32bit mode. In 64bit mode, thestructures are packed on 8 byte # boundaries, while in 32bit mode, they are packed on 1 byte boundaries. # This is important to get right for some API calls that fill out these # structures. if platform.architecture()[0] == "64bit": SETUPAPI_PACK = 8 elif platform.architecture()[0] == "32bit": SETUPAPI_PACK = 1 else: raise errors.HidError("Unknown architecture: %s" % platform.architecture()[0]) class DeviceInterfaceData(ctypes.Structure): _fields_ = [("cbSize", wintypes.DWORD), ("InterfaceClassGuid", GUID), ("Flags", wintypes.DWORD), ("Reserved", ctypes.POINTER(ctypes.c_ulong))] _pack_ = SETUPAPI_PACK class DeviceInterfaceDetailData(ctypes.Structure): _fields_ = [("cbSize", wintypes.DWORD), ("DevicePath", ctypes.c_byte * 1)] _pack_ = SETUPAPI_PACK class HidAttributes(ctypes.Structure): _fields_ = [("Size", ctypes.c_ulong), ("VendorID", ctypes.c_ushort), ("ProductID", ctypes.c_ushort), ("VersionNumber", ctypes.c_ushort)] class HidCapabilities(ctypes.Structure): _fields_ = [("Usage", ctypes.c_ushort), ("UsagePage", ctypes.c_ushort), ("InputReportByteLength", ctypes.c_ushort), ("OutputReportByteLength", ctypes.c_ushort), ("FeatureReportByteLength", ctypes.c_ushort), ("Reserved", ctypes.c_ushort * 17), ("NotUsed", ctypes.c_ushort * 10)] # Various void* aliases for readability. HDEVINFO = ctypes.c_void_p HANDLE = ctypes.c_void_p PHIDP_PREPARSED_DATA = ctypes.c_void_p # pylint: disable=invalid-name # This is a HANDLE. INVALID_HANDLE_VALUE = 0xffffffff # Status codes NTSTATUS = ctypes.c_long HIDP_STATUS_SUCCESS = 0x00110000 FILE_SHARE_READ = 0x00000001 FILE_SHARE_WRITE = 0x00000002 OPEN_EXISTING = 0x03 ERROR_ACCESS_DENIED = 0x05 # CreateFile Flags GENERIC_WRITE = 0x40000000 GENERIC_READ = 0x80000000 # Function signatures hid.HidD_GetHidGuid.restype = None hid.HidD_GetHidGuid.argtypes = [ctypes.POINTER(GUID)] hid.HidD_GetAttributes.restype = wintypes.BOOLEAN hid.HidD_GetAttributes.argtypes = [HANDLE, ctypes.POINTER(HidAttributes)] hid.HidD_GetPreparsedData.restype = wintypes.BOOLEAN hid.HidD_GetPreparsedData.argtypes = [HANDLE, ctypes.POINTER(PHIDP_PREPARSED_DATA)] hid.HidD_FreePreparsedData.restype = wintypes.BOOLEAN hid.HidD_FreePreparsedData.argtypes = [PHIDP_PREPARSED_DATA] hid.HidD_GetProductString.restype = wintypes.BOOLEAN hid.HidD_GetProductString.argtypes = [HANDLE, ctypes.c_void_p, ctypes.c_ulong] hid.HidP_GetCaps.restype = NTSTATUS hid.HidP_GetCaps.argtypes = [PHIDP_PREPARSED_DATA, ctypes.POINTER(HidCapabilities)] setupapi.SetupDiGetClassDevsA.argtypes = [ctypes.POINTER(GUID), ctypes.c_char_p, wintypes.HWND, wintypes.DWORD] setupapi.SetupDiGetClassDevsA.restype = HDEVINFO setupapi.SetupDiEnumDeviceInterfaces.restype = wintypes.BOOL setupapi.SetupDiEnumDeviceInterfaces.argtypes = [ HDEVINFO, ctypes.c_void_p, ctypes.POINTER(GUID), wintypes.DWORD, ctypes.POINTER(DeviceInterfaceData)] setupapi.SetupDiGetDeviceInterfaceDetailA.restype = wintypes.BOOL setupapi.SetupDiGetDeviceInterfaceDetailA.argtypes = [ HDEVINFO, ctypes.POINTER(DeviceInterfaceData), ctypes.POINTER(DeviceInterfaceDetailData), wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), ctypes.c_void_p] kernel32.CreateFileA.restype = HANDLE kernel32.CreateFileA.argtypes = [ ctypes.c_char_p, wintypes.DWORD, wintypes.DWORD, ctypes.c_void_p, wintypes.DWORD, wintypes.DWORD, HANDLE] kernel32.CloseHandle.restype = wintypes.BOOL kernel32.CloseHandle.argtypes = [HANDLE] kernel32.ReadFile.restype = wintypes.BOOL kernel32.ReadFile.argtypes = [ HANDLE, ctypes.c_void_p, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), ctypes.c_void_p] kernel32.WriteFile.restype = wintypes.BOOL kernel32.WriteFile.argtypes = [ HANDLE, ctypes.c_void_p, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), ctypes.c_void_p] def FillDeviceAttributes(device, descriptor): """Fill out the attributes of the device. Fills the devices HidAttributes and product string into the descriptor. Args: device: A handle to the open device descriptor: The DeviceDescriptor to populate with the attributes. Returns: None Raises: WindowsError when unable to obtain attributes or product string. """ attributes = HidAttributes() result = hid.HidD_GetAttributes(device, ctypes.byref(attributes)) if not result: raise ctypes.WinError() buf = ctypes.create_string_buffer(1024) result = hid.HidD_GetProductString(device, buf, 1024) if not result: raise ctypes.WinError() descriptor.vendor_id = attributes.VendorID descriptor.product_id = attributes.ProductID descriptor.product_string = ctypes.wstring_at(buf) def FillDeviceCapabilities(device, descriptor): """Fill out device capabilities. Fills the HidCapabilitites of the device into descriptor. Args: device: A handle to the open device descriptor: DeviceDescriptor to populate with the capabilities Returns: none Raises: WindowsError when unable to obtain capabilitites. """ preparsed_data = PHIDP_PREPARSED_DATA(0) ret = hid.HidD_GetPreparsedData(device, ctypes.byref(preparsed_data)) if not ret: raise ctypes.WinError() try: caps = HidCapabilities() ret = hid.HidP_GetCaps(preparsed_data, ctypes.byref(caps)) if ret != HIDP_STATUS_SUCCESS: raise ctypes.WinError() descriptor.usage = caps.Usage descriptor.usage_page = caps.UsagePage descriptor.internal_max_in_report_len = caps.InputReportByteLength descriptor.internal_max_out_report_len = caps.OutputReportByteLength finally: hid.HidD_FreePreparsedData(preparsed_data) # The python os.open() implementation uses the windows libc # open() function, which writes CreateFile but does so in a way # that doesn't let us open the device with the right set of permissions. # Therefore, we have to directly use the Windows API calls. # We could use PyWin32, which provides simple wrappers. However, to avoid # requiring a PyWin32 dependency for clients, we simply also implement it # using ctypes. def OpenDevice(path, enum=False): """Open the device and return a handle to it.""" desired_access = GENERIC_WRITE \| GENERIC_READ share_mode = FILE_SHARE_READ \| FILE_SHARE_WRITE if enum: desired_access = 0 h = kernel32.CreateFileA(path, desired_access, share_mode, None, OPEN_EXISTING, 0, None) if h == INVALID_HANDLE_VALUE: raise ctypes.WinError() return h class WindowsHidDevice(base.HidDevice): """Implementation of raw HID interface on Windows.""" @staticmethod def Enumerate(): """See base class.""" hid_guid = GUID() hid.HidD_GetHidGuid(ctypes.byref(hid_guid)) devices = setupapi.SetupDiGetClassDevsA( ctypes.byref(hid_guid), None, None, 0x12) index = 0 interface_info = DeviceInterfaceData() interface_info.cbSize = ctypes.sizeof(DeviceInterfaceData) # pylint: disable=invalid-name out = [] while True: result = setupapi.SetupDiEnumDeviceInterfaces( devices, 0, ctypes.byref(hid_guid), index, ctypes.byref(interface_info)) index += 1 if not result: break detail_len = wintypes.DWORD() result = setupapi.SetupDiGetDeviceInterfaceDetailA( devices, ctypes.byref(interface_info), None, 0, ctypes.byref(detail_len), None) detail_len = detail_len.value if detail_len == 0: # skip this device, some kind of error continue buf = ctypes.create_string_buffer(detail_len) interface_detail = DeviceInterfaceDetailData.from_buffer(buf) interface_detail.cbSize = ctypes.sizeof(DeviceInterfaceDetailData) result = setupapi.SetupDiGetDeviceInterfaceDetailA( devices, ctypes.byref(interface_info), ctypes.byref(interface_detail), detail_len, None, None) if not result: raise ctypes.WinError() descriptor = base.DeviceDescriptor() # This is a bit of a hack to work around a limitation of ctypes and # "header" structures that are common in windows. DevicePath is a # ctypes array of length 1, but it is backed with a buffer that is much # longer and contains a null terminated string. So, we read the null # terminated string off DevicePath here. Per the comment above, the # alignment of this struct varies depending on architecture, but # in all cases the path string starts 1 DWORD into the structure. # # The path length is: # length of detail buffer - header length (1 DWORD) path_len = detail_len - ctypes.sizeof(wintypes.DWORD) descriptor.path = ctypes.string_at( ctypes.addressof(interface_detail.DevicePath), path_len) device = None try: device = OpenDevice(descriptor.path, True) except WindowsError as e: # pylint: disable=undefined-variable if e.winerror == ERROR_ACCESS_DENIED: # Access Denied, e.g. a keyboard continue else: raise e try: FillDeviceAttributes(device, descriptor) FillDeviceCapabilities(device, descriptor) out.append(descriptor.ToPublicDict()) except WindowsError as e: continue # skip this device finally: kernel32.CloseHandle(device) return out def __init__(self, path): """See base class.""" base.HidDevice.__init__(self, path) self.dev = OpenDevice(path) self.desc = base.DeviceDescriptor() FillDeviceCapabilities(self.dev, self.desc) def GetInReportDataLength(self): """See base class.""" return self.desc.internal_max_in_report_len - 1 def GetOutReportDataLength(self): """See base class.""" return self.desc.internal_max_out_report_len - 1 def Write(self, packet): """See base class.""" if len(packet) != self.GetOutReportDataLength(): raise errors.HidError("Packet length must match report data length.") packet_data = [0] + packet # Prepend the zero-byte (report ID) out = bytes(bytearray(packet_data)) num_written = wintypes.DWORD() ret = ( kernel32.WriteFile( self.dev, out, len(out), ctypes.byref(num_written), None)) if num_written.value != len(out): raise errors.HidError( "Failed to write complete packet. " + "Expected %d, but got %d" % (len(out), num_written.value)) if not ret: raise ctypes.WinError() def Read(self): """See base class.""" buf = ctypes.create_string_buffer(self.desc.internal_max_in_report_len) num_read = wintypes.DWORD() ret = kernel32.ReadFile( self.dev, buf, len(buf), ctypes.byref(num_read), None) if num_read.value != self.desc.internal_max_in_report_len: raise errors.HidError("Failed to read full length report from device.") if not ret: raise ctypes.WinError() # Convert the string buffer to a list of numbers. Throw away the first # byte, which is the report id (which we don't care about). return list(bytearray(buf[1:])) def __del__(self): """Closes the file handle when object is GC-ed.""" if hasattr(self, 'dev'): kernel32.CloseHandle(self.dev)	12.45 KB	2025-02-24 20:22

Terminal