Inside Bitcoin

1001        assert_equal(wallet.gettransaction(txid=child_txid)["confirmations"], -1)
1002        assert_equal(wallet.gettransaction(txid=conflict_txid)["confirmations"], 1)
1003
1004        wallet.unloadwallet()
1005
1006    def test_hybrid_pubkey(self):
1007        self.log.info("Test migration when wallet contains a hybrid pubkey")
1008
1009        wallet = self.create_legacy_wallet("hybrid_keys")
1010
1011        # Get the hybrid pubkey for one of the keys in the wallet
1012        normal_pubkey = wallet.getaddressinfo(wallet.getnewaddress())["pubkey"]
1013        first_byte = bytes.fromhex(normal_pubkey)[0] + 4 # Get the hybrid pubkey first byte
1014        parsed_pubkey = ECPubKey()
1015        parsed_pubkey.set(bytes.fromhex(normal_pubkey))
1016        parsed_pubkey.compressed = False
1017        hybrid_pubkey_bytes = bytearray(parsed_pubkey.get_bytes())
1018        hybrid_pubkey_bytes[0] = first_byte # Make it hybrid
1019        hybrid_pubkey = hybrid_pubkey_bytes.hex()
1020
1021        # Import the hybrid pubkey
1022        wallet.importpubkey(hybrid_pubkey)
1023        p2pkh_addr = key_to_p2pkh(hybrid_pubkey)
1024        p2pkh_addr_info = wallet.getaddressinfo(p2pkh_addr)
1025        assert_equal(p2pkh_addr_info["iswatchonly"], True)
1026        assert_equal(p2pkh_addr_info["ismine"], False) # Things involving hybrid pubkeys are not spendable
1027
1028        # Also import the p2wpkh for the pubkey to make sure we don't migrate it
1029        p2wpkh_addr = key_to_p2wpkh(hybrid_pubkey)
1030        wallet.importaddress(p2wpkh_addr)
1031
1032        migrate_info, wallet = self.migrate_and_get_rpc("hybrid_keys")
1033
1034        # Both addresses should only appear in the watchonly wallet
1035        p2pkh_addr_info = wallet.getaddressinfo(p2pkh_addr)
1036        assert_equal(p2pkh_addr_info["ismine"], False)
1037        p2wpkh_addr_info = wallet.getaddressinfo(p2wpkh_addr)
1038        assert_equal(p2wpkh_addr_info["ismine"], False)
1039
1040        watchonly_wallet = self.master_node.get_wallet_rpc(migrate_info["watchonly_name"])
1041        watchonly_p2pkh_addr_info = watchonly_wallet.getaddressinfo(p2pkh_addr)
1042        assert_equal(watchonly_p2pkh_addr_info["ismine"], True)
1043        watchonly_p2wpkh_addr_info = watchonly_wallet.getaddressinfo(p2wpkh_addr)
1044        assert_equal(watchonly_p2wpkh_addr_info["ismine"], True)
1045
1046        # There should only be raw or addr descriptors
1047        for desc in watchonly_wallet.listdescriptors()["descriptors"]:
1048            if desc["desc"].startswith("raw(") or desc["desc"].startswith("addr("):
1049                continue
1050            assert False, "Hybrid pubkey watchonly wallet has more than just raw() and addr()"
1051
1052        wallet.unloadwallet()
1053
1054    def test_failed_migration_cleanup(self):
1055        self.log.info("Test that a failed migration is cleaned up")
1056        wallet = self.create_legacy_wallet("failed")
1057
1058        # Make a copy of the wallet with the solvables wallet name so that we are unable
1059        # to create the solvables wallet when migrating, thus failing to migrate
1060        wallet.unloadwallet()
1061        solvables_path = self.master_node.wallets_path / "failed_solvables"
1062        shutil.copytree(self.old_node.wallets_path / "failed", solvables_path)
1063        original_shasum = sha256sum_file(solvables_path / "wallet.dat")
1064
1065        self.old_node.loadwallet("failed")
1066
1067        # Add a multisig so that a solvables wallet is created
1068        wallet.addmultisigaddress(2, [wallet.getnewaddress(), get_generate_key().pubkey])
1069        wallet.importaddress(get_generate_key().p2pkh_addr)
1070
1071        self.old_node.unloadwallet("failed")
1072        shutil.copytree(self.old_node.wallets_path / "failed", self.master_node.wallets_path / "failed")
1073        assert_raises_rpc_error(-4, "Failed to create database", self.master_node.migratewallet, "failed")
1074
1075        assert all(wallet not in self.master_node.listwallets() for wallet in ["failed", "failed_watchonly", "failed_solvables"])
1076
1077        assert not (self.master_node.wallets_path / "failed_watchonly").exists()
1078        # Since the file in failed_solvables is one that we put there, migration shouldn't touch it
1079        assert solvables_path.exists()
1080        new_shasum = sha256sum_file(solvables_path / "wallet.dat")
1081        assert_equal(original_shasum, new_shasum)
1082
1083        # Check the wallet we tried to migrate is still BDB
1084        self.assert_is_bdb("failed")
1085
1086    def test_blank(self):
1087        self.log.info("Test that a blank wallet is migrated")
1088        wallet = self.create_legacy_wallet("blank", blank=True)
1089        assert_equal(wallet.getwalletinfo()["blank"], True)
1090        _, wallet = self.migrate_and_get_rpc("blank")
1091        assert_equal(wallet.getwalletinfo()["blank"], True)
1092
1093    def test_avoidreuse(self):
1094        self.log.info("Test that avoidreuse persists after migration")
1095        def_wallet = self.master_node.get_wallet_rpc(self.default_wallet_name)
1096
1097        wallet = self.create_legacy_wallet("avoidreuse")
1098        wallet.setwalletflag("avoid_reuse", True)
1099
1100        # Import a pubkey to the test wallet and send some funds to it
1101        reused_imported_addr = def_wallet.getnewaddress()
1102        wallet.importpubkey(def_wallet.getaddressinfo(reused_imported_addr)["pubkey"])
1103        imported_utxos = self.create_outpoints(def_wallet, outputs=[{reused_imported_addr: 2}])
1104        def_wallet.lockunspent(False, imported_utxos)
1105
1106        # Send funds to the test wallet
1107        reused_addr = wallet.getnewaddress()
1108        def_wallet.sendtoaddress(reused_addr, 2)
1109
1110        self.generate(self.master_node, 1)
1111
1112        # Send funds from the test wallet with both its own and the imported
1113        wallet.sendall([def_wallet.getnewaddress()])
1114        def_wallet.sendall(recipients=[def_wallet.getnewaddress()], inputs=imported_utxos)
1115        self.generate(self.master_node, 1)
1116        balances = wallet.getbalances()
1117        assert_equal(balances["mine"]["trusted"], 0)
1118        assert_equal(balances["watchonly"]["trusted"], 0)
1119
1120        # Reuse the addresses
1121        def_wallet.sendtoaddress(reused_addr, 1)
1122        def_wallet.sendtoaddress(reused_imported_addr, 1)
1123        self.generate(self.master_node, 1)
1124        balances = wallet.getbalances()
1125        assert_equal(balances["mine"]["used"], 1)
1126        # Reused watchonly will not show up in balances
1127        assert_equal(balances["watchonly"]["trusted"], 0)
1128        assert_equal(balances["watchonly"]["untrusted_pending"], 0)
1129        assert_equal(balances["watchonly"]["immature"], 0)
1130
1131        utxos = wallet.listunspent()
1132        assert_equal(len(utxos), 2)
1133        for utxo in utxos:
1134            assert_equal(utxo["reused"], True)
1135
1136        # Migrate
1137        _, wallet = self.migrate_and_get_rpc("avoidreuse")
1138        watchonly_wallet = self.master_node.get_wallet_rpc("avoidreuse_watchonly")
1139
1140        # One utxo in each wallet, marked used
1141        utxos = wallet.listunspent()
1142        assert_equal(len(utxos), 1)
1143        assert_equal(utxos[0]["reused"], True)
1144        watchonly_utxos = watchonly_wallet.listunspent()
1145        assert_equal(len(watchonly_utxos), 1)
1146        assert_equal(watchonly_utxos[0]["reused"], True)
1147
1148    def test_preserve_tx_extra_info(self):
1149        self.log.info("Test that tx extra data is preserved after migration")
1150        def_wallet = self.master_node.get_wallet_rpc(self.default_wallet_name)
1151
1152        # Create and fund wallet
1153        wallet = self.create_legacy_wallet("persist_comments")
1154        def_wallet.sendtoaddress(wallet.getnewaddress(), 2)
1155
1156        self.generate(self.master_node, 6)
1157
1158        # Create tx and bump it to store 'replaced_by_txid' and 'replaces_txid' data within the transactions.
1159        # Additionally, store an extra comment within the original tx.
1160        extra_comment = "don't discard me"
1161        original_tx_id = wallet.sendtoaddress(address=wallet.getnewaddress(), amount=1, comment=extra_comment)
1162        bumped_tx = wallet.bumpfee(txid=original_tx_id)
1163
1164        def check_comments():
1165            for record in wallet.listtransactions():
1166                if record["txid"] == original_tx_id:
1167                    assert_equal(record["replaced_by_txid"], bumped_tx["txid"])
1168                    assert_equal(record['comment'], extra_comment)
1169                elif record["txid"] == bumped_tx["txid"]:
1170                    assert_equal(record["replaces_txid"], original_tx_id)
1171
1172        # Pre-migration verification
1173        check_comments()
1174        # Migrate
1175        _, wallet = self.migrate_and_get_rpc("persist_comments")
1176        # Post-migration verification
1177        check_comments()
1178
1179        wallet.unloadwallet()
1180
1181    def test_migrate_simple_watch_only(self):
1182        self.log.info("Test migrating a watch-only p2pk script")
1183        wallet = self.create_legacy_wallet("bare_p2pk", blank=True)
1184        _, pubkey = generate_keypair()
1185        p2pk_script = key_to_p2pk_script(pubkey)
1186        wallet.importaddress(address=p2pk_script.hex())
1187        # Migrate wallet in the latest node
1188        res, _ = self.migrate_and_get_rpc("bare_p2pk")
1189        wo_wallet = self.master_node.get_wallet_rpc(res['wallet_name'])
1190        assert_equal(wo_wallet.listdescriptors()['descriptors'][0]['desc'], descsum_create(f'pk({pubkey.hex()})'))
1191        assert_equal(wo_wallet.getwalletinfo()["private_keys_enabled"], False)
1192
1193        # Ensure that migrating a wallet with watch-only scripts does not create a spendable wallet.
1194        assert_equal('bare_p2pk_watchonly', res['wallet_name'])
1195        assert "bare_p2pk" not in self.master_node.listwallets()
1196        assert "bare_p2pk" not in [w["name"] for w in self.master_node.listwalletdir()["wallets"]]
1197
1198        wo_wallet.unloadwallet()
1199
1200    def test_manual_keys_import(self):
1201        self.log.info("Test migrating standalone private keys")
1202        wallet = self.create_legacy_wallet("import_privkeys", blank=True)
1203        privkey, pubkey = generate_keypair(wif=True)
1204        wallet.importprivkey(privkey=privkey, label="hi", rescan=False)
1205
1206        # Migrate and verify
1207        res, wallet = self.migrate_and_get_rpc("import_privkeys")
1208
1209        # There should be descriptors containing the imported key for: pk(), pkh(), sh(wpkh()), wpkh()
1210        key_origin = hash160(pubkey)[:4].hex()
1211        pubkey_hex = pubkey.hex()
1212        combo_desc = descsum_create(f"combo([{key_origin}]{pubkey_hex})")
1213
1214        # Verify all expected descriptors were migrated
1215        migrated_desc = [item['desc'] for item in wallet.listdescriptors()['descriptors'] if pubkey.hex() in item['desc']]
1216        assert_equal([combo_desc], migrated_desc)
1217        wallet.unloadwallet()
1218
1219        ######################################################
1220        self.log.info("Test migrating standalone public keys")
1221        wallet = self.create_legacy_wallet("import_pubkeys", blank=True)
1222        wallet.importpubkey(pubkey=pubkey_hex, rescan=False)
1223
1224        res, _ = self.migrate_and_get_rpc("import_pubkeys")
1225
1226        # Same as before, there should be descriptors in the watch-only wallet for the imported pubkey
1227        wo_wallet = self.nodes[0].get_wallet_rpc(res['wallet_name'])
1228        # Assert this is a watch-only wallet
1229        assert_equal(wo_wallet.getwalletinfo()["private_keys_enabled"], False)
1230        # As we imported the pubkey only, there will be no key origin in the following descriptors
1231        pk_desc = descsum_create(f'pk({pubkey_hex})')
1232        pkh_desc = descsum_create(f'pkh({pubkey_hex})')
1233        sh_wpkh_desc = descsum_create(f'sh(wpkh({pubkey_hex}))')
1234        wpkh_desc = descsum_create(f'wpkh({pubkey_hex})')
1235        expected_descs = [pk_desc, pkh_desc, sh_wpkh_desc, wpkh_desc]
1236
1237        # Verify all expected descriptors were migrated
1238        migrated_desc = [item['desc'] for item in wo_wallet.listdescriptors()['descriptors']]
1239        assert_equal(expected_descs, migrated_desc)
1240        # Ensure that migrating a wallet with watch-only scripts does not create a spendable wallet.
1241        assert_equal('import_pubkeys_watchonly', res['wallet_name'])
1242        assert "import_pubkeys" not in self.master_node.listwallets()
1243        assert "import_pubkeys" not in [w["name"] for w in self.master_node.listwalletdir()["wallets"]]
1244        wo_wallet.unloadwallet()
1245
1246    def test_p2wsh(self):
1247        self.log.info("Test that non-multisig P2WSH output scripts are migrated")
1248        def_wallet = self.master_node.get_wallet_rpc(self.default_wallet_name)
1249
1250        wallet = self.create_legacy_wallet("p2wsh")