#!/usr/bin/env python3 """ $WHOLE Protocol - Single-File Miner =================================== Mines memory-hard Merkle proofs off-chain and submits them to the WholeToken contract on Base (mainnet). HOW TO USE (3 steps): 1. Install Python 3.10+ -> https://www.python.org/downloads/ 2. Install the one dependency: pip install web3 3. Create a file named ".env" next to this file (see the example below), put your wallet private key in it, then run: python whole-miner.py The contract address and network are ALREADY filled in for you. The ONLY thing you must set is your PRIVATE_KEY in the .env file. ------------------------------------------------------------------- Example .env file (create it in the same folder as this script): PRIVATE_KEY=0xyour_wallet_private_key_here # Optional - only change if you know what you are doing: # RPC_URL=https://mainnet.base.org # CONTRACT=0x2f8A1A11Ab90Eb03D83ad3250956A537ec9037D9 # DRY_RUN=false ------------------------------------------------------------------- WARNING: Never share your private key. Use a dedicated wallet, not your main wallet. You need a small amount of ETH on Base to pay the scan fee (0.0005 ETH per scan). Bridge ETH to Base at https://bridge.base.org. """ import os import sys import time from dataclasses import dataclass from typing import Optional # --------------------------------------------------------------------------- # Default configuration (already filled in - you normally do not touch this) # --------------------------------------------------------------------------- DEFAULT_RPC_URL = "https://mainnet.base.org" DEFAULT_CONTRACT = "0x2f8A1A11Ab90Eb03D83ad3250956A537ec9037D9" SCAN_FEE_ETH = 0.0005 # 0.0005 ETH per scan ARRAY_SIZE = 65_536 # memory-hard array size (64k x 32 bytes ~= 2 MB) TARGET_MIN_GAP = 10 # targetBlock must be at least n-10 TARGET_MAX_GAP = 200 # and at most n-200 (safety margin before 256) # --------------------------------------------------------------------------- # Tiny .env loader (so you only need `pip install web3`, nothing else) # --------------------------------------------------------------------------- def load_env(path: str = ".env") -> dict: """Reads a simple KEY=VALUE .env file. Ignores blank lines and #comments.""" values = {} if os.path.exists(path): with open(path, "r", encoding="utf-8") as fh: for raw in fh: line = raw.strip() if not line or line.startswith("#"): continue if "=" not in line: continue key, _, val = line.partition("=") values[key.strip()] = val.strip().strip('"').strip("'") # Environment variables win over the file for k in ("PRIVATE_KEY", "RPC_URL", "CONTRACT", "DRY_RUN"): if os.environ.get(k): values[k] = os.environ[k] return values try: from web3 import Web3 from eth_account import Account except ImportError: print("\n[ERROR] The 'web3' package is not installed.") print(" Fix it by running this command in your terminal:\n") print(" pip install web3\n") sys.exit(1) # Minimal ABI to interact with WholeToken ABI_MINIMAL = [ {"inputs": [], "name": "currentEra", "outputs": [{"type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "currentEpisode", "outputs": [{"type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getEffectiveDifficulty", "outputs": [{"type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "isEpisodeReady", "outputs": [{"type": "bool"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getCosmicReservoir", "outputs": [{"type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [ {"name": "merkleRoot", "type": "bytes32"}, {"name": "merkleProof", "type": "bytes32[]"}, {"name": "nonce", "type": "uint256"}, {"name": "targetBlock", "type": "uint256"}, {"name": "leafHash", "type": "bytes32"}, ], "name": "submitScan", "outputs": [], "stateMutability": "payable", "type": "function"}, ] # --------------------------------------------------------------------------- # Hashing helpers # --------------------------------------------------------------------------- def keccak256(data: bytes) -> bytes: return Web3.keccak(primitive=data) def count_leading_zero_nibbles(h: bytes) -> int: """Counts leading zero nibbles (half-bytes) of a hash.""" count = 0 for byte in h: high = (byte >> 4) & 0xF low = byte & 0xF if high != 0: break count += 1 if low != 0: break count += 1 return count # --------------------------------------------------------------------------- # Memory-hard Merkle proof # --------------------------------------------------------------------------- @dataclass class MerkleProof: root: bytes leaf_hash: bytes siblings: list leaf_index: int def build_memory_hard_array(seed: bytes, array_size: int) -> list: """Sequentially chained hashes - memory-hard, impossible to parallelize.""" print(f" [*] Building memory-hard array ({array_size} elements, " f"{array_size * 32 // 1024} KB)...") arr = [] prev = keccak256(seed + (0).to_bytes(8, "big")) arr.append(prev) for i in range(1, array_size): prev = keccak256(prev + i.to_bytes(8, "big")) arr.append(prev) return arr def build_merkle_tree(leaves: list) -> list: """Builds a sorted-pair Merkle tree (matches MerkleProofLib.sol).""" n = len(leaves) while n & (n - 1): leaves = leaves + [leaves[-1]] n += 1 levels = [leaves] current = leaves while len(current) > 1: parent = [] for i in range(0, len(current), 2): a, b = current[i], current[i + 1] if a <= b: parent.append(keccak256(a + b)) else: parent.append(keccak256(b + a)) current = parent levels.append(current) return levels def get_merkle_proof(levels: list, leaf_index: int) -> MerkleProof: root = levels[-1][0] leaf_hash = levels[0][leaf_index] siblings = [] idx = leaf_index for level in levels[:-1]: sibling_idx = idx ^ 1 if sibling_idx < len(level): siblings.append(level[sibling_idx]) else: siblings.append(level[idx]) idx //= 2 return MerkleProof(root=root, leaf_hash=leaf_hash, siblings=siblings, leaf_index=leaf_index) # --------------------------------------------------------------------------- # Mining logic # --------------------------------------------------------------------------- def compute_seed(miner_address: str, target_block: int, block_hash: bytes) -> bytes: """Rebuilds the seed exactly like the Solidity contract: keccak256(abi.encodePacked(msg.sender, targetBlock, blockhash(targetBlock)))""" addr_bytes = bytes.fromhex(miner_address[2:]) block_bytes = target_block.to_bytes(32, "big") return keccak256(addr_bytes + block_bytes + block_hash) def mine(seed: bytes, merkle_root: bytes, difficulty: int, hawk_diff: int, max_attempts: int = 50_000_000): """Searches for a nonce whose finalHash has enough leading zero nibbles.""" print(f" [*] Mining... difficulty={difficulty} (Hawking>={hawk_diff}), " f"up to {max_attempts:,} attempts") start_time = time.time() best_lz = 0 for nonce in range(max_attempts): nonce_bytes = nonce.to_bytes(32, "big") final_hash = keccak256(seed + merkle_root + nonce_bytes) lz = count_leading_zero_nibbles(final_hash) if lz > best_lz: best_lz = lz if lz >= difficulty: elapsed = time.time() - start_time print(f" [OK] SINGULARITY! nonce={nonce}, {lz} nibbles in {elapsed:.2f}s") return nonce, lz, "singularity" if lz >= hawk_diff: elapsed = time.time() - start_time print(f" [~] HAWKING! nonce={nonce}, {lz} nibbles in {elapsed:.2f}s") return nonce, lz, "hawking" if nonce % 500_000 == 0 and nonce > 0: elapsed = time.time() - start_time hashrate = nonce / elapsed if elapsed > 0 else 0 print(f" {nonce:>10,} attempts | {hashrate:,.0f} H/s | " f"best={best_lz} nibbles") elapsed = time.time() - start_time print(f" [x] Not found in {max_attempts:,} attempts ({elapsed:.2f}s). " f"Best: {best_lz} nibbles.") return None, best_lz, "miss" # --------------------------------------------------------------------------- # On-chain submission # --------------------------------------------------------------------------- def submit_scan(w3, contract, account, merkle_root, proof, nonce, target_block, leaf_hash, dry_run=False): scan_fee_wei = Web3.to_wei(SCAN_FEE_ETH, "ether") mr_hex = "0x" + merkle_root.hex() lh_hex = "0x" + leaf_hash.hex() p_hex = ["0x" + s.hex() for s in proof] print(f"\n merkleRoot : {mr_hex}") print(f" leafHash : {lh_hex}") print(f" proof nodes : {len(p_hex)}") print(f" nonce : {nonce}") print(f" targetBlock : {target_block}") print(f" fee : {SCAN_FEE_ETH} ETH") if dry_run: print(" [DRY-RUN] Transaction NOT submitted.") return None gas_estimate = contract.functions.submitScan( mr_hex, p_hex, nonce, target_block, lh_hex ).estimate_gas({"from": account.address, "value": scan_fee_wei}) tx = contract.functions.submitScan( mr_hex, p_hex, nonce, target_block, lh_hex ).build_transaction({ "from": account.address, "value": scan_fee_wei, "gas": int(gas_estimate * 1.2), "nonce": w3.eth.get_transaction_count(account.address), }) signed = account.sign_transaction(tx) tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction) print(f" [OK] Transaction sent: 0x{tx_hash.hex()}") receipt = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=120) status = "OK Confirmed" if receipt.status == 1 else "x Reverted" print(f" [{status}] Block #{receipt.blockNumber}, gas used: {receipt.gasUsed:,}") return tx_hash.hex() # --------------------------------------------------------------------------- # Main mining loop # --------------------------------------------------------------------------- def run_miner(rpc_url, contract_addr, private_key, dry_run): print("=" * 60) print(" $WHOLE Protocol - Single-File Miner") print("=" * 60) w3 = Web3(Web3.HTTPProvider(rpc_url)) if not w3.is_connected(): print(f"[ERROR] Could not connect to {rpc_url}") sys.exit(1) print(f" Network : chain ID {w3.eth.chain_id}") account = Account.from_key(private_key) contract = w3.eth.contract(address=Web3.to_checksum_address(contract_addr), abi=ABI_MINIMAL) balance_wei = w3.eth.get_balance(account.address) print(f" Miner : {account.address}") print(f" Balance : {Web3.from_wei(balance_wei, 'ether'):.5f} ETH") print(f" Contract : {contract_addr}") print(f" Dry-run : {dry_run}") if balance_wei < Web3.to_wei(SCAN_FEE_ETH, "ether") and not dry_run: print("\n [!] Your balance is lower than the scan fee (0.0005 ETH).") print(" Bridge ETH to Base at https://bridge.base.org\n") loop_count = 0 while True: loop_count += 1 print(f"\n{'-' * 60}") print(f" Loop #{loop_count}") try: era = contract.functions.currentEra().call() episode = contract.functions.currentEpisode().call() diff = contract.functions.getEffectiveDifficulty().call() ready = contract.functions.isEpisodeReady().call() reservoir = contract.functions.getCosmicReservoir().call() except Exception as e: print(f" [!] RPC connection error: {e}") print(f" [!] Reconnecting in 15s...") time.sleep(15) try: w3 = Web3(Web3.HTTPProvider(rpc_url)) contract = w3.eth.contract(address=Web3.to_checksum_address(contract_addr), abi=ABI) print(" [OK] Reconnected.") except Exception: pass continue print(f" Era #{era}, Episode #{episode}") print(f" Effective difficulty : {diff} nibbles") print(f" Episode ready : {ready}") print(f" Cosmic Reservoir : {Web3.from_wei(reservoir, 'ether'):.4f} ETH") if not ready: print(" [!] Cooldown active (300 blocks). Waiting ~600s...") time.sleep(600) continue try: current_block = w3.eth.block_number target_block = current_block - TARGET_MIN_GAP - 5 if target_block < current_block - TARGET_MAX_GAP: target_block = current_block - TARGET_MIN_GAP - 5 block_data = w3.eth.get_block(target_block) block_hash = bytes(block_data.hash) except Exception as e: print(f" [!] RPC error fetching block: {e}") print(f" [!] Retrying in 15s...") time.sleep(15) continue print(f" Target block : #{target_block} (hash: 0x{block_hash.hex()[:16]}...)") seed = compute_seed(account.address, target_block, block_hash) print(f" Seed : 0x{seed.hex()[:32]}...") arr = build_memory_hard_array(seed, ARRAY_SIZE) levels = build_merkle_tree(arr) leaf_index = int.from_bytes(seed[:4], "big") % ARRAY_SIZE mp = get_merkle_proof(levels, leaf_index) print(f" Merkle root : 0x{mp.root.hex()[:32]}...") print(f" Leaf # : {mp.leaf_index}") hawk_diff = max(1, diff // 2) nonce, lz, result_type = mine(seed, mp.root, diff, hawk_diff) if result_type == "miss": print(" Retrying next round in 5s...") time.sleep(5) continue print(f"\n Submitting proof ({result_type})...") try: submit_scan(w3, contract, account, mp.root, mp.siblings, nonce, target_block, mp.leaf_hash, dry_run=dry_run) except Exception as e: print(f" [!] Submit error: {e}") print(f" [!] Retrying loop in 15s...") time.sleep(15) continue if dry_run: print(" [DRY-RUN] Exiting after first proof generated.") break print(" Waiting for confirmation and cooldown before next attempt...") time.sleep(600 + 30) def main(): cfg = load_env(".env") private_key = cfg.get("PRIVATE_KEY", "").strip() rpc_url = cfg.get("RPC_URL", DEFAULT_RPC_URL).strip() or DEFAULT_RPC_URL contract = cfg.get("CONTRACT", DEFAULT_CONTRACT).strip() or DEFAULT_CONTRACT dry_run = cfg.get("DRY_RUN", "false").strip().lower() in ("1", "true", "yes") if not private_key or private_key.startswith("0xyour"): print("\n[ERROR] No PRIVATE_KEY found.") print(" Create a file named '.env' in this folder with this line:\n") print(" PRIVATE_KEY=0xyour_wallet_private_key_here\n") print(" Then run again: python whole-miner.py\n") sys.exit(1) if not private_key.startswith("0x"): private_key = "0x" + private_key run_miner(rpc_url, contract, private_key, dry_run) if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\n\n Stopped by user. Goodbye!") sys.exit(0)