Compare commits
19 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| b085719f98 | |||
| 1aae58a2f1 | |||
| 19ffb414de | |||
| cd5fd56a1d | |||
| 7c2f454592 | |||
| a008e506c7 | |||
| 79c1bcb178 | |||
| e2d984adb0 | |||
| 872aca7c83 | |||
| 1658eb92b3 | |||
| 94ef650f80 | |||
| 74fe60f2d4 | |||
| e10b70df4e | |||
| d6d4489142 | |||
| ef80f8f271 | |||
| 1fe46149b9 | |||
| e7d92fba8a | |||
| 6b1b0f479f | |||
| e5c3e9c824 |
+220
-38
@@ -17,12 +17,19 @@
|
||||
*/
|
||||
"""
|
||||
|
||||
import sys
|
||||
from typing import Optional
|
||||
import threading
|
||||
from .APDU import APDUResponse
|
||||
from .SecureChannel import SecureChannel
|
||||
from .RescuePicoKey import RescuePicoKey
|
||||
from .RescueMonitor import RescueMonitor, RescueMonitorObserver
|
||||
from .PhyData import PhyData
|
||||
from .core import NamedIntEnum
|
||||
from .core.exceptions import PicoKeyNotFoundError, PicoKeyInvalidStateError
|
||||
import usb.core
|
||||
from .core.log import get_logger
|
||||
|
||||
logger = get_logger("PicoKey")
|
||||
|
||||
class Platform(NamedIntEnum):
|
||||
RP2040 = 0
|
||||
@@ -36,54 +43,203 @@ class Product(NamedIntEnum):
|
||||
FIDO = 2
|
||||
OPENPGP = 3
|
||||
|
||||
class ConnectionType(NamedIntEnum):
|
||||
UNKNOWN = 0
|
||||
SMARTCARD = 1
|
||||
RESCUE = 2
|
||||
|
||||
class ConnectTimeout(Exception):
|
||||
pass
|
||||
|
||||
def connect_with_timeout(connection, timeout=2.0):
|
||||
result = {}
|
||||
|
||||
def worker():
|
||||
try:
|
||||
connection.connect()
|
||||
result["ok"] = True
|
||||
except Exception as e:
|
||||
result["error"] = e
|
||||
|
||||
t = threading.Thread(target=worker)
|
||||
t.daemon = True
|
||||
t.start()
|
||||
t.join(timeout)
|
||||
|
||||
if t.is_alive():
|
||||
raise ConnectTimeout("connection.connect() timed out")
|
||||
|
||||
if "error" in result:
|
||||
raise result["error"]
|
||||
|
||||
return True
|
||||
|
||||
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
|
||||
if (force_rescue):
|
||||
self.__card = None
|
||||
|
||||
if (not force_rescue):
|
||||
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
|
||||
|
||||
def update(self, observable, actions):
|
||||
(added, removed) = actions
|
||||
if added:
|
||||
pass
|
||||
if removed:
|
||||
self.__device.close()
|
||||
|
||||
def reader_has_card(reader):
|
||||
try:
|
||||
connection = reader.createConnection()
|
||||
connect_with_timeout(connection, timeout=1.0)
|
||||
return connection
|
||||
except smartcard.Exceptions.NoCardException:
|
||||
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:
|
||||
for i, reader in enumerate(rdrs):
|
||||
reader = rdrs[i]
|
||||
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
|
||||
|
||||
def update(self, actions: tuple[Optional[usb.core.Device], Optional[usb.core.Device]]):
|
||||
(connected, disconnected) = actions
|
||||
if connected:
|
||||
logger.debug("Observer: Rescue device connected")
|
||||
pass
|
||||
if disconnected:
|
||||
logger.debug("Observer: Rescue device disconnected, closing...")
|
||||
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:
|
||||
raise Exception('time-out: no card inserted')
|
||||
else:
|
||||
from smartcard.CardType import AnyCardType
|
||||
from smartcard.CardRequest import CardRequest
|
||||
cardtype = AnyCardType()
|
||||
try:
|
||||
# request card insertion
|
||||
readers = None
|
||||
if (slot >= 0):
|
||||
readers = CardRequest().getReaders()
|
||||
if (slot >= len(readers)):
|
||||
raise Exception('slot out of range')
|
||||
readers = [readers[slot]]
|
||||
cardrequest = CardRequest(timeout=1, cardType=cardtype, readers=readers)
|
||||
self.__card = cardrequest.waitforcard().connection
|
||||
|
||||
# connect to the card and perform a few transmits
|
||||
self.__card.connect()
|
||||
|
||||
except Exception:
|
||||
try:
|
||||
self.__card = RescuePicoKey()
|
||||
except Exception:
|
||||
raise Exception('time-out: no card inserted')
|
||||
resp, sw1, sw2 = self.select_applet(rescue=True)
|
||||
logger.error("No PicoKey device detected")
|
||||
raise PicoKeyNotFoundError('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)
|
||||
|
||||
def transmit(self, apdu):
|
||||
response, sw1, sw2 = self.__card.transmit(apdu)
|
||||
return response, sw1, sw2
|
||||
@property
|
||||
def device(self):
|
||||
return self.__card
|
||||
|
||||
def send(self, command, cla=0x00, p1=0x00, p2=0x00, ne=None, data=None, codes=[]):
|
||||
def has_device(self):
|
||||
return self.__card is not None
|
||||
|
||||
@property
|
||||
def connection_type(self):
|
||||
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...")
|
||||
if (self.__monitor and self.__observer):
|
||||
self.__monitor.deleteObserver(self.__observer)
|
||||
self.__observer = None
|
||||
self.__monitor = None
|
||||
logger.debug("Disconnecting and releasing card...")
|
||||
try:
|
||||
self.__card.disconnect()
|
||||
logger.debug("Card disconnected")
|
||||
self.__card.release()
|
||||
except Exception as e:
|
||||
logger.error("Error during card disconnect/release: " + str(e))
|
||||
self.__card = None
|
||||
|
||||
def transmit(self, apdu: list[int]):
|
||||
if (not self.__card):
|
||||
logger.error("No device connected")
|
||||
raise PicoKeyNotFoundError('No device connected')
|
||||
try:
|
||||
response, sw1, sw2 = self.__card.transmit(apdu)
|
||||
return response, sw1, sw2
|
||||
except Exception as e:
|
||||
logger.error("Transmission error: " + str(e))
|
||||
raise PicoKeyInvalidStateError("Transmission error: " + str(e))
|
||||
|
||||
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 PicoKeyNotFoundError('No device connected')
|
||||
lc = []
|
||||
dataf = []
|
||||
if (data):
|
||||
@@ -102,14 +258,27 @@ 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:
|
||||
self.__card.reconnect()
|
||||
response, sw1, sw2 = self.__card.transmit(apdu)
|
||||
logger.debug("Reconnecting card after transmit failure")
|
||||
|
||||
try:
|
||||
self.__card.reconnect()
|
||||
except Exception as e:
|
||||
logger.error("Reconnection failed: " + str(e))
|
||||
self.close()
|
||||
raise PicoKeyNotFoundError('Reconnection failed: ' + str(e))
|
||||
try:
|
||||
response, sw1, sw2 = self.__card.transmit(apdu)
|
||||
except Exception as e:
|
||||
logger.error("APDU transmission error after reconnect: " + str(e))
|
||||
raise PicoKeyInvalidStateError("APDU transmission error after reconnect: " + str(e))
|
||||
|
||||
code = (sw1<<8|sw2)
|
||||
if (sw1 != 0x90):
|
||||
@@ -130,8 +299,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
|
||||
@@ -144,35 +315,40 @@ 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, nonce, token, pbkeyBytes):
|
||||
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):
|
||||
raise Exception('Secure Channel token verification failed')
|
||||
self.__sc = sc
|
||||
|
||||
def select_applet(self, rescue=False):
|
||||
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=None):
|
||||
def phy(self, data : Optional[list[int]] = None):
|
||||
if (data is None):
|
||||
try:
|
||||
resp, sw = self.send(0x1E, cla=0x80, p1=0x00, ne=256)
|
||||
return PhyData.parse(resp)
|
||||
except APDUResponse:
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
else:
|
||||
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')
|
||||
@@ -191,6 +367,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,
|
||||
@@ -198,6 +375,11 @@ class PicoKey:
|
||||
'boot_key': resp[2]
|
||||
}
|
||||
|
||||
def secure_boot(self, bootkey_index=0, lock=False):
|
||||
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)
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
"""
|
||||
/*
|
||||
* 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/>.
|
||||
*/
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
import usb.core
|
||||
import threading
|
||||
import time
|
||||
from .RescuePicoKey import RescuePicoKey
|
||||
|
||||
class RescueMonitorObserver:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def notifyObservers(self, actions: tuple[Optional[usb.core.Device], Optional[usb.core.Device]]):
|
||||
func = getattr(self, "update", None)
|
||||
if callable(func):
|
||||
func(actions)
|
||||
|
||||
def on_connect(self, device: Optional[usb.core.Device]):
|
||||
self.notifyObservers((device, None))
|
||||
|
||||
def on_disconnect(self, device: Optional[usb.core.Device]):
|
||||
self.notifyObservers((None, device))
|
||||
|
||||
class RescueMonitor:
|
||||
def __init__(self, device: RescuePicoKey, cls_callback: RescueMonitorObserver, interval=0.5):
|
||||
self._dev = device
|
||||
self._cls_callback = cls_callback
|
||||
self.interval = interval
|
||||
self._running = False
|
||||
self._device_present = False
|
||||
self._thread = None
|
||||
self.start()
|
||||
|
||||
def start(self):
|
||||
if self._running:
|
||||
return
|
||||
self._running = True
|
||||
self._thread = threading.Thread(target=self._run, daemon=True)
|
||||
self._thread.start()
|
||||
|
||||
def stop(self):
|
||||
self._running = False
|
||||
#if self._thread:
|
||||
# self._thread.join()
|
||||
|
||||
def _run(self):
|
||||
while self._running:
|
||||
if (self._dev is None) or (self._dev.device is None):
|
||||
time.sleep(self.interval)
|
||||
continue
|
||||
dev = usb.core.find(idVendor=self._dev.device.idVendor, idProduct=self._dev.device.idProduct)
|
||||
|
||||
if dev and not self._device_present:
|
||||
# Device connected
|
||||
self._device_present = True
|
||||
if self._cls_callback:
|
||||
self._cls_callback.on_connect(dev)
|
||||
|
||||
if not dev and self._device_present:
|
||||
# Device disconnected
|
||||
self._device_present = False
|
||||
if self._cls_callback:
|
||||
self._cls_callback.on_disconnect(self._dev.device)
|
||||
|
||||
time.sleep(self.interval)
|
||||
+97
-19
@@ -17,14 +17,34 @@
|
||||
*/
|
||||
"""
|
||||
|
||||
import os
|
||||
import time
|
||||
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 RescuePicoKeyError(Exception):
|
||||
pass
|
||||
|
||||
class RescuePicoKeyNotFoundError(RescuePicoKeyError):
|
||||
pass
|
||||
|
||||
class RescuePicoKeyInvalidStateError(RescuePicoKeyError):
|
||||
pass
|
||||
|
||||
class RescuePicoKey:
|
||||
|
||||
def __init__(self):
|
||||
logger.debug("Initializing RescuePicoKey...")
|
||||
self.__dev = None
|
||||
self.__in = None
|
||||
self.__out = None
|
||||
self.__int = None
|
||||
self.__active = None
|
||||
|
||||
class find_class(object):
|
||||
def __init__(self, class_):
|
||||
@@ -37,21 +57,21 @@ class RescuePicoKey:
|
||||
if intf is not None:
|
||||
return True
|
||||
return False
|
||||
if (os.name == 'nt'):
|
||||
import libusb_package
|
||||
import usb.backend.libusb1
|
||||
backend = usb.backend.libusb1.get_backend(find_library=libusb_package.find_library)
|
||||
else:
|
||||
backend = None
|
||||
|
||||
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):
|
||||
@@ -68,46 +88,91 @@ 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
|
||||
try:
|
||||
self.powerOff()
|
||||
except Exception as e:
|
||||
print("RescuePicoKey: exception during power off:", e)
|
||||
logger.debug("Powering off device")
|
||||
self.powerOff()
|
||||
logger.debug("Device powered off")
|
||||
found = True
|
||||
break
|
||||
if (not found):
|
||||
raise Exception('Not found any Pico Key device')
|
||||
logger.error("No suitable device found")
|
||||
raise RescuePicoKeyNotFoundError('Not found any Pico Key device')
|
||||
|
||||
@property
|
||||
def device(self):
|
||||
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):
|
||||
return self.__dev is not None
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
self.close()
|
||||
|
||||
def __del__(self):
|
||||
self.close()
|
||||
|
||||
def __str__(self):
|
||||
return str(self.__dev)
|
||||
|
||||
def read(self, timeout=2000):
|
||||
ret = self.__dev.read(self.__in, 4096, timeout)
|
||||
return ret
|
||||
logger.debug("Reading data from device")
|
||||
try:
|
||||
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
|
||||
except Exception as e:
|
||||
logger.error("USB read error: " + str(e))
|
||||
raise RescuePicoKeyInvalidStateError("USB read error: " + str(e))
|
||||
|
||||
def write(self, data, timeout=2000):
|
||||
assert(self.__dev.write(self.__out, data, timeout) == len(data))
|
||||
logger.debug("Writing data to device")
|
||||
logger.trace(f"Data to write to device: {' '.join([f'{x:02X}' for x in data])}")
|
||||
try:
|
||||
assert(self.__dev.write(self.__out, data, timeout) == len(data))
|
||||
logger.debug("Wrote data to device")
|
||||
except Exception as e:
|
||||
logger.error("USB write error: " + str(e))
|
||||
raise RescuePicoKeyInvalidStateError("USB write error: " + str(e))
|
||||
|
||||
def exchange(self, data, timeout=2000):
|
||||
logger.debug("Exchanging data with device")
|
||||
try:
|
||||
self.write(data=data, timeout=timeout)
|
||||
except Exception as e:
|
||||
raise Exception("USB write error: " + str(e))
|
||||
logger.error("USB write error: " + str(e))
|
||||
raise RescuePicoKeyInvalidStateError("USB write error: " + str(e))
|
||||
try:
|
||||
ret = self.read(timeout=timeout)
|
||||
except Exception as e:
|
||||
raise Exception("USB read error: " + str(e))
|
||||
logger.error("USB read error: " + str(e))
|
||||
raise RescuePicoKeyInvalidStateError("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):
|
||||
@@ -115,3 +180,16 @@ class RescuePicoKey:
|
||||
self.powerOn()
|
||||
rapdu = self.__iccd.SendApdu(apdu=apdu)
|
||||
return rapdu[:-2], rapdu[-2], rapdu[-1]
|
||||
|
||||
def reconnect(self):
|
||||
logger.debug("Reconnecting to device")
|
||||
self.close()
|
||||
time.sleep(1)
|
||||
try:
|
||||
self.__init__()
|
||||
except Exception as e:
|
||||
logger.error("Reconnection failed: %s", e)
|
||||
self.close()
|
||||
raise e
|
||||
logger.debug("Reconnected to device")
|
||||
return self
|
||||
|
||||
@@ -6,3 +6,4 @@ from .SWCodes import SWCodes
|
||||
from .RescuePicoKey import RescuePicoKey
|
||||
from .PhyData import PhyData, PhyCurve, PhyUsbItf, PhyLedDriver, PhyOpt
|
||||
from .core import enums
|
||||
from .core.exceptions import PicoKeyError, PicoKeyNotFoundError, PicoKeyInvalidStateError
|
||||
|
||||
+1
-1
@@ -18,4 +18,4 @@
|
||||
*/
|
||||
"""
|
||||
|
||||
__version__ = "1.1.3"
|
||||
__version__ = "1.3"
|
||||
|
||||
@@ -1 +1,2 @@
|
||||
from .enums import *
|
||||
from .exceptions import *
|
||||
|
||||
@@ -1,5 +1,44 @@
|
||||
"""
|
||||
/*
|
||||
* 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 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
|
||||
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
"""
|
||||
/*
|
||||
* 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/>.
|
||||
*/
|
||||
"""
|
||||
|
||||
class PicoKeyError(Exception):
|
||||
pass
|
||||
|
||||
class PicoKeyNotFoundError(PicoKeyError):
|
||||
pass
|
||||
|
||||
class PicoKeyInvalidStateError(PicoKeyError):
|
||||
pass
|
||||
@@ -0,0 +1,58 @@
|
||||
"""
|
||||
/*
|
||||
* 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
|
||||
@@ -0,0 +1,51 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=61.0"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "pypicokey"
|
||||
dynamic = ["version", "readme"]
|
||||
description = "PicoKey for Python"
|
||||
requires-python = ">=3.8"
|
||||
license = { file = "LICENSE" }
|
||||
|
||||
authors = [
|
||||
{ name = "Pol Henarejos", email = "pol.henarejos@cttc.es" }
|
||||
]
|
||||
|
||||
dependencies = [
|
||||
"setuptools",
|
||||
"cryptography>=3.3",
|
||||
"pyusb",
|
||||
"pycvc",
|
||||
"pyscard>=2.3.1",
|
||||
"libusb",
|
||||
"libusb_package",
|
||||
]
|
||||
|
||||
classifiers = [
|
||||
"Development Status :: 5 - Production/Stable",
|
||||
"Environment :: Plugins",
|
||||
"Intended Audience :: System Administrators",
|
||||
"License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
|
||||
"Programming Language :: Python :: 3.8",
|
||||
"Topic :: Security",
|
||||
"Topic :: System :: Installation/Setup",
|
||||
"Topic :: System :: Networking",
|
||||
"Topic :: System :: Systems Administration",
|
||||
"Topic :: Utilities",
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://github.com/polhenarejos/pypicokey"
|
||||
|
||||
[tool.setuptools]
|
||||
packages = [
|
||||
"picokey",
|
||||
"picokey.core"
|
||||
]
|
||||
include-package-data = true
|
||||
|
||||
[tool.setuptools.dynamic]
|
||||
version = { attr = "picokey._version.__version__" }
|
||||
readme = { file = "README.md" }
|
||||
Reference in New Issue
Block a user