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