#!/usr/bin/env python3 # Copyright (c) 2020 The Bitcoin Core developers # Distributed under the MIT software license, see the accompanying # file COPYING or http://www.opensource.org/licenses/mit-license.php. import argparse import base64 import json import logging import math import os import re import struct import sys import time import subprocess import asyncio, websockets from hashlib import sha256 from io import BytesIO from random import uniform from decimal import Decimal PATH_BASE_CONTRIB_SIGNET = os.path.abspath(os.path.dirname(os.path.realpath(__file__))) PATH_BASE_TEST_FUNCTIONAL = os.path.abspath(os.path.join(PATH_BASE_CONTRIB_SIGNET, "..", "..", "test", "functional")) sys.path.insert(0, PATH_BASE_TEST_FUNCTIONAL) from test_framework.blocktools import WITNESS_COMMITMENT_HEADER, script_BIP34_coinbase_height # noqa: E402 from test_framework.messages import CBlock, CBlockHeader, COutPoint, CTransaction, CTxIn, CTxInWitness, CTxOut, from_hex, deser_string, hash256, ser_compact_size, ser_string, ser_uint256, tx_from_hex, uint256_from_str # noqa: E402 from test_framework.script import CScript, CScriptOp # noqa: E402 logging.basicConfig( format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO, datefmt='%Y-%m-%d %H:%M:%S') SIGNET_HEADER = b"\xec\xc7\xda\xa2" PSBT_SIGNET_BLOCK = b"\xfc\x06signetb" # proprietary PSBT global field holding the block being signed RE_MULTIMINER = re.compile("^(\d+)(-(\d+))?/(\d+)$") CC_URL = 'ws://localhost:9823/websocket' def json_dumps(obj, **k): def hook(dd): if isinstance(dd, Decimal): return float(dd) if hasattr(dd, 'strftime'): return str(dd) # isoformat logging.error("Unhandled JSON type: %r" % dd) raise TypeError k['default'] = hook return json.dumps(obj, **k) def message(action, *args): return json_dumps(dict(action=action, args=args)) POLICY = { 'never_log': False, 'must_log': False, 'priv_over_ux': True, 'boot_to_hsm': "123456", 'period': None, 'set_sl': None, 'allow_sl': None, 'rules': [ { "whitelist": "", "per_period": None, "max_amount": None, "users": "", "min_users": "all", "local_conf": False, "wallet": None } ], } # #### some helpers that could go into test_framework # like from_hex, but without the hex part def FromBinary(cls, stream): """deserialize a binary stream (or bytes object) into an object""" # handle bytes object by turning it into a stream was_bytes = isinstance(stream, bytes) if was_bytes: stream = BytesIO(stream) obj = cls() obj.deserialize(stream) if was_bytes: assert len(stream.read()) == 0 return obj class PSBTMap: """Class for serializing and deserializing PSBT maps""" def __init__(self, map=None): self.map = map if map is not None else {} def deserialize(self, f): m = {} while True: k = deser_string(f) if len(k) == 0: break v = deser_string(f) if len(k) == 1: k = k[0] assert k not in m m[k] = v self.map = m def serialize(self): m = b"" for k,v in self.map.items(): if isinstance(k, int) and 0 <= k and k <= 255: k = bytes([k]) m += ser_compact_size(len(k)) + k m += ser_compact_size(len(v)) + v m += b"\x00" return m class PSBT: """Class for serializing and deserializing PSBTs""" def __init__(self): self.g = PSBTMap() self.i = [] self.o = [] self.tx = None def deserialize(self, f): assert f.read(5) == b"psbt\xff" self.g = FromBinary(PSBTMap, f) assert 0 in self.g.map self.tx = FromBinary(CTransaction, self.g.map[0]) self.i = [FromBinary(PSBTMap, f) for _ in self.tx.vin] self.o = [FromBinary(PSBTMap, f) for _ in self.tx.vout] return self def serialize(self): assert isinstance(self.g, PSBTMap) assert isinstance(self.i, list) and all(isinstance(x, PSBTMap) for x in self.i) assert isinstance(self.o, list) and all(isinstance(x, PSBTMap) for x in self.o) assert 0 in self.g.map tx = FromBinary(CTransaction, self.g.map[0]) assert len(tx.vin) == len(self.i) assert len(tx.vout) == len(self.o) psbt = [x.serialize() for x in [self.g] + self.i + self.o] return b"psbt\xff" + b"".join(psbt) def to_base64(self): return base64.b64encode(self.serialize()).decode("utf8") @classmethod def from_base64(cls, b64psbt): return FromBinary(cls, base64.b64decode(b64psbt)) # ##### def create_coinbase(height, value, spk, miner_tag=''): cb = CTransaction() scriptsig = bytes(script_BIP34_coinbase_height(height)) if miner_tag is not None: scriptsig = CScript(scriptsig + CScriptOp.encode_op_pushdata(miner_tag.encode())) else: scriptsig = CScript(scriptsig) cb.vin = [CTxIn(COutPoint(0, 0xffffffff), scriptsig, 0xffffffff)] # Utiliser une seule sortie pour le miner (le relay recevra des fonds via une transaction normale) cb.vout = [CTxOut(value, spk)] logging.info(f"Coinbase reward: {value} sat to miner") return cb def get_witness_script(witness_root, witness_nonce): commitment = uint256_from_str(hash256(ser_uint256(witness_root) + ser_uint256(witness_nonce))) return b"\x6a" + CScriptOp.encode_op_pushdata(WITNESS_COMMITMENT_HEADER + ser_uint256(commitment)) def signet_txs(block, challenge): # assumes signet solution has not been added yet so does not need # to be removed txs = block.vtx[:] txs[0] = CTransaction(txs[0]) txs[0].vout[-1].scriptPubKey += CScriptOp.encode_op_pushdata(SIGNET_HEADER) hashes = [] for tx in txs: tx.rehash() hashes.append(ser_uint256(tx.sha256)) mroot = block.get_merkle_root(hashes) sd = b"" sd += struct.pack("> 24) & 0xff return (nbits & 0x00ffffff) * 2**(8*(shift - 3)) def target_to_nbits(target): tstr = "{0:x}".format(target) if len(tstr) < 6: tstr = ("000000"+tstr)[-6:] if len(tstr) % 2 != 0: tstr = "0" + tstr if int(tstr[0],16) >= 0x8: # avoid "negative" tstr = "00" + tstr fix = int(tstr[:6], 16) sz = len(tstr)//2 if tstr[6:] != "0"*(sz*2-6): fix += 1 return int("%02x%06x" % (sz,fix), 16) def seconds_to_hms(s): if s == 0: return "0s" neg = (s < 0) if neg: s = -s out = "" if s % 60 > 0: out = "%ds" % (s % 60) s //= 60 if s % 60 > 0: out = "%dm%s" % (s % 60, out) s //= 60 if s > 0: out = "%dh%s" % (s, out) if neg: out = "-" + out return out def send_to_relay(args, miner_addr, height): """Envoie des fonds au relay après avoir miné un bloc""" if not hasattr(args, 'relay_address') or not args.relay_address: return relay_addr = args.relay_address split_ratio = getattr(args, 'reward_split_ratio', 0.5) # Attendre que le bloc soit confirmé time.sleep(5) try: # Vérifier le solde du wallet balance = json.loads(args.bcli(f"-rpcwallet={args.MINING_WALLET}", "getbalance")) if balance < 0.001: # Minimum 0.001 BTC logging.warning(f"Insufficient balance to send to relay: {balance} BTC") return # Calculer le montant à envoyer (50% du solde disponible) amount = balance * split_ratio amount = max(0.001, min(amount, 0.1)) # Entre 0.001 et 0.1 BTC # Envoyer les fonds au relay txid = json.loads(args.bcli(f"-rpcwallet={args.MINING_WALLET}", "sendtoaddress", relay_addr, str(amount))) logging.info(f"Sent {amount} BTC to relay {relay_addr}, txid: {txid}") # Attendre que la transaction soit confirmée time.sleep(2) except Exception as e: logging.error(f"Failed to send funds to relay: {e}") def next_block_delta(last_nbits, last_hash, ultimate_target, do_poisson): # strategy: # 1) work out how far off our desired target we are # 2) cap it to a factor of 4 since that's the best we can do in a single retarget period # 3) use that to work out the desired average interval in this retarget period # 4) if doing poisson, use the last hash to pick a uniformly random number in [0,1), and work out a random multiplier to vary the average by # 5) cap the resulting interval between 1 second and 1 hour to avoid extremes INTERVAL = 600.0*2016/2015 # 10 minutes, adjusted for the off-by-one bug current_target = nbits_to_target(last_nbits) retarget_factor = ultimate_target / current_target retarget_factor = max(0.25, min(retarget_factor, 4.0)) avg_interval = INTERVAL * retarget_factor if do_poisson: det_rand = int(last_hash[-8:], 16) * 2**-32 this_interval_variance = -math.log1p(-det_rand) else: this_interval_variance = uniform(0.95,1.05) this_interval = avg_interval * this_interval_variance this_interval = max(1, min(this_interval, 3600)) return this_interval def next_block_is_mine(last_hash, my_blocks): det_rand = int(last_hash[-16:-8], 16) return my_blocks[0] <= (det_rand % my_blocks[2]) < my_blocks[1] def do_generate(args): if args.max_blocks is not None: if args.ongoing: logging.error("Cannot specify both --ongoing and --max-blocks") return 1 if args.max_blocks < 1: logging.error("N must be a positive integer") return 1 max_blocks = args.max_blocks elif args.ongoing: max_blocks = None else: max_blocks = 1 if args.set_block_time is not None and max_blocks != 1: logging.error("Cannot specify --ongoing or --max-blocks > 1 when using --set-block-time") return 1 if args.set_block_time is not None and args.set_block_time < 0: args.set_block_time = time.time() logging.info("Treating negative block time as current time (%d)" % (args.set_block_time)) if args.min_nbits: if args.nbits is not None: logging.error("Cannot specify --nbits and --min-nbits") return 1 args.nbits = "1e0377ae" logging.info("Using nbits=%s" % (args.nbits)) if args.set_block_time is None: if args.nbits is None or len(args.nbits) != 8: logging.error("Must specify --nbits (use calibrate command to determine value)") return 1 if args.multiminer is None: my_blocks = (0,1,1) else: if not args.ongoing: logging.error("Cannot specify --multiminer without --ongoing") return 1 m = RE_MULTIMINER.match(args.multiminer) if m is None: logging.error("--multiminer argument must be k/m or j-k/m") return 1 start,_,stop,total = m.groups() if stop is None: stop = start start, stop, total = map(int, (start, stop, total)) if stop < start or start <= 0 or total < stop or total == 0: logging.error("Inconsistent values for --multiminer") return 1 my_blocks = (start-1, stop, total) if args.MINER_TAG is not None: args.MINER_TAG = args.MINER_TAG.encode('utf-8') else: args.MINER_TAG = "default".encode('utf-8') ultimate_target = nbits_to_target(int(args.nbits,16)) mined_blocks = 0 bestheader = {"hash": None} lastheader = None while max_blocks is None or mined_blocks < max_blocks: # current status? bci = json.loads(args.bcli("getblockchaininfo")) if bestheader["hash"] != bci["bestblockhash"]: bestheader = json.loads(args.bcli("getblockheader", bci["bestblockhash"])) if lastheader is None: lastheader = bestheader["hash"] elif bestheader["hash"] != lastheader: next_delta = next_block_delta(int(bestheader["bits"], 16), bestheader["hash"], ultimate_target, args.poisson) next_delta += bestheader["time"] - time.time() next_is_mine = next_block_is_mine(bestheader["hash"], my_blocks) logging.info("Received new block at height %d; next in %s (%s)", bestheader["height"], seconds_to_hms(next_delta), ("mine" if next_is_mine else "backup")) lastheader = bestheader["hash"] # when is the next block due to be mined? now = time.time() if args.set_block_time is not None: logging.debug("Setting start time to %d", args.set_block_time) mine_time = args.set_block_time action_time = now is_mine = True elif bestheader["height"] == 0: time_delta = next_block_delta(int(bestheader["bits"], 16), bci["bestblockhash"], ultimate_target, args.poisson) time_delta *= 100 # 100 blocks logging.info("Backdating time for first block to %d minutes ago" % (time_delta/60)) mine_time = now - time_delta action_time = now is_mine = True else: time_delta = next_block_delta(int(bestheader["bits"], 16), bci["bestblockhash"], ultimate_target, args.poisson) mine_time = bestheader["time"] + time_delta is_mine = next_block_is_mine(bci["bestblockhash"], my_blocks) action_time = mine_time if not is_mine: action_time += args.backup_delay if args.standby_delay > 0: action_time += args.standby_delay elif mined_blocks == 0: # for non-standby, always mine immediately on startup, # even if the next block shouldn't be ours action_time = now # don't want fractional times so round down mine_time = int(mine_time) action_time = int(action_time) # can't mine a block 2h in the future; 1h55m for some safety action_time = max(action_time, mine_time - 6900) # ready to go? otherwise sleep and check for new block if now < action_time: sleep_for = min(action_time - now, 60) if mine_time < now: # someone else might have mined the block, # so check frequently, so we don't end up late # mining the next block if it's ours sleep_for = min(20, sleep_for) minestr = "mine" if is_mine else "backup" logging.debug("Sleeping for %s, next block due in %s (%s)" % (seconds_to_hms(sleep_for), seconds_to_hms(mine_time - now), minestr)) time.sleep(sleep_for) continue # gbt tmpl = json.loads(args.bcli("getblocktemplate", '{"rules":["signet","segwit"]}')) if tmpl["previousblockhash"] != bci["bestblockhash"]: logging.warning("GBT based off unexpected block (%s not %s), retrying", tmpl["previousblockhash"], bci["bestblockhash"]) time.sleep(1) continue logging.debug("GBT template: %s", tmpl) if tmpl["mintime"] > mine_time: logging.info("Updating block time from %d to %d", mine_time, tmpl["mintime"]) mine_time = tmpl["mintime"] if mine_time > now: logging.error("GBT mintime is in the future: %d is %d seconds later than %d", mine_time, (mine_time-now), now) return 1 # address for reward reward_addr, reward_spk = get_reward_addr_spk(args, tmpl["height"]) # Obtenir l'adresse et le scriptPubKey du relay relay_spk = None if hasattr(args, 'relay_address') and args.relay_address: try: relay_spk = bytes.fromhex(json.loads(args.bcli(f"-rpcwallet={args.WATCHONLY_WALLET}", "getaddressinfo", args.relay_address))["scriptPubKey"]) except: # Si l'adresse n'est pas dans le wallet, créer une adresse simple logging.warning(f"Relay address {args.relay_address} not in wallet, using simple address") # Pour l'instant, on utilise la même adresse que le miner relay_spk = reward_spk # mine block logging.debug("Mining block delta=%s start=%s mine=%s", seconds_to_hms(mine_time-bestheader["time"]), mine_time, is_mine) mined_blocks += 1 psbt = generate_psbt(tmpl, reward_spk, blocktime=mine_time, relay_spk=relay_spk, reward_split_ratio=getattr(args, 'reward_split_ratio', 0.5)) logging.info(f"psbt pre-processing: {psbt}") input_stream = os.linesep.join([psbt, "true", "ALL"]).encode('utf8') if args.signer == "coldcard": psbt = json.loads(args.bcli("-stdin", f"-rpcwallet={args.WATCHONLY_WALLET}", "walletprocesspsbt", input=input_stream))['psbt'] try: psbt_hash = asyncio.get_event_loop().run_until_complete(coldcard_upload_psbt(psbt)) except Exception as e: logging.error("Waiting 5s before trying again.") time.sleep(5) continue try: psbt_signed = asyncio.get_event_loop().run_until_complete(coldcard_sign_psbt(psbt_hash)) except Exception as e: logging.error("Please check your keys.") return 1 try: assert('local_download' in psbt_signed) except AssertionError as e: logging.error("Didn't receive signed psbt") logging.error(f"Received: {psbt_signed}") logging.error("Waiting 5s before trying again.") time.sleep(5) continue input_stream = os.linesep.join([psbt_signed['local_download']['data'], "false"]).encode('utf8') try: final_psbt = json.loads(args.bcli("-stdin", "finalizepsbt", input=input_stream)) except Exception as e: logging.error(f"Core can't finalize psbt: {e}") return 1 if not final_psbt.get("complete",False): logging.error("finalizepsbt return: %s" % (final_psbt,)) sys.stderr.write("PSBT finalization failed\n") return 1 final_psbt = final_psbt["psbt"] else: psbt_signed = json.loads(args.bcli("-stdin", f"-rpcwallet={args.MINING_WALLET}", "walletprocesspsbt", input=input_stream)) if not psbt_signed.get("complete",False): logging.debug("Generated PSBT: %s" % (psbt,)) sys.stderr.write("PSBT signing failed\n") return 1 final_psbt = psbt_signed["psbt"] block, signet_solution = do_decode_psbt(final_psbt) block = finish_block(block, signet_solution, args.grind_cmd) # submit block r = args.bcli("-stdin", "submitblock", input=block.serialize().hex().encode('utf8')) # report bstr = "block" if is_mine else "backup block" next_delta = next_block_delta(block.nBits, block.hash, ultimate_target, args.poisson) next_delta += block.nTime - time.time() next_is_mine = next_block_is_mine(block.hash, my_blocks) logging.debug("Block hash %s payout to %s", block.hash, reward_addr) logging.info("Mined %s at height %d; next in %s (%s)", bstr, tmpl["height"], seconds_to_hms(next_delta), ("mine" if next_is_mine else "backup")) if r != "": logging.warning("submitblock returned %s for height %d hash %s", r, tmpl["height"], block.hash) # Envoyer des fonds au relay si configuré if hasattr(args, 'relay_address') and args.relay_address and getattr(args, 'reward_split_ratio', 0) > 0: try: send_to_relay(args, reward_addr, tmpl["height"]) except Exception as e: logging.error(f"Failed to send funds to relay: {e}") lastheader = block.hash async def coldcard_upload_psbt(psbt): async with websockets.connect(CC_URL) as ws: psbt_bin = base64.b64decode(psbt) psbt_hash = sha256(psbt_bin).digest() payload = message("upload_psbt", len(psbt_bin), psbt_hash.hex(), psbt) await ws.send(payload) async for msg in ws: try: r = json.loads(msg) except json.JSONDecodeError as e: logging.error(f"{e}: {msg} is not valid json") continue if 'update_status' in r: logging.info(f"Received update: {r['update_status']}") elif 'success' in r: break return psbt_hash.hex() async def coldcard_sign_psbt(psbt_hash): async with websockets.connect(CC_URL) as ws: payload = message("submit_psbt", psbt_hash, False, False, True) await ws.send(payload) async for msg in ws: try: r = json.loads(msg) except json.JSONDecodeError as e: logging.error(f"{e}: {msg} is not valid json") continue if "update_status" in r: continue elif 'local_download' in r: break return r def do_calibrate(args): if args.nbits is not None and args.seconds is not None: sys.stderr.write("Can only specify one of --nbits or --seconds\n") return 1 if args.nbits is not None and len(args.nbits) != 8: sys.stderr.write("Must specify 8 hex digits for --nbits\n") return 1 TRIALS = 600 # gets variance down pretty low TRIAL_BITS = 0x1e3ea75f # takes about 5m to do 600 trials header = CBlockHeader() header.nBits = TRIAL_BITS targ = nbits_to_target(header.nBits) start = time.time() count = 0 for i in range(TRIALS): header.nTime = i header.nNonce = 0 headhex = header.serialize().hex() cmd = args.grind_cmd.split(" ") + [headhex] newheadhex = subprocess.run(cmd, stdout=subprocess.PIPE, input=b"", check=True).stdout.strip() avg = (time.time() - start) * 1.0 / TRIALS if args.nbits is not None: want_targ = nbits_to_target(int(args.nbits,16)) want_time = avg*targ/want_targ else: want_time = args.seconds if args.seconds is not None else 25 want_targ = int(targ*(avg/want_time)) print("nbits=%08x for %ds average mining time" % (target_to_nbits(want_targ), want_time)) return 0 def do_setup(args): if args.signer == "coldcard": sys.exit(asyncio.get_event_loop().run_until_complete(coldcard_setup())) else: logging.error("Unknown option") sys.exit(1) async def coldcard_status() -> dict: logging.info("Checking that server is up and connected to Coldcard") status = {} while 1: try: async with websockets.connect(CC_URL) as ws: payload = message("get_info") await ws.send(payload) async for msg in ws: try: r = json.loads(msg) except json.JSONDecodeError: logging.info(f"{msg} is not valid json") return 1 try: if 'status' in r: logging.info("Got status") logging.info(f"{r}") status = r['status'] break elif 'update_status' in r: # we ignore this as there's not all the details there logging.info("Received update_status, ignoring") continue elif 'error' in r: logging.error(f"{r['error']}") return 1 else: logging.info(f"{r}") logging.info("Still waiting for Coldcard connection") continue except Exception as e: logging.error(f"{e}") return 1 except ConnectionError as e: logging.error("Server seems to be offline. Trying again.") await asyncio.sleep(10) continue logging.info(f"{status}") return status async def coldcard_setup(): status = await coldcard_status() logging.info(f"{status}") if not status['hsm']['active']: logging.info("Entering Coldcard HSM setup") logging.info("Please follow the instructions on the coldcard") async with websockets.connect(CC_URL) as ws: payload = message("submit_policy", json_dumps(POLICY), False) await ws.send(payload) async for msg in ws: try: r = json.loads(msg) except json.JSONDecodeError: logging.info(f"{msg} is not valid json") return 1 try: if 'update_status' in r: if r['update_status']['hsm']['active']: logging.info("HSM is now activated. Proceeding.") break else: logging.info(f"Received update: {r['update_status']['hsm']}") except KeyError: logging.error(f"Received unexpected message: {r}") continue except Exception as e: logging.error(f"{e}") return 1 elif not status['sl_loaded']: while 1: status = await coldcard_status() if status['sl_loaded']: logging.info("Coldcard in HSM mode. Proceeding...") break await asyncio.sleep(10) else: logging.info("Coldcard already in HSM mode. Proceeding...") return 0 def bitcoin_cli(basecmd, args, **kwargs): cmd = basecmd + ["-signet"] + args logging.debug("Calling bitcoin-cli: %r", cmd) out = subprocess.run(cmd, stdout=subprocess.PIPE, **kwargs, check=True).stdout if isinstance(out, bytes): out = out.decode('utf8') return out.strip() def main(): parser = argparse.ArgumentParser() parser.add_argument("--cli", default="bitcoin-cli", type=str, help="bitcoin-cli command") parser.add_argument("--debug", action="store_true", help="Print debugging info") parser.add_argument("--quiet", action="store_true", help="Only print warnings/errors") cmds = parser.add_subparsers(help="sub-commands") genpsbt = cmds.add_parser("genpsbt", help="Generate a block PSBT for signing") genpsbt.set_defaults(fn=do_genpsbt) genpsbt.add_argument('--MINER_TAG', required=True, help='Miner tag') solvepsbt = cmds.add_parser("solvepsbt", help="Solve a signed block PSBT") solvepsbt.set_defaults(fn=do_solvepsbt) setup = cmds.add_parser("setup", help="Setup HSM for coldcard") setup.set_defaults(fn=do_setup) setup.add_argument("--signer", default=None, type=str, help="Who's signing blocks (default: Bitcoin Core)") generate = cmds.add_parser("generate", help="Mine blocks") generate.set_defaults(fn=do_generate) generate.add_argument("--ongoing", action="store_true", help="Keep mining blocks") generate.add_argument("--max-blocks", default=None, type=int, help="Max blocks to mine (default=1)") generate.add_argument("--set-block-time", default=None, type=int, help="Set block time (unix timestamp)") generate.add_argument("--nbits", default=None, type=str, help="Target nBits (specify difficulty)") generate.add_argument("--min-nbits", action="store_true", help="Target minimum nBits (use min difficulty)") generate.add_argument("--poisson", action="store_true", help="Simulate randomised block times") generate.add_argument("--multiminer", default=None, type=str, help="Specify which set of blocks to mine (eg: 1-40/100 for the first 40%%, 2/3 for the second 3rd)") generate.add_argument("--backup-delay", default=300, type=int, help="Seconds to delay before mining blocks reserved for other miners (default=300)") generate.add_argument("--standby-delay", default=0, type=int, help="Seconds to delay before mining blocks (default=0)") generate.add_argument("--signer", default=None, type=str, help="Who's signing blocks (default: Bitcoin Core)") generate.add_argument('--WATCHONLY_WALLET', required=True, help='Watch-only wallet') generate.add_argument('--MINING_WALLET', required=True, help='Mining wallet') generate.add_argument('--MINER_TAG', required=True, help='Miner tag') calibrate = cmds.add_parser("calibrate", help="Calibrate difficulty") calibrate.set_defaults(fn=do_calibrate) calibrate.add_argument("--nbits", type=str, default=None) calibrate.add_argument("--seconds", type=int, default=None) for sp in [genpsbt, generate]: sp.add_argument("--address", default=None, type=str, help="Address for block reward payment") sp.add_argument("--descriptor", default=None, type=str, help="Descriptor for block reward payment") sp.add_argument("--relay-address", default=None, type=str, help="Address for relay reward payment (50% of block reward)") sp.add_argument("--reward-split-ratio", default=0.5, type=float, help="Ratio of block reward to send to relay (default: 0.5)") for sp in [solvepsbt, generate, calibrate]: sp.add_argument("--grind-cmd", default=None, type=str, required=(sp==calibrate), help="Command to grind a block header for proof-of-work") args = parser.parse_args(sys.argv[1:]) args.bcli = lambda *a, input=b"", **kwargs: bitcoin_cli(args.cli.split(" "), list(a), input=input, **kwargs) if hasattr(args, "address") and hasattr(args, "descriptor"): if args.address is None and args.descriptor is None: sys.stderr.write("Must specify --address or --descriptor\n") return 1 elif args.address is not None and args.descriptor is not None: sys.stderr.write("Only specify one of --address or --descriptor\n") return 1 args.derived_addresses = {} if args.debug: logging.getLogger().setLevel(logging.DEBUG) elif args.quiet: logging.getLogger().setLevel(logging.WARNING) else: logging.getLogger().setLevel(logging.INFO) if hasattr(args, "fn"): return args.fn(args) else: logging.error("Must specify command") return 1 if __name__ == "__main__": main()