Skip to content

Commit 1ec0073

Browse files
committed
test: Test for exportwatchonlywallet
1 parent c87e0b8 commit 1ec0073

File tree

2 files changed

+288
-0
lines changed

2 files changed

+288
-0
lines changed

test/functional/test_runner.py

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

0 commit comments

Comments
 (0)