refactor(keycard_wallet): replace keycard-py (pyo3) with native keycard-rs

Rewrites the Keycard integration to talk to the LEE-flavored applet
directly from Rust via keycard-rs (pinned git dependency), dropping the
pyo3 shim, python_path.rs, and the vendored keycard-py Python library
entirely. Verified end-to-end against real hardware: pairing, PIN
verification, mnemonic loading, BIP340 Schnorr signing, and transfers
between keycard and public/private accounts.

Also cleans up the pyo3 usage that had leaked into wallet's signing
path (signing.rs, account_manager.rs) beyond the keycard CLI itself,
and trims wallet_with_keycard.sh's Python setup down to just pyscard,
which is only needed by the force_unpower.py test helper now.
This commit is contained in:
Marvin Jones 2026-07-03 15:47:05 -04:00
parent a0ba6008c3
commit 763464d27b
16 changed files with 728 additions and 703 deletions

556
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -163,6 +163,8 @@ logos-blockchain-chain-service = { git = "https://github.com/logos-blockchain/lo
logos-blockchain-zone-sdk = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "d8711bbc3d43d3ef9755ef9b73af32fd0f703160" }
logos-blockchain-http-api-common = { git = "https://github.com/logos-blockchain/logos-blockchain.git", rev = "d8711bbc3d43d3ef9755ef9b73af32fd0f703160" }
keycard-rs = { git = "https://github.com/keycard-tech/keycard-rs", rev = "98abc3ff1b26785a5631634d5bf24da54da541bf" }
rocksdb = { version = "0.24.0", default-features = false, features = [
"snappy",
"bindgen-runtime",
@ -183,7 +185,6 @@ actix-web = { version = "4.13.0", default-features = false, features = [
] }
clap = { version = "4.5.42", features = ["derive", "env"] }
reqwest = { version = "0.12", features = ["json", "rustls-tls", "stream"] }
pyo3 = { version = "0.29", features = ["auto-initialize"] }
zeroize = "1"
criterion = { version = "0.8", features = ["html_reports"] }

View File

@ -9,8 +9,11 @@ workspace = true
[dependencies]
lee.workspace = true
pyo3.workspace = true
keycard-rs.workspace = true
bip39.workspace = true
log.workspace = true
rand.workspace = true
serde = { workspace = true, features = ["derive"] }
serde_json.workspace = true
thiserror.workspace = true
zeroize.workspace = true

View File

@ -1,221 +0,0 @@
from smartcard.System import readers
from keycard.exceptions import APDUError, TransportError
from ecdsa import VerifyingKey, SECP256k1
from keycard.keycard import KeyCard
from keycard.commands.export_lee_key import export_lee_key
from mnemonic import Mnemonic
from keycard import constants
import os
import secrets
DEFAULT_PAIRING_PASSWORD = "KeycardDefaultPairing"
def _pairing_password() -> str:
return os.environ.get("KEYCARD_PAIRING_PASSWORD", DEFAULT_PAIRING_PASSWORD)
class KeycardWallet:
def __init__(self):
self.card = KeyCard()
def _is_smart_card_reader_detected(self) -> bool:
try:
return len(readers()) > 0
except Exception:
return False
def _is_keycard_detected(self) -> bool:
try:
KeyCard().select()
return True
except (TransportError, APDUError, Exception):
# No readers, no card, or card doesn't respond.
return False
def is_unpaired_keycard_available(self) -> bool:
if not self._is_smart_card_reader_detected():
return False
elif not self._is_keycard_detected():
return False
return True
def initialize(self, pin: str, pairing_password: str | None = None) -> bool:
try:
self.card.select()
if self.card.is_initialized:
raise RuntimeError("Card is already initialized")
puk = ''.join(secrets.choice('0123456789') for _ in range(12))
self.card.init(pin, puk, pairing_password or _pairing_password())
print(f"Keycard PUK: {puk}")
print("Record this PUK and store it somewhere safe. It cannot be recovered.")
return True
except Exception as e:
raise RuntimeError(f"Error initializing keycard: {e}") from e
def _reconnect(self) -> None:
self.card = KeyCard()
self.card.select()
def _pair(self, pin: str, password: str) -> tuple[int, bytes]:
self.card.select()
if not self.card.is_initialized:
raise RuntimeError("Card is not initialized — run 'wallet keycard init' first")
pairing_index, pairing_key = self.card.pair(password)
self.pairing_index = pairing_index
self.pairing_key = pairing_key
try:
self.card.open_secure_channel(pairing_index, pairing_key)
self.card.verify_pin(pin)
except Exception as e:
try:
self.card.unpair(pairing_index)
except Exception:
pass
raise RuntimeError(f"Error opening secure channel after fresh pair: {e}") from e
return pairing_index, pairing_key
def pair(self, pin: str, password: str | None = None) -> tuple[int, bytes]:
password = password or _pairing_password()
try:
return self._pair(pin, password)
except TransportError as e:
print(f"Transport error during fresh pair ({e}), attempting card reset and retry...")
try:
self._reconnect()
result = self._pair(pin, password)
print("Retry succeeded after card reset.")
return result
except TransportError as e2:
raise RuntimeError(
"Card lost power and did not recover after reset. "
"Try reseating the card in the reader."
) from e2
def _setup_communication_with_pairing(self, pin: str, pairing_index: int, pairing_key: bytes) -> bool:
self.card.select()
if not self.card.is_initialized:
raise RuntimeError("Card is not initialized — run 'wallet keycard init' first")
self.pairing_index = pairing_index
self.pairing_key = pairing_key
try:
self.card.open_secure_channel(pairing_index, pairing_key)
self.card.verify_pin(pin)
except Exception as e:
raise RuntimeError(f"Error setting up communication with stored pairing: {e}") from e
return True
def setup_communication_with_pairing(self, pin: str, pairing_index: int, pairing_key: bytes) -> bool:
try:
return self._setup_communication_with_pairing(pin, pairing_index, pairing_key)
except TransportError as e:
print(f"Transport error during stored pairing ({e}), attempting card reset and retry...")
try:
self._reconnect()
result = self._setup_communication_with_pairing(pin, pairing_index, pairing_key)
print("Retry succeeded after card reset.")
return result
except TransportError as e2:
raise RuntimeError(
"Card lost power and did not recover after reset. "
"Try reseating the card in the reader."
) from e2
def close_session(self) -> bool:
return True
def load_mnemonic(self, mnemonic: str) -> bool:
try:
# Convert mnemonic to seed
mnemo = Mnemonic("english")
if not mnemo.check(mnemonic):
raise RuntimeError("Invalid mnemonic phrase — check spelling and word count")
seed = mnemo.to_seed(mnemonic)
# Load the LEE seed onto the card
result = self.card.load_key(
key_type = constants.LoadKeyType.LEE_SEED,
lee_seed = seed
)
return True
except Exception as e:
raise RuntimeError(f"Error loading mnemonic: {e}") from e
def disconnect(self) -> bool:
try:
if not self.card.is_secure_channel_open:
return False
self.card.unpair(self.pairing_index)
return True
except Exception as e:
raise RuntimeError(f"Error during disconnect: {e}") from e
def get_public_key_for_path(self, path: str = "m/44'/60'/0'/0/0") -> bytes | None:
try:
if not self.card.is_secure_channel_open or not self.card.is_pin_verified:
return None
public_key = self.card.export_key(
derivation_option = constants.DerivationOption.DERIVE,
public_only = True,
keypath = path
)
public_key = public_key.public_key
public_key = VerifyingKey.from_string(public_key[1:], curve=SECP256k1)
public_key = public_key.to_string("compressed")[1:]
return public_key
except Exception as e:
raise RuntimeError(f"Error getting public key: {e}") from e
def sign_message_for_path(self, message: bytes, path: str = "m/44'/60'/0'/0/0") -> bytes | None:
try:
if not self.card.is_secure_channel_open or not self.card.is_pin_verified:
return None
signature = self.card.sign_with_path(
digest = message,
path = path,
algorithm = constants.SigningAlgorithm.SCHNORR_BIP340,
make_current = False
)
return signature.signature
except Exception as e:
raise RuntimeError(f"Error signing message: {e}") from e
def get_private_keys_for_path(self, path: str = "m/44'/60'/0'/0/0") -> bytes | None:
try:
if not self.card.is_secure_channel_open or not self.card.is_pin_verified:
return None
private_keys = export_lee_key(
self.card,
constants.DerivationOption.DERIVE,
path
)
nsk = private_keys.lee_nsk
vsk = private_keys.lee_vsk
return (nsk, vsk)
except Exception as e:
raise RuntimeError(f"Error getting private keys: {e}") from e

View File

@ -1,16 +1,48 @@
use std::path::PathBuf;
use std::{path::PathBuf, str::FromStr as _};
use keycard_rs::{
KeycardCommandSet, PcscChannel,
constants::sign_p2,
parsing::Bip32KeyPair,
secure_channel::Pairing,
tlv::{BerTlvReader, TLV_KEY_TEMPLATE, TLV_PUB_KEY, TLV_SIGNATURE_TEMPLATE},
};
use lee::{AccountId, PublicKey, Signature};
use pyo3::{prelude::*, types::PyAny};
use rand::Rng as _;
use serde::{Deserialize, Serialize};
use zeroize::Zeroizing;
pub mod python_path;
const DEFAULT_PAIRING_PASSWORD: &str = "KeycardDefaultPairing";
/// LEE-applet extension tags. These aren't part of upstream `keycard-rs`'s `tlv` module (which
/// only knows the standard Status Keycard tags) since they're specific to the LEE-flavored
/// applet (`LEE_keycard.cap`).
const TLV_LEE_NSK: u8 = 0x83;
const TLV_LEE_VSK: u8 = 0x84;
/// Raw Schnorr signature (64 bytes: `r || s`, no ASN.1/DER wrapping) nested inside the standard
/// `TLV_SIGNATURE_TEMPLATE` alongside the usual `TLV_PUB_KEY`. Confirmed against real hardware —
/// the LEE applet's `SIGN` response for `sign_p2::BIP340_SCHNORR` is
/// `0xA0 { 0x80 <65-byte pubkey>, 0x88 <64-byte r||s> }`.
const TLV_LEE_RAW_SIGNATURE: u8 = 0x88;
/// NSK (32 bytes) and VSK (64 bytes, the ML-KEM-768 seed `d || z`) as fixed-length zeroizing byte
/// arrays.
type PrivateKeyPair = (Zeroizing<[u8; 32]>, Zeroizing<[u8; 64]>);
#[derive(Debug, thiserror::Error)]
pub enum KeycardWalletError {
#[error(transparent)]
Keycard(#[from] keycard_rs::Error),
#[error("keycard is already initialized")]
AlreadyInitialized,
#[error("invalid mnemonic phrase: {0}")]
InvalidMnemonic(String),
#[error("invalid key material from keycard: {0}")]
InvalidKeyMaterial(String),
#[error("keycard returned a signature that does not verify against its own public key")]
SignatureVerificationFailed,
}
// TODO: encrypt at rest alongside broader wallet storage encryption work.
#[derive(Serialize, Deserialize)]
pub struct KeycardPairingData {
@ -24,136 +56,195 @@ impl KeycardPairingData {
}
}
/// Rust wrapper around the Python `KeycardWallet` class.
/// Rust wrapper around `keycard-rs`, talking to the LEE-flavored Keycard applet over PC/SC.
pub struct KeycardWallet {
instance: Py<PyAny>,
command_set: KeycardCommandSet,
}
impl KeycardWallet {
/// Create a new Python `KeycardWallet` instance.
pub fn new(py: Python) -> PyResult<Self> {
let module = py.import("keycard_wallet")?;
let class = module.getattr("KeycardWallet")?;
let instance = class.call0()?;
/// Connects to the first available PC/SC reader. Does not select the applet yet — callers
/// that need application info (`initialize`, `pair`, `connect`, ...) do that themselves.
pub fn new() -> Result<Self, KeycardWalletError> {
let channel = PcscChannel::connect()?;
Ok(Self {
instance: instance.into(),
command_set: KeycardCommandSet::new(channel),
})
}
pub fn is_unpaired_keycard_available(&self, py: Python) -> PyResult<bool> {
self.instance
.bind(py)
.call_method0("is_unpaired_keycard_available")?
.extract()
/// Returns whether a smart card reader and a selectable Keycard are both present.
#[must_use]
pub fn is_unpaired_keycard_available() -> bool {
let Ok(channel) = PcscChannel::connect() else {
return false;
};
KeycardCommandSet::new(channel)
.select()
.is_ok_and(|resp| resp.is_ok())
}
pub fn initialize(&self, py: Python<'_>, pin: &str) -> PyResult<bool> {
self.instance
.bind(py)
.call_method1("initialize", (pin,))?
.extract()
fn select(&mut self) -> Result<(), KeycardWalletError> {
self.command_set.select()?.check_ok()?;
Ok(())
}
pub fn pair(&self, py: Python<'_>, pin: &str) -> PyResult<(u8, Vec<u8>)> {
self.instance
.bind(py)
.call_method1("pair", (pin,))?
.extract()
fn open_and_verify(&mut self, pin: &str) -> Result<(), KeycardWalletError> {
self.command_set.auto_open_secure_channel()?;
self.command_set.verify_pin(pin)?.check_auth_ok()?;
Ok(())
}
/// Rebuilds the PC/SC channel and command set, then retries `op` once, if `op` failed with a
/// transport-level error (e.g. the card lost power mid-session). Mirrors the reconnect-once
/// behavior the previous `keycard-py`-based wallet had for `TransportError`.
fn with_reconnect_on_transport_error<T>(
&mut self,
op: impl Fn(&mut Self) -> Result<T, KeycardWalletError>,
) -> Result<T, KeycardWalletError> {
match op(self) {
Err(KeycardWalletError::Keycard(keycard_rs::Error::Io(io_err))) => {
log::warn!(
"transport error during keycard operation ({io_err}), reconnecting and retrying once"
);
*self = Self::new()?;
op(self)
}
result => result,
}
}
/// Initializes an uninitialized card, returning the generated PUK. The caller is responsible
/// for surfacing the PUK to the operator — it cannot be recovered afterward.
pub fn initialize(&mut self, pin: &str) -> Result<String, KeycardWalletError> {
self.select()?;
let already_initialized = self
.command_set
.app_info()
.expect("select() populates app_info on success")
.is_initialized();
if already_initialized {
return Err(KeycardWalletError::AlreadyInitialized);
}
let puk = generate_puk();
self.command_set
.init(pin, &puk, &pairing_password())?
.check_ok()?;
Ok(puk)
}
pub fn pair(&mut self, pin: &str) -> Result<(u8, [u8; 32]), KeycardWalletError> {
self.select()?;
self.command_set.auto_pair(&pairing_password())?;
let pairing = self
.command_set
.pairing()
.expect("auto_pair sets pairing data on success")
.clone();
if let Err(err) = self.open_and_verify(pin) {
drop(self.command_set.auto_unpair());
return Err(err);
}
Ok((pairing.pairing_index(), *pairing.pairing_key()))
}
pub fn setup_communication_with_pairing(
&self,
py: Python<'_>,
&mut self,
pin: &str,
index: u8,
key: &[u8],
) -> PyResult<bool> {
self.instance
.bind(py)
.call_method1(
"setup_communication_with_pairing",
(pin, index, key.to_vec()),
)?
.extract()
}
pub fn close_session(&self, py: Python<'_>) -> PyResult<bool> {
self.instance
.bind(py)
.call_method0("close_session")?
.extract()
key: &[u8; 32],
) -> Result<(), KeycardWalletError> {
self.select()?;
self.command_set.set_pairing(Pairing::new(*key, index));
self.open_and_verify(pin)
}
/// Connect using a stored pairing if available, falling back to a fresh pair.
/// Saves any newly established pairing to disk.
pub fn connect(&self, py: Python<'_>, pin: &str) -> PyResult<()> {
if let Some(pairing) = load_pairing().filter(KeycardPairingData::is_valid)
&& self
.setup_communication_with_pairing(py, pin, pairing.index, &pairing.key)
.is_ok()
{
return Ok(());
pub fn connect(&mut self, pin: &str) -> Result<(), KeycardWalletError> {
if let Some(pairing) = load_pairing().filter(KeycardPairingData::is_valid) {
let key: [u8; 32] = pairing
.key
.clone()
.try_into()
.expect("KeycardPairingData::is_valid checked the key is 32 bytes");
let reconnected = self.with_reconnect_on_transport_error(|wallet| {
wallet.setup_communication_with_pairing(pin, pairing.index, &key)
});
if reconnected.is_ok() {
return Ok(());
}
}
let (index, key) = self.pair(py, pin)?;
save_pairing(&KeycardPairingData { index, key });
let (index, key) = self.with_reconnect_on_transport_error(|wallet| wallet.pair(pin))?;
save_pairing(&KeycardPairingData {
index,
key: key.to_vec(),
});
Ok(())
}
pub fn disconnect(&self, py: Python) -> PyResult<bool> {
self.instance.bind(py).call_method0("disconnect")?.extract()
/// Unpairs the current session. Returns `false` if there was nothing to unpair.
pub fn disconnect(&mut self) -> Result<bool, KeycardWalletError> {
if self.command_set.pairing().is_none() {
return Ok(false);
}
self.command_set.auto_unpair()?;
Ok(true)
}
pub fn get_public_key_for_path(&self, py: Python, path: &str) -> PyResult<PublicKey> {
let public_key: Vec<u8> = self
.instance
.bind(py)
.call_method1("get_public_key_for_path", (path,))?
.extract()?;
pub fn get_public_key_for_path(&mut self, path: &str) -> Result<PublicKey, KeycardWalletError> {
let resp = self.command_set.export_key(path, false, true)?;
resp.check_ok()?;
let keypair = Bip32KeyPair::from_tlv(resp.data())?;
let public_key: [u8; 32] = public_key.try_into().map_err(|vec: Vec<u8>| {
PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
"expected 32-byte public key from keycard, got {} bytes",
vec.len()
))
})?;
// Uncompressed SEC1 point (0x04 || X || Y); the BIP340 x-only public key is just its X
// coordinate, since secp256k1 (what the card signs with) and k256's Schnorr verifying key
// are the same curve.
let public_key = keypair.public_key();
let x_only: [u8; 32] = match public_key.split_first() {
Some((&0x04, xy)) if xy.len() == 64 => xy
.split_at(32)
.0
.try_into()
.expect("split_at(32) of a 64-byte slice"),
_ => {
return Err(KeycardWalletError::InvalidKeyMaterial(format!(
"expected a 65-byte uncompressed secp256k1 public key from keycard, got {} bytes",
public_key.len()
)));
}
};
PublicKey::try_new(public_key)
.map_err(|e| PyErr::new::<pyo3::exceptions::PyValueError, _>(e.to_string()))
PublicKey::try_new(x_only).map_err(|e| KeycardWalletError::InvalidKeyMaterial(e.to_string()))
}
pub fn get_public_key_for_path_with_connect(pin: &str, path: &str) -> PyResult<PublicKey> {
Python::attach(|py| {
python_path::add_python_path(py)?;
let wallet = Self::new(py)?;
wallet.connect(py, pin)?;
let pub_key = wallet.get_public_key_for_path(py, path);
drop(wallet.close_session(py));
pub_key
})
pub fn get_public_key_for_path_with_connect(
pin: &str,
path: &str,
) -> Result<PublicKey, KeycardWalletError> {
let mut wallet = Self::new()?;
wallet.connect(pin)?;
wallet.get_public_key_for_path(path)
}
pub fn sign_message_for_path(
&self,
py: Python,
&mut self,
path: &str,
message: &[u8; 32],
) -> PyResult<(Signature, PublicKey)> {
let py_signature: Vec<u8> = self
.instance
.bind(py)
.call_method1("sign_message_for_path", (message, path))?
.extract()?;
) -> Result<(Signature, PublicKey), KeycardWalletError> {
let resp =
self.command_set
.sign_with_path_and_algo(message, path, sign_p2::BIP340_SCHNORR, false)?;
resp.check_ok()?;
let sig = Signature {
value: normalize_keycard_signature(py_signature)?,
value: parse_schnorr_signature(resp.data())?,
};
let pub_key = self.get_public_key_for_path(py, path)?;
let pub_key = self.get_public_key_for_path(path)?;
if !sig.is_valid_for(message, &pub_key) {
return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(
"keycard returned a signature that does not verify against its own public key",
));
return Err(KeycardWalletError::SignatureVerificationFailed);
}
Ok((sig, pub_key))
}
@ -162,39 +253,37 @@ impl KeycardWallet {
pin: &str,
path: &str,
message: &[u8; 32],
) -> PyResult<(Signature, PublicKey)> {
Python::attach(|py| {
python_path::add_python_path(py)?;
let wallet = Self::new(py)?;
wallet.connect(py, pin)?;
let result = wallet.sign_message_for_path(py, path, message);
drop(wallet.close_session(py));
result
})
) -> Result<(Signature, PublicKey), KeycardWalletError> {
let mut wallet = Self::new()?;
wallet.connect(pin)?;
wallet.sign_message_for_path(path, message)
}
pub fn load_mnemonic(&self, py: Python, mnemonic: &str) -> PyResult<()> {
self.instance
.bind(py)
.call_method1("load_mnemonic", (mnemonic,))?;
pub fn load_mnemonic(&mut self, mnemonic: &str) -> Result<(), KeycardWalletError> {
let mnemonic = bip39::Mnemonic::from_str(mnemonic)
.map_err(|e| KeycardWalletError::InvalidMnemonic(e.to_string()))?;
let seed = mnemonic.to_seed("");
self.command_set.load_lee_key(&seed)?.check_ok()?;
Ok(())
}
pub fn get_public_account_id_for_path_with_connect(
pin: &str,
key_path: &str,
) -> PyResult<String> {
) -> Result<String, KeycardWalletError> {
let public_key = Self::get_public_key_for_path_with_connect(pin, key_path)?;
Ok(format!("Public/{}", AccountId::from(&public_key)))
}
pub fn get_private_keys_for_path(&self, py: Python, path: &str) -> PyResult<PrivateKeyPair> {
let (raw_nsk, raw_vsk): (Vec<u8>, Vec<u8>) = self
.instance
.bind(py)
.call_method1("get_private_keys_for_path", (path,))?
.extract()?;
pub fn get_private_keys_for_path(&mut self, path: &str) -> Result<PrivateKeyPair, KeycardWalletError> {
let resp = self.command_set.export_lee_key(path)?;
resp.check_ok()?;
let mut reader = BerTlvReader::new(resp.data());
reader.enter_constructed(TLV_KEY_TEMPLATE)?;
let raw_nsk = reader.read_primitive(TLV_LEE_NSK)?;
let raw_vsk = reader.read_primitive(TLV_LEE_VSK)?;
let nsk = zeroizing_fixed_bytes::<32>("nullifier secret key", Zeroizing::new(raw_nsk))?;
let vsk = zeroizing_fixed_bytes::<64>("view secret key", Zeroizing::new(raw_vsk))?;
@ -205,46 +294,54 @@ impl KeycardWallet {
pub fn get_private_keys_for_path_with_connect(
pin: &str,
path: &str,
) -> PyResult<PrivateKeyPair> {
Python::attach(|py| {
python_path::add_python_path(py)?;
let wallet = Self::new(py)?;
wallet.connect(py, pin)?;
let result = wallet.get_private_keys_for_path(py, path);
drop(wallet.disconnect(py));
result
})
) -> Result<PrivateKeyPair, KeycardWalletError> {
let mut wallet = Self::new()?;
wallet.connect(pin)?;
let result = wallet.get_private_keys_for_path(path);
drop(wallet.disconnect());
result
}
}
/// The keycard Python library strips leading zeros from S when S < 2^(8k) for some k.
/// Left-pad S back to 32 bytes so the full signature is always 64 bytes (R || S).
#[expect(
clippy::arithmetic_side_effects,
reason = "64 - s_stripped.len() is safe: s_stripped.len() ≤ 31 because py_signature.len() is in [32, 63]"
)]
fn normalize_keycard_signature(py_signature: Vec<u8>) -> PyResult<[u8; 64]> {
if py_signature.len() < 64 {
if py_signature.len() < 32 {
return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
"signature from keycard too short: {} bytes",
py_signature.len()
)));
}
let s_stripped = &py_signature[32..];
let mut padded = [0_u8; 64];
padded[..32].copy_from_slice(&py_signature[..32]);
padded[(64 - s_stripped.len())..].copy_from_slice(s_stripped);
Ok(padded)
} else {
py_signature.try_into().map_err(|vec: Vec<u8>| {
PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
"Invalid signature length: expected 64 bytes, got {} (bytes: {:02x?})",
vec.len(),
vec
))
})
fn generate_puk() -> String {
let mut rng = rand::rngs::OsRng;
std::iter::repeat_with(|| char::from(rng.gen_range(b'0'..=b'9')))
.take(12)
.collect()
}
fn pairing_password() -> String {
std::env::var("KEYCARD_PAIRING_PASSWORD").unwrap_or_else(|_| DEFAULT_PAIRING_PASSWORD.to_owned())
}
/// Parses a BIP340 Schnorr signature from a LEE `SIGN` response.
///
/// Confirmed against real hardware: `TLV_SIGNATURE_TEMPLATE` (0xA0) contains the usual
/// `TLV_PUB_KEY` (0x80, 65 bytes, unused here — the caller fetches the pubkey separately via
/// `export_key`) followed by `TLV_LEE_RAW_SIGNATURE` (0x88, 64 bytes: `r || s` with no ASN.1/DER
/// wrapping). `keycard-rs`'s own `RecoverableSignature` parser doesn't apply — it's ECDSA-only
/// (attempts point recovery, which Schnorr doesn't have).
fn parse_schnorr_signature(data: &[u8]) -> Result<[u8; 64], KeycardWalletError> {
parse_schnorr_signature_inner(data).map_err(|e| {
KeycardWalletError::InvalidKeyMaterial(format!(
"failed to parse schnorr signature response ({e}); raw response bytes: {data:02x?}"
))
})
}
fn parse_schnorr_signature_inner(data: &[u8]) -> Result<[u8; 64], KeycardWalletError> {
let mut reader = BerTlvReader::new(data);
reader.enter_constructed(TLV_SIGNATURE_TEMPLATE)?;
if reader.next_tag_is(TLV_PUB_KEY) {
reader.read_primitive(TLV_PUB_KEY)?;
}
let sig = reader.read_primitive(TLV_LEE_RAW_SIGNATURE)?;
sig.try_into().map_err(|v: Vec<u8>| {
KeycardWalletError::InvalidKeyMaterial(format!(
"expected a 64-byte raw schnorr signature, got {} bytes",
v.len()
))
})
}
#[expect(
@ -254,9 +351,9 @@ fn normalize_keycard_signature(py_signature: Vec<u8>) -> PyResult<[u8; 64]> {
fn zeroizing_fixed_bytes<const N: usize>(
label: &str,
raw: Zeroizing<Vec<u8>>,
) -> PyResult<Zeroizing<[u8; N]>> {
) -> Result<Zeroizing<[u8; N]>, KeycardWalletError> {
if raw.len() != N {
return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
return Err(KeycardWalletError::InvalidKeyMaterial(format!(
"expected {N}-byte {label} from keycard, got {} bytes",
raw.len()
)));

View File

@ -1,69 +0,0 @@
use std::{env, path::PathBuf};
use pyo3::{prelude::*, types::PyList};
fn collect_python_paths() -> Vec<PathBuf> {
let current_dir = env::current_dir().expect("Failed to get current working directory");
let python_base = env::var("VIRTUAL_ENV")
.ok()
.and_then(|v| PathBuf::from(v).parent().map(PathBuf::from))
.unwrap_or_else(|| current_dir.clone());
let mut paths = vec![
python_base
.join("lez")
.join("keycard_wallet")
.join("python"),
python_base
.join("lez")
.join("keycard_wallet")
.join("python")
.join("keycard-py"),
];
// pyo3's embedded interpreter does not inherit sys.path from the shell,
// so venv site-packages must be added explicitly.
if let Ok(venv) = env::var("VIRTUAL_ENV") {
let lib = PathBuf::from(&venv).join("lib");
if let Ok(entries) = std::fs::read_dir(&lib) {
for entry in entries.flatten() {
let site_packages = entry.path().join("site-packages");
if site_packages.exists() {
paths.push(site_packages);
}
}
}
}
paths
}
/// Adds the project's `python/` directory and venv site-packages to Python's sys.path.
pub fn add_python_path(py: Python<'_>) -> PyResult<()> {
let paths = collect_python_paths();
for path in &paths {
if !path.exists() {
log::info!("Warning: Python path does not exist: {}", path.display());
}
}
let sys = PyModule::import(py, "sys")?;
let binding = sys.getattr("path")?;
let sys_path = binding.cast::<PyList>()?;
for path in &paths {
let path_str = path.to_str().expect("Invalid path");
let already_present = sys_path
.iter()
.any(|p| p.extract::<&str>().is_ok_and(|s| s == path_str));
if !already_present {
sys_path.insert(0, path_str)?;
}
}
Ok(())
}

View File

@ -7,8 +7,9 @@ the power-loss condition reported on some USB reader/driver combinations.
Either:
- pcscd re-powers the card on the next SCardConnect, so wallet
commands will succeed without triggering the retry path.
- the card stays unpowered, triggering TransportError
and exercising the retry wrapper in pair() / setup_communication_with_pairing().
- the card stays unpowered, triggering a PC/SC transport error
(keycard_rs::Error::Io) and exercising the reconnect-and-retry wrapper in
KeycardWallet::pair() / KeycardWallet::setup_communication_with_pairing().
"""
import sys
from smartcard.scard import (

View File

@ -5,8 +5,6 @@
# 1. Run wallet_with_keycard.sh once to install dependencies.
# 2. Keycard reader inserted with card loaded (wallet keycard load has been run).
source venv/bin/activate
cargo install --path lez/wallet --force --features keycard-debug
export KEYCARD_PIN=111111

View File

@ -1,8 +1,6 @@
#!/bin/bash
# Run wallet_with_keycard.sh first
source venv/bin/activate # Load the appropriate virtual environment
export KEYCARD_PIN=111111
# Tests wallet keycard available

View File

@ -23,7 +23,6 @@
# amm-lee-fund → public LEE holding used to seed the AMM pool
# (LP holding for amm new is created fresh each run — no persistent label)
source venv/bin/activate
export KEYCARD_PIN=111111
# =============================================================================

View File

@ -2,11 +2,9 @@
cargo install --path lez/wallet --force
# Install appropriate version of `keycard-py`.
git clone --branch lee-schnorr --single-branch https://github.com/bitgamma/keycard-py.git lez/keycard_wallet/python/keycard-py
# Set up virtual environment.
# `pyscard` is only needed by tests/force_unpower.py, a test-only helper that simulates a card
# power loss via PC/SC — the wallet CLI itself is pure Rust and talks to the card directly via
# `keycard-rs`.
python3 -m venv venv
source venv/bin/activate
pip install pyscard mnemonic ecdsa pyaes
pip install -e lez/keycard_wallet/python/keycard-py
pip install pyscard

View File

@ -25,7 +25,6 @@ system_accounts.workspace = true
associated_token_account_core.workspace = true
bip39.workspace = true
pyo3.workspace = true
rpassword = "7"
zeroize.workspace = true

View File

@ -2,7 +2,7 @@ use core::fmt;
use anyhow::Result;
use key_protocol::key_management::ephemeral_key_holder::EphemeralKeyHolder;
use keycard_wallet::{KeycardWallet, python_path};
use keycard_wallet::KeycardWallet;
use lee::{AccountId, PrivateKey, PublicKey, Signature};
use lee_core::{
CommitmentSetDigest, DUMMY_COMMITMENT_HASH, Identifier, InputAccountIdentity, MembershipProof,
@ -213,14 +213,7 @@ impl AccountManager {
if pin.is_none() {
pin = Some(
crate::helperfunctions::read_pin()
.map_err(|e| {
ExecutionFailureKind::KeycardError(pyo3::PyErr::new::<
pyo3::exceptions::PyRuntimeError,
_,
>(
e.to_string()
))
})?
.map_err(ExecutionFailureKind::SignError)?
.as_str()
.to_owned(),
);
@ -438,17 +431,11 @@ impl AccountManager {
.collect();
if let Some(pin) = self.pin.clone() {
pyo3::Python::attach(|py| -> pyo3::PyResult<()> {
python_path::add_python_path(py)?;
let wallet = KeycardWallet::new(py)?;
wallet.connect(py, &pin)?;
for path in keycard_paths {
sigs.push(wallet.sign_message_for_path(py, path, &message_hash)?);
}
let _res = wallet.close_session(py);
Ok(())
})
.map_err(anyhow::Error::from)?;
let mut wallet = KeycardWallet::new()?;
wallet.connect(&pin)?;
for path in keycard_paths {
sigs.push(wallet.sign_message_for_path(path, &message_hash)?);
}
}
Ok(sigs)

View File

@ -5,8 +5,7 @@
use anyhow::Result;
use clap::Subcommand;
use keycard_wallet::{KeycardWallet, clear_pairing, python_path};
use pyo3::prelude::*;
use keycard_wallet::{KeycardWallet, clear_pairing};
use crate::{
WalletCore,
@ -39,22 +38,11 @@ pub enum KeycardSubcommand {
impl KeycardSubcommand {
fn handle_available(_wallet_core: &mut WalletCore) -> SubcommandReturnValue {
Python::attach(|py| {
python_path::add_python_path(py)
.expect("`wallet::keycard::available`: unable to setup python path");
let wallet = KeycardWallet::new(py)
.expect("`wallet::keycard::available`: invalid data received for pin");
let available = wallet
.is_unpaired_keycard_available(py)
.expect("`wallet::keycard::available`: received invalid data from Keycard wrapper");
if available {
println!("\u{2705} Keycard is available.");
} else {
println!("\u{274c} Keycard is not available.");
}
});
if KeycardWallet::is_unpaired_keycard_available() {
println!("\u{2705} Keycard is available.");
} else {
println!("\u{274c} Keycard is not available.");
}
SubcommandReturnValue::Empty
}
@ -62,20 +50,9 @@ impl KeycardSubcommand {
fn handle_connect(_wallet_core: &mut WalletCore) -> Result<SubcommandReturnValue> {
let pin = read_pin()?;
Python::attach(|py| {
python_path::add_python_path(py)
.expect("`wallet::keycard::connect`: unable to setup python path");
let wallet = KeycardWallet::new(py)
.expect("`wallet::keycard::connect`: invalid keycard wallet provided");
wallet
.connect(py, &pin)
.expect("`wallet::keycard::connect`: failed to connect to keycard");
println!("\u{2705} Keycard paired and ready.");
drop(wallet.close_session(py));
});
let mut wallet = KeycardWallet::new()?;
wallet.connect(&pin)?;
println!("\u{2705} Keycard paired and ready.");
Ok(SubcommandReturnValue::Empty)
}
@ -83,24 +60,12 @@ impl KeycardSubcommand {
fn handle_disconnect(_wallet_core: &mut WalletCore) -> Result<SubcommandReturnValue> {
let pin = read_pin()?;
Python::attach(|py| {
python_path::add_python_path(py)
.expect("`wallet::keycard::disconnect`: unable to setup python path");
let mut wallet = KeycardWallet::new()?;
wallet.connect(&pin)?;
wallet.disconnect()?;
let wallet = KeycardWallet::new(py)
.expect("`wallet::keycard::disconnect`: invalid keycard wallet provided");
wallet
.connect(py, &pin)
.expect("`wallet::keycard::disconnect`: failed to open session");
wallet
.disconnect(py)
.expect("`wallet::keycard::disconnect`: failed to unpair keycard");
clear_pairing();
println!("\u{2705} Keycard unpaired and pairing cleared.");
});
clear_pairing();
println!("\u{2705} Keycard unpaired and pairing cleared.");
Ok(SubcommandReturnValue::Empty)
}
@ -108,22 +73,13 @@ impl KeycardSubcommand {
fn handle_init(_wallet_core: &mut WalletCore) -> Result<SubcommandReturnValue> {
let pin = read_pin()?;
Python::attach(|py| {
python_path::add_python_path(py)
.expect("`wallet::keycard::init`: unable to setup python path");
let mut wallet = KeycardWallet::new()?;
let puk = wallet.initialize(&pin)?;
let wallet = KeycardWallet::new(py)
.expect("`wallet::keycard::init`: invalid keycard wallet provided");
let initialized = wallet
.initialize(py, &pin)
.expect("`wallet::keycard::init`: failed to initialize keycard");
if initialized {
clear_pairing();
println!("\u{2705} Keycard initialized successfully.");
}
});
clear_pairing();
println!("Keycard PUK: {puk}");
println!("Record this PUK and store it somewhere safe. It cannot be recovered.");
println!("\u{2705} Keycard initialized successfully.");
Ok(SubcommandReturnValue::Empty)
}
@ -132,25 +88,15 @@ impl KeycardSubcommand {
let pin = read_pin()?;
let mnemonic = read_mnemonic()?;
Python::attach(|py| {
python_path::add_python_path(py)
.expect("`wallet::keycard::load`: unable to setup python path");
let mut wallet = KeycardWallet::new()?;
wallet.connect(&pin)?;
println!("\u{2705} Keycard is now connected to wallet.");
let wallet = KeycardWallet::new(py)
.expect("`wallet::keycard::load`: invalid keycard wallet provided");
wallet
.connect(py, &pin)
.expect("`wallet::keycard::load`: failed to connect to keycard");
println!("\u{2705} Keycard is now connected to wallet.");
if wallet.load_mnemonic(py, &mnemonic).is_ok() {
println!("\u{2705} Mnemonic phrase loaded successfully.");
} else {
println!("\u{274c} Failed to load mnemonic phrase.");
}
drop(wallet.close_session(py));
});
if wallet.load_mnemonic(&mnemonic).is_ok() {
println!("\u{2705} Mnemonic phrase loaded successfully.");
} else {
println!("\u{274c} Failed to load mnemonic phrase.");
}
Ok(SubcommandReturnValue::Empty)
}

View File

@ -79,8 +79,6 @@ pub enum ExecutionFailureKind {
TransactionBuildError(#[from] lee::error::LeeError),
#[error("Failed to sign transaction: {0}")]
SignError(anyhow::Error),
#[error(transparent)]
KeycardError(#[from] pyo3::PyErr),
}
#[expect(clippy::partial_pub_fields, reason = "TODO: make all fields private")]

View File

@ -1,5 +1,4 @@
use keycard_wallet::{KeycardWallet, python_path};
use pyo3::Python;
use keycard_wallet::{KeycardWallet, KeycardWalletError};
/// Lazily opens and reuses a single Keycard session for all keycard signers in one transaction.
pub struct KeycardSessionContext {
@ -15,19 +14,18 @@ impl KeycardSessionContext {
}
}
pub fn get_or_connect(&mut self, py: Python<'_>) -> pyo3::PyResult<&KeycardWallet> {
pub fn get_or_connect(&mut self) -> Result<&mut KeycardWallet, KeycardWalletError> {
if self.wallet.is_none() {
python_path::add_python_path(py)?;
let wallet = KeycardWallet::new(py)?;
wallet.connect(py, &self.pin)?;
let mut wallet = KeycardWallet::new()?;
wallet.connect(&self.pin)?;
self.wallet = Some(wallet);
}
Ok(self.wallet.as_ref().expect("wallet was just inserted"))
Ok(self.wallet.as_mut().expect("wallet was just inserted"))
}
pub fn close(self, py: Python<'_>) {
if let Some(w) = self.wallet {
let _res = w.close_session(py);
pub fn close(mut self) {
if let Some(wallet) = self.wallet.as_mut() {
drop(wallet.disconnect());
}
}
}