open-standards/wai/oer-conformance/generate_vectors.py

import json, hashlib, copy, sys, os

# Resolve everything against this script's own directory, so the README's
# instructions work wherever the pack is checked out. Previously hardcoded to
# /mnt/user-data/outputs/oer2, the sandbox it was first generated in.
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import oer2_verify as V
from cryptography.hazmat.primitives.asymmetric import ed25519
from cryptography.hazmat.primitives import serialization
from decimal import Decimal

data = json.load(open(os.path.join(HERE, "sw_export.json")))
seed = hashlib.sha256(b"settlement world node ledger vector key").digest()
priv = ed25519.Ed25519PrivateKey.from_private_bytes(seed)
pub = priv.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw).hex()
SIGNER = "ed25519:corridor-ledger-01"
def D(x): return "%.6e" % float(x)
RULES = "sha256:" + hashlib.sha256(b"rules@1").hexdigest()
exported = data["log"]; records = data["records"]
prior_sum = 0.0

def seal(r, prev):
    r = copy.deepcopy(r); r.pop("receipt_id", None); r["chain"] = {"prev_hash": prev, "receipt_hash": "", "signer": SIGNER, "signature": ""}
    body = V.canonical_bytes(r)
    r["receipt_id"] = V.h("sha-256", body)
    r["chain"]["receipt_hash"] = V.h("sha-256", prev.encode() + body)
    r["chain"]["signature"] = priv.sign(r["chain"]["receipt_hash"].encode()).hex()
    return r

def convert(b):
    served = b["settlement"]["served_from_record"]
    s = b["settlement"]; rec = records.get(s["record_id"]) if s.get("record_id") else None
    r = {"oer": "2.0", "kind": "decision",
         "capability": "oer.decision." + b["workload"]["name"], "intent": "replicate" if served else "create",
         "subject": {"inputs": ["sha256:" + s["question_key"], RULES], "output": b["workload"]["output"] if b["workload"]["output"] is not None else ""},
         "work": {"status": b["status"], "outcome": b["outcome"], "latency_s": D(b["workload"]["latency_s"])},
         "energy": {"joules_total": D(b["energy"]["joules_total"]), "boundary": b["energy"]["boundary"],
                    "breakdown": {k: D(v) for k, v in b["energy"]["breakdown"].items()}, "method": b["energy"]["method"], "temperature_k": D(data["fl_temp"])},
         "provenance": {"how": b["provenance"]["how"], "device": b["provenance"]["device"], "weakest_input": {"how": b["provenance"]["weakest_input"]["how"], "input": str(b["provenance"]["weakest_input"]["input"])}},
         "settlement": {"question_key": "sha256:" + s["question_key"], "served_from_record": served, "record_id": s["record_id"] or "", "record_hash": ("sha256:" + s["record_hash"]) if s["record_hash"] else "", "reuse_count": s["reuse_count"], "derivation_j": D(s["derivation_j"]), "valid_until": int(rec["valid_until"]) if rec else 0},
         "floor": {"bits_committed": b["floor"]["bits_destroyed"], "bits_delivered": D(b["floor"]["bits_resolved"]), "floor_j": D(b["floor"]["floor_j"]), "decades_above_floor": D(b["floor"]["decades_above_floor"] or 0)},
         "hash_alg": "sha-256", "sig_alg": "ed25519"}
    # total must equal the breakdown exactly in decimal strings
    r["energy"]["joules_total"] = str(sum(Decimal(v) for v in r["energy"]["breakdown"].values()))
    return r

chain, prev = [], "sha256:" + "0" * 64
for b in exported:
    r = seal(convert(b["body"]), prev); chain.append(r); prev = r["chain"]["receipt_hash"]
exported_sum = sum(Decimal(r["energy"]["joules_total"]) for r in chain)
carried = Decimal(repr(data["receiptJ"])) - sum(Decimal(repr(b["body"]["energy"]["joules_total"])) for b in exported)
ledger = {"meter_joules": D(data["meter"]), "unallocated_joules": D(data["pending"]), "carried_forward_joules": str(Decimal(D(data["meter"])) - exported_sum - Decimal(D(data["pending"])))}
records_v = {rid: rec for rid, rec in records.items()}

stages = [{"kind": r["kind"], "receipt_hash": r["chain"]["receipt_hash"], "joules": r["energy"]["joules_total"]} for r in chain]
root = V.h("sha-256", V.stage_list_bytes(stages))
completed = sum(1 for r in chain if r["work"]["status"] == 1); wasted = len(chain) - completed
wasted_j = sum(Decimal(r["energy"]["joules_total"]) for r in chain if r["work"]["status"] != 1)
bundle = {"oer": "2.0", "kind": "bundle", "capability": "oer.bundle.hour", "intent": "replicate",
          "subject": {"inputs": [r["chain"]["receipt_hash"] for r in chain], "output": root},
          "work": {"status": 1, "outcome": "ledger", "latency_s": "0.000000e+00"},
          "energy": {"joules_total": str(exported_sum), "boundary": "sum of the hour's decision receipts, refusals included", "breakdown": {"compute_j": str(exported_sum), "movement_j": "0.000000e+00", "idle_alloc_j": "0.000000e+00", "shared_alloc_j": "0.000000e+00", "upstream_j": "0.000000e+00"}, "method": "model_estimated", "temperature_k": D(data["fl_temp"])},
          "provenance": {"how": "projected", "device": "simulation", "weakest_input": {"how": "projected", "input": "wire_w"}},
          "efficiency": {"completed_units": completed, "wasted_units": wasted, "joules_per_completed": D(exported_sum / completed), "waste_fraction": D(wasted_j / exported_sum), "value_per_joule": None},
          "bundle": {"stages": stages, "provenance_root": root},
          "floor": {"bits_committed": data["fl_bits"] * len(chain), "bits_delivered": D(1.585 * len(chain)), "floor_j": D(data["fl_bits"] * len(chain) * 1.380649e-23 * data["fl_temp"] * 0.6931471805599453), "decades_above_floor": D(0)},
          "hash_alg": "sha-256", "sig_alg": "ed25519"}
bundle = seal(bundle, prev)
anchor = seal({"oer": "2.0", "kind": "anchor", "capability": "oer.anchor.iridium_imt", "intent": "replicate",
          "subject": {"inputs": [bundle["chain"]["receipt_hash"]], "output": root},
          "work": {"status": 1, "outcome": "delivered", "latency_s": "1.000000e+01"},
          "energy": {"joules_total": "2.560000e-01", "boundary": "gateway satellite modem, one message", "breakdown": {"compute_j": "0.000000e+00", "movement_j": "2.560000e-01", "idle_alloc_j": "0.000000e+00", "shared_alloc_j": "0.000000e+00", "upstream_j": "0.000000e+00"}, "method": "model_estimated", "temperature_k": D(data["fl_temp"])},
          "provenance": {"how": "measured", "device": "simulation", "weakest_input": {"how": "projected", "input": "iri_wake_s"}},
          "floor": {"bits_committed": 512, "bits_delivered": "2.560000e+02", "floor_j": D(512 * 1.380649e-23 * data["fl_temp"] * 0.6931471805599453), "decades_above_floor": "1.500000e+01"},
          "hash_alg": "sha-256", "sig_alg": "ed25519"}, bundle["chain"]["receipt_hash"])

vectors = []
vectors.append({"name": "v01 valid chain, records and ledger", "receipts": chain, "records": records_v, "ledger": ledger, "expected": []})
t = copy.deepcopy(records_v); cited = next(r["settlement"]["record_id"] for r in chain if r["settlement"]["served_from_record"]); t[cited]["answer"] += "_altered"
vectors.append({"name": "v02 a served record altered after the fact", "receipts": chain, "records": t, "expected": ["record_mismatch"]})
vectors.append({"name": "v03 valid bundle and anchor", "receipts": chain + [bundle, anchor], "expected": []})
tb = copy.deepcopy(bundle); tb["bundle"]["stages"] = tb["bundle"]["stages"][::-1]; tb = seal(tb, prev)
vectors.append({"name": "v04 bundle stages reordered", "receipts": chain + [tb], "expected": ["bundle_root_mismatch"]})
tm = copy.deepcopy(chain); tm[5]["oer"] = "3.0"
vectors.append({"name": "v05 unknown major version", "receipts": tm, "expected": ["unknown_major"]})
ts = copy.deepcopy(chain); x = ts[7]; x["energy"]["breakdown"]["compute_j"] = D(Decimal(x["energy"]["breakdown"]["compute_j"]) * 2); ts[7] = seal(x, ts[6]["chain"]["receipt_hash"]); ts[8] = seal(ts[8], ts[7]["chain"]["receipt_hash"])
vectors.append({"name": "v06 breakdown edited without the total, re-signed", "receipts": ts[:9], "expected": ["sum_mismatch"]})
tg = copy.deepcopy(chain); sig = bytearray(bytes.fromhex(tg[3]["chain"]["signature"])); sig[0] ^= 1; tg[3]["chain"]["signature"] = sig.hex()
vectors.append({"name": "v07 signature altered", "receipts": tg, "expected": ["signature_invalid"]})
tl = copy.deepcopy(ledger); tl["meter_joules"] = D(Decimal(tl["meter_joules"]) * Decimal("0.9"))
vectors.append({"name": "v08 meter figure does not match the receipts", "receipts": chain, "ledger": tl, "expected": ["ledger_mismatch"]})
tc = copy.deepcopy(chain); tc[10]["chain"]["prev_hash"] = "sha256:" + "f" * 64
vectors.append({"name": "v09 chain link broken", "receipts": tc, "expected": ["chain_break", "receipt_hash_mismatch"]})

out = {"oer": "2.0", "draft": "0.1", "source": "Settlement World v0.11, gatewayless corridor, one simulated day, last 150 receipts", "keys": {SIGNER: pub}, "vectors": vectors}
json.dump(out, open(os.path.join(HERE, "vectors.json"), "w"), indent=0)
print("receipts", len(chain), "bundle stages", len(stages), "completed", completed, "wasted", wasted, "hour joules", float(exported_sum))