blocktools.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2015-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 """Utilities for manipulating blocks and transactions."""
6
7 import struct
8 import time
9 import unittest
10
11 from .address import (
12 address_to_scriptpubkey,
13 key_to_p2sh_p2wpkh,
14 key_to_p2wpkh,
15 script_to_p2sh_p2wsh,
16 script_to_p2wsh,
17 )
18 from .messages import (
19 CBlock,
20 COIN,
21 COutPoint,
22 CTransaction,
23 CTxIn,
24 CTxInWitness,
25 CTxOut,
26 SEQUENCE_FINAL,
27 hash256,
28 ser_uint256,
29 tx_from_hex,
30 uint256_from_compact,
31 WITNESS_SCALE_FACTOR,
32 MAX_SEQUENCE_NONFINAL,
33 )
34 from .script import (
35 CScript,
36 CScriptNum,
37 CScriptOp,
38 OP_0,
39 OP_RETURN,
40 OP_TRUE,
41 )
42 from .script_util import (
43 key_to_p2pk_script,
44 key_to_p2wpkh_script,
45 keys_to_multisig_script,
46 script_to_p2wsh_script,
47 )
48 from .util import assert_equal
49
50 MAX_BLOCK_SIGOPS = 20000
51 MAX_BLOCK_SIGOPS_WEIGHT = MAX_BLOCK_SIGOPS * WITNESS_SCALE_FACTOR
52 MAX_STANDARD_TX_SIGOPS = 4000
53 MAX_STANDARD_TX_WEIGHT = 400000
54
55 # Genesis block time (regtest)
56 TIME_GENESIS_BLOCK = 1296688602
57
58 MAX_FUTURE_BLOCK_TIME = 2 * 60 * 60
59
60 # Coinbase transaction outputs can only be spent after this number of new blocks (network rule)
61 COINBASE_MATURITY = 100
62
63 # From BIP141
64 WITNESS_COMMITMENT_HEADER = b"\xaa\x21\xa9\xed"
65
66 NULL_OUTPOINT = COutPoint(0, 0xffffffff)
67
68 NORMAL_GBT_REQUEST_PARAMS = {"rules": ["segwit"]}
69 VERSIONBITS_LAST_OLD_BLOCK_VERSION = 4
70 MIN_BLOCKS_TO_KEEP = 288
71
72 REGTEST_RETARGET_PERIOD = 150
73
74 REGTEST_N_BITS = 0x207fffff # difficulty retargeting is disabled in REGTEST chainparams"
75 REGTEST_TARGET = 0x7fffff0000000000000000000000000000000000000000000000000000000000
76 assert_equal(uint256_from_compact(REGTEST_N_BITS), REGTEST_TARGET)
77
78 DIFF_1_N_BITS = 0x1d00ffff
79 DIFF_1_TARGET = 0x00000000ffff0000000000000000000000000000000000000000000000000000
80 assert_equal(uint256_from_compact(DIFF_1_N_BITS), DIFF_1_TARGET)
81
82 DIFF_4_N_BITS = 0x1c3fffc0
83 DIFF_4_TARGET = int(DIFF_1_TARGET / 4)
84 assert_equal(uint256_from_compact(DIFF_4_N_BITS), DIFF_4_TARGET)
85
86 # From BIP325
87 SIGNET_HEADER = b"\xec\xc7\xda\xa2"
88
89 # Number of blocks to create in temporary blockchain branch for reorg testing
90 FORK_LENGTH = 10
91
92 def nbits_str(nbits):
93 return f"{nbits:08x}"
94
95 def target_str(target):
96 return f"{target:064x}"
97
98 def create_block(hashprev=None, coinbase=None, *, ntime=None, height=None, version=None, tmpl=None, txlist=None):
99 """Create a block (with regtest difficulty)."""
100 block = CBlock()
101 if tmpl is None:
102 tmpl = {}
103 block.nVersion = version or tmpl.get('version') or VERSIONBITS_LAST_OLD_BLOCK_VERSION
104 block.nTime = ntime or tmpl.get('curtime') or int(time.time() + 600)
105 block.hashPrevBlock = hashprev or int(tmpl['previousblockhash'], 0x10)
106 if tmpl and tmpl.get('bits') is not None:
107 block.nBits = struct.unpack('>I', bytes.fromhex(tmpl['bits']))[0]
108 else:
109 block.nBits = REGTEST_N_BITS
110 if coinbase is None:
111 coinbase = create_coinbase(height=height or tmpl["height"])
112 block.vtx.append(coinbase)
113 if txlist:
114 for tx in txlist:
115 if type(tx) is str:
116 tx = tx_from_hex(tx)
117 block.vtx.append(tx)
118 block.hashMerkleRoot = block.calc_merkle_root()
119 return block
120
121 def create_empty_fork(node, fork_length=FORK_LENGTH):
122 '''
123 Creates a fork using node's chaintip as the starting point.
124 Returns a list of blocks to submit in order.
125 '''
126 tip = int(node.getbestblockhash(), 16)
127 height = node.getblockcount()
128 block_time = node.getblock(node.getbestblockhash())['time'] + 1
129
130 blocks = []
131 for _ in range(fork_length):
132 block = create_block(tip, height=height + 1, ntime=block_time)
133 block.solve()
134 blocks.append(block)
135 tip = block.hash_int
136 block_time += 1
137 height += 1
138
139 return blocks
140
141 def get_witness_script(witness_root, witness_nonce):
142 witness_commitment = hash256(ser_uint256(witness_root) + ser_uint256(witness_nonce))
143 output_data = WITNESS_COMMITMENT_HEADER + witness_commitment
144 return CScript([OP_RETURN, output_data])
145
146 def add_witness_commitment(block, nonce=0):
147 """Add a witness commitment to the block's coinbase transaction.
148
149 According to BIP141, blocks with witness rules active must commit to the
150 hash of all in-block transactions including witness."""
151 # First calculate the merkle root of the block's
152 # transactions, with witnesses.
153 witness_nonce = nonce
154 witness_root = block.calc_witness_merkle_root()
155 # witness_nonce should go to coinbase witness.
156 block.vtx[0].wit.vtxinwit = [CTxInWitness()]
157 block.vtx[0].wit.vtxinwit[0].scriptWitness.stack = [ser_uint256(witness_nonce)]
158
159 # witness commitment is the last OP_RETURN output in coinbase
160 block.vtx[0].vout.append(CTxOut(0, get_witness_script(witness_root, witness_nonce)))
161 block.hashMerkleRoot = block.calc_merkle_root()
162
163
164 def script_BIP34_coinbase_height(height, *, padding=True):
165 if height <= 16:
166 res = CScriptOp.encode_op_n(height)
167 if padding:
168 # Append dummy extraNonce to increase scriptSig size to 2 (see bad-cb-length consensus rule)
169 return CScript([res, OP_0])
170 return CScript([res])
171 return CScript([CScriptNum(height)])
172
173
174 def create_coinbase(height, pubkey=None, *, script_pubkey=None, extra_output_script=None, fees=0, nValue=50, halving_period=REGTEST_RETARGET_PERIOD):
175 """Create a coinbase transaction.
176
177 If pubkey is passed in, the coinbase output will be a P2PK output;
178 otherwise an anyone-can-spend output.
179
180 If extra_output_script is given, make a 0-value output to that
181 script. This is useful to pad block weight/sigops as needed. """
182 coinbase = CTransaction()
183 coinbase.nLockTime = height - 1
184 coinbase.vin.append(CTxIn(NULL_OUTPOINT, script_BIP34_coinbase_height(height), MAX_SEQUENCE_NONFINAL))
185 coinbaseoutput = CTxOut()
186 coinbaseoutput.nValue = nValue * COIN
187 if nValue == 50:
188 halvings = int(height / halving_period)
189 coinbaseoutput.nValue >>= halvings
190 coinbaseoutput.nValue += fees
191 if pubkey is not None:
192 coinbaseoutput.scriptPubKey = key_to_p2pk_script(pubkey)
193 elif script_pubkey is not None:
194 coinbaseoutput.scriptPubKey = script_pubkey
195 else:
196 coinbaseoutput.scriptPubKey = CScript([OP_TRUE])
197 coinbase.vout = [coinbaseoutput]
198 if extra_output_script is not None:
199 coinbaseoutput2 = CTxOut()
200 coinbaseoutput2.nValue = 0
201 coinbaseoutput2.scriptPubKey = extra_output_script
202 coinbase.vout.append(coinbaseoutput2)
203 return coinbase
204
205 def create_tx_with_script(prevtx, n, script_sig=b"", *, amount, output_script=None):
206 """Return one-input, one-output transaction object
207 spending the prevtx's n-th output with the given amount.
208
209 Can optionally pass scriptPubKey and scriptSig, default is anyone-can-spend output.
210 """
211 if output_script is None:
212 output_script = CScript()
213 tx = CTransaction()
214 assert n < len(prevtx.vout)
215 tx.vin.append(CTxIn(COutPoint(prevtx.txid_int, n), script_sig, SEQUENCE_FINAL))
216 tx.vout.append(CTxOut(amount, output_script))
217 return tx
218
219 def get_legacy_sigopcount_block(block, accurate=True):
220 count = 0
221 for tx in block.vtx:
222 count += get_legacy_sigopcount_tx(tx, accurate)
223 return count
224
225 def get_legacy_sigopcount_tx(tx, accurate=True):
226 count = 0
227 for i in tx.vout:
228 count += i.scriptPubKey.GetSigOpCount(accurate)
229 for j in tx.vin:
230 # scriptSig might be of type bytes, so convert to CScript for the moment
231 count += CScript(j.scriptSig).GetSigOpCount(accurate)
232 return count
233
234 def witness_script(use_p2wsh, pubkey):
235 """Create a scriptPubKey for a pay-to-witness TxOut.
236
237 This is either a P2WPKH output for the given pubkey, or a P2WSH output of a
238 1-of-1 multisig for the given pubkey. Returns the hex encoding of the
239 scriptPubKey."""
240 if not use_p2wsh:
241 # P2WPKH instead
242 pkscript = key_to_p2wpkh_script(pubkey)
243 else:
244 # 1-of-1 multisig
245 witness_script = keys_to_multisig_script([pubkey])
246 pkscript = script_to_p2wsh_script(witness_script)
247 return pkscript.hex()
248
249 def create_witness_tx(node, use_p2wsh, utxo, pubkey, encode_p2sh, amount):
250 """Return a transaction (in hex) that spends the given utxo to a segwit output.
251
252 Optionally wrap the segwit output using P2SH."""
253 if use_p2wsh:
254 program = keys_to_multisig_script([pubkey])
255 addr = script_to_p2sh_p2wsh(program) if encode_p2sh else script_to_p2wsh(program)
256 else:
257 addr = key_to_p2sh_p2wpkh(pubkey) if encode_p2sh else key_to_p2wpkh(pubkey)
258 if not encode_p2sh:
259 assert_equal(address_to_scriptpubkey(addr).hex(), witness_script(use_p2wsh, pubkey))
260 return node.createrawtransaction([utxo], {addr: amount})
261
262 def send_to_witness(use_p2wsh, node, utxo, pubkey, encode_p2sh, amount, sign=True, insert_redeem_script=""):
263 """Create a transaction spending a given utxo to a segwit output.
264
265 The output corresponds to the given pubkey: use_p2wsh determines whether to
266 use P2WPKH or P2WSH; encode_p2sh determines whether to wrap in P2SH.
267 sign=True will have the given node sign the transaction.
268 insert_redeem_script will be added to the scriptSig, if given."""
269 tx_to_witness = create_witness_tx(node, use_p2wsh, utxo, pubkey, encode_p2sh, amount)
270 if (sign):
271 signed = node.signrawtransactionwithwallet(tx_to_witness)
272 assert "errors" not in signed
273 return node.sendrawtransaction(signed["hex"])
274 else:
275 if (insert_redeem_script):
276 tx = tx_from_hex(tx_to_witness)
277 tx.vin[0].scriptSig += CScript([bytes.fromhex(insert_redeem_script)])
278 tx_to_witness = tx.serialize().hex()
279
280 return node.sendrawtransaction(tx_to_witness)
281
282 class TestFrameworkBlockTools(unittest.TestCase):
283 def test_create_block_prefers_explicit_height(self):
284 block = create_block(
285 hashprev=1,
286 tmpl={"height": 100},
287 height=200,
288 )
289 assert_equal(CScriptNum.decode(block.vtx[0].vin[0].scriptSig), 200)
290
291 def test_create_coinbase(self):
292 height = 20
293 coinbase_tx = create_coinbase(height=height)
294 assert_equal(CScriptNum.decode(coinbase_tx.vin[0].scriptSig), height)
295