git / moonbite-network / tools/vault/sweep_to_vault.py · raw
#!/usr/bin/env python3"""Sweep founder-mined coins into the ADR-014 time-locked vault (ADR-015).Dry run by default: it prints what it would send and sends nothing. python3 tools/vault/sweep_to_vault.py --cli "~/moonbite/moonbite-cli -datadir=$HOME/.moonbite -conf=$HOME/.moonbite/moonbite.conf" python3 tools/vault/sweep_to_vault.py --cli "..." --sendWhat it does, and the guard rails:- Spends ONLY mature coins (100+ confirmations) sitting at the founder addresses listed in ADR-015, never anything else in the wallet.- Rebuilds the vault address from ADR-014's witness script with the node's own `decodescript` and refuses to run if it doesn't match.- Sends in batches (one transaction can't spend thousands of block rewards), each paying everything, minus a small fee, to the vault. Every batch is checked with `testmempoolaccept` before it is broadcast.- Prints every txid and appends it to sweep-log.txt for the ADR-015 Log.Run it again after 100 more blocks to sweep coinbases that were stillimmature the first time."""from __future__ import annotationsimport argparseimport jsonimport mathimport shleximport subprocessimport sysimport timefrom decimal import ROUND_DOWN, DecimalVAULT = "moon1qw436kfcvdg70qkkl4lfnt7m9qd79mkzl0js5lnjnqyxfkckw4d5sj26lxq"WITNESS_SCRIPT = "04d0ec7770b17576a9143c195a5773ff6f186cd9ff76290c36528433d77288ac"FOUNDER_ADDRESSES = [ "moon1qa29zr498w5kzhjnc63c7zmmt536uylvn9tkxsk", "moon1qk2xagl4j3cg5rz6ayvzvtd4r7yypjpdaeg7rua",]MATURE = 100COIN = Decimal("0.00000001")# Virtual size of a P2WPKH-in, one-P2WSH-out transaction: ~11 vB overhead,# ~68 vB per input, 43 vB for the vault output.VB_BASE, VB_IN, VB_OUT = 11, 68, 43class Cli: def __init__(self, cmd: str): # Windows paths: forward slashes work for the node and survive shlex. self.argv = shlex.split(cmd.replace(chr(92), "/")) def __call__(self, method: str, *params): args = [p if isinstance(p, str) else json.dumps(p) for p in params] out = subprocess.run(self.argv + [method, *args], capture_output=True, text=True) if out.returncode != 0: raise SystemExit(f"{method} failed: {out.stderr.strip() or out.stdout.strip()}") text = out.stdout.strip() try: return json.loads(text, parse_float=Decimal) except json.JSONDecodeError: return textdef vbytes(n_inputs: int) -> int: return VB_BASE + VB_IN * n_inputs + VB_OUTdef plan(utxos: list[dict], batch: int, feerate: int) -> list[dict]: """Split coins into batches; each pays its total minus fee to the vault.""" utxos = sorted(utxos, key=lambda u: (u["txid"], u["vout"])) out = [] for i in range(0, len(utxos), batch): part = utxos[i:i + batch] total = sum(Decimal(str(u["amount"])) for u in part) fee = Decimal(math.ceil(vbytes(len(part)) * feerate)) * COIN out.append({"inputs": part, "total": total, "fee": fee, "send": (total - fee).quantize(COIN, rounding=ROUND_DOWN)}) return outdef check_vault(cli) -> None: decoded = cli("decodescript", WITNESS_SCRIPT) seg = decoded.get("segwit", {}) if isinstance(decoded, dict) else {} found = seg.get("address") or (seg.get("addresses") or [None])[0] if found != VAULT: raise SystemExit(f"Vault check FAILED: the witness script gives {found!r}, expected {VAULT}. Nothing sent.") asm = decoded.get("asm", "") if "1886907600 OP_CHECKLOCKTIMEVERIFY" not in asm: raise SystemExit(f"Vault check FAILED: unexpected script {asm!r}. Nothing sent.")def main(argv=None) -> int: ap = argparse.ArgumentParser(description="Sweep founder coins into the ADR-014 vault (ADR-015).") ap.add_argument("--cli", required=True, help='moonbite-cli command, e.g. "moonbite-cli -datadir=..."') ap.add_argument("--send", action="store_true", help="actually broadcast (default: dry run)") ap.add_argument("--batch", type=int, default=400, help="inputs per transaction (default 400)") ap.add_argument("--feerate", type=int, default=2, help="fee in units per vbyte (default 2)") ap.add_argument("--log", default="sweep-log.txt", help="where txids are appended") args = ap.parse_args(argv) if not 1 <= args.batch <= 1000: raise SystemExit("--batch must be between 1 and 1000") cli = Cli(args.cli) check_vault(cli) print(f"Vault address verified from the ADR-014 witness script: {VAULT}") utxos = cli("listunspent", MATURE, 9999999, FOUNDER_ADDRESSES) if not utxos: print("No mature founder coins in this wallet. (Is this the wallet that mined them?)") return 1 by_addr: dict[str, list] = {} for u in utxos: by_addr.setdefault(u["address"], []).append(u) for a, us in by_addr.items(): print(f" {a}: {len(us)} coins, {sum(Decimal(str(u['amount'])) for u in us)} MBITE") batches = plan(utxos, args.batch, args.feerate) print(f"Plan: {len(batches)} transaction(s), " f"{sum(b['send'] for b in batches)} MBITE to the vault, {sum(b['fee'] for b in batches)} in fees.") for n, b in enumerate(batches, 1): print(f" #{n}: {len(b['inputs'])} inputs, {b['total']} in, {b['send']} to vault, fee {b['fee']}") if not args.send: print("Dry run: nothing sent. Add --send to broadcast.") return 0 for n, b in enumerate(batches, 1): inputs = [{"txid": u["txid"], "vout": u["vout"]} for u in b["inputs"]] raw = cli("createrawtransaction", inputs, {VAULT: str(b["send"])}) signed = cli("signrawtransactionwithwallet", raw) if not signed.get("complete"): raise SystemExit(f"#{n}: wallet could not sign every input. Nothing more sent.") test = cli("testmempoolaccept", [signed["hex"]])[0] if not test.get("allowed"): raise SystemExit(f"#{n}: node refused it ({test.get('reject-reason')}). Nothing more sent.") txid = cli("sendrawtransaction", signed["hex"]) line = f"{time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime())} UTC {txid} {b['send']} MBITE {len(b['inputs'])} inputs" print(f" sent #{n}: {txid}") with open(args.log, "a", encoding="utf-8") as fh: fh.write(line + "\n") print(f"Done. Txids are in {args.log}; copy them into ADR-015's Log.") return 0if __name__ == "__main__": sys.exit(main())