1  #!/usr/bin/env python3
  2  # Copyright (c) 2024-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 miniscript multisig that starts as 4-of-4 and "decays" to 3-of-4, 2-of-4, and finally 1-of-4 at each future halvening block height.
  6  
  7  Spending policy: `thresh(4,pk(key_1),pk(key_2),pk(key_3),pk(key_4),after(t1),after(t2),after(t3))`
  8  This is similar to `test/functional/wallet_multisig_descriptor_psbt.py`.
  9  """
 10  
 11  import random
 12  from test_framework.test_framework import BitcoinTestFramework
 13  from test_framework.util import (
 14      assert_approx,
 15      assert_equal,
 16      assert_raises_rpc_error,
 17  )
 18  
 19  
 20  class WalletMiniscriptDecayingMultisigDescriptorPSBTTest(BitcoinTestFramework):
 21      def set_test_params(self):
 22          self.num_nodes = 1
 23          self.setup_clean_chain = True
 24          self.wallet_names = []
 25          self.extra_args = [["-keypool=100"]]
 26  
 27      def skip_test_if_missing_module(self):
 28          self.skip_if_no_wallet()
 29  
 30      @staticmethod
 31      def _get_xpub(wallet):
 32          """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)."""
 33          pkh_descriptor = next(filter(lambda d: d["desc"].startswith("pkh(") and not d["internal"], wallet.listdescriptors()["descriptors"]))
 34          # keep all key origin information (master key fingerprint and all derivation steps) for proper support of hardware devices
 35          # see section 'Key origin identification' in 'doc/descriptors.md' for more details...
 36          # Replace the change index with the multipath convention
 37          return pkh_descriptor["desc"].split("pkh(")[1].split(")")[0].replace("/0/*", "/<0;1>/*")
 38  
 39      def create_multisig(self, xpubs):
 40          """The multisig is created by importing a single multipath descriptor. The resulting wallet is watch-only and every signer can do this."""
 41          self.node.createwallet(wallet_name=f"{self.name}", blank=True, disable_private_keys=True)
 42          multisig = self.node.get_wallet_rpc(f"{self.name}")
 43          # spending policy: `thresh(4,pk(key_1),pk(key_2),pk(key_3),pk(key_4),after(t1),after(t2),after(t3))`
 44          # IMPORTANT: when backing up your descriptor, the order of key_1...key_4 must be correct!
 45          multisig_desc = f"wsh(thresh({self.N},pk({'),s:pk('.join(xpubs)}),sln:after({'),sln:after('.join(map(str, self.locktimes))})))"
 46          checksum = multisig.getdescriptorinfo(multisig_desc)["checksum"]
 47          result = multisig.importdescriptors([
 48              {  # Multipath descriptor expands to receive and change
 49                  "desc": f"{multisig_desc}#{checksum}",
 50                  "active": True,
 51                  "timestamp": "now",
 52              },
 53          ])
 54          assert all(r["success"] for r in result)
 55          return multisig
 56  
 57      def run_test(self):
 58          self.node = self.nodes[0]
 59          self.M = 4  # starts as 4-of-4
 60          self.N = 4
 61  
 62          self.locktimes = [104, 106, 108]
 63          assert_equal(len(self.locktimes), self.N - 1)
 64  
 65          self.name = f"{self.M}_of_{self.N}_decaying_multisig"
 66          self.log.info(f"Testing a miniscript multisig which starts as 4-of-4 and 'decays' to 3-of-4 at block height {self.locktimes[0]}, 2-of-4 at {self.locktimes[1]}, and finally 1-of-4 at {self.locktimes[2]}...")
 67  
 68          self.log.info("Create the signer wallets and get their xpubs...")
 69          signers = [self.node.get_wallet_rpc(self.node.createwallet(wallet_name=f"signer_{i}")["name"]) for i in range(self.N)]
 70          xpubs = [self._get_xpub(signer) for signer in signers]
 71  
 72          self.log.info("Create the watch-only decaying multisig using signers' xpubs...")
 73          multisig = self.create_multisig(xpubs)
 74  
 75          self.log.info("Get a mature utxo to send to the multisig...")
 76          coordinator_wallet = self.node.get_wallet_rpc(self.node.createwallet(wallet_name="coordinator")["name"])
 77          self.generatetoaddress(self.node, 101, coordinator_wallet.getnewaddress())
 78  
 79          self.log.info("Send funds to the multisig's receiving address...")
 80          deposit_amount = 6.15
 81          coordinator_wallet.sendtoaddress(multisig.getnewaddress(), deposit_amount)
 82          self.generate(self.node, 1)
 83          assert_approx(multisig.getbalance(), deposit_amount, vspan=0.001)
 84  
 85          self.log.info("Send transactions from the multisig as required signers decay...")
 86          amount = 1.5
 87          receiver = signers[0]
 88          sent = 0
 89          for locktime in [0] + self.locktimes:
 90              self.log.info(f"At block height >= {locktime} this multisig is {self.M}-of-{self.N}")
 91              current_height = self.node.getblock(self.node.getbestblockhash())['height']
 92  
 93              # in this test each signer signs the same psbt "in series" one after the other.
 94              # Another option is for each signer to sign the original psbt, and then combine
 95              # and finalize these. In some cases this may be more optimal for coordination.
 96              psbt = multisig.walletcreatefundedpsbt(inputs=[], outputs={receiver.getnewaddress(): amount}, feeRate=0.00010, locktime=locktime)
 97              # the random sample asserts that any of the signing keys can sign for the 3-of-4,
 98              # 2-of-4, and 1-of-4. While this is basic behavior of the miniscript thresh primitive,
 99              # it is a critical property of this wallet.
100              for i, m in enumerate(random.sample(range(self.M), self.M)):
101                  psbt = signers[m].walletprocesspsbt(psbt["psbt"])
102                  assert_equal(psbt["complete"], i == self.M - 1)
103  
104              if self.M < self.N:
105                  self.log.info(f"Check that the time-locked transaction is too immature to spend with {self.M}-of-{self.N} at block height {current_height}...")
106                  assert_equal(current_height >= locktime, False)
107                  assert_raises_rpc_error(-26, "non-final", multisig.sendrawtransaction, psbt["hex"])
108  
109                  self.log.info(f"Generate blocks to reach the time-lock block height {locktime} and broadcast the transaction...")
110                  self.generate(self.node, locktime - current_height)
111              else:
112                  self.log.info("All the signers are required to spend before the first locktime")
113  
114              multisig.sendrawtransaction(psbt["hex"])
115              sent += amount
116  
117              self.log.info("Check that balances are correct after the transaction has been included in a block...")
118              self.generate(self.node, 1)
119              assert_approx(multisig.getbalance(), deposit_amount - sent, vspan=0.001)
120              assert_equal(receiver.getbalance(), sent)
121  
122              self.M -= 1  # decay the number of required signers for the next locktime..
123  
124  
125  if __name__ == "__main__":
126      WalletMiniscriptDecayingMultisigDescriptorPSBTTest(__file__).main()