diff --git a/hwilib/_cli.py b/hwilib/_cli.py index 4a17a53f8..52077ac08 100644 --- a/hwilib/_cli.py +++ b/hwilib/_cli.py @@ -97,7 +97,7 @@ def signmessage_handler(args: argparse.Namespace, client: HardwareWalletClient) return signmessage(client, message=args.message, path=args.path) def signtx_handler(args: argparse.Namespace, client: HardwareWalletClient) -> Dict[str, Union[bool, str]]: - return signtx(client, psbt=args.psbt) + return signtx(client, psbt=args.psbt, registrations=args.registrations) def wipe_device_handler(args: argparse.Namespace, client: HardwareWalletClient) -> Dict[str, bool]: return wipe_device(client) @@ -173,6 +173,12 @@ def get_parser() -> HWIArgumentParser: signtx_parser = subparsers.add_parser('signtx', help='Sign a PSBT') signtx_parser.add_argument('psbt', help='The Partially Signed Bitcoin Transaction to sign') + signtx_parser.add_argument( + '--registration', + dest='registrations', + action='append', + help='Registration returned by the registerdescriptor command; may be specified multiple times', + ) signtx_parser.set_defaults(func=signtx_handler) getxpub_parser = subparsers.add_parser('getxpub', help='Get an extended public key') diff --git a/hwilib/commands.py b/hwilib/commands.py index eb47465d8..7371a5356 100644 --- a/hwilib/commands.py +++ b/hwilib/commands.py @@ -183,19 +183,28 @@ def getmasterxpub(client: HardwareWalletClient, addrtype: AddressType = AddressT """ return {"xpub": client.get_master_xpub(addrtype, account).to_string()} -def signtx(client: HardwareWalletClient, psbt: str) -> Dict[str, Union[bool, str]]: +def signtx( + client: HardwareWalletClient, + psbt: str, + registrations: Optional[List[str]] = None, +) -> Dict[str, Union[bool, str]]: """ Sign a Partially Signed Bitcoin Transaction (PSBT) with the client. :param client: The client to interact with :param psbt: The PSBT to sign + :param registrations: Serialized registered descriptors for BIP388 policy signing :return: A dictionary containing the processed PSBT serialized in Base64. Returned as ``{"psbt": }``. """ # Deserialize the transaction tx = PSBT() tx.deserialize(psbt) - result = client.sign_tx(tx).serialize() + registered_descriptors = { + RegisteredDescriptor.deserialize(registration) + for registration in registrations or [] + } + result = client.sign_tx(tx, registered_descriptors).serialize() return {"psbt": result, "signed": result != psbt} def getxpub(client: HardwareWalletClient, path: str, expert: bool = False) -> Dict[str, Any]: diff --git a/hwilib/descriptor.py b/hwilib/descriptor.py index 14c8098ad..1373ad2d2 100644 --- a/hwilib/descriptor.py +++ b/hwilib/descriptor.py @@ -17,7 +17,7 @@ multipath_to_string, path_to_string, ) -from .common import hash160, sha256 +from .common import AddressType from .errors import BadArgumentError, InvalidPolicyError from ._serialize import ( deser_compact_size, @@ -28,7 +28,6 @@ from base64 import b64decode, b64encode from binascii import unhexlify -from collections import namedtuple from copy import deepcopy from enum import Enum from io import BufferedReader, BytesIO @@ -42,8 +41,6 @@ MAX_TAPROOT_NODES = 128 -ExpandedScripts = namedtuple("ExpandedScripts", ["output_script", "redeem_script", "witness_script"]) - def PolyMod(c: int, val: int) -> int: """ :meta private: @@ -315,11 +312,9 @@ def to_string(self, hardened_char: str = "h") -> str: """ return AddChecksum(self.to_string_no_checksum(hardened_char)) - def expand(self, pos: int) -> "ExpandedScripts": - """ - Returns the scripts for a descriptor at the given `pos` for ranged descriptors. - """ - raise NotImplementedError("The Descriptor base class does not implement this method") + def get_address_type(self) -> Optional[AddressType]: + """Return the address type, or ``None`` for descriptors without an address encoding.""" + return None def get_bip388_template(self) -> str: """ @@ -394,9 +389,8 @@ def __init__( """ super().__init__([pubkey], [], "pkh") - def expand(self, pos: int) -> "ExpandedScripts": - script = b"\x76\xa9\x14" + hash160(self.pubkeys[0].get_pubkey_bytes(pos)) + b"\x88\xac" - return ExpandedScripts(script, None, None) + def get_address_type(self) -> Optional[AddressType]: + return AddressType.LEGACY class WPKHDescriptor(Descriptor): @@ -412,9 +406,8 @@ def __init__( """ super().__init__([pubkey], [], "wpkh") - def expand(self, pos: int) -> "ExpandedScripts": - script = b"\x00\x14" + hash160(self.pubkeys[0].get_pubkey_bytes(pos)) - return ExpandedScripts(script, None, None) + def get_address_type(self) -> Optional[AddressType]: + return AddressType.WIT class MultisigDescriptor(Descriptor): @@ -439,22 +432,6 @@ def __init__( def to_string_no_checksum(self, hardened_char: str = "h") -> str: return "{}({},{})".format(self.name, self.thresh, ",".join([p.to_string(hardened_char) for p in self.pubkeys])) - def expand(self, pos: int) -> "ExpandedScripts": - if self.thresh > 16: - m = b"\x01" + self.thresh.to_bytes(1, "big") - else: - m = (self.thresh + 0x50).to_bytes(1, "big") if self.thresh > 0 else b"\x00" - n = (len(self.pubkeys) + 0x50).to_bytes(1, "big") if len(self.pubkeys) > 0 else b"\x00" - script: bytes = m - der_pks = [p.get_pubkey_bytes(pos) for p in self.pubkeys] - if self.is_sorted: - der_pks.sort() - for pk in der_pks: - script += len(pk).to_bytes(1, "big") + pk - script += n + b"\xae" - - return ExpandedScripts(script, None, None) - def get_bip388_template(self) -> str: return "{}({},{})".format(self.name, self.thresh, ",".join([p.get_bip388_placeholder() for p in self.pubkeys])) @@ -472,11 +449,10 @@ def __init__( """ super().__init__([], [subdescriptor], "sh") - def expand(self, pos: int) -> "ExpandedScripts": - assert len(self.subdescriptors) == 1 - redeem_script, _, witness_script = self.subdescriptors[0].expand(pos) - script = b"\xa9\x14" + hash160(redeem_script) + b"\x87" - return ExpandedScripts(script, redeem_script, witness_script) + def get_address_type(self) -> Optional[AddressType]: + if self.subdescriptors[0].get_address_type() is AddressType.WIT: + return AddressType.SH_WIT + return AddressType.LEGACY class WSHDescriptor(Descriptor): @@ -492,11 +468,8 @@ def __init__( """ super().__init__([], [subdescriptor], "wsh") - def expand(self, pos: int) -> "ExpandedScripts": - assert len(self.subdescriptors) == 1 - witness_script, _, _ = self.subdescriptors[0].expand(pos) - script = b"\x00\x20" + sha256(witness_script) - return ExpandedScripts(script, None, witness_script) + def get_address_type(self) -> Optional[AddressType]: + return AddressType.WIT class TRDescriptor(Descriptor): @@ -517,6 +490,9 @@ def __init__( super().__init__([internal_key], subdescriptors, "tr") self.depths = depths + def get_address_type(self) -> Optional[AddressType]: + return AddressType.TAP + def to_string_no_checksum(self, hardened_char: str = "h") -> str: r = f"{self.name}({self.pubkeys[0].to_string(hardened_char)}" path: List[bool] = [] # Track left or right for each depth diff --git a/hwilib/devices/bitbox02.py b/hwilib/devices/bitbox02.py index d7902d5f3..94d64cb83 100644 --- a/hwilib/devices/bitbox02.py +++ b/hwilib/devices/bitbox02.py @@ -14,6 +14,7 @@ Tuple, List, Sequence, + Set, TypeVar, ) import base64 @@ -594,7 +595,11 @@ def display_multisig_address( return address @bitbox02_exception - def sign_tx(self, psbt: PSBT) -> PSBT: + def sign_tx( + self, + psbt: PSBT, + registered_descriptors: Optional[Set[RegisteredDescriptor]] = None, + ) -> PSBT: """ Sign a transaction with the BitBox02. @@ -604,6 +609,38 @@ def sign_tx(self, psbt: PSBT) -> PSBT: Transactions with legacy inputs are not supported. """ + policy_script_config: Optional[bitbox02.btc.BTCScriptConfigWithKeypath] = None + if registered_descriptors: + if len(registered_descriptors) > 1: + raise BadArgumentError("The BitBox02 can only sign with one registered policy at a time") + registered_descriptor = next(iter(registered_descriptors)) + descriptor = registered_descriptor.descriptor + device_fingerprint = self.get_master_fingerprint() + account_keypath = None + for pubkey in descriptor.get_pubkey_providers(): + if ( + pubkey.origin is None + or pubkey.origin.fingerprint != device_fingerprint + or pubkey.extkey is None + ): + continue + device_xpub = decode_check(self._get_xpub(pubkey.origin.path)) + if not _xpubs_equal_ignoring_version( + device_xpub, + pubkey.extkey.serialize(), + ): + continue + if account_keypath is not None: + raise BadArgumentError("This BitBox02 occurs more than once in the policy") + account_keypath = pubkey.origin.path + if account_keypath is None: + raise BadArgumentError("This BitBox02 is not one of the policy keys") + + policy_script_config = bitbox02.btc.BTCScriptConfigWithKeypath( + script_config=self._bip388_script_config(descriptor), + keypath=account_keypath, + ) + def find_our_key( keypaths: Dict[bytes, KeyOriginInfo] ) -> Tuple[Optional[bytes], Optional[Sequence[int]]]: @@ -650,6 +687,8 @@ def script_config_from_utxo( redeem_script: bytes, witness_script: bytes, ) -> bitbox02.btc.BTCScriptConfigWithKeypath: + if policy_script_config is not None: + return policy_script_config if is_p2pkh(output.scriptPubKey): raise BadArgumentError( "The BitBox02 does not support legacy p2pkh scripts" diff --git a/hwilib/devices/ckcc/protocol.py b/hwilib/devices/ckcc/protocol.py index 54c744e6d..200eb3651 100644 --- a/hwilib/devices/ckcc/protocol.py +++ b/hwilib/devices/ckcc/protocol.py @@ -91,11 +91,17 @@ def sha256(): return b'sha2' @staticmethod - def sign_transaction(length, file_sha, finalize=False, flags=0x0): + def sign_transaction(length, file_sha, finalize=False, flags=0x0, + miniscript_name=None): # must have already uploaded binary, and give expected sha256 assert len(file_sha) == 32 flags |= (STXN_FINALIZE if finalize else 0x00) - return pack('<4sII32s', b'stxn', length, int(flags), file_sha) + rv = pack('<4sII32s', b'stxn', length, int(flags), file_sha) + if miniscript_name: + name = miniscript_name.encode('ascii') + assert 1 <= len(name) <= 32, "name len" + rv += pack('B', len(name)) + name + return rv @staticmethod def sign_message(raw_msg, subpath='m', addr_fmt=AF_CLASSIC): diff --git a/hwilib/devices/coldcard.py b/hwilib/devices/coldcard.py index df8b61b92..5d4eba985 100644 --- a/hwilib/devices/coldcard.py +++ b/hwilib/devices/coldcard.py @@ -75,6 +75,7 @@ Any, Callable, Optional, + Set, ) CC_SIMULATOR_SOCK = '/tmp/ckcc-simulator.sock' @@ -154,25 +155,79 @@ def get_master_fingerprint(self) -> bytes: # quick method to get fingerprint of wallet return struct.pack(' PSBT: - """ - Sign a transaction with the Coldcard. + def _sign_tx_once( + self, + psbt: PSBT, + miniscript_name: Optional[str] = None, + ) -> PSBT: + # Get psbt in hex and then make binary + fd = io.BytesIO(base64.b64decode(psbt.serialize())) + + # learn size (portable way) + sz = fd.seek(0, 2) + fd.seek(0) + + left = sz + chk = sha256() + for pos in range(0, sz, MAX_BLK_LEN): + here = fd.read(min(MAX_BLK_LEN, left)) + if not here: + break + left -= len(here) + result = self.device.send_recv(CCProtocolPacker.upload(pos, sz, here)) + assert result == pos + chk.update(here) - - The Coldcard firmware only supports signing single key and multisig transactions. It cannot sign arbitrary scripts. - - Multisigs need to be registered on the device before a transaction spending that multisig will be signed by the device. - - Multisigs must use BIP 67. This can be accomplished in Bitcoin Core using the `sortedmulti()` descriptor, available in Bitcoin Core 0.20. - """ - self.device.check_mitm() + # do a verify + expect = chk.digest() + result = self.device.send_recv(CCProtocolPacker.sha256()) + assert len(result) == 32 + if result != expect: + raise DeviceFailureError("Wrong checksum:\nexpect: %s\n got: %s" % (b2a_hex(expect).decode('ascii'), b2a_hex(result).decode('ascii'))) - # Get this devices master key fingerprint - xpub = self.device.send_recv(CCProtocolPacker.get_xpub('m/0\''), timeout=None) - master_fp = get_xpub_fingerprint(xpub) + # start the signing process + ok = self.device.send_recv( + CCProtocolPacker.sign_transaction( + sz, + expect, + miniscript_name=miniscript_name, + ), + timeout=None, + ) + assert ok is None + if self.device.is_simulator: + self.device.send_recv(CCProtocolPacker.sim_keypress(b'y')) - # For multisigs, we may need to do multiple passes if we appear in an input multiple times + print("Waiting for OK on the Coldcard...", file=sys.stderr) + + while 1: + time.sleep(0.250) + done = self.device.send_recv(CCProtocolPacker.get_signed_txn(), timeout=None) + if done is None: + continue + break + + if len(done) != 2: + raise DeviceFailureError('Failed: %r' % done) + + result_len, result_sha = done + + result = self.device.download_file(result_len, result_sha, file_number=1) + + psbt = PSBT() + psbt.deserialize(base64.b64encode(result).decode()) + return psbt + + def _sign_without_policy_names( + self, + psbt: PSBT, + master_fp: bytes, + ) -> PSBT: + # For multisigs, we may need to do multiple passes if we appear in an + # input multiple times. passes = 1 if not self.is_edge: - for psbt_in in tx.inputs: + for psbt_in in psbt.inputs: our_keys = 0 for key in psbt_in.hd_keypaths.keys(): keypath = psbt_in.hd_keypaths[key] @@ -181,61 +236,66 @@ def sign_tx(self, tx: PSBT) -> PSBT: if our_keys > passes: passes = our_keys - if tx.version == 2 and not self._supports_psbt_v2(): - tx.convert_to_v0() - for _ in range(passes): - # Get psbt in hex and then make binary - fd = io.BytesIO(base64.b64decode(tx.serialize())) - - # learn size (portable way) - sz = fd.seek(0, 2) - fd.seek(0) - - left = sz - chk = sha256() - for pos in range(0, sz, MAX_BLK_LEN): - here = fd.read(min(MAX_BLK_LEN, left)) - if not here: - break - left -= len(here) - result = self.device.send_recv(CCProtocolPacker.upload(pos, sz, here)) - assert result == pos - chk.update(here) - - # do a verify - expect = chk.digest() - result = self.device.send_recv(CCProtocolPacker.sha256()) - assert len(result) == 32 - if result != expect: - raise DeviceFailureError("Wrong checksum:\nexpect: %s\n got: %s" % (b2a_hex(expect).decode('ascii'), b2a_hex(result).decode('ascii'))) - - # start the signing process - ok = self.device.send_recv(CCProtocolPacker.sign_transaction(sz, expect), timeout=None) - assert ok is None - if self.device.is_simulator: - self.device.send_recv(CCProtocolPacker.sim_keypress(b'y')) + psbt = self._sign_tx_once(psbt) - print("Waiting for OK on the Coldcard...", file=sys.stderr) + return psbt - while 1: - time.sleep(0.250) - done = self.device.send_recv(CCProtocolPacker.get_signed_txn(), timeout=None) - if done is None: - continue - break + def _sign_with_policy_names( + self, + psbt: PSBT, + registered_descriptors: Set[RegisteredDescriptor], + master_fp: bytes, + ) -> PSBT: + for registration in sorted( + registered_descriptors, + key=lambda registration: registration.serialize(), + ): + psbt = self._sign_tx_once(psbt, registration.name) + + # Named policy requests only cover matching inputs. Use an unnamed + # request to infer a policy for any remaining unsigned device inputs. + if any( + psbt_in.has_fingerprint(master_fp) + and not psbt_in.has_signature(master_fp) + for psbt_in in psbt.inputs + ): + psbt = self._sign_tx_once(psbt) - if len(done) != 2: - raise DeviceFailureError('Failed: %r' % done) + return psbt - result_len, result_sha = done + @coldcard_exception + def sign_tx( + self, + psbt: PSBT, + registered_descriptors: Optional[Set[RegisteredDescriptor]] = None, + ) -> PSBT: + """ + Sign a transaction with the Coldcard. - result = self.device.download_file(result_len, result_sha, file_number=1) + - The Coldcard firmware only supports signing single key and multisig transactions. It cannot sign arbitrary scripts. + - Multisigs need to be registered on the device before a transaction spending that multisig will be signed by the device. + - Multisigs must use BIP 67. This can be accomplished in Bitcoin Core using the `sortedmulti()` descriptor, available in Bitcoin Core 0.20. + """ + self.device.check_mitm() - tx = PSBT() - tx.deserialize(base64.b64encode(result).decode()) + # Get this devices master key fingerprint + xpub = self.device.send_recv(CCProtocolPacker.get_xpub('m/0\''), timeout=None) + master_fp = get_xpub_fingerprint(xpub) + + if psbt.version == 2 and not self._supports_psbt_v2(): + psbt.convert_to_v0() + + if self.is_edge and registered_descriptors: + psbt = self._sign_with_policy_names( + psbt, + registered_descriptors, + master_fp, + ) + else: + psbt = self._sign_without_policy_names(psbt, master_fp) - return tx + return psbt @coldcard_exception def sign_message(self, message: Union[str, bytes], keypath: str) -> str: diff --git a/hwilib/devices/digitalbitbox.py b/hwilib/devices/digitalbitbox.py index efab96e83..91440c427 100644 --- a/hwilib/devices/digitalbitbox.py +++ b/hwilib/devices/digitalbitbox.py @@ -23,6 +23,7 @@ Dict, List, Optional, + Set, Tuple, Union, ) @@ -391,17 +392,23 @@ def get_pubkey_at_path(self, path: str) -> ExtendedKey: return xpub @digitalbitbox_exception - def sign_tx(self, tx: PSBT) -> PSBT: - + def sign_tx( + self, + psbt: PSBT, + registered_descriptors: Optional[Set[RegisteredDescriptor]] = None, + ) -> PSBT: + + if registered_descriptors: + raise UnavailableActionError("The Digital Bitbox does not support BIP388 policy signing") # Create a transaction with all scriptsigs blanked out - blank_tx = tx.get_unsigned_tx() + blank_tx = psbt.get_unsigned_tx() # Get the master key fingerprint master_fp = self.get_master_fingerprint() # create sighashes sighash_tuples = [] - for txin, psbt_in, i_num in zip(blank_tx.vin, tx.inputs, range(len(blank_tx.vin))): + for txin, psbt_in, i_num in zip(blank_tx.vin, psbt.inputs, range(len(blank_tx.vin))): sighash = b"" utxo = None if psbt_in.witness_utxo: @@ -497,7 +504,7 @@ def sign_tx(self, tx: PSBT) -> PSBT: # Return early if nothing to do if len(sighash_tuples) == 0: - return tx + return psbt for i in range(0, len(sighash_tuples), 15): tups = sighash_tuples[i:i + 15] @@ -537,9 +544,9 @@ def sign_tx(self, tx: PSBT) -> PSBT: # add sigs to tx for tup, sig in zip(tups, der_sigs): - tx.inputs[tup[2]].partial_sigs[tup[3]] = sig + psbt.inputs[tup[2]].partial_sigs[tup[3]] = sig - return tx + return psbt @digitalbitbox_exception def sign_message(self, message: Union[str, bytes], keypath: str) -> str: diff --git a/hwilib/devices/jade.py b/hwilib/devices/jade.py index b1ed9f890..466b51216 100644 --- a/hwilib/devices/jade.py +++ b/hwilib/devices/jade.py @@ -17,6 +17,7 @@ List, Optional, Sequence, + Set, Tuple, Union ) @@ -374,16 +375,21 @@ def _split_at_last_hardened_element(path: Sequence[int]) -> Tuple[Sequence[int], # Sign tx PSBT - newer Jade firmware supports native PSBT signing, but old firmwares require # mapping to the legacy 'sign_tx' structures. @jade_exception - def sign_tx(self, tx: PSBT) -> PSBT: + def sign_tx( + self, + psbt: PSBT, + registered_descriptors: Optional[Set[RegisteredDescriptor]] = None, + ) -> PSBT: """ Sign a transaction with the Blockstream Jade. """ + # registered_descriptors are intentionally unused because the descriptors were stored by register_descriptor. # Old firmware does not have native PSBT handling - use legacy method if self.PSBT_SUPPORTED_FW_VERSION > self.fw_version.finalize_version(): - return self.legacy_sign_tx(tx) + return self.legacy_sign_tx(psbt) # Firmware 0.1.47 (March 2023) and later support native PSBT signing - psbt_b64 = tx.serialize() + psbt_b64 = psbt.serialize() psbt_bytes = base64.b64decode(psbt_b64.strip()) # NOTE: sign_psbt() does not use AE signatures, so sticks with default (rfc6979) diff --git a/hwilib/devices/keepkey.py b/hwilib/devices/keepkey.py index aefab5cc2..08f192d7c 100644 --- a/hwilib/devices/keepkey.py +++ b/hwilib/devices/keepkey.py @@ -15,6 +15,7 @@ handle_errors, UnavailableActionError, ) +from ..psbt import PSBT from .trezorlib import protobuf from .trezorlib.transport import ( hid, @@ -35,6 +36,7 @@ Dict, List, Optional, + Set, ) py_enumerate = enumerate # Need to use the enumerate built-in but there's another function already named that @@ -176,6 +178,15 @@ def can_sign_taproot(self) -> bool: """ return False + def sign_tx( + self, + psbt: PSBT, + registered_descriptors: Optional[Set[RegisteredDescriptor]] = None, + ) -> PSBT: + if registered_descriptors: + raise UnavailableActionError("The KeepKey does not support BIP388 policy signing") + return super().sign_tx(psbt, registered_descriptors) + def register_descriptor(self, name: str, descriptor: 'Descriptor') -> RegisteredDescriptor: """ The KeepKey does not support registering descriptors diff --git a/hwilib/devices/ledger.py b/hwilib/devices/ledger.py index 868c1c9c5..87f87505f 100644 --- a/hwilib/devices/ledger.py +++ b/hwilib/devices/ledger.py @@ -10,6 +10,7 @@ Dict, List, Optional, + Set, Tuple, Union, ) @@ -188,7 +189,11 @@ def get_pubkey_at_path(self, path: str) -> ExtendedKey: return ExtendedKey.deserialize(xpub_str) @ledger_exception - def sign_tx(self, tx: PSBT) -> PSBT: + def sign_tx( + self, + psbt: PSBT, + registered_descriptors: Optional[Set[RegisteredDescriptor]] = None, + ) -> PSBT: """ Sign a transaction with a Ledger device. Not all transactions can be signed by a Ledger. @@ -202,6 +207,8 @@ def sign_tx(self, tx: PSBT) -> PSBT: - Only keys derived with standard BIP 44, 49, 84, and 86 derivation paths are supported for single signature addresses. """ + if registered_descriptors and isinstance(self.client, LegacyClient): + raise UnavailableActionError("Legacy Ledger app does not support BIP388 policy signing") master_fp = self.get_master_fingerprint() def legacy_sign_tx() -> PSBT: @@ -209,24 +216,43 @@ def legacy_sign_tx() -> PSBT: if not isinstance(client, LegacyClient): client = LegacyClient(self.transport_client, self.chain) wallet = WalletPolicy("", "wpkh(@0/**)", [""]) - legacy_input_sigs = client.sign_psbt(tx, wallet, None) + legacy_input_sigs = client.sign_psbt(psbt, wallet, None) for idx, partial_sig in legacy_input_sigs: - psbt_in = tx.inputs[idx] + psbt_in = psbt.inputs[idx] psbt_in.partial_sigs[partial_sig.pubkey] = partial_sig.signature - return tx + return psbt if isinstance(self.client, LegacyClient): return legacy_sign_tx() # Make a deepcopy of this psbt. We will need to modify it to get signing to work, # which will affect the caller's detection for whether signing occured. - psbt2 = copy.deepcopy(tx) - if tx.version != 2: + psbt2 = copy.deepcopy(psbt) + if psbt.version != 2: psbt2.convert_to_v2() # Figure out which wallets are signing wallets: Dict[bytes, Tuple[int, AddressType, WalletPolicy, Optional[bytes]]] = {} + for registered_descriptor in sorted( + registered_descriptors or set(), + key=lambda registration: registration.serialize(), + ): + descriptor = registered_descriptor.descriptor + registered_wallet = WalletPolicy( + registered_descriptor.name, + descriptor.get_bip388_template(), + [p.get_bip388_key_info() for p in descriptor.get_pubkey_providers()], + ) + registered_addrtype = descriptor.get_address_type() + if registered_addrtype is None: + raise BadArgumentError("Registered descriptor does not have an address type") + wallets[registered_wallet.id] = ( + signing_priority[registered_addrtype], + registered_addrtype, + registered_wallet, + registered_descriptor.registration, + ) pubkeys: Dict[int, bytes] = {} for input_num, psbt_in in builtins.enumerate(psbt2.inputs): utxo = None @@ -374,13 +400,13 @@ def process_origin(origin: KeyOriginInfo) -> None: psbt_in.partial_sigs[yielded.pubkey] = yielded.signature # Extract the sigs from psbt2 and put them into tx - for sig_in, psbt_in in zip(psbt2.inputs, tx.inputs): + for sig_in, psbt_in in zip(psbt2.inputs, psbt.inputs): psbt_in.partial_sigs.update(sig_in.partial_sigs) psbt_in.tap_script_sigs.update(sig_in.tap_script_sigs) if len(sig_in.tap_key_sig) != 0 and len(psbt_in.tap_key_sig) == 0: psbt_in.tap_key_sig = sig_in.tap_key_sig - return tx + return psbt @ledger_exception def sign_message(self, message: Union[str, bytes], keypath: str) -> str: diff --git a/hwilib/devices/trezor.py b/hwilib/devices/trezor.py index 89e1b58ce..fe175e4a1 100644 --- a/hwilib/devices/trezor.py +++ b/hwilib/devices/trezor.py @@ -359,7 +359,11 @@ def get_pubkey_at_path(self, path: str) -> ExtendedKey: return xpub @trezor_exception - def sign_tx(self, tx: PSBT) -> PSBT: + def sign_tx( + self, + psbt: PSBT, + registered_descriptors: Optional[Set[RegisteredDescriptor]] = None, + ) -> PSBT: """ Sign a transaction with the Trezor. There are some limitations to what transactions can be signed. @@ -368,6 +372,8 @@ def sign_tx(self, tx: PSBT) -> PSBT: - Send-to-self transactions will result in no prompt for outputs as all outputs will be detected as change. - Transactions containing Taproot inputs cannot have external inputs. """ + if registered_descriptors: + raise UnavailableActionError("The Trezor does not support BIP388 policy signing") self._check_unlocked() # Get this devices master key fingerprint @@ -382,7 +388,7 @@ def sign_tx(self, tx: PSBT) -> PSBT: # Prepare inputs inputs = [] to_ignore = [] # Note down which inputs whose signatures we're going to ignore - for input_num, psbt_in in builtins.enumerate(tx.inputs): + for input_num, psbt_in in builtins.enumerate(psbt.inputs): assert psbt_in.prev_txid is not None assert psbt_in.prev_out is not None assert psbt_in.sequence is not None @@ -447,7 +453,7 @@ def ignore_input() -> None: to_ignore.append(input_num) # Check for multisig - is_ms, multisig = parse_multisig(scriptcode, tx.xpub, psbt_in) + is_ms, multisig = parse_multisig(scriptcode, psbt.xpub, psbt_in) if is_ms: # Add to txinputtype txinputtype.multisig = multisig @@ -533,7 +539,7 @@ def ignore_input() -> None: # prepare outputs outputs = [] - for psbt_out in tx.outputs: + for psbt_out in psbt.outputs: out = psbt_out.get_txout() txoutput = messages.TxOutputType(amount=out.nValue) txoutput.script_type = messages.OutputScriptType.PAYTOADDRESS @@ -582,7 +588,7 @@ def ignore_input() -> None: if psbt_out.witness_script or psbt_out.redeem_script: is_ms, multisig = parse_multisig( psbt_out.witness_script or psbt_out.redeem_script, - tx.xpub, psbt_out) + psbt.xpub, psbt_out) if is_ms: txoutput.multisig = multisig if not wit: @@ -593,7 +599,7 @@ def ignore_input() -> None: # Prepare prev txs prevtxs = {} - for psbt_in in tx.inputs: + for psbt_in in psbt.inputs: if psbt_in.non_witness_utxo: prev = psbt_in.non_witness_utxo @@ -622,20 +628,20 @@ def ignore_input() -> None: prevtxs[ser_uint256(psbt_in.non_witness_utxo.sha256)[::-1]] = t # Sign the transaction - assert tx.tx_version is not None + assert psbt.tx_version is not None signed_tx = btc.sign_tx( client=self.client, coin_name=self.coin_name, inputs=inputs, outputs=outputs, prev_txes=prevtxs, - version=tx.tx_version, - lock_time=tx.compute_lock_time(), + version=psbt.tx_version, + lock_time=psbt.compute_lock_time(), serialize=False, ) # Each input has one signature - for input_num, (psbt_in, sig) in py_enumerate(list(zip(tx.inputs, signed_tx[0]))): + for input_num, (psbt_in, sig) in py_enumerate(list(zip(psbt.inputs, signed_tx[0]))): if input_num in to_ignore: continue for pubkey in psbt_in.hd_keypaths.keys(): @@ -650,7 +656,7 @@ def ignore_input() -> None: p += 1 - return tx + return psbt @trezor_exception def sign_message(self, message: Union[str, bytes], keypath: str) -> str: diff --git a/hwilib/hwwclient.py b/hwilib/hwwclient.py index ab08b20c2..32672327c 100644 --- a/hwilib/hwwclient.py +++ b/hwilib/hwwclient.py @@ -8,6 +8,7 @@ from typing import ( Dict, Optional, + Set, Union, ) from .descriptor import ( @@ -82,11 +83,16 @@ def get_pubkey_at_path(self, bip32_path: str) -> ExtendedKey: raise NotImplementedError("The HardwareWalletClient base class " "does not implement this method") - def sign_tx(self, psbt: PSBT) -> PSBT: + def sign_tx( + self, + psbt: PSBT, + registered_descriptors: Optional[Set[RegisteredDescriptor]] = None, + ) -> PSBT: """ Sign a partially signed bitcoin transaction (PSBT). :param psbt: The PSBT to sign + :param registered_descriptors: The registered BIP388 descriptor policies to sign with :return: The PSBT after being processed by the hardware wallet """ raise NotImplementedError("The HardwareWalletClient base class " diff --git a/hwilib/psbt.py b/hwilib/psbt.py index 429cf355d..2e9c57093 100644 --- a/hwilib/psbt.py +++ b/hwilib/psbt.py @@ -164,6 +164,40 @@ def set_null(self) -> None: self.musig2_partial_sigs.clear() self.unknown.clear() + def has_fingerprint(self, fingerprint: bytes) -> bool: + """ + Return whether this input contains a key with the specified fingerprint. + """ + return any( + origin.fingerprint == fingerprint + for origin in self.hd_keypaths.values() + ) or any( + origin.fingerprint == fingerprint + for _, origin in self.tap_bip32_paths.values() + ) + + def has_signature(self, fingerprint: bytes) -> bool: + """ + Return whether a key with the specified fingerprint has a signature. + """ + for pubkey, origin in self.hd_keypaths.items(): + if ( + origin.fingerprint == fingerprint + and pubkey in self.partial_sigs + ): + return True + for pubkey, (leaf_hashes, origin) in self.tap_bip32_paths.items(): + if origin.fingerprint != fingerprint: + continue + if not leaf_hashes and self.tap_key_sig: + return True + if any( + (pubkey, leaf_hash) in self.tap_script_sigs + for leaf_hash in leaf_hashes + ): + return True + return False + def deserialize(self, f: Readable) -> None: """ Deserialize a serialized PSBT input. diff --git a/test/data/speculos-automation.json b/test/data/speculos-automation.json index 4ee4cfefb..c228efa54 100644 --- a/test/data/speculos-automation.json +++ b/test/data/speculos-automation.json @@ -55,6 +55,13 @@ [ "button", 2, false ] ] }, + { + "regexp": "^To \\([0-9]+/[0-9]+\\)$", + "actions": [ + [ "button", 2, true ], + [ "button", 2, false ] + ] + }, { "regexp": "^To.*", "y": 3, @@ -63,6 +70,13 @@ [ "button", 2, false ] ] }, + { + "regexp": "^From$", + "actions": [ + [ "button", 2, true ], + [ "button", 2, false ] + ] + }, { "regexp": "^(Message)$", "actions": [ @@ -71,7 +85,7 @@ ] }, { - "regexp": "^(Accept|Approve|Continue|Sign message|Sign transaction|Register account).*", + "regexp": "^(Accept|Approve|Continue|Sign message|Sign transaction|Register account|Account registered|Address verified|Transaction signed).*", "actions": [ [ "button", 3, true ], [ "button", 3, false ] diff --git a/test/test_bitbox02.py b/test/test_bitbox02.py index 5085c34f4..6dae9a907 100644 --- a/test/test_bitbox02.py +++ b/test/test_bitbox02.py @@ -129,7 +129,15 @@ def bitbox02_test_suite(simulator, bitcoind, interface): # TestSignMessage is removed, since its only testcase is for legacy p2pkh, which is not supported by BitBox02 # suite.addTest(DeviceTestCase.parameterize(TestSignMessage, bitcoind, emulator=dev_emulator, interface=interface)) suite.addTest(DeviceTestCase.parameterize(TestBitbox02GetXpub, bitcoind, emulator=dev_emulator, interface=interface)) - suite.addTest(DeviceTestCase.parameterize(TestRegisterDescriptor, bitcoind, emulator=dev_emulator, interface=interface, returns_registration=False, sorted=False)) + suite.addTest(DeviceTestCase.parameterize( + TestRegisterDescriptor, + bitcoind, + emulator=dev_emulator, + interface=interface, + returns_registration=False, + sorted=False, + supports_multiple_policies=False, + )) result = unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite) return result.wasSuccessful() diff --git a/test/test_coldcard.py b/test/test_coldcard.py index 2c2adf064..db978e41e 100755 --- a/test/test_coldcard.py +++ b/test/test_coldcard.py @@ -200,7 +200,14 @@ def coldcard_test_suite(simulator, bitcoind, interface, is_edge=False): suite.addTest(DeviceTestCase.parameterize(TestDisplayAddress, bitcoind, emulator=dev_emulator, interface=interface)) suite.addTest(DeviceTestCase.parameterize(TestSignMessage, bitcoind, emulator=dev_emulator, interface=interface)) suite.addTest(DeviceTestCase.parameterize(TestSignTx, bitcoind, emulator=dev_emulator, interface=interface, signtx_cases=signtx_cases)) - suite.addTest(DeviceTestCase.parameterize(TestRegisterDescriptor, bitcoind, emulator=dev_emulator, interface=interface, returns_registration=False)) + suite.addTest(DeviceTestCase.parameterize( + TestRegisterDescriptor, + bitcoind, + emulator=dev_emulator, + interface=interface, + returns_registration=False, + supports_multiple_policies=is_edge, + )) if is_edge: suite.addTest(DeviceTestCase.parameterize(TestColdcardEdgeDisplayAddress, bitcoind, emulator=dev_emulator, interface=interface)) diff --git a/test/test_descriptor.py b/test/test_descriptor.py index 4755b815b..dca4fbe80 100755 --- a/test/test_descriptor.py +++ b/test/test_descriptor.py @@ -9,10 +9,9 @@ WPKHDescriptor, WSHDescriptor, ) +from hwilib.common import AddressType from hwilib.errors import InvalidPolicyError -from binascii import unhexlify - import unittest class TestDescriptor(unittest.TestCase): @@ -31,6 +30,23 @@ def test_derive(self): ) self.assertEqual(descriptor.to_string_no_checksum(), descriptor_str) + def test_get_address_type(self): + key = "02c97dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c7" + descriptors = { + f"pk({key})": None, + f"multi(1,{key})": None, + f"pkh({key})": AddressType.LEGACY, + f"sh(multi(1,{key}))": AddressType.LEGACY, + f"wpkh({key})": AddressType.WIT, + f"wsh(multi(1,{key}))": AddressType.WIT, + f"sh(wpkh({key}))": AddressType.SH_WIT, + f"sh(wsh(multi(1,{key})))": AddressType.SH_WIT, + f"tr({key})": AddressType.TAP, + } + for descriptor, address_type in descriptors.items(): + with self.subTest(descriptor=descriptor): + self.assertEqual(parse_descriptor(descriptor).get_address_type(), address_type) + def test_parse_descriptor_with_origin(self): d = "wpkh([00000001/84h/1h/0h]tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/0)" desc = parse_descriptor(d) @@ -41,10 +57,6 @@ def test_parse_descriptor_with_origin(self): self.assertEqual(desc.pubkeys[0].deriv_path, [[0], [0]]) self.assertEqual(desc.pubkeys[0].expr_index, 0) self.assertEqual(desc.to_string_no_checksum(), d) - e = desc.expand(0) - self.assertEqual(e.output_script, unhexlify("0014d95fc47eada9e4c3cf59a2cbf9e96517c3ba2efa")) - self.assertEqual(e.redeem_script, None) - self.assertEqual(e.witness_script, None) def test_parse_multisig_descriptor_with_origin(self): d = "wsh(multi(2,[00000001/48h/0h/0h/2h]tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/0,[00000002/48h/0h/0h/2h]tpubDFHiBJDeNvqPWNJbzzxqDVXmJZoNn2GEtoVcFhMjXipQiorGUmps3e5ieDGbRrBPTFTh9TXEKJCwbAGW9uZnfrVPbMxxbFohuFzfT6VThty/0/0))" @@ -63,10 +75,6 @@ def test_parse_multisig_descriptor_with_origin(self): self.assertEqual(desc.subdescriptors[0].pubkeys[1].deriv_path, [[0], [0]]) self.assertEqual(desc.subdescriptors[0].pubkeys[1].expr_index, 1) self.assertEqual(desc.to_string_no_checksum(), d) - e = desc.expand(0) - self.assertEqual(e.output_script, unhexlify("002084b64b2b8651df8fd3e9735f6269edbf9e03abf619ae0788be9f17bf18e83d59")) - self.assertEqual(e.redeem_script, None) - self.assertEqual(e.witness_script, unhexlify("522102c97dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c721033a4f18d2b498273ed7439c59f6d8a673d5b9c67a03163d530e12c941ca22be3352ae")) d = "sh(multi(2,[00000001/48h/0h/0h/2h]tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/0,[00000002/48h/0h/0h/2h]tpubDFHiBJDeNvqPWNJbzzxqDVXmJZoNn2GEtoVcFhMjXipQiorGUmps3e5ieDGbRrBPTFTh9TXEKJCwbAGW9uZnfrVPbMxxbFohuFzfT6VThty/0/0))" desc = parse_descriptor(d) @@ -84,10 +92,6 @@ def test_parse_multisig_descriptor_with_origin(self): self.assertEqual(desc.subdescriptors[0].pubkeys[1].deriv_path, [[0], [0]]) self.assertEqual(desc.subdescriptors[0].pubkeys[1].expr_index, 1) self.assertEqual(desc.to_string_no_checksum(), d) - e = desc.expand(0) - self.assertEqual(e.output_script, unhexlify("a91495ee6326805b1586bb821fc3c0eeab2c68441b4187")) - self.assertEqual(e.redeem_script, unhexlify("522102c97dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c721033a4f18d2b498273ed7439c59f6d8a673d5b9c67a03163d530e12c941ca22be3352ae")) - self.assertEqual(e.witness_script, None) d = "sh(wsh(multi(2,[00000001/48h/0h/0h/2h]tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/0,[00000002/48h/0h/0h/2h]tpubDFHiBJDeNvqPWNJbzzxqDVXmJZoNn2GEtoVcFhMjXipQiorGUmps3e5ieDGbRrBPTFTh9TXEKJCwbAGW9uZnfrVPbMxxbFohuFzfT6VThty/0/0)))" desc = parse_descriptor(d) @@ -106,10 +110,6 @@ def test_parse_multisig_descriptor_with_origin(self): self.assertEqual(desc.subdescriptors[0].subdescriptors[0].pubkeys[1].deriv_path, [[0], [0]]) self.assertEqual(desc.subdescriptors[0].subdescriptors[0].pubkeys[1].expr_index, 1) self.assertEqual(desc.to_string_no_checksum(), d) - e = desc.expand(0) - self.assertEqual(e.output_script, unhexlify("a914779ae0f6958e98b997cc177f9b554289905fbb5587")) - self.assertEqual(e.redeem_script, unhexlify("002084b64b2b8651df8fd3e9735f6269edbf9e03abf619ae0788be9f17bf18e83d59")) - self.assertEqual(e.witness_script, unhexlify("522102c97dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c721033a4f18d2b498273ed7439c59f6d8a673d5b9c67a03163d530e12c941ca22be3352ae")) def test_parse_descriptor_without_origin(self): d = "wpkh(tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/0)" @@ -119,10 +119,6 @@ def test_parse_descriptor_without_origin(self): self.assertEqual(desc.pubkeys[0].pubkey, "tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B") self.assertEqual(desc.pubkeys[0].deriv_path, [[0], [0]]) self.assertEqual(desc.to_string_no_checksum(), d) - e = desc.expand(0) - self.assertEqual(e.output_script, unhexlify("0014d95fc47eada9e4c3cf59a2cbf9e96517c3ba2efa")) - self.assertEqual(e.redeem_script, None) - self.assertEqual(e.witness_script, None) def test_parse_descriptor_with_origin_fingerprint_only(self): d = "wpkh([00000001]tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/0)" @@ -134,10 +130,6 @@ def test_parse_descriptor_with_origin_fingerprint_only(self): self.assertEqual(desc.pubkeys[0].deriv_path, [[0], [0]]) self.assertEqual(desc.pubkeys[0].expr_index, 0) self.assertEqual(desc.to_string_no_checksum(), d) - e = desc.expand(0) - self.assertEqual(e.output_script, unhexlify("0014d95fc47eada9e4c3cf59a2cbf9e96517c3ba2efa")) - self.assertEqual(e.redeem_script, None) - self.assertEqual(e.witness_script, None) def test_parse_descriptor_with_key_at_end_with_origin(self): d = "wpkh([00000001/84h/1h/0h/0/0]02c97dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c7)" @@ -149,10 +141,6 @@ def test_parse_descriptor_with_key_at_end_with_origin(self): self.assertEqual(desc.pubkeys[0].deriv_path, None) self.assertEqual(desc.pubkeys[0].expr_index, 0) self.assertEqual(desc.to_string_no_checksum(), d) - e = desc.expand(0) - self.assertEqual(e.output_script, unhexlify("0014d95fc47eada9e4c3cf59a2cbf9e96517c3ba2efa")) - self.assertEqual(e.redeem_script, None) - self.assertEqual(e.witness_script, None) d = "pkh([00000001/84h/1h/0h/0/0]02c97dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c7)" desc = parse_descriptor(d) @@ -163,10 +151,6 @@ def test_parse_descriptor_with_key_at_end_with_origin(self): self.assertEqual(desc.pubkeys[0].deriv_path, None) self.assertEqual(desc.pubkeys[0].expr_index, 0) self.assertEqual(desc.to_string_no_checksum(), d) - e = desc.expand(0) - self.assertEqual(e.output_script, unhexlify("76a914d95fc47eada9e4c3cf59a2cbf9e96517c3ba2efa88ac")) - self.assertEqual(e.redeem_script, None) - self.assertEqual(e.witness_script, None) def test_parse_descriptor_with_key_at_end_without_origin(self): d = "wpkh(02c97dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c7)" diff --git a/test/test_device.py b/test/test_device.py index 421054dbd..4c09c548a 100644 --- a/test/test_device.py +++ b/test/test_device.py @@ -198,6 +198,29 @@ def setup_wallets(self): self.wrpc = self.bitcoind.get_wallet_rpc(wallet_name) self.wpk_rpc = self.bitcoind.get_wallet_rpc("supply") + def sign_and_finalize(self, psbt: str, *signtx_args: str) -> Dict: + """Sign a PSBT with the device and assert that it finalizes.""" + sign_res = self.do_command( + self.dev_args + ["signtx", *signtx_args, psbt] + ) + self.assertNotIn("error", sign_res) + self.assertTrue(sign_res["signed"]) + + finalize_res = self.wrpc.finalizepsbt(sign_res["psbt"]) + self.assertTrue(finalize_res["complete"]) + return finalize_res + + def _set_global_xpubs( + self, + psbt: str, + xpubs: Dict[bytes, KeyOriginInfo], + ) -> str: + """Replace a PSBT's global xpub map.""" + psbt_obj = PSBT() + psbt_obj.deserialize(psbt) + psbt_obj.xpub = xpubs + return psbt_obj.serialize() + def setUp(self): self.emulator.start() @@ -372,10 +395,7 @@ def _generate_and_finalize(self, unknown_inputs, psbt): if not unknown_inputs: # Just do the normal signing process to test "all inputs" case - sign_res = self.do_command(self.dev_args + ['signtx', psbt]) - finalize_res = self.wrpc.finalizepsbt(sign_res['psbt']) - self.assertTrue(sign_res["signed"]) - self.assertTrue(finalize_res["complete"]) + finalize_res = self.sign_and_finalize(psbt) else: # Sign only input one on first pass # then rest on second pass to test ability to successfully @@ -576,10 +596,7 @@ def _test_signtx(self, input_types, multisig_types, external, op_return: bool): )["psbt"] # We need to modify the psbt to include our xpubs as Core does not include xpubs - psbt_obj = PSBT() - psbt_obj.deserialize(psbt) - psbt_obj.xpub = xpubs - psbt = psbt_obj.serialize() + psbt = self._set_global_xpubs(psbt, xpubs) if external: # Sign with unknown inputs in two steps @@ -783,22 +800,41 @@ def test_bad_path(self): self.assertEqual(result['code'], -7) class TestRegisterDescriptor(DeviceTestCase): - def __init__(self, *args, returns_registration, sorted=True, **kwargs): + def __init__( + self, + *args, + returns_registration, + sorted=True, + supports_multiple_policies=True, + **kwargs, + ): super().__init__(*args, **kwargs) self.returns_registration = returns_registration self.sorted = sorted + self.supports_multiple_policies = supports_multiple_policies - def test_register_descriptor(self): - account_path = "m/48h/1h/0h/2h" + def setUp(self): + super().setUp() + self.setup_wallets() + + def _descriptor(self, account: int) -> str: + account_path = f"m/48h/1h/{account}h/2h" account_xpub = self.do_command(self.dev_args + ["getxpub", account_path])["xpub"] multi = "sortedmulti" if self.sorted else "multi" - descriptor = f"wsh({multi}(1,[{self.emulator.fingerprint}{account_path[1:]}]{account_xpub}/<0;1>/*,[1a0f5425{account_path[1:]}]tpubDF23ETNjCC283QmYZtJp26GqHkSa6Yw6vPqp3UkMsPCvBzRC4dMQzE1U3WwKsFsx3apUkQA4JHQDSmcC3N1yhE2gF1aKJA1CiVtNyA9Rv4H/<0;1>/*))" # noqa: E702 - desc_name = "HWI_testing" - result = self.do_command(self.dev_args + ["registerdescriptor", desc_name, descriptor]) + cosigner_origin = "[1a0f5425/48h/1h/0h/2h]" + return f"wsh({multi}(1,[{self.emulator.fingerprint}{account_path[1:]}]{account_xpub}/<0;1>/*,{cosigner_origin}tpubDF23ETNjCC283QmYZtJp26GqHkSa6Yw6vPqp3UkMsPCvBzRC4dMQzE1U3WwKsFsx3apUkQA4JHQDSmcC3N1yhE2gF1aKJA1CiVtNyA9Rv4H/<0;1>/*))" # noqa: E702 + + def _register_descriptor(self, name: str, account: int) -> tuple[str, str]: + descriptor = self._descriptor(account) + result = self.do_command(self.dev_args + ["registerdescriptor", name, descriptor]) self.assertNotIn("error", result) self.assertIn("registration", result) - reg_str = result["registration"] + return descriptor, result["registration"] + + def test_register_descriptor(self): + descriptor, reg_str = self._register_descriptor("HWI_testing", 0) + desc_name = "HWI_testing" reg = RegisteredDescriptor.deserialize(reg_str) self.assertEqual(reg.name, desc_name) self.assertEqual(reg.descriptor.to_string_no_checksum(), descriptor) @@ -817,7 +853,7 @@ def test_register_descriptor(self): "displayaddress", "--index", str(address_index), "--multipath-index", str(multipath_index), - "--registration", result["registration"], + "--registration", reg_str, ]) self.assertNotIn("error", result) self.assertNotIn("code", result) @@ -828,3 +864,103 @@ def test_register_descriptor(self): bech32.decode("bcrt", expected_address), bech32.decode("tb", result["address"]), ) + import_result = self.wrpc.importdescriptors([{ + "desc": AddChecksum(descriptor), + "timestamp": "now", + "active": True, + }]) + self.assertTrue(import_result[0]["success"]) + + self.wpk_rpc.sendtoaddress(self.wrpc.getnewaddress(), 1) + self.wpk_rpc.generatetoaddress(6, self.wpk_rpc.getnewaddress()) + psbt = self.wrpc.walletcreatefundedpsbt( + [], + [{self.wpk_rpc.getnewaddress(): 0.5}], + 0, + {}, + True, + )["psbt"] + for include_xpubs in (False, True): + with self.subTest(include_xpubs=include_xpubs): + xpubs = {} + if include_xpubs: + for provider in reg.descriptor.get_pubkey_providers(): + assert provider.extkey is not None + assert provider.origin is not None + xpubs[provider.extkey.serialize()] = provider.origin + + self.sign_and_finalize( + self._set_global_xpubs(psbt, xpubs), + "--registration", + reg_str, + ) + + def test_sign_multiple_registered_descriptors(self): + if not self.supports_multiple_policies: + self.skipTest("Device only supports one registered policy per signing call") + + descriptors = [] + registrations = [] + for account in range(2): + descriptor, registration = self._register_descriptor(f"HWI_testing_{account}", account) + descriptors.append(descriptor) + registrations.append(registration) + + imports = [{ + "desc": AddChecksum(descriptor), + "timestamp": "now", + "range": [0, 0], + } for descriptor in descriptors] + import_result = self.wrpc.importdescriptors(imports) + self.assertTrue(all(result["success"] for result in import_result)) + + addresses = [ + self.rpc.deriveaddresses(AddChecksum(descriptor), [0, 0])[0][0] + for descriptor in descriptors + ] + + device_descriptors = self.do_command(self.dev_args + ["getdescriptors"]) + inferred_imports = [] + for key, internal in (("receive", False), ("internal", True)): + descriptor = next( + descriptor + for descriptor in device_descriptors[key] + if descriptor.startswith("wpkh(") + ) + inferred_imports.append({ + "desc": descriptor, + "timestamp": "now", + "range": [0, 0], + "active": True, + "internal": internal, + }) + inferred_result = self.wrpc.importdescriptors(inferred_imports) + self.assertTrue(all(result["success"] for result in inferred_result)) + addresses.append(self.wrpc.getnewaddress("", "bech32")) + + funding_txids = { + self.wpk_rpc.sendtoaddress(address, 1) + for address in addresses + } + self.wpk_rpc.generatetoaddress(6, self.wpk_rpc.getnewaddress()) + + inputs = [ + {"txid": utxo["txid"], "vout": utxo["vout"]} + for utxo in self.wrpc.listunspent(0, 9999999, addresses) + if utxo["txid"] in funding_txids + ] + self.assertEqual(len(inputs), 3) + psbt = self.wrpc.walletcreatefundedpsbt( + inputs, + [{self.wpk_rpc.getnewaddress(): 3}], + 0, + {"subtractFeeFromOutputs": [0]}, + True, + )["psbt"] + self.sign_and_finalize( + self._set_global_xpubs(psbt, {}), + "--registration", + registrations[0], + "--registration", + registrations[1], + ) diff --git a/test/test_psbt.py b/test/test_psbt.py index aecce121e..4d172b65a 100755 --- a/test/test_psbt.py +++ b/test/test_psbt.py @@ -65,5 +65,45 @@ def test_convert_to_v0(self): self.assertEqual(psbt.tx.vin[0].nSequence, 0xffffffff) + def test_has_fingerprint(self): + cases = [ + ("BIP32", 7, {"19542eb0": True, "00000001": False}), + ("Taproot BIP32", 22, {"7c461e5d": True, "00000001": False}), + ] + for name, vector, fingerprints in cases: + with self.subTest(name=name): + psbt = PSBT() + psbt.deserialize(self.data["valid"][vector]) + for fingerprint, expected in fingerprints.items(): + self.assertEqual( + psbt.inputs[0].has_fingerprint(bytes.fromhex(fingerprint)), + expected, + ) + + def test_has_signature(self): + cases = [ + ("partial signature", 4, {"b4a6ba67": True}), + ( + "partial signature for another fingerprint", + 7, + {"19542eb0": True, "e81a5744": False}, + ), + ("Taproot key path", 9, {"772b2da7": True}), + ( + "Taproot script path", + 22, + {"2680dd6e": True, "580b0887": False}, + ), + ] + for name, vector, fingerprints in cases: + with self.subTest(name=name): + psbt = PSBT() + psbt.deserialize(self.data["valid"][vector]) + for fingerprint, expected in fingerprints.items(): + self.assertEqual( + psbt.inputs[0].has_signature(bytes.fromhex(fingerprint)), + expected, + ) + if __name__ == "__main__": unittest.main()