python-fido2/fido2/hid/macos.py

# Original work 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.
#
# Modified work Copyright 2020 Yubico AB. All Rights Reserved.
# This file, with modifications, is licensed under the above Apache License.

from .base import HidDescriptor, CtapHidConnection, FIDO_USAGE_PAGE, FIDO_USAGE

import ctypes
import ctypes.util
import threading
from queue import Queue, Empty

import logging

logger = logging.getLogger(__name__)

# Constants

HID_DEVICE_PROPERTY_VENDOR_ID = b"VendorID"
HID_DEVICE_PROPERTY_PRODUCT_ID = b"ProductID"
HID_DEVICE_PROPERTY_PRODUCT = b"Product"
HID_DEVICE_PROPERTY_SERIAL_NUMBER = b"SerialNumber"
HID_DEVICE_PROPERTY_PRIMARY_USAGE = b"PrimaryUsage"
HID_DEVICE_PROPERTY_PRIMARY_USAGE_PAGE = b"PrimaryUsagePage"
HID_DEVICE_PROPERTY_MAX_INPUT_REPORT_SIZE = b"MaxInputReportSize"
HID_DEVICE_PROPERTY_MAX_OUTPUT_REPORT_SIZE = b"MaxOutputReportSize"
HID_DEVICE_PROPERTY_REPORT_ID = b"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


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_DICTIONARY_REF = ctypes.c_void_p
CF_MUTABLE_DICTIONARY_REF = ctypes.c_void_p
CF_TYPE_ID = ctypes.c_ulong
CF_INDEX = ctypes.c_long
CF_TIME_INTERVAL = ctypes.c_double
CF_STRING_ENCODING = ctypes.c_uint32
CF_STRING_BUILTIN_ENCODINGS_UTF8 = 134217984
IO_RETURN = ctypes.c_uint
IO_HID_REPORT_TYPE = ctypes.c_uint
IO_OPTION_BITS = ctypes.c_uint
IO_OBJECT_T = ctypes.c_uint
MACH_PORT_T = ctypes.c_uint
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,
)
IO_HID_CALLBACK = ctypes.CFUNCTYPE(None, ctypes.py_object, IO_RETURN, ctypes.c_void_p)

# Define C constants
K_CF_NUMBER_SINT32_TYPE = 3
K_CF_ALLOCATOR_DEFAULT = None

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
# NOTE: find_library doesn't currently work on Big Sur, requiring the hardcoded paths
iokit = ctypes.cdll.LoadLibrary(
    ctypes.util.find_library("IOKit")
    or "/System/Library/Frameworks/IOKit.framework/IOKit"
)
cf = ctypes.cdll.LoadLibrary(
    ctypes.util.find_library("CoreFoundation")
    or "/System/Library/Frameworks/CoreFoundation.framework/CoreFoundation"
)

# 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_bool
cf.CFStringGetCString.argtypes = [
    CF_TYPE_REF,
    ctypes.c_char_p,
    CF_INDEX,
    CF_STRING_ENCODING,
]
cf.CFGetTypeID.restype = CF_TYPE_ID
cf.CFGetTypeID.argtypes = [CF_TYPE_REF]
cf.CFNumberGetTypeID.restype = CF_TYPE_ID
cf.CFStringGetTypeID.restype = CF_TYPE_ID
cf.CFNumberGetValue.restype = ctypes.c_int
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]
cf.CFRelease.restype = IO_RETURN
cf.CFRelease.argtypes = [CF_TYPE_REF]

iokit.IOObjectRelease.argtypes = [IO_OBJECT_T]

iokit.IOHIDManagerCreate.restype = IO_HID_MANAGER_REF
iokit.IOHIDManagerCreate.argtypes = [CF_ALLOCATOR_REF, IO_OPTION_BITS]
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.IORegistryEntryIDMatching.restype = CF_MUTABLE_DICTIONARY_REF
iokit.IORegistryEntryIDMatching.argtypes = [ctypes.c_uint64]
iokit.IORegistryEntryGetRegistryEntryID.restype = IO_RETURN
iokit.IORegistryEntryGetRegistryEntryID.argtypes = [
    IO_REGISTRY_ENTRY_T,
    ctypes.POINTER(ctypes.c_uint64),
]

iokit.IOHIDDeviceCreate.restype = IO_HID_DEVICE_REF
iokit.IOHIDDeviceCreate.argtypes = [CF_ALLOCATOR_REF, IO_SERVICE_T]
iokit.IOHIDDeviceClose.restype = IO_RETURN
iokit.IOHIDDeviceClose.argtypes = [IO_HID_DEVICE_REF, ctypes.c_uint32]
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.IOHIDDeviceGetProperty.restype = CF_TYPE_REF
iokit.IOHIDDeviceGetProperty.argtypes = [IO_HID_DEVICE_REF, CF_STRING_REF]
iokit.IOHIDDeviceSetReport.restype = IO_RETURN
iokit.IOHIDDeviceSetReport.argtypes = [
    IO_HID_DEVICE_REF,
    IO_HID_REPORT_TYPE,
    CF_INDEX,
    ctypes.c_void_p,
    CF_INDEX,
]
iokit.IOServiceGetMatchingService.restype = IO_SERVICE_T
iokit.IOServiceGetMatchingService.argtypes = [MACH_PORT_T, CF_DICTIONARY_REF]


def _hid_read_callback(
    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

    read_queue.put(ctypes.string_at(report, report_length))


# C wrapper around ReadCallback()
# Declared in this scope so it doesn't get GC-ed
REGISTERED_READ_CALLBACK = IO_HID_REPORT_CALLBACK(_hid_read_callback)


def _hid_removal_callback(hid_device, result, sender):
    del result, sender
    cf.CFRunLoopStop(hid_device.run_loop_ref)


REMOVAL_CALLBACK = IO_HID_CALLBACK(_hid_removal_callback)


def _dev_read_thread(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.handle, hid_device.run_loop_ref, K_CF_RUNLOOP_DEFAULT_MODE
    )

    iokit.IOHIDDeviceRegisterRemovalCallback(
        hid_device.handle, REMOVAL_CALLBACK, ctypes.py_object(hid_device)
    )

    # Run the run loop
    run_loop_run_result = cf.CFRunLoopRunInMode(
        K_CF_RUNLOOP_DEFAULT_MODE, 4, True  # Timeout in seconds
    )  # Return after source handled

    # log any unexpected run loop exit
    if run_loop_run_result != K_CF_RUN_LOOP_RUN_HANDLED_SOURCE:
        logger.error("Unexpected run loop exit code: %d", run_loop_run_result)

    # Unschedule from run loop
    iokit.IOHIDDeviceUnscheduleFromRunLoop(
        hid_device.handle, hid_device.run_loop_ref, K_CF_RUNLOOP_DEFAULT_MODE
    )


class MacCtapHidConnection(CtapHidConnection):
    def __init__(self, descriptor):
        self.descriptor = descriptor
        self.handle = _handle_from_path(descriptor.path)

        # Open device
        result = iokit.IOHIDDeviceOpen(self.handle, 0)
        if result != K_IO_RETURN_SUCCESS:
            raise OSError("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

        # Register read callback
        self.in_report_buffer = (ctypes.c_uint8 * descriptor.report_size_in)()
        iokit.IOHIDDeviceRegisterInputReportCallback(
            self.handle,
            self.in_report_buffer,
            self.descriptor.report_size_in,
            REGISTERED_READ_CALLBACK,
            ctypes.py_object(self.read_queue),
        )

    def close(self):
        iokit.IOHIDDeviceRegisterInputReportCallback(
            self.handle,
            self.in_report_buffer,
            self.descriptor.report_size_in,
            ctypes.cast(0, IO_HID_REPORT_CALLBACK),
            None,
        )

    def write_packet(self, packet):
        result = iokit.IOHIDDeviceSetReport(
            self.handle,
            K_IO_HID_REPORT_TYPE_OUTPUT,
            0,
            packet,
            len(packet),
        )

        # Non-zero status indicates failure
        if result != K_IO_RETURN_SUCCESS:
            raise OSError("Failed to write report to device: {}".format(result))

    def read_packet(self):
        read_thread = threading.Thread(target=_dev_read_thread, args=(self,))
        read_thread.start()
        read_thread.join()
        try:
            return self.read_queue.get(False)
        except Empty:
            raise OSError("Failed reading a response")


def get_int_property(dev, key):
    """Reads int property from the HID device."""
    cf_key = cf.CFStringCreateWithCString(None, key, 0)
    type_ref = iokit.IOHIDDeviceGetProperty(dev, cf_key)
    cf.CFRelease(cf_key)
    if not type_ref:
        return None

    if cf.CFGetTypeID(type_ref) != cf.CFNumberGetTypeID():
        raise OSError("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 get_string_property(dev, key):
    """Reads string property from the HID device."""
    cf_key = cf.CFStringCreateWithCString(None, key, 0)
    type_ref = iokit.IOHIDDeviceGetProperty(dev, cf_key)
    cf.CFRelease(cf_key)
    if not type_ref:
        return None

    if cf.CFGetTypeID(type_ref) != cf.CFStringGetTypeID():
        raise OSError("Expected string type, got {}".format(cf.CFGetTypeID(type_ref)))

    out = ctypes.create_string_buffer(128)
    ret = cf.CFStringGetCString(
        type_ref, out, ctypes.sizeof(out), CF_STRING_BUILTIN_ENCODINGS_UTF8
    )
    if not ret:
        return None

    try:
        return out.value.decode("utf-8") or None
    except UnicodeDecodeError:
        return None


def get_device_id(handle):
    """Obtains the unique IORegistry entry ID for the device.

    Args:
    handle: reference to the device

    Returns:
    A unique ID for the device, obtained from the IO Registry
    """
    # Obtain device entry ID from IO Registry
    io_service_obj = iokit.IOHIDDeviceGetService(handle)
    entry_id = ctypes.c_uint64()
    result = iokit.IORegistryEntryGetRegistryEntryID(
        io_service_obj, ctypes.byref(entry_id)
    )
    if result != K_IO_RETURN_SUCCESS:
        raise OSError("Failed to obtain IORegistry entry ID: {}".format(result))

    return entry_id.value


def _handle_from_path(path):
    # Resolve the path to device handle
    entry_id = ctypes.c_uint64(int(path))
    matching_dict = iokit.IORegistryEntryIDMatching(entry_id)
    device_entry = iokit.IOServiceGetMatchingService(
        K_IO_MASTER_PORT_DEFAULT, matching_dict
    )
    if not device_entry:
        raise OSError(
            "Device ID {} does not match any HID device on the system".format(path)
        )

    return iokit.IOHIDDeviceCreate(K_CF_ALLOCATOR_DEFAULT, device_entry)


def open_connection(descriptor):
    return MacCtapHidConnection(descriptor)


def _get_descriptor_from_handle(handle):
    usage_page = get_int_property(handle, HID_DEVICE_PROPERTY_PRIMARY_USAGE_PAGE)
    usage = get_int_property(handle, HID_DEVICE_PROPERTY_PRIMARY_USAGE)
    if usage_page == FIDO_USAGE_PAGE and usage == FIDO_USAGE:
        device_id = get_device_id(handle)
        vid = get_int_property(handle, HID_DEVICE_PROPERTY_VENDOR_ID)
        pid = get_int_property(handle, HID_DEVICE_PROPERTY_PRODUCT_ID)
        product = get_string_property(handle, HID_DEVICE_PROPERTY_PRODUCT)
        serial = get_string_property(handle, HID_DEVICE_PROPERTY_SERIAL_NUMBER)
        size_in = get_int_property(handle, HID_DEVICE_PROPERTY_MAX_INPUT_REPORT_SIZE)
        size_out = get_int_property(handle, HID_DEVICE_PROPERTY_MAX_OUTPUT_REPORT_SIZE)
        return HidDescriptor(
            str(device_id), vid, pid, size_in, size_out, product, serial
        )
    raise ValueError("Not a CTAP device")


def get_descriptor(path):
    return _get_descriptor_from_handle(_handle_from_path(path))


def list_descriptors():
    # Init a HID manager
    hid_mgr = iokit.IOHIDManagerCreate(None, 0)
    if not hid_mgr:
        raise OSError("Unable to obtain HID manager reference")
    try:
        iokit.IOHIDManagerSetDeviceMatching(hid_mgr, None)

        # Get devices from HID manager
        device_set_ref = iokit.IOHIDManagerCopyDevices(hid_mgr)
        if not device_set_ref:
            raise OSError("Failed to obtain devices from HID manager")
        try:
            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 handle in devices:
                try:
                    descriptor = _get_descriptor_from_handle(handle)
                    descriptors.append(descriptor)
                    logger.debug("Found CTAP device: %s", descriptor.path)
                except ValueError:
                    continue  # Not a CTAP device, ignore it
            return descriptors
        finally:
            cf.CFRelease(device_set_ref)
    finally:
        cf.CFRelease(hid_mgr)