wallet_exported_watchonly.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2025-present The Bitcoin Core developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or https://www.opensource.org/licenses/mit-license.php.
5
6 import os
7
8 from test_framework.descriptors import descsum_create
9 from test_framework.key import H_POINT
10 from test_framework.test_framework import BitcoinTestFramework
11 from test_framework.util import (
12 assert_equal,
13 assert_not_equal,
14 assert_raises_rpc_error,
15 )
16 from test_framework.wallet_util import generate_keypair
17
18 KEYPOOL_SIZE = 10
19
20 class WalletExportedWatchOnly(BitcoinTestFramework):
21 def set_test_params(self):
22 self.setup_clean_chain = True
23 self.num_nodes = 2
24 self.extra_args = [[], [f"-keypool={KEYPOOL_SIZE}"]]
25
26 def setup_network(self):
27 # Setup the nodes but don't connect them to each other
28 self.setup_nodes()
29
30 def skip_test_if_missing_module(self):
31 self.skip_if_no_wallet()
32
33 def export_and_restore(self, wallet, export_name):
34 export_path = os.path.join(self.export_path, f"{export_name}.dat")
35 res = wallet.exportwatchonlywallet(export_path)
36 assert_equal(res["exported_file"], export_path)
37 self.online.restorewallet(export_name, res["exported_file"])
38 return self.online.get_wallet_rpc(export_name)
39
40 def test_basic_export(self):
41 self.log.info("Test basic watchonly wallet export")
42 self.offline.createwallet("basic")
43 offline_wallet = self.offline.get_wallet_rpc("basic")
44
45 # Bad RPC args
46 assert_raises_rpc_error(-4, "Error: Export ", offline_wallet.exportwatchonlywallet, "")
47 assert_raises_rpc_error(-4, "Error: Export destination '.' already exists", offline_wallet.exportwatchonlywallet, ".")
48 assert_raises_rpc_error(-4, f"Error: Export destination '{self.export_path}' already exists", offline_wallet.exportwatchonlywallet, self.export_path)
49
50 # Export the watchonly wallet file and load onto online node
51 online_wallet = self.export_and_restore(offline_wallet, "basic_watchonly")
52
53 # Exporting watchonly from a watchonly also works
54 online_wallet2 = self.export_and_restore(online_wallet, "basic_watchonly2")
55
56 # Verify that the wallets have the same descriptors
57 addr = offline_wallet.getnewaddress()
58 assert_equal(addr, online_wallet.getnewaddress())
59 assert_equal(addr, online_wallet2.getnewaddress())
60 assert_equal(offline_wallet.listdescriptors()["descriptors"], online_wallet.listdescriptors()["descriptors"])
61 assert_equal(offline_wallet.listdescriptors()["descriptors"], online_wallet2.listdescriptors()["descriptors"])
62
63 # Verify that online wallet cannot spend, but offline can
64 self.funder.sendtoaddress(online_wallet.getnewaddress(), 10)
65 self.generate(self.online, 1, sync_fun=self.no_op)
66 assert_equal(online_wallet.getbalances()["mine"]["trusted"], 10)
67 assert_equal(offline_wallet.getbalances()["mine"]["trusted"], 0)
68 funds_addr = self.funder.getnewaddress()
69 send_res = online_wallet.send([{funds_addr: 5}])
70 assert_equal(send_res["complete"], False)
71 assert "psbt" in send_res
72 signed_psbt = offline_wallet.walletprocesspsbt(send_res["psbt"])["psbt"]
73 finalized = self.online.finalizepsbt(signed_psbt)["hex"]
74 self.online.sendrawtransaction(finalized)
75
76 # Verify that the change address is known to both wallets
77 dec_tx = self.online.decoderawtransaction(finalized)
78 for txout in dec_tx["vout"]:
79 if txout["scriptPubKey"]["address"] == funds_addr:
80 continue
81 assert_equal(online_wallet.getaddressinfo(txout["scriptPubKey"]["address"])["ismine"], True)
82 assert_equal(offline_wallet.getaddressinfo(txout["scriptPubKey"]["address"])["ismine"], True)
83
84 # Both wallets should agree on addresses generated past the end of the keypool
85 offline_wallet.getnewaddress()
86 offline_wallet.getrawchangeaddress()
87 for _ in range(KEYPOOL_SIZE * 2):
88 assert_equal(online_wallet.getnewaddress(), offline_wallet.getnewaddress())
89 assert_equal(online_wallet.getrawchangeaddress(), offline_wallet.getrawchangeaddress())
90
91 self.generate(self.online, 1, sync_fun=self.no_op)
92 offline_wallet.unloadwallet()
93 online_wallet.unloadwallet()
94
95 def test_export_with_address_book(self):
96 self.log.info("Test all address book entries appear in the exported wallet")
97 self.offline.createwallet("addrbook")
98 offline_wallet = self.offline.get_wallet_rpc("addrbook")
99
100 # Create some address book entries
101 receive_addr = offline_wallet.getnewaddress(label="addrbook_receive")
102 send_addr = self.funder.getnewaddress()
103 offline_wallet.setlabel(send_addr, "addrbook_send") # Sets purpose "send"
104
105 # Export the watchonly wallet file and load onto online node
106 online_wallet = self.export_and_restore(offline_wallet, "addrbook_watchonly")
107
108 # Verify the labels are in both wallets
109 for wallet in [online_wallet, offline_wallet]:
110 for purpose in ["receive", "send"]:
111 label = f"addrbook_{purpose}"
112 assert_equal(wallet.listlabels(purpose), [label])
113 addr = send_addr if purpose == "send" else receive_addr
114 assert_equal(wallet.getaddressesbylabel(label), {addr: {"purpose": purpose}})
115
116 offline_wallet.unloadwallet()
117 online_wallet.unloadwallet()
118
119 def test_export_with_txs_and_locked_coins(self):
120 self.log.info("Test all transactions and locked coins appear in the exported wallet")
121 self.offline.createwallet("txs")
122 offline_wallet = self.offline.get_wallet_rpc("txs")
123
124 # In order to make transactions in the offline wallet, briefly connect offline to online
125 self.connect_nodes(self.offline.index, self.online.index)
126 txids = [self.funder.sendtoaddress(offline_wallet.getnewaddress("funds"), i) for i in range(1, 4)]
127 self.generate(self.online, 1)
128 self.disconnect_nodes(self.offline.index, self.online.index)
129
130 # lock some coins
131 utxos = offline_wallet.listunspent()
132 persistent_lock = [{"txid": utxos[0]["txid"], "vout": utxos[0]["vout"]}]
133 temp_lock = [{"txid": utxos[1]["txid"], "vout": utxos[1]["vout"]}]
134 offline_wallet.lockunspent(unlock=False, transactions=persistent_lock, persistent=True)
135 offline_wallet.lockunspent(unlock=False, transactions=temp_lock, persistent=False)
136
137 # Export the watchonly wallet file and load onto online node
138 # There should not be a rescan as the nodes were at the same height
139 with self.online.assert_debug_log(expected_msgs=[], unexpected_msgs=["Rescanning last"]):
140 online_wallet = self.export_and_restore(offline_wallet, "txs_watchonly")
141
142 # Verify the transactions are in both wallets
143 for txid in txids:
144 assert_equal(online_wallet.gettransaction(txid), offline_wallet.gettransaction(txid))
145
146 # Verify the transactions are all output by listtransactions, in the same order
147 assert_equal(online_wallet.listtransactions(), offline_wallet.listtransactions())
148
149 # New transactions should be the newest (last) transaction in listtransactions
150 new_txid = self.funder.sendtoaddress(online_wallet.getnewaddress(), 1)
151 online_wallet.syncwithvalidationinterfacequeue()
152 assert_equal(online_wallet.listtransactions()[-1]["txid"], new_txid)
153
154 # Verify that the persistent locked coin is locked in both wallets
155 assert_equal(online_wallet.listlockunspent(), persistent_lock)
156 assert_equal(sorted(offline_wallet.listlockunspent(), key=lambda x: x["txid"]), sorted(persistent_lock + temp_lock, key=lambda x: x["txid"]))
157
158 offline_wallet.unloadwallet()
159 online_wallet.unloadwallet()
160
161 def test_export_imported_descriptors(self):
162 self.log.info("Test imported descriptors are exported to the watchonly wallet")
163 self.offline.createwallet("imports")
164 offline_wallet = self.offline.get_wallet_rpc("imports")
165
166 import_res = offline_wallet.importdescriptors(
167 [
168 # A single key, non-ranged
169 {"desc": descsum_create(f"pkh({generate_keypair(wif=True)[0]})"), "timestamp": "now"},
170 # hardened derivation
171 {"desc": descsum_create("sh(wpkh(tprv8ZgxMBicQKsPeuVhWwi6wuMQGfPKi9Li5GtX35jVNknACgqe3CY4g5xgkfDDJcmtF7o1QnxWDRYw4H5P26PXq7sbcUkEqeR4fg3Kxp2tigg/0'/*'))"), "timestamp": "now", "active": True},
172 # multisig
173 {"desc": descsum_create("wsh(multi(1,tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/*,tprv8ZgxMBicQKsPeuVhWwi6wuMQGfPKi9Li5GtX35jVNknACgqe3CY4g5xgkfDDJcmtF7o1QnxWDRYw4H5P26PXq7sbcUkEqeR4fg3Kxp2tigg/*))"), "timestamp": "now", "active": True, "internal": True},
174 # taproot multi scripts
175 {"desc": descsum_create(f"tr({H_POINT},{{pk(tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/*),pk(tprv8ZgxMBicQKsPeuVhWwi6wuMQGfPKi9Li5GtX35jVNknACgqe3CY4g5xgkfDDJcmtF7o1QnxWDRYw4H5P26PXq7sbcUkEqeR4fg3Kxp2tigg/0h/*)}})"), "timestamp": "now", "active": True},
176 # miniscript
177 {"desc": descsum_create(f"tr({H_POINT},or_b(pk(tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/1/2/*),s:pk(tprv8ZgxMBicQKsPeuVhWwi6wuMQGfPKi9Li5GtX35jVNknACgqe3CY4g5xgkfDDJcmtF7o1QnxWDRYw4H5P26PXq7sbcUkEqeR4fg3Kxp2tigg/1h/2/*)))"), "timestamp": "now", "active": True, "internal": True},
178 ]
179 )
180 assert_equal(all([r["success"] for r in import_res]), True)
181
182 # Export the watchonly wallet file and load onto online node
183 online_wallet = self.export_and_restore(offline_wallet, "imports_watchonly")
184
185 # Verify public descriptors are the same
186 assert_equal(offline_wallet.listdescriptors()["descriptors"], online_wallet.listdescriptors()["descriptors"])
187
188 # Verify all the addresses are the same
189 for address_type in ["legacy", "p2sh-segwit", "bech32", "bech32m"]:
190 for internal in [False, True]:
191 if internal:
192 addr = offline_wallet.getrawchangeaddress(address_type=address_type)
193 assert_equal(addr, online_wallet.getrawchangeaddress(address_type=address_type))
194 else:
195 addr = offline_wallet.getnewaddress(address_type=address_type)
196 assert_equal(addr, online_wallet.getnewaddress(address_type=address_type))
197 self.funder.sendtoaddress(addr, 1)
198 self.generate(self.online, 1, sync_fun=self.no_op)
199
200 # The hardened derivation should have KEYPOOL_SIZE - 1 remaining addresses
201 for _ in range(KEYPOOL_SIZE - 1):
202 online_wallet.getnewaddress(address_type="p2sh-segwit")
203 assert_raises_rpc_error(-12, "No addresses available", online_wallet.getnewaddress, address_type="p2sh-segwit")
204
205 # Verify that the offline wallet can sign and send
206 send_res = online_wallet.sendall([self.funder.getnewaddress()])
207 assert_equal(send_res["complete"], False)
208 assert "psbt" in send_res
209 signed_psbt = offline_wallet.walletprocesspsbt(send_res["psbt"])["psbt"]
210 finalized = self.online.finalizepsbt(signed_psbt)["hex"]
211 self.online.sendrawtransaction(finalized)
212
213 self.generate(self.online, 1, sync_fun=self.no_op)
214 offline_wallet.unloadwallet()
215 online_wallet.unloadwallet()
216
217 def test_avoid_reuse(self):
218 self.log.info("Test that the avoid reuse flag appears in the exported wallet")
219 self.offline.createwallet(wallet_name="avoidreuse", avoid_reuse=True)
220 offline_wallet = self.offline.get_wallet_rpc("avoidreuse")
221 assert_equal(offline_wallet.getwalletinfo()["avoid_reuse"], True)
222
223 # The avoid_reuse flag also sets some specific address book entries to track reused addresses
224 # In order for these to be set, a few transactions need to be made, so briefly connect offline to online
225 self.connect_nodes(self.offline.index, self.online.index)
226 addr = offline_wallet.getnewaddress()
227 self.funder.sendtoaddress(addr, 1)
228 self.generate(self.online, 1)
229 # Spend funds in order to mark addr as previously spent
230 offline_wallet.sendall([self.funder.getnewaddress()])
231 self.funder.sendtoaddress(addr, 1)
232 self.generate(self.online, 1)
233 assert_equal(offline_wallet.listunspent(addresses=[addr])[0]["reused"], True)
234 self.disconnect_nodes(self.offline.index, self.online.index)
235
236 # Export the watchonly wallet file and load onto online node
237 online_wallet = self.export_and_restore(offline_wallet, "avoidreuse_watchonly")
238
239 # check avoid_reuse is still set
240 assert_equal(online_wallet.getwalletinfo()["avoid_reuse"], True)
241 assert_equal(online_wallet.listunspent(addresses=[addr])[0]["reused"], True)
242
243 offline_wallet.unloadwallet()
244 online_wallet.unloadwallet()
245
246 def test_encrypted_wallet(self):
247 self.log.info("Test that a watchonly wallet can be exported from a locked wallet")
248 self.offline.createwallet(wallet_name="encrypted", passphrase="pass")
249 offline_wallet = self.offline.get_wallet_rpc("encrypted")
250 assert_equal(offline_wallet.getwalletinfo()["unlocked_until"], 0)
251
252 # Export the watchonly wallet file and load onto online node
253 online_wallet = self.export_and_restore(offline_wallet, "encrypted_watchonly")
254
255 # watchonly wallet does not have encryption because it doesn't have private keys
256 assert "unlocked_until" not in online_wallet.getwalletinfo()
257 # But it still has all of the public descriptors
258 assert_equal(offline_wallet.listdescriptors()["descriptors"], online_wallet.listdescriptors()["descriptors"])
259
260 offline_wallet.unloadwallet()
261 online_wallet.unloadwallet()
262
263 def test_export_blank_wallet(self):
264 self.log.info("Test that exporting a blank/no-private-key wallet returns an error")
265
266 # blank=True + disable_private_keys: wallet has no descriptors at all
267 self.offline.createwallet(wallet_name="blank_no_keys", blank=True, disable_private_keys=True)
268 blank_wallet = self.offline.get_wallet_rpc("blank_no_keys")
269 assert_equal(blank_wallet.listdescriptors()["descriptors"], [])
270 export_path = os.path.join(self.export_path, "blank_export.dat")
271 assert_raises_rpc_error(-4, "Wallet has no descriptors to export", blank_wallet.exportwatchonlywallet, export_path)
272 blank_wallet.unloadwallet()
273
274 # disable_private_keys only also results in a blank wallet (no descriptors)
275 self.offline.createwallet(wallet_name="no_keys_only", disable_private_keys=True)
276 no_keys_wallet = self.offline.get_wallet_rpc("no_keys_only")
277 assert_equal(no_keys_wallet.listdescriptors()["descriptors"], [])
278 assert_raises_rpc_error(-4, "Wallet has no descriptors to export", no_keys_wallet.exportwatchonlywallet, export_path)
279 no_keys_wallet.unloadwallet()
280
281 def run_test(self):
282 self.online = self.nodes[0]
283 self.offline = self.nodes[1]
284 self.funder = self.online.get_wallet_rpc(self.default_wallet_name)
285 self.export_path = os.path.join(self.options.tmpdir, "exported_wallets")
286 os.makedirs(self.export_path, exist_ok=True)
287
288 # Mine some blocks, and verify disconnected
289 self.generate(self.online, 101, sync_fun=self.no_op)
290 assert_not_equal(self.online.getbestblockhash(), self.offline.getbestblockhash())
291 assert_equal(self.online.getblockcount(), 101)
292 assert_equal(self.offline.getblockcount(), 0)
293
294 self.test_basic_export()
295 self.test_export_with_address_book()
296 self.test_export_with_txs_and_locked_coins()
297 self.test_export_imported_descriptors()
298 self.test_avoid_reuse()
299 self.test_encrypted_wallet()
300 self.test_export_blank_wallet()
301
302 if __name__ == '__main__':
303 WalletExportedWatchOnly(__file__).main()
304