Compare commits
14 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 577de21fbe | |||
| 823f2e32a0 | |||
| 71ff27f099 | |||
| 85639fc8e3 | |||
| d6f6ddbafd | |||
| 4b41ebc243 | |||
| eb6d371354 | |||
| f8b79e6313 | |||
| b42064a73e | |||
| b085719f98 | |||
| 1aae58a2f1 | |||
| 19ffb414de | |||
| cd5fd56a1d | |||
| 7c2f454592 |
@@ -1,6 +1,9 @@
|
||||
# pypicokey
|
||||
PicoKey tools for Python
|
||||
|
||||
Forked From https://github.com/polhenarejos/pypicokey,
|
||||
If fork is unproper please tell me and i'd delete it.
|
||||
|
||||
## Introduction
|
||||
|
||||
PicoKey firmware allows to convert a Raspberry Pico into a Hardware Security Module (HSM), FIDO2 device or OpenPGP card, to store private and secret keys, perform signing and ciphering operations, without exposing the key.
|
||||
|
||||
+323
@@ -0,0 +1,323 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Interactive commissioning helper for PicoKey
|
||||
|
||||
This script collects commissioning options (VID:PID, LED, options,
|
||||
product name, curves, etc.) and writes them to a JSON file
|
||||
(`commission_config.json`). It can optionally attempt to apply the
|
||||
configuration when `--apply` is passed; the apply action is a safe
|
||||
stub and will ask for confirmation before performing any hardware
|
||||
changes.
|
||||
|
||||
Usage:
|
||||
python configure.py # interactive
|
||||
python configure.py --out mycfg.json
|
||||
python configure.py --yes --out mycfg.json
|
||||
python configure.py --apply # will prompt before touching device
|
||||
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
|
||||
CFG_OUT = "commission_config.json"
|
||||
VIDPID_RE = re.compile(r"^[0-9a-fA-F]{4}:[0-9a-fA-F]{4}$")
|
||||
|
||||
|
||||
def prompt(text: str, default: Optional[str] = None) -> Optional[str]:
|
||||
if default is None:
|
||||
val = input(f"{text} ")
|
||||
else:
|
||||
val = input(f"{text} [Default: {default}] ")
|
||||
if val.strip() == "":
|
||||
return None
|
||||
return val.strip()
|
||||
|
||||
|
||||
def prompt_int(text: str, default: Optional[int] = None, allow_empty=True) -> Optional[int]:
|
||||
while True:
|
||||
s = prompt(text, str(default) if default is not None else None)
|
||||
if s is None:
|
||||
return None if allow_empty else default
|
||||
try:
|
||||
return int(s)
|
||||
except ValueError:
|
||||
print("Please enter a number, or press enter to leave blank.")
|
||||
|
||||
|
||||
def prompt_bool(text: str, default: Optional[bool] = None) -> Optional[bool]:
|
||||
while True:
|
||||
defstr = None if default is None else ("y" if default else "n")
|
||||
s = prompt(text + " (y/n)", defstr)
|
||||
if s is None:
|
||||
return None
|
||||
if s.lower() in ("y", "yes", "t", "1"):
|
||||
return True
|
||||
if s.lower() in ("n", "no", "f", "0"):
|
||||
return False
|
||||
print("Please enter either "Y" or "N", or leave blank to keep the option unchanged.")
|
||||
|
||||
|
||||
def choose(text: str, choices: list[str], default: Optional[int] = None) -> Optional[str]:
|
||||
print(text)
|
||||
for i, c in enumerate(choices, start=1):
|
||||
print(f" {i}. {c}")
|
||||
sel = prompt_int("Enter number, or leave blank to keep current setting:", default=None)
|
||||
if sel is None:
|
||||
return None
|
||||
if 1 <= sel <= len(choices):
|
||||
return choices[sel - 1]
|
||||
print("Sorry, this input's not recognised. Either enter a number, or leave blank to keep the current setting:")
|
||||
return None
|
||||
|
||||
|
||||
def validate_vidpid(v: str) -> bool:
|
||||
return bool(VIDPID_RE.match(v))
|
||||
|
||||
|
||||
def interactive_build() -> dict:
|
||||
print("Welcome to the Python PicoKey interactive Comissioner,")
|
||||
|
||||
# Vendor / VID:PID
|
||||
VENDORS = {
|
||||
"Nitrokey HSM": {"vid": "20a0", "pid": "4230"},
|
||||
"Nitrokey FIDO2": {"vid": "20a0", "pid": "42b1"},
|
||||
"Nitrokey Pro": {"vid": "20a0", "pid": "4108"},
|
||||
"Nitrokey 3": {"vid": "20a0", "pid": "42b2"},
|
||||
"Nitrokey Start": {"vid": "20a0", "pid": "4211"},
|
||||
"Yubikey 4/5": {"vid": "1050", "pid": "0407"},
|
||||
"Yubikey NEO": {"vid": "1050", "pid": "0116"},
|
||||
"Yubico YubiHSM": {"vid": "1050", "pid": "0030"},
|
||||
"FSIJ Gnuk": {"vid": "234b", "pid": "0000"},
|
||||
"GnuPG e.V.": {"vid": "1209", "pid": "2440"},
|
||||
}
|
||||
|
||||
vendor_choices = list(VENDORS.keys()) + ["Custom VID:PID"]
|
||||
vendor_choice = choose("Select a known vendor...", vendor_choices, None)
|
||||
vidpid = None
|
||||
if vendor_choice == "Custom VID:PID":
|
||||
while True:
|
||||
v = prompt("Type VID:PID in hex form (0123:abcd):")
|
||||
if v is None:
|
||||
vidpid = None
|
||||
break
|
||||
if validate_vidpid(v):
|
||||
vidpid = v.lower()
|
||||
break
|
||||
print("Wrong format. It should be in hexadecimal format. An example is 0123:abcd.")
|
||||
elif vendor_choice in VENDORS:
|
||||
e = VENDORS[vendor_choice]
|
||||
vidpid = f"{e['vid']}:{e['pid']}"
|
||||
|
||||
presence = prompt_int("Presence Button Timeout (seconds, 0=disabled):", default=None)
|
||||
led_brightness = prompt_int("LED brightness (0=off):", default=None)
|
||||
|
||||
# Options
|
||||
print("Options: 留空保持当前")
|
||||
led_dimmable = prompt_bool("LED dimmable?", default=None)
|
||||
initialize = prompt_bool("Initialize device (will reset some state)?", default=None)
|
||||
secure_boot = prompt_bool("Enable Secure Boot? (WebUSB required)", default=None)
|
||||
secure_lock = prompt_bool("Enable Secure Lock? (WebUSB required)", default=None)
|
||||
power_cycle = prompt_bool("Power Cycle on Reset? (Pico FIDO only)", default=None)
|
||||
led_steady = prompt_bool("LED steady (always on)?", default=None)
|
||||
secp256k1 = prompt_bool("Enable secp256k1 curve? (Android may not support)", default=None)
|
||||
|
||||
led_gpio = prompt_int("LED GPIO pin (number):", default=None)
|
||||
|
||||
drivers = ["PICO", "PIMORONI", "WS2812", "CYW43", "NEOPIXEL", "NONE"]
|
||||
led_driver = choose("Select a LED driver:", drivers, None)
|
||||
|
||||
product_name = prompt("Product Name (max 14 chars):")
|
||||
if product_name is not None and len(product_name) > 14:
|
||||
print("Product name 超过 14 字符,将被截断。")
|
||||
product_name = product_name[:14]
|
||||
|
||||
cfg = {
|
||||
"vendor_choice": vendor_choice,
|
||||
"vidpid": vidpid,
|
||||
"presence_timeout": presence,
|
||||
"led_brightness": led_brightness,
|
||||
"options": {
|
||||
"led_dimmable": led_dimmable,
|
||||
"initialize": initialize,
|
||||
"secure_boot": secure_boot,
|
||||
"secure_lock": secure_lock,
|
||||
"power_cycle_on_reset": power_cycle,
|
||||
"led_steady": led_steady,
|
||||
"secp256k1": secp256k1,
|
||||
},
|
||||
"led_gpio": led_gpio,
|
||||
"led_driver": led_driver,
|
||||
"product_name": product_name,
|
||||
}
|
||||
|
||||
return cfg
|
||||
|
||||
|
||||
def build_phy_bytes(cfg: dict) -> bytes:
|
||||
# mirror getPhyData() TLV construction
|
||||
PHY_VID = 0x0
|
||||
PHY_LED_GPIO = 0x4
|
||||
PHY_LED_BTNESS = 0x5
|
||||
PHY_OPTS = 0x6
|
||||
PHY_OPT_WCID = 0x1
|
||||
PHY_OPT_DIMM = 0x2
|
||||
PHY_OPT_DISABLE_POWER_RESET = 0x4
|
||||
PHY_OPT_LED_STEADY = 0x8
|
||||
PHY_UP_BUTTON = 0x8
|
||||
PHY_USB_PRODUCT = 0x9
|
||||
PHY_ENABLED_CURVES = 0xA
|
||||
PHY_LED_DRIVER = 0xC
|
||||
|
||||
PHY_CURVE_SECP256K1 = 0x8
|
||||
|
||||
PHY_LED_DRIVER_SINGLE = 0x1
|
||||
PHY_LED_DRIVER_WS2812 = 0x3
|
||||
|
||||
b = bytearray()
|
||||
|
||||
# VID/PID
|
||||
vidpid = cfg.get("vidpid")
|
||||
if vidpid:
|
||||
try:
|
||||
vid_str, pid_str = vidpid.split(":")
|
||||
vid = int(vid_str, 16)
|
||||
pid = int(pid_str, 16)
|
||||
b += bytes([PHY_VID, 4, (vid >> 8) & 0xFF, vid & 0xFF, (pid >> 8) & 0xFF, pid & 0xFF])
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# LED GPIO
|
||||
lg = cfg.get("led_gpio")
|
||||
if lg is not None:
|
||||
b += bytes([PHY_LED_GPIO, 1, int(lg) & 0xFF])
|
||||
|
||||
# LED brightness
|
||||
lb = cfg.get("led_brightness")
|
||||
if lb is None:
|
||||
lb = 0
|
||||
b += bytes([PHY_LED_BTNESS, 1, int(lb) & 0xFF])
|
||||
|
||||
# opts
|
||||
opts = 0
|
||||
opts_map = cfg.get("options", {})
|
||||
if opts_map.get("led_dimmable"):
|
||||
opts |= PHY_OPT_DIMM
|
||||
if not opts_map.get("power_cycle_on_reset"):
|
||||
opts |= PHY_OPT_DISABLE_POWER_RESET
|
||||
if opts_map.get("led_steady"):
|
||||
opts |= PHY_OPT_LED_STEADY
|
||||
|
||||
b += bytes([PHY_OPTS, 2, (opts >> 8) & 0xFF, opts & 0xFF])
|
||||
|
||||
# Presence / Up button timeout
|
||||
btn = cfg.get("presence_timeout")
|
||||
if btn is None:
|
||||
btn = 0
|
||||
b += bytes([PHY_UP_BUTTON, 1, int(btn) & 0xFF])
|
||||
|
||||
# curves
|
||||
curves = 0
|
||||
if opts_map.get("secp256k1"):
|
||||
curves |= PHY_CURVE_SECP256K1
|
||||
b += bytes([PHY_ENABLED_CURVES, 4, (curves >> 24) & 0xFF, (curves >> 16) & 0xFF, (curves >> 8) & 0xFF, curves & 0xFF])
|
||||
|
||||
# USB product string
|
||||
pn = cfg.get("product_name")
|
||||
if pn:
|
||||
s = pn.encode("ascii", "ignore")
|
||||
b += bytes([PHY_USB_PRODUCT, len(s) + 1]) + s + b"\x00"
|
||||
|
||||
# led driver
|
||||
ld = cfg.get("led_driver")
|
||||
leddrv = 0
|
||||
if ld is not None:
|
||||
if ld.upper().startswith("WS"):
|
||||
leddrv = PHY_LED_DRIVER_WS2812
|
||||
elif ld.upper().startswith("PICO"):
|
||||
leddrv = PHY_LED_DRIVER_SINGLE
|
||||
elif ld.upper() == "NONE":
|
||||
leddrv = 0xFF
|
||||
b += bytes([PHY_LED_DRIVER, 1, leddrv & 0xFF])
|
||||
|
||||
return bytes(b)
|
||||
|
||||
|
||||
def save_config(cfg: dict, out: Path):
|
||||
out.write_text(json.dumps(cfg, ensure_ascii=False, indent=2))
|
||||
print(f"配置已保存到 {out}")
|
||||
|
||||
|
||||
def apply_stub(cfg: dict) -> None:
|
||||
print("\n-- APPLY (STUB) --")
|
||||
print("脚本将在此处尝试应用配置到本机已连接的 PicoKey(本地方式)。")
|
||||
print("配置摘要:")
|
||||
print(json.dumps(cfg, ensure_ascii=False, indent=2))
|
||||
confirm = input("确认要尝试直接应用配置到本机已连接设备吗?(y/N) ")
|
||||
if confirm.lower() != "y":
|
||||
print("取消应用。配置已保存到文件,可使用本地 picokey 库或其他工具手动应用。")
|
||||
return
|
||||
# 尝试使用本地 picokey 库把 PHY bytes 写入设备
|
||||
try:
|
||||
import picokey # type: ignore
|
||||
except Exception as e:
|
||||
print("未检测到可用的 picokey 库:", e)
|
||||
print("请确保已安装 pypicokey 并在正确的 Python 环境下运行此脚本。")
|
||||
return
|
||||
|
||||
def apply_local(cfg: dict) -> None:
|
||||
try:
|
||||
pk = picokey.PicoKey()
|
||||
except Exception as e:
|
||||
print("无法连接到 PicoKey 设备:", e)
|
||||
return
|
||||
try:
|
||||
phy = build_phy_bytes(cfg)
|
||||
if not phy:
|
||||
print("未生成任何 PHY 字节,跳过写入。")
|
||||
return
|
||||
print(f"准备写入 {len(phy)} 字节到设备...")
|
||||
try:
|
||||
# PicoKey.phy expects Optional[list[int]] for data
|
||||
pk.phy(list(phy))
|
||||
print("已成功写入 PHY 配置到设备。")
|
||||
except Exception as e:
|
||||
print("写入设备时发生错误:", e)
|
||||
finally:
|
||||
try:
|
||||
pk.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
apply_local(cfg)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="PicoKey commissioning configurator")
|
||||
parser.add_argument("--out", "-o", help="输出 JSON 文件路径", default=CFG_OUT)
|
||||
parser.add_argument("--apply", action="store_true", help="交互式确认后尝试应用配置到设备(stub)")
|
||||
parser.add_argument("--yes", "-y", action="store_true", help="自动确认所有提示(留空将被视为保留当前值)")
|
||||
args = parser.parse_args()
|
||||
|
||||
cfg = interactive_build()
|
||||
|
||||
outpath = Path(args.out)
|
||||
if outpath.exists():
|
||||
if not args.yes:
|
||||
overwrite = input(f"{outpath} 已存在,是否覆盖?(y/N) ")
|
||||
if overwrite.lower() != "y":
|
||||
print("取消,未保存文件。")
|
||||
return
|
||||
save_config(cfg, outpath)
|
||||
|
||||
if args.apply:
|
||||
apply_stub(cfg)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+56
-21
@@ -25,6 +25,7 @@ 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
|
||||
|
||||
@@ -156,8 +157,10 @@ class PicoKey:
|
||||
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()
|
||||
@@ -169,20 +172,32 @@ class PicoKey:
|
||||
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')
|
||||
raise PicoKeyNotFoundError('time-out: no card inserted')
|
||||
try:
|
||||
logger.debug("Selecting applet in rescue mode...")
|
||||
resp, sw1, sw2 = self.select_applet(rescue=True)
|
||||
resp, sw1, sw2 = self.select_applet()
|
||||
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])
|
||||
if (len(resp) >= 12):
|
||||
self.serial_number = int.from_bytes(resp[4:12], 'big')
|
||||
else:
|
||||
self.serial_number = 0
|
||||
logger.debug(f"Device platform: {self.platform.name}, product: {self.product.name}, version: {self.version}, serial number: {self.serial_number:016X}")
|
||||
else:
|
||||
logger.error("Unexpected response code during applet selection")
|
||||
self.platform = Platform(Platform.RP2040)
|
||||
self.product = Product(Product.UNKNOWN)
|
||||
self.version = (0, 0)
|
||||
self.serial_number = 0
|
||||
except APDUResponse:
|
||||
logger.error("APDU response error during applet selection")
|
||||
self.platform = Platform(Platform.RP2040)
|
||||
self.product = Product(Product.UNKNOWN)
|
||||
self.version = (0, 0)
|
||||
self.serial_number = 0
|
||||
|
||||
@property
|
||||
def device(self):
|
||||
@@ -208,27 +223,35 @@ class PicoKey:
|
||||
self.__card.close()
|
||||
else:
|
||||
logger.debug("Removing card monitor observer...")
|
||||
self.__monitor.deleteObserver(self.__observer)
|
||||
self.__observer = None
|
||||
self.__monitor = None
|
||||
if (self.__monitor and self.__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()
|
||||
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 Exception('No device connected')
|
||||
response, sw1, sw2 = self.__card.transmit(apdu)
|
||||
return response, sw1, sw2
|
||||
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 Exception('No device connected')
|
||||
raise PicoKeyNotFoundError('No device connected')
|
||||
lc = []
|
||||
dataf = []
|
||||
if (data):
|
||||
@@ -256,12 +279,18 @@ class PicoKey:
|
||||
response, sw1, sw2 = self.__card.transmit(apdu)
|
||||
except Exception:
|
||||
logger.debug("Reconnecting card after transmit failure")
|
||||
self.__card.reconnect()
|
||||
|
||||
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 Exception("APDU transmission error after reconnect: " + str(e))
|
||||
raise PicoKeyInvalidStateError("APDU transmission error after reconnect: " + str(e))
|
||||
|
||||
code = (sw1<<8|sw2)
|
||||
if (sw1 != 0x90):
|
||||
@@ -312,17 +341,15 @@ class PicoKey:
|
||||
raise Exception('Secure Channel token verification failed')
|
||||
self.__sc = sc
|
||||
|
||||
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 select_applet(self):
|
||||
logger.debug("Selecting rescue applet")
|
||||
return self.transmit([0x00, 0xA4, 0x04, 0x04, 0x08, 0xA0, 0x58, 0x3F, 0xC1, 0x9B, 0x7E, 0x4F, 0x21, 0x00])
|
||||
|
||||
def phy(self, data : Optional[list[int]] = None):
|
||||
if (data is None):
|
||||
try:
|
||||
resp, sw = self.send(0x1E, cla=0x80, p1=0x00, ne=256)
|
||||
self.select_applet()
|
||||
resp, sw = self.send(0x1E, cla=0x80, p1=0x01, ne=256)
|
||||
return PhyData.parse(resp)
|
||||
except Exception:
|
||||
pass
|
||||
@@ -333,6 +360,7 @@ class PicoKey:
|
||||
def flash_info(self):
|
||||
logger.debug("Retrieving flash info")
|
||||
try:
|
||||
self.select_applet()
|
||||
resp, sw = self.send(0x1E, cla=0x80, p1=0x02)
|
||||
free = int.from_bytes(resp[0:4], 'big')
|
||||
used = int.from_bytes(resp[4:8], 'big')
|
||||
@@ -351,6 +379,7 @@ class PicoKey:
|
||||
|
||||
def secure_info(self):
|
||||
logger.debug("Retrieving secure boot info")
|
||||
self.select_applet()
|
||||
resp, sw = self.send(0x1E, cla=0x80, p1=0x03)
|
||||
return {
|
||||
'enabled': resp[0] != 0,
|
||||
@@ -360,5 +389,11 @@ 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}")
|
||||
self.select_applet()
|
||||
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.select_applet()
|
||||
self.send(0x1F, cla=0x80, p1=0x01 if bootsel else 0x00)
|
||||
|
||||
@@ -62,6 +62,9 @@ class RescueMonitor:
|
||||
|
||||
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:
|
||||
|
||||
@@ -17,6 +17,7 @@
|
||||
*/
|
||||
"""
|
||||
|
||||
import time
|
||||
import usb.core
|
||||
import usb.util
|
||||
import libusb_package
|
||||
@@ -26,6 +27,15 @@ 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):
|
||||
@@ -89,12 +99,16 @@ class RescuePicoKey:
|
||||
break
|
||||
if (not found):
|
||||
logger.error("No suitable device found")
|
||||
raise Exception('Not found any Pico Key device')
|
||||
raise RescuePicoKeyNotFoundError('Not found any Pico Key device')
|
||||
|
||||
@property
|
||||
def device(self):
|
||||
return self.__dev
|
||||
|
||||
@property
|
||||
def serial_number(self) -> str:
|
||||
return usb.util.get_string(self.__dev, self.__dev.iSerialNumber)
|
||||
|
||||
def close(self):
|
||||
logger.debug("Closing device")
|
||||
if self.__dev:
|
||||
@@ -117,16 +131,24 @@ class RescuePicoKey:
|
||||
|
||||
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
|
||||
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):
|
||||
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")
|
||||
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")
|
||||
@@ -134,12 +156,12 @@ class RescuePicoKey:
|
||||
self.write(data=data, timeout=timeout)
|
||||
except Exception as e:
|
||||
logger.error("USB write error: " + str(e))
|
||||
raise Exception("USB write error: " + str(e))
|
||||
raise RescuePicoKeyInvalidStateError("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))
|
||||
raise RescuePicoKeyInvalidStateError("USB read error: " + str(e))
|
||||
return ret
|
||||
|
||||
def powerOn(self):
|
||||
@@ -162,3 +184,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.2"
|
||||
__version__ = "1.3.2"
|
||||
|
||||
@@ -1 +1,2 @@
|
||||
from .enums import *
|
||||
from .exceptions import *
|
||||
|
||||
@@ -1,3 +1,22 @@
|
||||
"""
|
||||
/*
|
||||
* 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
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -56,4 +56,3 @@ def get_logger(name: str):
|
||||
logger.addHandler(handler)
|
||||
|
||||
return logger
|
||||
|
||||
|
||||
+1
-1
@@ -1,5 +1,5 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=61.0"]
|
||||
requires = ["setuptools"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
|
||||
Reference in New Issue
Block a user