Inside Bitcoin

1#!/usr/bin/env python3
2# Copyright (c) 2020-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
6import argparse
7import json
8import logging
9import math
10import os
11import re
12import shlex
13import sys
14import time
15import subprocess
16
17PATH_BASE_CONTRIB_SIGNET = os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
18PATH_BASE_TEST_FUNCTIONAL = os.path.abspath(os.path.join(PATH_BASE_CONTRIB_SIGNET, "..", "..", "test", "functional"))
19sys.path.insert(0, PATH_BASE_TEST_FUNCTIONAL)
20
21from test_framework.blocktools import get_witness_script, script_BIP34_coinbase_height, SIGNET_HEADER # noqa: E402
22from test_framework.messages import CBlock, CBlockHeader, COutPoint, CTransaction, CTxIn, CTxInWitness, CTxOut, from_binary, from_hex, ser_string, ser_uint256, tx_from_hex, MAX_SEQUENCE_NONFINAL # noqa: E402
23from test_framework.psbt import PSBT, PSBTMap, PSBT_GLOBAL_UNSIGNED_TX, PSBT_IN_FINAL_SCRIPTSIG, PSBT_IN_FINAL_SCRIPTWITNESS, PSBT_IN_NON_WITNESS_UTXO, PSBT_IN_SIGHASH_TYPE # noqa: E402
24from test_framework.script import CScript, CScriptOp # noqa: E402
25
26logging.basicConfig(
27    format='%(asctime)s %(levelname)s %(message)s',
28    level=logging.INFO,
29    datefmt='%Y-%m-%d %H:%M:%S')
30
31PSBT_SIGNET_BLOCK = b"\xfc\x06signetb"    # proprietary PSBT global field holding the block being signed
32RE_MULTIMINER = re.compile(r"^(\d+)(-(\d+))?/(\d+)$")
33
34def signet_txs(block, challenge):
35    # assumes signet solution has not been added yet so does not need
36    # to be removed
37
38    txs = block.vtx[:]
39    txs[0] = CTransaction(txs[0])
40    txs[0].vout[-1].scriptPubKey += CScriptOp.encode_op_pushdata(SIGNET_HEADER)
41    hashes = []
42    for tx in txs:
43        hashes.append(ser_uint256(tx.txid_int))
44    mroot = block.get_merkle_root(hashes)
45
46    sd = b""
47    sd += block.nVersion.to_bytes(4, "little", signed=True)
48    sd += ser_uint256(block.hashPrevBlock)
49    sd += ser_uint256(mroot)
50    sd += block.nTime.to_bytes(4, "little")
51
52    to_spend = CTransaction()
53    to_spend.version = 0
54    to_spend.nLockTime = 0
55    to_spend.vin = [CTxIn(COutPoint(0, 0xFFFFFFFF), b"\x00" + CScriptOp.encode_op_pushdata(sd), 0)]
56    to_spend.vout = [CTxOut(0, challenge)]
57
58    spend = CTransaction()
59    spend.version = 0
60    spend.nLockTime = 0
61    spend.vin = [CTxIn(COutPoint(to_spend.txid_int, 0), b"", 0)]
62    spend.vout = [CTxOut(0, b"\x6a")]
63
64    return spend, to_spend
65
66def decode_challenge_psbt(b64psbt):
67    psbt = PSBT.from_base64(b64psbt)
68
69    assert len(psbt.i) == 1
70    assert len(psbt.o) == 1
71    assert PSBT_SIGNET_BLOCK in psbt.g.map
72    return psbt
73
74def get_block_from_psbt(psbt):
75    return from_binary(CBlock, psbt.g.map[PSBT_SIGNET_BLOCK])
76
77def get_solution_from_psbt(psbt, emptyok=False):
78    scriptSig = psbt.i[0].map.get(PSBT_IN_FINAL_SCRIPTSIG, b"")
79    scriptWitness = psbt.i[0].map.get(PSBT_IN_FINAL_SCRIPTWITNESS, b"\x00")
80    if emptyok and len(scriptSig) == 0 and scriptWitness == b"\x00":
81        return None
82    return ser_string(scriptSig) + scriptWitness
83
84def finish_block(block, signet_solution, grind_cmd):
85    if signet_solution is None:
86        pass # Don't need to add a signet commitment if there's no signet signature needed
87    else:
88        block.vtx[0].vout[-1].scriptPubKey += CScriptOp.encode_op_pushdata(SIGNET_HEADER + signet_solution)
89        block.hashMerkleRoot = block.calc_merkle_root()
90    if grind_cmd is None:
91        block.solve()
92    else:
93        headhex = CBlockHeader.serialize(block).hex()
94        cmd = shlex.split(grind_cmd) + [headhex]
95        newheadhex = subprocess.run(cmd, stdout=subprocess.PIPE, input=b"", check=True).stdout.strip()
96        newhead = from_hex(CBlockHeader(), newheadhex.decode('utf8'))
97        block.nNonce = newhead.nNonce
98    return block
99
100def new_block(tmpl, reward_spk, *, blocktime=None, poolid=None):
101    scriptSig = script_BIP34_coinbase_height(tmpl["height"])
102    if poolid is not None:
103        scriptSig = CScript(b"" + scriptSig + CScriptOp.encode_op_pushdata(poolid))
104
105    cbtx = CTransaction()
106    cbtx.nLockTime = tmpl["height"] - 1
107    cbtx.vin = [CTxIn(COutPoint(0, 0xffffffff), scriptSig, MAX_SEQUENCE_NONFINAL)]
108    cbtx.vout = [CTxOut(tmpl["coinbasevalue"], reward_spk)]
109    cbtx.vin[0].nSequence = 2**32-2
110
111    block = CBlock()
112    block.nVersion = tmpl["version"]
113    block.hashPrevBlock = int(tmpl["previousblockhash"], 16)
114    block.nTime = tmpl["curtime"] if blocktime is None else blocktime
115    if block.nTime < tmpl["mintime"]:
116        block.nTime = tmpl["mintime"]
117    block.nBits = int(tmpl["bits"], 16)
118    block.nNonce = 0
119    block.vtx = [cbtx] + [tx_from_hex(t["data"]) for t in tmpl["transactions"]]
120
121    witnonce = 0
122    witroot = block.calc_witness_merkle_root()
123    cbwit = CTxInWitness()
124    cbwit.scriptWitness.stack = [ser_uint256(witnonce)]
125    block.vtx[0].wit.vtxinwit = [cbwit]
126    block.vtx[0].vout.append(CTxOut(0, bytes(get_witness_script(witroot, witnonce))))
127
128    block.hashMerkleRoot = block.calc_merkle_root()
129
130    return block
131
132def generate_psbt(block, signet_spk):
133    signet_spk_bin = bytes.fromhex(signet_spk)
134    signme, spendme = signet_txs(block, signet_spk_bin)
135    psbt = PSBT()
136    psbt.g = PSBTMap( {PSBT_GLOBAL_UNSIGNED_TX: signme.serialize(),
137                       PSBT_SIGNET_BLOCK: block.serialize()
138                     } )
139    psbt.i = [ PSBTMap( {PSBT_IN_NON_WITNESS_UTXO: spendme.serialize(),
140                         PSBT_IN_SIGHASH_TYPE: bytes([1,0,0,0])})
141             ]
142    psbt.o = [ PSBTMap() ]
143    return psbt.to_base64()
144
145def get_poolid(args):
146    if args.poolid is not None:
147        return args.poolid.encode('utf8')
148    elif args.poolnum is not None:
149        return b"/signet:%d/" % (args.poolnum)
150    else:
151        return None
152
153def get_reward_addr_spk(args, height):
154    assert args.address is not None or args.descriptor is not None
155
156    if hasattr(args, "reward_spk"):
157        return args.address, args.reward_spk
158
159    if args.address is not None:
160        reward_addr = args.address
161    elif '*' not in args.descriptor:
162        reward_addr = args.address = json.loads(args.bcli("deriveaddresses", args.descriptor))[0]
163    else:
164        remove = [k for k in args.derived_addresses.keys() if k+20 <= height]
165        for k in remove:
166            del args.derived_addresses[k]
167        if height not in args.derived_addresses:
168            addrs = json.loads(args.bcli("deriveaddresses", args.descriptor, "[%d,%d]" % (height, height+20)))
169            for k, a in enumerate(addrs):
170                args.derived_addresses[height+k] = a
171        reward_addr = args.derived_addresses[height]
172
173    reward_spk = bytes.fromhex(json.loads(args.bcli("getaddressinfo", reward_addr))["scriptPubKey"])
174    if args.address is not None:
175        # will always be the same, so cache
176        args.reward_spk = reward_spk
177
178    return reward_addr, reward_spk
179
180def do_genpsbt(args):
181    poolid = get_poolid(args)
182    tmpl = json.load(sys.stdin)
183    signet_spk = tmpl["signet_challenge"]
184    _, reward_spk = get_reward_addr_spk(args, tmpl["height"])
185    block = new_block(tmpl, reward_spk, poolid=poolid)
186    psbt = generate_psbt(block, signet_spk)
187    print(psbt)
188
189def do_solvepsbt(args):
190    psbt = decode_challenge_psbt(sys.stdin.read())
191    block = get_block_from_psbt(psbt)
192    signet_solution = get_solution_from_psbt(psbt, emptyok=True)
193    block = finish_block(block, signet_solution, args.grind_cmd)
194    print(block.serialize().hex())
195
196def nbits_to_target(nbits):
197    shift = (nbits >> 24) & 0xff
198    return (nbits & 0x00ffffff) * 2**(8*(shift - 3))
199
200def target_to_nbits(target):
201    tstr = "{0:x}".format(target)
202    if len(tstr) < 6:
203        tstr = ("000000"+tstr)[-6:]
204    if len(tstr) % 2 != 0:
205        tstr = "0" + tstr
206    if int(tstr[0],16) >= 0x8:
207        # avoid "negative"
208        tstr = "00" + tstr
209    fix = int(tstr[:6], 16)
210    sz = len(tstr)//2
211    if tstr[6:] != "0"*(sz*2-6):
212        fix += 1
213
214    return int("%02x%06x" % (sz,fix), 16)
215
216def seconds_to_hms(s):
217    if s == 0:
218        return "0s"
219    neg = (s < 0)
220    if neg:
221        s = -s
222    out = ""
223    if s % 60 > 0:
224        out = "%ds" % (s % 60)
225    s //= 60
226    if s % 60 > 0:
227        out = "%dm%s" % (s % 60, out)
228    s //= 60
229    if s > 0:
230        out = "%dh%s" % (s, out)
231    if neg:
232        out = "-" + out
233    return out
234
235def trivial_challenge(spkhex):
236    """
237    BIP325 allows omitting the signet commitment when scriptSig and
238    scriptWitness are both empty. This is the case for trivial
239    challenges such as OP_TRUE or a single data push.
240    """
241    spk = bytes.fromhex(spkhex)
242    if len(spk) == 1 and 0x51 <= spk[0] <= 0x60:
243        # OP_TRUE/OP_1...OP_16
244        return True
245    elif 2 <= len(spk) <= 76 and spk[0] + 1 == len(spk):
246        # Single fixed push of 1-75 bytes
247        return True
248    return False
249
250class Generate: