rpc_createmultisig.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2015-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 multisig RPCs"""
6 import decimal
7 import itertools
8 import json
9 import os
10
11 from test_framework.address import address_to_scriptpubkey
12 from test_framework.descriptors import descsum_create, drop_origins
13 from test_framework.key import ECPubKey
14 from test_framework.messages import COIN
15 from test_framework.script_util import keys_to_multisig_script
16 from test_framework.test_framework import LimenkaTestFramework
17 from test_framework.util import (
18 assert_raises_rpc_error,
19 assert_equal,
20 )
21 from test_framework.wallet_util import generate_keypair
22 from test_framework.wallet import (
23 MiniWallet,
24 getnewdestination,
25 )
26
27 class RpcCreateMultiSigTest(LimenkaTestFramework):
28 def add_options(self, parser):
29 self.add_wallet_options(parser)
30
31 def set_test_params(self):
32 self.setup_clean_chain = True
33 self.num_nodes = 3
34 self.supports_cli = False
35 self.enable_wallet_if_possible()
36
37 def create_keys(self, num_keys):
38 self.pub = []
39 self.priv = []
40 for _ in range(num_keys):
41 privkey, pubkey = generate_keypair(wif=True)
42 self.pub.append(pubkey.hex())
43 self.priv.append(privkey)
44
45 def create_wallet(self, node, wallet_name):
46 node.createwallet(wallet_name=wallet_name, disable_private_keys=True)
47 return node.get_wallet_rpc(wallet_name)
48
49 def run_test(self):
50 node0, node1, _node2 = self.nodes
51 self.wallet = MiniWallet(test_node=node0)
52
53 if self.is_wallet_compiled():
54 self.check_addmultisigaddress_errors()
55
56 self.log.info('Generating blocks ...')
57 self.generate(self.wallet, 149)
58
59 wallet_multi = self.create_wallet(node1, 'wmulti') if self._requires_wallet else None
60 self.create_keys(21) # max number of allowed keys + 1
61 m_of_n = [(2, 3), (3, 3), (2, 5), (3, 5), (10, 15), (15, 15)]
62 for (sigs, keys) in m_of_n:
63 for output_type in ["bech32", "p2sh-segwit", "legacy"]:
64 self.do_multisig(keys, sigs, output_type, wallet_multi)
65
66 self.test_multisig_script_limit(wallet_multi)
67 self.test_mixing_uncompressed_and_compressed_keys(node0, wallet_multi)
68 self.test_sortedmulti_descriptors_bip67()
69
70 # Check that bech32m is currently not allowed
71 assert_raises_rpc_error(-5, "createmultisig cannot create bech32m multisig addresses", self.nodes[0].createmultisig, 2, self.pub, "bech32m")
72
73 self.log.info('Check correct encoding of multisig script for all n (1..20)')
74 for nkeys in range(1, 20+1):
75 keys = [self.pub[0]]*nkeys
76 expected_ms_script = keys_to_multisig_script(keys, k=nkeys) # simply use n-of-n
77 # note that the 'legacy' address type fails for n values larger than 15
78 # due to exceeding the P2SH size limit (520 bytes), so we use 'bech32' instead
79 # (for the purpose of this encoding test, we don't care about the resulting address)
80 res = self.nodes[0].createmultisig(nrequired=nkeys, keys=keys, address_type='bech32')
81 assert_equal(res['redeemScript'], expected_ms_script.hex())
82
83 def check_addmultisigaddress_errors(self):
84 if self.options.descriptors:
85 return
86 self.log.info('Check that addmultisigaddress fails when the private keys are missing')
87 addresses = [self.nodes[1].getnewaddress(address_type='legacy') for _ in range(2)]
88 assert_raises_rpc_error(-5, 'no full public key for address', lambda: self.nodes[0].addmultisigaddress(nrequired=1, keys=addresses))
89 for a in addresses:
90 # Importing all addresses should not change the result
91 self.nodes[0].importaddress(a)
92 assert_raises_rpc_error(-5, 'no full public key for address', lambda: self.nodes[0].addmultisigaddress(nrequired=1, keys=addresses))
93
94 # Bech32m address type is disallowed for legacy wallets
95 pubs = [self.nodes[1].getaddressinfo(addr)["pubkey"] for addr in addresses]
96 assert_raises_rpc_error(-5, "Bech32m multisig addresses cannot be created with legacy wallets", self.nodes[0].addmultisigaddress, 2, pubs, "", "bech32m")
97
98 def test_multisig_script_limit(self, wallet_multi):
99 node1 = self.nodes[1]
100 pubkeys = self.pub[0:20]
101
102 self.log.info('Test legacy redeem script max size limit')
103 assert_raises_rpc_error(-8, "redeemScript exceeds size limit: 684 > 520", node1.createmultisig, 16, pubkeys, 'legacy')
104
105 self.log.info('Test valid 16-20 multisig p2sh-legacy and bech32 (no wallet)')
106 self.do_multisig(nkeys=20, nsigs=16, output_type="p2sh-segwit", wallet_multi=None)
107 self.do_multisig(nkeys=20, nsigs=16, output_type="bech32", wallet_multi=None)
108
109 self.log.info('Test invalid 16-21 multisig p2sh-legacy and bech32 (no wallet)')
110 assert_raises_rpc_error(-8, "Number of keys involved in the multisignature address creation > 20", node1.createmultisig, 16, self.pub, 'p2sh-segwit')
111 assert_raises_rpc_error(-8, "Number of keys involved in the multisignature address creation > 20", node1.createmultisig, 16, self.pub, 'bech32')
112
113 # Check legacy wallet related command
114 self.log.info('Test legacy redeem script max size limit (with wallet)')
115 if wallet_multi is not None and not self.options.descriptors:
116 assert_raises_rpc_error(-8, "redeemScript exceeds size limit: 684 > 520", wallet_multi.addmultisigaddress, 16, pubkeys, '', 'legacy')
117
118 self.log.info('Test legacy wallet unsupported operation. 16-20 multisig p2sh-legacy and bech32 generation')
119 # Due an internal limitation on legacy wallets, the redeem script limit also applies to p2sh-segwit and bech32 (even when the scripts are valid)
120 # We take this as a "good thing" to tell users to upgrade to descriptors.
121 assert_raises_rpc_error(-4, "Unsupported multisig script size for legacy wallet. Upgrade to descriptors to overcome this limitation for p2sh-segwit or bech32 scripts", wallet_multi.addmultisigaddress, 16, pubkeys, '', 'p2sh-segwit')
122 assert_raises_rpc_error(-4, "Unsupported multisig script size for legacy wallet. Upgrade to descriptors to overcome this limitation for p2sh-segwit or bech32 scripts", wallet_multi.addmultisigaddress, 16, pubkeys, '', 'bech32')
123
124 def do_multisig(self, nkeys, nsigs, output_type, wallet_multi):
125 node0, _node1, node2 = self.nodes
126 pub_keys = self.pub[0: nkeys]
127 priv_keys = self.priv[0: nkeys]
128
129 # Construct the expected descriptor
130 desc = 'multi({},{})'.format(nsigs, ','.join(pub_keys))
131 if output_type == 'legacy':
132 desc = 'sh({})'.format(desc)
133 elif output_type == 'p2sh-segwit':
134 desc = 'sh(wsh({}))'.format(desc)
135 elif output_type == 'bech32':
136 desc = 'wsh({})'.format(desc)
137 desc = descsum_create(desc)
138
139 msig = node2.createmultisig(nsigs, pub_keys, output_type)
140 assert 'warnings' not in msig
141 madd = msig["address"]
142 mredeem = msig["redeemScript"]
143 assert_equal(desc, msig['descriptor'])
144 if output_type == 'bech32':
145 assert madd[0:4] == "bcrt" # actually a bech32 address
146
147 if wallet_multi is not None:
148 # compare against addmultisigaddress
149 msigw = wallet_multi.addmultisigaddress(nsigs, pub_keys, None, output_type)
150 maddw = msigw["address"]
151 mredeemw = msigw["redeemScript"]
152 assert_equal(desc, drop_origins(msigw['descriptor']))
153 # addmultisigiaddress and createmultisig work the same
154 assert maddw == madd
155 assert mredeemw == mredeem
156
157 spk = address_to_scriptpubkey(madd)
158 value = decimal.Decimal("0.00004000")
159 tx = self.wallet.send_to(from_node=self.nodes[0], scriptPubKey=spk, amount=int(value * COIN))
160 prevtxs = [{"txid": tx["txid"], "vout": tx["sent_vout"], "scriptPubKey": spk.hex(), "redeemScript": mredeem, "amount": value}]
161
162 self.generate(node0, 1)
163
164 outval = value - decimal.Decimal("0.00002000") # deduce fee (must be higher than the min relay fee)
165 # send coins to node2 when wallet is enabled
166 node2_balance = node2.getbalances()['mine']['trusted'] if self.is_wallet_compiled() else 0
167 out_addr = node2.getnewaddress() if self.is_wallet_compiled() else getnewdestination('bech32')[2]
168 rawtx = node2.createrawtransaction([{"txid": tx["txid"], "vout": tx["sent_vout"]}], [{out_addr: outval}])
169
170 prevtx_err = dict(prevtxs[0])
171 del prevtx_err["redeemScript"]
172
173 assert_raises_rpc_error(-8, "Missing redeemScript/witnessScript", node2.signrawtransactionwithkey, rawtx, priv_keys[0:nsigs-1], [prevtx_err])
174
175 # if witnessScript specified, all ok
176 prevtx_err["witnessScript"] = prevtxs[0]["redeemScript"]
177 node2.signrawtransactionwithkey(rawtx, priv_keys[0:nsigs-1], [prevtx_err])
178
179 # both specified, also ok
180 prevtx_err["redeemScript"] = prevtxs[0]["redeemScript"]
181 node2.signrawtransactionwithkey(rawtx, priv_keys[0:nsigs-1], [prevtx_err])
182
183 # redeemScript mismatch to witnessScript
184 prevtx_err["redeemScript"] = "6a" # OP_RETURN
185 assert_raises_rpc_error(-8, "redeemScript does not correspond to witnessScript", node2.signrawtransactionwithkey, rawtx, priv_keys[0:nsigs-1], [prevtx_err])
186
187 # redeemScript does not match scriptPubKey
188 del prevtx_err["witnessScript"]
189 assert_raises_rpc_error(-8, "redeemScript/witnessScript does not match scriptPubKey", node2.signrawtransactionwithkey, rawtx, priv_keys[0:nsigs-1], [prevtx_err])
190
191 # witnessScript does not match scriptPubKey
192 prevtx_err["witnessScript"] = prevtx_err["redeemScript"]
193 del prevtx_err["redeemScript"]
194 assert_raises_rpc_error(-8, "redeemScript/witnessScript does not match scriptPubKey", node2.signrawtransactionwithkey, rawtx, priv_keys[0:nsigs-1], [prevtx_err])
195
196 rawtx2 = node2.signrawtransactionwithkey(rawtx, priv_keys[0:nsigs - 1], prevtxs)
197 assert_equal(rawtx2["complete"], False)
198 rawtx3 = node2.signrawtransactionwithkey(rawtx, [priv_keys[-1]], prevtxs)
199 assert_equal(rawtx3["complete"], False)
200 assert_raises_rpc_error(-22, "TX decode failed", node2.combinerawtransaction, [rawtx2['hex'], rawtx3['hex'] + "00"])
201 assert_raises_rpc_error(-22, "Missing transactions", node2.combinerawtransaction, [])
202 combined_rawtx = node2.combinerawtransaction([rawtx2["hex"], rawtx3["hex"]])
203
204 tx = node0.sendrawtransaction(combined_rawtx, 0)
205 blk = self.generate(node0, 1)[0]
206 assert tx in node0.getblock(blk)["tx"]
207
208 assert_raises_rpc_error(-25, "Input not found or already spent", node2.combinerawtransaction, [rawtx2['hex'], rawtx3['hex']])
209
210 # When the wallet is enabled, assert node2 sees the incoming amount
211 if self.is_wallet_compiled():
212 assert_equal(node2.getbalances()['mine']['trusted'], node2_balance + outval)
213
214 txinfo = node0.getrawtransaction(tx, True, blk)
215 self.log.info("n/m=%d/%d %s size=%d vsize=%d weight=%d" % (nsigs, nkeys, output_type, txinfo["size"], txinfo["vsize"], txinfo["weight"]))
216
217 def test_mixing_uncompressed_and_compressed_keys(self, node, wallet_multi):
218 self.log.info('Mixed compressed and uncompressed multisigs are not allowed')
219 pk0, pk1, pk2 = [getnewdestination('bech32')[0].hex() for _ in range(3)]
220
221 # decompress pk2
222 pk_obj = ECPubKey()
223 pk_obj.set(bytes.fromhex(pk2))
224 pk_obj.compressed = False
225 pk2 = pk_obj.get_bytes().hex()
226
227 # Check all permutations of keys because order matters apparently
228 for keys in itertools.permutations([pk0, pk1, pk2]):
229 # Results should be the same as this legacy one
230 legacy_addr = node.createmultisig(2, keys, 'legacy')['address']
231
232 if wallet_multi is not None:
233 # 'addmultisigaddress' should return the same address
234 result = wallet_multi.addmultisigaddress(2, keys, '', 'legacy')
235 assert_equal(legacy_addr, result['address'])
236 assert 'warnings' not in result
237
238 # Generate addresses with the segwit types. These should all make legacy addresses
239 err_msg = ["Unable to make chosen address type, please ensure no uncompressed public keys are present."]
240
241 for addr_type in ['bech32', 'p2sh-segwit']:
242 result = self.nodes[0].createmultisig(nrequired=2, keys=keys, address_type=addr_type)
243 assert_equal(legacy_addr, result['address'])
244 assert_equal(result['warnings'], err_msg)
245
246 if wallet_multi is not None:
247 result = wallet_multi.addmultisigaddress(nrequired=2, keys=keys, address_type=addr_type)
248 assert_equal(legacy_addr, result['address'])
249 assert_equal(result['warnings'], err_msg)
250
251 def test_sortedmulti_descriptors_bip67(self):
252 self.log.info('Testing sortedmulti descriptors with BIP 67 test vectors')
253 with open(os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data/rpc_bip67.json'), encoding='utf-8') as f:
254 vectors = json.load(f)
255
256 for t in vectors:
257 key_str = ','.join(t['keys'])
258 desc = descsum_create('sh(sortedmulti(2,{}))'.format(key_str))
259 assert_equal(self.nodes[0].deriveaddresses(desc)[0], t['address'])
260 sorted_key_str = ','.join(t['sorted_keys'])
261 sorted_key_desc = descsum_create('sh(multi(2,{}))'.format(sorted_key_str))
262 assert_equal(self.nodes[0].deriveaddresses(sorted_key_desc)[0], t['address'])
263
264
265 if __name__ == '__main__':
266 RpcCreateMultiSigTest(__file__).main()
267