ce9207ac3c
staged-conformance/ holds stage sets and the steps a sink takes with them: image layers over a pinned integer image base (identity, up2 to even and odd sizes) and over a PNG base; video layers per frame and per GOP, with prev runs, over integer motion bases and over a pinned neural-keyframe clip; audio layers over a pinned integer audio base, and a stereo rate ladder over a FLAC base; a splat attribute layer; and one case per refusal and order rule (another base, a wrong producer, a missing layer, conflicts held and after an application, out of order, a fallback base, geometry, truncation, a malformed table, q of 0, a word after the last symbol, a stream that runs out, clamped and bypassed residuals, malformed refines, a capability that is not a refinement, a base not admitted, inclusion-only evidence, a cost limit, a missed deadline, the hold bound). Each case's case.json states what a sink reports at every step and holds at the end; schedule.json holds arrival traces and what a sink presents. wai_staged_vectors gen writes the corpus; verify regenerates it byte for byte, replays every case through StageSink, decodes the integer bases with their pins against stage0.bin and checks their tiers. staged_verify.py is a standard-library reimplementation of the envelope, refines, WIR1, the sink's algorithm, tiers and stage hashes, and the PNG, FLAC and WIV1 keyframe-0 decodes; it checks every layer application in every case. The pure-Python BLAKE3 moves from claim_verify.py to _blake3.py, which both verifiers import. A unit held and decided within one step is reported by its decision alone. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1019 lines
48 KiB
Python
1019 lines
48 KiB
Python
#!/usr/bin/env python3
|
|
"""Second verifier for claim-conformance/ — typed claim receipts, the soft binding class,
|
|
and recovery under the ancestor rule (extensions/jwp-receipts.md §2.4.3-§2.4.9, §2.6,
|
|
§3.1-§3.2), and the arithmetic of a measurement report (energy-measurement.md §3.1, §6.2),
|
|
computed with Python's integers.
|
|
|
|
Written from the extension text. It shares no code with wai-rs (`crate::claim_receipt`),
|
|
which generates the corpus: the canonical form is Python's JSON serialiser with members
|
|
ordered by UTF-16 code units, BLAKE3 is written below from its specification, Ed25519's
|
|
equation is the `cryptography` package's and step 3's other checks (key decoding, the
|
|
order of the key and of R, S < L) are written below from the curve's definition, and OER
|
|
receipts named by an `energy` claim are read, verified and their figures read by the OER
|
|
reference verifier (`../oer-conformance/oer2_verify.py`), under oer-receipts' own rules.
|
|
|
|
python3 claim_verify.py [vectors.json] verify every vector and the canonical fixture
|
|
python3 claim_verify.py fixture print the fixture's canonical bytes,
|
|
signature and link hash
|
|
python3 claim_verify.py brackets print the figure and least declaration of
|
|
each bracket wai-rs pins
|
|
python3 claim_verify.py verdicts <file> print `name<TAB>verdict` for every vector of a
|
|
corpus-format file, ignoring any declared
|
|
outcome (the input to differential.py)
|
|
|
|
Exit status 0 only if every vector reports its declared outcome, and the fixture and the
|
|
bracket arithmetic this script computes equal the literals pinned in wai-rs.
|
|
|
|
A corpus-format file is read as the OER reference reads a vector document (`OER.load`,
|
|
oer-receipts §4 "The document"), since it holds OER receipts: a file that reading refuses
|
|
has no verdicts, and `verdicts` exits 1 with no output. No integer in a claim is converted
|
|
from more than twenty digits, so no verdict depends on an interpreter's limit on integer
|
|
digits.
|
|
"""
|
|
import json, os, re, sys, unicodedata
|
|
from decimal import localcontext
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.join(HERE, "..", "oer-conformance"))
|
|
import oer2_verify as OER # noqa: E402 (the OER reference verifier: grades + receipt checks)
|
|
sys.path.insert(0, HERE)
|
|
|
|
from cryptography.exceptions import InvalidSignature # noqa: E402
|
|
from cryptography.hazmat.primitives import serialization # noqa: E402
|
|
from cryptography.hazmat.primitives.asymmetric import ed25519 # noqa: E402
|
|
|
|
# ---------------------------------------------------------------- BLAKE3 (hash mode)
|
|
|
|
from _blake3 import blake3 # noqa: E402 (written from its specification; shared with staged_verify.py)
|
|
|
|
|
|
# ------------------------------------------------ Ed25519 (RFC 8032), §3.1 step 3's rule
|
|
#
|
|
# RFC 8032 verifiers differ on two choices that decide whether a signature verifies:
|
|
# whether keys and R of small order are refused (RFC 8032's procedure does not refuse
|
|
# them), and whether the cofactored or the cofactorless equation is checked (§5.1.7 states
|
|
# the first and allows the second). §3.1 step 3 fixes both, and restates RFC 8032's own decoding rule
|
|
# (one encoding per point) and range rule (S < L). Every check but the equation is made
|
|
# here, from the curve's definition: the `cryptography` package decodes keys more
|
|
# leniently than RFC 8032 §5.1.3 and accepts keys and R of small order. It does check
|
|
# S < L, which is made here as well. The equation itself is left to it.
|
|
|
|
_P = 2 ** 255 - 19
|
|
_L = 2 ** 252 + 27742317777372353535851937790883648493
|
|
_D = (-121665 * pow(121666, _P - 2, _P)) % _P
|
|
_SQRT_M1 = pow(2, (_P - 1) // 4, _P)
|
|
|
|
|
|
def point_decode(b: bytes):
|
|
"""RFC 8032 §5.1.3: the affine point (x, y), or None where decoding fails — including
|
|
y >= p and x = 0 with the sign bit set, so a point has exactly one encoding."""
|
|
y = int.from_bytes(b, "little")
|
|
sign, y = y >> 255, y & ((1 << 255) - 1)
|
|
if y >= _P:
|
|
return None
|
|
u, v = (y * y - 1) % _P, (_D * y * y + 1) % _P
|
|
x2 = u * pow(v, _P - 2, _P) % _P
|
|
if x2 == 0:
|
|
return (0, y) if sign == 0 else None
|
|
x = pow(x2, (_P + 3) // 8, _P)
|
|
if (x * x - x2) % _P:
|
|
x = x * _SQRT_M1 % _P
|
|
if (x * x - x2) % _P:
|
|
return None
|
|
if x & 1 != sign:
|
|
x = _P - x
|
|
return (x, y)
|
|
|
|
|
|
def _add(p, q):
|
|
"""Twisted Edwards addition (a = -1), complete on this curve."""
|
|
(x1, y1), (x2, y2) = p, q
|
|
t = _D * x1 * x2 * y1 * y2 % _P
|
|
x3 = (x1 * y2 + y1 * x2) * pow(1 + t, _P - 2, _P) % _P
|
|
y3 = (y1 * y2 + x1 * x2) * pow(1 - t, _P - 2, _P) % _P
|
|
return (x3, y3)
|
|
|
|
|
|
def small_order(pt) -> bool:
|
|
"""Whether [8]pt is the identity: the point's order divides 8."""
|
|
for _ in range(3):
|
|
pt = _add(pt, pt)
|
|
return pt == (0, 1)
|
|
|
|
|
|
def check_signature(pk: bytes, sig: bytes, msg: bytes):
|
|
"""§3.1 step 3, in its order: the key decodes and is not of small order
|
|
(`bad_signer_key`); then S < L, R is not a point of small order, and the cofactorless
|
|
equation holds with R compared byte for byte (`signature_verify_failed`)."""
|
|
a = point_decode(pk)
|
|
need(a is not None and not small_order(a), 3, "bad_signer_key")
|
|
r = point_decode(sig[:32])
|
|
need(int.from_bytes(sig[32:], "little") < _L, 3, "signature_verify_failed")
|
|
need(r is None or not small_order(r), 3, "signature_verify_failed")
|
|
try:
|
|
ed25519.Ed25519PublicKey.from_public_bytes(pk).verify(sig, msg)
|
|
except (InvalidSignature, ValueError):
|
|
raise Reject(3, "signature_verify_failed")
|
|
|
|
|
|
# ------------------------------------------------------------- canonical form (RFC 8785)
|
|
|
|
MAX_INT = 2 ** 53 - 1
|
|
|
|
|
|
def jcs(v):
|
|
if v is None:
|
|
return "null"
|
|
if v is True or v is False:
|
|
return "true" if v else "false"
|
|
if isinstance(v, int):
|
|
if not 0 <= v <= MAX_INT:
|
|
raise ValueError("integer outside the canonical subset")
|
|
return str(v)
|
|
if isinstance(v, str):
|
|
return json.dumps(v, ensure_ascii=False)
|
|
if isinstance(v, list):
|
|
return "[" + ",".join(jcs(x) for x in v) + "]"
|
|
if isinstance(v, dict):
|
|
keys = sorted(v, key=lambda k: k.encode("utf-16-be"))
|
|
return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + jcs(v[k]) for k in keys) + "}"
|
|
raise ValueError("value outside the canonical subset")
|
|
|
|
|
|
def link_hash(claim: dict) -> str:
|
|
return blake3(b"moq-jwp:claim-link\x01" + jcs(claim).encode()).hex()
|
|
|
|
|
|
def signing_input(claim: dict) -> bytes:
|
|
return b"moq-jwp:claim\x01" + jcs({k: v for k, v in claim.items() if k != "sig"}).encode()
|
|
|
|
|
|
MAX_DEPTH = 32 # §3.1 step 1: arrays and objects nested deeper, the receipt being level 1
|
|
|
|
|
|
class OutsideSubset:
|
|
"""A JSON number that is not an unsigned integer written without a sign, fraction or
|
|
exponent: whatever its value, not one a claim may carry (`field_type`, never
|
|
`malformed`). Kept as a marker rather than a float so that no reading depends on how a
|
|
float is rounded, or on whether it overflows."""
|
|
|
|
|
|
OUTSIDE_SUBSET = OutsideSubset()
|
|
|
|
|
|
def _scalar_text(s):
|
|
"""Whether `s` is Unicode text: no surrogate code point, which is what an escape that
|
|
leaves a surrogate unpaired decodes to here."""
|
|
return not any("\ud800" <= ch <= "\udfff" for ch in s)
|
|
|
|
|
|
def _check_tree(v, level):
|
|
if isinstance(v, str):
|
|
if not _scalar_text(v):
|
|
raise ValueError("an escape that leaves a surrogate unpaired")
|
|
elif isinstance(v, (list, dict)):
|
|
if level + 1 > MAX_DEPTH:
|
|
raise ValueError("nested too deeply")
|
|
for x in (v.values() if isinstance(v, dict) else v):
|
|
_check_tree(x, level + 1)
|
|
|
|
|
|
def parse_strict(text):
|
|
"""§3.1 step 1's reading of a JSON text; ValueError is `malformed`. Refuses what RFC
|
|
8259 does, a duplicate member name, a string or member name holding an unpaired
|
|
surrogate, and nesting deeper than MAX_DEPTH. Never refuses a number: one outside the
|
|
claim subset is OUTSIDE_SUBSET."""
|
|
def pairs(ps):
|
|
d = {}
|
|
for k, v in ps:
|
|
if not _scalar_text(k):
|
|
raise ValueError("an escape that leaves a surrogate unpaired")
|
|
if k in d:
|
|
raise ValueError("duplicate member " + k)
|
|
d[k] = v
|
|
return d
|
|
|
|
def no_constant(c):
|
|
raise ValueError("not JSON: " + c)
|
|
|
|
def integer(s):
|
|
# A plain unsigned integer beyond 20 digits exceeds every u64, as wai-rs reads it.
|
|
return OUTSIDE_SUBSET if s.startswith("-") or len(s) > 20 else int(s)
|
|
|
|
if not isinstance(text, str):
|
|
raise TypeError("a claim is given as its wire form")
|
|
try:
|
|
v = json.loads(text, object_pairs_hook=pairs, parse_constant=no_constant,
|
|
parse_float=lambda s: OUTSIDE_SUBSET, parse_int=integer)
|
|
except RecursionError:
|
|
raise ValueError("nested too deeply")
|
|
_check_tree(v, 0)
|
|
return v
|
|
|
|
|
|
# -------------------------------------------------------------------- the extension
|
|
|
|
ENVELOPE = ["kind", "class", "subject_type", "subject", "seq", "issued_at", "body",
|
|
"signer_pubkey", "signer_id", "parent_receipt_hash", "sig"]
|
|
ADMITS = {"energy": {"object"}, "lawful-basis": {"object", "receipt"}, "log-inclusion": {"receipt"},
|
|
"soft-binding": {"object"}, "binding-recovery": {"object"},
|
|
"measurement": {"object", "receipt"}, "operating-point": {"object", "receipt"}}
|
|
FIGURE = ["figure_profile", "figure_index", "group", "joules_micro", "energy_provenance", "energy_uncertainty",
|
|
"energy_counter_filtering", "energy_calibration_ref"]
|
|
BODY = {
|
|
"energy": ["joules_micro", "energy_provenance", "energy_uncertainty", "energy_counter_filtering",
|
|
"energy_calibration_ref", "coverage", "grade", "method", "work_units", "work_unit", "oer_receipt"],
|
|
"log-inclusion": ["log", "profile", "entry", "log_receipt"],
|
|
"soft-binding": ["alg", "registry", "binding_type", "value", "resolution_endpoint"],
|
|
"binding-recovery": ["alg", "binding_type", "value", "declaration", "resolution_endpoint", "observed", "attests"],
|
|
"measurement": FIGURE + ["bracket", "build", "power_reader"],
|
|
"operating-point": FIGURE + ["vocabulary", "op_requested", "op_committed", "op_delivered", "prediction"],
|
|
}
|
|
BRACKET = ["idle", "busy", "window_us", "completions", "in_flight", "opening"]
|
|
SUMMARY = ["n", "mean_uw", "sd_uw", "live_intervals", "longest_live_run"]
|
|
BUILD = ["profile", "toolchain", "target"]
|
|
PREDICTION = ["predictor", "descriptor_hash", "predicted_joules_micro"]
|
|
# energy-measurement §6.1: profile -> (subject_type, carries several figures, an Ed25519
|
|
# key signs the figure, names a group).
|
|
PROFILES = {"moq-jwp:object": ("object", True, True, True), "moq-jwp:claim-link": ("receipt", False, True, False),
|
|
"wai:energy-binding-id": ("receipt", False, False, False)}
|
|
for _n in ["video-receipt-id", "video-manifest-receipt-id", "video-ssai-receipt-id", "video-keys-receipt-id",
|
|
"video-package-receipt-id", "video-channel-receipt-id", "world-receipt-id", "film-receipt-id",
|
|
"haptic-receipt-id", "score-receipt-id", "splat4d-receipt-id", "audio-receipt-id", "ternary-receipt-id",
|
|
"quantum-receipt-id", "quantum-calibration-id", "quantum-job-id", "quantum-mitigate-id",
|
|
"quantum-decode-id", "quantum-compile-id", "quantum-rearrange-id", "phasor-receipt-id"]:
|
|
PROFILES["wai:" + _n] = ("receipt", False, True, False)
|
|
for _n in ["asset-receipt-id", "quantum-qbom-id"]:
|
|
PROFILES["wai:" + _n] = ("receipt", True, True, False)
|
|
DECLARATION = ["record", "vocabulary", "fragment", "purposes"]
|
|
PURPOSE = ["purpose", "basis", "data"]
|
|
DECISION = ["record", "vocabulary", "declaration", "purpose", "outcome"]
|
|
UNCERTAINTY = ["relative_ppm", "window_us"]
|
|
# energy-measurement.md §2's acquisition classes, each with the strongest oer-receipts.md
|
|
# §2 grade it supports. Grades and their ranking come from the OER reference verifier.
|
|
STRONGEST = {"CalibratedInstrument": "measured", "OnChipCounter": "measured", "HwShunt": "measured",
|
|
"ModelBased": "derived", "Estimator": "projected"}
|
|
FILTERING = {"On", "Off", "Unknown"} # energy-measurement.md §2.4
|
|
RESOLUTION_FLOOR = 288676 # energy-measurement.md §2.3: joules_micro x relative_ppm, at least
|
|
# Unicode's White_Space property: what a calibration reference must hold something besides
|
|
# (energy-measurement.md §2.5, "non-empty"). Written out, because str.isspace() also
|
|
# counts U+001C to U+001F, which are not White_Space.
|
|
WHITE_SPACE = ("\t\n\x0b\x0c\r \x85\xa0\u1680\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007"
|
|
"\u2008\u2009\u200a\u2028\u2029\u202f\u205f\u3000")
|
|
COVERAGE = {"Whole", "Partial"}
|
|
GRADES = OER.GRADES
|
|
|
|
|
|
class Reject(Exception):
|
|
def __init__(self, step, reason):
|
|
super().__init__(f"reject step={step} reason={reason}")
|
|
self.step, self.reason = step, reason
|
|
|
|
|
|
def is_int(v):
|
|
return isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= MAX_INT
|
|
|
|
|
|
def is_hex(v, nbytes=None):
|
|
if not isinstance(v, str) or len(v) % 2 or not re.fullmatch(r"[0-9a-f]*", v):
|
|
return False
|
|
return nbytes is None or len(v) == 2 * nbytes
|
|
|
|
|
|
def is_token(s):
|
|
return (isinstance(s, str) and s != "" and len(s.encode()) <= 2048
|
|
and not any(c.isspace() or unicodedata.category(c) == "Cc" for c in s))
|
|
|
|
|
|
def is_uri(s):
|
|
return is_token(s) and re.fullmatch(r"[A-Za-z][A-Za-z0-9+.\-]*:.+", s, re.S) is not None
|
|
|
|
|
|
def is_https(s):
|
|
"""§2.6.2: `https://`, then an authority (up to the first `/`, `?` or `#`) with no `@`,
|
|
that is a non-empty host — the text before the first `:`, or what an IP literal's
|
|
brackets enclose — optionally followed by `:` and a port of ASCII digits."""
|
|
if not isinstance(s, str) or not s.startswith("https://"):
|
|
return False
|
|
authority = re.split(r"[/?#]", s[len("https://"):], maxsplit=1)[0]
|
|
if not is_uri(s) or "@" in authority:
|
|
return False
|
|
if authority.startswith("["):
|
|
host, closed, after = authority[1:].partition("]")
|
|
if not closed:
|
|
return False
|
|
else:
|
|
host, colon, port = authority.partition(":")
|
|
after = colon + port
|
|
return host != "" and (after == "" or (after[0] == ":" and all(c in "0123456789" for c in after[1:])))
|
|
|
|
|
|
def registry_form(alg):
|
|
labels = alg.split(".")
|
|
return len(labels) >= 2 and all(re.fullmatch(r"[A-Za-z0-9_+\-]+", l) for l in labels)
|
|
|
|
|
|
def days_in(year, month):
|
|
"""Days in a month of the proleptic Gregorian calendar (RFC 3339 Appendix C)."""
|
|
if month == 2:
|
|
return 29 if year % 4 == 0 and (year % 100 != 0 or year % 400 == 0) else 28
|
|
return 30 if month in (4, 6, 9, 11) else 31
|
|
|
|
|
|
def is_rfc3339(s):
|
|
"""RFC 3339 §5.6 date-time with §5.7's day-of-month rule; ASCII digits only (a regex
|
|
`\\d` would also take any Unicode decimal digit). A second of 60 is accepted at any
|
|
time of day: whether a leap second occurred then is not a property of the syntax."""
|
|
m = re.fullmatch(r"([0-9]{4})-([0-9]{2})-([0-9]{2})[Tt]([0-9]{2}):([0-9]{2}):([0-9]{2})(\.[0-9]+)?"
|
|
r"([Zz]|[+-]([0-9]{2}):([0-9]{2}))", s)
|
|
if not m:
|
|
return False
|
|
yr, mo, da, hh, mi, ss = (int(m.group(i)) for i in (1, 2, 3, 4, 5, 6))
|
|
if not (1 <= mo <= 12 and 1 <= da <= days_in(yr, mo) and hh <= 23 and mi <= 59 and ss <= 60):
|
|
return False
|
|
return m.group(9) is None or (int(m.group(9)) <= 23 and int(m.group(10)) <= 59)
|
|
|
|
|
|
def is_oer_id(s):
|
|
return isinstance(s, str) and re.fullmatch(r"(sha256|blake3):[0-9a-f]{64}", s) is not None
|
|
|
|
|
|
def members(obj, required, step):
|
|
if any(k not in obj for k in required):
|
|
raise Reject(step, "missing_required_field")
|
|
if any(k not in required for k in obj):
|
|
raise Reject(step, "unknown_member")
|
|
|
|
|
|
def need(ok, step, reason="field_type"):
|
|
if not ok:
|
|
raise Reject(step, reason)
|
|
|
|
|
|
def check_body(cls, b):
|
|
"""§3.1 step 2: the members, then each member's type and encoding together, in the
|
|
member table's order, a nested object's members checked where the table reaches it.
|
|
Where every failure left is `field_type` the checks are grouped; the order shows only
|
|
where a nested object's missing or unknown member could be reported instead."""
|
|
need(isinstance(b, dict), 2)
|
|
if cls == "lawful-basis":
|
|
# §2.4.6: `record` is read before anything else, and names the table.
|
|
if "record" not in b:
|
|
raise Reject(2, "missing_required_field")
|
|
need(isinstance(b["record"], str), 2)
|
|
if b["record"] == "declaration":
|
|
members(b, DECLARATION, 2)
|
|
need(is_uri(b["vocabulary"]), 2)
|
|
need(b["fragment"] is None or is_hex(b["fragment"], 32), 2)
|
|
need(isinstance(b["purposes"], list), 2)
|
|
for p in b["purposes"]:
|
|
need(isinstance(p, dict), 2)
|
|
members(p, PURPOSE, 2)
|
|
need(is_token(p["purpose"]), 2)
|
|
need(p["basis"] == "" or is_token(p["basis"]), 2) # an empty basis is step 4's
|
|
need(isinstance(p["data"], list) and all(is_token(d) for d in p["data"]), 2)
|
|
elif b["record"] == "decision":
|
|
members(b, DECISION, 2)
|
|
need(all(isinstance(b[k], str) for k in ("vocabulary", "purpose", "outcome")), 2)
|
|
need(is_hex(b["declaration"], 32), 2)
|
|
need(is_uri(b["vocabulary"]) and is_token(b["purpose"]) and is_token(b["outcome"]), 2)
|
|
else:
|
|
raise Reject(2, "field_type")
|
|
return
|
|
members(b, BODY[cls], 2)
|
|
if cls in ("measurement", "operating-point"):
|
|
check_figure_header(b)
|
|
if cls == "measurement":
|
|
br = b["bracket"]
|
|
if br is not None:
|
|
need(isinstance(br, dict), 2)
|
|
members(br, BRACKET, 2)
|
|
for side in ("idle", "busy"):
|
|
need(isinstance(br[side], dict), 2)
|
|
members(br[side], SUMMARY, 2)
|
|
need(all(is_int(br[side][k]) for k in SUMMARY), 2)
|
|
need(all(is_int(br[k]) for k in ("window_us", "completions", "in_flight")), 2)
|
|
need(br["opening"] in ("from-rest", "running"), 2)
|
|
bd = b["build"]
|
|
need(isinstance(bd, dict), 2)
|
|
members(bd, BUILD, 2)
|
|
need(bd["profile"] in ("release", "debug"), 2)
|
|
need(bd["toolchain"] is None or isinstance(bd["toolchain"], str), 2)
|
|
need(bd["target"] is None or is_token(bd["target"]), 2)
|
|
need(b["power_reader"] is None or isinstance(b["power_reader"], str), 2)
|
|
else:
|
|
need(is_uri(b["vocabulary"]), 2)
|
|
for k in ("op_requested", "op_committed", "op_delivered"):
|
|
pt = b[k]
|
|
if pt is not None:
|
|
need(isinstance(pt, dict), 2)
|
|
for name, val in pt.items():
|
|
need(is_token(name) and is_int(val), 2)
|
|
pr = b["prediction"]
|
|
if pr is not None:
|
|
need(isinstance(pr, dict), 2)
|
|
members(pr, PREDICTION, 2)
|
|
need(is_uri(pr["predictor"]), 2)
|
|
need(is_hex(pr["descriptor_hash"], 32), 2)
|
|
need(is_int(pr["predicted_joules_micro"]), 2)
|
|
return
|
|
if cls == "energy":
|
|
# In the member table's order: a nested object's member check can report another
|
|
# reason than a type check, so the order is the one every verifier uses.
|
|
need(is_int(b["joules_micro"]), 2)
|
|
need(isinstance(b["energy_provenance"], str), 2)
|
|
u = b["energy_uncertainty"]
|
|
if u is not None:
|
|
need(isinstance(u, dict), 2)
|
|
members(u, UNCERTAINTY, 2)
|
|
need(is_int(u["relative_ppm"]) and is_int(u["window_us"]), 2)
|
|
need(b["energy_counter_filtering"] is None or isinstance(b["energy_counter_filtering"], str), 2)
|
|
need(b["energy_calibration_ref"] is None or isinstance(b["energy_calibration_ref"], str), 2)
|
|
need(all(isinstance(b[k], str) for k in ("coverage", "grade", "method")), 2)
|
|
need(is_int(b["work_units"]), 2)
|
|
need(isinstance(b["work_unit"], str), 2)
|
|
need(b["oer_receipt"] is None or isinstance(b["oer_receipt"], str), 2)
|
|
need(is_uri(b["method"]) and is_token(b["work_unit"]), 2)
|
|
need(b["oer_receipt"] is None or is_oer_id(b["oer_receipt"]), 2)
|
|
elif cls == "log-inclusion":
|
|
need(all(isinstance(b[k], str) for k in ("log", "profile", "entry")), 2)
|
|
need(is_hex(b["log_receipt"]), 2)
|
|
need(is_uri(b["log"]) and is_uri(b["profile"]) and is_token(b["entry"]), 2)
|
|
else: # soft-binding, binding-recovery
|
|
need(isinstance(b["alg"], str) and b["binding_type"] in ("watermark", "fingerprint"), 2)
|
|
need(is_hex(b["value"]) and isinstance(b["resolution_endpoint"], str), 2)
|
|
if cls == "soft-binding":
|
|
need(isinstance(b["registry"], str), 2)
|
|
need(is_token(b["alg"]) and is_uri(b["registry"]) and is_uri(b["resolution_endpoint"]), 2)
|
|
else:
|
|
need(is_hex(b["declaration"], 32), 2)
|
|
need(b["observed"] is None or is_hex(b["observed"], 32), 2)
|
|
need(isinstance(b["attests"], str), 2)
|
|
need(is_token(b["alg"]) and is_uri(b["resolution_endpoint"]), 2)
|
|
|
|
|
|
def check_figure_header(b):
|
|
"""§2.4.8's header at step 2, in its table's order: `figure_profile` a term, as it is
|
|
read; the index an integer or null; the group a hash or null; then the figure and the
|
|
class's members as §2.4.5 types them."""
|
|
need(is_token(b["figure_profile"]), 2)
|
|
need(b["figure_index"] is None or is_int(b["figure_index"]), 2)
|
|
need(b["group"] is None or is_hex(b["group"], 32), 2)
|
|
need(is_int(b["joules_micro"]), 2)
|
|
need(isinstance(b["energy_provenance"], str), 2)
|
|
u = b["energy_uncertainty"]
|
|
if u is not None:
|
|
need(isinstance(u, dict), 2)
|
|
members(u, UNCERTAINTY, 2)
|
|
need(is_int(u["relative_ppm"]) and is_int(u["window_us"]), 2)
|
|
need(b["energy_counter_filtering"] is None or isinstance(b["energy_counter_filtering"], str), 2)
|
|
need(b["energy_calibration_ref"] is None or isinstance(b["energy_calibration_ref"], str), 2)
|
|
|
|
|
|
# ------------------------------------------- a measurement report's arithmetic (§3.1, §6.2)
|
|
|
|
def summary_live(s):
|
|
"""energy-measurement §3.1: a rail is live over a bracket only if it advanced in two
|
|
consecutive intervals, in at least five, and in at least four of every five — facts
|
|
that describe one sequence of `n` intervals. A power sample advances only where it
|
|
differs from the one before it, so a bracket whose samples are all equal (`sd_uw` 0)
|
|
advanced once at most, and is never live whatever facts it carries."""
|
|
n, live, run = s["n"], s["live_intervals"], s["longest_live_run"]
|
|
return s["sd_uw"] > 0 and run <= live <= n and run >= 2 and live >= 5 and live * 5 >= n * 4
|
|
|
|
|
|
def work(br):
|
|
c, n = br["completions"], br["in_flight"]
|
|
return (2 * c + n, 2) if br["opening"] == "from-rest" else (c, 1)
|
|
|
|
|
|
def bracket_figure(br):
|
|
"""F = floor((2·M·T·Wd + 10^6·Wn) / (2·10^6·Wn)), or None."""
|
|
m = br["busy"]["mean_uw"] - br["idle"]["mean_uw"]
|
|
t, c, n = br["window_us"], br["completions"], br["in_flight"]
|
|
if m <= 0 or t == 0 or c == 0 or n == 0:
|
|
return None
|
|
wn, wd = work(br)
|
|
f = (2 * m * t * wd + 10 ** 6 * wn) // (2 * 10 ** 6 * wn)
|
|
return f if 0 < f <= MAX_INT else None
|
|
|
|
|
|
def k_and_p(br, f):
|
|
m = br["busy"]["mean_uw"] - br["idle"]["mean_uw"]
|
|
t, n = br["window_us"], br["in_flight"]
|
|
wn, wd = work(br)
|
|
ni, nb, si, sb = br["idle"]["n"], br["busy"]["n"], br["idle"]["sd_uw"], br["busy"]["sd_uw"]
|
|
k = 12 * ni * nb * wn ** 4 * f * f
|
|
p = 12 * wn ** 2 * t ** 2 * wd ** 2 * (si * si * nb + sb * sb * ni) + 4 * ni * nb * m * m * t * t * wd * wd * n * n \
|
|
+ 10 ** 12 * ni * nb * wn ** 4
|
|
return k, p
|
|
|
|
|
|
def least_ppm(br, f):
|
|
"""The least p with p^2·K >= P, or None above 2^32 - 1."""
|
|
k, p = k_and_p(br, f)
|
|
if k == 0:
|
|
return None
|
|
q = -(-p // k) # p^2 >= ceil(P / K)
|
|
r = __import__("math").isqrt(q)
|
|
while r * r * k < p:
|
|
r += 1
|
|
while r > 1 and (r - 1) * (r - 1) * k >= p:
|
|
r -= 1
|
|
return r if 1 <= r <= 2 ** 32 - 1 else None
|
|
|
|
|
|
def bracket_rules(br, joules, label, u):
|
|
"""§2.4.8 rule 5, in its order."""
|
|
hw = label in ("OnChipCounter", "CalibratedInstrument")
|
|
need(br["idle"]["n"] >= 2 and br["busy"]["n"] >= 2, 4, "measurement_too_few_samples")
|
|
need(not hw or (summary_live(br["idle"]) and summary_live(br["busy"])), 4, "measurement_rail_not_live")
|
|
need(br["window_us"] != 0 and br["completions"] != 0 and br["in_flight"] != 0, 4, "measurement_empty_burst")
|
|
m = br["busy"]["mean_uw"] - br["idle"]["mean_uw"]
|
|
need(m > 0, 4, "measurement_no_marginal")
|
|
need(not hw or m >= br["idle"]["sd_uw"], 4, "measurement_below_attribution_floor")
|
|
need(bracket_figure(br) == joules, 4, "measurement_figure_mismatch")
|
|
if hw:
|
|
need(u["window_us"] == br["window_us"], 4, "measurement_window_mismatch")
|
|
k, p = k_and_p(br, joules)
|
|
need(u["relative_ppm"] ** 2 * k >= p, 4, "measurement_declaration_below_bracket")
|
|
|
|
|
|
def figure_rules(c):
|
|
"""§2.4.8 rules 1-3: a non-zero figure, a profile for the subject type with an index
|
|
and a group exactly where it has them, and the class."""
|
|
b = c["body"]
|
|
need(b["joules_micro"] != 0, 4, "energy_zero_figure")
|
|
prof = PROFILES.get(b["figure_profile"])
|
|
need(prof is not None and prof[0] == c["subject_type"], 4, "figure_unknown_profile")
|
|
need((b["figure_index"] is not None) == prof[1], 4, "figure_index_mismatch")
|
|
need((b["group"] is not None) == prof[3], 4, "figure_group_mismatch")
|
|
acquisition(b)
|
|
|
|
|
|
def acquisition(b):
|
|
"""§2.4.5 rule 2: the class is one energy-measurement.md §2 names, not the legacy
|
|
HwShunt, and carries exactly the evidence §2.3-§2.5 give it: a refined hardware class
|
|
a non-zero declared uncertainty whose relative part fits in 32 bits (§2.7) and which
|
|
reaches the resolution floor, an on-chip counter a filtering state and no calibration
|
|
reference, a calibrated instrument a calibration reference that is not all white space
|
|
and no filtering state; every other class none of the three."""
|
|
label = b["energy_provenance"]
|
|
need(label in STRONGEST, 4, "energy_unknown_acquisition")
|
|
need(label != "HwShunt", 4, "energy_legacy_acquisition")
|
|
u, f, ref = b["energy_uncertainty"], b["energy_counter_filtering"], b["energy_calibration_ref"]
|
|
ev = "energy_acquisition_evidence"
|
|
if label in ("OnChipCounter", "CalibratedInstrument"):
|
|
need(u is not None and 0 < u["relative_ppm"] <= 2 ** 32 - 1 and u["window_us"] > 0, 4, ev)
|
|
if label == "OnChipCounter":
|
|
need(f in FILTERING and ref is None, 4, ev)
|
|
else:
|
|
need(f is None and ref is not None and ref.strip(WHITE_SPACE) != "", 4, ev)
|
|
need(b["joules_micro"] * u["relative_ppm"] >= RESOLUTION_FLOOR, 4, ev)
|
|
else:
|
|
need(u is None and f is None and ref is None, 4, ev)
|
|
|
|
|
|
def class_rules(c):
|
|
"""§3.1 step 4 — §2.4.5-§2.4.7 and §2.6."""
|
|
cls, b = c["class"], c["body"]
|
|
if cls == "energy":
|
|
need(b["joules_micro"] != 0, 4, "energy_zero_figure")
|
|
acquisition(b)
|
|
need(b["coverage"] in COVERAGE, 4, "energy_unknown_coverage")
|
|
need(b["grade"] in GRADES, 4, "energy_unknown_grade")
|
|
need(GRADES.index(b["grade"]) <= GRADES.index(STRONGEST[b["energy_provenance"]]), 4,
|
|
"energy_grade_exceeds_acquisition")
|
|
need(b["work_units"] != 0, 4, "energy_zero_work")
|
|
elif cls == "lawful-basis" and b["record"] == "declaration":
|
|
need(len(b["purposes"]) > 0, 4, "lawful_basis_no_purposes")
|
|
seen = set()
|
|
for p in b["purposes"]:
|
|
need(p["basis"] != "", 4, "lawful_basis_purpose_without_basis")
|
|
need(p["purpose"] not in seen, 4, "lawful_basis_duplicate_purpose")
|
|
seen.add(p["purpose"])
|
|
elif cls == "measurement":
|
|
figure_rules(c)
|
|
# Rule 4: a hardware figure is a reading, and its report names the reader.
|
|
need(b["energy_provenance"] not in ("OnChipCounter", "CalibratedInstrument") or b["power_reader"] is not None,
|
|
4, "measurement_no_power_reader")
|
|
if b["bracket"] is not None:
|
|
bracket_rules(b["bracket"], b["joules_micro"], b["energy_provenance"], b["energy_uncertainty"])
|
|
elif cls == "operating-point":
|
|
figure_rules(c)
|
|
pts = [b[k] for k in ("op_requested", "op_committed", "op_delivered")]
|
|
need(all(p is None or len(p) > 0 for p in pts), 4, "operating_point_empty_point")
|
|
req, com, dlv = pts
|
|
need(not (dlv is not None and com is None), 4, "operating_point_delivered_alone")
|
|
need(not ((req is not None or com is not None) and dlv is None), 4, "operating_point_incomplete")
|
|
need(dlv is not None or b["prediction"] is not None, 4, "operating_point_empty")
|
|
elif cls == "log-inclusion":
|
|
need(b["log_receipt"] != "", 4, "log_inclusion_no_evidence")
|
|
elif cls in ("soft-binding", "binding-recovery"):
|
|
need(registry_form(b["alg"]), 4, "soft_binding_alg_not_registry_form")
|
|
need(b["value"] != "", 4, "soft_binding_empty_value")
|
|
need(is_https(b["resolution_endpoint"]), 4, "soft_binding_endpoint_not_https")
|
|
if cls == "binding-recovery":
|
|
need(b["attests"] == "ancestor", 4, "recovery_not_ancestor")
|
|
need(b["observed"] != c["subject"], 4, "recovery_of_hard_bound_object")
|
|
|
|
|
|
def verify_claim(text):
|
|
"""§3.1 steps 1-4. Returns the claim; raises Reject."""
|
|
try:
|
|
c = parse_strict(text)
|
|
except ValueError:
|
|
raise Reject(1, "malformed")
|
|
need(isinstance(c, dict), 1, "malformed")
|
|
if "kind" not in c:
|
|
raise Reject(1, "missing_required_field")
|
|
need(c["kind"] == "claim", 1, "wrong_kind")
|
|
members(c, ENVELOPE, 1)
|
|
need(isinstance(c["class"], str), 1)
|
|
need(c["class"] != "delivery", 1, "class_reserved")
|
|
need(c["class"] in ADMITS, 1, "unknown_class")
|
|
need(c["subject_type"] in ("object", "receipt"), 1)
|
|
need(is_hex(c["subject"], 32), 1)
|
|
need(is_int(c["seq"]), 1)
|
|
need(isinstance(c["issued_at"], str) and is_rfc3339(c["issued_at"]), 1)
|
|
need(is_hex(c["signer_pubkey"], 32), 1)
|
|
need(isinstance(c["signer_id"], str), 1)
|
|
need(c["parent_receipt_hash"] is None or is_hex(c["parent_receipt_hash"], 32), 1)
|
|
need(is_hex(c["sig"], 64), 1)
|
|
need(c["subject_type"] in ADMITS[c["class"]], 1, "subject_type_not_admitted")
|
|
check_body(c["class"], c["body"])
|
|
check_signature(bytes.fromhex(c["signer_pubkey"]), bytes.fromhex(c["sig"]), signing_input(c))
|
|
class_rules(c)
|
|
return c
|
|
|
|
|
|
def check_chain(c, previous):
|
|
"""§3.1 step 5."""
|
|
parent = c["parent_receipt_hash"]
|
|
if previous is None:
|
|
need(parent is None, 5, "parent_not_held")
|
|
return
|
|
need(parent is not None, 5, "chain_break")
|
|
need(previous["signer_pubkey"] == c["signer_pubkey"] and parent == link_hash(previous)
|
|
and c["seq"] > previous["seq"], 5, "chain_break")
|
|
|
|
|
|
def match(alg, declared, observed, exact_algs):
|
|
need(alg in exact_algs, 6, "recovery_no_matcher")
|
|
need(declared == observed, 6, "recovery_value_mismatch")
|
|
|
|
|
|
def dig(v, *path):
|
|
"""The member at `path` through JSON objects, or None."""
|
|
for k in path:
|
|
v = v.get(k) if isinstance(v, dict) else None
|
|
return v
|
|
|
|
|
|
def figure_rounds_to_micro(figure, uj):
|
|
"""§2.4.5 check 4: whether an OER receipt's joules_total is `uj` microjoules to the
|
|
nearest microjoule, a figure half-way between two admitting either.
|
|
|
|
The figure is read as the OER reference reads a quantity for its own composition
|
|
checks (`OER.dec`: a decimal string of at most 34 significand digits, or an integer
|
|
within I-JSON's range) and compared in its exact context, so no digit is rounded
|
|
away: 0.0120004 followed by twenty-five nines is below half-way, although rounding it
|
|
to 28 digits would make it a tie. Check 2 has already refused a receipt whose figure
|
|
is anything else; such a value is no figure here either."""
|
|
try:
|
|
with localcontext(OER.EXACT):
|
|
return abs(OER.dec(figure) * 1000000 - uj) * 2 <= 1
|
|
except TypeError:
|
|
return False
|
|
|
|
|
|
def check_context(c, ctx):
|
|
"""§3.1 step 6."""
|
|
b = c["body"]
|
|
if "object" in ctx:
|
|
o = ctx["object"]
|
|
# The corpus gives a delivered object's microjoules as an integer within 64 bits, as
|
|
# wai-rs's harness requires; anything else (`-0`, which is read as a double, among
|
|
# them) is a malformed corpus, not a verdict.
|
|
j = o.get("joules_micro")
|
|
if not (isinstance(j, int) and not isinstance(j, bool) and 0 <= j < 2 ** 64):
|
|
raise ValueError("bad object.joules_micro")
|
|
need(c["class"] == "energy" and o["content_hash"] == c["subject"], 6, "context_mismatch")
|
|
need(o["joules_micro"] == b["joules_micro"], 6, "energy_disagrees_with_delivery")
|
|
label = class_of_json(o) # the label in the object's leaf, where its leaf is labelled
|
|
need(label is None or label == class_of_body(b), 6, "energy_class_disagrees_with_delivery")
|
|
if "figure" in ctx:
|
|
check_figure(c, figure_view(ctx["figure"]))
|
|
elif c["class"] == "measurement":
|
|
# §2.4.8: a report is its figure signer's only once step 6 has run against the
|
|
# figure (check 5); checked without it, it is not accepted.
|
|
raise Reject(6, "figure_not_held")
|
|
if "oer" in ctx:
|
|
r, keys = ctx["oer"]["receipt"], ctx["oer"]["keys"]
|
|
need(c["class"] == "energy" and b["oer_receipt"] is not None
|
|
and dig(r, "receipt_id") == b["oer_receipt"], 6, "context_mismatch")
|
|
# The reference reports every failure as a reason (oer-receipts §7); an exception
|
|
# here is a defect in it, and is left to surface as one rather than read as a verdict.
|
|
need(OER.verify_receipt(r, keys) == [], 6, "energy_oer_invalid")
|
|
need(dig(r, "subject", "output") == c["subject"], 6, "context_mismatch")
|
|
need(dig(r, "provenance", "how") == b["grade"], 6, "energy_disagrees_with_oer")
|
|
need(figure_rounds_to_micro(dig(r, "energy", "joules_total"), b["joules_micro"]), 6, "energy_disagrees_with_oer")
|
|
if "declaration" in ctx:
|
|
# In §3.1 step 6's order: the claim must be one that names a declaration; the
|
|
# declaration must itself pass steps 1-4 (context_invalid) — its signer_pubkey is
|
|
# only its signer's once that is established; then it must be of the class named.
|
|
recovery = c["class"] == "binding-recovery"
|
|
decision = c["class"] == "lawful-basis" and b["record"] == "decision"
|
|
# §2.6.3: the declaration is context only together with the publisher key of a
|
|
# verified delivery receipt for its subject. A corpus that gives one without the
|
|
# other is malformed, not a verdict.
|
|
if recovery and "publisher" not in ctx:
|
|
raise ValueError("a recovery record's declaration is context only with a publisher key")
|
|
need(recovery or decision, 6, "context_mismatch")
|
|
if not isinstance(ctx["declaration"], str):
|
|
raise TypeError("a declaration is given as its wire form")
|
|
try:
|
|
d = verify_claim(ctx["declaration"])
|
|
except Reject:
|
|
raise Reject(6, "context_invalid")
|
|
db = d["body"]
|
|
if recovery:
|
|
need(d["class"] == "soft-binding", 6, "context_mismatch")
|
|
need(b["declaration"] == link_hash(d) and c["subject"] == d["subject"]
|
|
and b["alg"] == db["alg"] and b["binding_type"] == db["binding_type"],
|
|
6, "recovery_declaration_mismatch")
|
|
need(d["signer_pubkey"] == ctx["publisher"], 6, "recovery_signer_mismatch")
|
|
match(db["alg"], db["value"], b["value"], ctx.get("exact_match_algs", []))
|
|
else:
|
|
need(d["class"] == "lawful-basis" and db["record"] == "declaration", 6, "context_mismatch")
|
|
need(b["declaration"] == link_hash(d) and c["subject"] == d["subject"]
|
|
and c["subject_type"] == d["subject_type"] and b["vocabulary"] == db["vocabulary"],
|
|
6, "decision_declaration_mismatch")
|
|
need(any(p["purpose"] == b["purpose"] for p in db["purposes"]), 6, "decision_purpose_not_declared")
|
|
|
|
|
|
def class_of_json(m):
|
|
"""A class as energy-measurement §2.7 reads one from a receipt's JSON members: None
|
|
for no class; (label, ppm, window, filtering, reference) otherwise, an on-chip
|
|
counter's absent filtering state read as Unknown. A label that cannot be read is a
|
|
malformed corpus (ValueError), not a verdict."""
|
|
label = m.get("energy_provenance")
|
|
u, f, r = m.get("energy_uncertainty"), m.get("energy_counter_filtering"), m.get("energy_calibration_ref")
|
|
if label is None:
|
|
if u is not None or f is not None or r is not None:
|
|
raise ValueError("evidence without a class")
|
|
return None
|
|
if label in ("OnChipCounter", "CalibratedInstrument"):
|
|
if not isinstance(u, dict):
|
|
raise ValueError("a refined class without its uncertainty")
|
|
ppm, win = u["relative_ppm"], u["window_us"]
|
|
if label == "OnChipCounter":
|
|
return (label, ppm, win, "Unknown" if f is None else f, None)
|
|
return (label, ppm, win, None, r)
|
|
if label not in STRONGEST:
|
|
raise ValueError("unknown class " + str(label))
|
|
return (label, None, None, None, None)
|
|
|
|
|
|
def class_of_body(b):
|
|
"""The class a claim's body restates, in class_of_json's form (step 4 has admitted it)."""
|
|
return class_of_json(b)
|
|
|
|
|
|
def figure_view(f):
|
|
"""A `figure` context as (profile, identity, group, {index: (joules, class)}, signer),
|
|
or None for a receipt that does not verify. A `claim` is an energy claim's wire form,
|
|
viewed under `moq-jwp:claim-link`; a `view` is given as the verifier holds it."""
|
|
if "claim" in f:
|
|
if not isinstance(f["claim"], str):
|
|
raise TypeError("a figure claim is given as its wire form")
|
|
try:
|
|
e = verify_claim(f["claim"])
|
|
except Reject:
|
|
return None
|
|
figs = {}
|
|
if e["class"] == "energy":
|
|
figs[None] = (e["body"]["joules_micro"], class_of_body(e["body"]))
|
|
return ("moq-jwp:claim-link", link_hash(e), None, figs, e["signer_pubkey"])
|
|
v = f["view"]
|
|
figs = {x["index"]: (x["joules_micro"], class_of_json(x)) for x in v["figures"]}
|
|
return (v["profile"], v["identity"], v["group"], figs, v["signer"])
|
|
|
|
|
|
def check_figure(c, view):
|
|
"""§2.4.8 / §2.4.9 against the figure, in order."""
|
|
b = c["body"]
|
|
need(c["class"] in ("measurement", "operating-point"), 6, "context_mismatch")
|
|
need(view is not None, 6, "context_invalid")
|
|
profile, identity, group, figs, signer = view
|
|
need(profile == b["figure_profile"] and identity == c["subject"] and group == b["group"]
|
|
and b["figure_index"] in figs, 6, "context_mismatch")
|
|
joules, cls = figs[b["figure_index"]]
|
|
need(joules == b["joules_micro"], 6, "figure_disagrees")
|
|
need(cls is not None, 6, "figure_unclassified")
|
|
need(cls == class_of_body(b), 6, "figure_class_disagrees")
|
|
if c["class"] == "measurement":
|
|
need(PROFILES[b["figure_profile"]][2], 6, "figure_signer_unknown")
|
|
need(signer == c["signer_pubkey"], 6, "figure_signer_mismatch")
|
|
|
|
|
|
def held_previous(ctx):
|
|
"""§3.1 step 5's previous claim: one that does not itself pass steps 1-4 is not held —
|
|
failing steps 1-3 it is not known to be its signer's, failing step 4 it breaks its
|
|
class's rules."""
|
|
if "previous" not in ctx:
|
|
return None
|
|
if not isinstance(ctx["previous"], str):
|
|
raise TypeError("a previous claim is given as its wire form")
|
|
try:
|
|
return verify_claim(ctx["previous"])
|
|
except Reject:
|
|
return None
|
|
|
|
|
|
def recover(ctx):
|
|
"""§3.2. Returns the ancestor; raises Reject."""
|
|
d = verify_claim(ctx["declaration"])
|
|
need(d["class"] == "soft-binding", 6, "context_mismatch")
|
|
need(d["signer_pubkey"] == ctx["publisher"], 6, "recovery_signer_mismatch")
|
|
need(ctx.get("observed_hash") != d["subject"], 6, "recovery_of_hard_bound_object")
|
|
match(d["body"]["alg"], d["body"]["value"], ctx["observed_value"], ctx.get("exact_match_algs", []))
|
|
return d["subject"]
|
|
|
|
|
|
def verdict(v):
|
|
"""This verifier's verdict on one vector: `accept link_hash=<hex>` for an accepted
|
|
claim, `accept ancestor=<hex>` for a recovery, or `reject step=<n> reason=<label>`."""
|
|
ctx = v.get("context", {})
|
|
try:
|
|
if v["procedure"] == "recover":
|
|
return "accept ancestor=" + recover(ctx)
|
|
c = verify_claim(v["receipt"])
|
|
check_chain(c, held_previous(ctx))
|
|
check_context(c, ctx)
|
|
return "accept link_hash=" + link_hash(c)
|
|
except Reject as r:
|
|
return f"reject step={r.step} reason={r.reason}"
|
|
|
|
|
|
def run_vector(v):
|
|
got = verdict(v)
|
|
if got.startswith("accept link_hash="):
|
|
# An accepted claim must also reproduce its pinned link hash, when one is pinned.
|
|
pinned = v["expected"].get("link_hash")
|
|
return "accept" if pinned in (None, got[len("accept link_hash="):]) else got
|
|
return got
|
|
|
|
|
|
def verdicts(path):
|
|
"""`name<TAB>verdict` for every vector in a corpus-format file, ignoring any declared
|
|
outcome: the input to a differential comparison with wai-rs (`differential.py`).
|
|
A vector this script cannot interpret is `error`; an exception of any other kind is
|
|
reported as `crash: <type>`, which a conforming run never prints."""
|
|
try:
|
|
doc = OER.load(open(path, encoding="utf-8"))
|
|
except ValueError as e:
|
|
print(f"{path}: not read: {e}", file=sys.stderr)
|
|
return 1
|
|
for v in doc["vectors"]:
|
|
try:
|
|
got = verdict(v)
|
|
except UnicodeError as e:
|
|
got = "crash: " + type(e).__name__
|
|
except (ValueError, KeyError, TypeError):
|
|
got = "error"
|
|
except Exception as e: # noqa: BLE001 -- reported, never swallowed
|
|
got = "crash: " + type(e).__name__
|
|
print(f"{v['name']}\t{got}")
|
|
return 0
|
|
|
|
|
|
def expected_of(e):
|
|
if e["outcome"] == "accept":
|
|
return "accept ancestor=" + e["ancestor"] if "ancestor" in e else "accept"
|
|
return f"reject step={e['step']} reason={e['reason']}"
|
|
|
|
|
|
# ------------------------------------------------------------------ canonical fixture
|
|
|
|
FIXTURE_SEED = bytes([0x22] * 32)
|
|
|
|
|
|
def fixture():
|
|
"""The canonical fixture wai-rs pins: computed here, from the members and the seed."""
|
|
sk = ed25519.Ed25519PrivateKey.from_private_bytes(FIXTURE_SEED)
|
|
pk = sk.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw).hex()
|
|
c = {
|
|
"signer_id": "did:example:météo\u001f",
|
|
"body": {"work_unit": "subpixel", "oer_receipt": None, "energy_calibration_ref": None, "joules_micro": 4096,
|
|
"grade": "measured", "energy_uncertainty": {"window_us": 1000000, "relative_ppm": 20000},
|
|
"method": "https://transaction.science/wai/extensions/energy-measurement",
|
|
"coverage": "Whole", "energy_counter_filtering": "Unknown", "work_units": 49152,
|
|
"energy_provenance": "OnChipCounter"},
|
|
"seq": 7, "parent_receipt_hash": None, "kind": "claim",
|
|
"subject": blake3(b"moq-jwp:object\x01" + b"claim-conformance: canonical fixture object").hex(),
|
|
"issued_at": "2026-09-27T12:00:00.5+02:00", "class": "energy", "subject_type": "object",
|
|
"signer_pubkey": pk,
|
|
}
|
|
c["sig"] = sk.sign(signing_input(c)).hex()
|
|
return c, jcs(c), link_hash(c)
|
|
|
|
|
|
def fixture_matches_wai_rs():
|
|
"""The literals `check_canonical_form` pins in wai-rs equal what this script computes."""
|
|
src_path = os.path.join(HERE, "..", "wai-rs", "src", "claim_receipt.rs")
|
|
try:
|
|
src = open(src_path, encoding="utf-8").read()
|
|
except OSError as e:
|
|
return [f"cannot read {src_path}: {e}"]
|
|
c, canon, link = fixture()
|
|
fails = []
|
|
m = re.search(r'const CANONICAL: &str = "((?:[^"\\]|\\.)*)";', src)
|
|
pinned = re.sub(r'\\(["\\])', r"\1", m.group(1)) if m else None
|
|
if pinned != canon:
|
|
fails.append("canonical bytes differ from wai-rs")
|
|
m = re.search(r'const LINK: &str = "([0-9a-f]{64})";', src)
|
|
if not m or m.group(1) != link:
|
|
fails.append("link hash differs from wai-rs")
|
|
if f'"sig": "{c["sig"]}"' not in src:
|
|
fails.append("signature differs from wai-rs")
|
|
return fails
|
|
|
|
|
|
def bracket_fixture():
|
|
"""The brackets wai-rs pins in `energy_report::pinned_brackets`, read from its source,
|
|
with the figure and least declaration this script computes for each:
|
|
[(name, values, F, p)]."""
|
|
src_path = os.path.join(HERE, "..", "wai-rs", "src", "energy_report.rs")
|
|
src = open(src_path, encoding="utf-8").read()
|
|
body = src[src.index("fn pinned_brackets()"):]
|
|
body = body[:body.index("\n}\n")]
|
|
out = []
|
|
num = r"(x|[0-9_]+)"
|
|
pat = (r'\("(\w+)", v\(s\(' + num + r', ' + num + r', ' + num + r'\), s\(' + num + r', ' + num + r', ' + num
|
|
+ r'\), ' + num + r', ' + num + r', ' + num + r', (FromRest|Running)\), ' + num + r', ' + num + r'\)')
|
|
for m in re.finditer(pat, body):
|
|
g = [MAX_INT if x == "x" else int(x.replace("_", "")) if x not in ("FromRest", "Running") else x for x in m.groups()[1:]]
|
|
br = {"idle": {"n": g[0], "mean_uw": g[1], "sd_uw": g[2]}, "busy": {"n": g[3], "mean_uw": g[4], "sd_uw": g[5]},
|
|
"window_us": g[6], "completions": g[7], "in_flight": g[8],
|
|
"opening": "from-rest" if g[9] == "FromRest" else "running"}
|
|
out.append((m.group(1), br, g[10], g[11]))
|
|
return out
|
|
|
|
|
|
def brackets_match_wai_rs():
|
|
"""The figures and least declarations wai-rs pins equal what this script computes, and
|
|
one ppm less is not supported."""
|
|
fails, cases = [], bracket_fixture()
|
|
if len(cases) < 7:
|
|
return ["fewer pinned brackets than expected were read from energy_report.rs"]
|
|
for name, br, f, p in cases:
|
|
got_f = bracket_figure(br)
|
|
got_p = least_ppm(br, f) if got_f is not None else None
|
|
if got_f != f or got_p != p:
|
|
fails.append(f"{name}: computed F={got_f} p={got_p}, pinned F={f} p={p}")
|
|
return fails
|
|
|
|
|
|
def main(argv):
|
|
if argv[1:2] == ["brackets"]:
|
|
for name, br, _, _ in bracket_fixture():
|
|
f = bracket_figure(br)
|
|
print(name, f, least_ppm(br, f))
|
|
return 0
|
|
if argv[1:2] == ["verdicts"] and len(argv) == 3:
|
|
return verdicts(argv[2])
|
|
if argv[1:2] == ["fixture"]:
|
|
c, canon, link = fixture()
|
|
print("sig ", c["sig"])
|
|
print("canonical", canon)
|
|
print("link_hash", link)
|
|
return 0
|
|
fail = 0
|
|
# BLAKE3's published values for the empty input and "abc", before anything relies on it.
|
|
b3 = blake3(b"").hex() == "af1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262" and \
|
|
blake3(b"abc").hex() == "6437b3ac38465133ffb63b75273a8db548c558465d79db03fd359c6cd5bd9d85"
|
|
print(("PASS" if b3 else "FAIL") + " BLAKE3 reproduces its published values")
|
|
fail += not b3
|
|
ff = fixture_matches_wai_rs()
|
|
print(("PASS" if not ff else "FAIL") + " canonical fixture equals the literals pinned in wai-rs " + ",".join(ff))
|
|
fail += bool(ff)
|
|
bm = brackets_match_wai_rs()
|
|
print(("PASS" if not bm else "FAIL") + " bracket arithmetic equals the literals pinned in wai-rs " + ",".join(bm))
|
|
fail += bool(bm)
|
|
rd = OER.check_canonical_form() # the OER reference's own fixtures, its reading of -0 among them
|
|
print(("PASS" if not rd else "FAIL") + " the OER reference's canonical form and number reading hold " + ",".join(rd))
|
|
fail += bool(rd)
|
|
path = argv[1] if len(argv) > 1 else os.path.join(HERE, "vectors.json")
|
|
doc = OER.load(open(path, encoding="utf-8"))
|
|
for v in doc["vectors"]:
|
|
want, got = expected_of(v["expected"]), run_vector(v)
|
|
ok = want == got
|
|
fail += not ok
|
|
print(f"{'PASS' if ok else 'FAIL'} {v['name']} expected={want} got={got}")
|
|
print(f"{len(doc['vectors'])} vectors, {fail} failure(s)")
|
|
return 1 if fail else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|