Merge bitcoin#25625: test: add test for decoding PSBT with per-input preimage types

71a751f test: add test for decoding PSBT with per-input preimage types (Sebastian Falbesoner)
faf4337 refactor: move helper `random_bytes` to util library (Sebastian Falbesoner)
fdc1ca3 test: add constants for PSBT key types (BIP 174) (Sebastian Falbesoner)
1b035c0 refactor: move PSBT(Map) helpers from signet miner to test framework (Sebastian Falbesoner)
7c0dfec refactor: move `from_binary` helper from signet miner to test framework (Sebastian Falbesoner)
597a4b3 scripted-diff: rename `FromBinary` helper to `from_binary` (signet miner) (Sebastian Falbesoner)

Pull request description:

  This PR adds missing test coverage for the `decodepsbt` RPC in the case that a PSBT with on of the per-input preimage types (`PSBT_IN_RIPEMD160`, `PSBT_IN_SHA256`, `PSBT_IN_HASH160`, `PSBT_IN_HASH256`; see [BIP 174](https://github.com/bitcoin/bips/blob/master/bip-0174.mediawiki#Specification)) is passed. As preparation, the first four commits move the already existing helpers for (de)serialization of PSBTs and PSBTMaps from the signet miner to the test framework (in a new module `psbt.py`), which should be quite useful for further tests to easily create PSBTs.

ACKs for top commit:
  achow101:
    ACK 71a751f

Tree-SHA512: 04f2671612d94029da2ac8dc15ff93c4c8fcb73fe0b8cf5970509208564df1f5e32319b53ae998dd6e544d37637a9b75609f27a3685da51f603f6ed0555635fb

Author: achow101

contrib/signet/miner

@@ -4,7 +4,6 @@
 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
 import argparse
-import base64
 import json
 import logging
 import math
@@ -15,14 +14,13 @@ import sys
 import time
 import subprocess
 
-from io import BytesIO
-
 PATH_BASE_CONTRIB_SIGNET = os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
 PATH_BASE_TEST_FUNCTIONAL = os.path.abspath(os.path.join(PATH_BASE_CONTRIB_SIGNET, "..", "..", "test", "functional"))
 sys.path.insert(0, PATH_BASE_TEST_FUNCTIONAL)
 
 from test_framework.blocktools import get_witness_script, script_BIP34_coinbase_height # noqa: E402
-from test_framework.messages import CBlock, CBlockHeader, COutPoint, CTransaction, CTxIn, CTxInWitness, CTxOut, from_hex, deser_string, ser_compact_size, ser_string, ser_uint256, tx_from_hex # noqa: E402
+from test_framework.messages import CBlock, CBlockHeader, COutPoint, CTransaction, CTxIn, CTxInWitness, CTxOut, from_binary, from_hex, ser_string, ser_uint256, tx_from_hex # noqa: E402
+from test_framework.psbt import PSBT, PSBTMap, PSBT_GLOBAL_UNSIGNED_TX, PSBT_IN_FINAL_SCRIPTSIG, PSBT_IN_FINAL_SCRIPTWITNESS, PSBT_IN_NON_WITNESS_UTXO, PSBT_IN_SIGHASH_TYPE # noqa: E402
 from test_framework.script import CScriptOp # noqa: E402
 
 logging.basicConfig(
@@ -34,89 +32,6 @@ SIGNET_HEADER = b"\xec\xc7\xda\xa2"
 PSBT_SIGNET_BLOCK = b"\xfc\x06signetb"    # proprietary PSBT global field holding the block being signed
 RE_MULTIMINER = re.compile("^(\d+)(-(\d+))?/(\d+)$")
 
-# #### some helpers that could go into test_framework
-
-# like from_hex, but without the hex part
-def FromBinary(cls, stream):
-    """deserialize a binary stream (or bytes object) into an object"""
-    # handle bytes object by turning it into a stream
-    was_bytes = isinstance(stream, bytes)
-    if was_bytes:
-        stream = BytesIO(stream)
-    obj = cls()
-    obj.deserialize(stream)
-    if was_bytes:
-        assert len(stream.read()) == 0
-    return obj
-
-class PSBTMap:
-    """Class for serializing and deserializing PSBT maps"""
-
-    def __init__(self, map=None):
-        self.map = map if map is not None else {}
-
-    def deserialize(self, f):
-        m = {}
-        while True:
-            k = deser_string(f)
-            if len(k) == 0:
-                break
-            v = deser_string(f)
-            if len(k) == 1:
-                k = k[0]
-            assert k not in m
-            m[k] = v
-        self.map = m
-
-    def serialize(self):
-        m = b""
-        for k,v in self.map.items():
-            if isinstance(k, int) and 0 <= k and k <= 255:
-                k = bytes([k])
-            m += ser_compact_size(len(k)) + k
-            m += ser_compact_size(len(v)) + v
-        m += b"\x00"
-        return m
-
-class PSBT:
-    """Class for serializing and deserializing PSBTs"""
-
-    def __init__(self):
-        self.g = PSBTMap()
-        self.i = []
-        self.o = []
-        self.tx = None
-
-    def deserialize(self, f):
-        assert f.read(5) == b"psbt\xff"
-        self.g = FromBinary(PSBTMap, f)
-        assert 0 in self.g.map
-        self.tx = FromBinary(CTransaction, self.g.map[0])
-        self.i = [FromBinary(PSBTMap, f) for _ in self.tx.vin]
-        self.o = [FromBinary(PSBTMap, f) for _ in self.tx.vout]
-        return self
-
-    def serialize(self):
-        assert isinstance(self.g, PSBTMap)
-        assert isinstance(self.i, list) and all(isinstance(x, PSBTMap) for x in self.i)
-        assert isinstance(self.o, list) and all(isinstance(x, PSBTMap) for x in self.o)
-        assert 0 in self.g.map
-        tx = FromBinary(CTransaction, self.g.map[0])
-        assert len(tx.vin) == len(self.i)
-        assert len(tx.vout) == len(self.o)
-
-        psbt = [x.serialize() for x in [self.g] + self.i + self.o]
-        return b"psbt\xff" + b"".join(psbt)
-
-    def to_base64(self):
-        return base64.b64encode(self.serialize()).decode("utf8")
-
-    @classmethod
-    def from_base64(cls, b64psbt):
-        return FromBinary(cls, base64.b64decode(b64psbt))
-
-# #####
-
 def create_coinbase(height, value, spk):
     cb = CTransaction()
     cb.vin = [CTxIn(COutPoint(0, 0xffffffff), script_BIP34_coinbase_height(height), 0xffffffff)]
@@ -159,11 +74,11 @@ def signet_txs(block, challenge):
 
 def do_createpsbt(block, signme, spendme):
     psbt = PSBT()
-    psbt.g = PSBTMap( {0: signme.serialize(),
+    psbt.g = PSBTMap( {PSBT_GLOBAL_UNSIGNED_TX: signme.serialize(),
                        PSBT_SIGNET_BLOCK: block.serialize()
                      } )
-    psbt.i = [ PSBTMap( {0: spendme.serialize(),
-                         3: bytes([1,0,0,0])})
+    psbt.i = [ PSBTMap( {PSBT_IN_NON_WITNESS_UTXO: spendme.serialize(),
+                         PSBT_IN_SIGHASH_TYPE: bytes([1,0,0,0])})
              ]
     psbt.o = [ PSBTMap() ]
     return psbt.to_base64()
@@ -175,10 +90,10 @@ def do_decode_psbt(b64psbt):
     assert len(psbt.tx.vout) == 1
     assert PSBT_SIGNET_BLOCK in psbt.g.map
 
-    scriptSig = psbt.i[0].map.get(7, b"")
-    scriptWitness = psbt.i[0].map.get(8, b"\x00")
+    scriptSig = psbt.i[0].map.get(PSBT_IN_FINAL_SCRIPTSIG, b"")
+    scriptWitness = psbt.i[0].map.get(PSBT_IN_FINAL_SCRIPTWITNESS, b"\x00")
 
-    return FromBinary(CBlock, psbt.g.map[PSBT_SIGNET_BLOCK]), ser_string(scriptSig) + scriptWitness
+    return from_binary(CBlock, psbt.g.map[PSBT_SIGNET_BLOCK]), ser_string(scriptSig) + scriptWitness
 
 def finish_block(block, signet_solution, grind_cmd):
     block.vtx[0].vout[-1].scriptPubKey += CScriptOp.encode_op_pushdata(SIGNET_HEADER + signet_solution)

test/functional/feature_taproot.py

@@ -91,7 +91,11 @@
     script_to_p2wsh_script,
 )
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import assert_raises_rpc_error, assert_equal
+from test_framework.util import (
+    assert_raises_rpc_error,
+    assert_equal,
+    random_bytes,
+)
 from test_framework.key import generate_privkey, compute_xonly_pubkey, sign_schnorr, tweak_add_privkey, ECKey
 from test_framework.address import (
     hash160,
@@ -566,10 +570,6 @@ def random_checksig_style(pubkey):
         ret = CScript([pubkey, opcode])
     return bytes(ret)
 
-def random_bytes(n):
-    """Return a random bytes object of length n."""
-    return bytes(random.getrandbits(8) for i in range(n))
-
 def bitflipper(expr):
     """Return a callable that evaluates expr and returns it with a random bitflip."""
     def fn(ctx):

test/functional/rpc_psbt.py

@@ -11,10 +11,23 @@
 from test_framework.descriptors import descsum_create
 from test_framework.key import ECKey, H_POINT
 from test_framework.messages import (
+    COutPoint,
+    CTransaction,
+    CTxIn,
+    CTxOut,
     MAX_BIP125_RBF_SEQUENCE,
     WITNESS_SCALE_FACTOR,
     ser_compact_size,
 )
+from test_framework.psbt import (
+    PSBT,
+    PSBTMap,
+    PSBT_GLOBAL_UNSIGNED_TX,
+    PSBT_IN_RIPEMD160,
+    PSBT_IN_SHA256,
+    PSBT_IN_HASH160,
+    PSBT_IN_HASH256,
+)
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_approx,
@@ -23,6 +36,7 @@
     assert_raises_rpc_error,
     find_output,
     find_vout_for_address,
+    random_bytes,
 )
 from test_framework.wallet_util import bytes_to_wif
 
@@ -775,5 +789,37 @@ def test_psbt_input_keys(psbt_input, keys):
             self.nodes[0].sendrawtransaction(rawtx)
             self.generate(self.nodes[0], 1)
 
+        self.log.info("Test decoding PSBT with per-input preimage types")
+        # note that the decodepsbt RPC doesn't check whether preimages and hashes match
+        hash_ripemd160, preimage_ripemd160 = random_bytes(20), random_bytes(50)
+        hash_sha256, preimage_sha256 = random_bytes(32), random_bytes(50)
+        hash_hash160, preimage_hash160 = random_bytes(20), random_bytes(50)
+        hash_hash256, preimage_hash256 = random_bytes(32), random_bytes(50)
+
+        tx = CTransaction()
+        tx.vin = [CTxIn(outpoint=COutPoint(hash=int('aa' * 32, 16), n=0), scriptSig=b""),
+                  CTxIn(outpoint=COutPoint(hash=int('bb' * 32, 16), n=0), scriptSig=b""),
+                  CTxIn(outpoint=COutPoint(hash=int('cc' * 32, 16), n=0), scriptSig=b""),
+                  CTxIn(outpoint=COutPoint(hash=int('dd' * 32, 16), n=0), scriptSig=b"")]
+        tx.vout = [CTxOut(nValue=0, scriptPubKey=b"")]
+        psbt = PSBT()
+        psbt.g = PSBTMap({PSBT_GLOBAL_UNSIGNED_TX: tx.serialize()})
+        psbt.i = [PSBTMap({bytes([PSBT_IN_RIPEMD160]) + hash_ripemd160: preimage_ripemd160}),
+                  PSBTMap({bytes([PSBT_IN_SHA256]) + hash_sha256: preimage_sha256}),
+                  PSBTMap({bytes([PSBT_IN_HASH160]) + hash_hash160: preimage_hash160}),
+                  PSBTMap({bytes([PSBT_IN_HASH256]) + hash_hash256: preimage_hash256})]
+        psbt.o = [PSBTMap()]
+        res_inputs = self.nodes[0].decodepsbt(psbt.to_base64())["inputs"]
+        assert_equal(len(res_inputs), 4)
+        preimage_keys = ["ripemd160_preimages", "sha256_preimages", "hash160_preimages", "hash256_preimages"]
+        expected_hashes = [hash_ripemd160, hash_sha256, hash_hash160, hash_hash256]
+        expected_preimages = [preimage_ripemd160, preimage_sha256, preimage_hash160, preimage_hash256]
+        for res_input, preimage_key, hash, preimage in zip(res_inputs, preimage_keys, expected_hashes, expected_preimages):
+            assert preimage_key in res_input
+            assert_equal(len(res_input[preimage_key]), 1)
+            assert hash.hex() in res_input[preimage_key]
+            assert_equal(res_input[preimage_key][hash.hex()], preimage.hex())
+
+
 if __name__ == '__main__':
     PSBTTest().main()

test/functional/test_framework/messages.py

@@ -208,6 +208,20 @@ def tx_from_hex(hex_string):
     return from_hex(CTransaction(), hex_string)
 
 
+# like from_hex, but without the hex part
+def from_binary(cls, stream):
+    """deserialize a binary stream (or bytes object) into an object"""
+    # handle bytes object by turning it into a stream
+    was_bytes = isinstance(stream, bytes)
+    if was_bytes:
+        stream = BytesIO(stream)
+    obj = cls()
+    obj.deserialize(stream)
+    if was_bytes:
+        assert len(stream.read()) == 0
+    return obj
+
+
 # Objects that map to bitcoind objects, which can be serialized/deserialized
 
 

test/functional/test_framework/psbt.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+# Copyright (c) 2022 The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+import base64
+
+from .messages import (
+    CTransaction,
+    deser_string,
+    from_binary,
+    ser_compact_size,
+)
+
+
+# global types
+PSBT_GLOBAL_UNSIGNED_TX = 0x00
+PSBT_GLOBAL_XPUB = 0x01
+PSBT_GLOBAL_TX_VERSION = 0x02
+PSBT_GLOBAL_FALLBACK_LOCKTIME = 0x03
+PSBT_GLOBAL_INPUT_COUNT = 0x04
+PSBT_GLOBAL_OUTPUT_COUNT = 0x05
+PSBT_GLOBAL_TX_MODIFIABLE = 0x06
+PSBT_GLOBAL_VERSION = 0xfb
+PSBT_GLOBAL_PROPRIETARY = 0xfc
+
+# per-input types
+PSBT_IN_NON_WITNESS_UTXO = 0x00
+PSBT_IN_WITNESS_UTXO = 0x01
+PSBT_IN_PARTIAL_SIG = 0x02
+PSBT_IN_SIGHASH_TYPE = 0x03
+PSBT_IN_REDEEM_SCRIPT = 0x04
+PSBT_IN_WITNESS_SCRIPT = 0x05
+PSBT_IN_BIP32_DERIVATION = 0x06
+PSBT_IN_FINAL_SCRIPTSIG = 0x07
+PSBT_IN_FINAL_SCRIPTWITNESS = 0x08
+PSBT_IN_POR_COMMITMENT = 0x09
+PSBT_IN_RIPEMD160 = 0x0a
+PSBT_IN_SHA256 = 0x0b
+PSBT_IN_HASH160 = 0x0c
+PSBT_IN_HASH256 = 0x0d
+PSBT_IN_PREVIOUS_TXID = 0x0e
+PSBT_IN_OUTPUT_INDEX = 0x0f
+PSBT_IN_SEQUENCE = 0x10
+PSBT_IN_REQUIRED_TIME_LOCKTIME = 0x11
+PSBT_IN_REQUIRED_HEIGHT_LOCKTIME = 0x12
+PSBT_IN_TAP_KEY_SIG = 0x13
+PSBT_IN_TAP_SCRIPT_SIG = 0x14
+PSBT_IN_TAP_LEAF_SCRIPT = 0x15
+PSBT_IN_TAP_BIP32_DERIVATION = 0x16
+PSBT_IN_TAP_INTERNAL_KEY = 0x17
+PSBT_IN_TAP_MERKLE_ROOT = 0x18
+PSBT_IN_PROPRIETARY = 0xfc
+
+# per-output types
+PSBT_OUT_REDEEM_SCRIPT = 0x00
+PSBT_OUT_WITNESS_SCRIPT = 0x01
+PSBT_OUT_BIP32_DERIVATION = 0x02
+PSBT_OUT_AMOUNT = 0x03
+PSBT_OUT_SCRIPT = 0x04
+PSBT_OUT_TAP_INTERNAL_KEY = 0x05
+PSBT_OUT_TAP_TREE = 0x06
+PSBT_OUT_TAP_BIP32_DERIVATION = 0x07
+PSBT_OUT_PROPRIETARY = 0xfc
+
+
+class PSBTMap:
+    """Class for serializing and deserializing PSBT maps"""
+
+    def __init__(self, map=None):
+        self.map = map if map is not None else {}
+
+    def deserialize(self, f):
+        m = {}
+        while True:
+            k = deser_string(f)
+            if len(k) == 0:
+                break
+            v = deser_string(f)
+            if len(k) == 1:
+                k = k[0]
+            assert k not in m
+            m[k] = v
+        self.map = m
+
+    def serialize(self):
+        m = b""
+        for k,v in self.map.items():
+            if isinstance(k, int) and 0 <= k and k <= 255:
+                k = bytes([k])
+            m += ser_compact_size(len(k)) + k
+            m += ser_compact_size(len(v)) + v
+        m += b"\x00"
+        return m
+
+class PSBT:
+    """Class for serializing and deserializing PSBTs"""
+
+    def __init__(self):
+        self.g = PSBTMap()
+        self.i = []
+        self.o = []
+        self.tx = None
+
+    def deserialize(self, f):
+        assert f.read(5) == b"psbt\xff"
+        self.g = from_binary(PSBTMap, f)
+        assert 0 in self.g.map
+        self.tx = from_binary(CTransaction, self.g.map[0])
+        self.i = [from_binary(PSBTMap, f) for _ in self.tx.vin]
+        self.o = [from_binary(PSBTMap, f) for _ in self.tx.vout]
+        return self
+
+    def serialize(self):
+        assert isinstance(self.g, PSBTMap)
+        assert isinstance(self.i, list) and all(isinstance(x, PSBTMap) for x in self.i)
+        assert isinstance(self.o, list) and all(isinstance(x, PSBTMap) for x in self.o)
+        assert 0 in self.g.map
+        tx = from_binary(CTransaction, self.g.map[0])
+        assert len(tx.vin) == len(self.i)
+        assert len(tx.vout) == len(self.o)
+
+        psbt = [x.serialize() for x in [self.g] + self.i + self.o]
+        return b"psbt\xff" + b"".join(psbt)
+
+    def to_base64(self):
+        return base64.b64encode(self.serialize()).decode("utf8")
+
+    @classmethod
+    def from_base64(cls, b64psbt):
+        return from_binary(cls, base64.b64decode(b64psbt))