764 lines
32 KiB
Python
764 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the browser-sink corpus (extensions/browser-sink.md).
|
|
|
|
Written from the extension's text. It shares no code with wai-rs, which generates the
|
|
corpus, and never runs it. It reads from the texts themselves the tables it works from:
|
|
the user agent's decoders (§4.1), the WebCodecs Codec Registry's rows (§4.2), the reasons
|
|
(§11), and the parameter register of capability-constraints §3.1. A capability's kind,
|
|
status and replicate class come from the registry export, registry/capabilities.json. It
|
|
hashes with ../claim-conformance/_blake3.py and checks Ed25519 signatures with
|
|
../claim-conformance/_ed25519.py, WAI's strict rule (SPEC §7 *Signatures*), both written
|
|
from their specifications with the standard library.
|
|
|
|
For every case of vectors.json it recomputes, from the text:
|
|
|
|
- `routes`: the report read or refused (§3), and each object's route, reason, legs,
|
|
`asked`, tier and energy entry (§4, §6);
|
|
- `constraints`: §5's outcome and reason;
|
|
- `bindings`: §7.1's reading and canonical form; `frame_tiers`: §7.2's tier;
|
|
- `chain`: each group receipt's signature (jwp-receipts §2.3, legacy form), its root and
|
|
each object's inclusion (§2.2), its canonical bytes and link hash (§3 step 3), the
|
|
second group's link to the first, whether each object verifies, and the binding a page
|
|
writes for it on the wasm route;
|
|
- `frames`: §10's state; `tables`: §9.2's side table over each sequence.
|
|
|
|
It requires every reason of §11 to appear in the corpus and §11 to list exactly the codes
|
|
it knows. Then it mutates copies of the corpus, one defect each, and requires every copy to
|
|
fail; and it changes the answers the corpus gives for codec strings no registry row lists,
|
|
and requires the corpus still to pass, since those answers are never read (§4.2).
|
|
|
|
python3 -I browser_sink_verify.py <live-conformance/browser>
|
|
|
|
Exit status 0 only if every check holds and every mutated copy fails. Python 3.8 or later;
|
|
standard library only.
|
|
"""
|
|
import copy
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
WAI = os.path.normpath(os.path.join(HERE, ".."))
|
|
sys.path.insert(0, os.path.join(WAI, "claim-conformance"))
|
|
import _blake3 as B3 # noqa: E402
|
|
import _ed25519 as ED # noqa: E402
|
|
|
|
TIERS = ["none", "entropy-consistency", "decode-equivalence"]
|
|
ALG = "jwp-object-1"
|
|
MAX_SAFE = 2 ** 53 - 1
|
|
CLASSES = ["CalibratedInstrument", "OnChipCounter", "ModelBased", "Estimator"]
|
|
# The registry export's labels of the replicate classes §2's table names.
|
|
CLASS_LABEL = {"StandardDefined": "standard-defined", "NeuralIntegerExact": "neural-int-exact",
|
|
"NeuralEntropyExact": "neural-entropy-exact"}
|
|
|
|
|
|
def read(p):
|
|
with open(p, encoding="utf-8") as f:
|
|
return f.read()
|
|
|
|
|
|
def section(text, heading, stop):
|
|
i = text.index(heading)
|
|
j = text.index(stop, i + len(heading))
|
|
return text[i:j]
|
|
|
|
|
|
def ticks(s):
|
|
return re.findall(r"`([^`]*)`", s)
|
|
|
|
|
|
# ───────────────────────────── the texts' tables ─────────────────────────────
|
|
|
|
SPEC = read(os.path.join(WAI, "extensions", "browser-sink.md"))
|
|
CC = read(os.path.join(WAI, "extensions", "capability-constraints.md"))
|
|
|
|
|
|
def ua_rows():
|
|
rows = []
|
|
for line in section(SPEC, "### 4.1 ", "### 4.2 ").splitlines():
|
|
if not line.startswith("| `"):
|
|
continue
|
|
cells = [c.strip() for c in line.strip().strip("|").split("|")]
|
|
caps, decoder, forms, tier = ticks(cells[0]), ticks(cells[1])[0], ticks(cells[2]), ticks(cells[3])[0]
|
|
rows.append({"caps": caps, "decoder": decoder, "forms": forms, "prefix": "and more" in cells[2], "tier": tier})
|
|
assert rows, "§4.1 has no rows"
|
|
return rows
|
|
|
|
|
|
def registry_rows():
|
|
rows = []
|
|
for line in section(SPEC, "### 4.2 ", "### 4.3 ").splitlines():
|
|
if line.startswith("| video |") or line.startswith("| audio |"):
|
|
rows += ticks(line)
|
|
assert rows, "§4.2 has no rows"
|
|
return rows
|
|
|
|
|
|
def reasons():
|
|
out = []
|
|
for line in section(SPEC, "## 11. Reasons", "## 12. ").splitlines():
|
|
if line.startswith("| `"):
|
|
out.append(ticks(line)[0])
|
|
return out
|
|
|
|
|
|
def register():
|
|
return [ticks(line)[0] for line in section(CC, "### 3.1 ", "\n## 4.").splitlines() if line.startswith("| `")]
|
|
|
|
|
|
def tier_by_class():
|
|
out = {}
|
|
for line in section(SPEC, "**Registered tier**", "`StandardDefined` is read").splitlines():
|
|
line = line.strip()
|
|
if line.startswith("| `") and "every other class" not in line:
|
|
cells = line.strip().strip("|").split("|")
|
|
for c in ticks(cells[0]):
|
|
out[CLASS_LABEL[c]] = ticks(cells[1])[0]
|
|
return out
|
|
|
|
|
|
UA_ROWS = ua_rows()
|
|
REGISTRY_ROWS = registry_rows()
|
|
REASONS = reasons()
|
|
REGISTER = register()
|
|
TIER_BY_CLASS = tier_by_class()
|
|
CAPS = {c["capability"]: c for c in json.loads(read(os.path.join(WAI, "registry", "capabilities.json")))["capabilities"]}
|
|
|
|
|
|
def rank(t):
|
|
return TIERS.index(t) if t in TIERS else 0
|
|
|
|
|
|
def read_tier(t):
|
|
return t if t in TIERS else "none"
|
|
|
|
|
|
def registered_tier(cap):
|
|
c = CAPS.get(cap)
|
|
return TIER_BY_CLASS.get(c["replicate"], "none") if c else "none"
|
|
|
|
|
|
def dispatchable(cap):
|
|
c = CAPS.get(cap)
|
|
if c is None:
|
|
return True # a string this revision does not register (SPEC §4)
|
|
return c["kind"] == "payload" and c["status"] != "candidate"
|
|
|
|
|
|
def row_matches(row, s):
|
|
if row.endswith("*"):
|
|
p = row[:-1]
|
|
return len(s) > len(p) and s.startswith(p)
|
|
return s == row
|
|
|
|
|
|
def gate(codec):
|
|
return any(row_matches(r, codec) for r in REGISTRY_ROWS)
|
|
|
|
|
|
def ua_row(cap):
|
|
for r in UA_ROWS:
|
|
if cap in r["caps"]:
|
|
return r
|
|
return None
|
|
|
|
|
|
def form_ok(row, codec):
|
|
for f in row["forms"]:
|
|
if row["prefix"] and f.endswith("."):
|
|
if len(codec) > len(f) and codec.startswith(f):
|
|
return True
|
|
elif codec == f:
|
|
return True
|
|
return False
|
|
|
|
|
|
# ─────────────────────────────── §5 ───────────────────────────────
|
|
|
|
def integer(v):
|
|
return v if isinstance(v, int) and not isinstance(v, bool) and -MAX_SAFE <= v <= MAX_SAFE else None
|
|
|
|
|
|
def is_value(v):
|
|
return isinstance(v, str) or integer(v) is not None
|
|
|
|
|
|
def constraint_ok(c):
|
|
if not isinstance(c, dict) or not c or any(k not in ("enum", "minimum", "maximum") for k in c):
|
|
return False
|
|
if "enum" in c and not (isinstance(c["enum"], list) and c["enum"] and all(is_value(x) for x in c["enum"])):
|
|
return False
|
|
for k in ("minimum", "maximum"):
|
|
if k in c and integer(c[k]) is None:
|
|
return False
|
|
if "minimum" in c and "maximum" in c and c["minimum"] > c["maximum"]:
|
|
return False
|
|
return True
|
|
|
|
|
|
def meets(c, v):
|
|
if v is None or not is_value(v):
|
|
return False
|
|
if "enum" in c:
|
|
hit = False
|
|
for x in c["enum"]:
|
|
if isinstance(x, str) and isinstance(v, str) and x == v:
|
|
hit = True
|
|
if integer(x) is not None and integer(v) is not None and x == v:
|
|
hit = True
|
|
if not hit:
|
|
return False
|
|
if "minimum" in c and (integer(v) is None or v < c["minimum"]):
|
|
return False
|
|
if "maximum" in c and (integer(v) is None or v > c["maximum"]):
|
|
return False
|
|
return True
|
|
|
|
|
|
def evaluate(sets, params):
|
|
"""(outcome, reason)."""
|
|
if sets is None:
|
|
return "unconstrained", None
|
|
if not isinstance(sets, list) or not sets:
|
|
return "unevaluable", "constraint_malformed"
|
|
for s in sets:
|
|
if not isinstance(s, dict) or not all(constraint_ok(c) for c in s.values()):
|
|
return "unevaluable", "constraint_malformed"
|
|
for s in sets:
|
|
if any(k not in REGISTER for k in s):
|
|
return "unevaluable", "parameter_unregistered"
|
|
params = params if isinstance(params, dict) else {}
|
|
for s in sets:
|
|
if all(meets(c, params.get(k)) for k, c in s.items()):
|
|
return "satisfied", None
|
|
return "unsatisfied", "constraints_unmet"
|
|
|
|
|
|
# ─────────────────────────────── §3, §4, §6 ───────────────────────────────
|
|
|
|
def is_bool_or_null(v):
|
|
return v is None or isinstance(v, bool)
|
|
|
|
|
|
def report_reads(r):
|
|
if not isinstance(r, dict):
|
|
return False
|
|
v = r.get("browser_sink")
|
|
if isinstance(v, bool) or v != 1 or not isinstance(v, int):
|
|
return False
|
|
ua = r.get("ua")
|
|
if not isinstance(ua, dict):
|
|
return False
|
|
for a in ua.values():
|
|
if not (isinstance(a, dict) and isinstance(a.get("supported"), bool) and "smooth" in a and is_bool_or_null(a["smooth"])
|
|
and "power_efficient" in a and is_bool_or_null(a["power_efficient"]) and a.get("probe") in ("match", "mismatch", "not_run")):
|
|
return False
|
|
if not (isinstance(r.get("host"), dict) and all(isinstance(x, str) for x in r["host"].values())):
|
|
return False
|
|
if not (isinstance(r.get("wasm"), list) and all(isinstance(x, str) for x in r["wasm"])):
|
|
return False
|
|
if not isinstance(r.get("constraints"), dict):
|
|
return False
|
|
if not (isinstance(r.get("energy"), dict) and all(isinstance(x, dict) for x in r["energy"].values())):
|
|
return False
|
|
objs = r.get("objects")
|
|
if not isinstance(objs, list):
|
|
return False
|
|
for o in objs:
|
|
if not (isinstance(o, dict) and isinstance(o.get("name"), str) and isinstance(o.get("capability"), str)
|
|
and "codec" in o and (o["codec"] is None or isinstance(o["codec"], str))
|
|
and isinstance(o.get("need"), str) and isinstance(o.get("params"), dict)):
|
|
return False
|
|
return True
|
|
|
|
|
|
def positive(v):
|
|
i = integer(v)
|
|
return i is not None and i > 0
|
|
|
|
|
|
def read_measurement(m):
|
|
cls = m.get("class")
|
|
if cls not in CLASSES:
|
|
return None, "energy_class_invalid"
|
|
if integer(m.get("joules_micro")) == 0:
|
|
return None, "unmetered_as_zero"
|
|
if not all(positive(m.get(k)) for k in ("joules_micro", "decodes", "window_us")):
|
|
return None, "energy_count_invalid"
|
|
if cls in ("CalibratedInstrument", "OnChipCounter") and not positive(m.get("relative_ppm")):
|
|
return None, "energy_evidence_missing"
|
|
if cls == "OnChipCounter" and m.get("filtering") not in ("On", "Off", "Unknown"):
|
|
return None, "energy_evidence_missing"
|
|
if cls == "CalibratedInstrument" and not (isinstance(m.get("calibration_ref"), str) and m["calibration_ref"]):
|
|
return None, "energy_evidence_missing"
|
|
if m.get("coverage") not in ("Whole", "Partial"):
|
|
return None, "energy_evidence_missing"
|
|
f = m.get("fabrics")
|
|
if not (isinstance(f, list) and f and all(isinstance(x, str) for x in f) and len(set(f)) == len(f)):
|
|
return None, "energy_evidence_missing"
|
|
out = dict(m)
|
|
out["per_decode_micro"] = m["joules_micro"] // m["decodes"]
|
|
return out, None
|
|
|
|
|
|
def energy_entry(r, cap, codec, route):
|
|
pe = None
|
|
if route == "ua" and codec in r["ua"]:
|
|
pe = r["ua"][codec]["power_efficient"]
|
|
figure, reason = None, None
|
|
if route != "none":
|
|
m = r["energy"].get(cap + " " + route)
|
|
if m is not None:
|
|
figure, reason = read_measurement(m)
|
|
return {"power_efficient": pe, "figure": figure, "reason": reason}
|
|
|
|
|
|
def route_object(r, o):
|
|
cap, codec, need = o["capability"], o["codec"], read_tier(o["need"])
|
|
reg = registered_tier(cap)
|
|
nr = "not_reached"
|
|
|
|
def decided(reason):
|
|
return {"name": o["name"], "route": "none", "reason": reason, "legs": {"ua": nr, "host": nr, "wasm": nr},
|
|
"asked": False, "tier": "none", "energy": energy_entry(r, cap, codec, "none")}
|
|
|
|
if not dispatchable(cap):
|
|
return decided("not_dispatchable")
|
|
if rank(need) > rank(reg):
|
|
return decided("tier_exceeds_class")
|
|
outcome, _ = evaluate(r["constraints"].get(cap), o["params"])
|
|
if outcome == "unevaluable":
|
|
return decided("constraints_unevaluable")
|
|
if outcome == "unsatisfied":
|
|
return decided("constraints_unmet")
|
|
|
|
asked = False
|
|
ua, ua_tier = None, None
|
|
row = ua_row(cap)
|
|
if codec is None:
|
|
ua = "no_ua_mapping"
|
|
elif not gate(codec):
|
|
ua = "codec_unregistered"
|
|
elif row is None:
|
|
ua = "no_ua_mapping"
|
|
elif not form_ok(row, codec):
|
|
ua = "codec_mismatch"
|
|
else:
|
|
asked = True
|
|
a = r["ua"].get(codec)
|
|
if a is None:
|
|
ua = "ua_not_asked"
|
|
elif a["supported"] is not True:
|
|
ua = "ua_unsupported"
|
|
elif rank(need) > rank(row["tier"]):
|
|
ua = "ua_inexact"
|
|
elif need == "decode-equivalence" and a["probe"] != "match":
|
|
ua = "ua_probe_mismatch" if a["probe"] == "mismatch" else "ua_unprobed"
|
|
else:
|
|
t = row["tier"] if a["probe"] == "match" else "none"
|
|
ua_tier = TIERS[min(rank(t), rank(reg))]
|
|
host, host_tier = None, None
|
|
if cap not in r["host"]:
|
|
host = "no_host"
|
|
elif rank(read_tier(r["host"][cap])) < rank(need):
|
|
host = "host_inexact"
|
|
else:
|
|
host_tier = TIERS[min(rank(read_tier(r["host"][cap])), rank(reg))]
|
|
wasm, wasm_tier = None, None
|
|
if cap in r["wasm"]:
|
|
wasm_tier = reg
|
|
else:
|
|
wasm = "no_wasm"
|
|
|
|
if ua_tier is not None:
|
|
route, tier, legs = "ua", ua_tier, {"ua": "taken", "host": nr, "wasm": nr}
|
|
elif host_tier is not None:
|
|
route, tier, legs = "host", host_tier, {"ua": ua, "host": "taken", "wasm": nr}
|
|
elif wasm_tier is not None:
|
|
route, tier, legs = "wasm", wasm_tier, {"ua": ua, "host": host, "wasm": "taken"}
|
|
else:
|
|
route, tier, legs = "none", "none", {"ua": ua, "host": host, "wasm": wasm}
|
|
return {"name": o["name"], "route": route, "reason": "no_decoder" if route == "none" else None, "legs": legs,
|
|
"asked": asked, "tier": tier, "energy": energy_entry(r, cap, codec, route)}
|
|
|
|
|
|
def route_report(r):
|
|
if not report_reads(r):
|
|
return None, "report_malformed"
|
|
return [route_object(r, o) for o in r["objects"]], None
|
|
|
|
|
|
# ─────────────────────────────── §7 ───────────────────────────────
|
|
|
|
def hex32(s):
|
|
return isinstance(s, str) and re.fullmatch(r"[0-9a-f]{64}", s) is not None
|
|
|
|
|
|
def read_binding(v):
|
|
"""(binding dict in member order, None) or (None, reason)."""
|
|
if not isinstance(v, dict):
|
|
return None, "binding_malformed"
|
|
for k in ("alg", "digest", "tier"):
|
|
if not isinstance(v.get(k), str):
|
|
return None, "binding_malformed"
|
|
if not hex32(v["digest"]):
|
|
return None, "binding_malformed"
|
|
if "receipt" in v and not hex32(v["receipt"]):
|
|
return None, "binding_malformed"
|
|
if v["alg"] != ALG:
|
|
return None, "binding_alg_unknown"
|
|
b = {"alg": ALG, "digest": v["digest"]}
|
|
if "receipt" in v:
|
|
b["receipt"] = v["receipt"]
|
|
b["tier"] = read_tier(v["tier"])
|
|
return b, None
|
|
|
|
|
|
def canonical(b):
|
|
return json.dumps(b, separators=(",", ":"), ensure_ascii=False)
|
|
|
|
|
|
def frame_tier(route, route_tier, format_null, altered):
|
|
if altered or (route == "ua" and format_null):
|
|
return "none"
|
|
return route_tier
|
|
|
|
|
|
def frame_state(binding, obj, chain, prov):
|
|
b, reason = (None, None) if binding is None else read_binding(binding)
|
|
if b is None:
|
|
return {"state": "unbound", "reason": reason, "tier": None, "digest": None, "receipt": None}
|
|
if "refused" in (obj, chain, prov):
|
|
s = "refused"
|
|
elif obj == "unverified" or chain == "held" or prov == "missing":
|
|
s = "pending"
|
|
else:
|
|
s = "verified"
|
|
return {"state": s, "reason": None, "tier": b["tier"], "digest": b["digest"], "receipt": b.get("receipt")}
|
|
|
|
|
|
# ─────────────────────────────── §9.2 ───────────────────────────────
|
|
|
|
def run_table(bound, ops):
|
|
pending, frames, refusals, struck = {}, {}, [], 0
|
|
STRUCK = object()
|
|
for i, op in enumerate(ops):
|
|
k = op["op"]
|
|
if k == "chunk":
|
|
q = pending.setdefault(op["decoder"], [])
|
|
b = None
|
|
if op.get("binding") is not None:
|
|
b, why = read_binding(op["binding"])
|
|
if b is None:
|
|
raise ValueError("op %d: binding %s" % (i, why))
|
|
hit = [e for e in q if e[0] == op["timestamp"]]
|
|
if hit:
|
|
if hit[0][1] is not STRUCK:
|
|
refusals.append({"at": i, "reason": "binding_ambiguous"})
|
|
hit[0][1] = STRUCK
|
|
else:
|
|
q.append([op["timestamp"], b])
|
|
if len(q) > bound:
|
|
q.pop(0)
|
|
struck += 1
|
|
elif k == "output":
|
|
q = pending.setdefault(op["decoder"], [])
|
|
b = None
|
|
for j, e in enumerate(q):
|
|
if e[0] == op["timestamp"]:
|
|
q.pop(j)
|
|
b = None if e[1] is STRUCK else e[1]
|
|
break
|
|
frames[op["frame"]] = b
|
|
elif k == "clone":
|
|
frames[op["frame"]] = frames[op["from"]]
|
|
elif k == "transfer":
|
|
b = frames.pop(op["from"])
|
|
frames[op["frame"]] = b if op.get("with_binding") is True else None
|
|
elif k == "close":
|
|
frames.pop(op["frame"], None)
|
|
elif k in ("reset", "flush"):
|
|
struck += len(pending.pop(op["decoder"], []))
|
|
else:
|
|
raise ValueError("op %d: %r" % (i, k))
|
|
return {"frames": {f: (canonical(b) if b is not None else None) for f, b in frames.items()},
|
|
"refusals": refusals, "struck": struck}
|
|
|
|
|
|
# ─────────────────────────────── the chain ───────────────────────────────
|
|
|
|
_HASHES = {}
|
|
|
|
|
|
def b3(data):
|
|
data = bytes(data)
|
|
h = _HASHES.get(data)
|
|
if h is None:
|
|
h = _HASHES[data] = B3.blake3(data)
|
|
return h
|
|
|
|
|
|
def u64(n):
|
|
return n.to_bytes(8, "big")
|
|
|
|
|
|
def lp(s):
|
|
b = s.encode("utf-8")
|
|
return len(b).to_bytes(4, "big") + b
|
|
|
|
|
|
def check_chain(chain, load, fails):
|
|
prev_link = None
|
|
for gi, g in enumerate(chain["groups"]):
|
|
r = g["receipt"]
|
|
pk, sig = bytes.fromhex(r["signer_pubkey"]), bytes.fromhex(r["sig"])
|
|
payload = (b"moq-jwp:group-sign\x01" + u64(r["track_alias"]) + u64(r["group_id"]) + bytes.fromhex(r["root_hash"])
|
|
+ u64(r["leaf_count"]) + u64(r["joules_total"]) + lp(r["group_start"]) + lp(r["group_end"]) + pk + lp(r["signer_id"]))
|
|
if ED.verify(pk, payload, sig) is not None:
|
|
fails.append("chain group %d: the signature does not verify" % gi)
|
|
if r["signer_pubkey"] != chain["signer_pubkey"]:
|
|
fails.append("chain group %d: another key" % gi)
|
|
can = payload + sig
|
|
if can.hex() != g["canonical"]:
|
|
fails.append("chain group %d: canonical bytes" % gi)
|
|
link = b3(b"moq-jwp:receipt-link\x01" + can)
|
|
if link.hex() != g["link_hash"]:
|
|
fails.append("chain group %d: link hash" % gi)
|
|
want_parent = prev_link.hex() if prev_link is not None else None
|
|
if r["parent_receipt_hash"] != want_parent:
|
|
fails.append("chain group %d: parent link" % gi)
|
|
prev_link = link
|
|
for o in g["objects"]:
|
|
env = load(o["envelope"]["file"]) if isinstance(o["envelope"], dict) else bytes.fromhex(o["envelope"])
|
|
orc = o["receipt"]
|
|
ch = b3(b"moq-jwp:object\x01" + env)
|
|
ok = ch.hex() == orc["content_hash"]
|
|
if (orc["track_alias"], orc["group_id"]) != (r["track_alias"], r["group_id"]) or orc["merkle_proof"]["leaf_index"] != orc["object_id"]:
|
|
ok = False
|
|
h = b3(b"moq-jwp:merkle-leaf\x01" + bytes.fromhex(orc["content_hash"]))
|
|
bits = orc["merkle_proof"]["direction_bits"]
|
|
for i, s in enumerate(orc["merkle_proof"]["siblings"]):
|
|
s = bytes.fromhex(s)
|
|
h = b3(b"moq-jwp:merkle-node\x01" + (s + h if (bits >> i) & 1 else h + s))
|
|
if h.hex() != r["root_hash"]:
|
|
ok = False
|
|
want = "verified" if ok else "refused"
|
|
if o["outcome"] != want:
|
|
fails.append("chain %s: %s, the corpus says %s" % (o["name"], want, o["outcome"]))
|
|
if ok:
|
|
tier = frame_tier(chain["route"], registered_tier(chain["capability"]), False, False)
|
|
b = canonical({"alg": ALG, "digest": ch.hex(), "receipt": link.hex(), "tier": tier})
|
|
if o["binding"] != b:
|
|
fails.append("chain %s: binding %s" % (o["name"], o["binding"]))
|
|
elif o["binding"] is not None:
|
|
fails.append("chain %s: a binding for an object that does not verify" % o["name"])
|
|
|
|
|
|
# ─────────────────────────────── the corpus ───────────────────────────────
|
|
|
|
def verify(doc, load):
|
|
fails = []
|
|
seen = set()
|
|
|
|
def take(x):
|
|
if isinstance(x, str):
|
|
seen.add(x)
|
|
|
|
for c in doc["routes"]:
|
|
table, reason = route_report(c["report"])
|
|
if reason is not None:
|
|
if c["outcome"] != "refused" or c["reason"] != reason or c["table"] is not None:
|
|
fails.append("route %s: refused %s" % (c["name"], reason))
|
|
elif c["outcome"] != "routed" or c["reason"] is not None or c["table"] != table:
|
|
got = {e["name"]: (e["route"], e["reason"]) for e in table}
|
|
fails.append("route %s: %s" % (c["name"], json.dumps(got)[:300]))
|
|
take(c["reason"])
|
|
for e in c["table"] or []:
|
|
take(e["reason"])
|
|
take(e["energy"]["reason"])
|
|
for leg in e["legs"].values():
|
|
take(leg)
|
|
for c in doc["constraints"]:
|
|
o, r = evaluate(c["sets"], c["params"])
|
|
if (o, r) != (c["outcome"], c["reason"]):
|
|
fails.append("constraints %s: %s %s" % (c["name"], o, r))
|
|
take(r)
|
|
for c in doc["bindings"]:
|
|
b, r = read_binding(c["value"])
|
|
got = ("read", None, canonical(b), b["tier"]) if b else ("refused", r, None, None)
|
|
if got != (c["outcome"], c["reason"], c["canonical"], c["tier"]):
|
|
fails.append("binding %s: %s" % (c["name"], got))
|
|
take(r)
|
|
for c in doc["frame_tiers"]:
|
|
if frame_tier(c["route"], c["route_tier"], c["format_null"], c["altered"]) != c["tier"]:
|
|
fails.append("frame tier %s" % c)
|
|
check_chain(doc["chain"], load, fails)
|
|
for g in doc["chain"]["groups"]:
|
|
for o in g["objects"]:
|
|
if o["outcome"] == "refused":
|
|
seen.add("binding_unverified_reference")
|
|
for c in doc["frames"]:
|
|
s = frame_state(c["binding"], c["object"], c["chain"], c["provenance"])
|
|
if s != c["result"]:
|
|
fails.append("frame %s: %s" % (c["name"], s))
|
|
for c in doc["tables"]:
|
|
try:
|
|
r = run_table(c["bound"], c["ops"])
|
|
except (KeyError, ValueError) as e:
|
|
fails.append("table %s: %s" % (c["name"], e))
|
|
continue
|
|
if r != c["result"]:
|
|
fails.append("table %s: %s" % (c["name"], r))
|
|
for x in r["refusals"]:
|
|
take(x["reason"])
|
|
for code in REASONS:
|
|
if code not in seen:
|
|
fails.append("nothing in the corpus gives %s" % code)
|
|
return fails
|
|
|
|
|
|
KNOWN = ["report_malformed", "not_dispatchable", "tier_exceeds_class", "constraints_unevaluable", "constraints_unmet",
|
|
"no_ua_mapping", "codec_mismatch", "codec_unregistered", "ua_not_asked", "ua_unsupported", "ua_inexact",
|
|
"ua_unprobed", "ua_probe_mismatch", "no_host", "host_inexact", "no_wasm", "no_decoder", "constraint_malformed",
|
|
"parameter_unregistered", "energy_class_invalid", "unmetered_as_zero", "energy_count_invalid",
|
|
"energy_evidence_missing", "binding_malformed", "binding_alg_unknown", "binding_unverified_reference",
|
|
"binding_ambiguous"]
|
|
|
|
|
|
def find(doc, family, name):
|
|
for c in doc[family]:
|
|
if c["name"] == name:
|
|
return c
|
|
raise KeyError(name)
|
|
|
|
|
|
def entry(doc, case, obj):
|
|
for e in find(doc, "routes", case)["table"]:
|
|
if e["name"] == obj:
|
|
return e
|
|
raise KeyError(obj)
|
|
|
|
|
|
def mutations(doc):
|
|
"""(name, change to a copy of the document, change to the files or None)."""
|
|
def m(f):
|
|
return f
|
|
|
|
def flip_hex(s, i=0):
|
|
return s[:i] + ("0" if s[i] != "0" else "1") + s[i + 1:]
|
|
|
|
out = []
|
|
out.append(("a route changed", m(lambda d: entry(d, "room_tracks", "wai.video.vp9").update(route="wasm")), None))
|
|
out.append(("a supported answer withdrawn", m(lambda d: find(d, "routes", "room_tracks")["report"]["ua"]["vp09.00.10.08"].update(supported=False)), None))
|
|
out.append(("a probe matched", m(lambda d: find(d, "routes", "need_decode_equivalence")["report"]["ua"]["vp09.00.10.08"].update(probe="match")), None))
|
|
out.append(("the registry gate ignored", m(lambda d: entry(d, "unregistered_codecs", "av2").update(asked=True)), None))
|
|
out.append(("a leg's reason changed", m(lambda d: entry(d, "fallbacks", "int_motion")["legs"].update(host="no_host")), None))
|
|
out.append(("a figure's floor changed", m(lambda d: entry(d, "room_tracks", "wai.video.vp9")["energy"]["figure"].update(per_decode_micro=3051)), None))
|
|
out.append(("powerEfficient stated off the ua route", m(lambda d: entry(d, "energy", "estimator_wasm")["energy"].update(power_efficient=True)), None))
|
|
out.append(("powerEfficient inverted", m(lambda d: find(d, "routes", "room_tracks")["report"]["ua"]["vp09.00.10.08"].update(power_efficient=False)), None))
|
|
out.append(("a zero read as a figure", m(lambda d: entry(d, "energy", "zero_with_class")["energy"].update(reason=None)), None))
|
|
out.append(("a constraint bound moved", m(lambda d: find(d, "constraints", "bounds_inclusive")["sets"][0]["wai:frame_height"].update(minimum=1081)), None))
|
|
out.append(("an unregistered parameter registered", m(lambda d: find(d, "constraints", "unregistered").update(outcome="unsatisfied", reason="constraints_unmet")), None))
|
|
out.append(("a digest upper-cased", m(lambda d: find(d, "bindings", "canonical")["value"].update(digest="1F" * 32)), None))
|
|
out.append(("an unknown tier kept", m(lambda d: find(d, "bindings", "unknown_tier").update(tier="bit-exact")), None))
|
|
out.append(("a frame tier kept through an alteration", m(lambda d: [c.update(tier=c["route_tier"]) for c in d["frame_tiers"] if c["altered"]]), None))
|
|
out.append(("a group signature changed", m(lambda d: d["chain"]["groups"][0]["receipt"].update(sig=flip_hex(d["chain"]["groups"][0]["receipt"]["sig"], 3))), None))
|
|
out.append(("an object's bytes changed", m(lambda d: d["chain"]["groups"][0]["objects"].__setitem__(1, dict(d["chain"]["groups"][0]["objects"][1], envelope=flip_hex(d["chain"]["groups"][0]["objects"][1]["envelope"], 40)))), None))
|
|
out.append(("a link hash changed", m(lambda d: d["chain"]["groups"][0].update(link_hash=flip_hex(d["chain"]["groups"][0]["link_hash"]))), None))
|
|
out.append(("a binding names another receipt", m(lambda d: d["chain"]["groups"][1]["objects"][0].update(binding=d["chain"]["groups"][0]["objects"][0]["binding"])), None))
|
|
out.append(("a changed object bound", m(lambda d: d["chain"]["groups"][0]["objects"][2].update(binding=d["chain"]["groups"][0]["objects"][1]["binding"], outcome="verified")), None))
|
|
out.append(("the second group unlinked", m(lambda d: d["chain"]["groups"][1]["receipt"].update(parent_receipt_hash=None)), None))
|
|
|
|
def forged(d):
|
|
# The last group's signature changed, and everything derived from it made to agree:
|
|
# its canonical bytes, its link hash, and the binding of its object. Only the
|
|
# signature check can catch it.
|
|
g = d["chain"]["groups"][-1]
|
|
sig = flip_hex(g["receipt"]["sig"], 5)
|
|
g["receipt"]["sig"] = sig
|
|
can = bytes.fromhex(g["canonical"][:-128] + sig)
|
|
g["canonical"] = can.hex()
|
|
link = b3(b"moq-jwp:receipt-link\x01" + can).hex()
|
|
g["link_hash"] = link
|
|
for o in g["objects"]:
|
|
if o["binding"]:
|
|
b = json.loads(o["binding"])
|
|
b["receipt"] = link
|
|
o["binding"] = canonical(b)
|
|
out.append(("a signature forged, everything derived from it agreeing", forged, None))
|
|
out.append(("the envelope file changed", None, lambda name, b: b[:-1] + bytes([b[-1] ^ 1]) if name == "int_motion.wai" else b))
|
|
out.append(("refused provenance shown verified", m(lambda d: find(d, "frames", "cmaf_provenance_refused")["result"].update(state="verified")), None))
|
|
out.append(("a held chain shown verified", m(lambda d: find(d, "frames", "chain_held").update(chain="linked")), None))
|
|
out.append(("an ambiguous timestamp bound", m(lambda d: find(d, "tables", "same_timestamp").update(ops=find(d, "tables", "same_timestamp")["ops"][:1] + find(d, "tables", "same_timestamp")["ops"][3:])), None))
|
|
out.append(("a clone without its binding", m(lambda d: find(d, "tables", "clone_and_close")["result"]["frames"].update(f0c=None)), None))
|
|
out.append(("a bound raised", m(lambda d: find(d, "tables", "bound").update(bound=3)), None))
|
|
out.append(("a reason's only case removed", m(lambda d: d["tables"].remove(find(d, "tables", "same_timestamp"))), None))
|
|
return out
|
|
|
|
|
|
def never_read(doc):
|
|
"""Answers the corpus gives for codec strings no registry row lists, changed every way."""
|
|
out = []
|
|
for flip in ({"supported": False}, {"probe": "mismatch"}, {"power_efficient": False, "smooth": False}):
|
|
d = copy.deepcopy(doc)
|
|
for c in d["routes"]:
|
|
ua = c["report"].get("ua") if isinstance(c["report"], dict) else None
|
|
for k, a in (ua or {}).items():
|
|
if not gate(k):
|
|
a.update(flip)
|
|
out.append((json.dumps(flip), d))
|
|
return out
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) != 2:
|
|
print(__doc__)
|
|
return 2
|
|
d = sys.argv[1]
|
|
with open(os.path.join(d, "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
files = {}
|
|
|
|
def loader(change=None):
|
|
def load(name):
|
|
if name not in files:
|
|
with open(os.path.join(d, name), "rb") as f:
|
|
files[name] = f.read()
|
|
b = files[name]
|
|
return change(name, b) if change else b
|
|
return load
|
|
|
|
bad = 0
|
|
if sorted(REASONS) != sorted(KNOWN) or len(REASONS) != len(set(REASONS)):
|
|
print("FAIL §11 lists %s, this verifier knows %s" % (sorted(set(REASONS) ^ set(KNOWN)), len(KNOWN)))
|
|
bad += 1
|
|
fails = verify(doc, loader())
|
|
for f in fails:
|
|
print("FAIL " + f)
|
|
n = sum(len(doc[k]) for k in ("routes", "constraints", "bindings", "frame_tiers", "frames", "tables"))
|
|
n += sum(len(g["objects"]) for g in doc["chain"]["groups"])
|
|
print("%s every case gets the outcome the corpus states (%d cases; every reason of §11 given)" % ("PASS" if not fails else "FAIL", n))
|
|
bad += bool(fails)
|
|
for flip, d2 in never_read(doc):
|
|
f2 = verify(d2, loader())
|
|
print("%s answers for unregistered codec strings changed (%s) are never read" % ("PASS" if not f2 else "FAIL", flip))
|
|
bad += bool(f2)
|
|
caught = 0
|
|
muts = mutations(doc)
|
|
for name, change, file_change in muts:
|
|
d2 = copy.deepcopy(doc)
|
|
if change:
|
|
change(d2)
|
|
f2 = verify(d2, loader(file_change))
|
|
if f2:
|
|
caught += 1
|
|
else:
|
|
print("FAIL the mutation '%s' is not caught" % name)
|
|
print("%s %d of %d mutated copies fail" % ("PASS" if caught == len(muts) else "FAIL", caught, len(muts)))
|
|
bad += caught != len(muts)
|
|
return 1 if bad else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|