From 79c1bcb178367d596f75497618d124f3cd30eee4 Mon Sep 17 00:00:00 2001 From: Pol Henarejos Date: Wed, 26 Nov 2025 12:49:46 +0100 Subject: [PATCH] Add logger and debug levels in PICOKEY_LOG environ. Signed-off-by: Pol Henarejos --- picokey/PicoKey.py | 53 +++++++++++++++++++++++++++++++++++- picokey/RescuePicoKey.py | 33 ++++++++++++++++++++-- picokey/core/log.py | 59 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 142 insertions(+), 3 deletions(-) create mode 100644 picokey/core/log.py diff --git a/picokey/PicoKey.py b/picokey/PicoKey.py index c6bef14..5908fc1 100644 --- a/picokey/PicoKey.py +++ b/picokey/PicoKey.py @@ -26,6 +26,9 @@ from .RescueMonitor import RescueMonitor, RescueMonitorObserver from .PhyData import PhyData from .core import NamedIntEnum import usb.core +from .core.log import get_logger + +logger = get_logger("PicoKey") class Platform(NamedIntEnum): RP2040 = 0 @@ -73,6 +76,11 @@ def connect_with_timeout(connection, timeout=2.0): class PicoKey: def __init__(self, slot=-1, force_rescue=False): + logger.debug("Initializing PicoKey...") + self.__apdu = [] + self.__connection_type = ConnectionType.UNKNOWN + self.__monitor = None + self.__observer = None self.__sc = None self.__card = None @@ -80,7 +88,7 @@ class PicoKey: from smartcard.System import readers import smartcard.Exceptions from smartcard.CardMonitoring import CardMonitor, CardObserver - + logger.debug("Searching for smartcard readers...") class PicoCardObserver(CardObserver): def __init__(self, device): self.__device = device @@ -101,18 +109,23 @@ class PicoKey: return None return None + logger.debug("Checking available smartcard readers...") try: rdrs = readers() except Exception as e: + logger.error("Error accessing smartcard readers: " + str(e)) rdrs = [] if len(rdrs) > 0: if (slot >= 0 and slot >= len(rdrs)): + logger.error("Slot number out of range") raise Exception('Slot number out of range') if (slot >= 0 and slot < len(rdrs)): + logger.debug(f"Checking reader slot {slot}") reader = rdrs[slot] connection = reader_has_card(reader) if (connection is None): + logger.error(f"No card in reader slot {slot}") raise Exception(f'No card in reader slot {slot}') self.__card = connection else: @@ -121,15 +134,21 @@ class PicoKey: connection = reader_has_card(reader) if (connection is None): continue + logger.debug(f"Card found in reader slot {i}") self.__card = connection self.__connection_type = ConnectionType.SMARTCARD + logger.debug("Setting up card monitor...") self.__monitor = CardMonitor() + logger.debug("Creating card observer...") self.__observer = PicoCardObserver(self) + logger.debug("Adding observer to monitor...") self.__monitor.addObserver(self.__observer) + logger.debug("Observer added to monitor") break if (self.__card is None): + logger.debug("Attempting to connect in rescue mode...") class PicoRescueObserver(RescueMonitorObserver): def __init__(self, device): self.__device = device @@ -142,18 +161,25 @@ class PicoKey: self.__device.close() try: self.__card = RescuePicoKey() + logger.debug("Rescue mode card initialized") self.__connection_type = ConnectionType.RESCUE + logger.debug("Setting up rescue monitor...") self.__observer = PicoRescueObserver(self) + logger.debug("Creating rescue monitor...") self.__monitor = RescueMonitor(device=self.__card, cls_callback=self.__observer) except Exception: + logger.error("No PicoKey device detected") raise Exception('time-out: no card inserted') try: + logger.debug("Selecting applet in rescue mode...") resp, sw1, sw2 = self.select_applet(rescue=True) + logger.debug(f"Applet selected with response code: 0x{sw1:02X}{sw2:02X}") if (sw1 == 0x90 and sw2 == 0x00): self.platform = Platform(resp[0]) self.product = Product(resp[1]) self.version = (resp[2], resp[3]) except APDUResponse: + logger.error("APDU response error during applet selection") self.platform = Platform(Platform.RP2040) self.product = Product(Product.UNKNOWN) self.version = (0, 0) @@ -170,26 +196,38 @@ class PicoKey: return self.__connection_type def close(self): + logger.debug("Closing device...") if (not self.__card): + logger.debug("No device to close") return if isinstance(self.__card, RescuePicoKey): + logger.debug("Stopping rescue monitor...") self.__monitor.stop() self.__monitor = None self.__observer = None self.__card.close() else: + logger.debug("Removing card monitor observer...") + self.__monitor.deleteObserver(self.__observer) + self.__observer = None + self.__monitor = None + logger.debug("Disconnecting and releasing card...") self.__card.disconnect() + logger.debug("Card disconnected") self.__card.release() self.__card = None def transmit(self, apdu: list[int]): if (not self.__card): + logger.error("No device connected") raise Exception('No device connected') response, sw1, sw2 = self.__card.transmit(apdu) return response, sw1, sw2 def send(self, command: int, cla: int = 0x00, p1: int =0x00, p2: int=0x00, ne : Optional[int] = None, data : Optional[list[int]] = None, codes : list[int] = []): + logger.debug(f"Sending command {hex(command)} with cla={hex(cla)}, p1={hex(p1)}, p2={hex(p2)}, ne={ne}") if (not self.__card): + logger.error("No device connected") raise Exception('No device connected') lc = [] dataf = [] @@ -209,16 +247,20 @@ class PicoKey: apdu = apdu + [p1, p2] + lc + dataf + le self.__apdu = apdu + logger.trace(f"APDU -> {' '.join([f'{x:02X}' for x in apdu])}") if (self.__sc): apdu = self.__sc.wrap_apdu(apdu) + logger.trace(f"Wrapped APDU -> {' '.join([f'{x:02X}' for x in apdu])}") try: response, sw1, sw2 = self.__card.transmit(apdu) except Exception: + logger.debug("Reconnecting card after transmit failure") self.__card.reconnect() try: response, sw1, sw2 = self.__card.transmit(apdu) except Exception as e: + logger.error("APDU transmission error after reconnect: " + str(e)) raise Exception("APDU transmission error after reconnect: " + str(e)) code = (sw1<<8|sw2) @@ -240,8 +282,10 @@ class PicoKey: code = (sw1<<8|sw2) if (code not in codes and code != 0x9000): raise APDUResponse(sw1, sw2) + logger.trace(f"Response APDU <- {' '.join([f'{x:02X}' for x in response])}, SW1={sw1:02X}, SW2={sw2:02X}") if (self.__sc): response, code = self.__sc.unwrap_rapdu(response) + logger.trace(f"Unwrapped RAPDU <- {' '.join([f'{x:02X}' for x in response])}, Code={code:04X}") if (code not in codes and code != 0x9000): raise APDUResponse(code >> 8, code & 0xff) return bytes(response), code @@ -254,12 +298,14 @@ class PicoKey: try: response, sw1, sw2 = self.__card.transmit(apdu) except Exception: + logger.debug("Reconnecting card after transmit failure") self.__card.reconnect() response, sw1, sw2 = self.__card.transmit(apdu) return bytes(response), sw1, sw2 def open_secure_channel(self, shared: bytes, nonce: bytes, token: bytes, pbkeyBytes: bytes): + logger.debug("Opening secure channel") sc = SecureChannel(shared=shared, nonce=nonce) res = sc.verify_token(token, pbkeyBytes) if (not res): @@ -268,7 +314,9 @@ class PicoKey: def select_applet(self, rescue : bool = False): if (rescue): + logger.debug("Selecting rescue applet") return self.transmit([0x00, 0xA4, 0x04, 0x04, 0x08, 0xA0, 0x58, 0x3F, 0xC1, 0x9B, 0x7E, 0x4F, 0x21, 0x00]) + logger.debug("Selecting standard applet") return self.transmit([0x00, 0xA4, 0x04, 0x00, 0x0B, 0xE8, 0x2B, 0x06, 0x01, 0x04, 0x01, 0x81, 0xC3, 0x1F, 0x02, 0x01, 0x00]) def phy(self, data : Optional[list[int]] = None): @@ -283,6 +331,7 @@ class PicoKey: self.send(0x1C, cla=0x80, p1=0x01, data=data) def flash_info(self): + logger.debug("Retrieving flash info") try: resp, sw = self.send(0x1E, cla=0x80, p1=0x02) free = int.from_bytes(resp[0:4], 'big') @@ -301,6 +350,7 @@ class PicoKey: } def secure_info(self): + logger.debug("Retrieving secure boot info") resp, sw = self.send(0x1E, cla=0x80, p1=0x03) return { 'enabled': resp[0] != 0, @@ -309,5 +359,6 @@ class PicoKey: } def secure_boot(self, bootkey_index: int = 0, lock: bool = False): + logger.debug(f"Setting secure boot: bootkey_index={bootkey_index}, lock={lock}") data = bytes([bootkey_index & 0xFF, 1 if lock else 0]) self.send(0x1C, cla=0x80, p1=0x02, data=data) diff --git a/picokey/RescuePicoKey.py b/picokey/RescuePicoKey.py index 3b66f06..463e07d 100644 --- a/picokey/RescuePicoKey.py +++ b/picokey/RescuePicoKey.py @@ -17,16 +17,19 @@ */ """ -import os import usb.core import usb.util import libusb_package import usb.backend.libusb1 from .ICCD import ICCD +from .core.log import get_logger + +logger = get_logger("RescuePicoKey") class RescuePicoKey: def __init__(self): + logger.debug("Initializing RescuePicoKey...") self.__dev = None self.__in = None self.__out = None @@ -45,16 +48,20 @@ class RescuePicoKey: return True return False + logger.debug("Searching for USB device...") backend = usb.backend.libusb1.get_backend(find_library=libusb_package.find_library) try: devs = usb.core.find(find_all=True, custom_match=find_class(0x0B), backend=backend) except Exception as e: - print("RescuePicoKey: exception during usb.core.find:", e) + logger.error("Exception during usb.core.find: %s", e) devs = [] found = False for dev in devs: if (dev.manufacturer == 'Pol Henarejos'): + logger.debug("Found device") dev.set_configuration() + logger.debug("Device configuration set") + logger.debug("Getting active configuration...") cfg = dev.get_active_configuration() for intf in cfg: if (intf.bInterfaceClass == 0xFF): @@ -71,12 +78,17 @@ class RescuePicoKey: self.__in = epin.bEndpointAddress self.__out = epout[0].bEndpointAddress self.__int = epint.bEndpointAddress if epint else None + logger.debug(f"Endpoints - IN: 0x{self.__in:02X}, OUT: 0x{self.__out:02X}, INT: {self.__int}") self.__iccd = ICCD(self) + logger.debug("ICCD interface initialized") self.__active = None + logger.debug("Powering off device") self.powerOff() + logger.debug("Device powered off") found = True break if (not found): + logger.error("No suitable device found") raise Exception('Not found any Pico Key device') @property @@ -84,8 +96,11 @@ class RescuePicoKey: return self.__dev def close(self): + logger.debug("Closing device") if self.__dev: + logger.debug("Disposing USB resources") usb.util.dispose_resources(self.__dev) + logger.debug("Device closed") self.__dev = None def has_card(self): @@ -101,31 +116,45 @@ class RescuePicoKey: return str(self.__dev) def read(self, timeout=2000): + logger.debug("Reading data from device") ret = self.__dev.read(self.__in, 4096, timeout) + logger.trace(f"Data read from device: {' '.join([f'{x:02X}' for x in ret])}") + logger.debug("Read data from device") return ret def write(self, data, timeout=2000): + logger.debug("Writing data to device") + logger.trace(f"Data to write to device: {' '.join([f'{x:02X}' for x in data])}") assert(self.__dev.write(self.__out, data, timeout) == len(data)) + logger.debug("Wrote data to device") def exchange(self, data, timeout=2000): + logger.debug("Exchanging data with device") try: self.write(data=data, timeout=timeout) except Exception as e: + logger.error("USB write error: " + str(e)) raise Exception("USB write error: " + str(e)) try: ret = self.read(timeout=timeout) except Exception as e: + logger.error("USB read error: " + str(e)) raise Exception("USB read error: " + str(e)) return ret def powerOn(self): + logger.debug("Powering on device") if (not self.__active): self.__active = True + logger.debug("Device powered on") return self.__iccd.IccPowerOn() def powerOff(self): + logger.debug("Powering off device") if (self.__active or self.__active is None): + logger.debug("Device powered off") self.__iccd.IccPowerOff() + logger.debug("ICCD powered off") self.__active = False def transmit(self, apdu): diff --git a/picokey/core/log.py b/picokey/core/log.py new file mode 100644 index 0000000..3071912 --- /dev/null +++ b/picokey/core/log.py @@ -0,0 +1,59 @@ +""" +/* + * This file is part of the pypicokey distribution (https://github.com/polhenarejos/pypicokey). + * Copyright (c) 2025 Pol Henarejos. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published by + * the Free Software Foundation, version 3. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +""" + +import os +import logging + +TRACE_LEVEL = 5 +logging.addLevelName(TRACE_LEVEL, "TRACE") + +def trace(self, message, *args, **kwargs): + if self.isEnabledFor(TRACE_LEVEL): + self._log(TRACE_LEVEL, message, args, **kwargs) + +# Afegeix logger.trace() +logging.Logger.trace = trace + +def get_logger(name: str): + env_level = os.getenv("PICOKEY_LOG", "CRITICAL").upper() + + valid_levels = { + "TRACE": TRACE_LEVEL, + "DEBUG": logging.DEBUG, + "INFO": logging.INFO, + "WARNING": logging.WARNING, + "ERROR": logging.ERROR, + "CRITICAL": logging.CRITICAL, + } + + level = valid_levels.get(env_level, logging.CRITICAL) + + logger = logging.getLogger(name) + logger.setLevel(level) + + if not logger.handlers: + handler = logging.StreamHandler() + handler.setFormatter(logging.Formatter( + fmt="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S" + )) + logger.addHandler(handler) + + return logger +