#!/usr/bin/env python3 """Independent verifier of the live-metering corpus (extensions/live-metering.md). Written from the extension's text. It shares no code with wai-rs, which generates the corpus, and never runs it. It checks Ed25519 signatures with ../claim-conformance/_ed25519.py, WAI's strict rule (SPEC §7 *Signatures*) written from RFC 8032 with the standard library. Each room is verified by the live-rooms corpus's own second implementation, live_rooms_verify.py (live-rooms §9); each take by capture_take_verify.py (capture-binding §7.2, §9.2); each composition receipt by live_compose_verify.py (live-compose §8.3). Everything else is here, from the text: - the per-hop session record (§3): read in version 1 or 2 by its own layout, its class, its rules (version 2's DVR figures held to its entries), the size bound of its version, its ledger and the ledger's root; - the room-session receipt (§4): read, its rules, who may sign it; - the room's replay (§5.3): who publishes, the grants in effect, the media tracks published; - the units (§4.4), the joules per hop and per class (§4.5), each grant's consumption and its verdicts (§4.6), and the QoE cell (§4.7) from the logs' session metrics (delivery-telemetry §7), all recomputed with Python's integers. For every vector it requires the outcome, reason and place, and the digest, units, records (each one's digest, size, bound and version, and a version 2 record's DVR figures), totals, coverage, consumption entries, root, grants past a ceiling and QoE cell the corpus states; of the corpus, that every reason of §9 refuses a vector and that §9 lists exactly those reasons. Then it mutates copies of the corpus, one defect each, and requires every copy to fail. python3 live_metering_verify.py Exit status 0 only if every vector verifies and every mutated copy fails. Python 3.8 or later; standard library only. """ import copy import hashlib import json import os import re import struct import sys HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance")) import live_events_verify as LEV # noqa: E402 import live_rooms_verify as LRV # noqa: E402 import capture_take_verify as CT # noqa: E402 import live_compose_verify as LCV # noqa: E402 import _ed25519 as ED # noqa: E402 Refuse = LEV.Refuse HOP_MAGIC = b"WMH1" HOP_VERSIONS = (1, 2) SESSION_MAGIC = b"WMR1" SESSION_VERSIONS = (1,) DVR_FIELDS = ("dvr_objects_out", "dvr_bytes_out", "dvr_objects_stored", "dvr_bytes_stored", "dvr_byte_s", "dvr_dropped") HOP_DOMAIN = b"wai:metering-hop\x01" SESSION_DOMAIN = b"wai:metering-session\x01" LEDGER_DOMAIN = b"wai:metering-ledger\x01" CONSUMPTION_DOMAIN = b"wai:metering-consumption\x01" NODE_DOMAIN = b"wai:out-node\x01" NO_CEILING = 2 ** 64 - 1 NO_FUNDS = 2 ** 64 - 1 CLASSES = 9 GAP_US = 1000000 QOE_MIN = 50 FLOOR = 288676 LABELS = {2: "ModelBased", 3: "Estimator", 4: "OnChipCounter", 5: "CalibratedInstrument"} HOP_AUTHORITY = {1: "live/relay", 2: "live/compositor", 3: "live/ingest"} METER_AUTHORITY = "live/meter" ZERO = bytes(32) JOULES_VERDICT = ["none", "within", "past", "partial", "unmetered"] FUNDS_VERDICT = ["none", "within", "past", "no_tariff"] LIFECYCLE = range(0x10, 0x17) REASONS = [ "session_invalid", "signer_not_authorised", "signer_revoked", "record_missing", "record_too_large", "energy_invalid", "record_invalid", "hop_not_authorised", "hop_revoked", "span_outside_session", "track_unknown", "ledger_invalid", "ledger_mismatch", "unit_mismatch", "classes_summed", "ceiling_mismatch", "qoe_invalid", "qoe_not_suppressed", ] class Stop(Exception): def __init__(self, reason, at): super().__init__(reason) self.reason, self.at = reason, at def sha(*parts): h = hashlib.sha256() for p in parts: h.update(p) return h.digest() def bound(m, version=1): """§3.7: B(M) = 552 + 180 × (M + 2) for version 1, B₂(M) = 600 + 180 × (M + 2) for version 2.""" return (600 if version == 2 else 552) + 180 * (m + 2) def bound_of(b, m): """§3.7: the bound of the version the fifth byte names, read before anything else; B(M) for a fifth byte other than 2, or none.""" return bound(m, 2 if len(b) >= 5 and b[4] == 2 else 1) def given_digest(b): """§5.1 step 4: SHA-256 over all but the last 64 bytes.""" return sha(b[:max(len(b) - 64, 0)]) def node(a, b): return sha(NODE_DOMAIN, a, b) def mth(leaves): """RFC 6962 §2.1, with the node of staged-measurement Appendix C; zero for none.""" if not leaves: return ZERO if len(leaves) == 1: return leaves[0] k = 1 while k * 2 < len(leaves): k *= 2 return node(mth(leaves[:k]), mth(leaves[k:])) class R: """A cursor; running out is `truncated`.""" def __init__(self, b, p=0): self.b, self.p = b, p def take(self, n): if self.p + n > len(self.b): raise Refuse("truncated") s = self.b[self.p:self.p + n] self.p += n return s def u8(self): return self.take(1)[0] def u16(self): return struct.unpack("I", self.take(4))[0] def be64(self): return struct.unpack(">Q", self.take(8))[0] def done(self): return self.p == len(self.b) def read_head(b, magic, versions): """live-events §3.3's first two steps: the version, one of `versions`.""" if len(b) < 4: raise Refuse("truncated") if b[:4] != magic: raise Refuse("malformed") if len(b) < 5: raise Refuse("truncated") if b[4] not in versions: raise Refuse("unknown_version") return b[4] INVALID = "invalid" def read_class(r): """energy-measurement §2.7's tag and evidence: None (tag 0), a tuple, or INVALID.""" tag = r.u8() if tag == 0: return None if tag in (2, 3): return (tag,) if tag == 4: ppm, window, f = r.be32(), r.be64(), r.u8() return (4, ppm, window, f) if ppm > 0 and window > 0 and f in (0, 1, 2) else INVALID if tag == 5: ppm, window, n = r.be32(), r.be64(), r.be32() if n > 255: raise Refuse("malformed") ref = r.take(n) try: text = ref.decode("utf-8") except UnicodeDecodeError: return INVALID return (5, ppm, window, text) if ppm > 0 and window > 0 and text.strip() != "" else INVALID return INVALID def energy_ok(joules, cls): """§3.6's `energy_invalid` rules.""" if cls == INVALID: return False if cls is None: return joules == 0 if joules == 0: return False return cls[0] not in (4, 5) or joules * cls[1] >= FLOOR def check_sig(key, domain, body, sig): bad = ED.verify(key, domain + body, sig) if bad: raise Refuse(bad) # ---------------------------------------------------------------- §3 the per-hop record def read_hop(b, room): """§3.6: the record, of version 1 or 2, its class, its rules, its room, its signature.""" version = read_head(b, HOP_MAGIC, HOP_VERSIONS) r = R(b, 5) x = {"version": version, "room": r.take(32), "hop": r.take(32), "role": r.u8(), "start": r.u64(), "end": r.u64(), "joules": r.u64()} x["class"] = read_class(r) x["tracks"] = [] for _ in range(r.u16()): e = {"kind": r.u8(), "track": r.take(32), "dir": r.u8()} for k in ("objects", "bytes", "timeout", "refused", "duplicate", "closed", "receipted"): e[k] = r.u64() x["tracks"].append(e) x["sessions"] = r.u32() x["sub"] = [r.u64() for _ in range(CLASSES)] x["events"], x["grants"], x["ledger"] = r.u64(), r.u32(), r.take(32) # §3.1: version 2 has `dvr`, six u64, after `ledger` and before `sig` (§3.9). x["dvr"] = {k: r.u64() for k in DVR_FIELDS} if version == 2 else None sig = r.take(64) if not r.done(): raise Refuse("malformed") if not energy_ok(x["joules"], x["class"]): raise Refuse("energy_invalid") order = [(e["kind"], e["track"], e["dir"]) for e in x["tracks"]] def entry_ok(e): if e["kind"] == 1: kind_ok = e["track"] != ZERO elif e["kind"] in (2, 3): kind_ok = e["track"] == ZERO else: return False if e["dir"] == 1: return kind_ok and e["timeout"] == 0 and e["closed"] == 0 and e["receipted"] == 0 \ and e["refused"] + e["duplicate"] <= e["objects"] if e["dir"] == 2: return kind_ok and e["refused"] == 0 and e["duplicate"] == 0 and e["receipted"] <= e["timeout"] + e["closed"] return False def dvr_ok(d): """§3.6, version 2: what it sent from its store within its out entries, what it stored within its in entries.""" if d is None: return True out = [e for e in x["tracks"] if e["dir"] == 2] inn = [e for e in x["tracks"] if e["dir"] == 1] return (d["dvr_objects_out"] <= sum(e["objects"] for e in out) and d["dvr_bytes_out"] <= sum(e["bytes"] for e in out) and d["dvr_objects_stored"] <= sum(e["objects"] for e in inn) and d["dvr_bytes_stored"] <= sum(e["bytes"] for e in inn)) rules = (x["role"] in (1, 2, 3) and x["start"] < x["end"] and all(a < z for a, z in zip(order, order[1:])) and all(entry_ok(e) for e in x["tracks"]) and (x["sessions"] != 0 or (not any(x["sub"]) and x["events"] == 0)) and ((x["grants"] == 0) == (x["ledger"] == ZERO)) and dvr_ok(x["dvr"])) if not rules: raise Refuse("record_invalid") if x["room"] != room: raise Refuse("wrong_room") check_sig(x["hop"], HOP_DOMAIN, b[:-64], sig) x["digest"] = sha(b[:-64]) return x def read_ledger(b): """§3.6: the ledger's entries, or Refuse("ledger_invalid").""" if len(b) % 143: raise Refuse("ledger_invalid") out = [] for k in range(0, len(b), 143): c = b[k:k + 143] e = {"raw": c, "grant": c[:32], "jc": int.from_bytes(c[32:40], "little"), "fc": int.from_bytes(c[40:48], "little"), "currency": c[48:51], "sessions": int.from_bytes(c[51:55], "little"), "in": int.from_bytes(c[55:63], "little"), "out": int.from_bytes(c[63:71], "little"), "sub": [int.from_bytes(c[71 + 8 * i:79 + 8 * i], "little") for i in range(CLASSES)]} currency_ok = e["currency"] == b"\x00\x00\x00" if e["fc"] == NO_CEILING else all(0x41 <= x <= 0x5A for x in e["currency"]) zero_ok = e["grant"] != ZERO or (e["jc"] == NO_CEILING and e["fc"] == NO_CEILING) if not currency_ok or not zero_ok or (out and out[-1]["grant"] >= e["grant"]): raise Refuse("ledger_invalid") out.append(e) return out def ledger_root(entries): return mth([sha(LEDGER_DOMAIN, e["raw"]) for e in entries]) # ---------------------------------------------------------------- §4 the receipt def read_receipt(b, room): """§4.2: the receipt, `classes_summed` for a total of tag 0, its rules, its room, its signature.""" read_head(b, SESSION_MAGIC, SESSION_VERSIONS) r = R(b, 5) x = {"room": r.take(32), "signer": r.take(32), "start": r.u64(), "end": r.u64()} x["hops"], classes_ok = [], True for _ in range(r.u8()): h = {"record": r.take(32), "hop": r.take(32), "role": r.u8(), "joules": r.u64()} h["class"] = read_class(r) classes_ok = classes_ok and energy_ok(h["joules"], h["class"]) x["hops"].append(h) x["takes"] = [r.take(32) for _ in range(r.u8())] x["compose_count"], x["compose_digest"] = r.u32(), r.take(32) x["qoe_count"], x["qoe_digest"] = r.u32(), r.take(32) x["units"] = units_of([r.u64() for _ in range(15)]) x["unmetered"] = r.u8() x["totals"] = [(r.u8(), r.u32(), r.u64()) for _ in range(r.u8())] x["currency"] = r.take(3) x["prices"] = [r.u32() for _ in range(13)] x["grants"], x["root"] = r.u32(), r.take(32) x["past"] = [(r.take(32), r.u8()) for _ in range(r.u32())] cell = [r.u8()] + [r.u32() for _ in range(3)] + [r.u64() for _ in range(3)] + [r.u32(), r.u64()] x["cell"] = cell_dict(cell) sig = r.take(64) if not r.done(): raise Refuse("malformed") if any(t[0] == 0 for t in x["totals"]): raise Refuse("classes_summed") tariff_ok = all(p == 0 for p in x["prices"]) if x["currency"] == b"\x00\x00\x00" else all(0x41 <= c <= 0x5A for c in x["currency"]) recs, takes, tags, past = [h["record"] for h in x["hops"]], x["takes"], [t[0] for t in x["totals"]], [p[0] for p in x["past"]] rules = (x["start"] < x["end"] and all(a < z for a, z in zip(recs, recs[1:])) and all(h["role"] in (1, 2, 3) for h in x["hops"]) and classes_ok and all(a < z for a, z in zip(takes, takes[1:])) and ((x["compose_count"] == 0) == (x["compose_digest"] == ZERO)) and ((x["qoe_count"] == 0) == (x["qoe_digest"] == ZERO)) and all(a < z for a, z in zip(tags, tags[1:])) and all(2 <= t[0] <= 5 and t[1] > 0 for t in x["totals"]) and tariff_ok and all(a < z for a, z in zip(past, past[1:])) and all(f in (1, 2, 3) for _, f in x["past"]) and (cell[0] == 0 or (cell[0] == 1 and not any(cell[1:])))) if not rules: raise Refuse("session_invalid") if x["room"] != room: raise Refuse("wrong_room") check_sig(x["signer"], SESSION_DOMAIN, b[:-64], sig) x["digest"] = sha(b[:-64]) return x def units_of(v): return {"publisher_us": v[0], "subscriber_us": v[1:10], "sessions": v[10], "composition_us": v[11], "events_sent": v[12], "events_delivered": v[13], "recorded_us": v[14]} def cell_dict(c): return {"suppressed": c[0] == 1, "started": c[1], "start_failed": c[2], "exit_before_start": c[3], "stalls": c[4], "stall_ms": c[5], "engaged_ms": c[6], "vst_count": c[7], "vst_ms": c[8]} # ---------------------------------------------------------------- §5.3 the room EMPTY = {"present": {}, "own": {}, "slots": None} class Ctx: """A room that verified: its descriptor, grants, key log, accepted records and replay.""" def __init__(self, room): descriptor = bytes.fromhex(room["descriptor"]) grants = [bytes.fromhex(g) for g in room["grants"]] head = room["key_log_head"] self.chains = [[bytes.fromhex(x) for x in c] for c in room["chains"]] out = LRV.verify_room(descriptor, grants, head, self.chains) if out["outcome"] == "refuse": raise Stop(out["reason"], ["room"] + out["at"]) self.d = LRV.read_descriptor(descriptor) self.id = self.d["room"] if out["outcome"] == "gap" and any(LRV.is_room_key(self.d, bytes.fromhex(g["principal"])) for g in out["gaps"]): raise Stop("lifecycle_incomplete", ["room"]) self.gaps = out.get("gaps", []) held, _ = LRV.check_grants(self.d, grants) self.grants = {g["digest"]: g for g in held} self.log = json.loads(head) if self.d["origin"] != b"" else None self.view = LCV.view_room(room) self.seen, digests, instants = [], set(), set() for c, i, g in out["order"]: ev = LEV.read_record(self.chains[c][i]) if ev["digest"] in digests: continue digests.add(ev["digest"]) self.seen.append({"chain": c, "record": i, "ev": ev, "grant": bytes.fromhex(g) if g else None}) if LRV.is_room_key(self.d, ev["principal"]) and ev["kind"] in LIFECYCLE: instants.add(ev["t"]) empty = {"present": {}, "own": {}, "slots": [{"live": None, "standby": None} for _ in self.d["slots"]]} self.segments = [[0, empty]] for x in sorted(instants): st = self.view.at(x) if x == 0: self.segments[0][1] = st else: self.segments.append([x, st]) def over(self, s, e): out = [] for k, (a, st) in enumerate(self.segments): z = self.segments[k + 1][0] if k + 1 < len(self.segments) else 2 ** 64 lo, hi = max(a, s), min(z, e) if lo < hi: out.append((st, lo, hi)) return out def at(self, x): return next(st for a, st in reversed(self.segments) if a <= x) def published(self, st): out = set() for (p, m), (_, layers) in st["own"].items(): out |= {LCV.track_digest(self.id, p.hex(), m, str(n)) for n in range(layers)} for slot, s in zip(self.d["slots"], st["slots"]): if s["live"]: out |= {LCV.track_digest(self.id, "_" + slot["name"].decode(), slot["media"], str(n)) for n in range(s["live"][2])} return out def ceilings(self, g): if g == ZERO: return (NO_CEILING, NO_CEILING, b"\x00\x00\x00") x = self.grants.get(g) return None if x is None else (x["joules"], x["funds"], x["currency"]) def entry(self, key): if self.log is None: return None return next((e for e in self.log["keys"] if e["key"] == key.hex()), None) def publishers(st): out = {p for p, _ in st["own"]} for s in st["slots"]: out |= {h[0] for h in (s["live"], s["standby"]) if h} return out def hop_authorised(ctx, x): """§5.2.""" if ctx.d["origin"] == b"": if not LRV.is_room_key(ctx.d, x["hop"]): raise Refuse("hop_not_authorised") return e = ctx.entry(x["hop"]) if e is None: raise Refuse("hop_not_authorised") need = [HOP_AUTHORITY[x["role"]]] + (["energy/" + LABELS[x["class"][0]]] if x["class"] else []) if not all(a in e["may_sign"] for a in need): raise Refuse("hop_not_authorised") if e["status"] == "revoked": raise Refuse("hop_revoked") def signer_authorised(ctx, rc): """§4.3.""" d = ctx.d if d["origin"] == b"": if rc["signer"] != d["authority"]: raise Refuse("signer_not_authorised") return e = ctx.entry(rc["signer"]) if e is None or not (rc["signer"] == d["authority"] or METER_AUTHORITY in e["may_sign"]): raise Refuse("signer_not_authorised") if not all("energy/" + LABELS[t[0]] in e["may_sign"] for t in rc["totals"]): raise Refuse("signer_not_authorised") if e["status"] == "revoked": raise Refuse("signer_revoked") # ---------------------------------------------------------------- delivery-telemetry §2, §7 EVENTS = {"request", "response", "start", "present", "switch", "layer-change", "stall-start", "stall-end", "frame-drop", "capable", "abandon", "end"} MEMBERS = {"v", "sid", "sn", "ts_ms", "event", "cid", "st", "sf", "object", "present", "switch", "layer_change", "capable", "cohort", "metrics", "claim"} WITH = {"object": ("request", "response"), "present": ("present",), "switch": ("switch",), "layer_change": ("layer-change",), "capable": ("capable",)} def is_int(v): return isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= 2 ** 53 - 1 def read_observation(o): """delivery-telemetry §2.1 and §2.4, as far as the session metrics read an observation.""" if not isinstance(o, dict) or set(o) != MEMBERS or o["v"] != 1 or isinstance(o["v"], bool): raise Refuse("qoe_invalid") if not isinstance(o["sid"], str) or not re.fullmatch(r"[0-9a-f]{32}", o["sid"]): raise Refuse("qoe_invalid") if not is_int(o["sn"]) or not (o["ts_ms"] is None or is_int(o["ts_ms"])) or o["event"] not in EVENTS: raise Refuse("qoe_invalid") if o["st"] not in ("v", "l", "ll") or o["sf"] not in ("o", "h", "d"): raise Refuse("qoe_invalid") for k, events in WITH.items(): if (o[k] is not None) != (o["event"] in events): raise Refuse("qoe_invalid") p = o["present"] if p is not None: if not isinstance(p, dict) or set(p) != {"unit", "layers", "first", "late_ms", "decode_us"}: raise Refuse("qoe_invalid") if not is_int(p["unit"]) or not is_int(p["layers"]) or p["layers"] == 0 or not isinstance(p["first"], bool): raise Refuse("qoe_invalid") if o["claim"] is not None and o["event"] != "end": raise Refuse("qoe_invalid") event_mode = o["event"] != "request" and (o["event"] != "present" or p["first"]) if event_mode and o["ts_ms"] is None: raise Refuse("qoe_invalid") return o def session_metrics(log): """delivery-telemetry §7: one log's metrics, or Refuse("qoe_invalid") for one that is not a log (§7.1).""" m = {"started": False, "vst": None, "exit_before_start": False, "start_failed": False, "stalls": 0, "stall_ms": 0, "engaged_ms": 0} start = present = stall = None last, ended = 0, False for i, o in enumerate(log): if o["sid"] != log[0]["sid"] or (i > 0 and o["sn"] <= log[i - 1]["sn"]) or ended: raise Refuse("qoe_invalid") t = o["ts_ms"] if t is not None: if t < last: raise Refuse("qoe_invalid") last = t e = o["event"] if e == "start" and start is None: m["started"], start = True, t elif e == "present": if o["present"]["first"]: if present is not None: raise Refuse("qoe_invalid") present = t m["vst"] = t - min(start, t) if start is not None and t is not None else None elif e == "stall-start" and present is not None and stall is None: stall = t elif e == "stall-end": if stall is not None and t is not None: m["stalls"] += 1 m["stall_ms"] += t - stall stall = None elif e in ("abandon", "end"): ended = True if stall is not None: m["stalls"] += 1 m["stall_ms"] += t - stall stall = None if present is not None: m["engaged_ms"] = t - present elif e == "abandon": m["exit_before_start"] = True else: m["start_failed"] = m["started"] return m def read_logs(b): """§6.1: each log's metrics.""" try: doc = json.loads(b.decode("utf-8")) except (UnicodeDecodeError, ValueError): raise Refuse("qoe_invalid") if not isinstance(doc, list): raise Refuse("qoe_invalid") out = [] for log in doc: if not isinstance(log, list) or not log: raise Refuse("qoe_invalid") out.append(session_metrics([read_observation(o) for o in log])) return out def cell_of(ms): """§4.7.""" if len(ms) < QOE_MIN: return cell_dict([1] + [0] * 8) return cell_dict([0, sum(m["started"] for m in ms), sum(m["start_failed"] for m in ms), sum(m["exit_before_start"] for m in ms), sum(m["stalls"] for m in ms), sum(m["stall_ms"] for m in ms), sum(m["engaged_ms"] for m in ms), sum(m["vst"] is not None for m in ms), sum(m["vst"] for m in ms if m["vst"] is not None)]) # ---------------------------------------------------------------- §5 verification def u64s(*v): return b"".join(struct.pack(" b_: raise Refuse("record_too_large") x = read_hop(rb, room) if entry is not None and (entry["hop"], entry["role"], entry["joules"], entry["class"]) != (x["hop"], x["role"], x["joules"], x["class"]): raise Refuse("unit_mismatch") hop_authorised(ctx, x) if x["start"] < s or x["end"] > e: raise Refuse("span_outside_session") if any(t["kind"] == 1 and t["track"] not in published for t in x["tracks"]): raise Refuse("track_unknown") led = read_ledger(lb) if len(led) != x["grants"] or ledger_root(led) != x["ledger"] \ or any(ctx.ceilings(l["grant"]) != (l["jc"], l["fc"], l["currency"]) for l in led): raise Refuse("ledger_mismatch") if any(sum(l["sub"][c] for l in led) != x["sub"][c] for c in range(CLASSES)): raise Refuse("unit_mismatch") except Refuse as err: raise Stop(err.reason, at) x["ledger_entries"], x["size"], x["bound"] = led, len(rb), b_ records.append(x) take_digests, recorded = [], [] for i, given in enumerate(takes): try: if given is None: raise Refuse("record_missing") mf = CT.read_manifest(given, room) _, _, _, commits = CT.find(ctx.d, ctx.grants, ctx.seen, ctx.gaps, mf) except Refuse as err: raise Stop(err.reason, ["take", i]) except CT.Stop as err: raise Stop(err.reason, ["take", i]) take_digests.append(sha(given[:-64])) us = 0 if commits: o, size = commits[0][1]["offset"], commits[-1][1]["size"] for c in mf["chunks"][:min(size, len(mf["chunks"]))]: us += max(min(c["end"] + o, e) - max(c["start"] + o, s), 0) recorded.append((us, commits[0][0]["grant"] if commits else None)) digests = [given_digest(b) for b in compositions] outputs = {} for i, b in enumerate(compositions): try: rc = LCV.read_receipt(b) dbytes = next((d for d in descriptors if given_digest(d) == rc["descriptor"]), None) if dbytes is None: raise Refuse("record_missing") out = LCV.verify_receipt(ctx.view, dbytes, b) if out["outcome"] == "refuse": raise Refuse(out["reason"]) if rc["t"] < s or rc["t"] >= e: raise Refuse("span_outside_session") except Refuse as err: raise Stop(err.reason, ["compose", i]) outputs.setdefault((rc["compositor"], rc["media"], rc["output"]["track"]), []).append((rc["t"], rc["compositor"])) metrics = [] if logs is not None: try: metrics = read_logs(logs) except Refuse as err: raise Stop(err.reason, ["qoe"]) # §4.4 the units and §4.6 the consumption. units = {"publisher_us": 0, "subscriber_us": [0] * CLASSES, "sessions": 0, "composition_us": 0, "events_sent": 0, "events_delivered": 0, "recorded_us": 0} per = {} def grant(g): if g not in per: jc, fc, cur = ctx.ceilings(g) or (NO_CEILING, NO_CEILING, b"\x00\x00\x00") per[g] = {"grant": g, "jc": jc, "fc": fc, "currency": cur, "pub": 0, "sub": [0] * CLASSES, "comp": 0, "rec": 0, "ev": 0, "attr": [0, 0, 0, 0], "unmetered": False, "metered": False} return per[g] for st, a, z in ctx.over(s, e): for p in publishers(st): units["publisher_us"] += z - a if p in st["present"]: grant(st["present"][p])["pub"] += z - a for g in st["present"].values(): grant(g) for x in records: for c in range(CLASSES): units["subscriber_us"][c] += x["sub"][c] units["sessions"] += x["sessions"] units["events_delivered"] += x["events"] moved = sum(l["in"] + l["out"] for l in x["ledger_entries"]) for l in x["ledger_entries"]: g = grant(l["grant"]) for c in range(CLASSES): g["sub"][c] += l["sub"][c] b = l["in"] + l["out"] if b == 0: continue if x["class"] is None: g["unmetered"] = True else: g["attr"][x["class"][0] - 2] += x["joules"] * b // moved g["metered"] = True for runs in outputs.values(): runs.sort() for (t1, comp), (t2, _) in zip(runs, runs[1:]): if t2 - t1 <= GAP_US: units["composition_us"] += t2 - t1 g = ctx.at(t1)["present"].get(comp) if g is not None: grant(g)["comp"] += t2 - t1 for x in ctx.seen: if s <= x["ev"]["t"] < e: units["events_sent"] += 1 if x["grant"] is not None: grant(x["grant"])["ev"] += 1 for us, g in recorded: units["recorded_us"] += us if g is not None: grant(g)["rec"] += us totals = {} for x in records: if x["class"] is not None: f, j = totals.get(x["class"][0], (0, 0)) totals[x["class"][0]] = (f + 1, j + x["joules"]) consumption = [] for g in sorted(per): c = per[g] attributed = sum(c["attr"]) if c["jc"] == NO_CEILING: jv = 0 elif attributed > c["jc"]: jv = 2 elif not c["metered"]: jv = 4 elif c["unmetered"]: jv = 3 else: jv = 1 priced = currency != b"\x00\x00\x00" and currency == c["currency"] funds = NO_FUNDS if priced: n = c["pub"] * prices[0] + sum(c["sub"][k] * prices[1 + k] for k in range(CLASSES)) + c["comp"] * prices[10] \ + c["rec"] * prices[11] + c["ev"] * prices[12] * 10 ** 6 funds = min(-(-n // 10 ** 12), NO_FUNDS - 1) if c["fc"] == NO_CEILING: fv = 0 elif not priced: fv = 3 elif funds > c["fc"]: fv = 2 else: fv = 1 raw = (g + u64s(c["jc"], c["fc"]) + c["currency"] + u64s(c["pub"], *c["sub"], c["comp"], c["rec"], c["ev"], *c["attr"]) + bytes([1 if c["unmetered"] else 0]) + u64s(funds) + bytes([jv, fv])) assert len(raw) == 198 c.update(funds=funds, jv=jv, fv=fv, leaf=sha(CONSUMPTION_DOMAIN, raw)) consumption.append(c) return {"records": records, "m": m, "units": units, "totals": totals, "takes": take_digests, "compose_digests": digests, "metrics": metrics, "consumption": consumption} def list_digest(digests): return sha(b"".join(digests)) if digests else ZERO def check_totals(rc, comp): """§5.1 step 9.""" unmetered = sum(1 for x in comp["records"] if x["class"] is None) if rc["unmetered"] != unmetered: raise Refuse("unit_mismatch") tt = comp["totals"] for tag, figures, joules in rc["totals"]: if tag not in tt: raise Refuse("unit_mismatch") if joules != tt[tag][1]: others = [c for c in tt if c != tag] summed = any(tt[tag][1] + sum(tt[o][1] for k, o in enumerate(others) if mask >> k & 1) == joules for mask in range(1, 1 << len(others))) raise Refuse("classes_summed" if summed else "unit_mismatch") if figures != tt[tag][0]: raise Refuse("unit_mismatch") if len(rc["totals"]) != len(tt): raise Refuse("unit_mismatch") def record_facts(x): """§5.4: a hop record's digest, size, bound and version, and a version 2 record's DVR figures.""" out = {"digest": x["digest"].hex(), "size": x["size"], "bound": x["bound"], "version": x["version"]} if x["dvr"] is not None: out.update(x["dvr"]) return out def verify_vector(v, doc, cache): """§5: the outcome the vector should state.""" try: ctx = cache.ctx(doc["rooms"][v["room"]]) except Stop as st: return {"outcome": "refuse", "reason": st.reason, "at": st.at} given_hops = [(bytes.fromhex(h["record"]), bytes.fromhex(h["ledger"])) for h in v["hops"]] descriptors = [bytes.fromhex(x) for x in v["descriptors"]] compositions = [bytes.fromhex(x) for x in v["compositions"]] takes = [bytes.fromhex(x) for x in v["takes"]] logs = v["logs"].encode("utf-8") if v["logs"] is not None else None try: rc = read_receipt(bytes.fromhex(v["receipt"]), ctx.id) signer_authorised(ctx, rc) except Refuse as err: return {"outcome": "refuse", "reason": err.reason, "at": ["receipt"]} hops = [(next(((r, l) for r, l in given_hops if given_digest(r) == h["record"]), None), h) for h in rc["hops"]] named_takes = [next((t for t in takes if given_digest(t) == d), None) for d in rc["takes"]] comps = sorted(compositions, key=given_digest) s, e = rc["start"], rc["end"] try: if len(comps) != rc["compose_count"] or list_digest([given_digest(b) for b in comps]) != rc["compose_digest"]: evaluate(ctx, s, e, hops, named_takes, [], descriptors, None, rc["prices"], rc["currency"]) raise Stop("record_missing", ["compose"]) named_logs = None if rc["qoe_count"] != 0: if logs is None or sha(logs) != rc["qoe_digest"]: evaluate(ctx, s, e, hops, named_takes, comps, descriptors, None, rc["prices"], rc["currency"]) raise Stop("record_missing", ["qoe"]) named_logs = logs comp = evaluate(ctx, s, e, hops, named_takes, comps, descriptors, named_logs, rc["prices"], rc["currency"]) if len(comp["metrics"]) != rc["qoe_count"]: raise Stop("unit_mismatch", ["qoe"]) if comp["units"] != rc["units"]: raise Stop("unit_mismatch", ["units"]) try: check_totals(rc, comp) except Refuse as err: raise Stop(err.reason, ["energy"]) cell = cell_of(comp["metrics"]) if len(comp["metrics"]) < QOE_MIN and not rc["cell"]["suppressed"]: raise Stop("qoe_not_suppressed", ["qoe"]) if cell != rc["cell"]: raise Stop("unit_mismatch", ["qoe"]) root = mth([c["leaf"] for c in comp["consumption"]]) past = [(c["grant"], (c["jv"] == 2) | ((c["fv"] == 2) << 1)) for c in comp["consumption"] if c["jv"] == 2 or c["fv"] == 2] if len(comp["consumption"]) != rc["grants"] or root != rc["root"] or past != rc["past"]: raise Stop("ceiling_mismatch", ["ceilings"]) except Stop as st: return {"outcome": "refuse", "reason": st.reason, "at": st.at} whole = not any(x["class"] is None for x in comp["records"]) and len(comp["records"]) > 0 def ceiling(x): return None if x == NO_CEILING else x return { "outcome": "accept", "digest": rc["digest"].hex(), "media_tracks": comp["m"], "records": [record_facts(x) for x in comp["records"]], "units": comp["units"], "unmetered": rc["unmetered"], "totals": [{"class": LABELS[t], "figures": f, "joules": j} for t, (f, j) in sorted(comp["totals"].items())], "coverage": "whole" if whole else "partial", "consumption": [{ "grant": c["grant"].hex(), "joules_ceiling": ceiling(c["jc"]), "funds_ceiling": ceiling(c["fc"]), "currency": None if c["currency"] == b"\x00\x00\x00" else c["currency"].decode(), "publisher_us": c["pub"], "subscriber_us": c["sub"], "composition_us": c["comp"], "recorded_us": c["rec"], "events_sent": c["ev"], "attributed": {LABELS[t]: c["attr"][t - 2] for t in (2, 3, 4, 5)}, "unmetered": c["unmetered"], "funds": None if c["funds"] == NO_FUNDS else c["funds"], "joules_verdict": JOULES_VERDICT[c["jv"]], "funds_verdict": FUNDS_VERDICT[c["fv"]], "leaf": c["leaf"].hex(), } for c in comp["consumption"]], "consumption_root": root.hex(), "past": [{"grant": g.hex(), "flags": f} for g, f in past], "qoe_cell": cell, } class Cache: """Each room of the corpus verified once, by its content.""" def __init__(self): self.rooms = {} def ctx(self, room): k = json.dumps(room, sort_keys=True) if k not in self.rooms: try: self.rooms[k] = Ctx(room) except Stop as st: self.rooms[k] = st got = self.rooms[k] if isinstance(got, Stop): raise got return got OUTCOME_KEYS = ("outcome", "reason", "at", "digest", "media_tracks", "records", "units", "unmetered", "totals", "coverage", "consumption", "consumption_root", "past", "qoe_cell") def check(doc, cache, quiet=False, only=None): fails = [] for v in doc["vectors"]: if only is not None and v["name"] not in only: continue try: got = verify_vector(v, doc, cache) except Exception as e: # a crash on a mutated copy is a failure to read it got, problems = None, ["the vector does not verify: %s: %s" % (type(e).__name__, e)] else: problems = [k for k in OUTCOME_KEYS if v.get(k) != got.get(k)] if not quiet: what = "?" if got is None else got["outcome"] + (" " + got["reason"] if got["outcome"] == "refuse" else "") print(("PASS" if not problems else "FAIL") + " vector %s: %s" % (v["name"], what)) for k in problems: fails.append("vector %s: %s is %s, the corpus states %s" % ( v["name"], k, json.dumps(got.get(k))[:200] if got else k, json.dumps(v.get(k))[:200])) return fails def flip(h, at, bit=0x01): b = bytearray(bytes.fromhex(h)) b[at] ^= bit return b.hex() def mutations(doc): """Copies of the corpus, each with one defect a conforming verifier must catch, and the vectors it touches.""" idx = {v["name"]: i for i, v in enumerate(doc["vectors"])} def vec(d, name): return d["vectors"][idx[name]] def mutate(label, names, f): d = copy.deepcopy(doc) f(d) return label, names, d def relay_hop(d, name): relay = doc["principals"]["relay"] return next(h for h in vec(d, name)["hops"] if h["record"][74:138] == relay) def drop_join(d): clock = doc["principals"]["relay"] chain = next(c for c in d["rooms"]["lesson"]["chains"] if LEV.read_record(bytes.fromhex(c[0]))["principal"].hex() == clock) chain.pop(4) def cons(d, name, k): return vec(d, name)["consumption"][k] def dvr_record(d): return next(r for r in vec(d, "dvr_relay")["records"] if r["version"] == 2) def as_version_1(d): r = dvr_record(d) for k in DVR_FIELDS: del r[k] r.update(version=1, size=r["size"] - 48, bound=r["bound"] - 48) def dvr_byte(d, k): h = relay_hop(d, "dvr_relay") n = len(h["record"]) // 2 h.update(record=flip(h["record"], n - 64 - 48 + k)) lesson_units = ["lesson", "signed_by_authority"] return [ mutate("lesson: a bit of the receipt's signature", ["lesson"], lambda d: vec(d, "lesson").update(receipt=flip(vec(d, "lesson")["receipt"], len(vec(d, "lesson")["receipt"]) // 2 - 1))), mutate("lesson: publisher-seconds stated one more", ["lesson"], lambda d: vec(d, "lesson")["units"].update(publisher_us=vec(d, "lesson")["units"]["publisher_us"] + 1)), mutate("lesson: recorded-seconds stated one less", ["lesson"], lambda d: vec(d, "lesson")["units"].update(recorded_us=vec(d, "lesson")["units"]["recorded_us"] - 1)), mutate("lesson: a byte of the relay's record", ["lesson"], lambda d: relay_hop(d, "lesson").update(record=flip(relay_hop(d, "lesson")["record"], 200))), mutate("lesson: a byte of the relay's ledger", ["lesson"], lambda d: relay_hop(d, "lesson").update(ledger=flip(relay_hop(d, "lesson")["ledger"], 60))), mutate("lesson: a composition receipt not given", ["lesson"], lambda d: vec(d, "lesson")["compositions"].pop()), mutate("lesson: a take not given", ["lesson"], lambda d: vec(d, "lesson")["takes"].pop()), mutate("lesson: a log not given", ["lesson"], lambda d: vec(d, "lesson").update(logs=None)), mutate("lesson: a stated verdict", ["lesson"], lambda d: cons(d, "lesson", 2).update(joules_verdict="past")), mutate("lesson: a stated attributed share", ["lesson"], lambda d: cons(d, "lesson", 3)["attributed"].update(OnChipCounter=1)), mutate("lesson: the stated coverage", ["lesson"], lambda d: vec(d, "lesson").update(coverage="partial")), mutate("lesson: a record's stated bound", ["lesson"], lambda d: vec(d, "lesson")["records"][0].update(bound=1)), mutate("mixed_classes: two totals stated as one class", ["mixed_classes"], lambda d: vec(d, "mixed_classes")["totals"][0].update(**{"class": "OnChipCounter"})), mutate("ceilings_past: a grant past stated within", ["ceilings_past"], lambda d: vec(d, "ceilings_past").update(past=vec(d, "ceilings_past")["past"][:1])), mutate("large_audience: the QoE cell stated suppressed", ["large_audience"], lambda d: vec(d, "large_audience")["qoe_cell"].update(suppressed=True)), mutate("large_audience: an observation's sn", ["large_audience"], lambda d: vec(d, "large_audience").update(logs=vec(d, "large_audience")["logs"].replace('"sn":1,', '"sn":0,', 1))), mutate("relay_recorded: a byte of the relay's own record", ["relay_recorded"], lambda d: vec(d, "relay_recorded")["hops"][0].update(record=flip(vec(d, "relay_recorded")["hops"][0]["record"], 120))), mutate("relay_recorded: subscriber-seconds stated other", ["relay_recorded"], lambda d: vec(d, "relay_recorded")["units"]["subscriber_us"].__setitem__(0, 1)), mutate("track_unknown: stated record_invalid", ["track_unknown"], lambda d: vec(d, "track_unknown").update(reason="record_invalid")), mutate("classes_summed: stated unit_mismatch", ["classes_summed"], lambda d: vec(d, "classes_summed").update(reason="unit_mismatch")), mutate("record_too_large: stated at another hop", ["record_too_large"], lambda d: vec(d, "record_too_large").update(at=["hop", 0])), mutate("lesson: a record stated as version 2", ["lesson"], lambda d: vec(d, "lesson")["records"][0].update(version=2)), mutate("dvr_relay: its record stated as version 1, with no DVR figures", ["dvr_relay"], as_version_1), mutate("dvr_relay: the stated byte-seconds one more", ["dvr_relay"], lambda d: dvr_record(d).update(dvr_byte_s=dvr_record(d)["dvr_byte_s"] + 1)), mutate("dvr_relay: the stated drops left out", ["dvr_relay"], lambda d: dvr_record(d).pop("dvr_dropped")), mutate("dvr_relay: the stated bound version 1's", ["dvr_relay"], lambda d: dvr_record(d).update(bound=dvr_record(d)["bound"] - 48)), mutate("dvr_relay: a byte of the relay's DVR figures", ["dvr_relay"], lambda d: dvr_byte(d, 16)), mutate("dvr_relay: the relay's record without its DVR figures", ["dvr_relay"], lambda d: relay_hop(d, "dvr_relay").update(record=relay_hop(d, "dvr_relay")["record"][:-2 * (64 + 48)] + relay_hop(d, "dvr_relay")["record"][-128:])), mutate("dvr_out_exceeds: stated unknown_version", ["dvr_out_exceeds"], lambda d: vec(d, "dvr_out_exceeds").update(reason="unknown_version")), mutate("dvr_stored_exceeds: stated accepted", ["dvr_stored_exceeds"], lambda d: vec(d, "dvr_stored_exceeds").update(outcome="accept")), mutate("record_unknown_version: stated record_invalid", ["record_unknown_version"], lambda d: vec(d, "record_unknown_version").update(reason="record_invalid")), mutate("dvr_record_too_large: stated record_invalid", ["dvr_record_too_large"], lambda d: vec(d, "dvr_record_too_large").update(reason="record_invalid")), mutate("lesson room: a lifecycle record not delivered", lesson_units, drop_join), mutate("lesson room: a byte of a grant", lesson_units, lambda d: d["rooms"]["lesson"]["grants"].__setitem__(4, flip(d["rooms"]["lesson"]["grants"][4], 230))), ] def reasons_listed(): """The codes §9's table lists, in order.""" with open(os.path.join(HERE, "..", "extensions", "live-metering.md"), encoding="utf-8") as f: text = f.read() at = text.index("\n## 9. Reasons") end = text.find("\n## ", at + 1) section = text[at:end if end != -1 else len(text)] return [m.group(1) for m in re.finditer(r"^\| `([a-z_]+)` \|", section, re.M)] def main(corpus_dir): fails = [] st = ED._self_test() print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st)) if st: fails.append("ed25519 self-test") with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f: doc = json.load(f) cache = Cache() fails += check(doc, cache) seen = {v.get("reason") for v in doc["vectors"]} missing = [r for r in REASONS if r not in seen] if missing: fails.append("no vector is refused %s" % ", ".join(missing)) print("%s every reason of §9 refuses a vector" % ("PASS" if not missing else "FAIL")) listed = reasons_listed() if listed != REASONS: fails.append("§9 lists %s" % listed) print("%s §9 lists exactly the reasons this verifier knows" % ("PASS" if listed == REASONS else "FAIL")) for label, names, mutated in mutations(doc): caught = check(mutated, Cache() if label.startswith("lesson room") else cache, quiet=True, only=set(names)) print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:120] if caught else "")) if not caught: fails.append("a mutated copy verifies: %s" % label) for f in fails: print("FAIL " + f) print(("PASS" if not fails else "FAIL") + " live-metering corpus: %d vectors" % len(doc["vectors"])) return 1 if fails else 0 if __name__ == "__main__": if len(sys.argv) != 2: print("usage: live_metering_verify.py ", file=sys.stderr) sys.exit(2) sys.exit(main(sys.argv[1]))