1 #!/usr/bin/env python3
2 # Copyright (c) 2018-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 avoid_reuse and setwalletflag features."""
6 7 from test_framework.address import address_to_scriptpubkey
8 from test_framework.test_framework import LimenkaTestFramework
9 from test_framework.util import (
10 assert_approx,
11 assert_equal,
12 assert_raises_rpc_error,
13 )
14 15 def reset_balance(node, discardaddr):
16 '''Throw away all owned coins by the node so it gets a balance of 0.'''
17 balance = node.getbalance(avoid_reuse=False)
18 if balance > 0.5:
19 node.sendtoaddress(address=discardaddr, amount=balance, subtractfeefromamount=True, avoid_reuse=False)
20 21 def count_unspent(node):
22 '''Count the unspent outputs for the given node and return various statistics'''
23 r = {
24 "total": {
25 "count": 0,
26 "sum": 0,
27 },
28 "reused": {
29 "count": 0,
30 "sum": 0,
31 },
32 }
33 supports_reused = True
34 for utxo in node.listunspent(minconf=0):
35 r["total"]["count"] += 1
36 r["total"]["sum"] += utxo["amount"]
37 if supports_reused and "reused" in utxo:
38 if utxo["reused"]:
39 r["reused"]["count"] += 1
40 r["reused"]["sum"] += utxo["amount"]
41 else:
42 supports_reused = False
43 r["reused"]["supported"] = supports_reused
44 return r
45 46 def assert_unspent(node, total_count=None, total_sum=None, reused_supported=None, reused_count=None, reused_sum=None, margin=0.001):
47 '''Make assertions about a node's unspent output statistics'''
48 stats = count_unspent(node)
49 if total_count is not None:
50 assert_equal(stats["total"]["count"], total_count)
51 if total_sum is not None:
52 assert_approx(stats["total"]["sum"], total_sum, margin)
53 if reused_supported is not None:
54 assert_equal(stats["reused"]["supported"], reused_supported)
55 if reused_count is not None:
56 assert_equal(stats["reused"]["count"], reused_count)
57 if reused_sum is not None:
58 assert_approx(stats["reused"]["sum"], reused_sum, margin)
59 60 def assert_balances(node, mine, margin=0.001):
61 '''Make assertions about a node's getbalances output'''
62 got = node.getbalances()["mine"]
63 for k,v in mine.items():
64 assert_approx(got[k], v, margin)
65 66 class AvoidReuseTest(LimenkaTestFramework):
67 def add_options(self, parser):
68 self.add_wallet_options(parser)
69 70 def set_test_params(self):
71 self.num_nodes = 2
72 # whitelist peers to speed up tx relay / mempool sync
73 self.noban_tx_relay = True
74 75 def skip_test_if_missing_module(self):
76 self.skip_if_no_wallet()
77 78 def run_test(self):
79 '''Set up initial chain and run tests defined below'''
80 81 self.test_persistence()
82 self.test_immutable()
83 84 self.generate(self.nodes[0], 110)
85 self.test_change_remains_change(self.nodes[1])
86 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
87 self.test_sending_from_reused_address_without_avoid_reuse()
88 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
89 self.test_sending_from_reused_address_fails("legacy")
90 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
91 self.test_sending_from_reused_address_fails("p2sh-segwit")
92 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
93 self.test_sending_from_reused_address_fails("bech32")
94 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
95 self.test_getbalances_used()
96 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
97 self.test_full_destination_group_is_preferred()
98 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
99 self.test_all_destination_groups_are_used()
100 101 def test_persistence(self):
102 '''Test that wallet files persist the avoid_reuse flag.'''
103 self.log.info("Test wallet files persist avoid_reuse flag")
104 105 # Configure node 1 to use avoid_reuse
106 self.nodes[1].setwalletflag('avoid_reuse')
107 108 # Flags should be node1.avoid_reuse=false, node2.avoid_reuse=true
109 assert_equal(self.nodes[0].getwalletinfo()["avoid_reuse"], False)
110 assert_equal(self.nodes[1].getwalletinfo()["avoid_reuse"], True)
111 112 self.restart_node(1)
113 self.connect_nodes(0, 1)
114 115 # Flags should still be node1.avoid_reuse=false, node2.avoid_reuse=true
116 assert_equal(self.nodes[0].getwalletinfo()["avoid_reuse"], False)
117 assert_equal(self.nodes[1].getwalletinfo()["avoid_reuse"], True)
118 119 # Attempting to set flag to its current state should throw
120 assert_raises_rpc_error(-8, "Wallet flag is already set to false", self.nodes[0].setwalletflag, 'avoid_reuse', False)
121 assert_raises_rpc_error(-8, "Wallet flag is already set to true", self.nodes[1].setwalletflag, 'avoid_reuse', True)
122 123 assert_raises_rpc_error(-8, "Unknown wallet flag: abc", self.nodes[0].setwalletflag, 'abc', True)
124 125 # Create a wallet with avoid reuse, and test that disabling it afterwards persists
126 self.nodes[1].createwallet(wallet_name="avoid_reuse_persist", avoid_reuse=True)
127 w = self.nodes[1].get_wallet_rpc("avoid_reuse_persist")
128 assert_equal(w.getwalletinfo()["avoid_reuse"], True)
129 w.setwalletflag("avoid_reuse", False)
130 assert_equal(w.getwalletinfo()["avoid_reuse"], False)
131 w.unloadwallet()
132 self.nodes[1].loadwallet("avoid_reuse_persist")
133 assert_equal(w.getwalletinfo()["avoid_reuse"], False)
134 w.unloadwallet()
135 136 def test_immutable(self):
137 '''Test immutable wallet flags'''
138 self.log.info("Test immutable wallet flags")
139 140 # Attempt to set the disable_private_keys flag; this should not work
141 assert_raises_rpc_error(-8, "Wallet flag is immutable", self.nodes[1].setwalletflag, 'disable_private_keys')
142 143 tempwallet = ".wallet_avoidreuse.py_test_immutable_wallet.dat"
144 145 # Create a wallet with disable_private_keys set; this should work
146 self.nodes[1].createwallet(wallet_name=tempwallet, disable_private_keys=True)
147 w = self.nodes[1].get_wallet_rpc(tempwallet)
148 149 # Attempt to unset the disable_private_keys flag; this should not work
150 assert_raises_rpc_error(-8, "Wallet flag is immutable", w.setwalletflag, 'disable_private_keys', False)
151 152 # Unload temp wallet
153 self.nodes[1].unloadwallet(tempwallet)
154 155 def test_change_remains_change(self, node):
156 self.log.info("Test that change doesn't turn into non-change when spent")
157 158 reset_balance(node, node.getnewaddress())
159 addr = node.getnewaddress()
160 txid = node.sendtoaddress(addr, 1)
161 out = node.listunspent(minconf=0, query_options={'minimumAmount': 2})
162 assert_equal(len(out), 1)
163 assert_equal(out[0]['txid'], txid)
164 changeaddr = out[0]['address']
165 166 # Make sure it's starting out as change as expected
167 assert node.getaddressinfo(changeaddr)['ischange']
168 for logical_tx in node.listtransactions():
169 assert logical_tx.get('address') != changeaddr
170 171 # Spend it
172 reset_balance(node, node.getnewaddress())
173 174 # It should still be change
175 assert node.getaddressinfo(changeaddr)['ischange']
176 for logical_tx in node.listtransactions():
177 assert logical_tx.get('address') != changeaddr
178 179 def test_sending_from_reused_address_without_avoid_reuse(self):
180 '''
181 Test the same as test_sending_from_reused_address_fails, except send the 10 BTC with
182 the avoid_reuse flag set to false. This means the 10 BTC send should succeed,
183 where it fails in test_sending_from_reused_address_fails.
184 '''
185 self.log.info("Test sending from reused address with avoid_reuse=false")
186 187 fundaddr = self.nodes[1].getnewaddress()
188 retaddr = self.nodes[0].getnewaddress()
189 190 self.nodes[0].sendtoaddress(fundaddr, 10)
191 self.generate(self.nodes[0], 1)
192 193 # listunspent should show 1 single, unused 10 btc output
194 assert_unspent(self.nodes[1], total_count=1, total_sum=10, reused_supported=True, reused_count=0)
195 # getbalances should show no used, 10 btc trusted
196 assert_balances(self.nodes[1], mine={"used": 0, "trusted": 10})
197 # node 0 should not show a used entry, as it does not enable avoid_reuse
198 assert "used" not in self.nodes[0].getbalances()["mine"]
199 200 self.nodes[1].sendtoaddress(retaddr, 5)
201 self.generate(self.nodes[0], 1)
202 203 # listunspent should show 1 single, unused 5 btc output
204 assert_unspent(self.nodes[1], total_count=1, total_sum=5, reused_supported=True, reused_count=0)
205 # getbalances should show no used, 5 btc trusted
206 assert_balances(self.nodes[1], mine={"used": 0, "trusted": 5})
207 208 self.nodes[0].sendtoaddress(fundaddr, 10)
209 self.generate(self.nodes[0], 1)
210 211 # listunspent should show 2 total outputs (5, 10 btc), one unused (5), one reused (10)
212 assert_unspent(self.nodes[1], total_count=2, total_sum=15, reused_count=1, reused_sum=10)
213 # getbalances should show 10 used, 5 btc trusted
214 assert_balances(self.nodes[1], mine={"used": 10, "trusted": 5})
215 216 self.nodes[1].sendtoaddress(address=retaddr, amount=10, avoid_reuse=False)
217 218 # listunspent should show 1 total outputs (5 btc), unused
219 assert_unspent(self.nodes[1], total_count=1, total_sum=5, reused_count=0)
220 # getbalances should show no used, 5 btc trusted
221 assert_balances(self.nodes[1], mine={"used": 0, "trusted": 5})
222 223 # node 1 should now have about 5 btc left (for both cases)
224 assert_approx(self.nodes[1].getbalance(), 5, 0.001)
225 assert_approx(self.nodes[1].getbalance(avoid_reuse=False), 5, 0.001)
226 227 def test_sending_from_reused_address_fails(self, second_addr_type):
228 '''
229 Test the simple case where [1] generates a new address A, then
230 [0] sends 10 BTC to A.
231 [1] spends 5 BTC from A. (leaving roughly 5 BTC useable)
232 [0] sends 10 BTC to A again.
233 [1] tries to spend 10 BTC (fails; dirty).
234 [1] tries to spend 4 BTC (succeeds; change address sufficient)
235 '''
236 self.log.info("Test sending from reused {} address fails".format(second_addr_type))
237 238 fundaddr = self.nodes[1].getnewaddress(label="", address_type="legacy")
239 retaddr = self.nodes[0].getnewaddress()
240 241 self.nodes[0].sendtoaddress(fundaddr, 10)
242 self.generate(self.nodes[0], 1)
243 244 # listunspent should show 1 single, unused 10 btc output
245 assert_unspent(self.nodes[1], total_count=1, total_sum=10, reused_supported=True, reused_count=0)
246 # getbalances should show no used, 10 btc trusted
247 assert_balances(self.nodes[1], mine={"used": 0, "trusted": 10})
248 249 self.nodes[1].sendtoaddress(retaddr, 5)
250 self.generate(self.nodes[0], 1)
251 252 # listunspent should show 1 single, unused 5 btc output
253 assert_unspent(self.nodes[1], total_count=1, total_sum=5, reused_supported=True, reused_count=0)
254 # getbalances should show no used, 5 btc trusted
255 assert_balances(self.nodes[1], mine={"used": 0, "trusted": 5})
256 257 if not self.options.descriptors:
258 # For the second send, we transmute it to a related single-key address
259 # to make sure it's also detected as reuse
260 fund_spk = address_to_scriptpubkey(fundaddr).hex()
261 fund_decoded = self.nodes[0].decodescript(fund_spk)
262 if second_addr_type == "p2sh-segwit":
263 new_fundaddr = fund_decoded["segwit"]["p2sh-segwit"]
264 elif second_addr_type == "bech32":
265 new_fundaddr = fund_decoded["segwit"]["address"]
266 else:
267 new_fundaddr = fundaddr
268 assert_equal(second_addr_type, "legacy")
269 270 self.nodes[0].sendtoaddress(new_fundaddr, 10)
271 self.generate(self.nodes[0], 1)
272 273 # listunspent should show 2 total outputs (5, 10 btc), one unused (5), one reused (10)
274 assert_unspent(self.nodes[1], total_count=2, total_sum=15, reused_count=1, reused_sum=10)
275 # getbalances should show 10 used, 5 btc trusted
276 assert_balances(self.nodes[1], mine={"used": 10, "trusted": 5})
277 278 # node 1 should now have a balance of 5 (no dirty) or 15 (including dirty)
279 assert_approx(self.nodes[1].getbalance(), 5, 0.001)
280 assert_approx(self.nodes[1].getbalance(avoid_reuse=False), 15, 0.001)
281 282 assert_raises_rpc_error(-6, "Insufficient funds", self.nodes[1].sendtoaddress, retaddr, 10)
283 284 self.nodes[1].sendtoaddress(retaddr, 4)
285 286 # listunspent should show 2 total outputs (1, 10 btc), one unused (1), one reused (10)
287 assert_unspent(self.nodes[1], total_count=2, total_sum=11, reused_count=1, reused_sum=10)
288 # getbalances should show 10 used, 1 btc trusted
289 assert_balances(self.nodes[1], mine={"used": 10, "trusted": 1})
290 291 # node 1 should now have about 1 btc left (no dirty) and 11 (including dirty)
292 assert_approx(self.nodes[1].getbalance(), 1, 0.001)
293 assert_approx(self.nodes[1].getbalance(avoid_reuse=False), 11, 0.001)
294 295 def test_getbalances_used(self):
296 '''
297 getbalances and listunspent should pick up on reused addresses
298 immediately, even for address reusing outputs created before the first
299 transaction was spending from that address
300 '''
301 self.log.info("Test getbalances used category")
302 303 # node under test should be completely empty
304 assert_equal(self.nodes[1].getbalance(avoid_reuse=False), 0)
305 306 new_addr = self.nodes[1].getnewaddress()
307 ret_addr = self.nodes[0].getnewaddress()
308 309 # send multiple transactions, reusing one address
310 for _ in range(101):
311 self.nodes[0].sendtoaddress(new_addr, 1)
312 313 self.generate(self.nodes[0], 1)
314 315 # send transaction that should not use all the available outputs
316 # per the current coin selection algorithm
317 self.nodes[1].sendtoaddress(ret_addr, 5)
318 319 # getbalances and listunspent should show the remaining outputs
320 # in the reused address as used/reused
321 assert_unspent(self.nodes[1], total_count=2, total_sum=96, reused_count=1, reused_sum=1, margin=0.01)
322 assert_balances(self.nodes[1], mine={"used": 1, "trusted": 95}, margin=0.01)
323 324 def test_full_destination_group_is_preferred(self):
325 '''
326 Test the case where [1] only has 101 outputs of 1 BTC in the same reused
327 address and tries to send a small payment of 0.5 BTC. The wallet
328 should use 100 outputs from the reused address as inputs and not a
329 single 1 BTC input, in order to join several outputs from the reused
330 address.
331 '''
332 self.log.info("Test that full destination groups are preferred in coin selection")
333 334 # Node under test should be empty
335 assert_equal(self.nodes[1].getbalance(avoid_reuse=False), 0)
336 337 new_addr = self.nodes[1].getnewaddress()
338 ret_addr = self.nodes[0].getnewaddress()
339 340 # Send 101 outputs of 1 BTC to the same, reused address in the wallet
341 for _ in range(101):
342 self.nodes[0].sendtoaddress(new_addr, 1)
343 344 self.generate(self.nodes[0], 1)
345 346 # Sending a transaction that is smaller than each one of the
347 # available outputs
348 txid = self.nodes[1].sendtoaddress(address=ret_addr, amount=0.5)
349 inputs = self.nodes[1].getrawtransaction(txid, 1)["vin"]
350 351 # The transaction should use 100 inputs exactly
352 assert_equal(len(inputs), 100)
353 354 def test_all_destination_groups_are_used(self):
355 '''
356 Test the case where [1] only has 202 outputs of 1 BTC in the same reused
357 address and tries to send a payment of 200.5 BTC. The wallet
358 should use all 202 outputs from the reused address as inputs.
359 '''
360 self.log.info("Test that all destination groups are used")
361 362 # Node under test should be empty
363 assert_equal(self.nodes[1].getbalance(avoid_reuse=False), 0)
364 365 new_addr = self.nodes[1].getnewaddress()
366 ret_addr = self.nodes[0].getnewaddress()
367 368 # Send 202 outputs of 1 BTC to the same, reused address in the wallet
369 for _ in range(202):
370 self.nodes[0].sendtoaddress(new_addr, 1)
371 372 self.generate(self.nodes[0], 1)
373 374 # Sending a transaction that needs to use the full groups
375 # of 100 inputs but also the incomplete group of 2 inputs.
376 txid = self.nodes[1].sendtoaddress(address=ret_addr, amount=200.5)
377 inputs = self.nodes[1].getrawtransaction(txid, 1)["vin"]
378 379 # The transaction should use 202 inputs exactly
380 assert_equal(len(inputs), 202)
381 382 383 if __name__ == '__main__':
384 AvoidReuseTest(__file__).main()
385