Version 1.2.1

Signed-off-by: Pol Henarejos <pol.henarejos@cttc.es>

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,15 +109,23 @@ class PicoKey:
                     return None
                 return None
 
-            rdrs = readers()
+            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:
@@ -118,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
@@ -139,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)
@@ -167,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 = []
@@ -206,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)
@@ -237,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
@@ -251,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):
@@ -265,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):
@@ -280,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')
@@ -298,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,
@@ -306,5 +359,10 @@ 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)
+
+    def reboot(self, bootsel: bool = False):
+        logger.debug("Rebooting device into BOOTSEL mode" if bootsel else "Rebooting device into normal mode")
+        self.send(0x1F, cla=0x80, p1=0x01 if bootsel else 0x00)

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):

picokey/_version.py

@@ -18,4 +18,4 @@
  */
 """
 
-__version__ = "1.1.7"
+__version__ = "1.2.1"

picokey/core/enums.py

@@ -1,5 +1,25 @@
 import enum
+from typing import Union
 
 class NamedIntEnum(enum.IntEnum):
     def __str__(self):
         return self.name
+
+    def __format__(self, fmt):
+        if any(c in fmt for c in "xXod"):
+            return format(self.value, fmt)
+        return self.name
+
+    @classmethod
+    def from_string(cls, value: Union[str, int]) -> "NamedIntEnum":
+        if not value:
+            return cls.UNKNOWN
+
+        value = value.strip().lower()
+
+        for member in cls:
+            if member.value == value or member.name.lower() == value:
+                return member
+
+        return cls.UNKNOWN
+

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 <https://www.gnu.org/licenses/>.
+ */
+"""
+
+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
+