From ef2afc6a0a3cf91e9f21b7cc89b4e1bb04ddcef8 Mon Sep 17 00:00:00 2001 From: Ava Chow Date: Mon, 8 Jun 2026 19:02:10 -0700 Subject: [PATCH] test: Test for wallet txs with alternate wtxids --- test/functional/wallet_listtransactions.py | 115 +++++++++++++++++++++ 1 file changed, 115 insertions(+) diff --git a/test/functional/wallet_listtransactions.py b/test/functional/wallet_listtransactions.py index 59ed1d4e54f..0391109cc5c 100755 --- a/test/functional/wallet_listtransactions.py +++ b/test/functional/wallet_listtransactions.py @@ -11,11 +11,13 @@ import shutil from test_framework.blocktools import MAX_FUTURE_BLOCK_TIME from test_framework.descriptors import descsum_create +from test_framework.extendedkey import ExtendedPrivateKey from test_framework.test_framework import BitcoinTestFramework from test_framework.util import ( assert_not_equal, assert_array_result, assert_equal, + assert_greater_than, assert_raises_rpc_error, find_vout_for_address, ) @@ -96,6 +98,7 @@ class ListTransactionsTest(BitcoinTestFramework): self.run_coinjoin_test() self.run_invalid_parameters_test() self.test_op_return() + self.test_alternate_witness_tx() self.test_from_me_status_change() def run_externally_generated_address_test(self): @@ -242,5 +245,117 @@ class ListTransactionsTest(BitcoinTestFramework): assert "fee" in tx_info assert_equal(any(detail["category"] == "send" for detail in tx_info["details"]), True) + def check_tx_variants(self, wallet, txid, canonical_tx_hex, canonical_wtxid, alternate_wtxids): + """Assert gettransaction and listtransactions report tx variants properly""" + tx_info = wallet.gettransaction(txid) + assert_equal(tx_info["hex"], canonical_tx_hex) + assert_equal(tx_info["wtxid"], canonical_wtxid) + # alternate_wtxids lists the other variants, never the canonical one + assert canonical_wtxid not in tx_info["alternate_wtxids"] + assert_equal(set(tx_info["alternate_wtxids"]), set(alternate_wtxids)) + + # listtransactions exposes the same alternate_wtxids field as gettransaction + list_entry = next(entry for entry in wallet.listtransactions() if entry["txid"] == txid) + assert_equal(list_entry["alternate_wtxids"], tx_info["alternate_wtxids"]) + + # Returns the finalized psbt transaction and its wtxid + def finalize_tx_variant(self, wallet, psbt, spend_path): + # First check the expected spend path is being used + sig_field = {"script": "taproot_script_path_sigs", "key": "taproot_key_path_sig"} + present, absent = sig_field[spend_path], sig_field["key" if spend_path == "script" else "script"] + psbt_input = self.nodes[0].decodepsbt(psbt)["inputs"][0] + assert present in psbt_input and absent not in psbt_input + + # Then finalize and decode + tx = wallet.finalizepsbt(psbt)["hex"] + return tx, wallet.decoderawtransaction(tx)["hash"] + + def test_alternate_witness_tx(self): + self.log.info("Test gettransaction and listtransactions report alternate witnesses and canonical variant") + self.nodes[0].createwallet("altwit") + default_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name) + wallet = self.nodes[0].get_wallet_rpc("altwit") + + xprvs = [ExtendedPrivateKey.generate() for _ in range(0, 2)] + xpubs = [xprv.pubkey() for xprv in xprvs] + + # Import a taproot descriptor with script paths + desc = descsum_create(f"tr({xpubs[0].to_string()}/*,pk({xprvs[1].to_string()}/*))") + assert_equal(wallet.importdescriptors([{"desc": desc, "active": True, "timestamp": "now"}])[0]["success"], True) + default_wallet.sendtoaddress(wallet.getnewaddress(address_type="bech32m"), 1) + self.generate(self.nodes[0], 1, sync_fun=self.no_op) + # Isolate node0 for later reorg coverage + self.disconnect_nodes(0, 1) + self.disconnect_nodes(0, 2) + + # Create output psbt + psbt = wallet.walletcreatefundedpsbt(outputs=[{default_wallet.getnewaddress(): 0.5}])["psbt"] + + # Create a script path spend and relay it. With only one variant known it + # is trivially canonical and has no alternates + self.log.info("Test the only known variant is canonical with no alternates") + script_path_psbt = wallet.walletprocesspsbt(psbt=psbt, finalize=False)["psbt"] + script_path_tx, script_path_wtxid = self.finalize_tx_variant(wallet, script_path_psbt, spend_path="script") + txid = self.nodes[0].sendrawtransaction(script_path_tx) + self.check_tx_variants(wallet, txid, script_path_tx, script_path_wtxid, alternate_wtxids=[]) + + # Make a key path spend separate from the wallet + key_path_desc = descsum_create(f"tr({xprvs[0].to_string()}/*,pk({xpubs[1].to_string()}/*))") + key_path_psbt = self.nodes[0].descriptorprocesspsbt(psbt=psbt, descriptors=[{"desc": key_path_desc}], finalize=False)["psbt"] + key_path_tx, key_path_wtxid = self.finalize_tx_variant(wallet, key_path_psbt, spend_path="key") + + # Ensure variants share the same txid but differ in wtxid, and the key path is the lighter of the two + assert_equal(txid, self.nodes[0].decoderawtransaction(key_path_tx)["txid"]) + assert_not_equal(script_path_wtxid, key_path_wtxid) + assert_greater_than( + self.nodes[0].decoderawtransaction(script_path_tx)["weight"], + self.nodes[0].decoderawtransaction(key_path_tx)["weight"], + ) + + # The wallet only learns the key path witness from a block (the mempool + # holds one transaction per txid). Mine the key path: a confirmed variant + # is canonical, with the script path now listed as its alternate. + block = self.generateblock(self.nodes[0], default_wallet.getnewaddress(), [key_path_tx], sync_fun=self.no_op)["hash"] + self.check_tx_variants(wallet, txid, key_path_tx, key_path_wtxid, alternate_wtxids=[script_path_wtxid]) + + # Reorg that block out so both variants are known and unconfirmed. With no + # confirmation to force the choice, the lighter key path is canonical. + self.log.info("Test the lighter variant is canonical when both are known and unconfirmed") + self.nodes[0].invalidateblock(block) + self.nodes[0].syncwithvalidationinterfacequeue() + assert_equal(wallet.gettransaction(txid)["confirmations"], 0) + self.check_tx_variants(wallet, txid, key_path_tx, key_path_wtxid, alternate_wtxids=[script_path_wtxid]) + + # The canonical choice and alternates survive a wallet reload + wallet.unloadwallet() + self.nodes[0].loadwallet("altwit") + self.check_tx_variants(wallet, txid, key_path_tx, key_path_wtxid, alternate_wtxids=[script_path_wtxid]) + + # Now confirm the heavier script path instead, on a longer competing chain + # from node1, and reconnect so node0 reorgs onto it. The confirmed variant + # is canonical even though it is the heavier one. + self.log.info("Test a confirmed variant is canonical even when it is the heavier one") + self.generate(self.nodes[1], 3, sync_fun=self.no_op) + block = self.generateblock(self.nodes[1], default_wallet.getnewaddress(), [script_path_tx], sync_fun=self.no_op)["hash"] + self.connect_nodes(0, 1) + self.connect_nodes(0, 2) + self.sync_all() + self.check_tx_variants(wallet, txid, script_path_tx, script_path_wtxid, alternate_wtxids=[key_path_wtxid]) + + # The confirmed-canonical choice survive a reload + wallet.unloadwallet() + self.nodes[0].loadwallet("altwit") + self.check_tx_variants(wallet, txid, script_path_tx, script_path_wtxid, alternate_wtxids=[key_path_wtxid]) + + self.log.info("Test canonical reverts to the lighter variant when the confirmed one is reorged out") + # Both variants are unconfirmed again, so the lighter key path is canonical once more + self.disconnect_nodes(0, 1) + self.disconnect_nodes(0, 2) + self.nodes[0].invalidateblock(block) + self.nodes[0].syncwithvalidationinterfacequeue() + assert_equal(wallet.gettransaction(txid)["confirmations"], 0) + self.check_tx_variants(wallet, txid, key_path_tx, key_path_wtxid, alternate_wtxids=[script_path_wtxid]) + + if __name__ == '__main__': ListTransactionsTest(__file__).main()