1 #!/usr/bin/env python3
2 # Copyright (c) 2018-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 the avoid_reuse and setwalletflag features."""
6 7 from test_framework.test_framework import BitcoinTestFramework
8 from test_framework.util import (
9 assert_not_equal,
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(BitcoinTestFramework):
67 def set_test_params(self):
68 self.num_nodes = 2
69 # whitelist peers to speed up tx relay / mempool sync
70 self.noban_tx_relay = True
71 72 def skip_test_if_missing_module(self):
73 self.skip_if_no_wallet()
74 75 def run_test(self):
76 '''Set up initial chain and run tests defined below'''
77 78 self.test_persistence()
79 self.test_immutable()
80 81 self.generate(self.nodes[0], 110)
82 self.test_change_remains_change(self.nodes[1])
83 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
84 self.test_sending_from_reused_address_without_avoid_reuse()
85 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
86 self.test_sending_from_reused_address_fails("legacy")
87 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
88 self.test_sending_from_reused_address_fails("p2sh-segwit")
89 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
90 self.test_sending_from_reused_address_fails("bech32")
91 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
92 self.test_getbalances_used()
93 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
94 self.test_full_destination_group_is_preferred()
95 reset_balance(self.nodes[1], self.nodes[0].getnewaddress())
96 self.test_all_destination_groups_are_used()
97 98 def test_persistence(self):
99 '''Test that wallet files persist the avoid_reuse flag.'''
100 self.log.info("Test wallet files persist avoid_reuse flag")
101 102 # Configure node 1 to use avoid_reuse
103 self.nodes[1].setwalletflag('avoid_reuse')
104 105 # Flags should be node1.avoid_reuse=false, node2.avoid_reuse=true
106 assert_equal(self.nodes[0].getwalletinfo()["avoid_reuse"], False)
107 assert_equal(sorted(self.nodes[0].getwalletinfo()["flags"]), sorted(["descriptor_wallet", "last_hardened_xpub_cached"]))
108 assert_equal(self.nodes[1].getwalletinfo()["avoid_reuse"], True)
109 assert_equal(sorted(self.nodes[1].getwalletinfo()["flags"]), sorted(["descriptor_wallet", "last_hardened_xpub_cached", "avoid_reuse"]))
110 111 self.restart_node(1)
112 self.connect_nodes(0, 1)
113 114 # Flags should still be node1.avoid_reuse=false, node2.avoid_reuse=true
115 assert_equal(self.nodes[0].getwalletinfo()["avoid_reuse"], False)
116 assert_equal(self.nodes[1].getwalletinfo()["avoid_reuse"], True)
117 118 # Attempting to set flag to its current state should throw
119 assert_raises_rpc_error(-8, "Wallet flag is already set to false", self.nodes[0].setwalletflag, 'avoid_reuse', False)
120 assert_raises_rpc_error(-8, "Wallet flag is already set to true", self.nodes[1].setwalletflag, 'avoid_reuse', True)
121 122 assert_raises_rpc_error(-8, "Unknown wallet flag: abc", self.nodes[0].setwalletflag, 'abc', True)
123 124 # Create a wallet with avoid reuse, and test that disabling it afterwards persists
125 self.nodes[1].createwallet(wallet_name="avoid_reuse_persist", avoid_reuse=True)
126 w = self.nodes[1].get_wallet_rpc("avoid_reuse_persist")
127 assert_equal(w.getwalletinfo()["avoid_reuse"], True)
128 w.setwalletflag("avoid_reuse", False)
129 assert_equal(w.getwalletinfo()["avoid_reuse"], False)
130 w.unloadwallet()
131 self.nodes[1].loadwallet("avoid_reuse_persist")
132 assert_equal(w.getwalletinfo()["avoid_reuse"], False)
133 w.unloadwallet()
134 135 def test_immutable(self):
136 '''Test immutable wallet flags'''
137 self.log.info("Test immutable wallet flags")
138 139 # Attempt to set the disable_private_keys flag; this should not work
140 assert_raises_rpc_error(-8, "Wallet flag is immutable", self.nodes[1].setwalletflag, 'disable_private_keys')
141 142 tempwallet = ".wallet_avoidreuse.py_test_immutable_wallet.dat"
143 144 # Create a wallet with disable_private_keys set; this should work
145 self.nodes[1].createwallet(wallet_name=tempwallet, disable_private_keys=True)
146 w = self.nodes[1].get_wallet_rpc(tempwallet)
147 148 # Attempt to unset the disable_private_keys flag; this should not work
149 assert_raises_rpc_error(-8, "Wallet flag is immutable", w.setwalletflag, 'disable_private_keys', False)
150 151 # Unload temp wallet
152 self.nodes[1].unloadwallet(tempwallet)
153 154 def test_change_remains_change(self, node):
155 self.log.info("Test that change doesn't turn into non-change when spent")
156 157 reset_balance(node, node.getnewaddress())
158 addr = node.getnewaddress()
159 txid = node.sendtoaddress(addr, 1)
160 out = node.listunspent(minconf=0, query_options={'minimumAmount': 2})
161 assert_equal(len(out), 1)
162 assert_equal(out[0]['txid'], txid)
163 changeaddr = out[0]['address']
164 165 # Make sure it's starting out as change as expected
166 assert node.getaddressinfo(changeaddr)['ischange']
167 for logical_tx in node.listtransactions():
168 assert_not_equal(logical_tx.get('address'), changeaddr)
169 170 # Spend it
171 reset_balance(node, node.getnewaddress())
172 173 # It should still be change
174 assert node.getaddressinfo(changeaddr)['ischange']
175 for logical_tx in node.listtransactions():
176 assert_not_equal(logical_tx.get('address'), changeaddr)
177 178 def test_sending_from_reused_address_without_avoid_reuse(self):
179 '''
180 Test the same as test_sending_from_reused_address_fails, except send the 10 BTC with
181 the avoid_reuse flag set to false. This means the 10 BTC send should succeed,
182 where it fails in test_sending_from_reused_address_fails.
183 '''
184 self.log.info("Test sending from reused address with avoid_reuse=false")
185 186 fundaddr = self.nodes[1].getnewaddress()
187 retaddr = self.nodes[0].getnewaddress()
188 189 self.nodes[0].sendtoaddress(fundaddr, 10)
190 self.generate(self.nodes[0], 1)
191 192 # listunspent should show 1 single, unused 10 btc output
193 assert_unspent(self.nodes[1], total_count=1, total_sum=10, reused_supported=True, reused_count=0)
194 # getbalances should show no used, 10 btc trusted
195 assert_balances(self.nodes[1], mine={"used": 0, "trusted": 10})
196 # node 0 should not show a used entry, as it does not enable avoid_reuse
197 assert "used" not in self.nodes[0].getbalances()["mine"]
198 199 self.nodes[1].sendtoaddress(retaddr, 5)
200 self.generate(self.nodes[0], 1)
201 202 # listunspent should show 1 single, unused 5 btc output
203 assert_unspent(self.nodes[1], total_count=1, total_sum=5, reused_supported=True, reused_count=0)
204 # getbalances should show no used, 5 btc trusted
205 assert_balances(self.nodes[1], mine={"used": 0, "trusted": 5})
206 207 self.nodes[0].sendtoaddress(fundaddr, 10)
208 self.generate(self.nodes[0], 1)
209 210 # listunspent should show 2 total outputs (5, 10 btc), one unused (5), one reused (10)
211 assert_unspent(self.nodes[1], total_count=2, total_sum=15, reused_count=1, reused_sum=10)
212 # getbalances should show 10 used, 5 btc trusted
213 assert_balances(self.nodes[1], mine={"used": 10, "trusted": 5})
214 215 self.nodes[1].sendtoaddress(address=retaddr, amount=10, avoid_reuse=False)
216 217 # listunspent should show 1 total outputs (5 btc), unused
218 assert_unspent(self.nodes[1], total_count=1, total_sum=5, reused_count=0)
219 # getbalances should show no used, 5 btc trusted
220 assert_balances(self.nodes[1], mine={"used": 0, "trusted": 5})
221 222 # node 1 should now have about 5 btc left (for both cases)
223 assert_approx(self.nodes[1].getbalance(), 5, 0.001)
224 assert_approx(self.nodes[1].getbalance(avoid_reuse=False), 5, 0.001)
225 226 def test_sending_from_reused_address_fails(self, second_addr_type):
227 '''
228 Test the simple case where [1] generates a new address A, then
229 [0] sends 10 BTC to A.
230 [1] spends 5 BTC from A. (leaving roughly 5 BTC useable)
231 [0] sends 10 BTC to A again.
232 [1] tries to spend 10 BTC (fails; dirty).
233 [1] tries to spend 4 BTC (succeeds; change address sufficient)
234 '''
235 self.log.info("Test sending from reused {} address fails".format(second_addr_type))
236 237 fundaddr = self.nodes[1].getnewaddress(label="", address_type="legacy")
238 retaddr = self.nodes[0].getnewaddress()
239 240 self.nodes[0].sendtoaddress(fundaddr, 10)
241 self.generate(self.nodes[0], 1)
242 243 # listunspent should show 1 single, unused 10 btc output
244 assert_unspent(self.nodes[1], total_count=1, total_sum=10, reused_supported=True, reused_count=0)
245 # getbalances should show no used, 10 btc trusted
246 assert_balances(self.nodes[1], mine={"used": 0, "trusted": 10})
247 248 self.nodes[1].sendtoaddress(retaddr, 5)
249 self.generate(self.nodes[0], 1)
250 251 # listunspent should show 1 single, unused 5 btc output
252 assert_unspent(self.nodes[1], total_count=1, total_sum=5, reused_supported=True, reused_count=0)
253 # getbalances should show no used, 5 btc trusted
254 assert_balances(self.nodes[1], mine={"used": 0, "trusted": 5})
255 256 def test_getbalances_used(self):
257 '''
258 getbalances and listunspent should pick up on reused addresses
259 immediately, even for address reusing outputs created before the first
260 transaction was spending from that address
261 '''
262 self.log.info("Test getbalances used category")
263 264 # node under test should be completely empty
265 assert_equal(self.nodes[1].getbalance(avoid_reuse=False), 0)
266 267 new_addr = self.nodes[1].getnewaddress()
268 ret_addr = self.nodes[0].getnewaddress()
269 270 # send multiple transactions, reusing one address
271 for _ in range(101):
272 self.nodes[0].sendtoaddress(new_addr, 1)
273 274 self.generate(self.nodes[0], 1)
275 276 # send transaction that should not use all the available outputs
277 # per the current coin selection algorithm
278 self.nodes[1].sendtoaddress(ret_addr, 5)
279 280 # getbalances and listunspent should show the remaining outputs
281 # in the reused address as used/reused
282 assert_unspent(self.nodes[1], total_count=2, total_sum=96, reused_count=1, reused_sum=1, margin=0.01)
283 assert_balances(self.nodes[1], mine={"used": 1, "trusted": 95}, margin=0.01)
284 285 def test_full_destination_group_is_preferred(self):
286 '''
287 Test the case where [1] only has 101 outputs of 1 BTC in the same reused
288 address and tries to send a small payment of 0.5 BTC. The wallet
289 should use 100 outputs from the reused address as inputs and not a
290 single 1 BTC input, in order to join several outputs from the reused
291 address.
292 '''
293 self.log.info("Test that full destination groups are preferred in coin selection")
294 295 # Node under test should be empty
296 assert_equal(self.nodes[1].getbalance(avoid_reuse=False), 0)
297 298 new_addr = self.nodes[1].getnewaddress()
299 ret_addr = self.nodes[0].getnewaddress()
300 301 # Send 101 outputs of 1 BTC to the same, reused address in the wallet
302 for _ in range(101):
303 self.nodes[0].sendtoaddress(new_addr, 1)
304 305 self.generate(self.nodes[0], 1)
306 307 # Sending a transaction that is smaller than each one of the
308 # available outputs
309 txid = self.nodes[1].sendtoaddress(address=ret_addr, amount=0.5)
310 inputs = self.nodes[1].getrawtransaction(txid, 1)["vin"]
311 312 # The transaction should use 100 inputs exactly
313 assert_equal(len(inputs), 100)
314 315 def test_all_destination_groups_are_used(self):
316 '''
317 Test the case where [1] only has 202 outputs of 1 BTC in the same reused
318 address and tries to send a payment of 200.5 BTC. The wallet
319 should use all 202 outputs from the reused address as inputs.
320 '''
321 self.log.info("Test that all destination groups are used")
322 323 # Node under test should be empty
324 assert_equal(self.nodes[1].getbalance(avoid_reuse=False), 0)
325 326 new_addr = self.nodes[1].getnewaddress()
327 ret_addr = self.nodes[0].getnewaddress()
328 329 # Send 202 outputs of 1 BTC to the same, reused address in the wallet
330 for _ in range(202):
331 self.nodes[0].sendtoaddress(new_addr, 1)
332 333 self.generate(self.nodes[0], 1)
334 335 # Sending a transaction that needs to use the full groups
336 # of 100 inputs but also the incomplete group of 2 inputs.
337 txid = self.nodes[1].sendtoaddress(address=ret_addr, amount=200.5)
338 inputs = self.nodes[1].getrawtransaction(txid, 1)["vin"]
339 340 # The transaction should use 202 inputs exactly
341 assert_equal(len(inputs), 202)
342 343 344 if __name__ == '__main__':
345 AvoidReuseTest(__file__).main()
346