#!/usr/bin/env python3 # Copyright (c) 2021 The Bitcoin Core developers # Distributed under the MIT software license, see the accompanying # file COPYING or http://www.opensource.org/licenses/mit-license.php. """Test a basic M-of-N multisig setup between multiple people using descriptor wallets and PSBTs, as well as a signing flow. This is meant to be documentation as much as functional tests, so it is kept as simple and readable as possible. """ from test_framework.address import base58_to_byte from test_framework.test_framework import BitcoinTestFramework from test_framework.util import ( assert_approx, assert_equal, ) class WalletMultisigDescriptorPSBTTest(BitcoinTestFramework): def set_test_params(self): self.num_nodes = 3 self.setup_clean_chain = True self.wallet_names = [] self.extra_args = [["-keypool=100"]] * self.num_nodes def skip_test_if_missing_module(self): self.skip_if_no_wallet() self.skip_if_no_sqlite() @staticmethod def _get_xpub(wallet): """Extract the wallet's xpubs using `listdescriptors` and pick the one from the `pkh` descriptor since it's least likely to be accidentally reused (legacy addresses).""" descriptor = next(filter(lambda d: d["desc"].startswith("pkh"), wallet.listdescriptors()["descriptors"])) return descriptor["desc"].split("]")[-1].split("/")[0] @staticmethod def _check_psbt(psbt, to, value, multisig): """Helper function for any of the N participants to check the psbt with decodepsbt and verify it is OK before signing.""" tx = multisig.decodepsbt(psbt)["tx"] amount = 0 for vout in tx["vout"]: address = vout["scriptPubKey"]["address"] assert_equal(multisig.getaddressinfo(address)["ischange"], address != to) if address == to: amount += vout["value"] assert_approx(amount, float(value), vspan=0.001) def participants_create_multisigs(self, xpubs): """The multisig is created by importing the following descriptors. The resulting wallet is watch-only and every participant can do this.""" # some simple validation assert_equal(len(xpubs), self.N) # a sanity-check/assertion, this will throw if the base58 checksum of any of the provided xpubs are invalid for xpub in xpubs: base58_to_byte(xpub) for i, node in enumerate(self.nodes): node.createwallet(wallet_name=f"{self.name}_{i}", blank=True, descriptors=True, disable_private_keys=True) multisig = node.get_wallet_rpc(f"{self.name}_{i}") external = multisig.getdescriptorinfo(f"wsh(sortedmulti({self.M},{f'/0/*,'.join(xpubs)}/0/*))") internal = multisig.getdescriptorinfo(f"wsh(sortedmulti({self.M},{f'/1/*,'.join(xpubs)}/1/*))") result = multisig.importdescriptors([ { # receiving addresses (internal: False) "desc": external["descriptor"], "active": True, "internal": False, "timestamp": "now", }, { # change addresses (internal: True) "desc": internal["descriptor"], "active": True, "internal": True, "timestamp": "now", }, ]) assert all(r["success"] for r in result) yield multisig def run_test(self): self.M = 2 self.N = self.num_nodes self.name = f"{self.M}_of_{self.N}_multisig" self.log.info(f"Testing {self.name}...") participants = { # Every participant generates an xpub. The most straightforward way is to create a new descriptor wallet. # This wallet will be the participant's `signer` for the resulting multisig. Avoid reusing this wallet for any other purpose (for privacy reasons). "signers": [node.get_wallet_rpc(node.createwallet(wallet_name=f"participant_{self.nodes.index(node)}", descriptors=True)["name"]) for node in self.nodes], # After participants generate and exchange their xpubs they will each create their own watch-only multisig. # Note: these multisigs are all the same, this just highlights that each participant can independently verify everything on their own node. "multisigs": [] } self.log.info("Generate and exchange xpubs...") xpubs = [self._get_xpub(signer) for signer in participants["signers"]] self.log.info("Every participant imports the following descriptors to create the watch-only multisig...") participants["multisigs"] = list(self.participants_create_multisigs(xpubs)) self.log.info("Check that every participant's multisig generates the same addresses...") for _ in range(10): # we check that the first 10 generated addresses are the same for all participant's multisigs receive_addresses = [multisig.getnewaddress() for multisig in participants["multisigs"]] all(address == receive_addresses[0] for address in receive_addresses) change_addresses = [multisig.getrawchangeaddress() for multisig in participants["multisigs"]] all(address == change_addresses[0] for address in change_addresses) self.log.info("Get a mature utxo to send to the multisig...") coordinator_wallet = participants["signers"][0] self.generatetoaddress(self.nodes[0], 101, coordinator_wallet.getnewaddress()) deposit_amount = 6.15 multisig_receiving_address = participants["multisigs"][0].getnewaddress() self.log.info("Send funds to the resulting multisig receiving address...") coordinator_wallet.sendtoaddress(multisig_receiving_address, deposit_amount) self.generate(self.nodes[0], 1) for participant in participants["multisigs"]: assert_approx(participant.getbalance(), deposit_amount, vspan=0.001) self.log.info("Send a transaction from the multisig!") to = participants["signers"][self.N - 1].getnewaddress() value = 1 self.log.info("First, make a sending transaction, created using `walletcreatefundedpsbt` (anyone can initiate this)...") psbt = participants["multisigs"][0].walletcreatefundedpsbt(inputs=[], outputs={to: value}, options={"feeRate": 0.00010}) psbts = [] self.log.info("Now at least M users check the psbt with decodepsbt and (if OK) signs it with walletprocesspsbt...") for m in range(self.M): signers_multisig = participants["multisigs"][m] self._check_psbt(psbt["psbt"], to, value, signers_multisig) signing_wallet = participants["signers"][m] partially_signed_psbt = signing_wallet.walletprocesspsbt(psbt["psbt"]) psbts.append(partially_signed_psbt["psbt"]) self.log.info("Finally, collect the signed PSBTs with combinepsbt, finalizepsbt, then broadcast the resulting transaction...") combined = coordinator_wallet.combinepsbt(psbts) finalized = coordinator_wallet.finalizepsbt(combined) coordinator_wallet.sendrawtransaction(finalized["hex"]) self.log.info("Check that balances are correct after the transaction has been included in a block.") self.generate(self.nodes[0], 1) assert_approx(participants["multisigs"][0].getbalance(), deposit_amount - value, vspan=0.001) assert_equal(participants["signers"][self.N - 1].getbalance(), value) self.log.info("Send another transaction from the multisig, this time with a daisy chained signing flow (one after another in series)!") psbt = participants["multisigs"][0].walletcreatefundedpsbt(inputs=[], outputs={to: value}, options={"feeRate": 0.00010}) for m in range(self.M): signers_multisig = participants["multisigs"][m] self._check_psbt(psbt["psbt"], to, value, signers_multisig) signing_wallet = participants["signers"][m] psbt = signing_wallet.walletprocesspsbt(psbt["psbt"]) assert_equal(psbt["complete"], m == self.M - 1) finalized = coordinator_wallet.finalizepsbt(psbt["psbt"]) coordinator_wallet.sendrawtransaction(finalized["hex"]) self.log.info("Check that balances are correct after the transaction has been included in a block.") self.generate(self.nodes[0], 1) assert_approx(participants["multisigs"][0].getbalance(), deposit_amount - (value * 2), vspan=0.001) assert_equal(participants["signers"][self.N - 1].getbalance(), value * 2) if __name__ == "__main__": WalletMultisigDescriptorPSBTTest().main()