Changeset View
Changeset View
Standalone View
Standalone View
test/functional/rpc_psbt.py
- This file was added.
| Property | Old Value | New Value |
|---|---|---|
| File Mode | null | 100755 |
| #!/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 json | |||||
| import os | |||||
| 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, 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": 0}) | |||||
| 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) | |||||
| # 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): | |||||
| for key in signer['privkeys']: | |||||
| self.nodes[i].importprivkey(key) | |||||
| signed_tx = self.nodes[i].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() | |||||