wallet_signer.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2017-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 external signer.
6
7 Verify that a limenkad node can use an external signer command
8 See also rpc_signer.py for tests without wallet context.
9 """
10 import os
11 import sys
12
13 from test_framework.test_framework import LimenkaTestFramework
14 from test_framework.util import (
15 assert_equal,
16 assert_greater_than,
17 assert_raises_rpc_error,
18 )
19
20
21 class WalletSignerTest(LimenkaTestFramework):
22 def add_options(self, parser):
23 self.add_wallet_options(parser, legacy=False)
24
25 def mock_signer_path(self):
26 path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'mocks', 'signer.py')
27 return sys.executable + " " + path
28
29 def mock_no_connected_signer_path(self):
30 path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'mocks', 'no_signer.py')
31 return sys.executable + " " + path
32
33 def mock_invalid_signer_path(self):
34 path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'mocks', 'invalid_signer.py')
35 return sys.executable + " " + path
36
37 def mock_multi_signers_path(self):
38 path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'mocks', 'multi_signers.py')
39 return sys.executable + " " + path
40
41 def set_test_params(self):
42 self.num_nodes = 2
43
44 self.extra_args = [
45 [],
46 [f"-signer={self.mock_signer_path()}", '-keypool=10'],
47 ]
48
49 def skip_test_if_missing_module(self):
50 self.skip_if_no_external_signer()
51 self.skip_if_no_wallet()
52
53 def set_mock_result(self, node, res):
54 with open(os.path.join(node.cwd, "mock_result"), "w", encoding="utf8") as f:
55 f.write(res)
56
57 def clear_mock_result(self, node):
58 os.remove(os.path.join(node.cwd, "mock_result"))
59
60 def run_test(self):
61 self.test_valid_signer()
62 self.test_disconnected_signer()
63 self.restart_node(1, [f"-signer={self.mock_invalid_signer_path()}", "-keypool=10"])
64 self.test_invalid_signer()
65 self.restart_node(1, [f"-signer={self.mock_multi_signers_path()}", "-keypool=10"])
66 self.test_multiple_signers()
67
68 def test_valid_signer(self):
69 self.log.debug(f"-signer={self.mock_signer_path()}")
70
71 # Create new wallets for an external signer.
72 # disable_private_keys and descriptors must be true:
73 assert_raises_rpc_error(-4, "Private keys must be disabled when using an external signer", self.nodes[1].createwallet, wallet_name='not_hww', disable_private_keys=False, descriptors=True, external_signer=True)
74 if self.is_bdb_compiled():
75 assert_raises_rpc_error(-4, "Descriptor support must be enabled when using an external signer", self.nodes[1].createwallet, wallet_name='not_hww', disable_private_keys=True, descriptors=False, external_signer=True)
76 else:
77 assert_raises_rpc_error(-4, "Compiled without bdb support (required for legacy wallets)", self.nodes[1].createwallet, wallet_name='not_hww', disable_private_keys=True, descriptors=False, external_signer=True)
78
79 self.nodes[1].createwallet(wallet_name='hww', disable_private_keys=True, descriptors=True, external_signer=True)
80 hww = self.nodes[1].get_wallet_rpc('hww')
81 assert_equal(hww.getwalletinfo()["external_signer"], True)
82
83 # Flag can't be set afterwards (could be added later for non-blank descriptor based watch-only wallets)
84 self.nodes[1].createwallet(wallet_name='not_hww', disable_private_keys=True, descriptors=True, external_signer=False)
85 not_hww = self.nodes[1].get_wallet_rpc('not_hww')
86 assert_equal(not_hww.getwalletinfo()["external_signer"], False)
87
88 # Flag can be set
89 not_hww.setwalletflag("external_signer", True)
90 assert_equal(not_hww.getwalletinfo()["external_signer"], True)
91
92 # Flag can be unset
93 not_hww.setwalletflag("external_signer", False)
94 assert_equal(not_hww.getwalletinfo()["external_signer"], False)
95
96 # assert_raises_rpc_error(-4, "Multiple signers found, please specify which to use", wallet_name='not_hww', disable_private_keys=True, descriptors=True, external_signer=True)
97
98 # TODO: Handle error thrown by script
99 # self.set_mock_result(self.nodes[1], "2")
100 # assert_raises_rpc_error(-1, 'Unable to parse JSON',
101 # self.nodes[1].createwallet, wallet_name='not_hww2', disable_private_keys=True, descriptors=True, external_signer=False
102 # )
103 # self.clear_mock_result(self.nodes[1])
104
105 assert_equal(hww.getwalletinfo()["keypoolsize"], 40)
106
107 address1 = hww.getnewaddress(address_type="bech32")
108 assert_equal(address1, "bcrt1qm90ugl4d48jv8n6e5t9ln6t9zlpm5th68x4f8g")
109 address_info = hww.getaddressinfo(address1)
110 assert_equal(address_info['solvable'], True)
111 assert_equal(address_info['ismine'], True)
112 assert_equal(address_info['hdkeypath'], "m/84h/1h/0h/0/0")
113
114 address2 = hww.getnewaddress(address_type="p2sh-segwit")
115 assert_equal(address2, "2N2gQKzjUe47gM8p1JZxaAkTcoHPXV6YyVp")
116 address_info = hww.getaddressinfo(address2)
117 assert_equal(address_info['solvable'], True)
118 assert_equal(address_info['ismine'], True)
119 assert_equal(address_info['hdkeypath'], "m/49h/1h/0h/0/0")
120
121 address3 = hww.getnewaddress(address_type="legacy")
122 assert_equal(address3, "n1LKejAadN6hg2FrBXoU1KrwX4uK16mco9")
123 address_info = hww.getaddressinfo(address3)
124 assert_equal(address_info['solvable'], True)
125 assert_equal(address_info['ismine'], True)
126 assert_equal(address_info['hdkeypath'], "m/44h/1h/0h/0/0")
127
128 address4 = hww.getnewaddress(address_type="bech32m")
129 assert_equal(address4, "bcrt1phw4cgpt6cd30kz9k4wkpwm872cdvhss29jga2xpmftelhqll62ms4e9sqj")
130 address_info = hww.getaddressinfo(address4)
131 assert_equal(address_info['solvable'], True)
132 assert_equal(address_info['ismine'], True)
133 assert_equal(address_info['hdkeypath'], "m/86h/1h/0h/0/0")
134
135 self.log.info('Test walletdisplayaddress')
136 for address in [address1, address2, address3]:
137 result = hww.walletdisplayaddress(address)
138 assert_equal(result, {"address": address})
139
140 # Handle error thrown by script
141 self.set_mock_result(self.nodes[1], "2")
142 assert_raises_rpc_error(-1, 'RunCommandParseJSON error',
143 hww.walletdisplayaddress, address1
144 )
145 self.clear_mock_result(self.nodes[1])
146
147 # Returned address MUST match:
148 address_fail = hww.getnewaddress(address_type="bech32")
149 assert_equal(address_fail, "bcrt1ql7zg7ukh3dwr25ex2zn9jse926f27xy2jz58tm")
150 assert_raises_rpc_error(-1, 'Signer echoed unexpected address wrong_address',
151 hww.walletdisplayaddress, address_fail
152 )
153
154 self.log.info('Prepare mock PSBT')
155 self.nodes[0].sendtoaddress(address4, 1)
156 self.generate(self.nodes[0], 1)
157
158 # Load private key into wallet to generate a signed PSBT for the mock
159 self.nodes[1].createwallet(wallet_name="mock", disable_private_keys=False, blank=True, descriptors=True)
160 mock_wallet = self.nodes[1].get_wallet_rpc("mock")
161 assert mock_wallet.getwalletinfo()['private_keys_enabled']
162
163 result = mock_wallet.importdescriptors([{
164 "desc": "tr([00000001/86h/1h/0']tprv8ZgxMBicQKsPd7Uf69XL1XwhmjHopUGep8GuEiJDZmbQz6o58LninorQAfcKZWARbtRtfnLcJ5MQ2AtHcQJCCRUcMRvmDUjyEmNUWwx8UbK/0/*)#7ew68cn8",
165 "timestamp": 0,
166 "range": [0,1],
167 "internal": False,
168 "active": True
169 },
170 {
171 "desc": "tr([00000001/86h/1h/0']tprv8ZgxMBicQKsPd7Uf69XL1XwhmjHopUGep8GuEiJDZmbQz6o58LninorQAfcKZWARbtRtfnLcJ5MQ2AtHcQJCCRUcMRvmDUjyEmNUWwx8UbK/1/*)#0dtm6drl",
172 "timestamp": 0,
173 "range": [0, 0],
174 "internal": True,
175 "active": True
176 }])
177 assert_equal(result[0], {'success': True})
178 assert_equal(result[1], {'success': True})
179 assert_equal(mock_wallet.getwalletinfo()["txcount"], 1)
180 dest = self.nodes[0].getnewaddress(address_type='bech32')
181 mock_psbt = mock_wallet.walletcreatefundedpsbt([], {dest:0.5}, 0, {'replaceable': True}, True)['psbt']
182 mock_psbt_signed = mock_wallet.walletprocesspsbt(psbt=mock_psbt, sign=True, sighashtype="ALL", bip32derivs=True)
183 mock_tx = mock_psbt_signed["hex"]
184 assert mock_wallet.testmempoolaccept([mock_tx])[0]["allowed"]
185
186 # # Create a new wallet and populate with specific public keys, in order
187 # # to work with the mock signed PSBT.
188 # self.nodes[1].createwallet(wallet_name="hww4", disable_private_keys=True, descriptors=True, external_signer=True)
189 # hww4 = self.nodes[1].get_wallet_rpc("hww4")
190 #
191 # descriptors = [{
192 # "desc": "wpkh([00000001/84h/1h/0']tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/*)#x30uthjs",
193 # "timestamp": "now",
194 # "range": [0, 1],
195 # "internal": False,
196 # "watchonly": True,
197 # "active": True
198 # },
199 # {
200 # "desc": "wpkh([00000001/84h/1h/0']tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/1/*)#h92akzzg",
201 # "timestamp": "now",
202 # "range": [0, 0],
203 # "internal": True,
204 # "watchonly": True,
205 # "active": True
206 # }]
207
208 # result = hww4.importdescriptors(descriptors)
209 # assert_equal(result[0], {'success': True})
210 # assert_equal(result[1], {'success': True})
211 assert_equal(hww.getwalletinfo()["txcount"], 1)
212
213 assert hww.testmempoolaccept([mock_tx])[0]["allowed"]
214
215 with open(os.path.join(self.nodes[1].cwd, "mock_psbt"), "w", encoding="utf8") as f:
216 f.write(mock_psbt_signed["psbt"])
217
218 self.log.info('Test send using hww1')
219
220 # Don't broadcast transaction yet so the RPC returns the raw hex
221 res = hww.send(outputs={dest:0.5},add_to_wallet=False)
222 assert res["complete"]
223 assert_equal(res["hex"], mock_tx)
224
225 self.log.info('Test sendall using hww1')
226
227 res = hww.sendall(recipients=[{dest:0.5}, hww.getrawchangeaddress()], add_to_wallet=False)
228 assert res["complete"]
229 assert_equal(res["hex"], mock_tx)
230 # Broadcast transaction so we can bump the fee
231 hww.sendrawtransaction(res["hex"])
232
233 self.log.info('Prepare fee bumped mock PSBT')
234
235 # Now that the transaction is broadcast, bump fee in mock wallet:
236 orig_tx_id = res["txid"]
237 mock_psbt_bumped = mock_wallet.psbtbumpfee(orig_tx_id)["psbt"]
238 mock_psbt_bumped_signed = mock_wallet.walletprocesspsbt(psbt=mock_psbt_bumped, sign=True, sighashtype="ALL", bip32derivs=True)
239
240 with open(os.path.join(self.nodes[1].cwd, "mock_psbt"), "w", encoding="utf8") as f:
241 f.write(mock_psbt_bumped_signed["psbt"])
242
243 self.log.info('Test bumpfee using hww1')
244
245 # Bump fee
246 res = hww.bumpfee(orig_tx_id)
247 assert_greater_than(res["fee"], res["origfee"])
248 assert_equal(res["errors"], [])
249
250 # # Handle error thrown by script
251 # self.set_mock_result(self.nodes[4], "2")
252 # assert_raises_rpc_error(-1, 'Unable to parse JSON',
253 # hww4.signerprocesspsbt, psbt_orig, "00000001"
254 # )
255 # self.clear_mock_result(self.nodes[4])
256
257 def test_disconnected_signer(self):
258 self.log.info('Test disconnected external signer')
259
260 # First create a wallet with the signer connected
261 self.nodes[1].createwallet(wallet_name='hww_disconnect', disable_private_keys=True, external_signer=True)
262 hww = self.nodes[1].get_wallet_rpc('hww_disconnect')
263 assert_equal(hww.getwalletinfo()["external_signer"], True)
264
265 # Fund wallet
266 self.nodes[0].sendtoaddress(hww.getnewaddress(address_type="bech32m"), 1)
267 self.generate(self.nodes[0], 1)
268
269 # Restart node with no signer connected
270 self.log.debug(f"-signer={self.mock_no_connected_signer_path()}")
271 self.restart_node(1, [f"-signer={self.mock_no_connected_signer_path()}", "-keypool=10"])
272 self.nodes[1].loadwallet('hww_disconnect')
273 hww = self.nodes[1].get_wallet_rpc('hww_disconnect')
274
275 # Try to spend
276 dest = hww.getrawchangeaddress()
277 assert_raises_rpc_error(-25, "External signer not found", hww.send, outputs=[{dest:0.5}])
278
279 def test_invalid_signer(self):
280 self.log.debug(f"-signer={self.mock_invalid_signer_path()}")
281 self.log.info('Test invalid external signer')
282 assert_raises_rpc_error(-1, "Invalid descriptor", self.nodes[1].createwallet, wallet_name='hww_invalid', disable_private_keys=True, descriptors=True, external_signer=True)
283
284 def test_multiple_signers(self):
285 self.log.debug(f"-signer={self.mock_multi_signers_path()}")
286 self.log.info('Test multiple external signers')
287
288 assert_raises_rpc_error(-1, "More than one external signer found", self.nodes[1].createwallet, wallet_name='multi_hww', disable_private_keys=True, descriptors=True, external_signer=True)
289
290 if __name__ == '__main__':
291 WalletSignerTest(__file__).main()
292