tool_utxo_to_sqlite.py raw
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()
157