#!/usr/bin/env python3 """Hermes Grows - verify a Proof of Growth snapshot. Python 3.9+, standard library only. python3 verify.py --latest | --day N | --date YYYY-MM-DD published files + Solana Memo anchor python3 verify.py --local public/snapshots/YYYY-MM-DD files on disk Exit 0 when every check passes, 1 otherwise. """ import argparse, hashlib, json, os, re, sys, time, urllib.parse, urllib.request BASE = "https://hermesgrows.online/public/" RPC = "https://solana-rpc.publicnode.com" MEMO_PROGRAM = "MemoSq4gqABAXKb96qnH8TysNcWxMyWCqXgDLGmfcHr" FAILED = [] sha256 = lambda data: hashlib.sha256(data).hexdigest() # noqa: E731 memo_text = lambda day, date, root: "GROWS v1 day=%d date=%s sha256=%s" % (day, date, root) # noqa: E731 def check(ok, label, detail=""): if not ok: FAILED.append(label) print(("✓ " if ok else "✗ ") + label + (" " + str(detail) if detail != "" else "")) return ok def canonical(obj): """Canonical manifest bytes: sorted keys, no whitespace, UTF-8, final newline.""" return (json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n").encode("utf-8") def http(url, body=None): # A User-Agent is required: publicnode answers 403 to Python's default one. headers = {"User-Agent": "grows-verify/1", "Content-Type": "application/json"} with urllib.request.urlopen(urllib.request.Request(url, data=body, headers=headers), timeout=30) as r: return r.read() def verify_snapshot(raw, read, listing=None): """raw: manifest.json bytes. read(path) -> bytes of a listed file. Returns (manifest, root hex).""" m = json.loads(raw.decode("utf-8")) check(canonical(m) == raw, "manifest.json is canonical", "sorted keys, no spaces, UTF-8, final newline") check(m.get("version") == 1, "schema version", m.get("version")) files = [] for f in m.get("files", []): p = f.get("path", "") if not re.fullmatch(r"[A-Za-z0-9._-]+(/[A-Za-z0-9._-]+)*", p) or ".." in p.split("/"): check(False, "unsafe path in manifest", repr(p)) continue try: data = read(p) except Exception as e: # noqa: BLE001 - report and continue check(False, p, "unavailable (%s)" % e) continue digest = sha256(data) check(digest == f.get("sha256") and len(data) == f.get("bytes"), p, "sha256 %s %d bytes" % (digest, len(data))) files.append({"bytes": len(data), "path": p, "sha256": digest}) if p == "stats.json": check(json.loads(data.decode("utf-8")) == m.get("stats"), "manifest stats == stats.json") if listing is not None: extra = sorted(set(listing) - {f.get("path") for f in m.get("files", [])}) check(not extra, "folder holds only listed files", ", ".join(extra)) rebuilt = canonical({"version": 1, "day": m.get("day"), "date": m.get("date"), "files": sorted(files, key=lambda f: f["path"]), "stats": m.get("stats", {})}) check(rebuilt == raw, "manifest rebuilt from files, byte for byte", "%d bytes" % len(rebuilt)) root = sha256(raw) print(" root sha256 %s\n memo %s" % (root, memo_text(m.get("day", 0), m.get("date", "?"), root))) return m, root def onchain_memos(rpc, sig): body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "getTransaction", "params": [ sig, {"encoding": "jsonParsed", "maxSupportedTransactionVersion": 0, "commitment": "confirmed"}]}).encode() res = json.loads(http(rpc, body)) tx = res.get("result") if not tx: raise RuntimeError((res.get("error") or {}).get("message", "transaction not found")) memos = [ix["parsed"] for ix in tx["transaction"]["message"]["instructions"] if ix.get("programId") == MEMO_PROGRAM and isinstance(ix.get("parsed"), str)] for line in [] if memos else (tx.get("meta") or {}).get("logMessages") or []: # fallback: program logs hit = re.match(r'Program log: Memo \(len \d+\): "(.*)"$', line) if hit: memos.append(hit.group(1)) return memos, tx def verify_anchor(anchor, m, root, rpc): memo = memo_text(m["day"], m["date"], root) check(anchor.get("sha256") == root, "anchors.json sha256 == root hash", anchor.get("sha256")) check(anchor.get("memo") == memo, "anchors.json memo text", anchor.get("memo")) sig = anchor.get("signature", "") try: memos, tx = onchain_memos(rpc, sig) except Exception as e: # noqa: BLE001 return check(False, "Solana transaction " + sig, "unavailable (%s)" % e) when = time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime(tx["blockTime"])) if tx.get("blockTime") else "n/a" check((tx.get("meta") or {}).get("err") is None, "transaction confirmed without error", "slot %s, %s" % (tx.get("slot"), when)) check(tx.get("slot") == anchor.get("slot"), "slot matches anchors.json", anchor.get("slot")) check(memo in memos, "on-chain memo == expected memo", memos[0] if memos else "no memo found") print(" solscan https://solscan.io/tx/" + sig) def main(): ap = argparse.ArgumentParser(description="Verify a Hermes Grows snapshot against its Solana Memo anchor.") pick = ap.add_mutually_exclusive_group(required=True) pick.add_argument("--latest", action="store_true", help="newest anchored snapshot") pick.add_argument("--day", type=int) pick.add_argument("--date", metavar="YYYY-MM-DD") pick.add_argument("--local", metavar="DIR", help="snapshot folder on disk") ap.add_argument("--base", default=BASE, help="URL of the published public/ folder (default %(default)s)") ap.add_argument("--repo", help="read from a GitHub mirror instead: owner/name (branch main)") ap.add_argument("--rpc", default=RPC, help="Solana RPC (default %(default)s)") ap.add_argument("--offline", action="store_true", help="with --local: skip the Solana check") a = ap.parse_args() if a.local: d = os.path.abspath(a.local) listing = [os.path.relpath(os.path.join(r, f), d).replace(os.sep, "/") for r, _, fs in os.walk(d) for f in fs if not f.startswith(".") and f != "manifest.json"] with open(os.path.join(d, "manifest.json"), "rb") as fh: m, root = verify_snapshot(fh.read(), lambda p: open(os.path.join(d, p), "rb").read(), listing) if re.fullmatch(r"\d{4}-\d{2}-\d{2}", os.path.basename(d)): check(m.get("date") == os.path.basename(d), "folder name == manifest date", m.get("date")) anchors = os.path.join(d, "..", "..", "anchors.json") hits = [x for x in json.load(open(anchors)) if x.get("date") == m.get("date")] if os.path.exists(anchors) else [] if hits and not a.offline: verify_anchor(hits[0], m, root, a.rpc) elif not hits: print(" anchor none on disk for this date (not anchored yet)") else: base = "https://raw.githubusercontent.com/%s/main/public/" % a.repo if a.repo else a.base anchors = json.loads(http(base + "anchors.json")) check(isinstance(anchors, list) and len(anchors) > 0, "anchors.json", "%d anchors %s" % (len(anchors), base)) anchor = max(anchors, key=lambda x: x["day"]) if a.latest and anchors else next( (x for x in anchors if x.get("day") == a.day or x.get("date") == a.date), None) if not anchor: check(False, "anchor for the requested day", "not found") sys.exit(1) folder = base + "snapshots/%s/" % anchor["date"] m, root = verify_snapshot(http(folder + "manifest.json"), lambda p: http(folder + urllib.parse.quote(p))) verify_anchor(anchor, m, root, a.rpc) if a.latest and json.loads(http(base + "latest.json")).get("day", 0) > anchor["day"]: print(" note a newer snapshot exists but is not anchored yet") print("VERIFIED" if not FAILED else "FAILED: %d check(s)" % len(FAILED)) sys.exit(1 if FAILED else 0) if __name__ == "__main__": try: main() except (OSError, ValueError) as err: # network, HTTP 404, malformed JSON sys.exit("\u2717 %s\nFAILED: could not read the snapshot (wrong --repo, nothing published yet, or offline?)" % err)