File size: 9,073 Bytes
21b0475 ad19be1 21b0475 ad19be1 21b0475 ad19be1 21b0475 e1b4e6e ad19be1 21b0475 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | """verify - check this install against the release's published hashes, and those hashes against their Solana anchor.
RELEASE_HASHES.json (beside oracle7.py) lists one SHA-256 per shipped part - every specialist directory, the admitted fold
chain, this workbench, the launcher, the store manifest - and a Merkle root over them. verify_install() recomputes every
part from the files on this computer with the same recipe and rebuilds the root. From 10.2 (recipe v4) the small data
read beside the stores - tables, sidecars, nltk_data, textbooks, vendor, VARIANT - is a part too, so nothing the release
reads is outside the root except what the store manifest lists by SHA-256. A changed, missing or extra specialist
shows by name. Stores are checked by oracle7.py --setup against PACK_MANIFEST.json, whose own hash is a part here.
check_anchor() is optional and the only step that uses the network: it asks a public Solana RPC (read-only) for the
anchor transaction named in RELEASE_HASHES.json and confirms its memo carries the same root, so the root you computed is
the one Prometheus7 committed to on-chain at release time."""
from __future__ import annotations
import hashlib
import json
import urllib.request
from pathlib import Path
HERE = Path(__file__).resolve().parent
INSTALL = HERE.parent
SKIP_DIRS = {"__pycache__", "state", "local"}
RPC = "https://api.mainnet-beta.solana.com"
def part_hash(base):
base = Path(base)
h = hashlib.sha256()
files = sorted((p for p in base.rglob("*") if p.is_file() and not SKIP_DIRS & set(p.relative_to(base).parts) and p.suffix != ".pyc"),
key=lambda p: p.relative_to(base).as_posix().encode())
n = 0
for p in files:
rel = p.relative_to(base).as_posix()
if rel.endswith(".zst"):
continue
h.update(rel.encode() + b"\0")
h.update(hashlib.sha256(p.read_bytes()).digest())
n += 1
return h.hexdigest(), n
def routes_hash(relstore):
"""the relstore's route files (its top-level .json files: router patterns, property aliases), same recipe as part_hash"""
h = hashlib.sha256()
files = sorted((p for p in Path(relstore).glob("*.json") if p.is_file()), key=lambda p: p.name.encode())
for p in files:
h.update(p.name.encode() + b"\0")
h.update(hashlib.sha256(p.read_bytes()).digest())
return h.hexdigest(), len(files)
def _files_hash(base, files):
"""part_hash's recipe over a given list of files under base"""
base = Path(base)
h = hashlib.sha256()
files = sorted(files, key=lambda p: p.relative_to(base).as_posix().encode())
for p in files:
h.update(p.relative_to(base).as_posix().encode() + b"\0")
h.update(hashlib.sha256(p.read_bytes()).digest())
return h.hexdigest(), len(files)
def data_parts(R, manifest):
"""recipe v4 (10.2): the small data read beside the stores. A file the store manifest lists is checked by its own
SHA-256 there, so it is left out here; runtime state is never part of a release."""
R = Path(R)
listed = {f["path"] for f in manifest.get("files", [])}
def rel(p):
return "root/" + p.relative_to(R).as_posix()
def keep(p):
return p.is_file() and rel(p) not in listed and p.suffix not in (".pyc", ".zst") and not SKIP_DIRS & set(p.relative_to(R).parts)
out = {}
for d in ("nltk_data", "textbooks", "vendor"):
if (R / d).is_dir():
out["data/" + d] = _files_hash(R / d, [p for p in (R / d).rglob("*") if keep(p)])
out["data/relstore-tables"] = _files_hash(R / "relstore", [p for p in (R / "relstore").glob("*") if keep(p) and p.suffix != ".json"])
m = R / "fable-content/oracle-m-series-2026-08-15"
out["data/fable-content"] = _files_hash(m, [p for p in m.rglob("*") if keep(p) and p.relative_to(m).parts[0] != "self_loop"])
if (R / "VARIANT").is_file():
out["data/variant"] = _files_hash(R, [R / "VARIANT"])
return out
def leaf(name, hexd):
return hashlib.sha256(("oracle7-part:%s:%s" % (name, hexd)).encode()).digest()
def merkle(leaves):
level = list(leaves)
if not level:
return hashlib.sha256(b"").digest()
while len(level) > 1:
nxt = [hashlib.sha256(level[i] + level[i + 1]).digest() for i in range(0, len(level) - 1, 2)]
if len(level) % 2:
nxt.append(level[-1])
level = nxt
return level[0]
def published():
p = INSTALL / "RELEASE_HASHES.json"
if not p.exists():
raise ValueError("RELEASE_HASHES.json is not beside oracle7.py, so there is nothing to verify against")
return json.loads(p.read_text(encoding="utf-8"))
def verify_install(root=None):
import os
pub = published()
R = Path(root or os.environ.get("ORACLE7_ROOT") or (INSTALL / "root"))
parts = {}
for d in sorted((R / "specialists").iterdir()):
if d.is_dir() and d.name not in SKIP_DIRS:
parts["specialists/" + d.name] = part_hash(d)[0]
parts["core/admitted"] = part_hash(R / "fable-content/oracle-m-series-2026-08-15/self_loop/admitted")[0]
parts["workbench"] = part_hash(HERE)[0]
if "core/wander-train" in pub["parts"]: # the second recipe (a later release): the core's reader modules
parts["core/wander-train"] = part_hash(R / "wander-train")[0]
if "core/relstore-routes" in pub["parts"]: # ... and the relstore's route files
parts["core/relstore-routes"] = routes_hash(R / "relstore")[0]
# recipe v3 (10.0): the knowledge the 5.1-10.0 organs read is a part per folder; what a user makes (their workspace,
# lessons, ledgers, packets, places, scratch) is never part of a release
for k in sorted(pub["parts"]):
if k.startswith("organs10/") and (R / k).is_dir():
parts[k] = part_hash(R / k)[0]
for f in ("oracle7.py", "PACK_MANIFEST.json"):
if (INSTALL / f).exists():
parts[f] = hashlib.sha256((INSTALL / f).read_bytes()).hexdigest()
for top in ("orbiters", "records"): # recipe v5 (10.5): each orbiter, and the records the changelog points to
for d in sorted(p for p in (INSTALL / top).glob("*") if p.is_dir() and p.name not in SKIP_DIRS):
if top + "/" + d.name in pub["parts"] or any(k.startswith(top + "/") for k in pub["parts"]):
parts[top + "/" + d.name] = part_hash(d)[0]
if any(k.startswith("data/") for k in pub["parts"]) and (INSTALL / "PACK_MANIFEST.json").exists(): # recipe v4 (10.2)
manifest = json.loads((INSTALL / "PACK_MANIFEST.json").read_text(encoding="utf-8"))
parts.update({k: v[0] for k, v in data_parts(R, manifest).items()})
want = {k: v["sha256"] for k, v in pub["parts"].items()}
changed = sorted(k for k in want if k in parts and parts[k] != want[k])
missing = sorted(k for k in want if k not in parts)
extra = sorted(k for k in parts if k not in want)
# the root is rebuilt from the PUBLISHED part list, substituting what this computer holds
mine = {k: parts.get(k, want[k] if k in missing else parts.get(k)) for k in want}
root = merkle([leaf(k, mine[k]) for k in sorted(want)]).hex()
ok_parts = len(want) - len(changed) - len(missing)
return {"release": pub.get("release"), "head": pub.get("head"), "published_root": pub["merkle_root"], "computed_root": root,
"root_matches": root == pub["merkle_root"] and not missing, "parts": len(want), "verified": ok_parts,
"specialists_verified": sum(1 for k in want if k.startswith("specialists/") and k not in changed and k not in missing),
"changed": changed, "missing": missing, "extra_local": extra, "anchor": pub.get("anchor")}
def check_anchor(signature=None, rpc=RPC):
pub = published()
sig = signature or (pub.get("anchor") or {}).get("signature")
if not sig:
raise ValueError("this release's anchor transaction is not recorded yet")
body = {"jsonrpc": "2.0", "id": 1, "method": "getTransaction",
"params": [sig, {"encoding": "jsonParsed", "maxSupportedTransactionVersion": 1, "commitment": "finalized"}]}
req = urllib.request.Request(rpc, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=20) as r:
tx = json.loads(r.read()).get("result")
if not tx:
raise ValueError("the RPC has no finalized transaction %s" % sig)
memos = []
for ins in tx["transaction"]["message"]["instructions"]:
if ins.get("program") == "spl-memo" or ins.get("programId") == (pub.get("anchor") or {}).get("program"):
memos.append(ins.get("parsed") if isinstance(ins.get("parsed"), str) else json.dumps(ins.get("parsed")))
logs = " ".join(tx.get("meta", {}).get("logMessages") or [])
found = any(pub["merkle_root"] in (m or "") for m in memos) or pub["merkle_root"] in logs
return {"signature": sig, "slot": tx.get("slot"), "block_time": tx.get("blockTime"), "memos": memos,
"root_on_chain": found, "explorer": "https://explorer.solana.com/tx/%s" % sig}
|