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 = 1
20 self.setup_clean_chain = True
21 self.extra_args = [["-keypool=100"]]
22 23 def skip_test_if_missing_module(self):
24 self.skip_if_no_wallet()
25 26 @staticmethod
27 def _get_xpub(wallet):
28 """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)."""
29 pkh_descriptor = next(filter(lambda d: d["desc"].startswith("pkh(") and not d["internal"], wallet.listdescriptors()["descriptors"]))
30 # Keep all key origin information (master key fingerprint and all derivation steps) for proper support of hardware devices
31 # See section 'Key origin identification' in 'doc/descriptors.md' for more details...
32 # Replace the change index with the multipath convention
33 return pkh_descriptor["desc"].split("pkh(")[1].split(")")[0].replace("/0/*", "/<0;1>/*")
34 35 @staticmethod
36 def _check_psbt(psbt, to, value, multisig):
37 """Helper function for any of the N participants to check the psbt with decodepsbt and verify it is OK before signing."""
38 decoded = multisig.decodepsbt(psbt)
39 amount = 0
40 for psbt_out in decoded["outputs"]:
41 address = psbt_out["script"]["address"]
42 assert_equal(multisig.getaddressinfo(address)["ischange"], address != to)
43 if address == to:
44 amount += psbt_out["amount"]
45 assert_approx(amount, float(value), vspan=0.001)
46 47 def participants_create_multisigs(self, xpubs):
48 """The multisig is created by importing the following descriptors. The resulting wallet is watch-only and every participant can do this."""
49 for i in range(self.N):
50 self.node.createwallet(wallet_name=f"{self.name}_{i}", blank=True, disable_private_keys=True)
51 multisig = self.node.get_wallet_rpc(f"{self.name}_{i}")
52 multisig_desc = f"wsh(sortedmulti({self.M},{','.join(xpubs)}))"
53 checksum = multisig.getdescriptorinfo(multisig_desc)["checksum"]
54 result = multisig.importdescriptors([
55 { # Multipath descriptor expands to receive and change
56 "desc": f"{multisig_desc}#{checksum}",
57 "active": True,
58 "timestamp": "now",
59 }
60 ])
61 assert all(r["success"] for r in result)
62 yield multisig
63 64 def run_test(self):
65 self.M = 2
66 self.N = 3
67 self.node = self.nodes[0]
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": [self.node.get_wallet_rpc(self.node.createwallet(wallet_name=f"participant_{i}")["name"]) for i in range(self.N)],
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 for address in receive_addresses:
90 assert_equal(address, receive_addresses[0])
91 change_addresses = [multisig.getrawchangeaddress() for multisig in participants["multisigs"]]
92 for address in change_addresses:
93 assert_equal(address, change_addresses[0])
94 95 self.log.info("Get a mature utxo to send to the multisig...")
96 coordinator_wallet = participants["signers"][0]
97 self.generatetoaddress(self.node, 101, coordinator_wallet.getnewaddress())
98 99 deposit_amount = 6.15
100 multisig_receiving_address = participants["multisigs"][0].getnewaddress()
101 self.log.info("Send funds to the resulting multisig receiving address...")
102 coordinator_wallet.sendtoaddress(multisig_receiving_address, deposit_amount)
103 self.generate(self.node, 1)
104 for participant in participants["multisigs"]:
105 assert_approx(participant.getbalance(), deposit_amount, vspan=0.001)
106 107 self.log.info("Send a transaction from the multisig!")
108 to = participants["signers"][self.N - 1].getnewaddress()
109 value = 1
110 self.log.info("First, make a sending transaction, created using `walletcreatefundedpsbt` (anyone can initiate this)...")
111 psbt = participants["multisigs"][0].walletcreatefundedpsbt(inputs=[], outputs={to: value}, feeRate=0.00010)
112 113 psbts = []
114 self.log.info("Now at least M users check the psbt with decodepsbt and (if OK) signs it with walletprocesspsbt...")
115 for m in range(self.M):
116 signers_multisig = participants["multisigs"][m]
117 self._check_psbt(psbt["psbt"], to, value, signers_multisig)
118 signing_wallet = participants["signers"][m]
119 partially_signed_psbt = signing_wallet.walletprocesspsbt(psbt["psbt"])
120 psbts.append(partially_signed_psbt["psbt"])
121 122 self.log.info("Finally, collect the signed PSBTs with combinepsbt, finalizepsbt, then broadcast the resulting transaction...")
123 combined = coordinator_wallet.combinepsbt(psbts)
124 finalized = coordinator_wallet.finalizepsbt(combined)
125 coordinator_wallet.sendrawtransaction(finalized["hex"])
126 127 self.log.info("Check that balances are correct after the transaction has been included in a block.")
128 self.generate(self.node, 1)
129 assert_approx(participants["multisigs"][0].getbalance(), deposit_amount - value, vspan=0.001)
130 assert_equal(participants["signers"][self.N - 1].getbalance(), value)
131 132 self.log.info("Send another transaction from the multisig, this time with a daisy chained signing flow (one after another in series)!")
133 psbt = participants["multisigs"][0].walletcreatefundedpsbt(inputs=[], outputs={to: value}, feeRate=0.00010)
134 for m in range(self.M):
135 signers_multisig = participants["multisigs"][m]
136 self._check_psbt(psbt["psbt"], to, value, signers_multisig)
137 signing_wallet = participants["signers"][m]
138 psbt = signing_wallet.walletprocesspsbt(psbt["psbt"])
139 assert_equal(psbt["complete"], m == self.M - 1)
140 coordinator_wallet.sendrawtransaction(psbt["hex"])
141 142 self.log.info("Check that balances are correct after the transaction has been included in a block.")
143 self.generate(self.node, 1)
144 assert_approx(participants["multisigs"][0].getbalance(), deposit_amount - (value * 2), vspan=0.001)
145 assert_equal(participants["signers"][self.N - 1].getbalance(), value * 2)
146 147 148 if __name__ == "__main__":
149 WalletMultisigDescriptorPSBTTest(__file__).main()
150