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