wallet_signer.py raw

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