tool_utxo_to_sqlite.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2024-present 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 utxo-to-sqlite conversion tool"""
6 import os
7 try:
8 import sqlite3
9 except ImportError:
10 pass
11 import platform
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 )
20 from test_framework.crypto.muhash import MuHash3072
21 from test_framework.script import (
22 CScript,
23 CScriptOp,
24 )
25 from test_framework.script_util import (
26 PAY_TO_ANCHOR,
27 key_to_p2pk_script,
28 key_to_p2pkh_script,
29 key_to_p2wpkh_script,
30 keys_to_multisig_script,
31 output_key_to_p2tr_script,
32 script_to_p2sh_script,
33 script_to_p2wsh_script,
34 )
35 from test_framework.test_framework import LimenkaTestFramework
36 from test_framework.util import (
37 assert_equal,
38 )
39 from test_framework.wallet import MiniWallet
40
41
42 def calculate_muhash_from_sqlite_utxos(filename):
43 muhash = MuHash3072()
44 con = sqlite3.connect(filename)
45 cur = con.cursor()
46 for (txid_hex, vout, value, coinbase, height, spk_hex) in cur.execute("SELECT * FROM utxos"):
47 # serialize UTXO for MuHash (see function `TxOutSer` in the coinstats module)
48 utxo_ser = COutPoint(int(txid_hex, 16), vout).serialize()
49 utxo_ser += (height * 2 + coinbase).to_bytes(4, 'little')
50 utxo_ser += CTxOut(value, bytes.fromhex(spk_hex)).serialize()
51 muhash.insert(utxo_ser)
52 con.close()
53 return muhash.digest()[::-1].hex()
54
55
56 class UtxoToSqliteTest(LimenkaTestFramework):
57 def set_test_params(self):
58 self.num_nodes = 1
59 # we want to create some UTXOs with non-standard output scripts
60 self.extra_args = [['-acceptnonstdtxn=1']]
61
62 def skip_test_if_missing_module(self):
63 self.skip_if_no_py_sqlite3()
64
65 def run_test(self):
66 node = self.nodes[0]
67 wallet = MiniWallet(node)
68 key = ECKey()
69
70 self.log.info('Test that oversized output scripts are rejected')
71 key.generate(compressed=False)
72 uncompressed_pubkey = key.get_pubkey().get_bytes()
73 key.generate(compressed=True)
74 pubkey = key.get_pubkey().get_bytes()
75
76 # Test that scripts exceeding MAX_OUTPUT_SCRIPT_SIZE=34 are rejected
77 invalid_scripts = [
78 (key_to_p2pk_script(pubkey), "P2PK compressed (35 bytes)"),
79 (key_to_p2pk_script(uncompressed_pubkey), "P2PK uncompressed (67 bytes)"),
80 (keys_to_multisig_script([pubkey]), "Bare multisig 1-of-1 (37 bytes)"),
81 (keys_to_multisig_script([uncompressed_pubkey]*2), "Bare multisig 2-of-2 uncompressed"),
82 (CScript([CScriptOp.encode_op_n(1)]*1000), "Large script (1000 bytes)"),
83 ]
84
85 for script, description in invalid_scripts:
86 try:
87 wallet.send_to(from_node=node, scriptPubKey=script, amount=1, fee=20000)
88 raise AssertionError(f"{description} should have been rejected")
89 except Exception as e:
90 assert 'bad-txns-vout-script-toolarge' in str(e), \
91 f"{description} rejected with wrong error: {e}"
92 self.log.info(f" ✓ {description} correctly rejected")
93
94 self.log.info('Create UTXOs with valid output script types (≤34 bytes)')
95 for i in range(1, 10+1):
96 key.generate(compressed=True)
97 pubkey = key.get_pubkey().get_bytes()
98
99 # Only include output scripts that comply with MAX_OUTPUT_SCRIPT_SIZE=34
100 output_scripts = (
101 key_to_p2pkh_script(pubkey), # 25 bytes
102 script_to_p2sh_script(key_to_p2pkh_script(pubkey)), # 23 bytes
103 key_to_p2wpkh_script(pubkey), # 22 bytes
104 script_to_p2wsh_script(key_to_p2pkh_script(pubkey)),# 34 bytes
105 output_key_to_p2tr_script(pubkey[1:]), # 34 bytes
106 PAY_TO_ANCHOR, # 4 bytes
107 )
108
109 # create outputs and mine them in a block
110 for output_script in output_scripts:
111 wallet.send_to(from_node=node, scriptPubKey=output_script, amount=i, fee=20000)
112 self.generate(wallet, 1)
113
114 self.log.info('Dump UTXO set via `dumptxoutset` RPC')
115 input_filename = os.path.join(self.options.tmpdir, "utxos.dat")
116 node.dumptxoutset(input_filename, "latest")
117
118 self.log.info('Convert UTXO set from compact-serialized format to sqlite format')
119 output_filename = os.path.join(self.options.tmpdir, "utxos.sqlite")
120 base_dir = self.config["environment"]["SRCDIR"]
121 utxo_to_sqlite_path = os.path.join(base_dir, "contrib", "utxo-tools", "utxo_to_sqlite.py")
122 subprocess.run([sys.executable, utxo_to_sqlite_path, input_filename, output_filename],
123 check=True, stderr=subprocess.STDOUT)
124
125 self.log.info('Verify that both UTXO sets match by comparing their MuHash')
126 muhash_sqlite = calculate_muhash_from_sqlite_utxos(output_filename)
127 muhash_compact_serialized = node.gettxoutsetinfo('muhash')['muhash']
128 assert_equal(muhash_sqlite, muhash_compact_serialized)
129
130 if platform.system() != "Windows": # FIFOs are not available on Windows
131 self.log.info('Convert UTXO set directly (without intermediate dump) via named pipe')
132 fifo_filename = os.path.join(self.options.tmpdir, "utxos.fifo")
133 os.mkfifo(fifo_filename)
134 output_direct_filename = os.path.join(self.options.tmpdir, "utxos_direct.sqlite")
135 p = subprocess.Popen([sys.executable, utxo_to_sqlite_path, fifo_filename, output_direct_filename],
136 stderr=subprocess.STDOUT)
137 node.dumptxoutset(fifo_filename, "latest")
138 p.wait(timeout=10)
139 muhash_direct_sqlite = calculate_muhash_from_sqlite_utxos(output_direct_filename)
140 assert_equal(muhash_sqlite, muhash_direct_sqlite)
141 os.remove(fifo_filename)
142
143
144 if __name__ == "__main__":
145 UtxoToSqliteTest(__file__).main()
146