Skip to content
Bitcoin Roots

v30.3-roots.1 · Commit 23: [PATCH 23/68] feat(wallet): add transient private-key sweep RPC

test/functional/wallet_sweepprivkeys.py

Before: /dev/null · After: test/functional/wallet_sweepprivkeys.py

File metadata
new file mode 100644
index 0000000000000000000000000000000000000000..b9c940b7cb566da373f74439311bb611d042d918

Download original patch · Release notes

Diff rows 1–144 of 144. Line numbers refer to the original before/after files.

+ Added− RemovedUnmarked lines are context.

Scroll the code area horizontally to read long lines. The two line-number columns refer to before and after.

test/functional/wallet_sweepprivkeys.py ADDED
@@ -0,0 +1,144 @@
1
+ #!/usr/bin/env python3
2
+ """Exercise transient-key sweep preview and broadcast without import."""
3
+ from decimal import Decimal
4
+ from pathlib import Path
5
+
6
+ from test_framework.address import key_to_p2pkh, key_to_p2sh_p2wpkh, key_to_p2wpkh
7
+ from test_framework.descriptors import descsum_create
8
+ from test_framework.test_framework import BitcoinTestFramework
9
+ from test_framework.util import assert_equal, assert_raises_rpc_error
10
+ from test_framework.wallet_util import generate_keypair
11
+
12
+ COIN = Decimal(100_000_000)
13
+
14
+ class SweepPrivKeysTest(BitcoinTestFramework):
15
+ def set_test_params(self):
16
+ self.num_nodes = 2
17
+ self.extra_args = [['-deprecatedrpc=settxfee'], []]
18
+
19
+ def skip_test_if_missing_module(self):
20
+ self.skip_if_no_wallet()
21
+
22
+ def run_test(self):
23
+ node, miner = self.nodes
24
+ wallet = node.get_wallet_rpc(node.listwallets()[0])
25
+ destination = wallet.getnewaddress()
26
+ wif, pubkey = generate_keypair(wif=True)
27
+ uncompressed_wif, uncompressed_pubkey = generate_keypair(compressed=False, wif=True)
28
+ race_wif, race_pubkey = generate_keypair(wif=True)
29
+ empty_wif, _ = generate_keypair(wif=True)
30
+ self.generate(miner, 101)
31
+ assert_raises_rpc_error(-6, 'No spendable UTXOs found', wallet.sweepprivkeys, [empty_wif], destination)
32
+ for address in (key_to_p2pkh(pubkey), key_to_p2wpkh(pubkey), key_to_p2sh_p2wpkh(pubkey), key_to_p2pkh(uncompressed_pubkey)):
33
+ miner.sendtoaddress(address, 1)
34
+ race_address = key_to_p2pkh(race_pubkey)
35
+ race_txid = miner.sendtoaddress(race_address, 1)
36
+ self.generate(miner, 1)
37
+ race_tx = miner.decoderawtransaction(miner.gettransaction(race_txid)['hex'])
38
+ race_vout = next(vout for vout in race_tx['vout'] if vout['scriptPubKey'].get('address') == race_address)
39
+ race_prevout = {'txid': race_txid, 'vout': race_vout['n'], 'scriptPubKey': race_vout['scriptPubKey']['hex'], 'amount': race_vout['value']}
40
+ race_raw = miner.createrawtransaction([race_prevout], {miner.getnewaddress(): Decimal('0.99990000')})
41
+ race_signed = miner.signrawtransactionwithkey(race_raw, [race_wif], [race_prevout])
42
+ assert_equal(race_signed['complete'], True)
43
+ miner.sendrawtransaction(race_signed['hex'])
44
+ race_block = self.generate(miner, 1)[0]
45
+ self.sync_blocks()
46
+ assert_raises_rpc_error(-6, 'No spendable UTXOs found', wallet.sweepprivkeys, [race_wif], destination)
47
+ node.invalidateblock(race_block)
48
+ mempool_before_preview = wallet.getrawmempool()
49
+ stale_preview = wallet.sweepprivkeys([race_wif], destination)
50
+ assert_equal(stale_preview['broadcast'], False)
51
+ assert_equal(stale_preview['inputs'], 1)
52
+ assert_equal(wallet.getrawmempool(), mempool_before_preview)
53
+ node.reconsiderblock(race_block)
54
+ assert_raises_rpc_error(-6, 'No spendable UTXOs found', wallet.sweepprivkeys, [race_wif], destination)
55
+ conflict_wif, conflict_pubkey = generate_keypair(wif=True)
56
+ conflict_address = key_to_p2pkh(conflict_pubkey)
57
+ conflict_txid = miner.sendtoaddress(conflict_address, 1)
58
+ self.generate(miner, 1)
59
+ self.sync_blocks()
60
+ conflict_tx = miner.decoderawtransaction(miner.gettransaction(conflict_txid)['hex'])
61
+ conflict_vout = next(v for v in conflict_tx['vout'] if v['scriptPubKey'].get('address') == conflict_address)
62
+ prevout = {'txid': conflict_txid, 'vout': conflict_vout['n'], 'scriptPubKey': conflict_vout['scriptPubKey']['hex'], 'amount': conflict_vout['value']}
63
+ raw = miner.createrawtransaction([prevout], {miner.getnewaddress(): Decimal('0.99990000')})
64
+ competing = miner.signrawtransactionwithkey(raw, [conflict_wif], [prevout])
65
+ conflict_id = miner.sendrawtransaction(competing['hex'])
66
+ self.sync_mempools()
67
+ rejected_state = wallet.listdescriptors(), wallet.listlabels(), wallet.getwalletinfo()['keypoolsize']
68
+ assert_raises_rpc_error(-4, 'Sweep transaction broadcast failed: insufficient fee, rejecting replacement', wallet.sweepprivkeys, [conflict_wif], destination, True)
69
+ assert_equal(wallet.getrawmempool(), [conflict_id])
70
+ assert_equal((wallet.listdescriptors(), wallet.listlabels(), wallet.getwalletinfo()['keypoolsize']), rejected_state)
71
+ assert_equal(wallet.getaddressinfo(conflict_address)['ismine'], False)
72
+ descriptors = wallet.listdescriptors()
73
+ labels = wallet.listlabels()
74
+ keypoolsize = wallet.getwalletinfo()['keypoolsize']
75
+
76
+ assert_raises_rpc_error(-5, 'Invalid private key', wallet.sweepprivkeys, ['invalid'], destination)
77
+ assert_raises_rpc_error(-5, 'Invalid private key', wallet.sweepprivkeys, ['KwDiBf89QgGbjEhKnhXJuH7LrciVrZi3qYjgd9M7rFU73sVHnoWn'], destination)
78
+ assert_raises_rpc_error(-8, 'Duplicate private key', wallet.sweepprivkeys, [wif, wif], destination)
79
+ assert_raises_rpc_error(-4, 'Destination is not controlled', wallet.sweepprivkeys, [wif], miner.getnewaddress())
80
+ node.createwallet(wallet_name='other')
81
+ other = node.get_wallet_rpc('other')
82
+ assert_raises_rpc_error(-4, 'Destination is not controlled', wallet.sweepprivkeys, [wif], other.getnewaddress())
83
+ node.createwallet(wallet_name='watch', disable_private_keys=True)
84
+ watch = node.get_wallet_rpc('watch')
85
+ watch.importdescriptors([{'desc': descsum_create(f"wpkh({wallet.getaddressinfo(destination)['pubkey']})"), 'timestamp': 'now'}])
86
+ assert_raises_rpc_error(-4, 'Destination is not controlled', watch.sweepprivkeys, [wif], destination)
87
+
88
+ wallet.settxfee(Decimal('0.00001000'))
89
+ preview = wallet.sweepprivkeys(privkeys=[wif, uncompressed_wif], destination=destination)
90
+ assert_equal(preview['broadcast'], False)
91
+ assert_equal(preview['inputs'], 4)
92
+ decoded = node.decoderawtransaction(preview['hex'])
93
+ required_fee = Decimal(decoded['vsize']) / COIN
94
+ assert preview['fee'] >= required_fee
95
+ assert preview['fee'] - required_fee <= Decimal('0.00000001')
96
+ assert_equal(preview['amount'] - sum(output['value'] for output in decoded['vout']), preview['fee'])
97
+ assert_equal(wallet.listdescriptors(), descriptors)
98
+ assert_equal(wallet.listlabels(), labels)
99
+ assert_equal(wallet.getwalletinfo()['keypoolsize'], keypoolsize)
100
+ assert_equal(wallet.getaddressinfo(key_to_p2pkh(pubkey))['ismine'], False)
101
+ assert_equal(wallet.getrawmempool(), [conflict_id])
102
+
103
+ wallet.encryptwallet('pass')
104
+ sent = wallet.sweepprivkeys([wif, uncompressed_wif], destination, True)
105
+ assert_equal(sent['broadcast'], True)
106
+ assert_equal(sent['txid'], preview['txid'])
107
+ assert sent['txid'] in wallet.getrawmempool()
108
+ rejected_descriptors = wallet.listdescriptors()
109
+ rejected_labels = wallet.listlabels()
110
+ rejected_keypoolsize = wallet.getwalletinfo()['keypoolsize']
111
+ duplicate = wallet.sweepprivkeys([wif, uncompressed_wif], destination, True)
112
+ assert_equal(duplicate['txid'], sent['txid'])
113
+ assert_equal(set(wallet.getrawmempool()), {conflict_id, sent['txid']})
114
+ assert_equal(wallet.listdescriptors(), rejected_descriptors)
115
+ assert_equal(wallet.listlabels(), rejected_labels)
116
+ assert_equal(wallet.getwalletinfo()['keypoolsize'], rejected_keypoolsize)
117
+ assert_equal(wallet.getaddressinfo(key_to_p2pkh(pubkey))['ismine'], False)
118
+ self.sync_mempools()
119
+ self.generate(miner, 1)
120
+ self.sync_blocks()
121
+ assert_raises_rpc_error(-6, 'No spendable UTXOs found', wallet.sweepprivkeys, [wif, uncompressed_wif], destination)
122
+
123
+ small_wif, small_pubkey = generate_keypair(wif=True)
124
+ miner.sendtoaddress(key_to_p2pkh(small_pubkey), Decimal('0.00010000'))
125
+ self.generate(miner, 1)
126
+ self.sync_blocks()
127
+ dust_destination = wallet.getnewaddress(address_type='legacy')
128
+ wallet.settxfee(Decimal('0.00052000'))
129
+ assert_raises_rpc_error(-4, 'Swept output would be dust', wallet.sweepprivkeys, [small_wif], dust_destination)
130
+ wallet.settxfee(Decimal('0.10000000'))
131
+ assert_raises_rpc_error(-6, 'Fee exceeds swept value', wallet.sweepprivkeys, [small_wif], destination)
132
+ assert wif not in Path(self.options.tmpdir, 'node0', 'regtest', 'debug.log').read_text()
133
+ wallet_name = wallet.getwalletinfo()['walletname']
134
+ self.restart_node(0)
135
+ wallet = self.nodes[0].get_wallet_rpc(wallet_name)
136
+ assert_equal(wallet.getaddressinfo(key_to_p2pkh(pubkey))['ismine'], False)
137
+ assert wif not in Path(self.options.tmpdir, 'node0', 'regtest', 'debug.log').read_text()
138
+ self.nodes[0].unloadwallet(wallet_name)
139
+ assert wallet_name not in self.nodes[0].listwallets()
140
+ self.nodes[0].loadwallet(wallet_name)
141
+ wallet = self.nodes[0].get_wallet_rpc(wallet_name)
142
+ assert_equal(wallet.getaddressinfo(key_to_p2pkh(pubkey))['ismine'], False)
143
+
144
+ if __name__ == '__main__': SweepPrivKeysTest(__file__).main()