1  #!/usr/bin/env python3
  2  # Copyright (c) 2021-present The Bitcoin Core developers
  3  # Distributed under the MIT software license, see the accompanying
  4  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
  5  """Test a basic M-of-N multisig setup between multiple people using descriptor wallets and PSBTs, as well as a signing flow.
  6  
  7  This is meant to be documentation as much as functional tests, so it is kept as simple and readable as possible.
  8  """
  9  
 10  from test_framework.test_framework import BitcoinTestFramework
 11  from test_framework.util import (
 12      assert_approx,
 13      assert_equal,
 14  )
 15  
 16  
 17  class WalletMultisigDescriptorPSBTTest(BitcoinTestFramework):
 18      def set_test_params(self):
 19          self.num_nodes = 3
 20          self.setup_clean_chain = True
 21          self.wallet_names = []
 22          self.extra_args = [["-keypool=100"]] * self.num_nodes
 23  
 24      def skip_test_if_missing_module(self):
 25          self.skip_if_no_wallet()
 26  
 27      @staticmethod
 28      def _get_xpub(wallet):
 29          """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)."""
 30          pkh_descriptor = next(filter(lambda d: d["desc"].startswith("pkh(") and not d["internal"], wallet.listdescriptors()["descriptors"]))
 31          # Keep all key origin information (master key fingerprint and all derivation steps) for proper support of hardware devices
 32          # See section 'Key origin identification' in 'doc/descriptors.md' for more details...
 33          # Replace the change index with the multipath convention
 34          return pkh_descriptor["desc"].split("pkh(")[1].split(")")[0].replace("/0/*", "/<0;1>/*")
 35  
 36      @staticmethod
 37      def _check_psbt(psbt, to, value, multisig):
 38          """Helper function for any of the N participants to check the psbt with decodepsbt and verify it is OK before signing."""
 39          tx = multisig.decodepsbt(psbt)["tx"]
 40          amount = 0
 41          for vout in tx["vout"]:
 42              address = vout["scriptPubKey"]["address"]
 43              assert_equal(multisig.getaddressinfo(address)["ischange"], address != to)
 44              if address == to:
 45                  amount += vout["value"]
 46          assert_approx(amount, float(value), vspan=0.001)
 47  
 48      def participants_create_multisigs(self, xpubs):
 49          """The multisig is created by importing the following descriptors. The resulting wallet is watch-only and every participant can do this."""
 50          for i, node in enumerate(self.nodes):
 51              node.createwallet(wallet_name=f"{self.name}_{i}", blank=True, disable_private_keys=True)
 52              multisig = node.get_wallet_rpc(f"{self.name}_{i}")
 53              multisig_desc = f"wsh(sortedmulti({self.M},{','.join(xpubs)}))"
 54              checksum = multisig.getdescriptorinfo(multisig_desc)["checksum"]
 55              result = multisig.importdescriptors([
 56                  {  # Multipath descriptor expands to receive and change
 57                      "desc": f"{multisig_desc}#{checksum}",
 58                      "active": True,
 59                      "timestamp": "now",
 60                  }
 61              ])
 62              assert all(r["success"] for r in result)
 63              yield multisig
 64  
 65      def run_test(self):
 66          self.M = 2
 67          self.N = self.num_nodes
 68          self.name = f"{self.M}_of_{self.N}_multisig"
 69          self.log.info(f"Testing {self.name}...")
 70  
 71          participants = {
 72              # Every participant generates an xpub. The most straightforward way is to create a new descriptor wallet.
 73              # This wallet will be the participant's `signer` for the resulting multisig. Avoid reusing this wallet for any other purpose (for privacy reasons).
 74              "signers": [node.get_wallet_rpc(node.createwallet(wallet_name=f"participant_{self.nodes.index(node)}")["name"]) for node in self.nodes],
 75              # After participants generate and exchange their xpubs they will each create their own watch-only multisig.
 76              # Note: these multisigs are all the same, this just highlights that each participant can independently verify everything on their own node.
 77              "multisigs": []
 78          }
 79  
 80          self.log.info("Generate and exchange xpubs...")
 81          xpubs = [self._get_xpub(signer) for signer in participants["signers"]]
 82  
 83          self.log.info("Every participant imports the following descriptors to create the watch-only multisig...")
 84          participants["multisigs"] = list(self.participants_create_multisigs(xpubs))
 85  
 86          self.log.info("Check that every participant's multisig generates the same addresses...")
 87          for _ in range(10):  # we check that the first 10 generated addresses are the same for all participant's multisigs
 88              receive_addresses = [multisig.getnewaddress() for multisig in participants["multisigs"]]
 89              assert all(address == receive_addresses[0] for address in receive_addresses)
 90              change_addresses = [multisig.getrawchangeaddress() for multisig in participants["multisigs"]]
 91              assert all(address == change_addresses[0] for address in change_addresses)
 92  
 93          self.log.info("Get a mature utxo to send to the multisig...")
 94          coordinator_wallet = participants["signers"][0]
 95          self.generatetoaddress(self.nodes[0], 101, coordinator_wallet.getnewaddress())
 96  
 97          deposit_amount = 6.15
 98          multisig_receiving_address = participants["multisigs"][0].getnewaddress()
 99          self.log.info("Send funds to the resulting multisig receiving address...")
100          coordinator_wallet.sendtoaddress(multisig_receiving_address, deposit_amount)
101          self.generate(self.nodes[0], 1)
102          for participant in participants["multisigs"]:
103              assert_approx(participant.getbalance(), deposit_amount, vspan=0.001)
104  
105          self.log.info("Send a transaction from the multisig!")
106          to = participants["signers"][self.N - 1].getnewaddress()
107          value = 1
108          self.log.info("First, make a sending transaction, created using `walletcreatefundedpsbt` (anyone can initiate this)...")
109          psbt = participants["multisigs"][0].walletcreatefundedpsbt(inputs=[], outputs={to: value}, feeRate=0.00010)
110  
111          psbts = []
112          self.log.info("Now at least M users check the psbt with decodepsbt and (if OK) signs it with walletprocesspsbt...")
113          for m in range(self.M):
114              signers_multisig = participants["multisigs"][m]
115              self._check_psbt(psbt["psbt"], to, value, signers_multisig)
116              signing_wallet = participants["signers"][m]
117              partially_signed_psbt = signing_wallet.walletprocesspsbt(psbt["psbt"])
118              psbts.append(partially_signed_psbt["psbt"])
119  
120          self.log.info("Finally, collect the signed PSBTs with combinepsbt, finalizepsbt, then broadcast the resulting transaction...")
121          combined = coordinator_wallet.combinepsbt(psbts)
122          finalized = coordinator_wallet.finalizepsbt(combined)
123          coordinator_wallet.sendrawtransaction(finalized["hex"])
124  
125          self.log.info("Check that balances are correct after the transaction has been included in a block.")
126          self.generate(self.nodes[0], 1)
127          assert_approx(participants["multisigs"][0].getbalance(), deposit_amount - value, vspan=0.001)
128          assert_equal(participants["signers"][self.N - 1].getbalance(), value)
129  
130          self.log.info("Send another transaction from the multisig, this time with a daisy chained signing flow (one after another in series)!")
131          psbt = participants["multisigs"][0].walletcreatefundedpsbt(inputs=[], outputs={to: value}, feeRate=0.00010)
132          for m in range(self.M):
133              signers_multisig = participants["multisigs"][m]
134              self._check_psbt(psbt["psbt"], to, value, signers_multisig)
135              signing_wallet = participants["signers"][m]
136              psbt = signing_wallet.walletprocesspsbt(psbt["psbt"])
137              assert_equal(psbt["complete"], m == self.M - 1)
138          coordinator_wallet.sendrawtransaction(psbt["hex"])
139  
140          self.log.info("Check that balances are correct after the transaction has been included in a block.")
141          self.generate(self.nodes[0], 1)
142          assert_approx(participants["multisigs"][0].getbalance(), deposit_amount - (value * 2), vspan=0.001)
143          assert_equal(participants["signers"][self.N - 1].getbalance(), value * 2)
144  
145  
146  if __name__ == "__main__":
147      WalletMultisigDescriptorPSBTTest(__file__).main()