#!/usr/bin/env python3 """Search the Hexis ergon from a computer. This uses the processor cores. It is not a graphics-card miner. A browser search and this program share the same 6-hour pass. Install the hash library once: pip install pycryptodome Then, after the person check on the Ergon page: python ergon.py --address HEXabc... --pass PASTE """ from __future__ import annotations import argparse import json import os import sys import threading import time import urllib.request DEFAULT_URL = "https://testnet.hexis.network/api/rpc" def die(message: str) -> None: raise SystemExit(message) def keccak256(data: bytes) -> bytes: try: from Crypto.Hash import keccak except ImportError: die("Install the hash library first: pip install pycryptodome") digest = keccak.new(digest_bits=256) digest.update(data) return digest.digest() def hex_int(value: str) -> int: return int(value, 16) def address_body(text: str) -> str: body = text.strip() if body[:3].lower() in ("hex", "kal"): body = body[3:] if body[:2].lower() == "0x": body = body[2:] if len(body) != 40 or any(char not in "0123456789abcdefABCDEF" for char in body): die("The address has to be HEX and 40 hex characters.") return body.lower() def preimage(chain_id: int, slot: int, parent: bytes, address: bytes, nonce: int) -> bytes: out = bytearray(82) out[0:6] = b"HEXIS1" out[6:14] = chain_id.to_bytes(8, "big") out[14:22] = slot.to_bytes(8, "big") out[22:54] = parent out[54:74] = address out[74:82] = nonce.to_bytes(8, "big") return bytes(out) def below_target(digest: bytes, target: bytes) -> bool: return digest < target class Search: def __init__(self, url: str, address: str, token: str, workers: int) -> None: self.url = url self.address = "HEX" + address_body(address) self.body = bytes.fromhex(address_body(address)) self.token = token.strip() self.workers = max(1, workers) self.stop = threading.Event() self.lock = threading.Lock() self.job: dict | None = None self.accepted = 0 def call(self, method: str, params: list) -> object: payload = { "jsonrpc": "2.0", "id": 1, "method": method, "params": params, "minePass": self.token, } request = urllib.request.Request( self.url, data=json.dumps(payload).encode(), headers={"content-type": "application/json"}, ) with urllib.request.urlopen(request, timeout=30) as response: incoming = json.loads(response.read().decode()) if incoming.get("ok") is False: die(str(incoming.get("error") or "The chain rejected that.")) if isinstance(incoming.get("error"), dict): die(str(incoming["error"].get("message") or "The chain rejected that.")) if isinstance(incoming.get("error"), str): die(incoming["error"]) return incoming.get("result") def beat(self) -> None: beat = self.url[:-8] + "/api/mine/beat" if self.url.endswith("/api/rpc") else self.url request = urllib.request.Request( beat, data=json.dumps({"pass": self.token}).encode(), headers={"content-type": "application/json"}, ) try: with urllib.request.urlopen(request, timeout=20) as response: incoming = json.loads(response.read().decode()) except Exception: return if incoming.get("ok") is False: die(str(incoming.get("error") or "The day's time is used.")) def refresh(self) -> dict: work = self.call("qrl_getWork", []) if not isinstance(work, dict): die("The chain did not return an ergon.") with self.lock: self.job = work difficulty = hex_int(work["difficulty"]) print( f"Keccak-256 difficulty {difficulty:,} slot {hex_int(work['slot'])}", flush=True, ) return work def submit(self, nonce: int, work: dict) -> None: result = self.call( "qrl_submitWork", [hex(nonce), self.address, work["timestamp"], work["parentHash"]], ) accepted = result is True or (isinstance(result, dict) and result.get("accepted")) if not accepted: return number = result.get("number") if isinstance(result, dict) else None with self.lock: self.accepted += 1 count = self.accepted label = f" Guess {number}." if number else "" print(f"Accepted.{label} This visit: {count}. One block can pay it once.", flush=True) def worker(self, index: int) -> None: nonce = index while not self.stop.is_set(): with self.lock: work = dict(self.job) if self.job else None if not work: time.sleep(0.2) continue chain_id = hex_int(work["chainId"]) slot = hex_int(work["slot"]) parent = bytes.fromhex(work["parentHash"][2:]) target = bytes.fromhex(work["target"][2:].rjust(64, "0")) for _ in range(2000): digest = keccak256(preimage(chain_id, slot, parent, self.body, nonce)) if below_target(digest, target): try: self.submit(nonce, work) except SystemExit as exc: print(exc, flush=True) self.stop.set() return except Exception as exc: print(exc, flush=True) break nonce += self.workers else: continue nonce += self.workers def run(self) -> None: if len(self.token) < 16: die("Paste the search pass from the Ergon page.") self.refresh() threads = [ threading.Thread(target=self.worker, args=(index,), daemon=True) for index in range(self.workers) ] for thread in threads: thread.start() print(f"Searching with {self.workers} cores. This does not spend a signature.", flush=True) while not self.stop.is_set(): time.sleep(10) try: self.beat() previous = self.job current = self.refresh() except SystemExit as exc: print(exc, flush=True) self.stop.set() return if previous and ( current["parentHash"] != previous["parentHash"] or current["timestamp"] != previous["timestamp"] ): print("The ergon moved.", flush=True) def main() -> None: parser = argparse.ArgumentParser(description="Search the Hexis ergon on the CPU.") parser.add_argument("--address", required=True) parser.add_argument("--pass", dest="token", required=True) parser.add_argument("--url", default=DEFAULT_URL) parser.add_argument("--workers", type=int, default=os.cpu_count() or 1) args = parser.parse_args() try: Search(args.url, args.address, args.token, args.workers).run() except KeyboardInterrupt: print("Stopped.", flush=True) if __name__ == "__main__": main()