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