Merge bitcoin/bitcoin#21945: test: add P2PK support to MiniWallet

4bea301 test: use P2PK-MiniWallet for feature_csv_activation.py (Sebastian Falbesoner)
dc7eb64 test: MiniWallet: add P2PK support (Sebastian Falbesoner)

Pull request description:

  This PR adds support for creating and spending transactions with raw pubkey (P2PK) outputs to MiniWallet, [as suggested by MarcoFalke](bitcoin/bitcoin#21900 (comment)). Using that  mode in the test `feature_csv_activation.py`, all txs submitted to the mempool follow the standard policy, i.e. `-acceptnonstdtxn=1` can be removed.

  Possible follow-ups:
  * Improve MiniWallet constructor Interface; an enum-like parameter instead of two booleans would probably be better
  * Look at other tests that could benefit from P2PK (e.g. feature_cltv.py?)
  * Check vsize also for P2PK txs (vsize varies due to signature, i.e. a range has to be asserted)

ACKs for top commit:
  laanwj:
    Code review ACK 4bea301

Tree-SHA512: 9b428e6b7cfde59a8c7955d5096cea88af1384a5f49723f00052e9884d819d952d20a5ab39bb02f9d8b6073769c44462aa265d84a33e33da33c2d21670c488a6

Author: MarcoFalke

test/functional/feature_csv_activation.py

@@ -90,7 +90,6 @@ def set_test_params(self):
         self.setup_clean_chain = True
         self.extra_args = [[
             '[email protected]',
-            '-acceptnonstdtxn=1',
             '-par=1',  # Use only one script thread to get the exact reject reason for testing
         ]]
         self.supports_cli = False
@@ -103,12 +102,14 @@ def create_self_transfer_from_utxo(self, input_tx):
     def create_bip112special(self, input, txversion):
         tx = self.create_self_transfer_from_utxo(input)
         tx.nVersion = txversion
+        self.miniwallet.sign_tx(tx)
         tx.vin[0].scriptSig = CScript([-1, OP_CHECKSEQUENCEVERIFY, OP_DROP] + list(CScript(tx.vin[0].scriptSig)))
         return tx
 
     def create_bip112emptystack(self, input, txversion):
         tx = self.create_self_transfer_from_utxo(input)
         tx.nVersion = txversion
+        self.miniwallet.sign_tx(tx)
         tx.vin[0].scriptSig = CScript([OP_CHECKSEQUENCEVERIFY] + list(CScript(tx.vin[0].scriptSig)))
         return tx
 
@@ -126,6 +127,7 @@ def create_bip68txs(self, bip68inputs, txversion, locktime_delta=0):
             tx = self.create_self_transfer_from_utxo(bip68inputs[i])
             tx.nVersion = txversion
             tx.vin[0].nSequence = locktime + locktime_delta
+            self.miniwallet.sign_tx(tx)
             tx.rehash()
             txs.append({'tx': tx, 'sdf': sdf, 'stf': stf})
 
@@ -143,6 +145,7 @@ def create_bip112txs(self, bip112inputs, varyOP_CSV, txversion, locktime_delta=0
             else:  # vary nSequence instead, OP_CSV is fixed
                 tx.vin[0].nSequence = locktime + locktime_delta
             tx.nVersion = txversion
+            self.miniwallet.sign_tx(tx)
             if (varyOP_CSV):
                 tx.vin[0].scriptSig = CScript([locktime, OP_CHECKSEQUENCEVERIFY, OP_DROP] + list(CScript(tx.vin[0].scriptSig)))
             else:
@@ -178,7 +181,7 @@ def send_blocks(self, blocks, success=True, reject_reason=None):
 
     def run_test(self):
         self.helper_peer = self.nodes[0].add_p2p_connection(P2PDataStore())
-        self.miniwallet = MiniWallet(self.nodes[0], raw_script=True)
+        self.miniwallet = MiniWallet(self.nodes[0], use_p2pk=True)
 
         self.log.info("Generate blocks in the past for coinbase outputs.")
         long_past_time = int(time.time()) - 600 * 1000  # enough to build up to 1000 blocks 10 minutes apart without worrying about getting into the future
@@ -285,7 +288,7 @@ def run_test(self):
         success_txs = []
         # BIP113 tx, -1 CSV tx and empty stack CSV tx should succeed
         bip113tx_v1.nLockTime = self.last_block_time - 600 * 5  # = MTP of prior block (not <) but < time put on current block
-        bip113tx_v1.rehash()
+        self.miniwallet.sign_tx(bip113tx_v1)
         success_txs.append(bip113tx_v1)
         success_txs.append(bip112tx_special_v1)
         success_txs.append(bip112tx_emptystack_v1)
@@ -305,7 +308,7 @@ def run_test(self):
         success_txs = []
         # BIP113 tx, -1 CSV tx and empty stack CSV tx should succeed
         bip113tx_v2.nLockTime = self.last_block_time - 600 * 5  # = MTP of prior block (not <) but < time put on current block
-        bip113tx_v2.rehash()
+        self.miniwallet.sign_tx(bip113tx_v2)
         success_txs.append(bip113tx_v2)
         success_txs.append(bip112tx_special_v2)
         success_txs.append(bip112tx_emptystack_v2)
@@ -331,16 +334,20 @@ def run_test(self):
         self.log.info("BIP 113 tests")
         # BIP 113 tests should now fail regardless of version number if nLockTime isn't satisfied by new rules
         bip113tx_v1.nLockTime = self.last_block_time - 600 * 5  # = MTP of prior block (not <) but < time put on current block
+        self.miniwallet.sign_tx(bip113tx_v1)
         bip113tx_v1.rehash()
         bip113tx_v2.nLockTime = self.last_block_time - 600 * 5  # = MTP of prior block (not <) but < time put on current block
+        self.miniwallet.sign_tx(bip113tx_v2)
         bip113tx_v2.rehash()
         for bip113tx in [bip113tx_v1, bip113tx_v2]:
             self.send_blocks([self.create_test_block([bip113tx])], success=False, reject_reason='bad-txns-nonfinal')
 
         # BIP 113 tests should now pass if the locktime is < MTP
         bip113tx_v1.nLockTime = self.last_block_time - 600 * 5 - 1  # < MTP of prior block
+        self.miniwallet.sign_tx(bip113tx_v1)
         bip113tx_v1.rehash()
         bip113tx_v2.nLockTime = self.last_block_time - 600 * 5 - 1  # < MTP of prior block
+        self.miniwallet.sign_tx(bip113tx_v2)
         bip113tx_v2.rehash()
         for bip113tx in [bip113tx_v1, bip113tx_v2]:
             self.send_blocks([self.create_test_block([bip113tx])])
@@ -465,6 +472,7 @@ def run_test(self):
         time_txs = []
         for tx in [tx['tx'] for tx in bip112txs_vary_OP_CSV_v2 if not tx['sdf'] and tx['stf']]:
             tx.vin[0].nSequence = BASE_RELATIVE_LOCKTIME | SEQ_TYPE_FLAG
+            self.miniwallet.sign_tx(tx)
             tx.rehash()
             time_txs.append(tx)
 

test/functional/test_framework/wallet.py

@@ -6,6 +6,7 @@
 
 from decimal import Decimal
 from test_framework.address import ADDRESS_BCRT1_P2WSH_OP_TRUE
+from test_framework.key import ECKey
 from test_framework.messages import (
     COIN,
     COutPoint,
@@ -16,8 +17,11 @@
 )
 from test_framework.script import (
     CScript,
+    LegacySignatureHash,
+    OP_CHECKSIG,
     OP_TRUE,
     OP_NOP,
+    SIGHASH_ALL,
 )
 from test_framework.util import (
     assert_equal,
@@ -27,12 +31,20 @@
 
 
 class MiniWallet:
-    def __init__(self, test_node, *, raw_script=False):
+    def __init__(self, test_node, *, raw_script=False, use_p2pk=False):
         self._test_node = test_node
         self._utxos = []
+        self._priv_key = None
+        self._address = None
+
         if raw_script:
-            self._address = None
             self._scriptPubKey = bytes(CScript([OP_TRUE]))
+        elif use_p2pk:
+            # use simple deterministic private key (k=1)
+            self._priv_key = ECKey()
+            self._priv_key.set((1).to_bytes(32, 'big'), True)
+            pub_key = self._priv_key.get_pubkey()
+            self._scriptPubKey = bytes(CScript([pub_key.get_bytes(), OP_CHECKSIG]))
         else:
             self._address = ADDRESS_BCRT1_P2WSH_OP_TRUE
             self._scriptPubKey = hex_str_to_bytes(self._test_node.validateaddress(self._address)['scriptPubKey'])
@@ -50,6 +62,13 @@ def scan_tx(self, tx):
             if out['scriptPubKey']['hex'] == self._scriptPubKey.hex():
                 self._utxos.append({'txid': tx['txid'], 'vout': out['n'], 'value': out['value']})
 
+    def sign_tx(self, tx):
+        """Sign tx that has been created by MiniWallet in P2PK mode"""
+        assert self._priv_key is not None
+        (sighash, err) = LegacySignatureHash(CScript(self._scriptPubKey), tx, 0, SIGHASH_ALL)
+        assert err is None
+        tx.vin[0].scriptSig = CScript([self._priv_key.sign_ecdsa(sighash) + bytes(bytearray([SIGHASH_ALL]))])
+
     def generate(self, num_blocks):
         """Generate blocks with coinbase outputs to the internal address, and append the outputs to the internal list"""
         blocks = self._test_node.generatetodescriptor(num_blocks, f'raw({self._scriptPubKey.hex()})')
@@ -99,7 +118,12 @@ def create_self_transfer(self, *, fee_rate=Decimal("0.003"), from_node, utxo_to_
         tx.vout = [CTxOut(int(send_value * COIN), self._scriptPubKey)]
         if not self._address:
             # raw script
-            tx.vin[0].scriptSig = CScript([OP_NOP] * 35)  # pad to identical size
+            if self._priv_key is not None:
+                # P2PK, need to sign
+                self.sign_tx(tx)
+            else:
+                # anyone-can-spend
+                tx.vin[0].scriptSig = CScript([OP_NOP] * 35)  # pad to identical size
         else:
             tx.wit.vtxinwit = [CTxInWitness()]
             tx.wit.vtxinwit[0].scriptWitness.stack = [CScript([OP_TRUE])]
@@ -108,7 +132,10 @@ def create_self_transfer(self, *, fee_rate=Decimal("0.003"), from_node, utxo_to_
         tx_info = from_node.testmempoolaccept([tx_hex])[0]
         assert_equal(mempool_valid, tx_info['allowed'])
         if mempool_valid:
-            assert_equal(tx_info['vsize'], vsize)
+            # TODO: for P2PK, vsize is not constant due to varying scriptSig length,
+            # so only check this for anyone-can-spend outputs right now
+            if self._priv_key is None:
+                assert_equal(tx_info['vsize'], vsize)
             assert_equal(tx_info['fees']['base'], fee)
         return {'txid': tx_info['txid'], 'wtxid': tx_info['wtxid'], 'hex': tx_hex, 'tx': tx}