wallet_descriptor.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2019-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 descriptor wallet function."""
6
7 try:
8 import sqlite3
9 except ImportError:
10 pass
11
12 import re
13
14 from test_framework.blocktools import COINBASE_MATURITY
15 from test_framework.descriptors import descsum_create
16 from test_framework.extendedkey import ExtendedPrivateKey
17 from test_framework.messages import ser_string
18 from test_framework.test_framework import BitcoinTestFramework
19 from test_framework.util import (
20 assert_not_equal,
21 assert_equal,
22 assert_raises_rpc_error
23 )
24 from test_framework.wallet_util import WalletUnlock
25
26
27 class WalletDescriptorTest(BitcoinTestFramework):
28 def set_test_params(self):
29 self.setup_clean_chain = True
30 self.num_nodes = 1
31 self.extra_args = [['-keypool=100']]
32
33 def skip_test_if_missing_module(self):
34 self.skip_if_no_wallet()
35 self.skip_if_no_py_sqlite3()
36
37 def test_parent_descriptors(self):
38 self.log.info("Check that parent_descs is the same for all RPCs and is normalized")
39 self.nodes[0].createwallet(wallet_name="parent_descs")
40 wallet = self.nodes[0].get_wallet_rpc("parent_descs")
41 default_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
42
43 addr = wallet.getnewaddress()
44 parent_desc = wallet.getaddressinfo(addr)["parent_desc"]
45
46 # Verify that the parent descriptor is normalized
47 # First remove the checksum
48 desc_verify = parent_desc.split("#")[0]
49 # Next extract the xpub
50 desc_verify = re.sub(r"tpub\w+?(?=/)", "", desc_verify)
51 # Extract origin info
52 origin_match = re.search(r'\[([\da-fh/]+)\]', desc_verify)
53 origin_part = origin_match.group(1) if origin_match else ""
54 # Split on "]" for everything after the origin info
55 after_origin = desc_verify.split("]", maxsplit=1)[-1]
56 # Look for the hardened markers “h” inside each piece
57 # We don't need to check for aspostrophe as normalization will not output aspostrophe
58 found_hardened_in_origin = "h" in origin_part
59 found_hardened_after_origin = "h" in after_origin
60 assert_equal(found_hardened_in_origin, True)
61 assert_equal(found_hardened_after_origin, False)
62
63 # Send some coins so we can check listunspent, listtransactions, listunspent, and gettransaction
64 since_block = self.nodes[0].getbestblockhash()
65 txid = default_wallet.sendtoaddress(addr, 1)
66 self.generate(self.nodes[0], 1)
67
68 unspent = wallet.listunspent()
69 assert_equal(len(unspent), 1)
70 assert_equal(unspent[0]["parent_descs"], [parent_desc])
71
72 txs = wallet.listtransactions()
73 assert_equal(len(txs), 1)
74 assert_equal(txs[0]["parent_descs"], [parent_desc])
75
76 txs = wallet.listsinceblock(since_block)["transactions"]
77 assert_equal(len(txs), 1)
78 assert_equal(txs[0]["parent_descs"], [parent_desc])
79
80 tx = wallet.gettransaction(txid=txid, verbose=True)
81 assert_equal(tx["details"][0]["parent_descs"], [parent_desc])
82
83 wallet.unloadwallet()
84
85 def run_test(self):
86 self.generate(self.nodes[0], COINBASE_MATURITY + 1)
87
88 # Make a descriptor wallet
89 self.log.info("Making a descriptor wallet")
90 self.nodes[0].createwallet(wallet_name="desc1")
91 wallet = self.nodes[0].get_wallet_rpc("desc1")
92
93 # A descriptor wallet should have 100 addresses * 4 types = 400 keys
94 self.log.info("Checking wallet info")
95 wallet_info = wallet.getwalletinfo()
96 assert_equal(wallet_info['format'], 'sqlite')
97 assert_equal(wallet_info['keypoolsize'], 400)
98 assert_equal(wallet_info['keypoolsize_hd_internal'], 400)
99 assert 'keypoololdest' not in wallet_info
100
101 # Check that getnewaddress works
102 self.log.info("Test that getnewaddress and getrawchangeaddress work")
103 addr = wallet.getnewaddress("", "legacy")
104 addr_info = wallet.getaddressinfo(addr)
105 assert addr_info['desc'].startswith('pkh(')
106 assert_equal(addr_info['hdkeypath'], 'm/44h/1h/0h/0/0')
107
108 addr = wallet.getnewaddress("", "p2sh-segwit")
109 addr_info = wallet.getaddressinfo(addr)
110 assert addr_info['desc'].startswith('sh(wpkh(')
111 assert_equal(addr_info['hdkeypath'], 'm/49h/1h/0h/0/0')
112
113 addr = wallet.getnewaddress("", "bech32")
114 addr_info = wallet.getaddressinfo(addr)
115 assert addr_info['desc'].startswith('wpkh(')
116 assert_equal(addr_info['hdkeypath'], 'm/84h/1h/0h/0/0')
117
118 addr = wallet.getnewaddress("", "bech32m")
119 addr_info = wallet.getaddressinfo(addr)
120 assert addr_info['desc'].startswith('tr(')
121 assert_equal(addr_info['hdkeypath'], 'm/86h/1h/0h/0/0')
122
123 # Check that getrawchangeaddress works
124 addr = wallet.getrawchangeaddress("legacy")
125 addr_info = wallet.getaddressinfo(addr)
126 assert addr_info['desc'].startswith('pkh(')
127 assert_equal(addr_info['hdkeypath'], 'm/44h/1h/0h/1/0')
128
129 addr = wallet.getrawchangeaddress("p2sh-segwit")
130 addr_info = wallet.getaddressinfo(addr)
131 assert addr_info['desc'].startswith('sh(wpkh(')
132 assert_equal(addr_info['hdkeypath'], 'm/49h/1h/0h/1/0')
133
134 addr = wallet.getrawchangeaddress("bech32")
135 addr_info = wallet.getaddressinfo(addr)
136 assert addr_info['desc'].startswith('wpkh(')
137 assert_equal(addr_info['hdkeypath'], 'm/84h/1h/0h/1/0')
138
139 addr = wallet.getrawchangeaddress("bech32m")
140 addr_info = wallet.getaddressinfo(addr)
141 assert addr_info['desc'].startswith('tr(')
142 assert_equal(addr_info['hdkeypath'], 'm/86h/1h/0h/1/0')
143
144 # Make a wallet to receive coins at
145 self.nodes[0].createwallet(wallet_name="desc2")
146 recv_wrpc = self.nodes[0].get_wallet_rpc("desc2")
147 send_wrpc = self.nodes[0].get_wallet_rpc("desc1")
148
149 # Generate some coins
150 self.generatetoaddress(self.nodes[0], COINBASE_MATURITY + 1, send_wrpc.getnewaddress())
151
152 # Make transactions
153 self.log.info("Test sending and receiving")
154 addr = recv_wrpc.getnewaddress()
155 send_wrpc.sendtoaddress(addr, 10)
156
157 self.log.info("Test encryption")
158 # Get the master fingerprint before encrypt
159 info1 = send_wrpc.getaddressinfo(send_wrpc.getnewaddress())
160
161 # Encrypt wallet 0
162 send_wrpc.encryptwallet('pass')
163 with WalletUnlock(send_wrpc, "pass"):
164 addr = send_wrpc.getnewaddress()
165 info2 = send_wrpc.getaddressinfo(addr)
166 assert_not_equal(info1['hdmasterfingerprint'], info2['hdmasterfingerprint'])
167 assert 'hdmasterfingerprint' in send_wrpc.getaddressinfo(send_wrpc.getnewaddress())
168 info3 = send_wrpc.getaddressinfo(addr)
169 assert_equal(info2['desc'], info3['desc'])
170
171 self.log.info("Test that getnewaddress still works after keypool is exhausted in an encrypted wallet")
172 for _ in range(500):
173 send_wrpc.getnewaddress()
174
175 self.log.info("Test that unlock is needed when deriving only hardened keys in an encrypted wallet")
176 with WalletUnlock(send_wrpc, "pass"):
177 send_wrpc.importdescriptors([{
178 "desc": descsum_create(f"wpkh({ExtendedPrivateKey.generate().to_string()}/0h/*h)"),
179 "timestamp": "now",
180 "range": [0,10],
181 "active": True
182 }])
183 # Exhaust keypool of 100
184 for _ in range(100):
185 send_wrpc.getnewaddress(address_type='bech32')
186 # This should now error
187 assert_raises_rpc_error(-12, "Keypool ran out, please call keypoolrefill first", send_wrpc.getnewaddress, '', 'bech32')
188
189 self.log.info("Test born encrypted wallets")
190 self.nodes[0].createwallet('desc_enc', False, False, 'pass', False, True)
191 enc_rpc = self.nodes[0].get_wallet_rpc('desc_enc')
192 enc_rpc.getnewaddress() # Makes sure that we can get a new address from a born encrypted wallet
193
194 self.log.info("Test blank descriptor wallets")
195 self.nodes[0].createwallet(wallet_name='desc_blank', blank=True)
196 blank_rpc = self.nodes[0].get_wallet_rpc('desc_blank')
197 assert_raises_rpc_error(-4, 'This wallet has no available keys', blank_rpc.getnewaddress)
198
199 self.log.info("Test descriptor wallet with disabled private keys")
200 self.nodes[0].createwallet(wallet_name='desc_no_priv', disable_private_keys=True)
201 nopriv_rpc = self.nodes[0].get_wallet_rpc('desc_no_priv')
202 assert_raises_rpc_error(-4, 'This wallet has no available keys', nopriv_rpc.getnewaddress)
203
204 self.log.info("Test descriptor exports")
205 self.nodes[0].createwallet(wallet_name='desc_export')
206 exp_rpc = self.nodes[0].get_wallet_rpc('desc_export')
207 self.nodes[0].createwallet(wallet_name='desc_import', disable_private_keys=True)
208 imp_rpc = self.nodes[0].get_wallet_rpc('desc_import')
209
210 addr_types = [('legacy', False, 'pkh(', '44h/1h/0h', -13),
211 ('p2sh-segwit', False, 'sh(wpkh(', '49h/1h/0h', -14),
212 ('bech32', False, 'wpkh(', '84h/1h/0h', -13),
213 ('bech32m', False, 'tr(', '86h/1h/0h', -13),
214 ('legacy', True, 'pkh(', '44h/1h/0h', -13),
215 ('p2sh-segwit', True, 'sh(wpkh(', '49h/1h/0h', -14),
216 ('bech32', True, 'wpkh(', '84h/1h/0h', -13),
217 ('bech32m', True, 'tr(', '86h/1h/0h', -13)]
218
219 for addr_type, internal, desc_prefix, deriv_path, int_idx in addr_types:
220 int_str = 'internal' if internal else 'external'
221
222 self.log.info("Testing descriptor address type for {} {}".format(addr_type, int_str))
223 if internal:
224 addr = exp_rpc.getrawchangeaddress(address_type=addr_type)
225 else:
226 addr = exp_rpc.getnewaddress(address_type=addr_type)
227 desc = exp_rpc.getaddressinfo(addr)['parent_desc']
228 assert_equal(desc_prefix, desc[0:len(desc_prefix)])
229 idx = desc.index('/') + 1
230 assert_equal(deriv_path, desc[idx:idx + 9])
231 if internal:
232 assert_equal('1', desc[int_idx])
233 else:
234 assert_equal('0', desc[int_idx])
235
236 self.log.info("Testing the same descriptor is returned for address type {} {}".format(addr_type, int_str))
237 for i in range(0, 10):
238 if internal:
239 addr = exp_rpc.getrawchangeaddress(address_type=addr_type)
240 else:
241 addr = exp_rpc.getnewaddress(address_type=addr_type)
242 test_desc = exp_rpc.getaddressinfo(addr)['parent_desc']
243 assert_equal(desc, test_desc)
244
245 self.log.info("Testing import of exported {} descriptor".format(addr_type))
246 imp_rpc.importdescriptors([{
247 'desc': desc,
248 'active': True,
249 'next_index': 11,
250 'timestamp': 'now',
251 'internal': internal
252 }])
253
254 for i in range(0, 10):
255 if internal:
256 exp_addr = exp_rpc.getrawchangeaddress(address_type=addr_type)
257 imp_addr = imp_rpc.getrawchangeaddress(address_type=addr_type)
258 else:
259 exp_addr = exp_rpc.getnewaddress(address_type=addr_type)
260 imp_addr = imp_rpc.getnewaddress(address_type=addr_type)
261 assert_equal(exp_addr, imp_addr)
262
263 self.log.info("Test that loading descriptor wallet containing legacy key types throws error")
264 self.nodes[0].createwallet(wallet_name="crashme")
265 self.nodes[0].unloadwallet("crashme")
266 wallet_db = self.nodes[0].wallets_path / "crashme" / self.wallet_data_filename
267 conn = sqlite3.connect(wallet_db)
268 with conn:
269 # add "cscript" entry: key type is uint160 (20 bytes), value type is CScript (zero-length here)
270 conn.execute('INSERT INTO main VALUES(?, ?)', (b'\x07cscript' + b'\x00'*20, b'\x00'))
271 conn.close()
272 assert_raises_rpc_error(-4, "Unexpected legacy entry in descriptor wallet found.", self.nodes[0].loadwallet, "crashme")
273
274 self.log.info("Test that loading descriptor wallet containing both unencrypted and encrypted keys for same descriptor fails to load")
275 wallet_name = "mixed_crypt"
276 self.nodes[0].createwallet(wallet_name)
277 self.nodes[0].unloadwallet(wallet_name)
278 wallet_db = self.nodes[0].wallets_path / wallet_name / self.wallet_data_filename
279 conn = sqlite3.connect(wallet_db)
280 with conn:
281 key_prefix = ser_string(b"walletdescriptorkey")
282 ckey_prefix = ser_string(b"walletdescriptorckey")
283 rows = conn.execute('SELECT key, value FROM main').fetchall()
284 key_rows = [(k, v) for k, v in rows if k.startswith(key_prefix)]
285 # Test the test, want to be sure there is at least one unencrypted key.
286 assert len(key_rows) >= 1
287 k, v = key_rows[0]
288 conn.execute('INSERT INTO main VALUES(?, ?)', (k.replace(key_prefix, ckey_prefix), v))
289 conn.close()
290 with self.nodes[0].assert_debug_log(["Wallet contains both unencrypted and encrypted keys"]):
291 assert_raises_rpc_error(-4, "Wallet corrupted", self.nodes[0].loadwallet, wallet_name)
292
293 self.test_parent_descriptors()
294
295 if __name__ == '__main__':
296 WalletDescriptorTest(__file__).main()
297