1 #!/usr/bin/env python3
2 # Copyright (c) 2014-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 the wallet keypool and interaction with wallet encryption/locking."""
6 7 import re
8 import time
9 from decimal import Decimal
10 11 from test_framework.descriptors import descsum_create
12 from test_framework.test_framework import LimenkaTestFramework
13 from test_framework.util import assert_equal, assert_raises_rpc_error
14 from test_framework.wallet_util import WalletUnlock
15 16 TEST_KEYPOOL_SIZE = 10
17 TEST_NEW_KEYPOOL_SIZE = TEST_KEYPOOL_SIZE + 2
18 19 class KeyPoolTest(LimenkaTestFramework):
20 def add_options(self, parser):
21 self.add_wallet_options(parser)
22 23 def set_test_params(self):
24 self.num_nodes = 1
25 self.extra_args = [[f"-keypool={TEST_KEYPOOL_SIZE}"]]
26 27 def skip_test_if_missing_module(self):
28 self.skip_if_no_wallet()
29 30 def run_test(self):
31 nodes = self.nodes
32 33 # Derive addresses from the wallet without removing them from keypool
34 addrs = []
35 if not self.options.descriptors:
36 path = str(self.nodes[0].datadir_path / 'wallet.dump')
37 nodes[0].dumpwallet(path)
38 file = open(path, "r", encoding="utf8")
39 m = re.search(r"masterkey: (\w+)", file.read())
40 file.close()
41 xpriv = m.group(1)
42 desc = descsum_create(f"wpkh({xpriv}/0h/0h/*h)")
43 addrs = nodes[0].deriveaddresses(descriptor=desc, range=[0, 9])
44 else:
45 list_descriptors = nodes[0].listdescriptors()
46 for desc in list_descriptors["descriptors"]:
47 if desc['active'] and not desc["internal"] and desc["desc"][:4] == "wpkh":
48 addrs = nodes[0].deriveaddresses(descriptor=desc["desc"], range=[0, 9])
49 50 addr0 = addrs[0]
51 addr9 = addrs[9] # arbitrary future address index
52 53 # Address is mine and active before it is removed from keypool by getnewaddress
54 addr0_before_getting_data = nodes[0].getaddressinfo(addr0)
55 assert addr0_before_getting_data['ismine']
56 assert addr0_before_getting_data['isactive']
57 58 addr_before_encrypting = nodes[0].getnewaddress()
59 addr_before_encrypting_data = nodes[0].getaddressinfo(addr_before_encrypting)
60 assert addr0 == addr_before_encrypting
61 # Address is still mine and active even after being removed from keypool
62 assert addr_before_encrypting_data['ismine']
63 assert addr_before_encrypting_data['isactive']
64 65 wallet_info_old = nodes[0].getwalletinfo()
66 if not self.options.descriptors:
67 assert addr_before_encrypting_data['hdseedid'] == wallet_info_old['hdseedid']
68 69 # Address is mine and active before wallet is encrypted (resetting keypool)
70 addr9_before_encrypting_data = nodes[0].getaddressinfo(addr9)
71 assert addr9_before_encrypting_data['ismine']
72 assert addr9_before_encrypting_data['isactive']
73 74 # Imported things are never considered active, no need to rescan
75 # Imported public keys / addresses can't be mine because they are not spendable
76 if self.options.descriptors:
77 nodes[0].importdescriptors([{
78 "desc": "addr(bcrt1q95gp4zeaah3qcerh35yhw02qeptlzasdtst55v)",
79 "timestamp": "now"
80 }])
81 else:
82 nodes[0].importaddress("bcrt1q95gp4zeaah3qcerh35yhw02qeptlzasdtst55v", "label", rescan=False)
83 import_addr_data = nodes[0].getaddressinfo("bcrt1q95gp4zeaah3qcerh35yhw02qeptlzasdtst55v")
84 assert import_addr_data["iswatchonly"] is not self.options.descriptors
85 assert not import_addr_data["ismine"]
86 assert not import_addr_data["isactive"]
87 88 if self.options.descriptors:
89 nodes[0].importdescriptors([{
90 "desc": "pk(02f893ca95b0d55b4ce4e72ae94982eb679158cb2ebc120ff62c17fedfd1f0700e)",
91 "timestamp": "now"
92 }])
93 else:
94 nodes[0].importpubkey("02f893ca95b0d55b4ce4e72ae94982eb679158cb2ebc120ff62c17fedfd1f0700e", "label", rescan=False)
95 import_pub_data = nodes[0].getaddressinfo("bcrt1q4v7a8wn5vqd6fk4026s5gzzxyu7cfzz23n576h")
96 assert import_pub_data["iswatchonly"] is not self.options.descriptors
97 assert not import_pub_data["ismine"]
98 assert not import_pub_data["isactive"]
99 100 nodes[0].importprivkey("cPMX7v5CNV1zCphFSq2hnR5rCjzAhA1GsBfD1qrJGdj4QEfu38Qx", "label", rescan=False)
101 import_priv_data = nodes[0].getaddressinfo("bcrt1qa985v5d53qqtrfujmzq2zrw3r40j6zz4ns02kj")
102 assert not import_priv_data["iswatchonly"]
103 assert import_priv_data["ismine"]
104 assert not import_priv_data["isactive"]
105 106 # Encrypt wallet and wait to terminate
107 nodes[0].encryptwallet('test')
108 addr9_after_encrypting_data = nodes[0].getaddressinfo(addr9)
109 # Key is from unencrypted seed, no longer considered active
110 assert not addr9_after_encrypting_data['isactive']
111 # ...however it *IS* still mine since we can spend with this key
112 assert addr9_after_encrypting_data['ismine']
113 114 if self.options.descriptors:
115 # Import hardened derivation only descriptors
116 nodes[0].walletpassphrase('test', 10)
117 nodes[0].importdescriptors([
118 {
119 "desc": "wpkh(tprv8ZgxMBicQKsPd7Uf69XL1XwhmjHopUGep8GuEiJDZmbQz6o58LninorQAfcKZWARbtRtfnLcJ5MQ2AtHcQJCCRUcMRvmDUjyEmNUWwx8UbK/0h/*h)#y4dfsj7n",
120 "timestamp": "now",
121 "range": [0,0],
122 "active": True
123 },
124 {
125 "desc": "pkh(tprv8ZgxMBicQKsPd7Uf69XL1XwhmjHopUGep8GuEiJDZmbQz6o58LninorQAfcKZWARbtRtfnLcJ5MQ2AtHcQJCCRUcMRvmDUjyEmNUWwx8UbK/1h/*h)#a0nyvl0k",
126 "timestamp": "now",
127 "range": [0,0],
128 "active": True
129 },
130 {
131 "desc": "sh(wpkh(tprv8ZgxMBicQKsPd7Uf69XL1XwhmjHopUGep8GuEiJDZmbQz6o58LninorQAfcKZWARbtRtfnLcJ5MQ2AtHcQJCCRUcMRvmDUjyEmNUWwx8UbK/2h/*h))#lmeu2axg",
132 "timestamp": "now",
133 "range": [0,0],
134 "active": True
135 },
136 {
137 "desc": "wpkh(tprv8ZgxMBicQKsPd7Uf69XL1XwhmjHopUGep8GuEiJDZmbQz6o58LninorQAfcKZWARbtRtfnLcJ5MQ2AtHcQJCCRUcMRvmDUjyEmNUWwx8UbK/3h/*h)#jkl636gm",
138 "timestamp": "now",
139 "range": [0,0],
140 "active": True,
141 "internal": True
142 },
143 {
144 "desc": "pkh(tprv8ZgxMBicQKsPd7Uf69XL1XwhmjHopUGep8GuEiJDZmbQz6o58LninorQAfcKZWARbtRtfnLcJ5MQ2AtHcQJCCRUcMRvmDUjyEmNUWwx8UbK/4h/*h)#l3crwaus",
145 "timestamp": "now",
146 "range": [0,0],
147 "active": True,
148 "internal": True
149 },
150 {
151 "desc": "sh(wpkh(tprv8ZgxMBicQKsPd7Uf69XL1XwhmjHopUGep8GuEiJDZmbQz6o58LninorQAfcKZWARbtRtfnLcJ5MQ2AtHcQJCCRUcMRvmDUjyEmNUWwx8UbK/5h/*h))#qg8wa75f",
152 "timestamp": "now",
153 "range": [0,0],
154 "active": True,
155 "internal": True
156 }
157 ])
158 nodes[0].walletlock()
159 # Keep creating keys until we run out
160 for _ in range(TEST_KEYPOOL_SIZE - 1):
161 nodes[0].getnewaddress()
162 addr = nodes[0].getnewaddress()
163 addr_data = nodes[0].getaddressinfo(addr)
164 wallet_info = nodes[0].getwalletinfo()
165 assert addr_before_encrypting_data['hdmasterfingerprint'] != addr_data['hdmasterfingerprint']
166 if not self.options.descriptors:
167 assert addr_data['hdseedid'] == wallet_info['hdseedid']
168 assert_raises_rpc_error(-12, "Error: Keypool ran out, please call keypoolrefill first", nodes[0].getnewaddress)
169 170 # put two new keys in the keypool
171 with WalletUnlock(nodes[0], 'test'):
172 nodes[0].keypoolrefill(TEST_NEW_KEYPOOL_SIZE)
173 wi = nodes[0].getwalletinfo()
174 if self.options.descriptors:
175 # Descriptors wallet: keypool size applies to both internal and external
176 # chains and there are four of each (legacy, nested, segwit, and taproot)
177 assert_equal(wi['keypoolsize_hd_internal'], TEST_NEW_KEYPOOL_SIZE * 4)
178 assert_equal(wi['keypoolsize'], TEST_NEW_KEYPOOL_SIZE * 4)
179 else:
180 # Legacy wallet: keypool size applies to both internal and external HD chains
181 assert_equal(wi['keypoolsize_hd_internal'], TEST_NEW_KEYPOOL_SIZE)
182 assert_equal(wi['keypoolsize'], TEST_NEW_KEYPOOL_SIZE)
183 184 # drain the internal keys
185 for _ in range(TEST_NEW_KEYPOOL_SIZE):
186 nodes[0].getrawchangeaddress()
187 # remember keypool sizes
188 wi = nodes[0].getwalletinfo()
189 kp_size_before = [wi['keypoolsize_hd_internal'], wi['keypoolsize']]
190 # the next one should fail
191 assert_raises_rpc_error(-12, "Keypool ran out", nodes[0].getrawchangeaddress)
192 # check that keypool sizes did not change
193 wi = nodes[0].getwalletinfo()
194 kp_size_after = [wi['keypoolsize_hd_internal'], wi['keypoolsize']]
195 assert_equal(kp_size_before, kp_size_after)
196 197 # drain the external keys
198 addr = set()
199 for _ in range(TEST_NEW_KEYPOOL_SIZE):
200 addr.add(nodes[0].getnewaddress(address_type="bech32"))
201 # remember keypool sizes
202 wi = nodes[0].getwalletinfo()
203 kp_size_before = [wi['keypoolsize_hd_internal'], wi['keypoolsize']]
204 # the next one should fail
205 assert_raises_rpc_error(-12, "Error: Keypool ran out, please call keypoolrefill first", nodes[0].getnewaddress)
206 # check that keypool sizes did not change
207 wi = nodes[0].getwalletinfo()
208 kp_size_after = [wi['keypoolsize_hd_internal'], wi['keypoolsize']]
209 assert_equal(kp_size_before, kp_size_after)
210 211 # refill keypool
212 nodes[0].walletpassphrase('test', 1)
213 # At this point the keypool has >45 keys in it
214 # calling keypoolrefill with anything smaller than that is a noop
215 nodes[0].keypoolrefill(50)
216 217 # test walletpassphrase timeout
218 time.sleep(1.1)
219 assert_equal(nodes[0].getwalletinfo()["unlocked_until"], 0)
220 221 # drain the keypool
222 for _ in range(50):
223 nodes[0].getnewaddress()
224 assert_raises_rpc_error(-12, "Keypool ran out", nodes[0].getnewaddress)
225 226 with WalletUnlock(nodes[0], 'test'):
227 nodes[0].keypoolrefill(100)
228 wi = nodes[0].getwalletinfo()
229 if self.options.descriptors:
230 assert_equal(wi['keypoolsize_hd_internal'], 400)
231 assert_equal(wi['keypoolsize'], 400)
232 else:
233 assert_equal(wi['keypoolsize_hd_internal'], 100)
234 assert_equal(wi['keypoolsize'], 100)
235 236 if not self.options.descriptors:
237 # Check that newkeypool entirely flushes the keypool
238 start_keypath = nodes[0].getaddressinfo(nodes[0].getnewaddress())['hdkeypath']
239 start_change_keypath = nodes[0].getaddressinfo(nodes[0].getrawchangeaddress())['hdkeypath']
240 # flush keypool and get new addresses
241 nodes[0].newkeypool()
242 end_keypath = nodes[0].getaddressinfo(nodes[0].getnewaddress())['hdkeypath']
243 end_change_keypath = nodes[0].getaddressinfo(nodes[0].getrawchangeaddress())['hdkeypath']
244 # The new keypath index should be 100 more than the old one
245 new_index = int(start_keypath.rsplit('/', 1)[1][:-1]) + 100
246 new_change_index = int(start_change_keypath.rsplit('/', 1)[1][:-1]) + 100
247 assert_equal(end_keypath, "m/0'/0'/" + str(new_index) + "'")
248 assert_equal(end_change_keypath, "m/0'/1'/" + str(new_change_index) + "'")
249 250 # create a blank wallet
251 nodes[0].createwallet(wallet_name='w2', blank=True, disable_private_keys=True)
252 w2 = nodes[0].get_wallet_rpc('w2')
253 254 # refer to initial wallet as w1
255 w1 = nodes[0].get_wallet_rpc(self.default_wallet_name)
256 257 # import private key and fund it
258 address = addr.pop()
259 desc = w1.getaddressinfo(address)['desc']
260 if self.options.descriptors:
261 res = w2.importdescriptors([{'desc': desc, 'timestamp': 'now'}])
262 else:
263 res = w2.importmulti([{'desc': desc, 'timestamp': 'now'}])
264 assert_equal(res[0]['success'], True)
265 266 with WalletUnlock(w1, 'test'):
267 res = w1.sendtoaddress(address=address, amount=0.00010000)
268 self.generate(nodes[0], 1)
269 destination = addr.pop()
270 271 # Using a fee rate (10 sat / byte) well above the minimum relay rate
272 # creating a 5,000 sat transaction with change should not be possible
273 assert_raises_rpc_error(-4, "Transaction needs a change address, but we can't generate it.", w2.walletcreatefundedpsbt, inputs=[], outputs=[{addr.pop(): 0.00005000}], subtractFeeFromOutputs=[0], feeRate=0.00010)
274 275 # creating a 10,000 sat transaction without change, with a manual input, should still be possible
276 res = w2.walletcreatefundedpsbt(inputs=w2.listunspent(), outputs=[{destination: 0.00010000}], subtractFeeFromOutputs=[0], feeRate=0.00010)
277 assert_equal("psbt" in res, True)
278 279 # creating a 10,000 sat transaction without change should still be possible
280 res = w2.walletcreatefundedpsbt(inputs=[], outputs=[{destination: 0.00010000}], subtractFeeFromOutputs=[0], feeRate=0.00010)
281 assert_equal("psbt" in res, True)
282 # should work without subtractFeeFromOutputs if the exact fee is subtracted from the amount
283 res = w2.walletcreatefundedpsbt(inputs=[], outputs=[{destination: 0.00008900}], feeRate=0.00010)
284 assert_equal("psbt" in res, True)
285 286 # dust change should be removed
287 res = w2.walletcreatefundedpsbt(inputs=[], outputs=[{destination: 0.00008800}], feeRate=0.00010)
288 assert_equal("psbt" in res, True)
289 290 # create a transaction without change at the maximum fee rate, such that the output is still spendable:
291 res = w2.walletcreatefundedpsbt(inputs=[], outputs=[{destination: 0.00010000}], subtractFeeFromOutputs=[0], feeRate=0.0008823)
292 assert_equal("psbt" in res, True)
293 assert_equal(res["fee"], Decimal("0.00009706"))
294 295 # creating a 10,000 sat transaction with a manual change address should be possible
296 res = w2.walletcreatefundedpsbt(inputs=[], outputs=[{destination: 0.00010000}], subtractFeeFromOutputs=[0], feeRate=0.00010, changeAddress=addr.pop())
297 assert_equal("psbt" in res, True)
298 299 if not self.options.descriptors:
300 msg = "Error: Private keys are disabled for this wallet"
301 assert_raises_rpc_error(-4, msg, w2.keypoolrefill, 100)
302 303 if __name__ == '__main__':
304 KeyPoolTest(__file__).main()
305