#!/usr/bin/env python3 # Copyright (c) 2018 The Bitcoin Core developers # Distributed under the MIT software license, see the accompanying # file COPYING or http://www.opensource.org/licenses/mit-license.php. """Test the Partially Signed Transaction RPCs. """ import os import json from test_framework.test_framework import BitcoinTestFramework from test_framework.util import assert_equal, assert_raises_rpc_error, find_output # Create one-input, one-output, no-fee transaction: class PSBTTest(BitcoinTestFramework): def set_test_params(self): self.setup_clean_chain = False self.num_nodes = 3 def run_test(self): # Create and fund a raw tx for sending 10 BTC psbtx1 = self.nodes[0].walletcreatefundedpsbt([], {self.nodes[2].getnewaddress():10})['psbt'] # Node 1 should not be able to add anything to it but still return the psbtx same as before psbtx = self.nodes[1].walletprocesspsbt(psbtx1)['psbt'] assert_equal(psbtx1, psbtx) # Sign the transaction and send signed_tx = self.nodes[0].walletprocesspsbt(psbtx)['psbt'] final_tx = self.nodes[0].finalizepsbt(signed_tx)['hex'] self.nodes[0].sendrawtransaction(final_tx) # Create p2sh and p2pkh addresses pubkey0 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())['pubkey'] pubkey1 = self.nodes[1].getaddressinfo(self.nodes[1].getnewaddress())['pubkey'] pubkey2 = self.nodes[2].getaddressinfo(self.nodes[2].getnewaddress())['pubkey'] p2sh = self.nodes[1].addmultisigaddress(2, [pubkey0, pubkey1, pubkey2])['address'] p2pkh = self.nodes[1].getnewaddress() # fund those addresses rawtx = self.nodes[0].createrawtransaction([], {p2sh:10, p2pkh:10}) rawtx = self.nodes[0].fundrawtransaction(rawtx, {"changePosition":2}) signed_tx = self.nodes[0].signrawtransactionwithwallet(rawtx['hex'])['hex'] txid = self.nodes[0].sendrawtransaction(signed_tx) self.nodes[0].generate(6) self.sync_all() # Find the output pos p2sh_pos = -1 p2pkh_pos = -1 decoded = self.nodes[0].decoderawtransaction(signed_tx) for out in decoded['vout']: if out['scriptPubKey']['addresses'][0] == p2sh: p2sh_pos = out['n'] elif out['scriptPubKey']['addresses'][0] == p2pkh: p2pkh_pos = out['n'] # spend single key from node 1 rawtx = self.nodes[1].walletcreatefundedpsbt([{"txid":txid,"vout":p2pkh_pos}], {self.nodes[1].getnewaddress():9.99})['psbt'] walletprocesspsbt_out = self.nodes[1].walletprocesspsbt(rawtx) assert_equal(walletprocesspsbt_out['complete'], True) self.nodes[1].sendrawtransaction(self.nodes[1].finalizepsbt(walletprocesspsbt_out['psbt'])['hex']) # partially sign multisig things with node 1 psbtx = self.nodes[1].walletcreatefundedpsbt([{"txid":txid,"vout":p2sh_pos}], {self.nodes[1].getnewaddress():9.99})['psbt'] walletprocesspsbt_out = self.nodes[1].walletprocesspsbt(psbtx) psbtx = walletprocesspsbt_out['psbt'] assert_equal(walletprocesspsbt_out['complete'], False) # partially sign with node 2. This should be complete and sendable walletprocesspsbt_out = self.nodes[2].walletprocesspsbt(psbtx) assert_equal(walletprocesspsbt_out['complete'], True) self.nodes[2].sendrawtransaction(self.nodes[2].finalizepsbt(walletprocesspsbt_out['psbt'])['hex']) # check that walletprocesspsbt fails to decode a non-psbt rawtx = self.nodes[1].createrawtransaction([{"txid":txid,"vout":p2pkh_pos}], {self.nodes[1].getnewaddress():9.99}) assert_raises_rpc_error(-22, "TX decode failed", self.nodes[1].walletprocesspsbt, rawtx) # Convert a non-psbt to psbt and make sure we can decode it rawtx = self.nodes[0].createrawtransaction([], {self.nodes[1].getnewaddress():10}) rawtx = self.nodes[0].fundrawtransaction(rawtx) new_psbt = self.nodes[0].converttopsbt(rawtx['hex']) self.nodes[0].decodepsbt(new_psbt) # Make sure that a psbt with signatures cannot be converted signedtx = self.nodes[0].signrawtransactionwithwallet(rawtx['hex']) assert_raises_rpc_error(-22, "Inputs must not have scriptSigs", self.nodes[0].converttopsbt, signedtx['hex']) # Explicilty allow converting non-empty txs new_psbt = self.nodes[0].converttopsbt(rawtx['hex']) self.nodes[0].decodepsbt(new_psbt) # Create outputs to nodes 1 and 2 node1_addr = self.nodes[1].getnewaddress() node2_addr = self.nodes[2].getnewaddress() txid1 = self.nodes[0].sendtoaddress(node1_addr, 13) txid2 =self.nodes[0].sendtoaddress(node2_addr, 13) self.nodes[0].generate(6) self.sync_all() vout1 = find_output(self.nodes[1], txid1, 13) vout2 = find_output(self.nodes[2], txid2, 13) # Create a psbt spending outputs from nodes 1 and 2 psbt_orig = self.nodes[0].createpsbt([{"txid":txid1, "vout":vout1}, {"txid":txid2, "vout":vout2}], {self.nodes[0].getnewaddress():25.999}) # Update psbts, should only have data for one input and not the other psbt1 = self.nodes[1].walletprocesspsbt(psbt_orig)['psbt'] psbt1_decoded = self.nodes[0].decodepsbt(psbt1) assert psbt1_decoded['inputs'][0] and not psbt1_decoded['inputs'][1] psbt2 = self.nodes[2].walletprocesspsbt(psbt_orig)['psbt'] psbt2_decoded = self.nodes[0].decodepsbt(psbt2) assert not psbt2_decoded['inputs'][0] and psbt2_decoded['inputs'][1] # Combine, finalize, and send the psbts combined = self.nodes[0].combinepsbt([psbt1, psbt2]) finalized = self.nodes[0].finalizepsbt(combined)['hex'] self.nodes[0].sendrawtransaction(finalized) self.nodes[0].generate(6) self.sync_all() # BIP 174 Test Vectors # Check that unknown values are just passed through unknown_psbt = "cHNidP8BAD8CAAAAAf//////////////////////////////////////////AAAAAAD/////AQAAAAAAAAAAA2oBAAAAAAAACg8BAgMEBQYHCAkPAQIDBAUGBwgJCgsMDQ4PAAA=" unknown_out = self.nodes[0].walletprocesspsbt(unknown_psbt)['psbt'] assert_equal(unknown_psbt, unknown_out) # Open the data file with open(os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data/rpc_psbt.json'), encoding='utf-8') as f: d = json.load(f) invalids = d['invalid'] valids = d['valid'] creators = d['creator'] signers = d['signer'] combiners = d['combiner'] finalizers = d['finalizer'] extractors = d['extractor'] # Invalid PSBTs for invalid in invalids: assert_raises_rpc_error(-22, "TX decode failed", self.nodes[0].decodepsbt, invalid) # Valid PSBTs for valid in valids: self.nodes[0].decodepsbt(valid) # Creator Tests for creator in creators: created_tx = self.nodes[0].createpsbt(creator['inputs'], creator['outputs']) assert_equal(created_tx, creator['result']) # Signer tests for i, signer in enumerate(signers): self.nodes[2].createwallet("wallet{}".format(i)) wrpc = self.nodes[2].get_wallet_rpc("wallet{}".format(i)) for key in signer['privkeys']: wrpc.importprivkey(key) signed_tx = wrpc.walletprocesspsbt(signer['psbt'])['psbt'] assert_equal(signed_tx, signer['result']) # Combiner test for combiner in combiners: combined = self.nodes[2].combinepsbt(combiner['combine']) assert_equal(combined, combiner['result']) # Finalizer test for finalizer in finalizers: finalized = self.nodes[2].finalizepsbt(finalizer['finalize'], False)['psbt'] assert_equal(finalized, finalizer['result']) # Extractor test for extractor in extractors: extracted = self.nodes[2].finalizepsbt(extractor['extract'], True)['hex'] assert_equal(extracted, extractor['result']) if __name__ == '__main__': PSBTTest().main()