45ae4d5dae
A tariff and its invoices of version 2 carry job prices (a capability, the unit jobs or subpixels, a divisor and a price) and bill a job source: a route record's verification document with its hops' OER/2 receipts and a decode receipt of its output with its codestream. A job is billed at its serving capability's prices from its billed hop alone; its energy is the serving hop's figure in the scope jobs, by class, the unmetered marker apart. Record kinds 10 to 12, scope kind 5, line source kind 5, reasons output_invalid and job_figure_invalid; one hop and one output belong to one job, and two records of one router for one job overlap. Version 1 reads and bills as before; the control of an invoice of version 2 now gives version 3. The corpus bills a month of jobs from the route corpus beside a package, with a control for each new rule; receipt_ledger_verify.py verifies each job with route_record_verify.py and its output with _int_motion_receipt.py. The route corpus binds each hop receipt to its job and adds a hop served unmetered and one whose figure has no class. oer::micro_joules reads an OER/2 figure as whole microjoules, a half to the even one. Co-Authored-By: Claude Opus 5.5 <[email protected]>
225 lines
9.4 KiB
Python
225 lines
9.4 KiB
Python
"""The `wai.video.int_motion` decode receipt (extensions/integer-payloads.md §15.1, §15.2),
|
||
read and checked as the text states it.
|
||
|
||
Written from the extension's text; it shares no code with wai-rs. BLAKE3 is
|
||
../claim-conformance/_blake3.py, Ed25519 WAI's strict rule ../claim-conformance/_ed25519.py, and
|
||
the `WIV1` decode the integer-payload corpus's second implementation,
|
||
../int-payload-conformance/int_payload_verify.py. Standard library only.
|
||
|
||
read(text) -> the receipt, or raise Bad(why) (§15.2)
|
||
consistent(r) -> None, or why it is not (§15.1)
|
||
rederived(r, p) -> None, or why the payload does not give it (§15.1)
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import struct
|
||
import sys
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
||
sys.path.insert(0, os.path.join(HERE, "..", "int-payload-conformance"))
|
||
import _ed25519 as ED # noqa: E402
|
||
from _blake3 import blake3 # noqa: E402
|
||
import int_payload_verify as IPV # noqa: E402
|
||
|
||
PROFILE = "wai.video.int_motion/receipt"
|
||
LEAF, NODE = b"moq-jwp:merkle-leaf\x01", b"moq-jwp:merkle-node\x01" # jwp-receipts §2.2
|
||
U32, U64 = 2 ** 32 - 1, 2 ** 64 - 1
|
||
RESOLUTION_FLOOR = 288676 # §15.1
|
||
TAGS = {"HwShunt": 1, "ModelBased": 2, "Estimator": 3, "OnChipCounter": 4, "CalibratedInstrument": 5}
|
||
FILTERING = {"Unknown": 0, "Off": 1, "On": 2}
|
||
# The Unicode White_Space characters: a calibration reference of these alone is empty.
|
||
WHITE_SPACE = set("\t\n\x0b\x0c\r \x85\xa0 "
|
||
"
")
|
||
MEMBERS = ["profile", "codestream_hash", "clip_hash", "h", "w", "frames", "merkle_root", "work_pixel_ops", "joules_micro",
|
||
"uj_per_kilopixel_op", "energy_provenance", "parent_receipt_hash", "signer_pubkey", "signer_id", "sig", "receipt_hash"]
|
||
EVIDENCE = {"OnChipCounter": ["energy_uncertainty", "energy_counter_filtering"],
|
||
"CalibratedInstrument": ["energy_uncertainty", "energy_calibration_ref"]}
|
||
|
||
|
||
class Bad(Exception):
|
||
pass
|
||
|
||
|
||
class Float:
|
||
"""A JSON number with a fraction or an exponent, kept as it is spelled."""
|
||
def __init__(self, text):
|
||
self.text = text
|
||
|
||
|
||
def _pairs(pairs):
|
||
d = {}
|
||
for k, v in pairs:
|
||
if k in d:
|
||
raise Bad("a member named twice")
|
||
d[k] = v
|
||
return d
|
||
|
||
|
||
def _hex(v, n):
|
||
if not isinstance(v, str) or len(v) != 2 * n or any(c not in "0123456789abcdef" for c in v):
|
||
raise Bad("not %d bytes of lowercase hexadecimal" % n)
|
||
return bytes.fromhex(v)
|
||
|
||
|
||
def _uint(v, top):
|
||
if isinstance(v, bool) or not isinstance(v, int) or not 0 <= v <= top:
|
||
raise Bad("not an integer in plain decimal")
|
||
return v
|
||
|
||
|
||
def _class(o):
|
||
"""energy-measurement §2.7's reading of the class and its evidence: (label, ppm, window,
|
||
filtering, calibration_ref), or None when unmetered."""
|
||
label = o.get("energy_provenance")
|
||
evidence = [k for k in ("energy_uncertainty", "energy_counter_filtering", "energy_calibration_ref") if o.get(k) is not None]
|
||
if label is None:
|
||
if evidence:
|
||
raise Bad("evidence with no class")
|
||
return None
|
||
if not isinstance(label, str) or label not in TAGS:
|
||
raise Bad("a label outside the vocabulary")
|
||
if label not in EVIDENCE:
|
||
if evidence:
|
||
raise Bad("evidence beside a class that carries none")
|
||
return (label, None, None, None, None)
|
||
u = o.get("energy_uncertainty")
|
||
if not isinstance(u, dict):
|
||
raise Bad("a refined class without its uncertainty")
|
||
ppm, win = u.get("relative_ppm"), u.get("window_us")
|
||
if isinstance(ppm, bool) or not isinstance(ppm, int) or ppm < 0 or isinstance(win, bool) or not isinstance(win, int) or win < 0:
|
||
raise Bad("an uncertainty that is not two integers")
|
||
if ppm > U32 or ppm == 0 or win == 0 or win > U64:
|
||
raise Bad("an uncertainty of zero, of no window, or out of range")
|
||
if label == "OnChipCounter":
|
||
if o.get("energy_calibration_ref") is not None:
|
||
raise Bad("a certificate beside a counter")
|
||
f = o.get("energy_counter_filtering")
|
||
f = "Unknown" if f is None else f
|
||
if not isinstance(f, str) or f not in FILTERING:
|
||
raise Bad("a filtering state outside the three")
|
||
return (label, ppm, win, f, None)
|
||
if o.get("energy_counter_filtering") is not None:
|
||
raise Bad("a filtering state beside an instrument")
|
||
ref = o.get("energy_calibration_ref")
|
||
if not isinstance(ref, str) or all(c in WHITE_SPACE for c in ref):
|
||
raise Bad("an instrument with no certificate")
|
||
return (label, ppm, win, None, ref)
|
||
|
||
|
||
def read(text):
|
||
"""§15.2: the receipt a text is the JSON form of."""
|
||
try:
|
||
o = json.loads(text, object_pairs_hook=_pairs, parse_float=Float, parse_constant=lambda s: Float(s))
|
||
except (ValueError, RecursionError) as e:
|
||
raise Bad("not JSON read strictly: %s" % e)
|
||
if not isinstance(o, dict) or o.get("profile") != PROFILE:
|
||
raise Bad("profile")
|
||
cls = _class(o)
|
||
want = MEMBERS + (EVIDENCE.get(cls[0], []) if cls else [])
|
||
if set(o) != set(want) or len(o) != len(want):
|
||
raise Bad("members")
|
||
if not isinstance(o["frames"], list) or not isinstance(o["signer_id"], str):
|
||
raise Bad("frames or signer id")
|
||
r = {
|
||
"codestream_hash": _hex(o["codestream_hash"], 32), "clip_hash": _hex(o["clip_hash"], 32),
|
||
"h": _uint(o["h"], U32), "w": _uint(o["w"], U32), "frames": [_hex(f, 32) for f in o["frames"]],
|
||
"merkle_root": _hex(o["merkle_root"], 32), "work": _uint(o["work_pixel_ops"], U64), "joules": _uint(o["joules_micro"], U64),
|
||
"class": cls, "parent": None if o["parent_receipt_hash"] is None else _hex(o["parent_receipt_hash"], 32),
|
||
"signer": _hex(o["signer_pubkey"], 32), "signer_id": o["signer_id"], "sig": _hex(o["sig"], 64),
|
||
}
|
||
rate = o["uj_per_kilopixel_op"]
|
||
want_rate = None
|
||
if cls is not None and r["joules"] > 0 and r["work"] > 0:
|
||
want_rate = "%.6f" % (float(r["joules"]) * 1000.0 / float(r["work"]))
|
||
got_rate = rate.text if isinstance(rate, Float) else (None if rate is None else json.dumps(rate))
|
||
if got_rate != want_rate or (rate is not None and not isinstance(rate, Float)):
|
||
raise Bad("uj_per_kilopixel_op is not the receipt's")
|
||
if _hex(o["receipt_hash"], 32) != receipt_hash(r):
|
||
raise Bad("receipt_hash is not the receipt's")
|
||
return r
|
||
|
||
|
||
def class_bytes(cls):
|
||
"""§15.1: the tag byte and its evidence."""
|
||
if cls is None:
|
||
return b"\x00"
|
||
label, ppm, win, filt, ref = cls
|
||
out = bytes([TAGS[label]])
|
||
if label == "OnChipCounter":
|
||
out += struct.pack(">IQ", ppm, win) + bytes([FILTERING[filt]])
|
||
elif label == "CalibratedInstrument":
|
||
ref = ref.encode("utf-8")
|
||
out += struct.pack(">IQI", ppm, win, len(ref)) + ref
|
||
return out
|
||
|
||
|
||
def payload(r):
|
||
"""§15.1's signing payload."""
|
||
o = b"wai:video-receipt\x01" + r["codestream_hash"] + r["clip_hash"] + struct.pack(">II", r["h"], r["w"])
|
||
o += r["merkle_root"] + struct.pack(">QQQ", len(r["frames"]), r["work"], r["joules"]) + class_bytes(r["class"])
|
||
o += b"\x00" if r["parent"] is None else b"\x01" + r["parent"]
|
||
for i, f in enumerate(r["frames"]):
|
||
o += struct.pack(">Q", i) + f
|
||
sid = r["signer_id"].encode("utf-8")
|
||
return o + r["signer"] + struct.pack(">I", len(sid)) + sid
|
||
|
||
|
||
def receipt_hash(r):
|
||
return blake3(b"wai:video-receipt-id\x01" + payload(r) + r["sig"])
|
||
|
||
|
||
def merkle_root(leaves):
|
||
"""jwp-receipts §2.2: the final leaf duplicated on an odd level; the empty root is zero."""
|
||
if not leaves:
|
||
return bytes(32)
|
||
level = [blake3(LEAF + h) for h in leaves]
|
||
while len(level) > 1:
|
||
if len(level) % 2:
|
||
level.append(level[-1])
|
||
level = [blake3(NODE + level[i] + level[i + 1]) for i in range(0, len(level), 2)]
|
||
return level[0]
|
||
|
||
|
||
def frame_hash(i, rgb):
|
||
return blake3(b"wai:video-rframe\x01" + struct.pack("<Q", i) + bytes(rgb))
|
||
|
||
|
||
def clip_hash(h, w, frames):
|
||
return blake3(b"wai:video-clip\x01" + struct.pack("<IIQ", h, w, len(frames))
|
||
+ b"".join(struct.pack("<Q", i) + f for i, f in enumerate(frames)))
|
||
|
||
|
||
def consistent(r):
|
||
"""§15.1: None when the receipt is consistent, otherwise why not."""
|
||
if merkle_root(r["frames"]) != r["merkle_root"]:
|
||
return "merkle root"
|
||
if clip_hash(r["h"], r["w"], r["frames"]) != r["clip_hash"]:
|
||
return "clip hash"
|
||
if min(len(r["frames"]) * r["h"] * r["w"] * 3, U64) != r["work"]:
|
||
return "work"
|
||
cls, j = r["class"], r["joules"]
|
||
if (cls is None) != (j == 0):
|
||
return "figure and class"
|
||
if cls is not None and cls[1] is not None and j * cls[1] < RESOLUTION_FLOOR:
|
||
return "below the resolution floor"
|
||
if ED.verify(r["signer"], payload(r), r["sig"]):
|
||
return "signature"
|
||
return None
|
||
|
||
|
||
def rederived(r, p):
|
||
"""§15.1: None when the payload re-derives the receipt, otherwise why not."""
|
||
if blake3(p) != r["codestream_hash"]:
|
||
return "codestream hash"
|
||
try:
|
||
v = IPV.parse_wiv1(p, 0)
|
||
frames = IPV.decode_wiv1_k0(v)
|
||
except Exception as e: # a refused payload re-derives nothing
|
||
return "the payload does not decode: %s" % e
|
||
if (v["h"], v["w"], len(frames)) != (r["h"], r["w"], len(r["frames"])):
|
||
return "geometry"
|
||
if any(frame_hash(i, f) != r["frames"][i] for i, f in enumerate(frames)):
|
||
return "a frame"
|
||
return consistent(r)
|