mempool_datacarrier.py raw

   1  #!/usr/bin/env python3
   2  # Copyright (c) 2020-2022 The Limenka developers
   3  # Distributed under the MIT software license, see the accompanying
   4  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
   5  """Test datacarrier functionality"""
   6  from test_framework.messages import (
   7      COutPoint,
   8      CTransaction,
   9      CTxIn,
  10      CTxInWitness,
  11      CTxOut,
  12      MAX_OP_RETURN_RELAY,
  13  )
  14  from test_framework.script import (
  15      CScript,
  16      OP_1,
  17      OP_2DROP,
  18      OP_DROP,
  19      OP_RETURN,
  20      taproot_construct,
  21  )
  22  from test_framework.test_framework import LimenkaTestFramework
  23  from test_framework.test_node import TestNode
  24  from test_framework.util import assert_raises_rpc_error
  25  from test_framework.wallet import MiniWallet
  26  
  27  from random import randbytes
  28  
  29  
  30  class DataCarrierTest(LimenkaTestFramework):
  31      def set_test_params(self):
  32          self.num_nodes = 4
  33          self.extra_args = [
  34              ["-acceptnonstddatacarrier=1", "-datacarrierfullcount"],
  35              ["-datacarrier=0"],
  36              ["-datacarrier=1", f"-datacarriersize={MAX_OP_RETURN_RELAY - 1}"],
  37              ["-datacarrier=1", "-datacarriersize=2", "-acceptnonstddatacarrier=1", "-datacarrierfullcount"],
  38          ]
  39  
  40      def test_null_data_transaction(self, node: TestNode, data, success: bool) -> None:
  41          tx = self.wallet.create_self_transfer(fee_rate=0)["tx"]
  42          data = [] if data is None else [data]
  43          tx.vout.append(CTxOut(nValue=0, scriptPubKey=CScript([OP_RETURN] + data)))
  44          tx.vout[0].nValue -= tx.get_vsize()  # simply pay 1sat/vbyte fee
  45  
  46          tx_hex = tx.serialize().hex()
  47  
  48          if success:
  49              self.wallet.sendrawtransaction(from_node=node, tx_hex=tx_hex)
  50              assert tx.rehash() in node.getrawmempool(True), f'{tx_hex} not in mempool'
  51          else:
  52              assert_raises_rpc_error(-26, "scriptpubkey", self.wallet.sendrawtransaction, from_node=node, tx_hex=tx_hex)
  53  
  54      def test_opnet_transaction(self, node: TestNode, success: bool) -> None:
  55          minimal_script = CScript([OP_2DROP, OP_DROP, b'op', OP_DROP, OP_1])
  56          internal_key = b'\x01' * 32
  57          tap = taproot_construct(internal_key, [("leaf", minimal_script), ("dummy", CScript([OP_1]))])
  58          leaf = tap.leaves["leaf"]
  59          control_block = bytes([leaf.version | tap.negflag]) + tap.internal_pubkey + leaf.merklebranch
  60          assert len(control_block) == 65
  61  
  62          utxo = self.wallet.get_utxo()
  63          funding_tx = CTransaction()
  64          funding_tx.vin = [CTxIn(COutPoint(int(utxo['txid'], 16), utxo['vout']))]
  65          funding_value = int(utxo['value'] * 100_000_000) - 1000
  66          funding_tx.vout = [CTxOut(funding_value, tap.scriptPubKey)]
  67          funding_tx.version = 2
  68          self.wallet.sign_tx(funding_tx)
  69          funding_tx.rehash()
  70          self.nodes[0].sendrawtransaction(funding_tx.serialize().hex())
  71          self.generate(self.nodes[0], 1, sync_fun=self.sync_blocks)
  72  
  73          spend_tx = CTransaction()
  74          spend_tx.version = 2
  75          spend_tx.vin = [CTxIn(COutPoint(int(funding_tx.hash, 16), 0))]
  76          spend_tx.vout = [CTxOut(funding_value - 1000, tap.scriptPubKey)]
  77          spend_tx.wit.vtxinwit = [CTxInWitness()]
  78          spend_tx.wit.vtxinwit[0].scriptWitness.stack = [
  79              b'',                    # stack[0]: empty (minimises opnet bytes)
  80              b'',                    # stack[1]: cleared by OP_2DROP
  81              b'',                    # stack[2]: cleared by OP_2DROP
  82              bytes(minimal_script),  # stack[3]: tapscript containing \x02op
  83              control_block,          # stack[4]: control block (65 bytes)
  84          ]
  85          tx_hex = spend_tx.serialize().hex()
  86  
  87          if success:
  88              self.wallet.sendrawtransaction(from_node=node, tx_hex=tx_hex)
  89              assert spend_tx.rehash() in node.getrawmempool(True)
  90          else:
  91              assert_raises_rpc_error(-26, "txn-datacarrier-exceeded",
  92                                      self.wallet.sendrawtransaction, from_node=node, tx_hex=tx_hex)
  93  
  94  
  95      def run_test(self):
  96          self.wallet = MiniWallet(self.nodes[0])
  97  
  98          # By default, only 80 bytes are used for data (+1 for OP_RETURN, +2 for the pushdata opcodes).
  99          default_size_data = randbytes(MAX_OP_RETURN_RELAY - 3)
 100          too_long_data = randbytes(MAX_OP_RETURN_RELAY - 2)
 101          small_data = randbytes(MAX_OP_RETURN_RELAY - 4)
 102          one_byte = randbytes(1)
 103          zero_bytes = randbytes(0)
 104  
 105          self.log.info("Testing null data transaction with default -datacarrier and -datacarriersize values.")
 106          self.test_null_data_transaction(node=self.nodes[0], data=default_size_data, success=True)
 107  
 108          self.log.info("Testing a null data transaction larger than allowed by the default -datacarriersize value.")
 109          self.test_null_data_transaction(node=self.nodes[0], data=too_long_data, success=False)
 110  
 111          self.log.info("Testing a null data transaction with -datacarrier=false.")
 112          self.test_null_data_transaction(node=self.nodes[1], data=default_size_data, success=False)
 113  
 114          self.log.info("Testing a null data transaction with a size larger than accepted by -datacarriersize.")
 115          self.test_null_data_transaction(node=self.nodes[2], data=default_size_data, success=False)
 116  
 117          self.log.info("Testing a null data transaction with a size smaller than accepted by -datacarriersize.")
 118          self.test_null_data_transaction(node=self.nodes[2], data=small_data, success=True)
 119  
 120          self.log.info("Testing a null data transaction with no data.")
 121          self.test_null_data_transaction(node=self.nodes[0], data=None, success=True)
 122          self.test_null_data_transaction(node=self.nodes[1], data=None, success=False)
 123          self.test_null_data_transaction(node=self.nodes[2], data=None, success=True)
 124          self.test_null_data_transaction(node=self.nodes[3], data=None, success=True)
 125  
 126          self.log.info("Testing a null data transaction with zero bytes of data.")
 127          self.test_null_data_transaction(node=self.nodes[0], data=zero_bytes, success=True)
 128          self.test_null_data_transaction(node=self.nodes[1], data=zero_bytes, success=False)
 129          self.test_null_data_transaction(node=self.nodes[2], data=zero_bytes, success=True)
 130          self.test_null_data_transaction(node=self.nodes[3], data=zero_bytes, success=True)
 131  
 132          self.log.info("Testing a null data transaction with one byte of data.")
 133          self.test_null_data_transaction(node=self.nodes[0], data=one_byte, success=True)
 134          self.test_null_data_transaction(node=self.nodes[1], data=one_byte, success=False)
 135          self.test_null_data_transaction(node=self.nodes[2], data=one_byte, success=True)
 136          self.test_null_data_transaction(node=self.nodes[3], data=one_byte, success=False)
 137  
 138          self.log.info("Testing an OPNet transaction (just pushing 'op') with default -datacarriersize.")
 139          self.test_opnet_transaction(node=self.nodes[0], success=True)
 140  
 141          self.log.info("Testing an OPNet transaction (just pushing 'op') with -datacarriersize=2.")
 142          self.test_opnet_transaction(node=self.nodes[3], success=False)
 143  
 144  
 145  if __name__ == '__main__':
 146      DataCarrierTest(__file__).main()
 147