1022 lines
38 KiB
Python
1022 lines
38 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the energy-steering corpus (extensions/energy-steering.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. The room is read by the live-rooms corpus's own second
|
|
implementation, live_rooms_verify.py (live-rooms §9.1 steps 1 and 2); each per-hop record by
|
|
the live-metering corpus's, live_metering_verify.py (live-metering §3.6, §5.2). Everything
|
|
else is here, from the text:
|
|
|
|
- the steering policy (§3) and the steering decision (§7): read, their rules, who signs;
|
|
- the client's report (§4), with an RFC 9651 parser written from that RFC's §4.2;
|
|
- each pathway's window, coverage, status and figure (§5);
|
|
- the decision rule (§6): the limits, the order, the draw and the bands, the first pathway,
|
|
the priority and the TTL, all with Python's integers;
|
|
- the steering manifest, the HLS tag and the DASH element (§8), byte for byte.
|
|
|
|
For every request it requires the report the corpus states; for every vector the outcome,
|
|
reason and place, and the digest, priority, TTL, reason for the first pathway, each
|
|
pathway's evidence and limits, the manifest and the URI; for every policy its HLS tag and
|
|
DASH element; of the corpus, that every reason of §10 refuses a vector and that §10 lists
|
|
exactly those reasons. Then it mutates copies of the corpus, one defect each, and requires
|
|
every copy to fail.
|
|
|
|
python3 energy_steering_verify.py <live-conformance/steering>
|
|
|
|
Exit status 0 only if every check holds and every mutated copy fails. Python 3.8 or later;
|
|
standard library only.
|
|
"""
|
|
import copy
|
|
import functools
|
|
import hashlib
|
|
import json
|
|
import os
|
|
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 live_metering_verify as LMV # noqa: E402
|
|
import _ed25519 as ED # noqa: E402
|
|
|
|
Refuse = LEV.Refuse
|
|
|
|
POLICY_MAGIC = b"WSP1"
|
|
DECISION_MAGIC = b"WSD1"
|
|
POLICY_DOMAIN = b"wai:steering-policy\x01"
|
|
DECISION_DOMAIN = b"wai:steering-decision\x01"
|
|
DRAW_DOMAIN = b"wai:steering-draw\x01"
|
|
STEER_AUTHORITY = "live/steer"
|
|
PPM = 1000000
|
|
NO_CAPACITY = 2 ** 32 - 1
|
|
NOT_CLONE = 255
|
|
U64 = 2 ** 64 - 1
|
|
MANIFEST_KEY = "science.transaction.wai-decision"
|
|
LABELS = {2: "ModelBased", 3: "Estimator", 4: "OnChipCounter", 5: "CalibratedInstrument"}
|
|
RANK = {5: 0, 4: 1, 2: 2, 3: 3}
|
|
STATUS = ["figure", "unmetered", "mixed", "stale", "empty"]
|
|
WHY = [None, "buffer", "floor", "share", "hold", "rank", "none"]
|
|
BUFFER, FLOOR, SHARE, HOLD, BY_RANK, NONE = 1, 2, 3, 4, 5, 6
|
|
HLS, DASH = 1, 2
|
|
LATENCY, QUALITY, CAPACITY, ALL = 1, 2, 4, 7
|
|
|
|
REASONS = [
|
|
"policy_invalid", "clone_unknown", "decision_invalid", "evidence_not_on_pathway",
|
|
"evidence_outside_window", "evidence_mismatch", "unmetered_as_zero", "unknown_preferred",
|
|
"decision_mismatch", "manifest_mismatch",
|
|
]
|
|
|
|
|
|
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()
|
|
|
|
|
|
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 int(self, n):
|
|
return int.from_bytes(self.take(n), "little")
|
|
|
|
def u8(self):
|
|
return self.take(1)[0]
|
|
|
|
def short(self):
|
|
return self.take(self.u8())
|
|
|
|
def long(self):
|
|
return self.take(self.int(2))
|
|
|
|
def done(self):
|
|
return self.p == len(self.b)
|
|
|
|
|
|
def read_head(b, magic):
|
|
if len(b) < 4:
|
|
raise Refuse("truncated")
|
|
if b[:4] != magic:
|
|
raise Refuse("malformed")
|
|
if len(b) < 5:
|
|
raise Refuse("truncated")
|
|
if b[4] != 1:
|
|
raise Refuse("unknown_version")
|
|
|
|
|
|
def check_sig(key, domain, body, sig):
|
|
bad = ED.verify(key, domain + body, sig)
|
|
if bad:
|
|
raise Refuse(bad)
|
|
|
|
|
|
# ---------------------------------------------------------------- §3.3 character rules
|
|
|
|
ALPHA_DIGIT = set(b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789")
|
|
HEX = set(b"0123456789abcdefABCDEF")
|
|
UNRESERVED = ALPHA_DIGIT | set(b"-._~")
|
|
URI_CHARS = UNRESERVED | set(b":/?[]@!$&'()*+,;=%")
|
|
|
|
|
|
def percent_ok(s):
|
|
i = 0
|
|
while i < len(s):
|
|
if s[i] == 0x25:
|
|
if i + 2 >= len(s) or s[i + 1] not in HEX or s[i + 2] not in HEX:
|
|
return False
|
|
i += 3
|
|
else:
|
|
i += 1
|
|
return True
|
|
|
|
|
|
def uri_ok(s):
|
|
return 1 <= len(s) <= 2048 and all(c in URI_CHARS for c in s) and percent_ok(s)
|
|
|
|
|
|
def id_ok(s):
|
|
return 1 <= len(s) <= 64 and all(c in ALPHA_DIGIT or c in b".-_" for c in s)
|
|
|
|
|
|
def host_ok(s):
|
|
return all(c in ALPHA_DIGIT or c in b"-." for c in s)
|
|
|
|
|
|
def param_ok(s):
|
|
return all(c in UNRESERVED or c == 0x25 for c in s) and percent_ok(s)
|
|
|
|
|
|
# ---------------------------------------------------------------- §3 the policy
|
|
|
|
def read_policy(b):
|
|
"""§3.3: the policy, or Refuse. Its room and signature are checked by §9.1 step 2."""
|
|
read_head(b, POLICY_MAGIC)
|
|
r = R(b, 5)
|
|
p = {"room": r.take(32), "signer": r.take(32), "period": r.int(8), "window": r.int(8), "lag": r.int(8),
|
|
"basis": r.u8(), "floor": r.u8(), "ceiling": r.int(4), "min_buffer": r.int(4), "h": r.int(4),
|
|
"ttl": r.int(4), "min_ttl": r.int(4), "flags": r.u8(), "initial": r.u8()}
|
|
p["tiers"] = [r.int(8) for _ in range(r.u8())]
|
|
p["server_uri"], p["reload_uri"], p["decision_uri"] = r.long(), r.long(), r.long()
|
|
p["pathways"] = []
|
|
for _ in range(r.u8()):
|
|
x = {"id": r.short(), "base": r.u8(), "host": r.short()}
|
|
x["params"] = [(r.short(), r.short()) for _ in range(r.u8())]
|
|
x["top"], x["latency"], x["capacity"], x["share"] = r.u8(), r.int(4), r.int(4), r.int(4)
|
|
x["hops"] = [(r.take(32), r.u8()) for _ in range(r.u8())]
|
|
p["pathways"].append(x)
|
|
sig = r.take(64)
|
|
if not r.done():
|
|
raise Refuse("malformed")
|
|
ps, n = p["pathways"], len(p["tiers"])
|
|
if any(x["base"] != NOT_CLONE and x["base"] >= i for i, x in enumerate(ps)):
|
|
raise Refuse("clone_unknown")
|
|
ids, keys = set(), set()
|
|
|
|
def pathway_ok(x):
|
|
if not id_ok(x["id"]) or x["id"] in ids:
|
|
return False
|
|
ids.add(x["id"])
|
|
if x["base"] == NOT_CLONE:
|
|
if x["host"] or x["params"]:
|
|
return False
|
|
elif not (host_ok(x["host"]) and all(1 <= len(k) <= 64 and param_ok(k) and param_ok(v) for k, v in x["params"])
|
|
and all(a[0] < z[0] for a, z in zip(x["params"], x["params"][1:]))):
|
|
return False
|
|
if x["top"] >= n or not x["hops"] or not any(pl == 1 for _, pl in x["hops"]):
|
|
return False
|
|
for k, pl in x["hops"]:
|
|
if pl not in (1, 2) or k in keys:
|
|
return False
|
|
keys.add(k)
|
|
return True
|
|
rules = (p["period"] > 0 and p["window"] > 0 and p["window"] % p["period"] == 0 and p["basis"] in (1, 2)
|
|
and 1 <= p["min_ttl"] <= p["ttl"] and p["h"] <= PPM and p["flags"] <= 1
|
|
and 1 <= n <= 8 and p["tiers"][0] > 0 and all(a < z for a, z in zip(p["tiers"], p["tiers"][1:]))
|
|
and p["floor"] < n and all(uri_ok(p[k]) for k in ("server_uri", "reload_uri", "decision_uri"))
|
|
and 1 <= len(ps) <= 32 and all([pathway_ok(x) for x in ps])
|
|
and p["initial"] < len(ps) and ps[p["initial"]]["base"] == NOT_CLONE
|
|
and sum(x["share"] for x in ps) <= PPM)
|
|
if not rules:
|
|
raise Refuse("policy_invalid")
|
|
p["body"], p["sig"] = b[:-64], sig
|
|
p["digest"] = sha(p["body"])
|
|
return p
|
|
|
|
|
|
def index_of(p, ident):
|
|
return next((i for i, x in enumerate(p["pathways"]) if x["id"] == ident), None)
|
|
|
|
|
|
# ---------------------------------------------------------------- §4 the report, and RFC 9651
|
|
|
|
class SF:
|
|
"""RFC 9651 §4.2: a Dictionary, parsed whole. Values are (type, value) pairs."""
|
|
|
|
def __init__(self, s):
|
|
self.b, self.i = s, 0
|
|
|
|
def peek(self):
|
|
return self.b[self.i] if self.i < len(self.b) else None
|
|
|
|
def fail(self):
|
|
raise ValueError("structured field")
|
|
|
|
def sp(self):
|
|
while self.peek() == 0x20:
|
|
self.i += 1
|
|
|
|
def ows(self):
|
|
while self.peek() in (0x20, 0x09):
|
|
self.i += 1
|
|
|
|
def key(self):
|
|
c = self.peek()
|
|
if c is None or not (0x61 <= c <= 0x7a or c == 0x2a):
|
|
self.fail()
|
|
s = self.i
|
|
while self.peek() is not None and (0x61 <= self.peek() <= 0x7a or 0x30 <= self.peek() <= 0x39 or self.peek() in b"_-.*"):
|
|
self.i += 1
|
|
return self.b[s:self.i].decode("ascii")
|
|
|
|
def params(self):
|
|
out = {}
|
|
while self.peek() == 0x3b:
|
|
self.i += 1
|
|
self.sp()
|
|
k = self.key()
|
|
v = ("boolean", True)
|
|
if self.peek() == 0x3d:
|
|
self.i += 1
|
|
v = self.bare()
|
|
out[k] = v
|
|
return out
|
|
|
|
def item(self):
|
|
return (self.bare(), self.params())
|
|
|
|
def number(self):
|
|
neg = self.peek() == 0x2d
|
|
if neg:
|
|
self.i += 1
|
|
if self.peek() is None or not 0x30 <= self.peek() <= 0x39:
|
|
self.fail()
|
|
s, dot = self.i, None
|
|
while self.peek() is not None:
|
|
c = self.peek()
|
|
if 0x30 <= c <= 0x39:
|
|
self.i += 1
|
|
elif c == 0x2e and dot is None:
|
|
if self.i - s > 12:
|
|
self.fail()
|
|
dot = self.i
|
|
self.i += 1
|
|
else:
|
|
break
|
|
if (dot is None and self.i - s > 15) or (dot is not None and self.i - s > 16):
|
|
self.fail()
|
|
text = self.b[s:self.i].decode("ascii")
|
|
sign = -1 if neg else 1
|
|
if dot is None:
|
|
return ("integer", sign * int(text))
|
|
whole, frac = text.split(".")
|
|
if not 1 <= len(frac) <= 3:
|
|
self.fail()
|
|
return ("decimal", sign * (int(whole) * 1000 + int(frac) * 10 ** (3 - len(frac))))
|
|
|
|
def bare(self):
|
|
c = self.peek()
|
|
if c is None:
|
|
self.fail()
|
|
if c == 0x2d or 0x30 <= c <= 0x39:
|
|
return self.number()
|
|
if c == 0x22:
|
|
self.i += 1
|
|
out = bytearray()
|
|
while True:
|
|
c = self.peek()
|
|
if c is None:
|
|
self.fail()
|
|
self.i += 1
|
|
if c == 0x5c:
|
|
n = self.peek()
|
|
if n not in (0x22, 0x5c):
|
|
self.fail()
|
|
self.i += 1
|
|
out.append(n)
|
|
elif c == 0x22:
|
|
return ("string", bytes(out))
|
|
elif 0x20 <= c <= 0x7e:
|
|
out.append(c)
|
|
else:
|
|
self.fail()
|
|
if (0x41 <= c <= 0x5a or 0x61 <= c <= 0x7a) or c == 0x2a:
|
|
s = self.i
|
|
self.i += 1
|
|
while self.peek() is not None and (self.peek() in ALPHA_DIGIT or self.peek() in b"!#$%&'*+-.^_`|~:/"):
|
|
self.i += 1
|
|
return ("token", self.b[s:self.i].decode("ascii"))
|
|
if c == 0x3a:
|
|
self.i += 1
|
|
s = self.i
|
|
while True:
|
|
c = self.peek()
|
|
if c is None:
|
|
self.fail()
|
|
if c == 0x3a:
|
|
break
|
|
if not (c in ALPHA_DIGIT or c in b"+/="):
|
|
self.fail()
|
|
self.i += 1
|
|
v = self.b[s:self.i]
|
|
self.i += 1
|
|
core = v.rstrip(b"=")
|
|
q = len(v) - len(core)
|
|
if q > 2 or b"=" in core or (len(core) % 4 == 1 if q == 0 else (len(core) + q) % 4 != 0):
|
|
self.fail()
|
|
return ("bytes", v)
|
|
if c == 0x3f:
|
|
self.i += 1
|
|
if self.peek() not in (0x30, 0x31):
|
|
self.fail()
|
|
v = self.peek() == 0x31
|
|
self.i += 1
|
|
return ("boolean", v)
|
|
if c == 0x40:
|
|
self.i += 1
|
|
t, v = self.number()
|
|
if t != "integer":
|
|
self.fail()
|
|
return ("date", v)
|
|
if c == 0x25:
|
|
self.i += 1
|
|
if self.peek() != 0x22:
|
|
self.fail()
|
|
self.i += 1
|
|
out = bytearray()
|
|
while True:
|
|
c = self.peek()
|
|
if c is None:
|
|
self.fail()
|
|
self.i += 1
|
|
if c == 0x25:
|
|
h = self.b[self.i:self.i + 2]
|
|
if len(h) != 2 or not all(x in b"0123456789abcdef" for x in h):
|
|
self.fail()
|
|
out.append(int(h, 16))
|
|
self.i += 2
|
|
elif c == 0x22:
|
|
return ("display", bytes(out).decode("utf-8"))
|
|
elif 0x20 <= c <= 0x7e:
|
|
out.append(c)
|
|
else:
|
|
self.fail()
|
|
self.fail()
|
|
|
|
def member(self):
|
|
if self.peek() == 0x28:
|
|
self.i += 1
|
|
items = []
|
|
while True:
|
|
self.sp()
|
|
if self.peek() == 0x29:
|
|
self.i += 1
|
|
return ("inner", items, self.params())
|
|
items.append(self.item())
|
|
if self.peek() not in (0x20, 0x29):
|
|
self.fail()
|
|
return ("item",) + self.item()
|
|
|
|
def dictionary(self):
|
|
self.sp()
|
|
out = {}
|
|
while self.i < len(self.b):
|
|
k = self.key()
|
|
if self.peek() == 0x3d:
|
|
self.i += 1
|
|
out[k] = self.member()
|
|
else:
|
|
out[k] = ("item", ("boolean", True), self.params())
|
|
self.ows()
|
|
if self.i == len(self.b):
|
|
break
|
|
if self.peek() != 0x2c:
|
|
self.fail()
|
|
self.i += 1
|
|
self.ows()
|
|
if self.i == len(self.b):
|
|
self.fail()
|
|
return out
|
|
|
|
|
|
def parse_dictionary(s):
|
|
"""The dictionary, or None when it does not parse (UTF-8 errors included)."""
|
|
try:
|
|
return SF(s).dictionary()
|
|
except (ValueError, UnicodeDecodeError):
|
|
return None
|
|
|
|
|
|
def percent_decode(s):
|
|
out, i = bytearray(), 0
|
|
while i < len(s):
|
|
if s[i] == 0x25:
|
|
h = s[i + 1:i + 3]
|
|
if len(h) != 2 or not all(c in HEX for c in h):
|
|
return None
|
|
out.append(int(h, 16))
|
|
i += 3
|
|
else:
|
|
out.append(s[i])
|
|
i += 1
|
|
return bytes(out)
|
|
|
|
|
|
def query_parts(target):
|
|
"""§4.1: the decoded parts, in order; a part that does not decode is skipped."""
|
|
q = target.find(b"?")
|
|
if q < 0:
|
|
return []
|
|
rest = target[q + 1:]
|
|
f = rest.find(b"#")
|
|
if f >= 0:
|
|
rest = rest[:f]
|
|
out = []
|
|
for part in rest.split(b"&"):
|
|
e = part.find(b"=")
|
|
n, v = (part[:e], part[e + 1:]) if e >= 0 else (part, b"")
|
|
n, v = percent_decode(n), percent_decode(v)
|
|
if n is not None and v is not None:
|
|
out.append((n, v))
|
|
return out
|
|
|
|
|
|
def first(parts, name):
|
|
return next((v for n, v in parts if n == name), None)
|
|
|
|
|
|
def unquote(v):
|
|
return v[1:-1] if len(v) >= 2 and v[:1] == b'"' and v[-1:] == b'"' else v
|
|
|
|
|
|
def integer(v):
|
|
if v is None or not 1 <= len(v) <= 20 or not all(0x30 <= c <= 0x39 for c in v):
|
|
return None
|
|
x = int(v)
|
|
return x if x <= U64 else None
|
|
|
|
|
|
def least(m):
|
|
"""§4.4: an Integer of 0 or above, or the least of a non-empty Inner List of them."""
|
|
if m is None:
|
|
return None
|
|
if m[0] == "item":
|
|
values = [m[1]]
|
|
elif m[0] == "inner" and m[1]:
|
|
values = [bare for bare, _ in m[1]]
|
|
else:
|
|
return None
|
|
if any(t != "integer" or x < 0 for t, x in values):
|
|
return None
|
|
return min(x for _, x in values)
|
|
|
|
|
|
def read_request(p, protocol, target, header):
|
|
"""§4: (c, T, B), each None when absent."""
|
|
parts = query_parts(target)
|
|
c = t = None
|
|
if protocol == HLS:
|
|
v = first(parts, b"_HLS_pathway")
|
|
c = index_of(p, unquote(v)) if v is not None else None
|
|
t = integer(first(parts, b"_HLS_throughput"))
|
|
else:
|
|
j = 0
|
|
v = first(parts, b"_DASH_pathway")
|
|
if v is not None:
|
|
for k, item in enumerate(unquote(v).split(b",")):
|
|
i = index_of(p, item)
|
|
if i is not None:
|
|
c, j = i, k
|
|
break
|
|
v = first(parts, b"_DASH_throughput")
|
|
if v is not None:
|
|
items = v.split(b",")
|
|
t = integer(items[j]) if j < len(items) else None
|
|
cmcd = first(parts, b"CMCD")
|
|
if cmcd is None:
|
|
cmcd = header
|
|
d = None
|
|
if cmcd is not None:
|
|
try:
|
|
d = parse_dictionary(cmcd.decode("utf-8").encode("utf-8"))
|
|
except UnicodeDecodeError:
|
|
d = None
|
|
d = d or {}
|
|
b = least(d.get("bl"))
|
|
if t is None:
|
|
m = least(d.get("mtp"))
|
|
t = m * 1000 if m is not None and m * 1000 <= U64 else None
|
|
return c, t, b
|
|
|
|
|
|
# ---------------------------------------------------------------- §5 the evidence
|
|
|
|
def window(p, t):
|
|
b = 0 if t < p["lag"] else (t - p["lag"]) // p["period"] * p["period"]
|
|
return (b - p["window"] if b >= p["window"] else 0), b
|
|
|
|
|
|
def tag(x):
|
|
return x["class"][0] if x["class"] else 0
|
|
|
|
|
|
def evidence(p, i, recs, t):
|
|
"""§5.2, §5.3: (status, class, J, D)."""
|
|
a, b = window(p, t)
|
|
path = p["pathways"][i]
|
|
for key, _ in path["hops"]:
|
|
mine = [x for x in recs if x["hop"] == key]
|
|
if not (mine and mine[0]["start"] == a and all(z["start"] == y["end"] for y, z in zip(mine, mine[1:]))
|
|
and mine[-1]["end"] == b):
|
|
return (3, 0, 0, 0)
|
|
if any(x["class"] is None for x in recs):
|
|
return (1, 0, 0, 0)
|
|
tags = {tag(x) for x in recs}
|
|
if len(tags) > 1:
|
|
return (2, 0, 0, 0)
|
|
edges = {k for k, pl in path["hops"] if pl == 1}
|
|
d = 0
|
|
for x in recs:
|
|
if x["hop"] in edges:
|
|
if p["basis"] == 1:
|
|
d += sum(e["bytes"] for e in x["tracks"] if e["kind"] == 1 and e["dir"] == 2)
|
|
else:
|
|
d += sum(x["sub"][p["floor"]:8])
|
|
if d == 0:
|
|
return (4, 0, 0, 0)
|
|
return (0, tags.pop(), sum(x["joules"] for x in recs), d)
|
|
|
|
|
|
def rank(e):
|
|
return RANK[e[1]] if e[0] == 0 else 4
|
|
|
|
|
|
# ---------------------------------------------------------------- §6 the rule
|
|
|
|
def order(p, ev):
|
|
def cmp(x, y):
|
|
ex, ey = ev[x], ev[y]
|
|
if rank(ex) != rank(ey):
|
|
return -1 if rank(ex) < rank(ey) else 1
|
|
if ex[0] == 0 and ey[0] == 0 and ex[1] == ey[1]:
|
|
l, r = ex[2] * ey[3], ey[2] * ex[3]
|
|
if l != r:
|
|
return -1 if l < r else 1
|
|
a, b = p["pathways"][x]["id"], p["pathways"][y]["id"]
|
|
return -1 if a < b else (1 if a > b else 0)
|
|
return sorted(range(len(p["pathways"])), key=functools.cmp_to_key(cmp))
|
|
|
|
|
|
def draw(digest, session):
|
|
return int.from_bytes(sha(DRAW_DOMAIN, digest, session)[:4], "little") % PPM
|
|
|
|
|
|
def beats(w, x, h):
|
|
return w[0] == 0 and x[0] == 0 and w[2] * x[3] * (PPM + h) < x[2] * w[3] * PPM
|
|
|
|
|
|
def decide(p, session, report, loads, ev):
|
|
"""§6: (draw, limits, priority, ttl, reason)."""
|
|
c, t, b = report
|
|
ps = p["pathways"]
|
|
limits = []
|
|
for i, x in enumerate(ps):
|
|
bits = 0
|
|
if x["latency"] <= p["ceiling"]:
|
|
bits |= LATENCY
|
|
if x["top"] >= p["floor"]:
|
|
bits |= QUALITY
|
|
if x["capacity"] == NO_CAPACITY or loads[i] < x["capacity"] or (c == i and loads[i] <= x["capacity"]):
|
|
bits |= CAPACITY
|
|
limits.append(bits)
|
|
|
|
def within(i):
|
|
return limits[i] == ALL
|
|
od = order(p, ev)
|
|
d = draw(p["digest"], session)
|
|
at_floor = False
|
|
if t is not None:
|
|
tiers = [k for k, rate in enumerate(p["tiers"]) if rate <= t]
|
|
at_floor = not tiers or max(tiers) <= p["floor"]
|
|
c_within = c is not None and within(c)
|
|
w = next((i for i in od if within(i)), None)
|
|
band, s = None, 0
|
|
for i, x in enumerate(ps):
|
|
if s <= d < s + x["share"]:
|
|
band = i
|
|
s += x["share"]
|
|
if c_within and b is not None and b < p["min_buffer"]:
|
|
head, why = c, BUFFER
|
|
elif c_within and at_floor:
|
|
head, why = c, FLOOR
|
|
elif w is None:
|
|
head, why = (c if c is not None else od[0]), NONE
|
|
elif band is not None and within(band):
|
|
head, why = band, SHARE
|
|
elif c_within and c != w and rank(ev[c]) == rank(ev[w]) and (rank(ev[w]) == 4 or not beats(ev[w], ev[c], p["h"])):
|
|
head, why = c, HOLD
|
|
else:
|
|
head, why = w, BY_RANK
|
|
priority = [head] + [i for i in od if i != head and within(i)] + [i for i in od if i != head and not within(i)]
|
|
ttl = p["min_ttl"] if why in (BUFFER, FLOOR, NONE) else p["ttl"]
|
|
return d, limits, priority, ttl, why
|
|
|
|
|
|
# ---------------------------------------------------------------- §7 the decision
|
|
|
|
def read_decision(b):
|
|
"""§7.2: the decision, or Refuse."""
|
|
read_head(b, DECISION_MAGIC)
|
|
r = R(b, 5)
|
|
x = {"room": r.take(32), "signer": r.take(32), "policy": r.take(32), "t": r.int(8), "session": r.take(16),
|
|
"draw": r.int(4), "protocol": r.u8(), "present": r.u8(), "pathway": r.u8(), "throughput": r.int(8),
|
|
"buffer": r.int(8)}
|
|
x["pathways"] = []
|
|
for _ in range(r.u8()):
|
|
s = {"load": r.int(4), "limits": r.u8(), "status": r.u8(), "class": r.u8(), "joules": r.int(16),
|
|
"delivery": r.int(16)}
|
|
s["evidence"] = [r.take(32) for _ in range(r.u8())]
|
|
x["pathways"].append(s)
|
|
x["ttl"], x["reason"] = r.int(4), r.u8()
|
|
x["priority"] = list(r.take(r.u8()))
|
|
sig = r.take(64)
|
|
if not r.done():
|
|
raise Refuse("malformed")
|
|
n, pr = len(x["pathways"]), x["present"]
|
|
report_ok = (pr & ~7 == 0 and (pr & 1 or x["pathway"] == 0) and (pr & 2 or x["throughput"] == 0)
|
|
and (pr & 4 or x["buffer"] == 0) and (not pr & 1 or x["pathway"] < n))
|
|
|
|
def state_ok(s):
|
|
if s["limits"] & ~ALL or s["status"] > 4:
|
|
return False
|
|
if s["status"] == 0:
|
|
return 2 <= s["class"] <= 5 and s["joules"] > 0 and s["delivery"] > 0
|
|
return s["class"] == 0 and s["joules"] == 0 and s["delivery"] == 0
|
|
digests = [g for s in x["pathways"] for g in s["evidence"]]
|
|
rules = (x["protocol"] in (HLS, DASH) and x["draw"] < PPM and x["ttl"] >= 1 and 1 <= x["reason"] <= 6
|
|
and report_ok and all(state_ok(s) for s in x["pathways"]) and len(set(digests)) == len(digests)
|
|
and sorted(x["priority"]) == list(range(n)))
|
|
if not rules:
|
|
raise Refuse("decision_invalid")
|
|
x["body"], x["sig"] = b[:-64], sig
|
|
x["digest"] = sha(x["body"])
|
|
return x
|
|
|
|
|
|
def report_of(x):
|
|
return (x["pathway"] if x["present"] & 1 else None, x["throughput"] if x["present"] & 2 else None,
|
|
x["buffer"] if x["present"] & 4 else None)
|
|
|
|
|
|
# ---------------------------------------------------------------- §8 the mappings
|
|
|
|
def manifest(p, x):
|
|
"""§8.1."""
|
|
ps = p["pathways"]
|
|
reload = p["reload_uri"].decode() + ("&" if b"?" in p["reload_uri"] else "?") + "wai_session=" + x["session"].hex()
|
|
ids = ",".join('"%s"' % ps[i]["id"].decode() for i in x["priority"])
|
|
out = '{"VERSION":1,"TTL":%d,"RELOAD-URI":"%s","PATHWAY-PRIORITY":[%s]' % (x["ttl"], reload, ids)
|
|
clones = []
|
|
for y in ps:
|
|
if y["base"] == NOT_CLONE:
|
|
continue
|
|
rep = []
|
|
if y["host"]:
|
|
rep.append('"HOST":"%s"' % y["host"].decode())
|
|
if y["params"]:
|
|
rep.append('"PARAMS":{%s}' % ",".join('"%s":"%s"' % (k.decode(), v.decode()) for k, v in y["params"]))
|
|
clones.append('{"BASE-ID":"%s","ID":"%s","URI-REPLACEMENT":{%s}}' % (ps[y["base"]]["id"].decode(), y["id"].decode(), ",".join(rep)))
|
|
if clones:
|
|
out += ',"PATHWAY-CLONES":[%s]' % ",".join(clones)
|
|
return out + ',"%s":"%s%s"}' % (MANIFEST_KEY, p["decision_uri"].decode(), x["digest"].hex())
|
|
|
|
|
|
def hls_tag(p):
|
|
return '#EXT-X-CONTENT-STEERING:SERVER-URI="%s",PATHWAY-ID="%s"' % (p["server_uri"].decode(), p["pathways"][p["initial"]]["id"].decode())
|
|
|
|
|
|
def dash_element(p):
|
|
return '<ContentSteering defaultServiceLocation="%s" queryBeforeStart="%s">%s</ContentSteering>' % (
|
|
p["pathways"][p["initial"]]["id"].decode(), "true" if p["flags"] & 1 else "false",
|
|
p["server_uri"].decode().replace("&", "&"))
|
|
|
|
|
|
# ---------------------------------------------------------------- §9 verification
|
|
|
|
class Keys:
|
|
"""What live_metering_verify.hop_authorised reads of a room: its descriptor and log."""
|
|
|
|
def __init__(self, d, log):
|
|
self.d, self.log = d, log
|
|
|
|
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 signer_ok(keys, signer, classes):
|
|
"""§3.5."""
|
|
d = keys.d
|
|
if d["origin"] == b"":
|
|
if signer != d["authority"]:
|
|
raise Refuse("signer_not_authorised")
|
|
return
|
|
e = keys.entry(signer)
|
|
if e is None or not (signer == d["authority"] or STEER_AUTHORITY in e["may_sign"]):
|
|
raise Refuse("signer_not_authorised")
|
|
if not all("energy/" + LABELS[c] in e["may_sign"] for c in classes):
|
|
raise Refuse("signer_not_authorised")
|
|
if e["status"] == "revoked":
|
|
raise Refuse("signer_revoked")
|
|
|
|
|
|
def at(where, f, *args):
|
|
try:
|
|
return f(*args)
|
|
except Refuse as e:
|
|
raise Stop(e.reason, where)
|
|
|
|
|
|
def verify(descriptor, head, policy, decision, records, given_manifest):
|
|
"""§9.1: the outcome as the corpus states it."""
|
|
d = at(["room"], LRV.read_descriptor, descriptor)
|
|
at(["room"], LRV.check_key_log, d, head)
|
|
keys = Keys(d, json.loads(head) if d["origin"] != b"" else None)
|
|
room = d["room"]
|
|
p = at(["policy"], read_policy, policy)
|
|
if p["room"] != room:
|
|
raise Stop("wrong_room", ["policy"])
|
|
at(["policy"], check_sig, p["signer"], POLICY_DOMAIN, p["body"], p["sig"])
|
|
at(["policy"], signer_ok, keys, p["signer"], [])
|
|
x = at(["decision"], read_decision, decision)
|
|
if x["room"] != room:
|
|
raise Stop("wrong_room", ["decision"])
|
|
if x["policy"] != p["digest"]:
|
|
raise Stop("record_missing", ["decision"])
|
|
if len(x["pathways"]) != len(p["pathways"]):
|
|
raise Stop("decision_invalid", ["decision"])
|
|
at(["decision"], check_sig, x["signer"], DECISION_DOMAIN, x["body"], x["sig"])
|
|
at(["decision"], signer_ok, keys, x["signer"], sorted({s["class"] for s in x["pathways"] if s["status"] == 0}))
|
|
a, b = window(p, x["t"])
|
|
named = []
|
|
for pi, s in enumerate(x["pathways"]):
|
|
hops = [k for k, _ in p["pathways"][pi]["hops"]]
|
|
out, last = [], None
|
|
for k, g in enumerate(s["evidence"]):
|
|
where = ["evidence", pi, k]
|
|
given = next((r for r in records if LMV.given_digest(r) == g), None)
|
|
if given is None:
|
|
raise Stop("record_missing", where)
|
|
rec = at(where, LMV.read_hop, given, room)
|
|
at(where, LMV.hop_authorised, keys, rec)
|
|
if rec["hop"] not in hops:
|
|
raise Stop("evidence_not_on_pathway", where)
|
|
if rec["start"] < a or rec["end"] > b:
|
|
raise Stop("evidence_outside_window", where)
|
|
place = (hops.index(rec["hop"]), rec["start"])
|
|
if last is not None and last >= place:
|
|
raise Stop("decision_invalid", where)
|
|
last = place
|
|
out.append(rec)
|
|
named.append(out)
|
|
ev = [evidence(p, i, named[i], x["t"]) for i in range(len(p["pathways"]))]
|
|
for pi, s in enumerate(x["pathways"]):
|
|
stated = (s["status"], s["class"], s["joules"], s["delivery"])
|
|
if stated != ev[pi]:
|
|
if stated[0] == 0 and ev[pi][0] == 1:
|
|
raise Stop("unmetered_as_zero", ["pathway", pi])
|
|
if stated[0] == 0 and ev[pi][0] == 2:
|
|
raise Stop("classes_summed", ["pathway", pi])
|
|
raise Stop("evidence_mismatch", ["pathway", pi])
|
|
loads = [s["load"] for s in x["pathways"]]
|
|
dr, limits, priority, ttl, why = decide(p, x["session"], report_of(x), loads, ev)
|
|
if (dr, limits, priority, ttl, why) != (x["draw"], [s["limits"] for s in x["pathways"]], x["priority"], x["ttl"], x["reason"]):
|
|
unknown_first = ev[x["priority"][0]][0] != 0 and ev[priority[0]][0] == 0
|
|
raise Stop("unknown_preferred" if unknown_first else "decision_mismatch", ["output"])
|
|
m = manifest(p, x)
|
|
if given_manifest is not None and given_manifest != m:
|
|
raise Stop("manifest_mismatch", ["manifest"])
|
|
return {
|
|
"outcome": "accept",
|
|
"digest": x["digest"].hex(),
|
|
"priority": [p["pathways"][i]["id"].decode() for i in priority],
|
|
"ttl": ttl,
|
|
"why": WHY[why],
|
|
"pathways": [{"id": y["id"].decode(), "status": STATUS[e[0]], "class": LABELS[e[1]] if e[0] == 0 else None,
|
|
"joules": str(e[2]), "delivery": str(e[3]), "limits": limits[i]}
|
|
for i, (y, e) in enumerate(zip(p["pathways"], ev))],
|
|
"steering_manifest": m,
|
|
"uri": p["decision_uri"].decode() + x["digest"].hex(),
|
|
}
|
|
|
|
|
|
def inputs_of(policy, decision):
|
|
try:
|
|
p, x = read_policy(policy), read_decision(decision)
|
|
except Refuse:
|
|
return None
|
|
if len(x["pathways"]) != len(p["pathways"]):
|
|
return None
|
|
return {"t": x["t"], "session": x["session"].hex(), "protocol": "hls" if x["protocol"] == HLS else "dash",
|
|
"report": report_json(p, report_of(x)), "loads": [s["load"] for s in x["pathways"]]}
|
|
|
|
|
|
def report_json(p, rep):
|
|
c, t, b = rep
|
|
return {"pathway": p["pathways"][c]["id"].decode() if c is not None else None, "throughput": t, "buffer": b}
|
|
|
|
|
|
# ---------------------------------------------------------------- the corpus
|
|
|
|
OUTCOME_KEYS = ("outcome", "reason", "at", "digest", "priority", "ttl", "why", "pathways", "steering_manifest", "uri")
|
|
|
|
|
|
def check(doc, quiet=False):
|
|
"""Every failure of the corpus against this text, as strings."""
|
|
fails = []
|
|
try:
|
|
desc = bytes.fromhex(doc["room"]["descriptor"])
|
|
head = doc["room"]["key_log_head"]
|
|
pols = {n: bytes.fromhex(v["policy"]) for n, v in doc["policies"].items()}
|
|
pool = {k: bytes.fromhex(v) for k, v in doc["records"].items()}
|
|
except (KeyError, ValueError, TypeError, AttributeError) as e:
|
|
return ["the corpus does not read: %r" % e]
|
|
for n, v in doc["policies"].items():
|
|
try:
|
|
p = read_policy(pols[n])
|
|
want = (hls_tag(p), dash_element(p))
|
|
except Refuse:
|
|
want = (None, None)
|
|
if (v.get("hls_tag"), v.get("dash_element")) != want:
|
|
fails.append("policy %s: its HLS tag or DASH element differs" % n)
|
|
for r in doc["requests"]:
|
|
try:
|
|
p = read_policy(pols[r["policy"]])
|
|
header = r["cmcd_header"].encode() if r["cmcd_header"] is not None else None
|
|
got = report_json(p, read_request(p, HLS if r["protocol"] == "hls" else DASH, r["target"].encode(), header))
|
|
except (Refuse, KeyError, AttributeError) as e:
|
|
fails.append("request %s does not read: %r" % (r.get("name"), e))
|
|
continue
|
|
if got != r["report"]:
|
|
fails.append("request %s: the report is %s, the corpus states %s" % (r["name"], got, r["report"]))
|
|
seen = set()
|
|
for v in doc["vectors"]:
|
|
name = v.get("name", "?")
|
|
try:
|
|
policy = pols[v["policy"]]
|
|
decision = bytes.fromhex(v["decision"])
|
|
records = [pool[g] for g in v["records"]]
|
|
except (KeyError, ValueError, TypeError) as e:
|
|
fails.append("%s does not read: %r" % (name, e))
|
|
continue
|
|
try:
|
|
got = verify(desc, head, policy, decision, records, v["manifest"])
|
|
except Stop as e:
|
|
got = {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
|
for k in OUTCOME_KEYS:
|
|
if v.get(k) != got.get(k):
|
|
fails.append("%s: %s is %s, the corpus states %s" % (name, k, got.get(k), v.get(k)))
|
|
if v.get("inputs") != inputs_of(policy, decision):
|
|
fails.append("%s: its inputs are not the decision's" % name)
|
|
if v.get("request") is not None:
|
|
try:
|
|
p, x = read_policy(policy), read_decision(decision)
|
|
rq = v["request"]
|
|
header = rq["cmcd_header"].encode() if rq["cmcd_header"] is not None else None
|
|
if read_request(p, HLS if rq["protocol"] == "hls" else DASH, rq["target"].encode(), header) != report_of(x):
|
|
fails.append("%s: its request does not give the decision's report" % name)
|
|
except (Refuse, KeyError, AttributeError):
|
|
fails.append("%s: its request does not read" % name)
|
|
if got["outcome"] == "refuse":
|
|
seen.add(got["reason"])
|
|
for r in REASONS + ["classes_summed", "hop_not_authorised", "hop_revoked", "record_missing"]:
|
|
if r not in seen:
|
|
fails.append("nothing in the corpus is refused %s" % r)
|
|
if not quiet:
|
|
for f in fails:
|
|
print("FAIL " + f)
|
|
return fails
|
|
|
|
|
|
def reasons_listed():
|
|
"""§10's table, in order."""
|
|
path = os.path.join(HERE, "..", "extensions", "energy-steering.md")
|
|
with open(path, encoding="utf-8") as f:
|
|
text = f.read()
|
|
at_ = text.index("\n## 10. Reasons")
|
|
end = text.find("\n## ", at_ + 1)
|
|
body = text[at_:end if end >= 0 else len(text)]
|
|
return [line.split("`")[1] for line in body.splitlines() if line.startswith("| `")]
|
|
|
|
|
|
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, by name."""
|
|
out = []
|
|
|
|
def mutate(name, f):
|
|
d = copy.deepcopy(doc)
|
|
f(d)
|
|
out.append((name, d))
|
|
|
|
def vec(d, n):
|
|
return next(v for v in d["vectors"] if v["name"] == n)
|
|
|
|
def first_record(d, n):
|
|
return vec(d, n)["records"][0]
|
|
mutate("a request's report", lambda d: d["requests"][0]["report"].update(throughput=4000001))
|
|
mutate("a request's target", lambda d: d["requests"][1].update(target="/v1/room?_HLS_pathway=edge-a"))
|
|
mutate("a CMCD request's header", lambda d: d["requests"][15].update(cmcd_header="bl=(9000;v)"))
|
|
mutate("an accepted decision's signature", lambda d: vec(d, "three_metered").update(decision=flip(vec(d, "three_metered")["decision"], -1)))
|
|
mutate("an accepted priority", lambda d: vec(d, "three_metered")["priority"].reverse())
|
|
mutate("an accepted TTL", lambda d: vec(d, "floor_client").update(ttl=300))
|
|
mutate("an accepted reason", lambda d: vec(d, "hysteresis_hold").update(why="rank"))
|
|
mutate("a pathway's joules", lambda d: vec(d, "mixed_classes")["pathways"][2].update(joules="1200000001"))
|
|
mutate("a pathway's class", lambda d: vec(d, "mixed_classes")["pathways"][0].update(**{"class": "OnChipCounter"}))
|
|
mutate("a pathway's status", lambda d: vec(d, "one_unmetered")["pathways"][1].update(status="figure"))
|
|
mutate("a pathway's limits", lambda d: vec(d, "latency")["pathways"][2].update(limits=7))
|
|
mutate("a given manifest", lambda d: vec(d, "clones").update(manifest=vec(d, "clones")["manifest"].replace("east-2", "east-3")))
|
|
mutate("a stated manifest", lambda d: vec(d, "dash").update(steering_manifest=vec(d, "dash")["steering_manifest"].replace('"TTL":300', '"TTL":301')))
|
|
mutate("a stated URI", lambda d: vec(d, "share").update(uri=vec(d, "share")["uri"][:-1] + ("1" if vec(d, "share")["uri"][-1] == "0" else "0")))
|
|
mutate("a record's byte", lambda d: d["records"].update({first_record(d, "stale"): flip(d["records"][first_record(d, "stale")], 40)}))
|
|
mutate("a record left out", lambda d: vec(d, "per_second")["records"].pop())
|
|
mutate("a refused reason", lambda d: vec(d, "classes_summed").update(reason="evidence_mismatch"))
|
|
mutate("a refused place", lambda d: vec(d, "record_missing").update(at=["evidence", 2, 3]))
|
|
mutate("a vector's inputs", lambda d: vec(d, "dash")["inputs"]["report"].update(buffer=30000))
|
|
mutate("a policy's HLS tag", lambda d: d["policies"]["three"].update(hls_tag=d["policies"]["three"]["hls_tag"].replace("edge-a", "edge-b")))
|
|
mutate("a policy's DASH element", lambda d: d["policies"]["three"].update(dash_element=d["policies"]["three"]["dash_element"].replace("&", "&")))
|
|
def key_log_sig(d):
|
|
h = d["room"]["key_log_head"]
|
|
i = h.rindex('"sig":"') + 7
|
|
d["room"]["key_log_head"] = h[:i] + ("1" if h[i] == "0" else "0") + h[i + 1:]
|
|
mutate("the key log's signature", key_log_sig)
|
|
mutate("the reason's only vector", lambda d: d["vectors"].remove(vec(d, "unknown_preferred")))
|
|
mutate("a policy's byte", lambda d: d["policies"]["explore"].update(policy=flip(d["policies"]["explore"]["policy"], -1)))
|
|
return out
|
|
|
|
|
|
def main(corpus_dir):
|
|
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
fail = 0
|
|
fails = check(doc)
|
|
n_req, n_vec = len(doc["requests"]), len(doc["vectors"])
|
|
print("%s %d requests and %d vectors verify from the text" % ("PASS" if not fails else "FAIL", n_req, n_vec))
|
|
fail += bool(fails)
|
|
listed = reasons_listed()
|
|
ok = listed == REASONS
|
|
print("%s §10 lists exactly the reasons this verifier raises" % ("PASS" if ok else "FAIL"))
|
|
fail += not ok
|
|
caught = 0
|
|
muts = mutations(doc)
|
|
for name, d in muts:
|
|
if check(d, quiet=True):
|
|
caught += 1
|
|
else:
|
|
print("FAIL the mutation '%s' goes unnoticed" % name)
|
|
fail += 1
|
|
print("%s %d of %d mutated copies of the corpus fail" % ("PASS" if caught == len(muts) else "FAIL", caught, len(muts)))
|
|
return 1 if fail else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: energy_steering_verify.py <live-conformance/steering>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|