#!/usr/bin/env python3 """Automatic HASHBATTLE miner: fetch, solve, verify, locally sign, confirm, repeat. Requires eth-account plus the bundled hb_work.py/hb_engines.py and C kernel. No RPC, contract or chain defaults ship here. Configure chain.json or supply --rpc --contract --chain. PRIVATE KEY is prompted invisibly once; alternatively use HB_PRIVATE_KEY or a locally owned MODE600 --key-file. Never pass a raw key on the command line. State contains signed transactions, NEVER private keys. """ from __future__ import annotations import argparse import getpass import fcntl import json import secrets import sys import time from contextlib import contextmanager, redirect_stdout import os from pathlib import Path import stat import warnings from decimal import Decimal, InvalidOperation, localcontext from urllib.parse import urlparse import hb_work KIT_DIR = Path(__file__).resolve().parent class MiningError(RuntimeError): """Fail-closed error, safe to show without secret material.""" def parse_args(argv=None): p = argparse.ArgumentParser(description=__doc__, allow_abbrev=False) p.add_argument("--rpc") p.add_argument("--contract") p.add_argument("--chain") p.add_argument("--config", type=Path, default=KIT_DIR / "chain.json") p.add_argument("--engine", choices=("cpu", "gpu"), default="cpu") p.add_argument("--max-mints", type=int, default=0, help="0: repeat until Ctrl-C") p.add_argument("--key-file", type=Path) p.add_argument("--state-dir", type=Path, default=KIT_DIR / ".miner-state") p.add_argument("--max-price-eth") p.add_argument("--max-gas-price-gwei") p.add_argument("--max-total-cost-eth", help="per-transaction value + full gas limit cost cap") a, unknown = p.parse_known_args(argv) if unknown: p.error("unrecognized arguments; raw private keys are never accepted on the CLI") cfg = {} if a.config.exists(): try: cfg = json.loads(a.config.read_text()) except (ValueError, OSError): p.error("chain configuration is unreadable or invalid JSON") if not isinstance(cfg, dict): p.error("chain configuration must be a JSON object") for field in ("rpc", "contract", "chainId"): if field in cfg and cfg[field] is None: p.error(f"chain configuration field {field} must not be null") for attr, field in (("rpc", "rpc"), ("contract", "contract"), ("chain", "chainId")): if getattr(a, attr) is None: setattr(a, attr, cfg.get(field)) if getattr(a, attr) is None: p.error(f"signed mining requires --{attr} or {field} in chain.json") if not isinstance(a.rpc, str) or urlparse(a.rpc).scheme not in ("http", "https") or not urlparse(a.rpc).hostname: p.error("RPC must be an explicit HTTP(S) URL") try: from eth_utils import is_address, to_checksum_address if not isinstance(a.contract, str) or not is_address(a.contract) or int(a.contract, 16) == 0: p.error("contract must be a nonzero 20-byte address") a.contract = to_checksum_address(a.contract) except ImportError: p.error("install the signing dependency: pip install eth-account") try: if isinstance(a.chain, bool) or isinstance(a.chain, float): raise ValueError a.chain = int(a.chain, 0) if isinstance(a.chain, str) else int(a.chain) if a.chain <= 0: raise ValueError except (TypeError, ValueError): p.error("expected chain id must be a positive integer") if a.max_mints < 0: p.error("max-mints must be nonnegative") for attr, unit in (("max_price_eth", 10**18), ("max_gas_price_gwei", 10**9), ("max_total_cost_eth", 10**18)): value = getattr(a, attr) if value is not None: try: value = Decimal(value) if not value.is_finite() or value < 0: raise ValueError with localcontext() as ctx: ctx.prec = max(100, len(value.as_tuple().digits) + 30) value = int(value * unit) except (InvalidOperation, ValueError, OverflowError): p.error("cost caps must be finite nonnegative decimal amounts") setattr(a, attr, value) return a def load_account(args): """Read a key locally, with no echo, diagnostic interpolation or persistence.""" from eth_account import Account key = None try: # Remove inherited env key before any engine subprocess can be spawned. env_key = os.environ.pop("HB_PRIVATE_KEY", None) if args.key_file is not None: try: fd = os.open(args.key_file, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) with os.fdopen(fd, "r") as stream: info = os.fstat(stream.fileno()) if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o600: raise MiningError("key file must be locally owned, regular and MODE600") key = stream.read(4097) except OSError: raise MiningError("key file must be locally owned, regular and MODE600 (no symlinks)") from None elif env_key is not None: key = env_key else: # getpass normally falls back to ECHOED stdin when no terminal exists. # Turn that fallback warning into a refusal instead. with warnings.catch_warnings(): warnings.simplefilter("error", getpass.GetPassWarning) key = getpass.getpass("PRIVATE KEY (local only, hidden): ") try: return Account.from_key(key.strip()) except (TypeError, ValueError): raise MiningError("invalid PRIVATE KEY; nothing was sent") from None except (getpass.GetPassWarning, EOFError): raise MiningError("a secure hidden terminal is required; use a MODE600 --key-file instead") from None finally: key = None env_key = None def _quantity(raw, name): try: if not isinstance(raw, str) or not raw.startswith("0x") or len(raw) < 3: raise ValueError number = int(raw, 16) if not 0 <= number < 2**256: raise ValueError return number except (TypeError, ValueError): raise MiningError(f"invalid RPC {name}") from None def _word(raw, name): if not isinstance(raw, str) or not raw.startswith("0x") or len(raw) != 66: raise MiningError(f"invalid RPC {name}: expected bytes32") _quantity(raw, name) return raw.lower() def _json_event(event): print(json.dumps(event, separators=(",", ":")), flush=True) class Miner: """One player, one outstanding signed transaction, bounded engine slices.""" hb_selectors = hb_work.SEL def __init__(self, args, account, engine, *, rpc_fn=None, sleep=None, emit=None): self.args, self.account, self.engine = args, account, engine self.rpc_fn = rpc_fn or hb_work.rpc self.sleep = sleep or time.sleep self.emit = emit or _json_event self.code = None self.state_dir = Path(args.state_dir) identity = f"{args.chain}-{account.address.lower()}" self.state_path = self.state_dir / f"{identity}.json" self.completed_path = self.state_path.with_suffix(".completed") self.lock_path = self.state_dir / f"{identity}.lock" self.confirmed = set() self.finalized = set() def rpc(self, method, params): return self.rpc_fn(self.args.rpc, method, params) def check_chain(self): if _quantity(self.rpc("eth_chainId", []), "chainId") != self.args.chain: raise MiningError("RPC chain differs from explicitly expected chain") def view(self, selector, tag): return self.rpc("eth_call", [{"to": self.args.contract, "data": "0x" + selector}, tag]) def snapshot(self): self.check_chain() tag = self.rpc("eth_blockNumber", []) _quantity(tag, "blockNumber") code = self.rpc("eth_getCode", [self.args.contract, tag]) try: if not isinstance(code, str) or not code.startswith("0x") or not bytes.fromhex(code[2:]): raise ValueError except ValueError: raise MiningError("configured contract has no valid live code") from None if self.code is not None and self.code != code.lower(): raise MiningError("live contract code changed; refusing to sign") self.code = code.lower() work = {"miner": self.account.address, "blockTag": tag, "prevHex": _word(self.view(self.hb_selectors["prev"], tag), "prevWork"), "anchorBlock": _quantity(self.view(self.hb_selectors["challengeBlock"], tag), "challengeBlock"), "anchorHash": _word(self.view(self.hb_selectors["challengeAnchor"], tag), "challengeAnchor"), "targetHex": _word(self.view(self.hb_selectors["target"] + hb_work.pad32(self.account.address), tag), "targetFor"), "priceWei": _quantity(self.view(self.hb_selectors["price"], tag), "enlistPrice")} if not work["anchorBlock"] or not int(work["anchorHash"], 16) or not int(work["targetHex"], 16): raise MiningError("canonical mining anchor or target unavailable") self.check_chain() return work @staticmethod def same_work(left, right): # A new offered anchor does not invalidate the solved one. challengeBlock # is the canonical L2 height minus one (not necessarily eth_blockNumber). age = right["anchorBlock"] + 1 - left["anchorBlock"] return (all(left[field] == right[field] for field in ("prevHex", "targetHex")) and 0 < age <= 250 and (left["anchorBlock"] != right["anchorBlock"] or left["anchorHash"] == right["anchorHash"])) def verify(self, work, hit): try: nonce = hit["nonce"] digest = hit["hash"].removeprefix("0x").lower() if isinstance(nonce, bool) or not isinstance(nonce, int) or not 0 <= nonce < 2**256 or len(digest) != 64: raise ValueError recomputed = hb_work.keccak256_hex(hb_work.make_preimage( self.account.address, nonce, work["prevHex"], work["anchorHash"]).hex()) if digest != recomputed or int(digest, 16) >= int(work["targetHex"], 16): raise ValueError return "0x" + digest except (KeyError, TypeError, AttributeError, ValueError): raise MiningError("engine proof failed local C verification; refusing gas/signature") from None @contextmanager def locked(self): self.state_dir.mkdir(mode=0o700, parents=True, exist_ok=True) info = self.state_dir.lstat() if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700: raise MiningError("state directory must be locally owned, not a symlink, and MODE700") fd = os.open(self.lock_path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600) try: try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise MiningError("wallet process lock held: another miner is running") from None yield finally: os.close(fd) def save_state(self, record, *, path=None): path = self.state_path if path is None else path temporary = self.state_path.with_suffix(".tmp") fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600) try: os.fchmod(fd, 0o600) with os.fdopen(fd, "w") as stream: fd = None json.dump(record, stream, separators=(",", ":")) stream.flush() os.fsync(stream.fileno()) os.replace(temporary, path) self.sync_directory() finally: if fd is not None: os.close(fd) def sync_directory(self): fd = os.open(self.state_dir, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(fd) finally: os.close(fd) def load_state(self, *, path=None): path = self.state_path if path is None else path try: fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) except FileNotFoundError: return None except OSError: raise MiningError("transaction state disk read failed; refusing new work") from None try: with os.fdopen(fd, "r") as stream: info = os.fstat(stream.fileno()) if stat.S_IMODE(info.st_mode) != 0o600 or info.st_uid != os.getuid() or not stat.S_ISREG(info.st_mode): raise MiningError("pending state must be locally owned and MODE600") record = json.load(stream) expected = ("confirmed", "reverted") if path == self.completed_path else ("signed",) if record["version"] != 1 or record["status"] not in expected: raise MiningError("invalid transaction state status; refusing new work") if path == self.completed_path: _quantity(record["receiptBlock"], "terminal receipt block") if record["chainId"] != self.args.chain or record["contract"].lower() != self.args.contract.lower() or record["miner"] != self.account.address: raise MiningError("pending transaction belongs to a different chain, contract or player") from eth_account import Account from eth_account._utils.legacy_transactions import Transaction from eth_utils import keccak raw = bytes.fromhex(record["rawTransaction"].removeprefix("0x")) if "0x" + keccak(raw).hex() != record["txHash"] or Account.recover_transaction(raw) != self.account.address: raise MiningError("invalid pending signature/hash; state retained") tx = Transaction.from_bytes(raw).as_dict() data = bytes(tx["data"]) if (tx["v"] not in (2 * self.args.chain + 35, 2 * self.args.chain + 36) or bytes(tx["to"]).hex() != self.args.contract[2:].lower() or tx["nonce"] != record["transactionNonce"] or len(data) != 68 or data[:4].hex() != self.hb_selectors["mine"] or int.from_bytes(data[4:36], "big") != record["nonce"] or int.from_bytes(data[36:], "big") != record["anchorBlock"]): raise MiningError("pending metadata differs from actual signed transaction") try: digest = self.verify(record["work"], {"nonce": record["nonce"], "hash": record["digest"]}) except MiningError: raise MiningError("pending proof failed local C verification; state retained") from None if digest != record["digest"]: raise MiningError("pending digest differs from signed proof") return record except (ValueError, KeyError, TypeError): raise MiningError("invalid pending transaction state; refusing new work") from None except OSError: raise MiningError("transaction state disk read failed; refusing new work") from None def clear_state(self, record): # An older completion can coexist with the next durable signed tx. # Never let recovery of that completion erase different signed bytes. pending = self.load_state() if pending is not None and pending["txHash"] == record["txHash"]: self.state_path.unlink(missing_ok=True) self.sync_directory() def event(self, name, record=None, **fields): event = {"event": name, "engine": self.args.engine} if record is not None: event.update({key: record[key] for key in ("nonce", "digest")}) if "txHash" in record: event["txHash"] = record["txHash"] self.emit({**event, **fields}) def check_caps(self, price, gas_price, gas=None): for actual, maximum, label in ( (price, self.args.max_price_eth, "price"), (gas_price, self.args.max_gas_price_gwei, "gas price"), (price + (gas or 0) * gas_price, self.args.max_total_cost_eth if gas is not None else None, "total cost"), ): if maximum is not None and actual > maximum: raise MiningError(f"user {label} cap exceeded; nothing signed") def prepare(self, work, nonce, digest): gas_price = _quantity(self.rpc("eth_gasPrice", []), "gasPrice") self.check_caps(work["priceWei"], gas_price) data = "0x" + self.hb_selectors["mine"] + format(nonce, "064x") + format(work["anchorBlock"], "064x") tx_nonce = _quantity(self.rpc("eth_getTransactionCount", [self.account.address, "pending"]), "pending nonce") call_tx = {"from": self.account.address, "to": self.args.contract, "data": data, "value": hex(work["priceWei"]), "gasPrice": hex(gas_price), "nonce": hex(tx_nonce)} try: estimate = _quantity(self.rpc("eth_estimateGas", [call_tx]), "gas estimate") except Exception: current = self.snapshot() if not self.same_work(work, current) or work["priceWei"] != current["priceWei"]: return None raise MiningError("gas estimation failed; no signature created") from None if not estimate: raise MiningError("empty gas estimate") gas = (estimate * 120 + 99) // 100 self.check_caps(work["priceWei"], gas_price, gas) # The estimate validates the original anchor's canonical hash/proof; # recheck its L2 age and live work/price after the estimate's race window. final_work = self.snapshot() if not self.same_work(work, final_work) or work["priceWei"] != final_work["priceWei"]: return None pending_now = _quantity(self.rpc("eth_getTransactionCount", [self.account.address, "pending"]), "pending nonce") if pending_now != tx_nonce: return None self.check_chain() # last network guard before the local signing operation # Legacy tx is completely formed locally; no eth_sendTransaction path. signed = self.account.sign_transaction({ "to": self.args.contract, "data": data, "value": work["priceWei"], "gas": gas, "gasPrice": gas_price, "nonce": tx_nonce, "chainId": self.args.chain, }) raw = "0x" + bytes(signed.raw_transaction).hex() record = {"version": 1, "status": "signed", "miner": self.account.address, "chainId": self.args.chain, "contract": self.args.contract, "nonce": nonce, "transactionNonce": tx_nonce, "digest": digest, "txHash": "0x" + bytes(signed.hash).hex(), "rawTransaction": raw, "code": self.code, "anchorBlock": work["anchorBlock"], "engine": self.args.engine, "work": work} self.save_state(record) # durable before ANY broadcast (including Ctrl-C) return record def broadcast(self, record): self.check_chain() code = self.rpc("eth_getCode", [self.args.contract, "latest"]) if not isinstance(code, str) or code.lower() != record["code"]: raise MiningError("live contract code differs from pending signed transaction") try: response = self.rpc("eth_sendRawTransaction", [record["rawTransaction"]]) except Exception: # Unknown acceptance. Never construct a replacement or change nonce. response = None if response is not None and response.lower() != record["txHash"].lower(): raise MiningError("RPC returned unexpected signed transaction hash; state retained") self.event("submitted", record) def validate_receipt(self, record, receipt): """Only receipt reads/validation belong to the retryable RPC phase.""" if receipt.get("transactionHash", "").lower() != record["txHash"].lower(): raise MiningError("receipt transaction hash mismatch; state retained") status = _quantity(receipt.get("status"), "receipt status") tag = receipt.get("blockNumber") _quantity(tag, "receipt block") if status == 0: return {**record, "status": "reverted", "receiptBlock": tag} if status != 1: raise MiningError("invalid receipt status; state retained") on_chain = _word(self.view(self.hb_selectors["prev"], tag), "receipt prevWork") if on_chain != record["digest"]: raise MiningError("receipt prevWork differs from mined digest; state retained") from eth_utils import keccak topic = "0x" + keccak(text="Transfer(address,address,uint256)").hex() destination = "0x" + hb_work.pad32(self.account.address) minted = any( isinstance(log, dict) and log.get("address", "").lower() == self.args.contract.lower() and not log.get("removed", False) and len(log.get("topics", [])) == 4 and log["topics"][0].lower() == topic and log["topics"][1].lower() == "0x" + "00" * 32 and log["topics"][2].lower() == destination for log in receipt.get("logs", []) ) if not minted: raise MiningError("receipt lacks mint Transfer to player; state retained") return {**record, "status": "confirmed", "receiptBlock": tag} def finalize_receipt(self, record): """Persist terminal evidence BEFORE deleting pending or emitting success. Keep one MODE600 completion (including its signed bytes/C proof) until replaced by the next terminal result. Restart counts this result before any new signing; this is recovery, not a global session mint budget. """ success = record["status"] == "confirmed" txhash = record["txHash"] if txhash in self.finalized: return success try: # Also resync recovered evidence: its prior directory fsync might # have been interrupted. Never delete pending on uncertain durability. self.save_state(record, path=self.completed_path) if success: self.confirmed.add(txhash) self.clear_state(record) except OSError: raise MiningError("receipt finalization disk durability failed; recovery state retained, restart required") from None # Mark before output: interruption after a consumer sees the event must # not repeat it in this process. A restart may report recovered success. self.finalized.add(txhash) if success: self.event("confirmed", record, blockNumber=record["receiptBlock"]) return success def reconcile(self, record, *, fresh=False): if record["txHash"] in self.finalized: return record["txHash"] in self.confirmed if record["status"] in ("confirmed", "reverted"): return self.finalize_receipt(record) if fresh: self.broadcast(record) while True: self.check_chain() try: receipt = self.rpc("eth_getTransactionReceipt", [record["txHash"]]) except Exception: self.sleep(2) continue if receipt is not None: # Archive/read outages retry; semantic proof failures stop closed. try: terminal = self.validate_receipt(record, receipt) except MiningError: raise except Exception: self.sleep(2) continue # Local durability/output failures are NOT archive outages. return self.finalize_receipt(terminal) try: known = self.rpc("eth_getTransactionByHash", [record["txHash"]]) except Exception: self.sleep(2) continue if known is None: pending_nonce = _quantity(self.rpc("eth_getTransactionCount", [self.account.address, "pending"]), "pending nonce") if pending_nonce > record["transactionNonce"]: raise MiningError("pending nonce consumed elsewhere; state retained, reconcile manually") self.broadcast(record) # EXACT same bytes only self.sleep(2) def run(self): reason = "limit" try: with self.locked(): self.event("start", miner=self.account.address, chainId=self.args.chain, contract=self.args.contract) pending = self.load_state() completed = self.load_state(path=self.completed_path) if completed is not None: self.code = completed["code"] self.finalize_receipt(completed) if pending is not None: self.code = pending["code"] if completed is None or pending["txHash"] != completed["txHash"]: self.reconcile(pending) start_nonce = secrets.randbits(48) while self.args.max_mints == 0 or len(self.confirmed) < self.args.max_mints: if start_nonce >= 2**64: raise MiningError("engine nonce range exhausted; restart miner") work = self.snapshot() with redirect_stdout(sys.stderr): hit, hashes = self.engine.search(work, start_nonce) if isinstance(hashes, bool) or not isinstance(hashes, int) or hashes <= 0: raise MiningError("engine must report a positive bounded nonce range") next_nonce = start_nonce + hashes # 2**64 is the valid exclusive end of the final window. # Process its hit now; stop before a subsequent search. if next_nonce > 2**64: raise MiningError("engine nonce range exceeds uint64") if hit is not None and not start_nonce <= hit.get("nonce", -1) < next_nonce: raise MiningError("engine proof nonce outside searched range") start_nonce = next_nonce if hit is None: continue digest = self.verify(work, hit) live = self.snapshot() if not self.same_work(work, live): continue self.event("found", {"nonce": hit["nonce"], "digest": digest}) record = self.prepare({**work, "priceWei": live["priceWei"]}, hit["nonce"], digest) if record is not None: self.reconcile(record, fresh=True) except KeyboardInterrupt: reason = "interrupted" finally: close = getattr(self.engine, "close", None) if close: close() self.event("done", mints=len(self.confirmed), reason=reason) return len(self.confirmed) def main(argv=None): args = parse_args(argv) try: account = load_account(args) # Engine import/build happens AFTER the environment key is removed. from hb_engines import CpuEngine, GpuEngine with redirect_stdout(sys.stderr): engine = (CpuEngine if args.engine == "cpu" else GpuEngine)(KIT_DIR) Miner(args, account, engine).run() return 0 except KeyboardInterrupt: return 130 except MiningError as error: print(f"miner stopped: {error}", file=sys.stderr) return 1 except Exception: # Do not display arbitrary exceptions: transports/signers can echo data. print("miner stopped: RPC/engine/signing failure; pending state retained if present", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())