1  #!/usr/bin/env python3
  2  # Copyright (c) 2016-2022 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 Hierarchical Deterministic wallet function."""
  6  
  7  import shutil
  8  
  9  from test_framework.blocktools import COINBASE_MATURITY
 10  from test_framework.test_framework import BitcoinTestFramework
 11  from test_framework.util import (
 12      assert_equal,
 13      wallet_importprivkey,
 14  )
 15  
 16  
 17  class WalletHDTest(BitcoinTestFramework):
 18      def set_test_params(self):
 19          self.setup_clean_chain = True
 20          self.num_nodes = 2
 21          self.extra_args = [[], ['-keypool=0']]
 22          # whitelist peers to speed up tx relay / mempool sync
 23          self.noban_tx_relay = True
 24  
 25      def skip_test_if_missing_module(self):
 26          self.skip_if_no_wallet()
 27  
 28      def run_test(self):
 29          # Make sure we use hd, keep masterkeyid
 30          hd_fingerprint = self.nodes[1].getaddressinfo(self.nodes[1].getnewaddress())['hdmasterfingerprint']
 31          assert_equal(len(hd_fingerprint), 8)
 32  
 33          # create an internal key
 34          change_addr = self.nodes[1].getrawchangeaddress()
 35          change_addrV = self.nodes[1].getaddressinfo(change_addr)
 36          assert_equal(change_addrV["hdkeypath"], "m/84h/1h/0h/1/0")
 37  
 38          # Import a non-HD private key in the HD wallet
 39          non_hd_add = 'bcrt1qmevj8zfx0wdvp05cqwkmr6mxkfx60yezwjksmt'
 40          non_hd_key = 'cS9umN9w6cDMuRVYdbkfE4c7YUFLJRoXMfhQ569uY4odiQbVN8Rt'
 41          wallet_importprivkey(self.nodes[1], non_hd_key, "now")
 42  
 43          # This should be enough to keep the master key and the non-HD key
 44          self.nodes[1].backupwallet(self.nodes[1].datadir_path / "hd.bak")
 45  
 46          # Derive some HD addresses and remember the last
 47          # Also send funds to each add
 48          self.generate(self.nodes[0], COINBASE_MATURITY + 1)
 49          hd_add = None
 50          NUM_HD_ADDS = 10
 51          for i in range(1, NUM_HD_ADDS + 1):
 52              hd_add = self.nodes[1].getnewaddress()
 53              hd_info = self.nodes[1].getaddressinfo(hd_add)
 54              assert_equal(hd_info["hdkeypath"], "m/84h/1h/0h/0/" + str(i))
 55              assert_equal(hd_info["hdmasterfingerprint"], hd_fingerprint)
 56              self.nodes[0].sendtoaddress(hd_add, 1)
 57              self.generate(self.nodes[0], 1)
 58          self.nodes[0].sendtoaddress(non_hd_add, 1)
 59          self.generate(self.nodes[0], 1)
 60  
 61          # create an internal key (again)
 62          change_addr = self.nodes[1].getrawchangeaddress()
 63          change_addrV = self.nodes[1].getaddressinfo(change_addr)
 64          assert_equal(change_addrV["hdkeypath"], "m/84h/1h/0h/1/1")
 65  
 66          self.sync_all()
 67          assert_equal(self.nodes[1].getbalance(), NUM_HD_ADDS + 1)
 68  
 69          self.log.info("Restore backup ...")
 70          self.stop_node(1)
 71          # we need to delete the complete chain directory
 72          # otherwise node1 would auto-recover all funds in flag the keypool keys as used
 73          shutil.rmtree(self.nodes[1].blocks_path)
 74          shutil.rmtree(self.nodes[1].chain_path / "chainstate")
 75          shutil.copyfile(
 76              self.nodes[1].datadir_path / "hd.bak",
 77              self.nodes[1].wallets_path / self.default_wallet_name / self.wallet_data_filename
 78          )
 79          self.start_node(1)
 80  
 81          # Assert that derivation is deterministic
 82          hd_add_2 = None
 83          for i in range(1, NUM_HD_ADDS + 1):
 84              hd_add_2 = self.nodes[1].getnewaddress()
 85              hd_info_2 = self.nodes[1].getaddressinfo(hd_add_2)
 86              assert_equal(hd_info_2["hdkeypath"], "m/84h/1h/0h/0/" + str(i))
 87              assert_equal(hd_info_2["hdmasterfingerprint"], hd_fingerprint)
 88          assert_equal(hd_add, hd_add_2)
 89          self.connect_nodes(0, 1)
 90          self.sync_all()
 91  
 92          # Needs rescan
 93          self.nodes[1].rescanblockchain()
 94          assert_equal(self.nodes[1].getbalance(), NUM_HD_ADDS + 1)
 95  
 96          # Try a RPC based rescan
 97          self.stop_node(1)
 98          shutil.rmtree(self.nodes[1].blocks_path)
 99          shutil.rmtree(self.nodes[1].chain_path / "chainstate")
100          shutil.copyfile(
101              self.nodes[1].datadir_path / "hd.bak",
102              self.nodes[1].wallets_path / self.default_wallet_name / self.wallet_data_filename
103          )
104          self.start_node(1, extra_args=self.extra_args[1])
105          self.connect_nodes(0, 1)
106          self.sync_all()
107          # Wallet automatically scans blocks older than key on startup
108          assert_equal(self.nodes[1].getbalance(), NUM_HD_ADDS + 1)
109          out = self.nodes[1].rescanblockchain(0, 1)
110          assert_equal(out['start_height'], 0)
111          assert_equal(out['stop_height'], 1)
112          out = self.nodes[1].rescanblockchain()
113          assert_equal(out['start_height'], 0)
114          assert_equal(out['stop_height'], self.nodes[1].getblockcount())
115          assert_equal(self.nodes[1].getbalance(), NUM_HD_ADDS + 1)
116  
117          # send a tx and make sure its using the internal chain for the changeoutput
118          txid = self.nodes[1].sendtoaddress(self.nodes[0].getnewaddress(), 1)
119          outs = self.nodes[1].gettransaction(txid=txid, verbose=True)['decoded']['vout']
120          keypath = ""
121          for out in outs:
122              if out['value'] != 1:
123                  keypath = self.nodes[1].getaddressinfo(out['scriptPubKey']['address'])['hdkeypath']
124  
125          assert_equal(keypath[0:14], "m/84h/1h/0h/1/")
126  
127  
128  if __name__ == '__main__':
129      WalletHDTest(__file__).main()