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