Inside Bitcoin

501
502        # Repeat with descriptorprocesspsbt
503        # Mismatching sighash type fails, including when no type is specified
504        for sighash in ["DEFAULT", "ALL", "NONE", "SINGLE", "NONE|ANYONECANPAY", "SINGLE|ANYONECANPAY", None]:
505            assert_raises_rpc_error(-22, "Specified sighash value does not match value stored in PSBT", self.nodes[0].descriptorprocesspsbt, psbt, descs, sighash)
506
507        # Matching sighash type succeeds
508        proc = self.nodes[0].descriptorprocesspsbt(psbt, descs, "ALL|ANYONECANPAY")
509        assert_equal(proc["complete"], True)
510
511        wallet.unloadwallet()
512
513    def test_sighash_adding(self):
514        self.log.info("Test adding of sighash type field")
515        self.nodes[0].createwallet("sighash_adding")
516        wallet = self.nodes[0].get_wallet_rpc("sighash_adding")
517        def_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
518
519        outputs = [{wallet.getnewaddress(address_type="bech32"): 1}]
520        outputs.append({wallet.getnewaddress(address_type="bech32m"): 1})
521        descs = wallet.listdescriptors(True)["descriptors"]
522        def_wallet.send(outputs)
523        self.generate(self.nodes[0], 6)
524        utxos = wallet.listunspent()
525
526        # Make a PSBT
527        psbt = wallet.walletcreatefundedpsbt(utxos, [{def_wallet.getnewaddress(): 0.5}])["psbt"]
528
529        # Process the PSBT with the wallet
530        wallet_psbt = wallet.walletprocesspsbt(psbt=psbt, sighashtype="ALL|ANYONECANPAY", finalize=False)["psbt"]
531
532        # Separately process the PSBT with descriptors
533        desc_psbt = self.nodes[0].descriptorprocesspsbt(psbt=psbt, descriptors=descs, sighashtype="ALL|ANYONECANPAY", finalize=False)["psbt"]
534
535        for psbt in [wallet_psbt, desc_psbt]:
536            # Check that the PSBT has a sighash field on all inputs
537            dec_psbt = self.nodes[0].decodepsbt(psbt)
538            for input in dec_psbt["inputs"]:
539                assert_equal(input["sighash"], "ALL|ANYONECANPAY")
540
541            # Make sure we can still finalize the transaction
542            fin_res = self.nodes[0].finalizepsbt(psbt)
543            assert_equal(fin_res["complete"], True)
544            fin_hex = fin_res["hex"]
545            assert_equal(self.nodes[0].testmempoolaccept([fin_hex])[0]["allowed"], True)
546
547            # Change the sighash field to a different value and make sure we can no longer finalize
548            mod_psbt = PSBT.from_base64(psbt)
549            mod_psbt.i[0].map[PSBT_IN_SIGHASH_TYPE] = (SIGHASH_ALL).to_bytes(4, byteorder="little")
550            mod_psbt.i[1].map[PSBT_IN_SIGHASH_TYPE] = (SIGHASH_ALL).to_bytes(4, byteorder="little")
551            psbt = mod_psbt.to_base64()
552            fin_res = self.nodes[0].finalizepsbt(psbt)
553            assert_equal(fin_res["complete"], False)
554
555        self.nodes[0].sendrawtransaction(fin_hex)
556        self.generate(self.nodes[0], 1)
557
558        wallet.unloadwallet()
559
560    def test_decodepsbt_long_sighash_type(self):
561        self.log.info("Test that decodepsbt rejects invalid trailing bytes in the sighash type field")
562        node = self.nodes[0]
563        psbt = PSBT.from_base64(node.createpsbt([{"txid": "00" * 32, "vout": 0}], [{"data": "00"}]))
564        # The first byte of this sighash type is ALL, but the type itself is not
565        psbt.i[0].map[PSBT_IN_SIGHASH_TYPE] = (0x101).to_bytes(4, "little")
566        assert_equal(node.decodepsbt(psbt.to_base64())["inputs"][0]["sighash"], "")
567
568    def test_combinepsbt_sighash_type(self):
569        self.log.info("Test that combining PSBTs preserves the sighash type field regardless of order")
570        node = self.nodes[0]
571        node.createwallet("combine_sighash")
572        wallet = node.get_wallet_rpc("combine_sighash")
573        def_wallet = node.get_wallet_rpc(self.default_wallet_name)
574
575        def_wallet.send([{wallet.getnewaddress(address_type="bech32"): 1}])
576        self.generate(node, 1)
577        psbt = wallet.walletcreatefundedpsbt(wallet.listunspent(), [{def_wallet.getnewaddress(): 0.5}])["psbt"]
578
579        signed = wallet.walletprocesspsbt(psbt=psbt, sighashtype="ALL|ANYONECANPAY", finalize=False)["psbt"]
580        assert_equal(node.decodepsbt(signed)["inputs"][0].get("sighash"), "ALL|ANYONECANPAY")
581        updated = wallet.walletprocesspsbt(psbt=psbt, sign=False)["psbt"]
582        assert "sighash" not in node.decodepsbt(updated)["inputs"][0]
583
584        finalized = []
585        for psbts in [[signed, updated], [updated, signed]]:
586            combined = node.combinepsbt(psbts)
587            assert_equal(node.decodepsbt(combined)["inputs"][0].get("sighash"), "ALL|ANYONECANPAY")
588            fin_res = node.finalizepsbt(combined)
589            assert_equal(fin_res["complete"], True)
590            assert_equal(node.testmempoolaccept([fin_res["hex"]])[0]["allowed"], True)
591            finalized.append(fin_res["hex"])
592        assert_equal(finalized[0], finalized[1])
593
594        wallet.unloadwallet()
595
596    def assert_change_type(self, psbtx, expected_type):
597        """Assert that the given PSBT has a change output with the given type."""
598
599        # The decodepsbt RPC is stateless and independent of any settings, we can always just call it on the first node
600        decoded_psbt = self.nodes[0].decodepsbt(psbtx["psbt"])
601        changepos = psbtx["changepos"]
602        assert_equal(decoded_psbt["outputs"][changepos]["script"]["type"], expected_type)
603
604    def test_psbt_named_parameter_handling(self):
605        """Test that PSBT Base64 parameters with '=' padding are handled correctly in -named mode"""
606        self.log.info("Testing PSBT Base64 parameter handling with '=' padding characters")
607        node = self.nodes[0]
608        psbt_with_padding = "cHNidP8BAJoCAAAAAqvNEjSrzRI0q80SNKvNEjSrzRI0q80SNKvNEjSrzRI0AAAAAAD9////NBLNqzQSzas0Es2rNBLNqzQSzas0Es2rNBLNqzQSzasBAAAAAP3///8CoIYBAAAAAAAWABQVQBGVs/sqFAmC8HZ8O+g1htqivkANAwAAAAAAFgAUir7MzgyzDnRMjdkVa7d+Dwr07jsAAAAAAAAAAAA="
609
610        # Test decodepsbt with explicit named parameter containing '=' padding
611        result = node.cli("-named", "decodepsbt", f"psbt={psbt_with_padding}").send_cli()
612        assert 'tx' in result
613
614        # Test decodepsbt with positional argument containing '=' padding
615        result = node.cli("-named", "decodepsbt", psbt_with_padding).send_cli()
616        assert 'tx' in result
617
618        # Test analyzepsbt with positional argument containing '=' padding
619        result = node.cli("-named", "analyzepsbt", psbt_with_padding).send_cli()
620        assert 'inputs' in result
621
622        # Test finalizepsbt with positional argument containing '=' padding
623        result = node.cli("-named", "finalizepsbt", psbt_with_padding, "extract=true").send_cli()
624        assert 'complete' in result
625
626        # Test walletprocesspsbt with positional argument containing '=' padding
627        result = node.cli("-named", "walletprocesspsbt", psbt_with_padding).send_cli()
628        assert 'complete' in result
629
630        # Test utxoupdatepsbt with positional argument containing '=' padding
631        result = node.cli("-named", "utxoupdatepsbt", psbt_with_padding).send_cli()
632        assert isinstance(result, str) and len(result) > 0
633
634        # Test that unknown parameter with '=' gets treated as positional and return error
635        unknown_psbt_param = "unknown_param_data=more_data="
636        # This should be treated as positional and fail with decode error, not parameter error
637        assert_raises_rpc_error(-22, "TX decode failed invalid base64", node.cli("-named", "finalizepsbt", unknown_psbt_param).send_cli)
638
639        self.log.info("PSBT parameter handling test completed successfully")
640
641    def test_psbt_roundtrip(self):
642        self.log.info("Test that PSBTs roundtrip when RPC does nothing")
643        utxo = self.nodes[0].listunspent()[0]
644        for ver in [0, 2]:
645            psbt = self.nodes[0].walletcreatefundedpsbt(inputs=[utxo], outputs=[{self.nodes[0].getnewaddress(): utxo["amount"] / 2}], psbt_version=ver)["psbt"]
646
647            rt_psbts = [
648                self.nodes[0].combinepsbt([psbt, psbt]),
649                self.nodes[0].finalizepsbt(psbt)["psbt"],
650                self.nodes[0].utxoupdatepsbt(psbt),
651                self.nodes[0].descriptorprocesspsbt(psbt, [])["psbt"],
652                self.nodes[0].walletprocesspsbt(psbt, sign=False)["psbt"],
653            ]
654            for p in rt_psbts:
655                assert_equal(psbt, p)
656
657    def test_psbt_version(self):
658        tobump = self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 1)
659        utxo = self.nodes[0].listunspent()[0]
660        outputs = [{self.nodes[0].getnewaddress(): utxo["amount"] / 2}]
661        rawtx = self.nodes[0].createrawtransaction(inputs=[utxo], outputs=outputs)
662        for ver in [0, 2]:
663            psbt = self.nodes[0].createpsbt(inputs=[utxo], outputs=outputs, psbt_version=ver)
664            dec = self.nodes[0].decodepsbt(psbt)
665            assert_equal(ver, dec["psbt_version"])
666
667            psbt = self.nodes[0].walletcreatefundedpsbt(inputs=[utxo], outputs=outputs, psbt_version=ver)
668            dec = self.nodes[0].decodepsbt(psbt["psbt"])
669            assert_equal(ver, dec["psbt_version"])
670
671            psbt = self.nodes[0].converttopsbt(hexstring=rawtx, psbt_version=ver)
672            dec = self.nodes[0].decodepsbt(psbt)
673            assert_equal(ver, dec["psbt_version"])
674
675            psbt = self.nodes[0].psbtbumpfee(txid=tobump, psbt_version=ver)
676            dec = self.nodes[0].decodepsbt(psbt["psbt"])
677            assert_equal(ver, dec["psbt_version"])
678
679        assert_raises_rpc_error(-8, "The PSBT version can only be 2 or 0", self.nodes[0].createpsbt, inputs=[utxo], outputs=outputs, psbt_version=1)
680        assert_raises_rpc_error(-8, "The PSBT version can only be 2 or 0", self.nodes[0].walletcreatefundedpsbt, inputs=[utxo], outputs=outputs, psbt_version=1)
681        assert_raises_rpc_error(-8, "The PSBT version can only be 2 or 0", self.nodes[0].converttopsbt, hexstring=rawtx, psbt_version=1)
682        assert_raises_rpc_error(-8, "The PSBT version can only be 2 or 0", self.nodes[0].psbtbumpfee, txid=tobump, psbt_version=1)
683
684    def test_psbt_with_invalid_signature(self):
685        self.log.info("Test descriptorprocesspsbt with invalid signature in signed PSBT")
686
687        def_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
688        outputs = [{def_wallet.getnewaddress(address_type="bech32m"): 1}]
689        def_wallet.send(outputs)
690        self.generate(self.nodes[0], 1)
691
692        utxos = [utxo for utxo in def_wallet.listunspent() if utxo["desc"].startswith("tr")]
693        unsigned_psbt = def_wallet.walletcreatefundedpsbt(utxos, [{def_wallet.getnewaddress(): 0.5}])["psbt"]
694        descs = [desc for desc in def_wallet.listdescriptors(True)["descriptors"] if desc["desc"].startswith("tr")]
695
696        # Unload the wallet to avoid using wallet RPC internals in descriptorprocesspsbt.
697        def_wallet.unloadwallet()
698
699        result = self.nodes[0].descriptorprocesspsbt(psbt=unsigned_psbt, descriptors=descs, finalize=True)
700        assert_equal(result["complete"], True)
701        assert_equal("hex" in result, True)
702
703        valid_sig_psbt = result["psbt"]
704        flawed_psbt = PSBT.from_base64(valid_sig_psbt)
705        valid_witness = flawed_psbt.i[0].map.get(PSBT_IN_FINAL_SCRIPTWITNESS)
706        # The witness format is [num_items: CompactSize] [item1_len: CompactSize] [item1_data] ...
707        # For taproot key-path spend [0x01] [0x40 or 0x41] [64 or 65 byte signature].
708        # Skip the first 2 prefix bytes and flip a bit in the signature bytes only.
709        assert_equal(valid_witness[0], 1)
710        prefix = valid_witness[:2]
711        sig_len = valid_witness[1]
712        signature = valid_witness[2:2 + sig_len]
713        invalid_sig = bitflipper(signature)
714        invalid_witness = prefix + invalid_sig + valid_witness[2 + sig_len:]
715        flawed_psbt.i[0].map[PSBT_IN_FINAL_SCRIPTWITNESS] = invalid_witness
716        invalid_sig_psbt = flawed_psbt.to_base64()
717
718        result = self.nodes[0].descriptorprocesspsbt(psbt=invalid_sig_psbt, descriptors=descs, finalize=True)
719        assert_equal(result["complete"], False)
720        assert_equal("hex" in result, False)
721
722        # Load the default wallet back for later test cases.
723        self.nodes[0].loadwallet(self.default_wallet_name)
724
725    def run_test(self):
726        # Create and fund a raw tx for sending 10 BTC
727        psbtx1 = self.nodes[0].walletcreatefundedpsbt([], {self.nodes[2].getnewaddress():10})['psbt']
728
729        self.log.info("Test for invalid maximum transaction weights")
730        dest_arg = [{self.nodes[0].getnewaddress(): 1}]
731        min_tx_weight = MIN_STANDARD_TX_NONWITNESS_SIZE * WITNESS_SCALE_FACTOR
732        assert_raises_rpc_error(-4, f"Maximum transaction weight must be between {min_tx_weight} and {MAX_STANDARD_TX_WEIGHT}", self.nodes[0].walletcreatefundedpsbt, [], dest_arg, 0, {"max_tx_weight": -1})
733        assert_raises_rpc_error(-4, f"Maximum transaction weight must be between {min_tx_weight} and {MAX_STANDARD_TX_WEIGHT}", self.nodes[0].walletcreatefundedpsbt, [], dest_arg, 0, {"max_tx_weight": 0})
734        assert_raises_rpc_error(-4, f"Maximum transaction weight must be between {min_tx_weight} and {MAX_STANDARD_TX_WEIGHT}", self.nodes[0].walletcreatefundedpsbt, [], dest_arg, 0, {"max_tx_weight": MAX_STANDARD_TX_WEIGHT + 1})
735
736        # Base transaction vsize: version (4) + locktime (4) + input count (1) + witness overhead (1) = 10 vbytes
737        base_tx_vsize = 10
738        # One P2WPKH output vsize: outpoint (31 vbytes)
739        p2wpkh_output_vsize = 31
740        # 1 vbyte for output count
741        output_count = 1
742        tx_weight_without_inputs = (base_tx_vsize + output_count + p2wpkh_output_vsize) * WITNESS_SCALE_FACTOR
743        # min_tx_weight is greater than transaction weight without inputs
744        assert_greater_than(min_tx_weight, tx_weight_without_inputs)
745
746        # In order to test for when the passed max weight is less than the transaction weight without inputs
747        # Define destination with two outputs.
748        dest_arg_large = [{self.nodes[0].getnewaddress(): 1}, {self.nodes[0].getnewaddress(): 1}]
749        large_tx_vsize_without_inputs = base_tx_vsize + output_count + (p2wpkh_output_vsize * 2)
750        large_tx_weight_without_inputs = large_tx_vsize_without_inputs * WITNESS_SCALE_FACTOR