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 40595604a4
commit 653c5d7b95
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

@ -178,6 +178,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-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" } 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 = [ rocksdb = { version = "0.24.0", default-features = false, features = [
"snappy", "snappy",
"bindgen-runtime", "bindgen-runtime",
@ -199,7 +201,6 @@ actix-web = { version = "4.13.0", default-features = false, features = [
axum = "0.8.4" axum = "0.8.4"
clap = { version = "4.5.42", features = ["derive", "env"] } clap = { version = "4.5.42", features = ["derive", "env"] }
reqwest = { version = "0.12", features = ["json", "rustls-tls", "stream"] } reqwest = { version = "0.12", features = ["json", "rustls-tls", "stream"] }
pyo3 = { version = "0.29", features = ["auto-initialize"] }
zeroize = "1" zeroize = "1"
criterion = { version = "0.8", features = ["html_reports"] } criterion = { version = "0.8", features = ["html_reports"] }

View File

@ -9,8 +9,11 @@ workspace = true
[dependencies] [dependencies]
lee.workspace = true lee.workspace = true
pyo3.workspace = true keycard-rs.workspace = true
bip39.workspace = true
log.workspace = true log.workspace = true
rand.workspace = true
serde = { workspace = true, features = ["derive"] } serde = { workspace = true, features = ["derive"] }
serde_json.workspace = true serde_json.workspace = true
thiserror.workspace = true
zeroize.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 lee::{AccountId, PublicKey, Signature};
use pyo3::{prelude::*, types::PyAny}; use rand::Rng as _;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use zeroize::Zeroizing; 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 /// NSK (32 bytes) and VSK (64 bytes, the ML-KEM-768 seed `d || z`) as fixed-length zeroizing byte
/// arrays. /// arrays.
type PrivateKeyPair = (Zeroizing<[u8; 32]>, Zeroizing<[u8; 64]>); 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. // TODO: encrypt at rest alongside broader wallet storage encryption work.
#[derive(Serialize, Deserialize)] #[derive(Serialize, Deserialize)]
pub struct KeycardPairingData { 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 { pub struct KeycardWallet {
instance: Py<PyAny>, command_set: KeycardCommandSet,
} }
impl KeycardWallet { impl KeycardWallet {
/// Create a new Python `KeycardWallet` instance. /// Connects to the first available PC/SC reader. Does not select the applet yet — callers
pub fn new(py: Python) -> PyResult<Self> { /// that need application info (`initialize`, `pair`, `connect`, ...) do that themselves.
let module = py.import("keycard_wallet")?; pub fn new() -> Result<Self, KeycardWalletError> {
let class = module.getattr("KeycardWallet")?; let channel = PcscChannel::connect()?;
let instance = class.call0()?;
Ok(Self { Ok(Self {
instance: instance.into(), command_set: KeycardCommandSet::new(channel),
}) })
} }
pub fn is_unpaired_keycard_available(&self, py: Python) -> PyResult<bool> { /// Returns whether a smart card reader and a selectable Keycard are both present.
self.instance #[must_use]
.bind(py) pub fn is_unpaired_keycard_available() -> bool {
.call_method0("is_unpaired_keycard_available")? let Ok(channel) = PcscChannel::connect() else {
.extract() return false;
};
KeycardCommandSet::new(channel)
.select()
.is_ok_and(|resp| resp.is_ok())
} }
pub fn initialize(&self, py: Python<'_>, pin: &str) -> PyResult<bool> { fn select(&mut self) -> Result<(), KeycardWalletError> {
self.instance self.command_set.select()?.check_ok()?;
.bind(py) Ok(())
.call_method1("initialize", (pin,))?
.extract()
} }
pub fn pair(&self, py: Python<'_>, pin: &str) -> PyResult<(u8, Vec<u8>)> { fn open_and_verify(&mut self, pin: &str) -> Result<(), KeycardWalletError> {
self.instance self.command_set.auto_open_secure_channel()?;
.bind(py) self.command_set.verify_pin(pin)?.check_auth_ok()?;
.call_method1("pair", (pin,))? Ok(())
.extract() }
/// 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( pub fn setup_communication_with_pairing(
&self, &mut self,
py: Python<'_>,
pin: &str, pin: &str,
index: u8, index: u8,
key: &[u8], key: &[u8; 32],
) -> PyResult<bool> { ) -> Result<(), KeycardWalletError> {
self.instance self.select()?;
.bind(py) self.command_set.set_pairing(Pairing::new(*key, index));
.call_method1( self.open_and_verify(pin)
"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()
} }
/// Connect using a stored pairing if available, falling back to a fresh pair. /// Connect using a stored pairing if available, falling back to a fresh pair.
/// Saves any newly established pairing to disk. /// Saves any newly established pairing to disk.
pub fn connect(&self, py: Python<'_>, pin: &str) -> PyResult<()> { pub fn connect(&mut self, pin: &str) -> Result<(), KeycardWalletError> {
if let Some(pairing) = load_pairing().filter(KeycardPairingData::is_valid) if let Some(pairing) = load_pairing().filter(KeycardPairingData::is_valid) {
&& self let key: [u8; 32] = pairing
.setup_communication_with_pairing(py, pin, pairing.index, &pairing.key) .key
.is_ok() .clone()
{ .try_into()
return Ok(()); .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(()) Ok(())
} }
pub fn disconnect(&self, py: Python) -> PyResult<bool> { /// Unpairs the current session. Returns `false` if there was nothing to unpair.
self.instance.bind(py).call_method0("disconnect")?.extract() 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> { pub fn get_public_key_for_path(&mut self, path: &str) -> Result<PublicKey, KeycardWalletError> {
let public_key: Vec<u8> = self let resp = self.command_set.export_key(path, false, true)?;
.instance resp.check_ok()?;
.bind(py) let keypair = Bip32KeyPair::from_tlv(resp.data())?;
.call_method1("get_public_key_for_path", (path,))?
.extract()?;
let public_key: [u8; 32] = public_key.try_into().map_err(|vec: Vec<u8>| { // Uncompressed SEC1 point (0x04 || X || Y); the BIP340 x-only public key is just its X
PyErr::new::<pyo3::exceptions::PyValueError, _>(format!( // coordinate, since secp256k1 (what the card signs with) and k256's Schnorr verifying key
"expected 32-byte public key from keycard, got {} bytes", // are the same curve.
vec.len() 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) PublicKey::try_new(x_only).map_err(|e| KeycardWalletError::InvalidKeyMaterial(e.to_string()))
.map_err(|e| PyErr::new::<pyo3::exceptions::PyValueError, _>(e.to_string()))
} }
pub fn get_public_key_for_path_with_connect(pin: &str, path: &str) -> PyResult<PublicKey> { pub fn get_public_key_for_path_with_connect(
Python::attach(|py| { pin: &str,
python_path::add_python_path(py)?; path: &str,
let wallet = Self::new(py)?; ) -> Result<PublicKey, KeycardWalletError> {
wallet.connect(py, pin)?; let mut wallet = Self::new()?;
let pub_key = wallet.get_public_key_for_path(py, path); wallet.connect(pin)?;
drop(wallet.close_session(py)); wallet.get_public_key_for_path(path)
pub_key
})
} }
pub fn sign_message_for_path( pub fn sign_message_for_path(
&self, &mut self,
py: Python,
path: &str, path: &str,
message: &[u8; 32], message: &[u8; 32],
) -> PyResult<(Signature, PublicKey)> { ) -> Result<(Signature, PublicKey), KeycardWalletError> {
let py_signature: Vec<u8> = self let resp =
.instance self.command_set
.bind(py) .sign_with_path_and_algo(message, path, sign_p2::BIP340_SCHNORR, false)?;
.call_method1("sign_message_for_path", (message, path))? resp.check_ok()?;
.extract()?;
let sig = Signature { 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) { if !sig.is_valid_for(message, &pub_key) {
return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>( return Err(KeycardWalletError::SignatureVerificationFailed);
"keycard returned a signature that does not verify against its own public key",
));
} }
Ok((sig, pub_key)) Ok((sig, pub_key))
} }
@ -162,39 +253,37 @@ impl KeycardWallet {
pin: &str, pin: &str,
path: &str, path: &str,
message: &[u8; 32], message: &[u8; 32],
) -> PyResult<(Signature, PublicKey)> { ) -> Result<(Signature, PublicKey), KeycardWalletError> {
Python::attach(|py| { let mut wallet = Self::new()?;
python_path::add_python_path(py)?; wallet.connect(pin)?;
let wallet = Self::new(py)?; wallet.sign_message_for_path(path, message)
wallet.connect(py, pin)?;
let result = wallet.sign_message_for_path(py, path, message);
drop(wallet.close_session(py));
result
})
} }
pub fn load_mnemonic(&self, py: Python, mnemonic: &str) -> PyResult<()> { pub fn load_mnemonic(&mut self, mnemonic: &str) -> Result<(), KeycardWalletError> {
self.instance let mnemonic = bip39::Mnemonic::from_str(mnemonic)
.bind(py) .map_err(|e| KeycardWalletError::InvalidMnemonic(e.to_string()))?;
.call_method1("load_mnemonic", (mnemonic,))?; let seed = mnemonic.to_seed("");
self.command_set.load_lee_key(&seed)?.check_ok()?;
Ok(()) Ok(())
} }
pub fn get_public_account_id_for_path_with_connect( pub fn get_public_account_id_for_path_with_connect(
pin: &str, pin: &str,
key_path: &str, key_path: &str,
) -> PyResult<String> { ) -> Result<String, KeycardWalletError> {
let public_key = Self::get_public_key_for_path_with_connect(pin, key_path)?; let public_key = Self::get_public_key_for_path_with_connect(pin, key_path)?;
Ok(format!("Public/{}", AccountId::from(&public_key))) Ok(format!("Public/{}", AccountId::from(&public_key)))
} }
pub fn get_private_keys_for_path(&self, py: Python, path: &str) -> PyResult<PrivateKeyPair> { pub fn get_private_keys_for_path(&mut self, path: &str) -> Result<PrivateKeyPair, KeycardWalletError> {
let (raw_nsk, raw_vsk): (Vec<u8>, Vec<u8>) = self let resp = self.command_set.export_lee_key(path)?;
.instance resp.check_ok()?;
.bind(py)
.call_method1("get_private_keys_for_path", (path,))? let mut reader = BerTlvReader::new(resp.data());
.extract()?; 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 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))?; 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( pub fn get_private_keys_for_path_with_connect(
pin: &str, pin: &str,
path: &str, path: &str,
) -> PyResult<PrivateKeyPair> { ) -> Result<PrivateKeyPair, KeycardWalletError> {
Python::attach(|py| { let mut wallet = Self::new()?;
python_path::add_python_path(py)?; wallet.connect(pin)?;
let wallet = Self::new(py)?; let result = wallet.get_private_keys_for_path(path);
wallet.connect(py, pin)?; drop(wallet.disconnect());
let result = wallet.get_private_keys_for_path(py, path); result
drop(wallet.disconnect(py));
result
})
} }
} }
/// The keycard Python library strips leading zeros from S when S < 2^(8k) for some k. fn generate_puk() -> String {
/// Left-pad S back to 32 bytes so the full signature is always 64 bytes (R || S). let mut rng = rand::rngs::OsRng;
#[expect( std::iter::repeat_with(|| char::from(rng.gen_range(b'0'..=b'9')))
clippy::arithmetic_side_effects, .take(12)
reason = "64 - s_stripped.len() is safe: s_stripped.len() ≤ 31 because py_signature.len() is in [32, 63]" .collect()
)] }
fn normalize_keycard_signature(py_signature: Vec<u8>) -> PyResult<[u8; 64]> {
if py_signature.len() < 64 { fn pairing_password() -> String {
if py_signature.len() < 32 { std::env::var("KEYCARD_PAIRING_PASSWORD").unwrap_or_else(|_| DEFAULT_PAIRING_PASSWORD.to_owned())
return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(format!( }
"signature from keycard too short: {} bytes",
py_signature.len() /// Parses a BIP340 Schnorr signature from a LEE `SIGN` response.
))); ///
} /// Confirmed against real hardware: `TLV_SIGNATURE_TEMPLATE` (0xA0) contains the usual
let s_stripped = &py_signature[32..]; /// `TLV_PUB_KEY` (0x80, 65 bytes, unused here — the caller fetches the pubkey separately via
let mut padded = [0_u8; 64]; /// `export_key`) followed by `TLV_LEE_RAW_SIGNATURE` (0x88, 64 bytes: `r || s` with no ASN.1/DER
padded[..32].copy_from_slice(&py_signature[..32]); /// wrapping). `keycard-rs`'s own `RecoverableSignature` parser doesn't apply — it's ECDSA-only
padded[(64 - s_stripped.len())..].copy_from_slice(s_stripped); /// (attempts point recovery, which Schnorr doesn't have).
Ok(padded) fn parse_schnorr_signature(data: &[u8]) -> Result<[u8; 64], KeycardWalletError> {
} else { parse_schnorr_signature_inner(data).map_err(|e| {
py_signature.try_into().map_err(|vec: Vec<u8>| { KeycardWalletError::InvalidKeyMaterial(format!(
PyErr::new::<pyo3::exceptions::PyValueError, _>(format!( "failed to parse schnorr signature response ({e}); raw response bytes: {data:02x?}"
"Invalid signature length: expected 64 bytes, got {} (bytes: {:02x?})", ))
vec.len(), })
vec }
))
}) 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( #[expect(
@ -254,9 +351,9 @@ fn normalize_keycard_signature(py_signature: Vec<u8>) -> PyResult<[u8; 64]> {
fn zeroizing_fixed_bytes<const N: usize>( fn zeroizing_fixed_bytes<const N: usize>(
label: &str, label: &str,
raw: Zeroizing<Vec<u8>>, raw: Zeroizing<Vec<u8>>,
) -> PyResult<Zeroizing<[u8; N]>> { ) -> Result<Zeroizing<[u8; N]>, KeycardWalletError> {
if raw.len() != N { 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", "expected {N}-byte {label} from keycard, got {} bytes",
raw.len() 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: Either:
- pcscd re-powers the card on the next SCardConnect, so wallet - pcscd re-powers the card on the next SCardConnect, so wallet
commands will succeed without triggering the retry path. commands will succeed without triggering the retry path.
- the card stays unpowered, triggering TransportError - the card stays unpowered, triggering a PC/SC transport error
and exercising the retry wrapper in pair() / setup_communication_with_pairing(). (keycard_rs::Error::Io) and exercising the reconnect-and-retry wrapper in
KeycardWallet::pair() / KeycardWallet::setup_communication_with_pairing().
""" """
import sys import sys
from smartcard.scard import ( from smartcard.scard import (

View File

@ -5,8 +5,6 @@
# 1. Run wallet_with_keycard.sh once to install dependencies. # 1. Run wallet_with_keycard.sh once to install dependencies.
# 2. Keycard reader inserted with card loaded (wallet keycard load has been run). # 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 cargo install --path lez/wallet --force --features keycard-debug
export KEYCARD_PIN=111111 export KEYCARD_PIN=111111

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -1,5 +1,4 @@
use keycard_wallet::{KeycardWallet, python_path}; use keycard_wallet::{KeycardWallet, KeycardWalletError};
use pyo3::Python;
/// Lazily opens and reuses a single Keycard session for all keycard signers in one transaction. /// Lazily opens and reuses a single Keycard session for all keycard signers in one transaction.
pub struct KeycardSessionContext { 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() { if self.wallet.is_none() {
python_path::add_python_path(py)?; let mut wallet = KeycardWallet::new()?;
let wallet = KeycardWallet::new(py)?; wallet.connect(&self.pin)?;
wallet.connect(py, &self.pin)?;
self.wallet = Some(wallet); 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<'_>) { pub fn close(mut self) {
if let Some(w) = self.wallet { if let Some(wallet) = self.wallet.as_mut() {
let _res = w.close_session(py); drop(wallet.disconnect());
} }
} }
} }