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

#!/usr/bin/env python3
"""Job vectors for OER/2 minor version 1 (extensions/oer-receipts.md §3.1).

A `job` receipt states one hop of a routed media job: the operation it ran, under
`oer.job.*`, and its figure, which minor version 1 lets it state as the unmetered marker
(a null `energy.joules_total`) or beside the acquisition class it was read with
(`energy.class`). This file holds receipts of both forms, valid and with one defect each,
and the same members under minor version 0, where §3 still refuses them.

Every `expected` and every `unmetered` below is written by hand from the rules of
§3.1 and §7. None is copied from a verifier's output. A receipt whose defect leaves the
seal intact is re-sealed (fresh receipt id, receipt hash and Ed25519 signature under the
fixed test seed below), so the declared reason is the only one it can produce.

Run: python3 generate_job_vectors.py   (writes job_vectors.json beside it)
"""
import copy, hashlib, json, os, sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import oer2_verify as V  # canonical form + hashing only; the fixtures pin those independently
from cryptography.hazmat.primitives.asymmetric import ed25519
from cryptography.hazmat.primitives import serialization

SEED = hashlib.sha256(b"oer job vectors: the hop key").digest()
PRIV = ed25519.Ed25519PrivateKey.from_private_bytes(SEED)
PUB = PRIV.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw).hex()
SIGNER = "ed25519:hop-01"
GENESIS = "sha256:" + "00" * 32

# A metered hop: an encode measured by the host's package counter.
METERED = {
    "oer": "2.1", "kind": "job", "capability": "oer.job.media.encode", "intent": "replicate",
    "subject": {"inputs": ["sha256:" + "11" * 32], "output": "blake3:" + "22" * 32},
    "work": {"status": 1, "outcome": "completed", "latency_s": "1.250000"},
    "energy": {"joules_total": "3.200000", "boundary": "the hop's host: CPU package and DRAM",
               "breakdown": {"compute_j": "3.000000", "movement_j": "0.200000", "idle_alloc_j": "0",
                             "shared_alloc_j": "0", "upstream_j": "0"},
               "method": "hardware_meter", "temperature_k": "300",
               "class": {"label": "OnChipCounter", "relative_ppm": 20000, "window_us": 1250000, "filtering": "Unknown"}},
    "provenance": {"how": "measured", "device": "hardware", "weakest_input": {"how": "measured", "input": "package counter"}},
    "floor": {"bits_committed": 0, "bits_delivered": "0", "floor_j": "0", "decades_above_floor": "0"},
    "hash_alg": "sha-256", "sig_alg": "ed25519",
    "chain": {"prev_hash": GENESIS, "receipt_hash": "", "signer": SIGNER, "signature": ""},
}

# An unmetered hop: no meter on the host, so the figure is the marker.
UNMETERED = copy.deepcopy(METERED)
UNMETERED["energy"] = {"joules_total": None, "boundary": "the hop's host: CPU package and DRAM",
                       "breakdown": {"compute_j": None, "movement_j": None, "idle_alloc_j": None,
                                     "shared_alloc_j": None, "upstream_j": None},
                       "method": "unmetered", "temperature_k": "300"}
UNMETERED["provenance"] = {"how": "projected", "device": "hardware", "weakest_input": {"how": "projected", "input": "no meter"}}
UNMETERED["subject"]["output"] = "blake3:" + "33" * 32


def seal(r, prev=GENESIS):
    """Recompute id, chained hash and signature so the receipt is otherwise valid."""
    r["chain"]["prev_hash"] = prev
    body = V.canonical_bytes(r)
    r["receipt_id"] = V.h(r["hash_alg"], body)
    r["chain"]["receipt_hash"] = V.h(r["hash_alg"], r["chain"]["prev_hash"].encode() + body)
    r["chain"]["signature"] = PRIV.sign(r["chain"]["receipt_hash"].encode()).hex()
    return dict(sorted(r.items(), key=lambda kv: V.FIELD_ORDER[""].index(kv[0])))


def edit(src, fn, sealed=True):
    r = copy.deepcopy(src)
    fn(r)
    return seal(r) if sealed else r


def cls(**kw):
    def f(r):
        r["energy"]["class"] = kw
    return f


def drop_class(r):
    r["energy"].pop("class")


M = seal(copy.deepcopy(METERED))
U = seal(copy.deepcopy(UNMETERED))
vectors = []


def vec(name, receipts, expected, unmetered, **extra):
    """`unmetered` names, by hand, the receipts this vector states the marker in: True for
    the vector's one receipt, or a list of receipt ids."""
    if unmetered is True:
        unmetered = [receipts[0]["receipt_id"]]
    v = {"name": name, "receipts": receipts, "expected": expected, "unmetered": unmetered}
    v.update(extra)
    vectors.append(v)


# --- valid ---
vec("j01 a metered hop with its counter class", [M], [], [])
vec("j02 an unmetered hop states the marker", [U], [], [U["receipt_id"]])
U2 = seal(copy.deepcopy(UNMETERED), prev=M["chain"]["receipt_hash"])
vec("j03 a failover chain: a metered hop, then an unmetered one", [M, U2], [], [U2["receipt_id"]])
vec("j04 a model-based figure graded derived",
    [edit(METERED, lambda r: (cls(label="ModelBased")(r), r["provenance"].update(how="derived"),
                              r["energy"].update(method="model_estimated")))], [], [])
vec("j05 an estimator's figure graded projected",
    [edit(METERED, lambda r: (cls(label="Estimator")(r), r["provenance"].update(how="projected")))], [], [])
vec("j06 a calibrated instrument's figure graded measured",
    [edit(METERED, cls(label="CalibratedInstrument", relative_ppm=5000, window_us=1250000,
                       calibration_ref="urn:example:calibration:0001"))], [], [])
vec("j07 a metered hop with no class", [edit(METERED, drop_class)], [], [])
vec("j08 the largest relative uncertainty a class states",
    [edit(METERED, cls(label="OnChipCounter", relative_ppm=2 ** 32 - 1, window_us=1, filtering="On"))], [], [])

# --- minor version 0 keeps §3's members ---
vec("j09 a class under minor version 0", [edit(M, lambda r: r.update(oer="2.0"), sealed=False)], ["canonical_failure"], [])
vec("j10 the marker under minor version 0", [edit(UNMETERED, lambda r: r.update(oer="2.0"))], ["malformed"], [])
vec("j11 a class under a minor version this text does not define", [edit(M, lambda r: r.update(oer="2.10"), sealed=False)],
    ["canonical_failure"], [])

# --- the unmetered marker's rules ---
vec("j12 the marker beside a breakdown term",
    [edit(UNMETERED, lambda r: r["energy"]["breakdown"].update(compute_j="1.0"))], ["unmetered_invalid"], True)
vec("j13 the marker graded measured", [edit(UNMETERED, lambda r: r["provenance"].update(how="measured"))],
    ["unmetered_invalid"], True)
vec("j14 the marker beside a class",
    [edit(UNMETERED, cls(label="Estimator"))], ["unmetered_invalid"], True)


def unmetered_bundle(r):
    r["kind"] = "bundle"
    r["capability"] = "oer.bundle.job"
    stages = [{"kind": "job", "receipt_hash": U["chain"]["receipt_hash"], "joules": "0"}]
    r["bundle"] = {"stages": stages, "provenance_root": V.h("sha-256", V.stage_list_bytes(stages))}
    r["efficiency"] = {"completed_units": 1}


vec("j15 an unmetered bundle", [edit(UNMETERED, unmetered_bundle)], ["unmetered_invalid"], True)
vec("j16 a ledger over a chain that holds the marker", [M, U2], ["ledger_mismatch"], [U2["receipt_id"]],
    ledger={"meter_joules": "3.200000", "unallocated_joules": "0"})

# --- the class's rules ---
vec("j17 the legacy hardware label", [edit(METERED, cls(label="HwShunt"))], ["class_invalid"], [])
vec("j18 a counter class with no uncertainty", [edit(METERED, cls(label="OnChipCounter", filtering="Off"))],
    ["class_invalid"], [])
vec("j19 a counter class with no filtering state",
    [edit(METERED, cls(label="OnChipCounter", relative_ppm=20000, window_us=1250000))], ["class_invalid"], [])
vec("j20 a filtering state outside the three",
    [edit(METERED, cls(label="OnChipCounter", relative_ppm=20000, window_us=1250000, filtering="Maybe"))],
    ["class_invalid"], [])
vec("j21 a relative uncertainty of zero",
    [edit(METERED, cls(label="OnChipCounter", relative_ppm=0, window_us=1250000, filtering="Off"))], ["class_invalid"], [])
vec("j22 a window of zero",
    [edit(METERED, cls(label="OnChipCounter", relative_ppm=20000, window_us=0, filtering="Off"))], ["class_invalid"], [])
vec("j23 a relative uncertainty past 2^32 - 1",
    [edit(METERED, cls(label="OnChipCounter", relative_ppm=2 ** 32, window_us=1250000, filtering="Off"))],
    ["class_invalid"], [])
vec("j24 a relative uncertainty written as a string",
    [edit(METERED, cls(label="OnChipCounter", relative_ppm="20000", window_us=1250000, filtering="Off"))],
    ["class_invalid"], [])
vec("j25 a relative uncertainty written as true",
    [edit(METERED, cls(label="OnChipCounter", relative_ppm=True, window_us=1250000, filtering="Off"))],
    ["class_invalid"], [])
vec("j26 a model's figure with evidence beside it",
    [edit(METERED, lambda r: (cls(label="ModelBased", relative_ppm=20000)(r), r["provenance"].update(how="derived")))],
    ["class_invalid"], [])
vec("j27 a calibrated instrument with an empty certificate reference",
    [edit(METERED, cls(label="CalibratedInstrument", relative_ppm=5000, window_us=1250000, calibration_ref=""))],
    ["class_invalid"], [])
vec("j28 a calibrated instrument with a filtering state",
    [edit(METERED, cls(label="CalibratedInstrument", relative_ppm=5000, window_us=1250000, filtering="Off"))],
    ["class_invalid"], [])
vec("j29 a class that is a string", [edit(METERED, lambda r: r["energy"].update({"class": "OnChipCounter"}))],
    ["class_invalid"], [])
vec("j30 a label outside the vocabulary", [edit(METERED, cls(label="Guess"))], ["class_invalid"], [])
vec("j31 a member the class does not list",
    [edit(M, cls(label="Estimator", source="nameplate"), sealed=False)], ["canonical_failure"], [])

# --- the class held to the grade (§2's correspondence) ---
vec("j32 a counter reading graded derived", [edit(METERED, lambda r: r["provenance"].update(how="derived"))],
    ["class_grade_mismatch"], [])
vec("j33 an estimator's figure graded measured", [edit(METERED, cls(label="Estimator"))],
    ["class_grade_mismatch"], [])
vec("j34 a model's figure graded simulated",
    [edit(METERED, lambda r: (cls(label="ModelBased")(r), r["provenance"].update(how="simulated")))],
    ["class_grade_mismatch"], [])
vec("j35 a calibrated instrument's figure graded projected",
    [edit(METERED, lambda r: (cls(label="CalibratedInstrument", relative_ppm=5000, window_us=1250000,
                                  calibration_ref="urn:example:calibration:0001")(r),
                              r["provenance"].update(how="projected")))],
    ["class_grade_mismatch"], [])
vec("j36 a class beside a grade outside the vocabulary", [edit(METERED, lambda r: r["provenance"].update(how="guessed"))],
    ["grade_unknown"], [])

# A tampered figure after sealing is caught by the seal, as in minor version 0.
vec("j37 an unmetered receipt given a figure after it was sealed",
    [edit(U, lambda r: (r["energy"].update(joules_total="0"), r["energy"]["breakdown"].update(
        compute_j="0", movement_j="0", idle_alloc_j="0", shared_alloc_j="0", upstream_j="0")), sealed=False)],
    ["receipt_hash_mismatch", "receipt_id_mismatch"], [])

# A failed hop is completed work of no use: status -1, and here no meter either.
vec("j38 a failed hop, unmetered", [edit(UNMETERED, lambda r: r["work"].update(status=-1, outcome="failed"))], [], True)

doc = {"oer": "2.1", "draft": "0.1",
       "source": "job receipts of minor version 1: the unmetered marker and the acquisition class (oer-receipts §3.1)",
       "keys": {SIGNER: PUB}, "vectors": vectors}
with open(os.path.join(HERE, "job_vectors.json"), "w", encoding="utf-8") as f:
    json.dump(doc, f, indent=1, ensure_ascii=False)
    f.write("\n")
print("wrote job_vectors.json: %d vectors" % len(vectors))