mirror of
https://github.com/cryptoadvance/specter-desktop.git
synced 2026-08-13 12:33:29 +02:00
Feature: Add Jade multisig support (#1520)
* update jadepy library * display multisig * add multisig change detection when signing tx with Jade * requests is available * update udev rules jade and hwi/jade.py * black * sort signers everywhere
This commit is contained in:
Stepan Snigirev
GitHub
parent
1cf27bfc7f
commit
57091ff2b8
5 changed files with 553 additions and 193 deletions
|
|
@ -3,11 +3,13 @@ Blockstream Jade Devices
|
|||
************************
|
||||
"""
|
||||
|
||||
from .jadepy import jade
|
||||
from .jadepy.jade import JadeAPI, JadeError
|
||||
from serial.tools import list_ports
|
||||
|
||||
from typing import List, Union
|
||||
from hwilib.descriptor import PubkeyProvider
|
||||
from functools import wraps
|
||||
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
|
||||
from hwilib.descriptor import PubkeyProvider, MultisigDescriptor
|
||||
from hwilib.hwwclient import HardwareWalletClient
|
||||
from hwilib.errors import (
|
||||
ActionCanceledError,
|
||||
|
|
@ -22,7 +24,7 @@ from hwilib.common import (
|
|||
AddressType,
|
||||
Chain,
|
||||
)
|
||||
from hwilib.key import ExtendedKey, parse_path
|
||||
from hwilib.key import ExtendedKey, parse_path, KeyOriginInfo, is_hardened
|
||||
from hwilib.psbt import PSBT
|
||||
from hwilib.tx import CTransaction
|
||||
from hwilib._script import (
|
||||
|
|
@ -30,10 +32,12 @@ from hwilib._script import (
|
|||
is_p2wpkh,
|
||||
is_p2wsh,
|
||||
is_witness,
|
||||
parse_multisig,
|
||||
)
|
||||
|
||||
import logging
|
||||
import os
|
||||
import hashlib
|
||||
|
||||
# embit-related things
|
||||
from embit import ec, bip32
|
||||
|
|
@ -44,8 +48,11 @@ from embit import hashes
|
|||
from embit.util import secp256k1
|
||||
from embit.liquid.finalizer import finalize_psbt
|
||||
|
||||
JADE_VENDOR_ID = 0x10C4
|
||||
JADE_DEVICE_ID = 0xEA60
|
||||
# The test emulator port
|
||||
SIMULATOR_PATH = "tcp:127.0.0.1:2222"
|
||||
|
||||
JADE_DEVICE_IDS = [(0x10C4, 0xEA60), (0x1A86, 0x55D4)]
|
||||
HAS_NETWORKING = hasattr(jade, "_http_request")
|
||||
|
||||
py_enumerate = (
|
||||
enumerate # To use the enumerate built-in, since the name is overridden below
|
||||
|
|
@ -54,28 +61,29 @@ py_enumerate = (
|
|||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def jade_exception(f):
|
||||
def func(*args, **kwargs):
|
||||
def jade_exception(f: Callable[..., Any]) -> Any:
|
||||
@wraps(f)
|
||||
def func(*args: Any, **kwargs: Any) -> Any:
|
||||
try:
|
||||
return f(*args, **kwargs)
|
||||
except ValueError as e:
|
||||
raise BadArgumentError(str(e))
|
||||
except JadeError as e:
|
||||
if e.code == -32000: # CBOR_RPC_USER_CANCELLED
|
||||
raise ActionCanceledError("{} canceled by user".format(f.__name__))
|
||||
elif e.code == -32602: # CBOR_RPC_BAD_PARAMETERS
|
||||
if e.code == JadeError.USER_CANCELLED:
|
||||
raise ActionCanceledError(f"{f.__name__} canceled by user")
|
||||
elif e.code == JadeError.BAD_PARAMETERS:
|
||||
raise BadArgumentError(e.message)
|
||||
elif e.code == -32603: # CBOR_RPC_INTERNAL_ERROR
|
||||
elif e.code == JadeError.INTERNAL_ERROR:
|
||||
raise DeviceFailureError(e.message)
|
||||
elif e.code == -32002: # CBOR_RPC_HW_LOCKED
|
||||
elif e.code == JadeError.HW_LOCKED:
|
||||
raise DeviceConnectionError("Device is locked")
|
||||
elif e.code == -32003: # CBOR_RPC_NETWORK_MISMATCH
|
||||
elif e.code == JadeError.NETWORK_MISMATCH:
|
||||
raise DeviceConnectionError("Network/chain selection error")
|
||||
elif e.code in [
|
||||
-32600,
|
||||
-32601,
|
||||
-32001,
|
||||
]: # CBOR_RPC_INVALID_REQUEST, CBOR_RPC_UNKNOWN_METHOD, CBOR_RPC_PROTOCOL_ERROR
|
||||
JadeError.INVALID_REQUEST,
|
||||
JadeError.UNKNOWN_METHOD,
|
||||
JadeError.PROTOCOL_ERROR,
|
||||
]:
|
||||
raise DeviceConnectionError("Messaging/communiciation error")
|
||||
else:
|
||||
raise e
|
||||
|
|
@ -86,7 +94,11 @@ def jade_exception(f):
|
|||
# This class extends the HardwareWalletClient for Blockstream Jade specific things
|
||||
class JadeClient(HardwareWalletClient):
|
||||
|
||||
NETWORKS = {Chain.MAIN: "mainnet", Chain.TEST: "testnet", Chain.REGTEST: "regtest"}
|
||||
NETWORKS = {
|
||||
Chain.MAIN: "mainnet",
|
||||
Chain.TEST: "testnet",
|
||||
Chain.REGTEST: "localtest",
|
||||
}
|
||||
liquid_network = None
|
||||
|
||||
def set_liquid_network(self, chain):
|
||||
|
|
@ -102,11 +114,31 @@ class JadeClient(HardwareWalletClient):
|
|||
AddressType.WIT: "wpkh(k)",
|
||||
AddressType.SH_WIT: "sh(wpkh(k))",
|
||||
}
|
||||
MULTI_ADDRTYPES = {
|
||||
AddressType.LEGACY: "sh(multi(k))",
|
||||
AddressType.WIT: "wsh(multi(k))",
|
||||
AddressType.SH_WIT: "sh(wsh(multi(k)))",
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _convertAddrType(addrType):
|
||||
def _convertAddrType(addrType, multisig=False):
|
||||
if multisig:
|
||||
return JadeClient.MULTI_ADDRTYPES[addrType]
|
||||
return JadeClient.ADDRTYPES[addrType]
|
||||
|
||||
@staticmethod
|
||||
def _get_multisig_name(
|
||||
type: str, threshold: int, signers: List[Tuple[bytes, Sequence[int]]]
|
||||
) -> str:
|
||||
# Concatenate script-type, threshold, and all signers fingerprints and derivation paths (sorted)
|
||||
summary = type + "|" + str(threshold) + "|"
|
||||
for fingerprint, path in sorted(signers):
|
||||
summary += fingerprint.hex() + "|" + str(path) + "|"
|
||||
|
||||
# Hash it, get the first 6-bytes as hex, prepend with 'hwi'
|
||||
hash_summary = hashlib.sha256(summary.encode()).digest().hex()
|
||||
return "hwi" + hash_summary[:12]
|
||||
|
||||
def __init__(self, path: str, password: str = "", expert: bool = False) -> None:
|
||||
super(JadeClient, self).__init__(path, password, expert)
|
||||
self.jade = JadeAPI.create_serial(path)
|
||||
|
|
@ -151,104 +183,306 @@ class JadeClient(HardwareWalletClient):
|
|||
def sign_tx(self, tx: PSBT) -> PSBT:
|
||||
"""
|
||||
Sign a transaction with the Blockstream Jade.
|
||||
|
||||
- Jade can only be used to sign single-key inputs at this time. It cannot sign multisig or arbitrary scripts.
|
||||
"""
|
||||
# Helper to get multisig record for change output
|
||||
def _parse_signers(
|
||||
hd_keypath_origins: List[KeyOriginInfo],
|
||||
) -> Tuple[List[Tuple[bytes, Sequence[int]]], List[Sequence[int]]]:
|
||||
# Split the path at the last hardened path element
|
||||
def _split_at_last_hardened_element(
|
||||
path: Sequence[int],
|
||||
) -> Tuple[Sequence[int], Sequence[int]]:
|
||||
for i in range(len(path), 0, -1):
|
||||
if is_hardened(path[i - 1]):
|
||||
return (path[:i], path[i:])
|
||||
return ([], path)
|
||||
|
||||
signers = []
|
||||
paths = []
|
||||
for origin in hd_keypath_origins:
|
||||
prefix, suffix = _split_at_last_hardened_element(origin.path)
|
||||
signers.append((origin.fingerprint, prefix))
|
||||
paths.append(suffix)
|
||||
# sort signers and paths like in multisig registration
|
||||
signers, paths = [list(a) for a in zip(*sorted(zip(signers, paths)))]
|
||||
|
||||
return signers, paths
|
||||
|
||||
c_txn = CTransaction(tx.tx)
|
||||
master_fp = self.get_master_fingerprint()
|
||||
signing_pubkeys = [None] * len(tx.inputs)
|
||||
signing_singlesigs = False
|
||||
signing_multisigs = {}
|
||||
need_to_sign = True
|
||||
|
||||
# Signing input details
|
||||
jade_inputs = []
|
||||
for n_vin, (txin, psbtin) in py_enumerate(zip(c_txn.vin, tx.inputs)):
|
||||
# Get bip32 path to use to sign, if required for this input
|
||||
path = None
|
||||
for pubkey, origin in psbtin.hd_keypaths.items():
|
||||
if origin.fingerprint == master_fp and len(origin.path) > 0:
|
||||
# Our input
|
||||
if (
|
||||
pubkey not in psbtin.partial_sigs
|
||||
or not psbtin.partial_sigs[pubkey]
|
||||
):
|
||||
# hw to sign this input - it is not already signed
|
||||
signing_pubkeys[n_vin] = pubkey
|
||||
path = origin.path
|
||||
while need_to_sign:
|
||||
signing_pubkeys: List[Optional[bytes]] = [None] * len(tx.inputs)
|
||||
need_to_sign = False
|
||||
|
||||
# Get the tx and prevout/scriptcode
|
||||
utxo = None
|
||||
input_txn_bytes = None
|
||||
if psbtin.witness_utxo:
|
||||
utxo = psbtin.witness_utxo
|
||||
if psbtin.non_witness_utxo:
|
||||
if txin.prevout.hash != psbtin.non_witness_utxo.sha256:
|
||||
raise BadArgumentError(
|
||||
"Input {} has a non_witness_utxo with the wrong hash".format(
|
||||
n_vin
|
||||
)
|
||||
# Signing input details
|
||||
jade_inputs = []
|
||||
for n_vin, (txin, psbtin) in py_enumerate(zip(c_txn.vin, tx.inputs)):
|
||||
# Get bip32 path to use to sign, if required for this input
|
||||
path = None
|
||||
multisig_input = len(psbtin.hd_keypaths) > 1
|
||||
for pubkey, origin in psbtin.hd_keypaths.items():
|
||||
if origin.fingerprint == master_fp and len(origin.path) > 0:
|
||||
if not multisig_input:
|
||||
signing_singlesigs = True
|
||||
|
||||
if psbtin.partial_sigs.get(pubkey, None) is None:
|
||||
# hw to sign this input - it is not already signed
|
||||
if signing_pubkeys[n_vin] is None:
|
||||
signing_pubkeys[n_vin] = pubkey
|
||||
path = origin.path
|
||||
else:
|
||||
# Additional signature needed for this input - ie. a multisig where this wallet is
|
||||
# multiple signers? Clumsy, but just loop and go through the signing procedure again.
|
||||
need_to_sign = True
|
||||
|
||||
# Get the tx and prevout/scriptcode
|
||||
utxo = None
|
||||
p2sh = False
|
||||
input_txn_bytes = None
|
||||
if psbtin.witness_utxo:
|
||||
utxo = psbtin.witness_utxo
|
||||
if psbtin.non_witness_utxo:
|
||||
utxo = psbtin.non_witness_utxo.vout[txin.prevout.n]
|
||||
input_txn_bytes = (
|
||||
psbtin.non_witness_utxo.serialize_without_witness()
|
||||
)
|
||||
utxo = psbtin.non_witness_utxo.vout[txin.prevout.n]
|
||||
input_txn_bytes = psbtin.non_witness_utxo.serialize_without_witness()
|
||||
if utxo is None:
|
||||
raise Exception(
|
||||
"PSBT is missing input utxo information, cannot sign"
|
||||
)
|
||||
scriptcode = utxo.scriptPubKey
|
||||
|
||||
scriptcode = utxo.scriptPubKey
|
||||
if is_p2sh(scriptcode):
|
||||
scriptcode = psbtin.redeem_script
|
||||
p2sh = True
|
||||
|
||||
if is_p2sh(scriptcode):
|
||||
scriptcode = psbtin.redeem_script
|
||||
witness_input, witness_version, witness_program = is_witness(scriptcode)
|
||||
|
||||
witness_input, witness_version, witness_program = is_witness(scriptcode)
|
||||
if witness_input:
|
||||
if is_p2wsh(scriptcode):
|
||||
scriptcode = psbtin.witness_script
|
||||
elif is_p2wpkh(scriptcode):
|
||||
scriptcode = b"\x76\xa9\x14" + witness_program + b"\x88\xac"
|
||||
else:
|
||||
continue
|
||||
|
||||
if witness_input:
|
||||
if is_p2wsh(scriptcode):
|
||||
scriptcode = psbtin.witness_script
|
||||
elif is_p2wpkh(scriptcode):
|
||||
scriptcode = b"\x76\xa9\x14" + witness_program + b"\x88\xac"
|
||||
else:
|
||||
scriptcode = None
|
||||
# If we are signing a multisig input, deduce the potential
|
||||
# registration details and cache as a potential change wallet
|
||||
if multisig_input and path and scriptcode and (p2sh or witness_input):
|
||||
parsed = parse_multisig(scriptcode)
|
||||
if parsed:
|
||||
addr_type = (
|
||||
AddressType.LEGACY
|
||||
if not witness_input
|
||||
else AddressType.WIT
|
||||
if not p2sh
|
||||
else AddressType.SH_WIT
|
||||
)
|
||||
script_variant = self._convertAddrType(addr_type, multisig=True)
|
||||
threshold = parsed[0]
|
||||
|
||||
# Build the input and add to the list
|
||||
jade_inputs.append(
|
||||
{
|
||||
"is_witness": witness_input,
|
||||
"input_tx": input_txn_bytes,
|
||||
"script": scriptcode,
|
||||
"path": path,
|
||||
pubkeys = parsed[1]
|
||||
hd_keypath_origins = [
|
||||
psbtin.hd_keypaths[pubkey] for pubkey in pubkeys
|
||||
]
|
||||
|
||||
signers, paths = _parse_signers(hd_keypath_origins)
|
||||
|
||||
multisig_name = self._get_multisig_name(
|
||||
script_variant, threshold, signers
|
||||
)
|
||||
signing_multisigs[multisig_name] = (
|
||||
script_variant,
|
||||
threshold,
|
||||
signers,
|
||||
)
|
||||
|
||||
# Build the input and add to the list - include some host entropy for AE sigs (although we won't verify)
|
||||
jade_inputs.append(
|
||||
{
|
||||
"is_witness": witness_input,
|
||||
"input_tx": input_txn_bytes,
|
||||
"script": scriptcode,
|
||||
"path": path,
|
||||
"ae_host_entropy": os.urandom(32),
|
||||
"ae_host_commitment": os.urandom(32),
|
||||
}
|
||||
)
|
||||
|
||||
# Change output details
|
||||
# This is optional, in that if we send it Jade validates the change output script
|
||||
# and the user need not confirm that ouptut. If not passed the change output must
|
||||
# be confirmed by the user on the hwwallet screen, like any other spend output.
|
||||
change: List[Optional[Dict[str, Any]]] = [None] * len(tx.outputs)
|
||||
|
||||
# If signing multisig inputs, get registered multisigs details in case we
|
||||
# see any multisig outputs which may be change which we can auto-validate.
|
||||
# ie. filter speculative 'signing multisigs' to ones actually registered on the hw
|
||||
candidate_multisigs = {}
|
||||
|
||||
if signing_multisigs:
|
||||
# register multisig if xpubs are known
|
||||
if tx.xpub and len(signing_multisigs) == 1:
|
||||
msigname = list(signing_multisigs.keys())[0]
|
||||
signers = []
|
||||
origins = []
|
||||
for xpub in tx.xpub:
|
||||
hd = bip32.HDKey.parse(xpub)
|
||||
origin = tx.xpub[xpub]
|
||||
origins.append((origin.fingerprint, origin.path))
|
||||
|
||||
signers.append(
|
||||
{
|
||||
"fingerprint": origin.fingerprint,
|
||||
"derivation": origin.path,
|
||||
"xpub": str(hd),
|
||||
"path": [],
|
||||
}
|
||||
)
|
||||
|
||||
# sort origins and signers together
|
||||
origins, signers = [
|
||||
list(a) for a in zip(*sorted(zip(origins, signers)))
|
||||
]
|
||||
|
||||
# Get a deterministic name for this multisig wallet
|
||||
script_variant = signing_multisigs[msigname][0]
|
||||
thresh = signing_multisigs[msigname][1]
|
||||
num_signers = signing_multisigs[msigname][2]
|
||||
multisig_name = self._get_multisig_name(
|
||||
script_variant, thresh, origins
|
||||
)
|
||||
# stupid sanity check of the fingerprints and origins
|
||||
if multisig_name == msigname:
|
||||
# Need to ensure this multisig wallet is registered first
|
||||
# (Note: 're-registering' is a no-op)
|
||||
self.jade.register_multisig(
|
||||
self._network(),
|
||||
multisig_name,
|
||||
script_variant,
|
||||
True, # always use sorted
|
||||
thresh,
|
||||
signers,
|
||||
)
|
||||
#
|
||||
registered_multisigs = self.jade.get_registered_multisigs()
|
||||
signing_multisigs = {
|
||||
k: v
|
||||
for k, v in signing_multisigs.items()
|
||||
if k in registered_multisigs
|
||||
and registered_multisigs[k]["variant"] == v[0]
|
||||
and registered_multisigs[k]["threshold"] == v[1]
|
||||
and registered_multisigs[k]["num_signers"] == len(v[2])
|
||||
}
|
||||
|
||||
# Look at every output...
|
||||
for n_vout, (txout, psbtout) in py_enumerate(zip(c_txn.vout, tx.outputs)):
|
||||
num_signers = len(psbtout.hd_keypaths)
|
||||
|
||||
if num_signers == 1 and signing_singlesigs:
|
||||
# Single-sig output - since we signed singlesig inputs this could be our change
|
||||
for pubkey, origin in psbtout.hd_keypaths.items():
|
||||
# Considers 'our' outputs as potential change as far as Jade is concerned
|
||||
# ie. can be verified and auto-confirmed.
|
||||
# Is this ok, or should check path also, assuming bip44-like ?
|
||||
if origin.fingerprint == master_fp and len(origin.path) > 0:
|
||||
change_addr_type = None
|
||||
if txout.is_p2pkh():
|
||||
change_addr_type = AddressType.LEGACY
|
||||
elif txout.is_witness()[0] and not txout.is_p2wsh():
|
||||
change_addr_type = AddressType.WIT # ie. p2wpkh
|
||||
elif (
|
||||
txout.is_p2sh() and is_witness(psbtout.redeem_script)[0]
|
||||
):
|
||||
change_addr_type = AddressType.SH_WIT
|
||||
else:
|
||||
continue
|
||||
|
||||
script_variant = self._convertAddrType(
|
||||
change_addr_type, multisig=False
|
||||
)
|
||||
change[n_vout] = {
|
||||
"path": origin.path,
|
||||
"variant": script_variant,
|
||||
}
|
||||
|
||||
elif num_signers > 1 and signing_multisigs:
|
||||
# Multisig output - since we signed multisig inputs this could be our change
|
||||
candidate_multisigs = {
|
||||
k: v
|
||||
for k, v in signing_multisigs.items()
|
||||
if len(v[2]) == num_signers
|
||||
}
|
||||
if not candidate_multisigs:
|
||||
continue
|
||||
|
||||
for pubkey, origin in psbtout.hd_keypaths.items():
|
||||
if origin.fingerprint == master_fp and len(origin.path) > 0:
|
||||
change_addr_type = None
|
||||
if (
|
||||
txout.is_p2sh()
|
||||
and not is_witness(psbtout.redeem_script)[0]
|
||||
):
|
||||
change_addr_type = AddressType.LEGACY
|
||||
scriptcode = psbtout.redeem_script
|
||||
elif txout.is_p2wsh() and not txout.is_p2sh():
|
||||
change_addr_type = AddressType.WIT
|
||||
scriptcode = psbtout.witness_script
|
||||
elif (
|
||||
txout.is_p2sh() and is_witness(psbtout.redeem_script)[0]
|
||||
):
|
||||
change_addr_type = AddressType.SH_WIT
|
||||
scriptcode = psbtout.witness_script
|
||||
else:
|
||||
continue
|
||||
|
||||
parsed = parse_multisig(scriptcode)
|
||||
if parsed:
|
||||
script_variant = self._convertAddrType(
|
||||
change_addr_type, multisig=True
|
||||
)
|
||||
threshold = parsed[0]
|
||||
|
||||
pubkeys = parsed[1]
|
||||
hd_keypath_origins = [
|
||||
psbtout.hd_keypaths[pubkey] for pubkey in pubkeys
|
||||
]
|
||||
|
||||
signers, paths = _parse_signers(hd_keypath_origins)
|
||||
|
||||
multisig_name = self._get_multisig_name(
|
||||
script_variant, threshold, signers
|
||||
)
|
||||
|
||||
matched_multisig = candidate_multisigs.get(
|
||||
multisig_name
|
||||
) == (script_variant, threshold, signers)
|
||||
if matched_multisig:
|
||||
change[n_vout] = {
|
||||
"paths": paths,
|
||||
"multisig_name": multisig_name,
|
||||
}
|
||||
|
||||
# The txn itself
|
||||
txn_bytes = c_txn.serialize_without_witness()
|
||||
|
||||
# Request Jade generate the signatures for our inputs.
|
||||
# Change details are passed to be validated on the hw (user does not confirm)
|
||||
signatures = self.jade.sign_tx(
|
||||
self._network(), txn_bytes, jade_inputs, change, True
|
||||
)
|
||||
|
||||
# Change output details
|
||||
# This is optional, in that if we send it Jade validates the change output script
|
||||
# and the user need not confirm that ouptut. If not passed the change output must
|
||||
# be confirmed by the user on the hwwallet screen, like any other spend output.
|
||||
change = [None] * len(tx.outputs)
|
||||
for n_vout, (txout, psbtout) in py_enumerate(zip(c_txn.vout, tx.outputs)):
|
||||
for pubkey, origin in psbtout.hd_keypaths.items():
|
||||
# Considers 'our' outputs as change as far as Jade is concerned
|
||||
# ie. can be auto-confirmed.
|
||||
# Is this ok, or should check path also, assuming bip44-like ?
|
||||
if origin.fingerprint == master_fp and len(origin.path) > 0:
|
||||
addr_type = None
|
||||
if txout.is_p2pkh():
|
||||
addr_type = AddressType.LEGACY
|
||||
elif txout.is_witness()[0] and not txout.is_p2wsh():
|
||||
addr_type = AddressType.WIT
|
||||
elif txout.is_p2sh():
|
||||
addr_type = AddressType.SH_WIT # is it really though ?
|
||||
|
||||
if addr_type:
|
||||
addr_type = self._convertAddrType(addr_type)
|
||||
change[n_vout] = {"path": origin.path, "variant": addr_type}
|
||||
|
||||
# The txn itself
|
||||
txn_bytes = c_txn.serialize_without_witness()
|
||||
|
||||
# Request Jade generate the signatures for our inputs.
|
||||
# Change details are passed to be validated on the hw (user does not confirm)
|
||||
signatures = self.jade.sign_tx(self._network(), txn_bytes, jade_inputs, change)
|
||||
|
||||
# Push sigs into PSBT structure as appropriate
|
||||
for psbtin, pubkey, sig in zip(tx.inputs, signing_pubkeys, signatures):
|
||||
if pubkey and sig:
|
||||
psbtin.partial_sigs[pubkey] = sig
|
||||
# Push sigs into PSBT structure as appropriate
|
||||
for psbtin, signer_pubkey, sigdata in zip(
|
||||
tx.inputs, signing_pubkeys, signatures
|
||||
):
|
||||
signer_commitment, sig = sigdata
|
||||
if signer_pubkey and sig:
|
||||
psbtin.partial_sigs[signer_pubkey] = sig
|
||||
|
||||
# Return the updated psbt
|
||||
return tx
|
||||
|
|
@ -273,17 +507,75 @@ class JadeClient(HardwareWalletClient):
|
|||
return address
|
||||
|
||||
def display_multisig_address(
|
||||
self, threshold: int, pubkeys: List[PubkeyProvider], addr_type: AddressType
|
||||
self,
|
||||
addr_type: AddressType,
|
||||
multisig: MultisigDescriptor,
|
||||
) -> str:
|
||||
"""
|
||||
The Blockstream Jade does not support multisig addresses.
|
||||
signer_origins = []
|
||||
signers = []
|
||||
paths = []
|
||||
for pubkey in multisig.pubkeys:
|
||||
if pubkey.extkey is None:
|
||||
raise BadArgumentError(
|
||||
"Blockstream Jade can only generate addresses for multisigs with full extended keys"
|
||||
)
|
||||
if pubkey.origin is None:
|
||||
raise BadArgumentError(
|
||||
"Blockstream Jade can only generate addresses for multisigs with key origin information"
|
||||
)
|
||||
if pubkey.deriv_path is None:
|
||||
raise BadArgumentError(
|
||||
"Blockstream Jade can only generate addresses for multisigs with key origin derivation path information"
|
||||
)
|
||||
|
||||
:raises UnavailableActionError: Always, this function is unavailable
|
||||
"""
|
||||
raise UnavailableActionError(
|
||||
"The Blockstream Jade does not support generic multisig P2SH address display"
|
||||
# Tuple to derive deterministic name for the registrtion
|
||||
signer_origins.append((pubkey.origin.fingerprint, pubkey.origin.path))
|
||||
|
||||
# We won't include the additional path in the multisig registration
|
||||
signers.append(
|
||||
{
|
||||
"fingerprint": pubkey.origin.fingerprint,
|
||||
"derivation": pubkey.origin.path,
|
||||
"xpub": pubkey.pubkey,
|
||||
"path": [],
|
||||
}
|
||||
)
|
||||
|
||||
# Instead hold it as the address path
|
||||
path = (
|
||||
pubkey.deriv_path[1:]
|
||||
if pubkey.deriv_path[0] == "/"
|
||||
else pubkey.deriv_path
|
||||
)
|
||||
paths.append(parse_path(path))
|
||||
|
||||
# sort origins, signers and paths according to origins (like in _get_multisig_name)
|
||||
signer_origins, signers, paths = [
|
||||
list(a) for a in zip(*sorted(zip(signer_origins, signers, paths)))
|
||||
]
|
||||
|
||||
# Get a deterministic name for this multisig wallet
|
||||
script_variant = self._convertAddrType(addr_type, multisig=True)
|
||||
multisig_name = self._get_multisig_name(
|
||||
script_variant, multisig.thresh, signer_origins
|
||||
)
|
||||
|
||||
# Need to ensure this multisig wallet is registered first
|
||||
# (Note: 're-registering' is a no-op)
|
||||
self.jade.register_multisig(
|
||||
self._network(),
|
||||
multisig_name,
|
||||
script_variant,
|
||||
True, # always use sorted
|
||||
multisig.thresh,
|
||||
signers,
|
||||
)
|
||||
address = self.jade.get_receive_address(
|
||||
self._network(), paths, multisig_name=multisig_name
|
||||
)
|
||||
|
||||
return str(address)
|
||||
|
||||
# Setup a new device
|
||||
def setup_device(self, label="", passphrase=""):
|
||||
"""
|
||||
|
|
@ -531,28 +823,45 @@ class JadeClient(HardwareWalletClient):
|
|||
return None
|
||||
|
||||
|
||||
def enumerate(password=""):
|
||||
def enumerate(password: str = "") -> List[Dict[str, Any]]:
|
||||
results = []
|
||||
|
||||
def _get_device_entry(device_model: str, device_path: str) -> Dict[str, Any]:
|
||||
d_data: Dict[str, Any] = {}
|
||||
d_data["type"] = "jade"
|
||||
d_data["model"] = device_model
|
||||
d_data["path"] = device_path
|
||||
d_data["needs_pin_sent"] = False
|
||||
d_data["needs_passphrase_sent"] = False
|
||||
|
||||
client = None
|
||||
with handle_errors(common_err_msgs["enumerate"], d_data):
|
||||
client = JadeClient(device_path, password, timeout=1)
|
||||
d_data["fingerprint"] = client.get_master_fingerprint().hex()
|
||||
|
||||
if client:
|
||||
client.close()
|
||||
|
||||
return d_data
|
||||
|
||||
# Jade is not really an HID device, it shows as a serial/com port device.
|
||||
# Scan com ports looking for the relevant vid and pid, and use 'path' to
|
||||
# hold the path to the serial port device, eg. /dev/ttyUSB0
|
||||
for devinfo in list_ports.comports():
|
||||
if devinfo.vid == JADE_VENDOR_ID and devinfo.pid == JADE_DEVICE_ID:
|
||||
d_data = {}
|
||||
d_data["type"] = "jade"
|
||||
d_data["path"] = devinfo.device
|
||||
d_data["needs_pin_sent"] = False
|
||||
d_data["needs_passphrase_sent"] = False
|
||||
if (devinfo.vid, devinfo.pid) in JADE_DEVICE_IDS:
|
||||
results.append(_get_device_entry("jade", devinfo.device))
|
||||
|
||||
client = None
|
||||
with handle_errors(common_err_msgs["enumerate"], d_data):
|
||||
client = JadeClient(devinfo.device, password)
|
||||
d_data["fingerprint"] = client.get_master_fingerprint().hex()
|
||||
# If we can connect to the simulator, add it too
|
||||
try:
|
||||
with JadeAPI.create_serial(SIMULATOR_PATH, timeout=1) as jade:
|
||||
verinfo = jade.get_version_info()
|
||||
|
||||
if client:
|
||||
client.close()
|
||||
if verinfo is not None:
|
||||
results.append(_get_device_entry("jade_simulator", SIMULATOR_PATH))
|
||||
|
||||
results.append(d_data)
|
||||
except Exception as e:
|
||||
# If we get any sort of error do not add the simulator
|
||||
logging.debug(f"Failed to connect to Jade simulator at {SIMULATOR_PATH}")
|
||||
logging.debug(e)
|
||||
|
||||
return results
|
||||
|
|
|
|||
|
|
@ -1,9 +0,0 @@
|
|||
# Python Jade Library
|
||||
|
||||
This is a slightly stripped down version of the official [Jade](https://github.com/Blockstream/Jade) python library.
|
||||
|
||||
This stripped down version was made at commit [dfdfc7e1d8b91227f4ea7555457640506b8c8aed](https://github.com/Blockstream/Jade/commit/dfdfc7e1d8b91227f4ea7555457640506b8c8aed).
|
||||
|
||||
## Changes
|
||||
|
||||
- Removed BLE module, reducing transitive dependencies
|
||||
|
|
@ -1,11 +1,14 @@
|
|||
import cbor
|
||||
import hashlib
|
||||
import json
|
||||
import time
|
||||
import logging
|
||||
import collections
|
||||
import collections.abc
|
||||
import traceback
|
||||
import requests
|
||||
import random
|
||||
import sys
|
||||
|
||||
|
||||
# JadeError
|
||||
from .jade_error import JadeError
|
||||
|
|
@ -14,11 +17,8 @@ from .jade_error import JadeError
|
|||
from .jade_serial import JadeSerialImpl
|
||||
from .jade_tcp import JadeTCPImpl
|
||||
|
||||
# Not used in HWI
|
||||
# Removed to reduce transitive dependencies
|
||||
# from .jade_ble import JadeBleImpl
|
||||
|
||||
|
||||
# Default serial connection
|
||||
DEFAULT_SERIAL_DEVICE = "/dev/ttyUSB0"
|
||||
DEFAULT_BAUD_RATE = 115200
|
||||
|
|
@ -34,6 +34,48 @@ logger = logging.getLogger("jade")
|
|||
device_logger = logging.getLogger("jade-device")
|
||||
|
||||
|
||||
# Helper to map bytes-like types into hex-strings
|
||||
# to make for prettier message-logging
|
||||
def _hexlify(data):
|
||||
if data is None:
|
||||
return None
|
||||
elif isinstance(data, bytes) or isinstance(data, bytearray):
|
||||
return data.hex()
|
||||
elif isinstance(data, list):
|
||||
return [_hexlify(item) for item in data]
|
||||
elif isinstance(data, dict):
|
||||
return {k: _hexlify(v) for k, v in data.items()}
|
||||
else:
|
||||
return data
|
||||
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
def _http_request(params):
|
||||
logger.debug("_http_request: {}".format(params))
|
||||
|
||||
# Use the first non-onion url
|
||||
url = [url for url in params["urls"] if not url.endswith(".onion")][0]
|
||||
if params["method"] == "GET":
|
||||
assert "data" not in params, "Cannot pass body to requests.get"
|
||||
f = requests.get(url)
|
||||
elif params["method"] == "POST":
|
||||
data = json.dumps(params["data"])
|
||||
f = requests.post(url, data)
|
||||
|
||||
logger.debug("http_request received reply: {}".format(f.text))
|
||||
|
||||
if f.status_code != 200:
|
||||
logger.error("http error {} : {}".format(f.status_code, f.text))
|
||||
raise ValueError(f.status_code)
|
||||
|
||||
assert params["accept"] == "json"
|
||||
f = f.json()
|
||||
|
||||
return {"body": f}
|
||||
|
||||
|
||||
#
|
||||
# High-Level Jade Client API
|
||||
# Builds on a JadeInterface to provide a meaningful API
|
||||
|
|
@ -88,33 +130,6 @@ class JadeAPI:
|
|||
def drain(self):
|
||||
self.jade.drain()
|
||||
|
||||
# Simple http request function which can be used when a Jade response requires
|
||||
# an external http call.
|
||||
# The default implementation used in _jadeRpc() below.
|
||||
@staticmethod
|
||||
def _http_request(params):
|
||||
logger.debug("_http_request: {}".format(params))
|
||||
|
||||
# Use the first non-onion url
|
||||
url = [url for url in params["urls"] if not url.endswith(".onion")][0]
|
||||
if params["method"] == "GET":
|
||||
assert "data" not in params, "Cannot pass body to requests.get"
|
||||
f = requests.get(url)
|
||||
elif params["method"] == "POST":
|
||||
data = json.dumps(params["data"])
|
||||
f = requests.post(url, data)
|
||||
|
||||
logger.debug("http_request received reply: {}".format(f.text))
|
||||
|
||||
if f.status_code != 200:
|
||||
logger.error("http error {} : {}".format(f.status_code, f.text))
|
||||
raise ValueError(f.status_code)
|
||||
|
||||
assert params["accept"] == "json"
|
||||
f = f.json()
|
||||
|
||||
return {"body": f}
|
||||
|
||||
# Raise any returned error as an exception
|
||||
@staticmethod
|
||||
def _get_result_or_raise_error(reply):
|
||||
|
|
@ -143,9 +158,14 @@ class JadeAPI:
|
|||
# code below acts as a dumb proxy and simply makes the http request and
|
||||
# forwards the response back to the Jade.
|
||||
# Note: the function called to make the http-request can be passed in,
|
||||
# or defaults to the simple _http_request() function above.
|
||||
if isinstance(result, collections.Mapping) and "http_request" in result:
|
||||
make_http_request = http_request_fn or self._http_request
|
||||
# or it can default to the simple _http_request() function above, if available.
|
||||
if isinstance(result, collections.abc.Mapping) and "http_request" in result:
|
||||
this_module = sys.modules[__name__]
|
||||
make_http_request = http_request_fn or getattr(
|
||||
this_module, "_http_request", None
|
||||
)
|
||||
assert make_http_request, "Default _http_request() function not available"
|
||||
|
||||
http_request = result["http_request"]
|
||||
http_response = make_http_request(http_request["params"])
|
||||
return self._jadeRpc(
|
||||
|
|
@ -169,18 +189,21 @@ class JadeAPI:
|
|||
# OTA new firmware
|
||||
def ota_update(self, fwcmp, fwlen, chunksize, cb):
|
||||
|
||||
compressed_size = len(fwcmp)
|
||||
cmphasher = hashlib.sha256()
|
||||
cmphasher.update(fwcmp)
|
||||
cmphash = cmphasher.digest()
|
||||
cmplen = len(fwcmp)
|
||||
|
||||
# Initiate OTA
|
||||
params = {"fwsize": fwlen, "cmpsize": compressed_size}
|
||||
params = {"fwsize": fwlen, "cmpsize": cmplen, "cmphash": cmphash}
|
||||
|
||||
result = self._jadeRpc("ota", params)
|
||||
assert result is True
|
||||
|
||||
# Write binary chunks
|
||||
written = 0
|
||||
while written < compressed_size:
|
||||
remaining = compressed_size - written
|
||||
while written < cmplen:
|
||||
remaining = cmplen - written
|
||||
length = min(remaining, chunksize)
|
||||
chunk = bytes(fwcmp[written : written + length])
|
||||
result = self._jadeRpc("ota_data", chunk)
|
||||
|
|
@ -188,23 +211,27 @@ class JadeAPI:
|
|||
written += length
|
||||
|
||||
if cb:
|
||||
cb(written, compressed_size)
|
||||
cb(written, cmplen)
|
||||
|
||||
# All binary data uploaded
|
||||
return self._jadeRpc("ota_complete")
|
||||
|
||||
# Run (debug) healthcheck on the hw
|
||||
def run_remote_selfcheck(self):
|
||||
return self._jadeRpc("debug_selfcheck")
|
||||
return self._jadeRpc("debug_selfcheck", long_timeout=True)
|
||||
|
||||
# Set the (debug) mnemonic
|
||||
def set_mnemonic(self, mnemonic):
|
||||
params = {"mnemonic": mnemonic}
|
||||
def set_mnemonic(self, mnemonic, passphrase=None, temporary_wallet=False):
|
||||
params = {
|
||||
"mnemonic": mnemonic,
|
||||
"passphrase": passphrase,
|
||||
"temporary_wallet": temporary_wallet,
|
||||
}
|
||||
return self._jadeRpc("debug_set_mnemonic", params)
|
||||
|
||||
# Set the (debug) seed
|
||||
def set_seed(self, seed):
|
||||
params = {"seed": seed}
|
||||
def set_seed(self, seed, temporary_wallet=False):
|
||||
params = {"seed": seed, "temporary_wallet": temporary_wallet}
|
||||
return self._jadeRpc("debug_set_mnemonic", params)
|
||||
|
||||
# Override the pinserver details on the hww
|
||||
|
|
@ -240,11 +267,35 @@ class JadeAPI:
|
|||
params = {"network": network, "path": path}
|
||||
return self._jadeRpc("get_xpub", params)
|
||||
|
||||
# Get registered multisig wallets
|
||||
def get_registered_multisigs(self):
|
||||
return self._jadeRpc("get_registered_multisigs")
|
||||
|
||||
# Register a multisig wallet
|
||||
def register_multisig(
|
||||
self, network, multisig_name, variant, sorted_keys, threshold, signers
|
||||
):
|
||||
params = {
|
||||
"network": network,
|
||||
"multisig_name": multisig_name,
|
||||
"descriptor": {
|
||||
"variant": variant,
|
||||
"sorted": sorted_keys,
|
||||
"threshold": threshold,
|
||||
"signers": signers,
|
||||
},
|
||||
}
|
||||
return self._jadeRpc("register_multisig", params)
|
||||
|
||||
# Get receive-address for parameters
|
||||
def get_receive_address(
|
||||
self, *args, recovery_xpub=None, csv_blocks=0, variant=None
|
||||
self, *args, recovery_xpub=None, csv_blocks=0, variant=None, multisig_name=None
|
||||
):
|
||||
if variant is not None:
|
||||
if multisig_name is not None:
|
||||
assert len(args) == 2
|
||||
keys = ["network", "paths", "multisig_name"]
|
||||
args += (multisig_name,)
|
||||
elif variant is not None:
|
||||
assert len(args) == 2
|
||||
keys = ["network", "path", "variant"]
|
||||
args += (variant,)
|
||||
|
|
@ -300,9 +351,13 @@ class JadeAPI:
|
|||
|
||||
# Get the shared secret to unblind a tx, given the receiving script on
|
||||
# our side and the pubkey of the sender (sometimes called "nonce" in
|
||||
# Liquid)
|
||||
def get_shared_nonce(self, script, their_pubkey):
|
||||
params = {"script": script, "their_pubkey": their_pubkey}
|
||||
# Liquid). Optionally fetch our blinding pubkey also.
|
||||
def get_shared_nonce(self, script, their_pubkey, include_pubkey=False):
|
||||
params = {
|
||||
"script": script,
|
||||
"their_pubkey": their_pubkey,
|
||||
"include_pubkey": include_pubkey,
|
||||
}
|
||||
return self._jadeRpc("get_shared_nonce", params)
|
||||
|
||||
# Get a "trusted" blinding factor to blind an output. Normally the blinding
|
||||
|
|
@ -424,6 +479,7 @@ class JadeAPI:
|
|||
"use_ae_signatures": use_ae_signatures,
|
||||
"change": change,
|
||||
}
|
||||
|
||||
reply = self._jadeRpc("sign_liquid_tx", params, str(base_id))
|
||||
assert reply
|
||||
|
||||
|
|
@ -442,6 +498,7 @@ class JadeAPI:
|
|||
"use_ae_signatures": use_ae_signatures,
|
||||
"change": change,
|
||||
}
|
||||
|
||||
reply = self._jadeRpc("sign_tx", params, str(base_id))
|
||||
assert reply
|
||||
|
||||
|
|
@ -497,15 +554,15 @@ class JadeInterface:
|
|||
)
|
||||
return JadeInterface(impl)
|
||||
|
||||
@staticmethod
|
||||
def create_ble(device_name=None, serial_number=None, scan_timeout=None, loop=None):
|
||||
impl = JadeBleImpl(
|
||||
device_name or DEFAULT_BLE_DEVICE_NAME,
|
||||
serial_number or DEFAULT_BLE_SERIAL_NUMBER,
|
||||
scan_timeout or DEFAULT_BLE_SCAN_TIMEOUT,
|
||||
loop=loop,
|
||||
)
|
||||
return JadeInterface(impl)
|
||||
# @staticmethod
|
||||
# def create_ble(device_name=None, serial_number=None, scan_timeout=None, loop=None):
|
||||
# impl = JadeBleImpl(
|
||||
# device_name or DEFAULT_BLE_DEVICE_NAME,
|
||||
# serial_number or DEFAULT_BLE_SERIAL_NUMBER,
|
||||
# scan_timeout or DEFAULT_BLE_SCAN_TIMEOUT,
|
||||
# loop=loop,
|
||||
# )
|
||||
# return JadeInterface(impl)
|
||||
|
||||
def connect(self):
|
||||
self.impl.connect()
|
||||
|
|
@ -557,7 +614,9 @@ class JadeInterface:
|
|||
)
|
||||
logger.info(msg)
|
||||
else:
|
||||
logger.info("Sending: {} as cbor of size {}".format(request, len_dump))
|
||||
logger.info(
|
||||
"Sending: {} as cbor of size {}".format(_hexlify(request), len_dump)
|
||||
)
|
||||
return dump
|
||||
|
||||
def write(self, bytes_):
|
||||
|
|
@ -586,7 +645,7 @@ class JadeInterface:
|
|||
|
||||
# A message response (to a prior request)
|
||||
if "id" in message:
|
||||
logger.info("Received msg: {}".format(message))
|
||||
logger.info("Received msg: {}".format(_hexlify(message)))
|
||||
return message
|
||||
|
||||
# A log message - handle as normal
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ class Jade(HWIDevice):
|
|||
icon = "jade_icon.svg"
|
||||
|
||||
supports_hwi_toggle_passphrase = False
|
||||
supports_hwi_multisig_display_address = False
|
||||
supports_hwi_multisig_display_address = True
|
||||
liquid_support = True
|
||||
|
||||
@classmethod
|
||||
|
|
|
|||
|
|
@ -1 +1,2 @@
|
|||
KERNEL=="ttyUSB*", SUBSYSTEMS=="usb", ATTRS{idVendor}=="10c4", ATTRS{idProduct}=="ea60", MODE="0660", GROUP="plugdev", TAG+="uaccess", TAG+="udev-acl", SYMLINK+="jade%n"
|
||||
KERNEL=="ttyACM*", SUBSYSTEMS=="usb", ATTRS{idVendor}=="1a86", ATTRS{idProduct}=="55d4", MODE="0660", GROUP="plugdev", TAG+="uaccess", TAG+="udev-acl", SYMLINK+="jade%n"
|
||||
Loading…
Add table
Add a link
Reference in a new issue