GlcoinCore/specter-desktop
1
0
Fork 0
mirror of https://github.com/cryptoadvance/specter-desktop.git synced 2026-08-13 12:33:29 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Fix Jade signing issues with Swan Vault (#2421)

Browse source 
* update jade api to version 2.0.2 and update jade hwi client to the latest hwi code (version

* change chain default for initialising the jade client back to MAIN

* add "unlock" to jade's enumerate method and its hwi client

* use is_startup property to prevent jade unlocking on startup + change to skip_hwi_initialisation + some simplifications

* pass chain param on every request for enumerate call as well + some changes to be on the safe side (avoid name collusion with built-in fetch + random id)

* always use timeout when calling enumerate from hwi.jinja

* add ui to register multisigs for multisig wallets using a jade

* updated requirements.txt

* rename myFetch to requestToHwiBridge

* address jamie's comments: change logic to skip_unlocking + move early return up in jade client
This commit is contained in:
Manolis Mandrapilias 2024-03-27 20:19:25 +01:00 • committed by GitHub
parent a0523732aa
commit 7970c3d8ab
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 2500 additions and 706 deletions
Download patch file Download diff file Expand all files Collapse all files

2
requirements.in
View file

@ -18,7 +18,7 @@ psutil==5.9.0
pyopenssl==23.0.0
flask_wtf==0.15.1
pgpy==0.6.0
cbor==1.0.0
cbor2==5.4.6
mnemonic==0.20
cryptography==39.0.1
Flask-APScheduler==1.12.4

47
requirements.txt
View file

@ -2,7 +2,7 @@
# This file is autogenerated by pip-compile with Python 3.10
# by the following command:
#
# pip-compile --generate-hashes --resolver=backtracking requirements.in
# pip-compile --generate-hashes requirements.in
#
aniso8601==9.0.1 \
--hash=sha256:1d2b7ef82963909e93c4f24ce48d4de9e66009a21bf1c1e1c85bdd0812fe412f \
@ -26,9 +26,45 @@ bitbox02==6.1.1 \
# via hwi
cbor==1.0.0 \
--hash=sha256:13225a262ddf5615cbd9fd55a76a0d53069d18b07d2e9f19c39e6acb8609bbb6
# via
# -r requirements.in
# hwi
# via hwi
cbor2==5.4.6 \
--hash=sha256:0b956f19e93ba3180c336282cd1b6665631f2d3a196a9c19b29a833bf979e7a4 \
--hash=sha256:0bd12c54a48949d11f5ffc2fa27f5df1b4754111f5207453e5fae3512ebb3cab \
--hash=sha256:0d2b926b024d3a1549b819bc82fdc387062bbd977b0299dd5fa5e0ea3267b98b \
--hash=sha256:1618d16e310f7ffed141762b0ff5d8bb6b53ad449406115cc465bf04213cefcf \
--hash=sha256:181ac494091d1f9c5bb373cd85514ce1eb967a8cf3ec298e8dfa8878aa823956 \
--hash=sha256:1835536e76ea16e88c934aac5e369ba9f93d495b01e5fa2d93f0b4986b89146d \
--hash=sha256:1c12c0ab78f5bc290b08a79152a8621822415836a86f8f4b50dadba371736fda \
--hash=sha256:24144822f8d2b0156f4cda9427f071f969c18683ffed39663dc86bc0a75ae4dd \
--hash=sha256:309fffbb7f561d67f02095d4b9657b73c9220558701c997e9bfcfbca2696e927 \
--hash=sha256:3316f09a77af85e7772ecfdd693b0f450678a60b1aee641bac319289757e3fa0 \
--hash=sha256:3545b16f9f0d5f34d4c99052829c3726020a07be34c99c250d0df87418f02954 \
--hash=sha256:39452c799453f5bf33281ffc0752c620b8bfa0b7c13070b87d370257a1311976 \
--hash=sha256:3950be57a1698086cf26d8710b4e5a637b65133c5b1f9eec23967d4089d8cfed \
--hash=sha256:456cdff668a50a52fdb8aa6d0742511e43ed46d6a5b463dba80a5a720fa0d320 \
--hash=sha256:4b9f3924da0e460a93b3674c7e71020dd6c9e9f17400a34e52a88c0af2dcd2aa \
--hash=sha256:4bbbdb2e3ef274865dc3f279aae109b5d94f4654aea3c72c479fb37e4a1e7ed7 \
--hash=sha256:4ce1a2c272ba8523a55ea2f1d66e3464e89fa0e37c9a3d786a919fe64e68dbd7 \
--hash=sha256:56dfa030cd3d67e5b6701d3067923f2f61536a8ffb1b45be14775d1e866b59ae \
--hash=sha256:6709d97695205cd08255363b54afa035306d5302b7b5e38308c8ff5a47e60f2a \
--hash=sha256:6e1b5aee920b6a2f737aa12e2b54de3826b09f885a7ce402db84216343368140 \
--hash=sha256:6f9c702bee2954fffdfa3de95a5af1a6b1c5f155e39490353d5654d83bb05bb9 \
--hash=sha256:78304df140b9e13b93bcbb2aecee64c9aaa9f1cadbd45f043b5e7b93cc2f21a2 \
--hash=sha256:79e048e623846d60d735bb350263e8fdd36cb6195d7f1a2b57eacd573d9c0b33 \
--hash=sha256:7bbd3470eb685325398023e335be896b74f61b014896604ed45049a7b7b6d8ac \
--hash=sha256:80ac8ba450c7a41c5afe5f7e503d3092442ed75393e1de162b0bf0d97edf7c7f \
--hash=sha256:9394ca49ecdf0957924e45d09a4026482d184a465a047f60c4044eb464c43de9 \
--hash=sha256:94f844d0e232aca061a86dd6ff191e47ba0389ddd34acb784ad9a41594dc99a4 \
--hash=sha256:96087fa5336ebfc94465c0768cd5de0fcf9af3840d2cf0ce32f5767855f1a293 \
--hash=sha256:b893500db0fe033e570c3adc956af6eefc57e280026bd2d86fd53da9f1e594d7 \
--hash=sha256:c285a2cb2c04004bfead93df89d92a0cef1874ad337d0cb5ea53c2c31e97bfdb \
--hash=sha256:d2984a488f350aee1d54fa9cb8c6a3c1f1f5b268abbc91161e47185de4d829f3 \
--hash=sha256:d54bd840b4fe34f097b8665fc0692c7dd175349e53976be6c5de4433b970daa4 \
--hash=sha256:db9eb582fce972f0fa429d8159b7891ff8deccb7affc4995090afc61ce0d328a \
--hash=sha256:e5094562dfe3e5583202b93ef7ca5082c2ba5571accb2c4412d27b7d0ba8a563 \
--hash=sha256:e73ca40dd3c7210ff776acff9869ddc9ff67bae7c425b58e5715dcf55275163f \
--hash=sha256:ff95b33e5482313a74648ca3620c9328e9f30ecfa034df040b828e476597d352
# via -r requirements.in
certifi==2022.12.7 \
--hash=sha256:35824b4c3a97115964b408844d64aa14db1cc518f6562e8d7261699d1350a9e3 \
--hash=sha256:4ad3232f5e926d6718ec31cfc1fcadfde020920e278684144551c91769c7bc18
@ -796,5 +832,6 @@ wtforms==3.0.1 \
# via flask-wtf
# WARNING: The following packages were not pinned, but pip requires them to be
# pinned when the requirements file includes hashes. Consider using the --allow-unsafe flag.
# pinned when the requirements file includes hashes and the requirement is not
# satisfied by a package already installed. Consider using the --allow-unsafe flag.
# setuptools

10
src/cryptoadvance/specter/cli/cli_server.py
View file

@ -58,9 +58,9 @@ def cli():
help="Start the hwi-bridge to use your HWWs with a remote specter.",
)
@click.option(
"--enforcehwiinitialisation",
"--skiphwiinitialisation",
is_flag=True,
help="calls enumerate() which is known to cause issues with certain usb-devices plugged in at startup.",
help="Skips to call HWI's enumerate() on start-up",
)
@click.option(
"--devstatus-threshold",
@ -88,7 +88,7 @@ def server(
filelog,
tor,
hwibridge,
enforcehwiinitialisation,
skiphwiinitialisation,
devstatus_threshold,
specter_data_folder,
config,
@ -154,8 +154,8 @@ def server(
kwargs = configure_ssl(kwargs, app.config, ssl)
app.app_context().push()
if enforcehwiinitialisation:
app.config["ENFORCE_HWI_INITIALISATION_AT_STARTUP"] = True
if skiphwiinitialisation:
app.config["SKIP_HWI_INITIALISATION_AT_STARTUP"] = True
init_app(app, hwibridge=hwibridge)
if filelog:

6
src/cryptoadvance/specter/config.py
View file

@ -82,9 +82,9 @@ class BaseConfig(object):
CERT = os.getenv("CERT", None)
KEY = os.getenv("KEY", None)
# It might be necessary to enforce the HWI initialisation
ENFORCE_HWI_INITIALISATION_AT_STARTUP = _get_bool_env_var(
"ENFORCE_HWI_INITIALISATION_AT_STARTUP", False
# It might be necessary / useful for testing to skip the HWI initialisation (i.e. calling enumerate on startup)
SKIP_HWI_INITIALISATION_AT_STARTUP = _get_bool_env_var(
"SKIP_HWI_INITIALISATION_AT_STARTUP", False
)
# This will be used to search for a bitcoin.conf in order to enable the

740
src/cryptoadvance/specter/devices/hwi/jade.py
View file

@ -9,33 +9,25 @@ from serial.tools import list_ports
from functools import wraps
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
from hwilib.descriptor import PubkeyProvider, MultisigDescriptor
from hwilib.descriptor import MultisigDescriptor
from hwilib.hwwclient import HardwareWalletClient
from hwilib.errors import (
ActionCanceledError,
BadArgumentError,
DeviceConnectionError,
DeviceFailureError,
DeviceNotReadyError,
UnavailableActionError,
common_err_msgs,
handle_errors,
)
from hwilib.common import (
AddressType,
Chain,
)
from hwilib.key import ExtendedKey, parse_path, KeyOriginInfo, is_hardened
from hwilib.common import AddressType, Chain, sha256
from hwilib.key import ExtendedKey, KeyOriginInfo, is_hardened, parse_path
from hwilib.psbt import PSBT
from hwilib.tx import CTransaction
from hwilib._script import (
is_p2sh,
is_p2wpkh,
is_p2wsh,
is_witness,
parse_multisig,
)
from hwilib._script import is_p2sh, is_p2wpkh, is_p2wsh, is_witness, parse_multisig
import logging
import semver
import os
import hashlib
@ -48,11 +40,17 @@ from embit import hashes
from embit.util import secp256k1
from embit.liquid.finalizer import finalize_psbt
from embit.liquid.transaction import write_commitment
from embit.descriptor import Descriptor
# The test emulator port
SIMULATOR_PATH = "tcp:127.0.0.1:2222"
SIMULATOR_PATH = "tcp:127.0.0.1:30121"
JADE_DEVICE_IDS = [(0x10C4, 0xEA60), (0x1A86, 0x55D4)]
JADE_DEVICE_IDS = [
(0x10C4, 0xEA60),
(0x1A86, 0x55D4),
(0x0403, 0x6001),
(0x1A86, 0x7523),
]
HAS_NETWORKING = hasattr(jade, "_http_request")
py_enumerate = (
@ -94,26 +92,19 @@ def jade_exception(f: Callable[..., Any]) -> Any:
# This class extends the HardwareWalletClient for Blockstream Jade specific things
class JadeClient(HardwareWalletClient):
MIN_SUPPORTED_FW_VERSION = semver.VersionInfo(0, 1, 32)
NETWORKS = {
Chain.MAIN: "mainnet",
Chain.TEST: "testnet",
Chain.SIGNET: "testnet", # same as far as Jade is concerned
Chain.REGTEST: "localtest",
}
liquid_network = None
def set_liquid_network(self, chain):
if chain == "liquidv1":
self.liquid_network = "liquid"
elif chain == "liquidtestnet":
self.liquid_network = "testnet-liquid"
else:
self.liquid_network = "localtest-liquid"
def _network(self):
if self.liquid_network:
return self.liquid_network
return JadeClient.NETWORKS.get(self.chain, "mainnet")
def _network(self) -> str:
if self.chain not in self.NETWORKS:
raise BadArgumentError(f"Unhandled network: {self.chain}")
return self.NETWORKS[self.chain]
ADDRTYPES = {
AddressType.LEGACY: "pkh(k)",
@ -126,12 +117,11 @@ class JadeClient(HardwareWalletClient):
AddressType.SH_WIT: "sh(wsh(multi(k)))",
}
@staticmethod
def _convertAddrType(addrType, multisig=False):
if multisig:
return JadeClient.MULTI_ADDRTYPES[addrType]
return JadeClient.ADDRTYPES[addrType]
@classmethod
def _convertAddrType(cls, addrType: AddressType, multisig: bool) -> str:
return cls.MULTI_ADDRTYPES[addrType] if multisig else cls.ADDRTYPES[addrType]
# Derive a deterministic name for a multisig registration record (ignoring bip67 key sorting)
@staticmethod
def _get_multisig_name(
type: str, threshold: int, signers: List[Tuple[bytes, Sequence[int]]]
@ -142,31 +132,58 @@ class JadeClient(HardwareWalletClient):
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()
hash_summary = sha256(summary.encode()).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)
def __init__(
self,
path: str,
password: Optional[str] = None,
expert: bool = False,
chain: Chain = Chain.MAIN,
skip_unlocking: bool = False,
timeout: Optional[int] = None,
) -> None:
super(JadeClient, self).__init__(path, password, expert, chain)
self.jade = JadeAPI.create_serial(path, timeout=timeout)
self.jade.connect()
# Push some host entropy into jade
self.jade.add_entropy(os.urandom(32))
verinfo = self.jade.get_version_info()
uninitialized = verinfo["JADE_STATE"] not in ["READY", "TEMP"]
# Do the PIN thing if required
# NOTE: uses standard 'requests' networking to connect to blind pinserver
try:
while not self.jade.auth_user(self._network()):
logging.debug("Incorrect PIN provided")
except:
try:
self.chain = Chain.TEST
while not self.jade.auth_user(self._network()):
logging.debug("Incorrect PIN provided")
except:
self.chain = Chain.REGTEST
while not self.jade.auth_user(self._network()):
logging.debug("Incorrect PIN provided")
# Check minimum supported firmware version (ignore candidate/build parts)
fw_version = semver.parse_version_info(verinfo["JADE_VERSION"])
if self.MIN_SUPPORTED_FW_VERSION > fw_version.finalize_version():
raise DeviceNotReadyError(
f"Jade fw version: {fw_version} - minimum required version: {self.MIN_SUPPORTED_FW_VERSION}. "
"Please update using a Blockstream Green companion app"
)
if path == SIMULATOR_PATH:
if uninitialized:
# Connected to simulator but it appears to have no wallet set
raise DeviceNotReadyError(
"Use JadeAPI.set_[seed|mnemonic] to set simulator wallet"
)
else:
if uninitialized:
if skip_unlocking:
# We don't want to prompt to unlock the device right now
return
if not HAS_NETWORKING:
# Wallet not initialised/unlocked nor do we have networking dependencies
# User must use 'Recovery Phrase Login' or 'QR Unlock' feature to access wallet
raise DeviceNotReadyError(
'Use "Recovery Phrase Login" or "QR PIN Unlock" feature on Jade hw to access wallet'
)
# Push some host entropy into jade
self.jade.add_entropy(os.urandom(32))
# Authenticate the user - this may require a PIN and pinserver interaction
# (if we have required networking dependencies)
authenticated = False
while not authenticated:
authenticated = self.jade.auth_user(self._network())
# Retrieves the public key at the specified BIP 32 derivation path
@jade_exception
@ -176,13 +193,6 @@ class JadeClient(HardwareWalletClient):
ext_key = ExtendedKey.deserialize(xpub)
return ext_key
@jade_exception
def get_master_blinding_key(self) -> str:
mbk = self.jade.get_master_blinding_key()
assert len(mbk) == 32
bkey = ec.PrivateKey(mbk)
return bkey.wif()
# Walk the PSBT looking for inputs we can sign. Push any signatures into the
# 'partial_sigs' map in the input, and return the updated PSBT.
@jade_exception
@ -209,12 +219,9 @@ class JadeClient(HardwareWalletClient):
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)
c_txn = tx.get_unsigned_tx()
master_fp = self.get_master_fingerprint()
signing_singlesigs = False
signing_multisigs = {}
@ -226,7 +233,7 @@ class JadeClient(HardwareWalletClient):
# Signing input details
jade_inputs = []
for n_vin, (txin, psbtin) in py_enumerate(zip(c_txn.vin, tx.inputs)):
for n_vin, psbtin in py_enumerate(tx.inputs):
# Get bip32 path to use to sign, if required for this input
path = None
multisig_input = len(psbtin.hd_keypaths) > 1
@ -252,7 +259,12 @@ class JadeClient(HardwareWalletClient):
if psbtin.witness_utxo:
utxo = psbtin.witness_utxo
if psbtin.non_witness_utxo:
utxo = psbtin.non_witness_utxo.vout[txin.prevout.n]
if psbtin.prev_txid != psbtin.non_witness_utxo.hash:
raise BadArgumentError(
f"Input {n_vin} has a non_witness_utxo with the wrong hash"
)
assert psbtin.prev_out is not None
utxo = psbtin.non_witness_utxo.vout[psbtin.prev_out]
input_txn_bytes = (
psbtin.non_witness_utxo.serialize_without_witness()
)
@ -260,6 +272,7 @@ class JadeClient(HardwareWalletClient):
raise Exception(
"PSBT is missing input utxo information, cannot sign"
)
sats_value = utxo.nValue
scriptcode = utxo.scriptPubKey
if is_p2sh(scriptcode):
@ -311,9 +324,10 @@ class JadeClient(HardwareWalletClient):
jade_inputs.append(
{
"is_witness": witness_input,
"input_tx": input_txn_bytes,
"satoshi": sats_value,
"script": scriptcode,
"path": path,
"input_tx": input_txn_bytes,
"ae_host_entropy": os.urandom(32),
"ae_host_commitment": os.urandom(32),
}
@ -321,157 +335,123 @@ class JadeClient(HardwareWalletClient):
# 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
# and the user need not confirm that output. 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 = {
# Skip automatic change validation in expert mode - user checks *every* output on hw
if not self.expert:
# 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
if signing_multisigs:
registered_multisigs = self.jade.get_registered_multisigs()
signing_multisigs = {
k: v
for k, v in signing_multisigs.items()
if len(v[2]) == num_signers
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])
}
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
# 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
parsed = parse_multisig(scriptcode)
if parsed:
script_variant = self._convertAddrType(
change_addr_type, multisig=True
change_addr_type, multisig=False
)
threshold = parsed[0]
change[n_vout] = {
"path": origin.path,
"variant": script_variant,
}
pubkeys = parsed[1]
hd_keypath_origins = [
psbtout.hd_keypaths[pubkey] for pubkey in pubkeys
]
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
signers, paths = _parse_signers(hd_keypath_origins)
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
multisig_name = self._get_multisig_name(
script_variant, threshold, signers
)
parsed = parse_multisig(scriptcode)
if parsed:
script_variant = self._convertAddrType(
change_addr_type, multisig=True
)
threshold = parsed[0]
matched_multisig = candidate_multisigs.get(
multisig_name
) == (script_variant, threshold, signers)
if matched_multisig:
change[n_vout] = {
"paths": paths,
"multisig_name": multisig_name,
}
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
)
if (
matched_multisig
and matched_multisig[0] == script_variant
and matched_multisig[1] == threshold
and sorted(matched_multisig[2])
== sorted(signers)
):
change[n_vout] = {
"paths": paths,
"multisig_name": multisig_name,
}
# The txn itself
txn_bytes = c_txn.serialize_without_witness()
@ -499,23 +479,25 @@ class JadeClient(HardwareWalletClient):
path = parse_path(bip32_path)
if isinstance(message, bytes) or isinstance(message, bytearray):
message = message.decode("utf-8")
signature = self.jade.sign_message(path, message)
return signature
# Display address of specified type on the device. Only supports single-key based addresses atm.
# NOTE: tests fail if we try to use AE signatures, so stick with default (rfc6979)
signature = self.jade.sign_message(path, message)
return str(signature)
# Display address of specified type on the device.
@jade_exception
def display_singlesig_address(self, bip32_path: str, addr_type: AddressType) -> str:
path = parse_path(bip32_path)
addr_type = self._convertAddrType(addr_type)
script_variant = self._convertAddrType(addr_type, multisig=False)
address = self.jade.get_receive_address(
self._network(), path, variant=addr_type
self._network(), path, variant=script_variant
)
return address
return str(address)
# Display multisig address of specified type on the device.
@jade_exception
def display_multisig_address(
self,
addr_type: AddressType,
multisig: MultisigDescriptor,
self, addr_type: AddressType, multisig: MultisigDescriptor
) -> str:
signer_origins = []
signers = []
@ -531,7 +513,15 @@ class JadeClient(HardwareWalletClient):
)
if pubkey.deriv_path is None:
raise BadArgumentError(
"Blockstream Jade can only generate addresses for multisigs with key origin derivation path information"
"Blockstream Jade can only generate addresses for multisigs with key derivation paths"
)
if pubkey.origin.path and not is_hardened(pubkey.origin.path[-1]):
logging.warning(
f"Final element of origin path {pubkey.origin.path} unhardened"
)
logging.warning(
"Blockstream Jade may not be able to identify change sent back to this descriptor"
)
# Tuple to derive deterministic name for the registrtion
@ -555,13 +545,13 @@ class JadeClient(HardwareWalletClient):
)
paths.append(parse_path(path))
# sort origins, signers and paths according to origins (like in _get_multisig_name)
# But, only sort if sorted_multi is used (and thus the order of xpubs is not relevant)
if multisig.is_sorted:
signer_origins, signers, paths = [
list(a) for a in zip(*sorted(zip(signer_origins, signers, paths)))
]
# Get a deterministic name for this multisig wallet
if multisig.is_sorted and paths[:-1] != paths[1:]:
logging.warning("Sorted multisig with different derivations per signer")
logging.warning(
"Blockstream Jade may not be able to validate change sent back to this descriptor"
)
# Get a deterministic name for this multisig wallet (ignoring bip67 key sorting)
script_variant = self._convertAddrType(addr_type, multisig=True)
multisig_name = self._get_multisig_name(
script_variant, multisig.thresh, signer_origins
@ -583,262 +573,152 @@ class JadeClient(HardwareWalletClient):
return str(address)
# Setup a new device
def setup_device(self, label="", passphrase=""):
"""
The Blockstream Jade does not support setup via software.
# Custom Specter method - register multisig on the Jade
@jade_exception
def register_multisig(self, descriptor: str) -> None:
:raises UnavailableActionError: Always, this function is unavailable
"""
raise UnavailableActionError(
"The Blockstream Jade does not support software setup"
descriptor = Descriptor.from_string(descriptor)
signer_origins = []
signers = []
paths = []
for key in descriptor.keys:
# Tuple to derive deterministic name for the registration
signer_origins.append((key.origin.fingerprint, key.origin.derivation))
# We won't include the additional path in the multisig registration
signers.append(
{
"fingerprint": key.fingerprint,
"derivation": key.derivation,
"xpub": key.key.to_string(),
"path": [],
}
)
# Get a deterministic name for this multisig wallet (ignoring bip67 key sorting)
if descriptor.wsh and not descriptor.sh:
addr_type = AddressType.WIT
elif descriptor.wsh and descriptor.sh:
addr_type = AddressType.SH_WIT
elif descriptor.wsh.is_legacy:
addr_type = AddressType.LEGACY
else:
raise BadArgumentError(
"The script type of the descriptor does not match any standard type."
)
script_variant = self._convertAddrType(addr_type, multisig=True)
threshold = descriptor.miniscript.args[0].num # hackish ...
multisig_name = self._get_multisig_name(
script_variant, threshold, signer_origins
)
# Wipe this device
def wipe_device(self):
# 're-registering' is a no-op
self.jade.register_multisig(
self._network(),
multisig_name,
script_variant,
descriptor.is_sorted,
threshold,
signers,
)
# Setup a new device
def setup_device(self, label: str = "", passphrase: str = "") -> bool:
"""
The Blockstream Jade does not support wiping via software.
Blockstream Jade does not support setup via software.
:raises UnavailableActionError: Always, this function is unavailable
"""
raise UnavailableActionError("Blockstream Jade does not support software setup")
# Wipe this device
def wipe_device(self) -> bool:
"""
Blockstream Jade does not support wiping via software.
:raises UnavailableActionError: Always, this function is unavailable
"""
raise UnavailableActionError(
"The Blockstream Jade does not support wiping via software"
"Blockstream Jade does not support wiping via software"
)
# Restore device from mnemonic or xprv
def restore_device(self, label="", word_count=24):
def restore_device(self, label: str = "", word_count: int = 24) -> bool:
"""
The Blockstream Jade does not support restoring via software.
Blockstream Jade does not support restoring via software.
:raises UnavailableActionError: Always, this function is unavailable
"""
raise UnavailableActionError(
"The Blockstream Jade does not support restoring via software"
"Blockstream Jade does not support restoring via software"
)
# Begin backup process
def backup_device(self, label="", passphrase=""):
def backup_device(self, label: str = "", passphrase: str = "") -> bool:
"""
The Blockstream Jade does not support backing up via software.
Blockstream Jade does not support backing up via software.
:raises UnavailableActionError: Always, this function is unavailable
"""
raise UnavailableActionError(
"The Blockstream Jade does not support creating a backup via software"
"Blockstream Jade does not support creating a backup via software"
)
# Close the device
def close(self):
def close(self) -> None:
self.jade.disconnect()
# Prompt pin
def prompt_pin(self):
def prompt_pin(self) -> bool:
"""
The Blockstream Jade does not need a PIN sent from the host.
Blockstream Jade does not need a PIN sent from the host.
:raises UnavailableActionError: Always, this function is unavailable
"""
raise UnavailableActionError(
"The Blockstream Jade does not need a PIN sent from the host"
"Blockstream Jade does not need a PIN sent from the host"
)
# Send pin
def send_pin(self, pin):
def send_pin(self, pin: str) -> bool:
"""
The Blockstream Jade does not need a PIN sent from the host.
Blockstream Jade does not need a PIN sent from the host.
:raises UnavailableActionError: Always, this function is unavailable
"""
raise UnavailableActionError(
"The Blockstream Jade does not need a PIN sent from the host"
"Blockstream Jade does not need a PIN sent from the host"
)
# Toggle passphrase
def toggle_passphrase(self):
def toggle_passphrase(self) -> bool:
"""
The Blockstream Jade does not support toggling passphrase from the host.
Blockstream Jade does not support toggling passphrase from the host.
:raises UnavailableActionError: Always, this function is unavailable
"""
raise UnavailableActionError(
"The Blockstream Jade does not support toggling passphrase from the host"
"Blockstream Jade does not support toggling passphrase from the host"
)
def _blind(self, pset, seed: bytes = None):
if seed is None:
seed = pset.unknown.get(b"\xfc\x07specter\x00", os.urandom(32))
txseed = pset.txseed(seed)
# assign blinding factors to all outputs
blinding_outs = []
commitments = []
# because we do sha once (cause taproot), and they want sha twice
hash_prevouts = hashes.sha256(pset.blinded_tx.hash_prevouts())
last_i = 0
last_commitment = {}
for i, out in py_enumerate(pset.outputs):
# skip ones where we don't need blinding
if out.blinding_pubkey is None:
commitments.append(None)
continue
commitment = self.jade.get_commitments(
bytes(reversed(out.asset)), out.value, hash_prevouts, i, vbf=None
)
commitment["blinding_key"] = out.blinding_pubkey
commitments.append(commitment)
last_i = i
last_commitment = commitments[-1]
out.asset_blinding_factor = commitment["abf"]
out.value_blinding_factor = commitment["vbf"]
blinding_outs.append(out)
if len(blinding_outs) == 0:
raise Exception("Nothing to blind")
# calculate last vbf
vals = []
abfs = []
vbfs = []
for sc in pset.inputs + blinding_outs:
value = sc.value if sc.value is not None else sc.utxo.value
asset = sc.asset or sc.utxo.asset
if not (isinstance(value, int) and len(asset) == 32):
continue
vals.append(value)
abfs.append(sc.asset_blinding_factor or b"\x00" * 32)
vbfs.append(sc.value_blinding_factor or b"\x00" * 32)
last_vbf = secp256k1.pedersen_blind_generator_blind_sum(
vals, abfs, vbfs, len(vals) - len(blinding_outs)
)
last_out = blinding_outs[-1]
new_last_commitment = self.jade.get_commitments(
bytes(reversed(last_out.asset)),
last_out.value,
hash_prevouts,
last_i,
vbf=last_vbf,
)
# check abf didn't change
assert new_last_commitment["abf"] == last_out.asset_blinding_factor
# set new values in the last commitment
last_commitment.update(new_last_commitment)
blinding_outs[-1].value_blinding_factor = last_vbf
@jade_exception
def can_sign_taproot(self) -> bool:
"""
Blockstream Jade does not currently support Taproot.
# calculate commitments (surj proof etc)
in_tags = []
in_gens = []
for inp in pset.inputs:
if inp.asset:
in_tags.append(inp.asset)
in_gens.append(secp256k1.generator_parse(inp.utxo.asset))
# if we have unconfidential input
elif len(inp.utxo.asset) == 32:
in_tags.append(inp.utxo.asset)
in_gens.append(secp256k1.generator_generate(inp.utxo.asset))
for i, out in py_enumerate(pset.outputs):
if None in [out.blinding_pubkey, out.value, out.asset_blinding_factor]:
continue
gen = secp256k1.generator_generate_blinded(
out.asset, out.asset_blinding_factor
)
out.asset_commitment = secp256k1.generator_serialize(gen)
value_commitment = secp256k1.pedersen_commit(
out.value_blinding_factor, out.value, gen
)
out.value_commitment = secp256k1.pedersen_commitment_serialize(
value_commitment
)
proof_seed = hashes.tagged_hash(
"liquid/surjection_proof", txseed + i.to_bytes(4, "little")
)
proof, in_idx = secp256k1.surjectionproof_initialize(
in_tags, out.asset, proof_seed
)
secp256k1.surjectionproof_generate(
proof, in_idx, in_gens, gen, abfs[in_idx], out.asset_blinding_factor
)
out.surjection_proof = secp256k1.surjectionproof_serialize(proof)
del proof
# generate range proof
rangeproof_nonce = hashes.tagged_hash(
"liquid/range_proof", txseed + i.to_bytes(4, "little")
)
# reblind with extra message for unblinding of change outs
extra_message = (
out.unknown.get(b"\xfc\x07specter\x01", b"")
if out.bip32_derivations
else b""
)
out.reblind(
rangeproof_nonce,
extra_message=extra_message,
)
return commitments
def sign_pset(self, b64pset: str) -> str:
"""Signs specter-desktop specific Liquid PSET transaction"""
mfp = self.get_master_fingerprint()
pset = PSET.from_string(b64pset)
commitments = self._blind(pset)
ins = [
{
"is_witness": True,
# "input_tx": inp.non_witness_utxo.serialize(),
"script": inp.witness_script.data
if inp.witness_script
else script.p2pkh_from_p2wpkh(inp.script_pubkey).data,
"value_commitment": write_commitment(inp.utxo.value),
"path": [
der
for der in inp.bip32_derivations.values()
if der.fingerprint == mfp
][0].derivation,
}
for inp in pset.inputs
]
change = [
{
"path": [
der
for pub, der in out.bip32_derivations.items()
if der.fingerprint == mfp
][0].derivation,
"variant": self._get_script_type(out),
}
if out.bip32_derivations and self._get_script_type(out) is not None
else None
for out in pset.outputs
]
rawtx = pset.blinded_tx.serialize()
signatures = self.jade.sign_liquid_tx(
self._network(), rawtx, ins, commitments, change
)
for i, inp in py_enumerate(pset.inputs):
inp.partial_sigs[
[
pub
for pub, der in inp.bip32_derivations.items()
if der.fingerprint == mfp
][0]
] = signatures[i]
# we must finalize here because it has different commitments and only supports singlesig
return str(finalize_psbt(pset))
def _get_script_type(self, out):
if out.script_pubkey.script_type() == "p2pkh":
return "pkh(k)"
elif out.script_pubkey.script_type() == "p2wpkh":
return "wpkh(k)"
elif out.script_pubkey.script_type() == "p2sh":
if out.redeem_script.script_type() == "p2wpkh":
return "sh(wpkh(k))"
# otherwise None
return None
:returns: False, always
"""
return False
def enumerate(password: str = "") -> List[Dict[str, Any]]:
def enumerate(
password: Optional[str] = None,
expert: bool = False,
chain: Chain = Chain.MAIN,
skip_unlocking=True,
) -> List[Dict[str, Any]]:
results = []
def _get_device_entry(device_model: str, device_path: str) -> Dict[str, Any]:
@ -851,9 +731,13 @@ def enumerate(password: str = "") -> List[Dict[str, Any]]:
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()
client = JadeClient(
device_path, password, expert, chain, skip_unlocking, timeout=1
)
# The Jade could already be unlocked upon startup (this is the only instance where unlock_required is False right now).
# But, we don't need the fingerpint then.
if client and not skip_unlocking:
d_data["fingerprint"] = client.get_master_fingerprint().hex()
if client:
client.close()
@ -874,9 +758,9 @@ def enumerate(password: str = "") -> List[Dict[str, Any]]:
if verinfo is not None:
results.append(_get_device_entry("jade_simulator", SIMULATOR_PATH))
except ConnectionRefusedError as e:
except Exception as e:
# If we get any sort of error do not add the simulator
logger.debug(f"Failed to connect to Jade simulator at {SIMULATOR_PATH}")
logger.debug(e)
logging.debug(f"Failed to connect to Jade simulator at {SIMULATOR_PATH}")
logging.debug(e)
return results

2
src/cryptoadvance/specter/devices/hwi/jadepy/__init__.py
View file

@ -1,4 +1,4 @@
from .jade import JadeAPI
from .jade_error import JadeError
__version__ = "0.0.1"
__version__ = "0.2.0"

1879
src/cryptoadvance/specter/devices/hwi/jadepy/jade.py
View file

File diff suppressed because it is too large Load diff

244
src/cryptoadvance/specter/devices/hwi/jadepy/jade_ble.py Normal file
View file

@ -0,0 +1,244 @@
import logging
import asyncio
import aioitertools
import collections
import subprocess
import platform
import bleak
from .jade_error import JadeError
logger = logging.getLogger(__name__)
#
# Low-level BLE backend interface to Jade
# Calls to send and receive bytes over the interface.
# Intended for use via JadeInterface wrapper.
#
# Either:
# a) use via JadeInterface.create_ble() (see JadeInterface)
# (recommended)
# or:
# b) use JadeBleImpl() directly, and call connect() before
# using, and disconnect() when finished,
# (caveat cranium)
#
class JadeBleImpl:
IO_SERVICE_UUID = "6e400001-b5a3-f393-e0a9-e50e24dcca9e"
IO_TX_CHAR_UUID = "6e400002-b5a3-f393-e0a9-e50e24dcca9e"
IO_RX_CHAR_UUID = "6e400003-b5a3-f393-e0a9-e50e24dcca9e"
BLE_MAX_WRITE_SIZE = 517 - 8
def __init__(self, device_name, serial_number, scan_timeout, loop=None):
self.device_name = device_name
self.serial_number = serial_number
self.scan_timeout = max(1, scan_timeout)
self.inputstream = None
self.write_task = None
self.client = None
self.rx_char_handle = None
if not loop:
loop = asyncio.get_event_loop()
self.loop = loop
# Helper to await async coroutines
def _run(self, coro):
assert coro and self.loop and not self.loop.is_closed()
return self.loop.run_until_complete(coro)
async def _connect_impl(self):
assert self.client is None
# Input received, buffered awaiting external read
inbufs = collections.deque()
# Async stream of those items for reading
async def _input_stream():
# Poll for new input all the time client exists
while self.client is not None:
while inbufs:
buf = inbufs.popleft()
for b in buf:
yield b
# No data, yield to event loop awaiting arrival of more data
await asyncio.sleep(0.01)
# Stream drained and client connection no longer exists
self.inputstream = None
self.inputstream = _input_stream()
# Scan for expected ble device
# Match device-name only if no serial number provided
device_mac = None
while not device_mac and self.scan_timeout > 0:
logger.info("Scanning, timeout = {}s".format(self.scan_timeout))
scan_time = min(2, self.scan_timeout)
self.scan_timeout -= scan_time
devices = await bleak.discover(scan_time)
for dev in devices:
logger.debug("Seen: {}".format(dev.name))
if (
dev.name
and dev.name.startswith(self.device_name)
and (
self.serial_number is None
or dev.name.endswith(self.serial_number)
)
):
# Map pretty name to mac-type address
device_mac = dev.address
full_name = dev.name
if not device_mac:
raise JadeError(
1,
"Unable to locate BLE device",
"Device name: {}, Serial number: {}".format(
self.device_name, self.serial_number or "<any>"
),
)
# Remove previous bt/ble pairing data for this device
if platform.system() == "Linux":
command = "bt-device --remove '{}'".format(device_mac)
process = subprocess.run(command, shell=True, stdout=subprocess.DEVNULL)
# Connect - seems pretty flaky so allow retries
connected = False
attempts_remaining = 3
while not connected:
try:
attempts_remaining -= 1
client = bleak.BleakClient(device_mac)
logger.info("Connecting to: {} ({})".format(full_name, device_mac))
await client.connect()
connected = client.is_connected
logger.info("Connected: {}".format(connected))
except Exception as e:
logger.warning("BLE connection exception: '{}'".format(e))
if not attempts_remaining:
logger.warning("Exhausted retries - BLE connection failed")
raise
# Peruse services and characteristics
# Get the 'handle' of the receiving charactersitic
for service in client.services:
for char in service.characteristics:
if char.uuid == JadeBleImpl.IO_RX_CHAR_UUID:
logger.debug(
"Found RX characterisitic - handle: ".format(char.handle)
)
self.rx_char_handle = char.handle
if "read" in char.properties:
await client.read_gatt_char(char.uuid)
for descriptor in char.descriptors:
await client.read_gatt_descriptor(descriptor.handle)
# Attach handler to be notified of new data on the receiving characteristic
def _notification_handler(char_handle, data):
assert char_handle == self.rx_char_handle
inbufs.append(data)
assert self.rx_char_handle
await client.start_notify(self.rx_char_handle, _notification_handler)
# Attach handler called when disconnected
def _disconnection_handler(client):
# Set the client to None - that will cause the receive
# generator to terminate and not wait forever for data.
assert client == self.client
self.client = None
# Also cancel any running task trying to write data,
# as otherwise that hangs forever too ...
if self.write_task:
self.write_task.cancel()
self.write_task = None
client.set_disconnected_callback(_disconnection_handler)
# Done
self.client = client
def connect(self):
return self._run(self._connect_impl())
async def _disconnect_impl(self):
try:
if self.client is not None and self.client.is_connected:
# Stop listening for incoming data
if self.rx_char_handle:
await self.client.stop_notify(self.rx_char_handle)
# Disconnect underlying client - this should trigger the _disconnection_handler()
# above to run before this returns from the 'await'
await self.client.disconnect()
except Exception as err:
# Sometimes get an exception when testing connection
# if the client has already internally disconnected ...
logger.warning("Exception when disconnecting ble: {}".format(err))
# Set the client to None in any case - that will cause the receive
# generator to terminate and not wait forever for data.
self.rx_char_handle = None
self.client = None
def disconnect(self):
return self._run(self._disconnect_impl())
async def _write_impl(self, bytes_):
assert self.client is not None
assert self.write_task is None
towrite = len(bytes_)
written = 0
async def _write():
if self.client is not None:
nonlocal written
# Write out in small chunks
while written < towrite:
remaining = towrite - written
length = min(remaining, JadeBleImpl.BLE_MAX_WRITE_SIZE)
ulimit = written + length
await self.client.write_gatt_char(
JadeBleImpl.IO_TX_CHAR_UUID,
bytearray(bytes_[written:ulimit]),
response=True,
)
written = ulimit
# Hold on to the write task in case we need to cancel it
# whie it is running (eg. unexpected disconnection)
self.write_task = asyncio.create_task(_write())
try:
await self.write_task
except asyncio.CancelledError:
logger.warning(
"write() task cancelled having written "
"{} of {} bytes".format(written, towrite)
)
finally:
self.write_task = None
return written
def write(self, bytes_):
return self._run(self._write_impl(bytes_))
async def _read_impl(self, n):
assert self.inputstream is not None
return bytes([b async for b in aioitertools.islice(self.inputstream, n)])
def read(self, n):
return self._run(self._read_impl(n))

41
src/cryptoadvance/specter/devices/hwi/jadepy/jade_serial.py
View file

@ -1,8 +1,9 @@
import serial
import logging
from serial.tools import list_ports
logger = logging.getLogger("jade.serial")
logger = logging.getLogger(__name__)
#
@ -19,8 +20,28 @@ logger = logging.getLogger("jade.serial")
# (caveat cranium)
#
class JadeSerialImpl:
# Used when searching for devices that might be a Jade/compatible hw
JADE_DEVICE_IDS = [
(0x10C4, 0xEA60),
(0x1A86, 0x55D4),
(0x0403, 0x6001),
(0x1A86, 0x7523),
]
@classmethod
def _get_first_compatible_device(cls):
jades = []
for devinfo in list_ports.comports():
if (devinfo.vid, devinfo.pid) in cls.JADE_DEVICE_IDS:
jades.append(devinfo.device)
if len(jades) > 1:
logger.warning(f"Multiple potential jade devices detected: {jades}")
return jades[0] if jades else None
def __init__(self, device, baud, timeout):
self.device = device
self.device = device or self._get_first_compatible_device()
self.baud = baud
self.timeout = timeout
self.ser = None
@ -33,12 +54,24 @@ class JadeSerialImpl:
self.device, self.baud, timeout=self.timeout, write_timeout=self.timeout
)
assert self.ser is not None
self.ser.__enter__()
if not self.ser.is_open:
self.ser.open()
# Ensure RTS and DTR are not set (as this can cause the hw to reboot)
self.ser.setRTS(False)
self.ser.setDTR(False)
logger.info("Connected")
def disconnect(self):
assert self.ser is not None
self.ser.__exit__()
# Ensure RTS and DTR are not set (as this can cause the hw to reboot)
# and then close the connection
self.ser.setRTS(False)
self.ser.setDTR(False)
self.ser.close()
# Reset state
self.ser = None

11
src/cryptoadvance/specter/devices/hwi/jadepy/jade_tcp.py
View file

@ -2,7 +2,7 @@ import socket
import logging
logger = logging.getLogger("jade.tcp")
logger = logging.getLogger(__name__)
#
@ -25,9 +25,10 @@ class JadeTCPImpl:
def isSupportedDevice(cls, device):
return device is not None and device.startswith(cls.PROTOCOL_PREFIX)
def __init__(self, device):
def __init__(self, device, timeout):
assert self.isSupportedDevice(device)
self.device = device
self.timeout = timeout
self.tcp_sock = None
def connect(self):
@ -36,6 +37,7 @@ class JadeTCPImpl:
logger.info("Connecting to {}".format(self.device))
self.tcp_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.tcp_sock.settimeout(self.timeout)
url = self.device[len(self.PROTOCOL_PREFIX) :].split(":")
self.tcp_sock.connect((url[0], int(url[1])))
@ -57,4 +59,7 @@ class JadeTCPImpl:
def read(self, n):
assert self.tcp_sock is not None
return self.tcp_sock.recv(n)
buf = self.tcp_sock.recv(n)
while len(buf) < n:
buf += self.tcp_sock.recv(n - len(buf))
return buf

1
src/cryptoadvance/specter/devices/jade.py
View file

@ -16,6 +16,7 @@ class Jade(HWIDevice):
supports_qr_message_signing = True
supports_hwi_toggle_passphrase = False
supports_hwi_multisig_display_address = True
supports_multisig_registration = True
liquid_support = True
@classmethod

113
src/cryptoadvance/specter/hwi_rpc.py
View file

@ -66,7 +66,7 @@ class HWIBridge(JSONRPC):
All methods of this class are callable over JSON-RPC, except _underscored.
"""
def __init__(self, enforce_hwi_initialisation=False):
def __init__(self, skip_hwi_initialisation=False):
self.exposed_rpc = {
"enumerate": self.enumerate,
"detect_device": self.detect_device,
@ -79,16 +79,20 @@ class HWIBridge(JSONRPC):
"sign_tx": self.sign_tx,
"sign_message": self.sign_message,
"extract_master_blinding_key": self.extract_master_blinding_key,
"register_multisig": self.register_multisig, # currently only Jade
"bitbox02_pairing": self.bitbox02_pairing,
}
if enforce_hwi_initialisation:
# Running enumerate after beginning an interaction with a specific device
# crashes python or make HWI misbehave. For now we just get all connected
# devices once per session and save them.
logger.info("Initializing HWI...") # to explain user why it takes so long
self.enumerate()
logger.info("Finished initializing HWI!")
self.devices = []
if skip_hwi_initialisation:
self.is_startup = False
self.devices = []
return
# Running enumerate after beginning an interaction with a specific device
# crashes python or make HWI misbehave. For now we just get all connected
# devices once per session and save them.
logger.info("Initializing HWI...")
self.is_startup = True # to explain user why it takes so long
self.enumerate()
logger.info("Finished initializing HWI!")
@locked(hwilock)
def enumerate(self, passphrase="", chain=""):
@ -97,37 +101,21 @@ class HWIBridge(JSONRPC):
Standard HWI enumerate() command + Specter.
"""
devices = []
# going through all device classes
# Call device-specific enumerate (can come from hwi lib or from the Specter code base) for each Specter device class
for devcls in hwi_classes:
try:
# calling device-specific enumerate
if passphrase:
devs = devcls.enumerate(passphrase)
# not sure if it will handle passphrase correctly
# so remove it if None
# Special handling of the Jade to unsure it is not prompting to unlock the device on startup
if devcls.__name__ == "Jade":
skip_unlocking = self.is_startup
client_chain = Chain.argparse(chain) # This returns an enum member
devs = devcls.enumerate(
skip_unlocking=skip_unlocking, chain=client_chain
)
else:
devs = devcls.enumerate()
# extracting fingerprint info
for dev in devs:
# we can't get fingerprint if device is locked
if "needs_pin_sent" in dev and dev["needs_pin_sent"]:
continue
# we can't get fingerprint if passphrase is not provided
if (
"needs_passphrase_sent" in dev
and dev["needs_passphrase_sent"]
and not passphrase
):
continue
client = None
try:
client = devcls.get_client(dev["path"], passphrase)
if isinstance(client, Bitbox02Client):
client.set_noise_config(BitBox02NoiseConfig())
dev["fingerprint"] = client.get_master_fingerprint().hex()
finally:
if client is not None:
client.close()
if passphrase:
devs = devcls.enumerate(passphrase)
else:
devs = devcls.enumerate()
devices += devs
except USBError as e:
logger.warning(
@ -138,6 +126,7 @@ class HWIBridge(JSONRPC):
)
self.devices = devices
self.is_startup = False
return self.devices
def detect_device(
@ -316,6 +305,34 @@ class HWIBridge(JSONRPC):
else:
raise Exception("Failed to validate address on device: Unknown Error")
@locked(hwilock)
def register_multisig(
self,
device_type=None,
path=None,
passphrase="",
fingerprint=None,
descriptor="",
chain="",
):
if descriptor == "":
raise Exception("Descriptor must not be empty")
with self._get_client(
device_type=device_type,
fingerprint=fingerprint,
path=path,
passphrase=passphrase,
chain=chain,
) as client:
try:
return client.register_multisig(descriptor)
except Exception as e:
logger.exception(e)
raise Exception(
f"Failed to register multisig on the device. Error: {e}"
)
@locked(hwilock)
def sign_tx(
self,
@ -426,22 +443,24 @@ class HWIBridge(JSONRPC):
)
devcls = get_device_class(device["type"])
if devcls:
client = devcls.get_client(device["path"], passphrase)
# Jade needs the chain/network already here for the the auth_call
if devcls.__name__ == "Jade":
client_chain = Chain.argparse(chain)
client = devcls.get_client(
path=device["path"],
password=passphrase,
expert=False,
chain=client_chain,
)
else:
client = devcls.get_client(device["path"], passphrase)
if not client:
raise Exception(
"The device was identified but could not be reached. Please check it is properly connected and try again"
)
try:
if is_liquid(chain):
client.chain = Chain.TEST if is_testnet(chain) else Chain.MAIN
else:
if type(client) is not JadeClient:
client.chain = Chain.argparse(chain)
# hack for Jade
if type(client) is JadeClient:
if is_liquid(chain):
client.set_liquid_network(chain)
elif chain == "signet":
client.chain = Chain.TEST
yield client
finally:
client.close()

2
src/cryptoadvance/specter/server.py
View file

@ -182,7 +182,7 @@ def init_app(app: SpecterFlask, hwibridge=False, specter=None):
specter.initialize()
# HWI
specter.hwi = HWIBridge(app.config["ENFORCE_HWI_INITIALISATION_AT_STARTUP"])
specter.hwi = HWIBridge(app.config["SKIP_HWI_INITIALISATION_AT_STARTUP"])
login_manager = LoginManager()
login_manager.session_protection = app.config.get("SESSION_PROTECTION", "strong")

8
src/cryptoadvance/specter/server_endpoints/wallets/wallets.py
View file

@ -788,6 +788,13 @@ def addresses(wallet_alias):
@login_required
def settings(wallet_alias):
wallet: Wallet = app.specter.wallet_manager.get_by_alias(wallet_alias)
# Check whether wallet has at least one device which supports multisig registrations (currently only Jade)
has_device_for_multisig_registration = any(
getattr(device, "supports_multisig_registration", False)
for device in wallet.devices
)
if request.method == "POST":
action = request.form["action"]
# Would like to refactor this to another endpoint as well
@ -812,6 +819,7 @@ def settings(wallet_alias):
purposes=purposes,
wallet_alias=wallet_alias,
wallet=wallet,
has_device_for_multisig_registration=has_device_for_multisig_registration,
specter=app.specter,
rand=rand,
scroll_to_rescan_blockchain=request.args.get("rescan_blockchain"),

45
src/cryptoadvance/specter/static/hwi.js
View file

@ -1,3 +1,7 @@
function generateId() {
return Date.now().toString(36) + Math.random().toString(36).substring(2);
}
class HWIBridge {
constructor(url, chain) {
this.url = url;
@ -5,7 +9,7 @@ class HWIBridge {
this.chain = chain;
this.in_progress = false;
}
async fetch(command, params={}, timeout=0){
async requestToHwiBridge(command, params={}, timeout=0){
if(this.in_progress){
throw "HWI is busy processing previous request.";
}
@ -17,9 +21,10 @@ class HWIBridge {
const timeoutId = setTimeout(() => controller.abort(), timeout);
}
try{
if (command != 'detect_device' && command != 'enumerate') {
if (command != 'detect_device') {
params.chain = this.chain;
}
const requestId = generateId()
data = await fetch(this.url, {
method: 'POST',
headers: {
@ -30,7 +35,7 @@ class HWIBridge {
body: JSON.stringify({
'jsonrpc': '2.0',
'method': command,
'id': 1,
'id': requestId,
params,
forwarded_request: (this.url !== '/hwi/api/'),
})
@ -44,13 +49,13 @@ class HWIBridge {
return data.result;
}
async enumerate(passphrase="", useTimeout){
return await this.fetch("enumerate", {
return await this.requestToHwiBridge("enumerate", {
passphrase
}, (useTimeout ? 60000 : 0));
}
async detectDevice(type, rescan=true){
// TODO: fingerprint, path, type
return await this.fetch("detect_device",
return await this.requestToHwiBridge("detect_device",
{ device_type: type, rescan_devices: rescan });
}
@ -59,7 +64,7 @@ class HWIBridge {
Tells the server to send the 'togglepassphrase' command to the device.
KeepKey and Trezor only.
**/
return await this.fetch('toggle_passphrase', {
return await this.requestToHwiBridge('toggle_passphrase', {
device_type: device.type,
path: device.path
});
@ -70,7 +75,7 @@ class HWIBridge {
Asks the HWI server for a pairing code for BitBox02.
Returns {"code": ""} with the code or an empty string if none found.
**/
return await this.fetch('bitbox02_pairing', {});
return await this.requestToHwiBridge('bitbox02_pairing', {});
}
async promptPin(device, passphrase="") {
@ -81,7 +86,7 @@ class HWIBridge {
if(!('passphrase' in device)){
device.passphrase = passphrase;
}
return await this.fetch('prompt_pin', {
return await this.requestToHwiBridge('prompt_pin', {
device_type: device.type,
path: device.path,
passphrase: device.passphrase,
@ -96,7 +101,7 @@ class HWIBridge {
if(!('passphrase' in device)){
device.passphrase = passphrase;
}
return await this.fetch('send_pin', {
return await this.requestToHwiBridge('send_pin', {
device_type: device.type,
path: device.path,
passphrase: device.passphrase,
@ -111,7 +116,7 @@ class HWIBridge {
if(!('passphrase' in device)){
device.passphrase = passphrase;
}
return await this.fetch('sign_tx', {
return await this.requestToHwiBridge('sign_tx', {
device_type: device.type,
path: device.path,
passphrase: device.passphrase,
@ -126,7 +131,7 @@ class HWIBridge {
if(!('passphrase' in device)){
device.passphrase = passphrase;
}
return await this.fetch('sign_message', {
return await this.requestToHwiBridge('sign_message', {
device_type: device.type,
path: device.path,
passphrase: device.passphrase,
@ -139,7 +144,7 @@ class HWIBridge {
if(!('passphrase' in device)){
device.passphrase = passphrase;
}
return await this.fetch('extract_xpubs', {
return await this.requestToHwiBridge('extract_xpubs', {
device_type: device.type,
account: account,
path: device.path,
@ -153,7 +158,7 @@ class HWIBridge {
if(!('passphrase' in device)){
device.passphrase = passphrase;
}
return await this.fetch('extract_xpub', {
return await this.requestToHwiBridge('extract_xpub', {
device_type: device.type,
derivation: derivation,
path: device.path,
@ -166,7 +171,7 @@ class HWIBridge {
if(!('passphrase' in device)){
device.passphrase = passphrase;
}
return await this.fetch('extract_master_blinding_key', {
return await this.requestToHwiBridge('extract_master_blinding_key', {
device_type: device.type,
path: device.path,
passphrase: device.passphrase,
@ -179,7 +184,7 @@ class HWIBridge {
if(!('passphrase' in device)){
device.passphrase = passphrase;
}
return await this.fetch('display_address', {
return await this.requestToHwiBridge('display_address', {
device_type: device.type,
path: device.path,
passphrase: device.passphrase,
@ -187,4 +192,14 @@ class HWIBridge {
xpubs_descriptor: xpubs_descriptor,
});
}
async registerMultisig(device, descriptor, fingerprint) {
return await this.requestToHwiBridge('register_multisig', {
device_type: device.type,
path: device.path,
passphrase: device.passphrase,
fingerprint: device.fingerprint,
descriptor: descriptor,
})
}
}

30
src/cryptoadvance/specter/templates/includes/hwi/components/wallet.jinja
View file

@ -96,6 +96,36 @@
}
}
}
async function registerMultisig(descriptor, fingerprint) {
const devices = await enumerate()
if (!devices || devices.length === 0) {
return
}
const device = await selectDevice(devices)
if (!device) {
return
}
if (fingerprint && device.fingerprint != fingerprint) {
handleHWIError("Device fingerprints don't match. You have probably selected the wrong device.")
return
}
showHWIProgress("Registering multisig ...", `Confirm on your ${capitalize(device.type)}`)
try {
await hwi.registerMultisig(device, descriptor)
} catch (error) {
handleHWIError(error)
return
}
hidePageOverlay()
showNotification("Multisig registered successfully!", 3000);
}
</script>
{% endif %}

4
src/cryptoadvance/specter/templates/includes/hwi/hwi.jinja
View file

@ -93,7 +93,7 @@
let retryCounter = 0;
try {
try {
result = await hwi.enumerate(passphrase, deviceTypes == 'bitbox02');
result = await hwi.enumerate(passphrase, true);
} catch (e) {
if (e.message !== 'Fetch is aborted' && e.message !== 'The user aborted a request.') {
throw e;
@ -122,7 +122,7 @@
}
console.log("Retrying enumerate...");
try {
result = await hwi.enumerate(passphrase, deviceTypes == 'bitbox02');
result = await hwi.enumerate(passphrase, true);
} catch (e) {
if (e.message !== 'Fetch is aborted' && e.message !== 'The user aborted a request.') {
throw e;

21
src/cryptoadvance/specter/templates/wallet/settings/wallet_settings.jinja
View file

@ -1,6 +1,7 @@
{% extends "wallet/components/wallet_tab.jinja" %}
{% include "includes/file-uploader.html" %}
{% include "includes/dnd-textarea.html" %}
{% include "includes/hwi/hwi.jinja" %}
{% set tab = 'settings' %}
{% block content %}
@ -21,6 +22,21 @@
<div id="wallet_info_settings_tab">
{% if wallet.is_multisig %}
<h3>{{ _("Devices") }}</h3>
{% if has_device_for_multisig_registration %}
<p>Register Multisig</p>
<div class="flex flex-col mt-1 mb-2">
{% for device in wallet.devices %}
{% if device.supports_multisig_registration %}
{% for wallet_key in wallet.keys %}
{% set matching_device_key = device.keys | selectattr("fingerprint", "eq", wallet_key.fingerprint) | first %}
{% if matching_device_key %}
<button class="button p-1" type="button" onclick="registerMultisigOnDevice('{{ matching_device_key.fingerprint }}');">Register on {{ device.name }}</button>
{% endif %}
{% endfor %}
{% endif %}
{% endfor %}
</div>
{% endif %}
<p>{{ wallet.sigs_required }} out of {{ wallet.keys|length }} multisig</p>
{% else %}
<h3>{{ _("Device") }}</h3>
@ -358,6 +374,11 @@
document.getElementById('export_specter_format').href = "data:text/json;charset=utf-8," + walletDataSpecterFormat;
});
const registerMultisigOnDevice = async (fingerprint) => {
const descriptor = '{{ wallet.descriptor }}'
await registerMultisig(descriptor, fingerprint)
}
function toggleKeysList() {
let titleButton = document.getElementById('toggle_keys_list');
let keysList = document.getElementById('keys_list');