1  #!/usr/bin/env python3
  2  # Copyright (c) 2024-present The Bitcoin Core 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 utxo-to-sqlite conversion tool"""
  6  from itertools import product
  7  import os.path
  8  try:
  9      import sqlite3
 10  except ImportError:
 11      pass
 12  import subprocess
 13  import sys
 14  
 15  from test_framework.key import ECKey
 16  from test_framework.messages import (
 17      COutPoint,
 18      CTxOut,
 19      uint256_from_str,
 20  )
 21  from test_framework.crypto.muhash import MuHash3072
 22  from test_framework.script import (
 23      CScript,
 24      CScriptOp,
 25  )
 26  from test_framework.script_util import (
 27      PAY_TO_ANCHOR,
 28      key_to_p2pk_script,
 29      key_to_p2pkh_script,
 30      key_to_p2wpkh_script,
 31      keys_to_multisig_script,
 32      output_key_to_p2tr_script,
 33      script_to_p2sh_script,
 34      script_to_p2wsh_script,
 35  )
 36  from test_framework.test_framework import BitcoinTestFramework
 37  from test_framework.util import (
 38      assert_equal,
 39  )
 40  from test_framework.wallet import MiniWallet
 41  
 42  
 43  def calculate_muhash_from_sqlite_utxos(filename, txid_format, spk_format):
 44      muhash = MuHash3072()
 45      con = sqlite3.connect(filename)
 46      cur = con.cursor()
 47      for (txid, vout, value, coinbase, height, spk) in cur.execute("SELECT * FROM utxos"):
 48          match txid_format:
 49              case "hex":
 50                  assert type(txid) is str
 51                  txid_bytes = bytes.fromhex(txid)[::-1]
 52              case "raw":
 53                  assert type(txid) is bytes
 54                  txid_bytes = txid
 55              case "rawle":
 56                  assert type(txid) is bytes
 57                  txid_bytes = txid[::-1]
 58          match spk_format:
 59              case "hex":
 60                  assert type(spk) is str
 61                  spk_bytes = bytes.fromhex(spk)
 62              case "raw":
 63                  assert type(spk) is bytes
 64                  spk_bytes = spk
 65  
 66          # serialize UTXO for MuHash (see function `TxOutSer` in the  coinstats module)
 67          utxo_ser = COutPoint(uint256_from_str(txid_bytes), vout).serialize()
 68          utxo_ser += (height * 2 + coinbase).to_bytes(4, 'little')
 69          utxo_ser += CTxOut(value, spk_bytes).serialize()
 70          muhash.insert(utxo_ser)
 71      con.close()
 72      return muhash.digest()[::-1].hex()
 73  
 74  
 75  class UtxoToSqliteTest(BitcoinTestFramework):
 76      def set_test_params(self):
 77          self.num_nodes = 1
 78          # we want to create some UTXOs with non-standard output scripts
 79          self.extra_args = [['-acceptnonstdtxn=1']]
 80  
 81      def skip_test_if_missing_module(self):
 82          self.skip_if_no_py_sqlite3()
 83  
 84      def run_test(self):
 85          node = self.nodes[0]
 86          wallet = MiniWallet(node)
 87          key = ECKey()
 88  
 89          self.log.info('Create UTXOs with various output script types')
 90          for i in range(1, 10+1):
 91              key.generate(compressed=False)
 92              uncompressed_pubkey = key.get_pubkey().get_bytes()
 93              key.generate(compressed=True)
 94              pubkey = key.get_pubkey().get_bytes()
 95  
 96              # add output scripts for compressed script type 0 (P2PKH), type 1 (P2SH),
 97              # types 2-3 (P2PK compressed), types 4-5 (P2PK uncompressed) and
 98              # for uncompressed scripts (bare multisig, segwit, etc.)
 99              output_scripts = (
100                  key_to_p2pkh_script(pubkey),
101                  script_to_p2sh_script(key_to_p2pkh_script(pubkey)),
102                  key_to_p2pk_script(pubkey),
103                  key_to_p2pk_script(uncompressed_pubkey),
104  
105                  keys_to_multisig_script([pubkey]*i),
106                  keys_to_multisig_script([uncompressed_pubkey]*i),
107                  key_to_p2wpkh_script(pubkey),
108                  script_to_p2wsh_script(key_to_p2pkh_script(pubkey)),
109                  output_key_to_p2tr_script(pubkey[1:]),
110                  PAY_TO_ANCHOR,
111                  CScript([CScriptOp.encode_op_n(i)]*(1000*i)),  # large script (up to 10000 bytes)
112              )
113  
114              # create outputs and mine them in a block
115              for output_script in output_scripts:
116                  wallet.send_to(from_node=node, scriptPubKey=output_script, amount=i, fee=20000)
117              self.generate(wallet, 1)
118  
119          self.log.info('Dump UTXO set via `dumptxoutset` RPC')
120          input_filename = os.path.join(self.options.tmpdir, "utxos.dat")
121          node.dumptxoutset(input_filename, "latest")
122  
123          for i, (txid_format, spk_format) in enumerate(product(["hex", "raw", "rawle"], ["hex", "raw"])):
124              self.log.info(f'Test utxo-to-sqlite script using txid format "{txid_format}" and spk format "{spk_format}" ({i+1})')
125              self.log.info('-> Convert UTXO set from compact-serialized format to sqlite format')
126              output_filename = os.path.join(self.options.tmpdir, f"utxos_{i+1}.sqlite")
127              base_dir = self.config["environment"]["SRCDIR"]
128              utxo_to_sqlite_path = os.path.join(base_dir, "contrib", "utxo-tools", "utxo_to_sqlite.py")
129              arguments = [input_filename, output_filename, f'--txid={txid_format}', f'--spk={spk_format}']
130              subprocess.run([sys.executable, utxo_to_sqlite_path] + arguments, check=True, stderr=subprocess.STDOUT)
131  
132              self.log.info('-> Verify that both UTXO sets match by comparing their MuHash')
133              muhash_sqlite = calculate_muhash_from_sqlite_utxos(output_filename, txid_format, spk_format)
134              muhash_compact_serialized = node.gettxoutsetinfo('muhash')['muhash']
135              assert_equal(muhash_sqlite, muhash_compact_serialized)
136              self.log.info('')
137  
138  
139  if __name__ == "__main__":
140      UtxoToSqliteTest(__file__).main()