1cffbc6ffd
- §7 step 4 as the text stands: a refinement of the pick is one whose refines.base, read
on its own, is the pick's id (container::refinements_naming, refinements_naming in
http_verify.py), so one whose refines is otherwise malformed is reported dropped
(components_nonconforming). Of a conforming envelope the set is the same as before.
- Objects OCF (ONC's conforming twin: refinements [2], dropped []), ONM (refines
{base: b} alone: reported) and ONX (refines {base: x}: not reported), on the whole and
the range paths; fetch_OR_malformed_refines now reports its refinement; the generator
checks each outcome.
- falsify: conforming-withheld and refines-read-whole, each with an http_verify.py
--mutate, wired into falsify/run.sh; nonconforming-offered's must-fail list extended.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
3486 lines
138 KiB
Python
3486 lines
138 KiB
Python
#!/usr/bin/env python3
|
||
"""Second verifier for http-conformance/ — HTTP delivery (extensions/http-delivery.md).
|
||
|
||
Written from the extension text and SPEC §7.1. It shares no code with wai-rs, which
|
||
generates the corpus: the head (§2, §5.1), covered_len, the head proof (§6), the
|
||
response checks (§7.1), the plans and classes of the reference client (§7, §10.1), the
|
||
byte-range procedure with its whole-object fallback (§7), the base presented as it
|
||
verifies and the refinement deadline on the caller's clock (§10.1, §10.2), SPEC §7.1's selection with
|
||
companions and refinements, the codecs grammar, union, floor and support (§4), all
|
||
written here. BLAKE3 is ../claim-conformance/_blake3.py (written from its specification,
|
||
shared with the other Python verifiers) and SHA-256 is hashlib. The registry's kinds,
|
||
roles and classes are read from ../registry/capabilities.json, the published export.
|
||
Only the standard library is needed.
|
||
|
||
Pins (SPEC §3.1) are read as far as the corpus needs: a pin applies to a capability that
|
||
decodes against a parameter set (registry class neural-int-exact, neural-entropy-exact,
|
||
neural-float or generative-exact) or that the registry does not hold; a form-less pin
|
||
resolves when the sink holds a file whose SHA-256 is the pin's; a pin in the set form
|
||
resolves to nothing here, since the corpus's sinks hold no set listing. Any other shape
|
||
stops the verifier, never passes.
|
||
|
||
python3 http_verify.py [dir] verify every case and the self-test
|
||
python3 http_verify.py [dir] --mutate NAME the same with a deliberate defect:
|
||
left-half | no-root | counter-zero | floor-as-codecs | blake3-iv | base-at-done |
|
||
tick-noop | refinement-unclipped | nonconforming-offered | conforming-withheld |
|
||
refines-read-whole | ... (falsify/run.sh runs every one)
|
||
python3 http_verify.py cases FILE [dir]
|
||
give every case of a case file (wai_http_vectors fuzz) its stated outcome
|
||
python3 http_verify.py diff FILE [--flip N --seed S]
|
||
verify every head proof `wai_http_vectors diff` wrote; with --flip, change one
|
||
chaining-value byte in N of them, chosen by S, and require exactly those refused
|
||
|
||
Exit status 0 only when every case gives its committed outcome (or, for diff, every
|
||
proof verifies, or exactly the flipped ones fail).
|
||
"""
|
||
import hashlib, json, os, random, re, sys, time
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
||
sys.path.insert(0, os.path.join(HERE, "..", "staged-conformance"))
|
||
import _blake3 as B3 # noqa: E402
|
||
import staged_verify as SV # noqa: E402 (strict JSON, the manifest members, refines, explode)
|
||
|
||
MUTATE = None
|
||
DOMAIN = b"moq-jwp:object\x01"
|
||
READABLE = ["wai.prior.set"]
|
||
|
||
|
||
class Refusal(Exception):
|
||
def __init__(self, code, skipped=None):
|
||
super().__init__(code)
|
||
self.code, self.skipped = code, skipped
|
||
|
||
|
||
def sha(b):
|
||
return hashlib.sha256(b).hexdigest()
|
||
|
||
|
||
def blake3(b):
|
||
return B3.blake3(b)
|
||
|
||
|
||
def content_hash(env):
|
||
return blake3(DOMAIN + env)
|
||
|
||
|
||
# ------------------------------------------------------------------ the registry
|
||
|
||
REG = {}
|
||
|
||
|
||
def load_registry():
|
||
d = json.load(open(os.path.join(HERE, "..", "registry", "capabilities.json")))
|
||
for c in d["capabilities"]:
|
||
REG[c["capability"]] = c
|
||
|
||
|
||
def kind(cap):
|
||
return REG[cap]["kind"] if cap in REG else None
|
||
|
||
|
||
def candidate(cap):
|
||
return cap in REG and REG[cap]["status"] == "candidate"
|
||
|
||
|
||
def dispatchable(cap):
|
||
k = kind(cap)
|
||
return k is None or (k == "payload" and not candidate(cap))
|
||
|
||
|
||
def enhancement(cap):
|
||
return cap in REG and REG[cap]["role"] == "enhancement"
|
||
|
||
|
||
def pin_applies(cap):
|
||
return cap not in REG or REG[cap]["replicate"] in ("neural-int-exact", "neural-entropy-exact", "neural-float", "generative-exact")
|
||
|
||
|
||
# ---------------------------------------------------------------------- the head
|
||
|
||
def u32(b, at):
|
||
return int.from_bytes(b[at:at + 4], "little")
|
||
|
||
|
||
def read_manifest(b, v2):
|
||
"""SPEC §2, §3, §7.1: the manifest read strictly, with the members SPEC requires and
|
||
their types (staged_verify's reader); `head_malformed` otherwise."""
|
||
try:
|
||
m = SV.strict_json(b)
|
||
(SV.check_manifest_v2 if v2 else SV.check_manifest_v1)(m)
|
||
except (SV.NotJSON, SV.Schema):
|
||
raise Refusal("head_malformed")
|
||
return m
|
||
|
||
|
||
def parse_head(prefix, n):
|
||
"""§2, §5.1: ("need", k) or a head dict; Refusal on a malformed or mis-sized object."""
|
||
if len(prefix) < 8:
|
||
return ("need", 8)
|
||
magic = prefix[:4]
|
||
if magic not in (b"WAI1", b"WAI2"):
|
||
raise Refusal("head_malformed")
|
||
ml = u32(prefix, 4)
|
||
if magic == b"WAI1":
|
||
head_len = 12 + ml
|
||
if len(prefix) < head_len:
|
||
return ("need", head_len)
|
||
comps = [(head_len, u32(prefix, 8 + ml))]
|
||
else:
|
||
if len(prefix) < 10 + ml:
|
||
return ("need", 10 + ml)
|
||
k = int.from_bytes(prefix[8 + ml:10 + ml], "little")
|
||
head_len = 10 + ml + 8 * k
|
||
if len(prefix) < head_len:
|
||
return ("need", head_len)
|
||
comps, expected = [], 0
|
||
for i in range(k):
|
||
off, ln = u32(prefix, 10 + ml + 8 * i), u32(prefix, 14 + ml + 8 * i)
|
||
if off != expected:
|
||
raise Refusal("head_malformed")
|
||
comps.append((head_len + off, ln))
|
||
expected += ln
|
||
if head_len + sum(c[1] for c in comps) != n:
|
||
raise Refusal("object_length_mismatch")
|
||
m = read_manifest(prefix[8:8 + ml], magic == b"WAI2")
|
||
if magic == b"WAI2" and len(m["renditions"]) != len(comps):
|
||
raise Refusal("head_malformed")
|
||
return {"version": magic.decode(), "head_len": head_len, "envelope_len": n, "components": comps, "manifest": m}
|
||
|
||
|
||
def is_layer(h):
|
||
"""§10.4: a WAI1 whose capability is refinement-only (staged delivery §3). A payload
|
||
capability that carries `refines` is an ordinary object: SPEC §4 dispatches it and
|
||
`refines` is not read."""
|
||
if h["version"] != "WAI1":
|
||
return False
|
||
return kind(h["manifest"]["model_requirement"]["capability"]) == "refinement"
|
||
|
||
|
||
def whole_head(env):
|
||
"""The head of a whole envelope; one shorter than its head is malformed."""
|
||
h = parse_head(env, len(env))
|
||
if isinstance(h, tuple):
|
||
raise Refusal("head_malformed")
|
||
return h
|
||
|
||
|
||
def covered_len(head_len, n):
|
||
return min(n, 1024 * -(-(15 + head_len) // 1024) - 15)
|
||
|
||
|
||
# ------------------------------------------------------------------ the head proof
|
||
|
||
def left(n):
|
||
if MUTATE == "left-half":
|
||
return max(1, (n // 2) // 1024) * 1024
|
||
return 1024 * 2 ** ((((n - 1) // 1024)).bit_length() - 1)
|
||
|
||
|
||
def merge(l, r, root):
|
||
if MUTATE == "no-root":
|
||
root = False
|
||
return B3.parent_cv(l, r, root)
|
||
|
||
|
||
def subtree(x, start):
|
||
return B3.subtree_cv(x, 0 if MUTATE == "counter-zero" else start)
|
||
|
||
|
||
def read_proof(b):
|
||
if len(b) < 53 or b[:4] != b"WHP1":
|
||
raise Refusal("head_proof_malformed")
|
||
k = b[52]
|
||
if k > 64 or len(b) != 53 + 32 * k:
|
||
raise Refusal("head_proof_malformed")
|
||
n, c = int.from_bytes(b[36:44], "little"), int.from_bytes(b[44:52], "little")
|
||
if n > 2 ** 64 - 1 - 15 or c > n:
|
||
raise Refusal("head_proof_malformed")
|
||
if not (c == n or (c < n and (15 + c) % 1024 == 0)) or (c == n and k):
|
||
raise Refusal("head_proof_malformed")
|
||
return {"cid": b[4:36], "n": n, "covered": c, "cvs": [b[53 + 32 * i:85 + 32 * i] for i in range(k)]}
|
||
|
||
|
||
def verify_proof(p, cid, prefix):
|
||
"""§6.3, checks 1-3, in order."""
|
||
if p["cid"] != cid or len(prefix) < p["covered"]:
|
||
raise Refusal("head_proof_malformed")
|
||
covered = prefix[:p["covered"]]
|
||
if p["covered"] == p["n"]:
|
||
if content_hash(covered) != cid:
|
||
raise Refusal("head_proof_mismatch")
|
||
else:
|
||
x, L, c = DOMAIN + covered, 15 + p["n"], 15 + p["covered"]
|
||
it = iter(p["cvs"])
|
||
|
||
def cv(start, ln):
|
||
if start >= c:
|
||
v = next(it, None)
|
||
if v is None:
|
||
raise Refusal("head_proof_mismatch")
|
||
return v
|
||
if start + ln <= c:
|
||
return subtree(x[start:start + ln], start)
|
||
l = left(ln)
|
||
return merge(cv(start, l), cv(start + l, ln - l), False)
|
||
|
||
l = left(L)
|
||
root = merge(cv(0, l), cv(l, L - l), True)
|
||
if next(it, None) is not None or root != cid:
|
||
raise Refusal("head_proof_mismatch")
|
||
try:
|
||
h = parse_head(covered, p["n"])
|
||
except Refusal as e:
|
||
if e.code == "object_length_mismatch":
|
||
raise Refusal("head_proof_malformed")
|
||
raise
|
||
if isinstance(h, tuple) or covered_len(h["head_len"], p["n"]) != p["covered"]:
|
||
raise Refusal("head_proof_malformed")
|
||
|
||
|
||
# ------------------------------------------------------------- responses (§7.1)
|
||
|
||
DIGITS = re.compile(r"[0-9]+", re.ASCII)
|
||
|
||
|
||
def digits(s):
|
||
"""RFC 9110's 1*DIGIT: ASCII digits only, no sign, separator, space or other script."""
|
||
return int(s) if DIGITS.fullmatch(s) else None
|
||
|
||
|
||
def check_response(req, status, cr, enc, body_len, n):
|
||
coded = enc is not None and enc.strip() and enc.strip().lower() != "identity"
|
||
if status == 206:
|
||
if req is None:
|
||
raise Refusal("http_status_unexpected")
|
||
if coded:
|
||
raise Refusal("content_coding_present")
|
||
field = (cr or "").strip(" \t")
|
||
if not field.startswith("bytes ") or "/" not in field:
|
||
raise Refusal("content_range_mismatch")
|
||
span, complete = field[len("bytes "):].split("/", 1)
|
||
if "-" not in span:
|
||
raise Refusal("content_range_mismatch")
|
||
a, b = span.split("-", 1)
|
||
a, b = digits(a), digits(b)
|
||
if a is None or b is None or a != req[0] or b != min(req[1], max(n - 1, 0)) or b < a or body_len != b - a + 1:
|
||
raise Refusal("content_range_mismatch")
|
||
if digits(complete) != n:
|
||
raise Refusal("complete_length_mismatch")
|
||
return (a, b)
|
||
if status == 200:
|
||
if coded:
|
||
raise Refusal("content_coding_present")
|
||
if body_len != n or n == 0:
|
||
raise Refusal("complete_length_mismatch")
|
||
return (0, n - 1)
|
||
raise Refusal("http_status_unexpected")
|
||
|
||
|
||
def origin_serve(n, rng):
|
||
"""An origin's answer to a Range header on n bytes (RFC 9110 §14), as the corpus's
|
||
origin gives it: (status, Content-Range, first, end)."""
|
||
if rng is None or not rng.startswith("bytes=") or "," in rng:
|
||
return 200, None, 0, n
|
||
a, b = rng[len("bytes="):].split("-")
|
||
if a == "":
|
||
s = int(b)
|
||
return 206, "bytes %d-%d/%d" % (max(0, n - s), n - 1, n), max(0, n - s), n
|
||
first = int(a)
|
||
last = n - 1 if b == "" else min(int(b), n - 1)
|
||
if first >= n:
|
||
return 416, "bytes */%d" % n, 0, 0
|
||
return 206, "bytes %d-%d/%d" % (first, last, n), first, last + 1
|
||
|
||
|
||
# ------------------------------------------------------------------ selection
|
||
|
||
def selectable(e):
|
||
return e.get("role") not in ("companion", "refinement") and dispatchable(e["capability"])
|
||
|
||
|
||
def resolve_pin(cap, prior, held):
|
||
if not isinstance(prior, dict) or "sha256" not in prior:
|
||
raise SystemExit("http_verify: a pin shape this verifier does not read")
|
||
form = prior.get("form")
|
||
if form is None:
|
||
return "ok" if any(sha(h) == prior["sha256"] for h in held) else "prior_not_held"
|
||
if form in READABLE:
|
||
return "prior_not_held"
|
||
raise SystemExit("http_verify: a pin form this verifier does not read")
|
||
|
||
|
||
def plaintext_v2(m):
|
||
"""SPEC §6.4 step 2 for a WAI2: `security` absent or null, or an object whose
|
||
`encryption` is absent, null or "none"; a member that is not a security object (a
|
||
string, `true`, a mistyped member) means the payloads cannot be read here."""
|
||
sec = m.get("security")
|
||
if sec is None:
|
||
return True
|
||
if not isinstance(sec, dict):
|
||
return False
|
||
enc, kid = sec.get("encryption"), sec.get("kid")
|
||
if enc is not None and not isinstance(enc, str):
|
||
return False
|
||
if kid is not None and not SV.a_u64(kid):
|
||
return False
|
||
return enc in (None, "none")
|
||
|
||
|
||
def select(m, order, adv, held, digest_ok):
|
||
"""SPEC §7.1 step 2 in `order`: (pick, skipped) or Refusal('inert', skipped)."""
|
||
entries, skipped = m["renditions"], []
|
||
if not plaintext_v2(m):
|
||
raise Refusal("inert", [[i, "unsupported_encryption"] for i in range(len(entries))])
|
||
for i in order:
|
||
if i >= len(entries):
|
||
continue
|
||
e = entries[i]
|
||
if not selectable(e):
|
||
skipped.append([i, "not_selectable"]); continue
|
||
if e["capability"] not in adv:
|
||
skipped.append([i, "missing_capability"]); continue
|
||
if not digest_ok(i, e):
|
||
skipped.append([i, "component_digest_mismatch"]); continue
|
||
if e.get("prior") is not None and pin_applies(e["capability"]):
|
||
r = resolve_pin(e["capability"], e["prior"], held)
|
||
if r != "ok":
|
||
skipped.append([i, r]); continue
|
||
return i, skipped
|
||
raise Refusal("inert", skipped)
|
||
|
||
|
||
def dispatch_wai1(m, adv, held):
|
||
"""SPEC §4 with §6.4 step 2 first: an encrypted payload is inert at a sink that
|
||
opens no scheme."""
|
||
sec = m.get("security")
|
||
if sec is not None and sec.get("encryption") not in (None, "none"):
|
||
raise Refusal("inert", [[0, "unsupported_encryption"]])
|
||
mr = m["model_requirement"]
|
||
cap, fb = mr["capability"], mr.get("fallback")
|
||
if kind(cap) in ("companion", "refinement"):
|
||
raise Refusal("inert", [[0, "missing_capability"]])
|
||
first = None
|
||
for c in [cap, fb]:
|
||
if c is None or not dispatchable(c) or c not in adv:
|
||
continue
|
||
p = m.get("conditioning", {}).get("prior")
|
||
if p is not None and pin_applies(c):
|
||
r = resolve_pin(c, p, held)
|
||
if r == "ok":
|
||
return 0
|
||
first = first or r
|
||
continue
|
||
return 0
|
||
raise Refusal("inert", [[0, first or "missing_capability"]])
|
||
|
||
|
||
def companions_of(entries, i):
|
||
base = entries[i]
|
||
out = []
|
||
for j, e in enumerate(entries):
|
||
if e.get("role") != "companion":
|
||
continue
|
||
cid = e.get("companion_of_id")
|
||
if (cid is not None and base.get("id") == cid) or (cid is None and e.get("companion_of") == base["capability"]):
|
||
out.append(j)
|
||
return out
|
||
|
||
|
||
def refinements_of(entries, i):
|
||
"""The refinement components whose well-formed `refines` (staged-delivery §3, ids in
|
||
place of hashes) names entry i's id as its base."""
|
||
bid = entries[i].get("id")
|
||
if bid is None:
|
||
return []
|
||
out = []
|
||
for j, e in enumerate(entries):
|
||
if e.get("role") != "refinement":
|
||
continue
|
||
try:
|
||
if SV.parse_refines_ids(e.get("refines"))["base"] == bid:
|
||
out.append(j)
|
||
except (SV.Malformed, KeyError, TypeError):
|
||
pass
|
||
return out
|
||
|
||
|
||
def refinements_naming(entries, i):
|
||
"""The refinement components whose `refines.base`, read on its own, is entry i's id:
|
||
refinements_of without the rest of `refines` read. Of a conforming envelope the two
|
||
agree; of one that does not, these are the refinements a plan drops and reports
|
||
(components_nonconforming, http-delivery §7 step 4), a malformed `refines` naming the
|
||
pick among them."""
|
||
if MUTATE == "refines-read-whole":
|
||
return refinements_of(entries, i)
|
||
bid = entries[i].get("id")
|
||
if bid is None:
|
||
return []
|
||
return [j for j, e in enumerate(entries) if e.get("role") == "refinement"
|
||
and isinstance(e.get("refines"), dict) and isinstance(e["refines"].get("base"), str)
|
||
and e["refines"]["base"] == bid]
|
||
|
||
|
||
def payload_ok(e, payload):
|
||
return e.get("sha256") is None or e["sha256"] == sha(payload)
|
||
|
||
|
||
# ------------------------------------------------------------------- the plan
|
||
|
||
def plan(head, wanted, held, permille):
|
||
"""§7 steps 5-6 and the threshold, over base-class bytes only:
|
||
("rest", (first, last), why) or ("ranges", [...], dropped)."""
|
||
n, cov = head["envelope_len"], covered_len(head["head_len"], head["envelope_len"])
|
||
if cov == n:
|
||
return ("ranges", [], [])
|
||
rest = (cov, n - 1)
|
||
entries = head["manifest"]["renditions"]
|
||
for i, cl in wanted:
|
||
if cl == "base" and entries[i].get("sha256") is None:
|
||
return ("rest", rest, "component_digest_absent")
|
||
pieces, dropped = [], []
|
||
for i, cl in wanted:
|
||
if entries[i].get("sha256") is None:
|
||
dropped.append([i, "component_digest_absent"])
|
||
continue
|
||
s, ln = head["components"][i]
|
||
segs = [(s, s + ln)]
|
||
for (hf, hl) in held:
|
||
nxt = []
|
||
for a, b in segs:
|
||
hs, he = hf, hl + 1
|
||
if he <= a or hs >= b:
|
||
nxt.append((a, b)); continue
|
||
if a < hs:
|
||
nxt.append((a, hs))
|
||
if he < b:
|
||
nxt.append((he, b))
|
||
segs = nxt
|
||
pieces += [(0 if cl == "base" else 1, a, b) for a, b in segs]
|
||
pieces.sort()
|
||
merged = []
|
||
for c, a, b in pieces:
|
||
if merged and merged[-1][0] == c and merged[-1][2] >= a:
|
||
merged[-1][2] = max(merged[-1][2], b)
|
||
else:
|
||
merged.append([c, a, b])
|
||
base = sum(b - a for c, a, b in merged if c == 0)
|
||
if base > 0 and base * 1000 >= permille * (n - cov):
|
||
return ("rest", rest, "range_threshold")
|
||
return ("ranges", [(a, b - 1, "base" if c == 0 else "refinement") for c, a, b in merged if b > a], dropped)
|
||
|
||
|
||
# ---------------------------------------------------------------- the codecs
|
||
|
||
def cap_ok(cap):
|
||
parts = cap.split(".")
|
||
lc = lambda s, extra="": s and all(ch in "abcdefghijklmnopqrstuvwxyz0123456789" + extra for ch in s)
|
||
return len(parts) >= 2 and lc(parts[0]) and all(lc(p, "_") for p in parts[1:])
|
||
|
||
|
||
def check_cap(c, first):
|
||
if not cap_ok(c):
|
||
raise Refusal("codecs_malformed")
|
||
full = "wai." + c
|
||
k = kind(full)
|
||
if k in ("derivation", "record"):
|
||
raise Refusal("codecs_not_media")
|
||
if candidate(full):
|
||
raise Refusal("codecs_candidate")
|
||
if first and k in ("companion", "refinement"):
|
||
raise Refusal("codecs_malformed")
|
||
if not first and k == "payload":
|
||
raise Refusal("codecs_malformed")
|
||
|
||
|
||
def check_alt(alt):
|
||
for i, c in enumerate(alt):
|
||
check_cap(c, i == 0)
|
||
|
||
|
||
def parse_element(s):
|
||
if "." not in s:
|
||
raise Refusal("codecs_malformed")
|
||
prefix, rest = s.split(".", 1)
|
||
if prefix not in ("wai1", "wai2"):
|
||
raise Refusal("codecs_malformed")
|
||
alts = []
|
||
for a in rest.split("|"):
|
||
alt = a.split("+")
|
||
check_alt(alt)
|
||
if len(set(alt)) != len(alt) or any(x[0] == alt[0] for x in alts):
|
||
raise Refusal("codecs_malformed")
|
||
alts.append(alt)
|
||
return (prefix, alts)
|
||
|
||
|
||
def render(e):
|
||
return e[0] + "." + "|".join("+".join(a) for a in e[1])
|
||
|
||
|
||
def parse_codecs(value):
|
||
out = []
|
||
for e in (x.strip(" \t") for x in value.split(",")):
|
||
if e == "":
|
||
raise Refusal("codecs_malformed")
|
||
if e.encode("utf-8")[:3].lower() == b"wai":
|
||
out.append(render(parse_element(e)))
|
||
else:
|
||
out.append(None)
|
||
return out
|
||
|
||
|
||
def short(c):
|
||
return c[4:] if c.startswith("wai.") else c
|
||
|
||
|
||
def element_of(head):
|
||
if is_layer(head):
|
||
raise Refusal("codecs_malformed")
|
||
m = head["manifest"]
|
||
if head["version"] == "WAI1":
|
||
mr = m["model_requirement"]
|
||
alts, prefix = [], "wai1"
|
||
for c in [mr["capability"], mr.get("fallback")]:
|
||
if c is not None and dispatchable(c) and short(c) not in [a[0] for a in alts]:
|
||
alts.append([short(c)])
|
||
else:
|
||
e, alts, prefix = m["renditions"], [], "wai2"
|
||
for i, r in enumerate(e):
|
||
if not selectable(r):
|
||
continue
|
||
cap = short(r["capability"])
|
||
a = next((x for x in alts if x[0] == cap), None)
|
||
if a is None:
|
||
a = [cap]; alts.append(a)
|
||
for j in sorted(companions_of(e, i) + refinements_of(e, i)):
|
||
p = short(e[j]["capability"])
|
||
if p not in a[1:]:
|
||
a.append(p)
|
||
pos = lambda p: next(k for k, r in enumerate(e) if short(r["capability"]) == p)
|
||
alts = [[a[0]] + sorted(a[1:], key=pos) for a in alts]
|
||
if not alts:
|
||
raise Refusal("codecs_not_media")
|
||
for a in alts:
|
||
check_alt(a)
|
||
return (prefix, alts)
|
||
|
||
|
||
def seq_object(env):
|
||
h = whole_head(env)
|
||
comps = []
|
||
if h["version"] == "WAI2":
|
||
try:
|
||
comps = [content_hash(x).hex() if x is not None else None for x in SV.explode_each(env)]
|
||
except (SV.NotWai2, SV.NotJSON, SV.Schema, ValueError, KeyError):
|
||
comps = []
|
||
return {"cid": content_hash(env).hex(), "head": h, "components": comps}
|
||
|
||
|
||
def base_capability(b, base):
|
||
"""The capability through which object b is the base whose content hash is `base`: a
|
||
WAI1's capability when its cid is `base`, or the capability of the selectable component
|
||
of a WAI2 whose exploded envelope has that hash; None otherwise (§4.2)."""
|
||
if is_layer(b["head"]):
|
||
return None
|
||
if b["cid"] == base and b["head"]["version"] == "WAI1":
|
||
return short(b["head"]["manifest"]["model_requirement"]["capability"])
|
||
if base in b["components"]:
|
||
r = b["head"]["manifest"]["renditions"][b["components"].index(base)]
|
||
return short(r["capability"]) if selectable(r) else None
|
||
return None
|
||
|
||
|
||
def described(objects):
|
||
els = [None if is_layer(o["head"]) else element_of(o["head"]) for o in objects]
|
||
for o in objects:
|
||
if not is_layer(o["head"]):
|
||
continue
|
||
m = o["head"]["manifest"]
|
||
cap = short(m["model_requirement"]["capability"])
|
||
check_cap(cap, False)
|
||
try:
|
||
base = SV.parse_refines(m.get("refines"))["base"]
|
||
except SV.Malformed:
|
||
raise Refusal("codecs_malformed")
|
||
target = None
|
||
for k, b in enumerate(objects):
|
||
bc = base_capability(b, base)
|
||
if bc is not None:
|
||
target = (k, bc)
|
||
break
|
||
if target is None:
|
||
raise Refusal("codecs_layer_unbound")
|
||
alt = next((a for a in els[target[0]][1] if a[0] == target[1]), None)
|
||
if alt is None:
|
||
raise Refusal("codecs_layer_unbound")
|
||
if cap not in alt[1:]:
|
||
alt.append(cap)
|
||
return [e for e in els if e is not None]
|
||
|
||
|
||
def union_of(objects):
|
||
if MUTATE == "floor-as-codecs":
|
||
return floor_of(objects)
|
||
each = described(objects)
|
||
prefix = "wai2" if any(o["head"]["version"] == "WAI2" for o in objects) else "wai1"
|
||
alts = []
|
||
for p, al in each:
|
||
for a in al:
|
||
x = next((x for x in alts if x[0] == a[0]), None)
|
||
if x is None:
|
||
alts.append(list(a))
|
||
else:
|
||
x += [c for c in a[1:] if c not in x[1:]]
|
||
if not alts:
|
||
raise Refusal("codecs_no_common_capability")
|
||
return (prefix, alts)
|
||
|
||
|
||
def floor_of(objects):
|
||
each = described(objects)
|
||
prefix = "wai2" if any(o["head"]["version"] == "WAI2" for o in objects) else "wai1"
|
||
seen = []
|
||
for e in each:
|
||
for a in e[1]:
|
||
x = next((s for s in seen if s[0] == a[0]), None)
|
||
if x is None:
|
||
seen.append(list(a))
|
||
else:
|
||
x += [c for c in a[1:] if c not in x[1:]]
|
||
out = []
|
||
for a in seen:
|
||
found = [next((x for x in e[1] if x[0] == a[0]), None) for e in each]
|
||
if any(f is None for f in found):
|
||
continue
|
||
out.append([a[0]] + [c for c in a[1:] if all(c in f[1:] for f in found)])
|
||
if not out:
|
||
raise Refusal("codecs_no_common_capability")
|
||
return (prefix, out)
|
||
|
||
|
||
def check_declared(head, union, base):
|
||
"""§4.2: a layer is declared only after `+` in the alt of its own base's capability,
|
||
the base (`base`, a sequence object) being the one its `refines.base` names."""
|
||
if is_layer(head):
|
||
m = head["manifest"]
|
||
cap = short(m["model_requirement"]["capability"])
|
||
try:
|
||
named = SV.parse_refines(m.get("refines"))["base"]
|
||
except SV.Malformed:
|
||
raise Refusal("codecs_undeclared")
|
||
bc = base_capability(base, named) if base is not None else None
|
||
if bc is None or not any(a[0] == bc and cap in a[1:] for a in union[1]):
|
||
raise Refusal("codecs_undeclared")
|
||
return
|
||
p, alts = element_of(head)
|
||
if p > union[0]:
|
||
raise Refusal("codecs_undeclared")
|
||
for a in alts:
|
||
u = next((x for x in union[1] if x[0] == a[0]), None)
|
||
if u is None or any(c not in u[1:] for c in a[1:]):
|
||
raise Refusal("codecs_undeclared")
|
||
|
||
|
||
def supports(floor, containers, caps):
|
||
reads = "WAI1" in containers and (floor[0] == "wai1" or "WAI2" in containers)
|
||
return reads and any(("wai." + a[0]) in caps for a in floor[1])
|
||
|
||
|
||
def parse_layer_tag(line):
|
||
"""§10.4's #WAI-LAYER line, as an HLS attribute list."""
|
||
bad = Refusal("layer_tag_malformed")
|
||
if not line.startswith("#WAI-LAYER:"):
|
||
raise bad
|
||
rest = line[len("#WAI-LAYER:"):].encode("utf-8")
|
||
n, uri, names = None, None, []
|
||
while True:
|
||
eq = rest.find(b"=")
|
||
if eq < 0:
|
||
raise bad
|
||
name = rest[:eq]
|
||
if not name or not re.fullmatch(rb"[A-Z0-9-]+", name) or name in names:
|
||
raise bad
|
||
names.append(name)
|
||
rest = rest[eq + 1:]
|
||
if rest[:1] == b'"':
|
||
close = rest.find(b'"', 1)
|
||
if close < 0:
|
||
raise bad
|
||
value, quoted, after = rest[1:close], True, rest[close + 1:]
|
||
if b"\r" in value or b"\n" in value:
|
||
raise bad
|
||
else:
|
||
end = rest.find(b",")
|
||
end = len(rest) if end < 0 else end
|
||
value, quoted, after = rest[:end], False, rest[end:]
|
||
if not value or b'"' in value or re.search(rb"[ \t\n\r\x0c]", value):
|
||
raise bad
|
||
if name == b"N":
|
||
if quoted or len(value) > 3 or not re.fullmatch(rb"[0-9]+", value) or not 1 <= int(value) <= 255:
|
||
raise bad
|
||
n = int(value)
|
||
elif name == b"URI":
|
||
if not quoted:
|
||
raise bad
|
||
try:
|
||
uri = value.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
raise bad
|
||
if not after:
|
||
break
|
||
if after[:1] != b",":
|
||
raise bad
|
||
rest = after[1:]
|
||
if n is None or uri is None:
|
||
raise bad
|
||
return {"n": n, "uri": uri}
|
||
|
||
|
||
# ------------------------------------------------------------- the session (§7)
|
||
|
||
class Session:
|
||
"""The reference client of §7, step by step as the requests are answered: it says
|
||
which requests to make, checks each response, and decides as each arrives."""
|
||
|
||
def __init__(self, cid, frag, permille, sink, layer=False, deadline=None):
|
||
self.cid, self.frag, self.permille, self.sink, self.deadline = cid, frag, permille, sink, deadline
|
||
self.next_id, self.queued, self.issued, self.early = 0, [], {}, []
|
||
self.proof, self.held, self.head, self.stage = None, [], None, ("head",)
|
||
self.abandoned, self.dropped, self.body, self.requests, self.result = [], [], 0, 0, None
|
||
# §10.1, §10.2: the caller's clock as last given; the base once it verified, with
|
||
# the time of the response that verified it; each verified component, frozen.
|
||
self.now, self.base, self.base_at, self.verified = 0, None, None, {}
|
||
if layer:
|
||
self.stage = ("layer",)
|
||
self.queue(("whole", "refinement"))
|
||
return
|
||
n = frag.get("n")
|
||
if n is not None and n < 1009:
|
||
self.stage = ("rest", "below_one_chunk", "base")
|
||
self.queue(("whole", "base"))
|
||
return
|
||
e = frag.get("h") or 16369
|
||
if n is not None:
|
||
e = min(e, n)
|
||
self.queue(("range", (0, e - 1), "base"))
|
||
self.queue(("proof",))
|
||
|
||
@staticmethod
|
||
def cls(w):
|
||
return "base" if w[0] == "proof" else w[-1]
|
||
|
||
def queue(self, w):
|
||
self.queued.append((self.next_id, w)); self.next_id += 1
|
||
|
||
def wants(self):
|
||
out, self.queued = self.queued, []
|
||
for i, w in out:
|
||
self.issued[i] = w
|
||
self.requests += 1
|
||
return out
|
||
|
||
def finish(self, r):
|
||
self.stage, self.queued, self.result = ("done",), [], r
|
||
|
||
def abandon(self, w):
|
||
st = self.stage[0]
|
||
if st == "layer" or (st == "rest" and self.stage[2] == "refinement"):
|
||
if 0 not in self.abandoned:
|
||
self.abandoned.append(0)
|
||
elif st == "components" and w[0] == "range" and w[2] == "refinement":
|
||
(a, b) = w[1]
|
||
for i, cl in self.stage[3]:
|
||
s, ln = self.head["components"][i]
|
||
if cl == "refinement" and s <= b and a < s + ln and i not in self.abandoned:
|
||
self.abandoned.append(i)
|
||
|
||
def on_deadline(self, i):
|
||
if self.result is not None:
|
||
return
|
||
w = self.issued.get(i)
|
||
if w is not None and self.cls(w) == "refinement":
|
||
del self.issued[i]
|
||
self.abandon(w)
|
||
self.advance()
|
||
|
||
def base_ready(self):
|
||
if self.base is not None:
|
||
return self.base
|
||
return self.result if self.result is not None and "error" not in self.result else None
|
||
|
||
def tick(self, now):
|
||
"""§10.2: the deadline runs from the response that verified the base; the sum
|
||
saturates at the end of u64, and a clock that steps back has not passed it."""
|
||
self.now = now
|
||
if self.base_at is None or self.deadline is None or MUTATE == "tick-noop":
|
||
return []
|
||
if self.result is not None or now < min(self.base_at + self.deadline, 2 ** 64 - 1):
|
||
return []
|
||
ids = sorted(i for i, w in self.issued.items() if self.cls(w) == "refinement")
|
||
for i in ids:
|
||
self.on_deadline(i)
|
||
return ids
|
||
|
||
def on_response(self, i, status, cr, enc, body, now):
|
||
self.now = now
|
||
if self.result is not None and "error" in self.result:
|
||
raise Refusal(self.result["error"], self.result.get("skipped"))
|
||
w = self.issued.pop(i, None)
|
||
if w is None or self.result is not None:
|
||
return
|
||
self.body += len(body)
|
||
try:
|
||
self.take(w, status, cr, enc, body)
|
||
self.advance()
|
||
except Refusal as e:
|
||
self.finish(refusal(e))
|
||
raise
|
||
|
||
def n(self):
|
||
return self.proof["n"] if self.proof else None
|
||
|
||
def take(self, w, status, cr, enc, body):
|
||
if w[0] == "proof":
|
||
if status != 200:
|
||
raise Refusal("http_status_unexpected")
|
||
if enc is not None and enc.strip() and enc.strip().lower() != "identity":
|
||
raise Refusal("content_coding_present")
|
||
p = read_proof(body)
|
||
if p["cid"] != self.cid:
|
||
raise Refusal("head_proof_malformed")
|
||
self.proof = p
|
||
early, self.early = self.early, []
|
||
for x in early:
|
||
self.take(*x)
|
||
return
|
||
if w[0] == "whole":
|
||
try:
|
||
check_response(None, status, cr, enc, len(body), len(body))
|
||
except Refusal:
|
||
if w[1] == "refinement":
|
||
self.abandon(w); return
|
||
raise
|
||
if content_hash(body) != self.cid:
|
||
raise Refusal("object_digest_mismatch")
|
||
if body:
|
||
self.held.append((0, len(body) - 1, body))
|
||
return
|
||
if self.proof is None:
|
||
self.early.append((w, status, cr, enc, body)); return
|
||
try:
|
||
a, b = check_response(w[1], status, cr, enc, len(body), self.n())
|
||
except Refusal:
|
||
if w[2] == "refinement":
|
||
self.abandon(w); return
|
||
raise
|
||
if w[2] == "refinement" and MUTATE != "refinement-unclipped":
|
||
# A refinement's response adds the bytes it asked for and no others (§10.1).
|
||
first, last = max(a, w[1][0]), min(b, w[1][1])
|
||
if first <= last:
|
||
self.held.append((first, last, body[first - a:last + 1 - a]))
|
||
return
|
||
self.held.append((a, b, body))
|
||
|
||
def have(self, first, end):
|
||
out, at = b"", first
|
||
while at < end:
|
||
best = max((h for h in self.held if h[0] <= at <= h[1]), key=lambda h: h[1], default=None)
|
||
if best is None:
|
||
return None
|
||
take = min(best[1] + 1, end)
|
||
out += best[2][at - best[0]:take - best[0]]
|
||
at = take
|
||
return out
|
||
|
||
def drop_from(self, frm):
|
||
kept = []
|
||
for a, b, body in self.held:
|
||
if a >= frm:
|
||
continue
|
||
kept.append((a, b, body) if b < frm else (a, frm - 1, body[:frm - a]))
|
||
self.held = kept
|
||
|
||
def prefix_len(self):
|
||
at = 0
|
||
while True:
|
||
e = max((h[1] + 1 for h in self.held if h[0] <= at <= h[1]), default=None)
|
||
if e is None:
|
||
return at
|
||
at = e
|
||
|
||
def advance(self):
|
||
if self.result is not None:
|
||
return
|
||
st = self.stage[0]
|
||
if st == "head":
|
||
self.advance_head()
|
||
elif st == "components":
|
||
self.advance_components()
|
||
elif st in ("rest", "layer"):
|
||
self.advance_rest()
|
||
|
||
def to_rest(self, why, cl):
|
||
self.stage = ("rest", why, cl)
|
||
n, cov = self.head["envelope_len"], covered_len(self.head["head_len"], self.head["envelope_len"])
|
||
if cov < n and self.have(cov, n) is None:
|
||
self.queue(("rest", (cov, n - 1), cl))
|
||
|
||
def nonconforming(self, entries, wanted):
|
||
"""§7 step 4: the refinement components among `wanted` of an envelope whose
|
||
components do not all conform (SPEC §7.1) — every one of them where it does not,
|
||
none where it does."""
|
||
conforms = SV.component_roles(entries) is None and MUTATE != "conforming-withheld"
|
||
if conforms or MUTATE == "nonconforming-offered":
|
||
return []
|
||
return [i for i, _ in wanted if entries[i].get("role") == "refinement"]
|
||
|
||
def applied(self, entries, pick):
|
||
adv, applies = self.sink[0], self.sink[1]
|
||
out = [(j, "refinement" if enhancement(entries[j]["capability"]) else "base")
|
||
for j in companions_of(entries, pick) if entries[j]["capability"] in applies]
|
||
out += [(j, "refinement") for j in refinements_naming(entries, pick) if entries[j]["capability"] in applies]
|
||
return out
|
||
|
||
def advance_head(self):
|
||
if self.proof is None:
|
||
return
|
||
cov = self.proof["covered"]
|
||
h0 = self.prefix_len()
|
||
if h0 < cov:
|
||
pending = [w for w in list(self.issued.values()) + [w for _, w in self.queued] if w[0] in ("range", "rest") and w[-1] == "base"]
|
||
if not pending:
|
||
self.queue(("range", (h0, cov - 1), "base"))
|
||
return
|
||
prefix = self.have(0, cov)
|
||
verify_proof(self.proof, self.cid, prefix)
|
||
head = parse_head(prefix, self.proof["n"])
|
||
if isinstance(head, tuple):
|
||
raise Refusal("head_malformed")
|
||
adv, applies, held_sets, order, declared, base = self.sink
|
||
if declared is not None:
|
||
check_declared(head, declared, base)
|
||
self.head = head
|
||
if is_layer(head):
|
||
self.to_rest("layer", "refinement")
|
||
elif head["version"] == "WAI1":
|
||
dispatch_wai1(head["manifest"], adv, held_sets)
|
||
self.to_rest("single_component", "base")
|
||
else:
|
||
entries = head["manifest"]["renditions"]
|
||
ordr = order(head) if callable(order) else (order if order is not None else list(range(len(entries))))
|
||
pick, skipped = select(head["manifest"], ordr, adv, held_sets, lambda i, e: True)
|
||
wanted = [(pick, "base")] + self.applied(entries, pick)
|
||
nc = self.nonconforming(entries, wanted)
|
||
planned = [(i, c) for i, c in wanted if i not in nc or entries[i].get("sha256") is None]
|
||
pl = plan(head, planned, [(a, b) for a, b, _ in self.held], self.permille)
|
||
if pl[0] == "rest":
|
||
self.to_rest(pl[2], "base")
|
||
else:
|
||
for (a, b, cl) in pl[1]:
|
||
self.queue(("range", (a, b), cl))
|
||
self.dropped = pl[2] + [[i, "components_nonconforming"] for i in nc]
|
||
gone = [i for i, _ in pl[2]]
|
||
self.stage = ("components", pick, skipped, [(i, c) for i, c in wanted if i not in gone])
|
||
if self.stage[0] == "components":
|
||
self.advance_components()
|
||
elif self.stage[0] == "rest":
|
||
self.advance_rest()
|
||
|
||
def advance_components(self):
|
||
_, pick, skipped, wanted = self.stage
|
||
entries = self.head["manifest"]["renditions"]
|
||
parts, pending, base_pending = {}, False, False
|
||
for i, cl in wanted:
|
||
if i in self.abandoned or any(d == i for d, _ in self.dropped):
|
||
continue
|
||
if i in self.verified:
|
||
parts[i] = self.verified[i]
|
||
continue
|
||
s, ln = self.head["components"][i]
|
||
bts = self.have(s, s + ln)
|
||
if bts is None:
|
||
pending = True
|
||
base_pending = base_pending or cl == "base"
|
||
elif entries[i].get("sha256") is not None and sha(bts) == entries[i]["sha256"]:
|
||
parts[i] = self.verified[i] = bts
|
||
elif cl == "refinement":
|
||
self.dropped.append([i, "component_digest_mismatch"])
|
||
else:
|
||
self.issued, self.queued, self.abandoned, self.dropped, self.verified = {}, [], [], [], {}
|
||
self.drop_from(covered_len(self.head["head_len"], self.head["envelope_len"]))
|
||
self.to_rest("component_digest_mismatch", "base")
|
||
self.advance_rest()
|
||
return
|
||
classes = dict(wanted)
|
||
|
||
def assemble(only_base):
|
||
js = [j for j, _ in wanted[1:] if j in parts and not (only_base and classes[j] == "refinement")]
|
||
comps = [j for j in js if entries[j].get("role") == "companion"]
|
||
refs = [j for j in js if entries[j].get("role") != "companion"]
|
||
return self.fetched("base", pick, parts[pick], comps, refs, self.abandoned, self.dropped, skipped, "head-proof", None)
|
||
if not (pending if MUTATE == "base-at-done" else base_pending) and self.base is None:
|
||
self.base, self.base_at = assemble(True), self.now
|
||
if pending:
|
||
return
|
||
self.finish(assemble(False))
|
||
|
||
def fetched(self, role, pick, comp, comps, refs, abandoned, dropped, skipped, by, why):
|
||
return {"role": role, "pick": pick, "verified_by": by, "why_whole": why, "body_bytes": self.body, "requests": self.requests,
|
||
"companions": comps, "refinements": refs, "abandoned": list(abandoned), "dropped": [list(x) for x in dropped],
|
||
"skipped": skipped, "component_sha256": sha(comp)}
|
||
|
||
def advance_rest(self):
|
||
layer_listed = self.stage[0] == "layer"
|
||
why = "layer" if layer_listed else self.stage[1]
|
||
if 0 in self.abandoned and (layer_listed or why == "layer"):
|
||
self.finish(self.fetched("layer", 0, b"", [], [], [0], [], [], "whole-object", "layer"))
|
||
return
|
||
if self.proof is not None:
|
||
n = self.proof["n"]
|
||
elif self.held and self.held[0][0] == 0:
|
||
n = self.held[0][1] + 1
|
||
else:
|
||
return
|
||
envb = self.have(0, n)
|
||
if envb is None:
|
||
return
|
||
if content_hash(envb) != self.cid:
|
||
raise Refusal("object_digest_mismatch")
|
||
head = parse_head(envb, n)
|
||
adv, applies, held_sets, order, declared, base = self.sink
|
||
if declared is not None:
|
||
check_declared(head, declared, base)
|
||
if is_layer(head):
|
||
s, ln = head["components"][0]
|
||
self.finish(self.fetched("layer", 0, envb[s:s + ln], [], [], [], [], [], "whole-object", "layer"))
|
||
return
|
||
if layer_listed:
|
||
raise Refusal("not_a_layer")
|
||
m = head["manifest"]
|
||
if head["version"] == "WAI1":
|
||
dispatch_wai1(m, adv, held_sets)
|
||
s, ln = head["components"][0]
|
||
self.finish(self.fetched("base", 0, envb[s:s + ln], [], [], [], [], [], "whole-object", why))
|
||
return
|
||
entries = m["renditions"]
|
||
pay = [envb[s:s + ln] for s, ln in head["components"]]
|
||
ordr = order(head) if callable(order) else (order if order is not None else list(range(len(entries))))
|
||
pick, skipped = select(m, ordr, adv, held_sets, lambda i, e: payload_ok(e, pay[i]))
|
||
comps, refs, dropped = [], [], []
|
||
applied = self.applied(entries, pick)
|
||
nc = self.nonconforming(entries, applied)
|
||
for j, cl in applied:
|
||
# §7 step 4: not offered from an envelope that does not conform; one stating no
|
||
# digest is dropped for that too, as on the range path.
|
||
if j in nc:
|
||
if entries[j].get("sha256") is None:
|
||
dropped.append([j, "component_digest_absent"])
|
||
continue
|
||
# A refinement stating no digest is dropped here as on the range path (§7 step 5).
|
||
if cl == "refinement" and entries[j].get("sha256") is None:
|
||
dropped.append([j, "component_digest_absent"])
|
||
continue
|
||
if not payload_ok(entries[j], pay[j]):
|
||
if cl == "refinement":
|
||
dropped.append([j, "component_digest_mismatch"])
|
||
continue
|
||
(comps if entries[j].get("role") == "companion" else refs).append(j)
|
||
dropped += [[j, "components_nonconforming"] for j in nc]
|
||
self.finish(self.fetched("base", pick, pay[pick], comps, refs, [], dropped, skipped, "whole-object", why))
|
||
|
||
|
||
def drive(s, answer):
|
||
"""Run a session: `answer(want)` gives (status, Content-Range, Content-Encoding,
|
||
body) or None for a request that never arrives."""
|
||
while True:
|
||
if s.result is not None:
|
||
if "error" in s.result:
|
||
raise Refusal(s.result["error"], s.result.get("skipped"))
|
||
return s.result
|
||
ws = s.wants()
|
||
if not ws:
|
||
for i in [i for i, w in s.issued.items() if Session.cls(w) == "refinement"]:
|
||
s.on_deadline(i)
|
||
if s.result is not None:
|
||
continue
|
||
if not s.queued:
|
||
if s.issued:
|
||
raise Refusal("base_unanswered")
|
||
raise SystemExit("http_verify: the session stalled")
|
||
continue
|
||
for i, w in ws:
|
||
r = answer(w)
|
||
if r is not None:
|
||
s.on_response(i, *r, s.now)
|
||
|
||
|
||
def fetch(env, served, frag, sink, permille, tamper, listed_as_layer=False):
|
||
"""The reference client against an origin serving `served` under the name of `env`,
|
||
with the corpus's tampers."""
|
||
proof_bytes = make_proof_file(env) if not listed_as_layer else b""
|
||
n_real = len(served)
|
||
kind_t, ref_t, first_flip = tamper.get("kind"), tamper.get("refinement"), tamper.get("first_range_flip")
|
||
state = {"first": True}
|
||
|
||
def answer(w):
|
||
cl = Session.cls(w)
|
||
if kind_t == "no_refinement" and cl == "refinement":
|
||
return None
|
||
if w[0] == "proof":
|
||
return None if kind_t == "no_proof" else (200, None, None, proof_bytes)
|
||
rng = None if w[0] == "whole" else ("bytes=%d-%d" % w[1] if w[0] == "range" else "bytes=%d-" % w[1][0])
|
||
status, cr, a, b = origin_serve(n_real, rng)
|
||
body = served[a:b]
|
||
enc = "gzip" if kind_t == "gzip" else None
|
||
if first_flip is not None and w[0] == "range" and state["first"]:
|
||
state["first"] = False
|
||
if a <= first_flip < b:
|
||
body = bytearray(body); body[first_flip - a] ^= 1; body = bytes(body)
|
||
if ref_t is not None and cl == "refinement" and w[0] == "range":
|
||
(ra, rb) = w[1]
|
||
if ref_t == "gzip":
|
||
enc = "gzip"
|
||
elif ref_t == "bad_range":
|
||
cr = "bytes %d-%d/%d" % (ra + 1, rb + 1, n_real)
|
||
elif ref_t == "416":
|
||
return (416, "bytes */%d" % n_real, None, b"")
|
||
elif ref_t == "complete_length":
|
||
cr = "bytes %d-%d/1" % (ra, rb)
|
||
elif ref_t == "plus_sign":
|
||
cr = cr.replace("bytes ", "bytes +", 1)
|
||
elif ref_t == "short_200":
|
||
return (200, None, None, body)
|
||
return (status, cr, enc, body)
|
||
|
||
s = Session(content_hash(env), frag, permille, sink, layer=listed_as_layer)
|
||
return drive(s, answer)
|
||
|
||
|
||
def fetch_timed(env, frag, sink, permille, deadline, script, flip):
|
||
"""The reference client on a clock (§10.1, §10.2), driven by the corpus's script."""
|
||
proof_bytes = make_proof_file(env)
|
||
n = len(env)
|
||
|
||
def answer(w):
|
||
if w[0] == "proof":
|
||
return (200, None, None, proof_bytes)
|
||
rng = None if w[0] == "whole" else ("bytes=%d-%d" % w[1] if w[0] == "range" else "bytes=%d-" % w[1][0])
|
||
status, cr, a, b = origin_serve(n, rng)
|
||
return (status, cr, None, env[a:b])
|
||
|
||
def summary(f):
|
||
return {"pick": f["pick"], "verified_by": f["verified_by"], "component_sha256": f["component_sha256"],
|
||
"companions": f["companions"], "refinements": f["refinements"]}
|
||
|
||
s = Session(content_hash(env), frag, permille, sink, deadline=deadline)
|
||
pending, ready_step, first, ticks = [], None, None, []
|
||
stop = False
|
||
for k, step in enumerate(script):
|
||
kind, t = step[0], step[1]
|
||
if kind in ("base", "base_once", "refinement"):
|
||
cl = "refinement" if kind == "refinement" else "base"
|
||
while True:
|
||
pending += s.wants()
|
||
now = [(i, w) for i, w in pending if Session.cls(w) == cl]
|
||
if not now:
|
||
break
|
||
pending = [(i, w) for i, w in pending if Session.cls(w) != cl]
|
||
for i, w in now:
|
||
if len(step) > 2 and step[2] == "whole_200":
|
||
b = bytearray(env)
|
||
if flip is not None:
|
||
b[flip] ^= 1
|
||
r = (200, None, None, bytes(b))
|
||
else:
|
||
r = answer(w)
|
||
try:
|
||
s.on_response(i, *r, t)
|
||
except Refusal:
|
||
stop = True
|
||
break
|
||
if stop or kind != "base":
|
||
break
|
||
elif kind == "tick":
|
||
ids = s.tick(t)
|
||
pending = [(i, w) for i, w in pending if i not in ids]
|
||
ticks.append([t, ids])
|
||
else:
|
||
raise SystemExit("unknown step " + kind)
|
||
if ready_step is None and s.base_ready() is not None:
|
||
ready_step, first = k, summary(s.base_ready())
|
||
if stop:
|
||
break
|
||
end = s.base_ready()
|
||
return {"ready_step": ready_step, "ready_at": s.base_at, "base": first, "base_end": summary(end) if end is not None else None,
|
||
"ticks": ticks, "outcome": s.result}
|
||
|
||
|
||
def make_proof_file(env):
|
||
"""A producer's head proof (§6.2), written from the text: the chaining values of the
|
||
subtrees at or after c, in the order §6.3 consumes them."""
|
||
h = whole_head(env)
|
||
cov = covered_len(h["head_len"], len(env))
|
||
cvs = []
|
||
if cov < len(env):
|
||
x, L, c = DOMAIN + env, 15 + len(env), 15 + cov
|
||
|
||
def walk(start, ln):
|
||
if start >= c:
|
||
cvs.append(B3.subtree_cv(x[start:start + ln], start))
|
||
elif start + ln > c:
|
||
l = 1024 * 2 ** ((((ln - 1) // 1024)).bit_length() - 1)
|
||
walk(start, l); walk(start + l, ln - l)
|
||
|
||
l = 1024 * 2 ** ((((L - 1) // 1024)).bit_length() - 1)
|
||
walk(0, l); walk(l, L - l)
|
||
return b"WHP1" + content_hash(env) + len(env).to_bytes(8, "little") + cov.to_bytes(8, "little") + bytes([len(cvs)]) + b"".join(cvs)
|
||
|
||
|
||
# ----------------------------------------------------------------- the cases
|
||
|
||
def apply_patch(b, patch):
|
||
b = bytearray(b)
|
||
for off, hx in patch or []:
|
||
bs = bytes.fromhex(hx)
|
||
b[off:off + len(bs)] = bs
|
||
return bytes(b)
|
||
|
||
|
||
def frag_of(s):
|
||
out = {}
|
||
if s and "#" in s:
|
||
for kv in s.split("#", 1)[1].split("&"):
|
||
if "=" in kv:
|
||
k, v = kv.split("=", 1)
|
||
if k in ("n", "h") and v.isdigit():
|
||
out[k] = int(v)
|
||
return out
|
||
|
||
|
||
def sink_of(d, inp, objs):
|
||
"""The sink a case describes. Its declared union is parsed, and the base a layer is
|
||
declared against read, before anything is fetched: either refuses the case."""
|
||
held = [open(os.path.join(d, p), "rb").read() for p in inp.get("held") or []]
|
||
declared = parse_element(inp["declared"]) if isinstance(inp.get("declared"), str) else None
|
||
base = None
|
||
if inp.get("layer_base") is not None or inp.get("layer_base_hex") is not None:
|
||
b = bytes.fromhex(inp["layer_base_hex"]) if isinstance(inp.get("layer_base_hex"), str) else objs[inp["layer_base"]]
|
||
base = seq_object(b)
|
||
order = inp.get("order")
|
||
if inp.get("abr") is not None:
|
||
adv, pol, st = inp.get("advertises") or [], inp["abr"]["policy"], abr_state(inp["abr"]["state"])
|
||
check_policy(pol)
|
||
check_state(st)
|
||
|
||
def order(head):
|
||
try:
|
||
return order_entries(head, adv, pol, st)
|
||
except Refusal:
|
||
return list(range(len(head["manifest"]["renditions"])))
|
||
return (inp.get("advertises") or [], inp.get("applies") or [], held, order, declared, base)
|
||
|
||
|
||
# ------------------------------------------------- adaptation (Appendix F)
|
||
|
||
LOG2_TABLE = [0, 5732, 11136, 16248, 21098, 25711, 30109, 34312, 38336, 42196, 45904, 49472, 52911, 56229, 59434, 62534, 65536]
|
||
|
||
|
||
def ilog2_q16(a, b):
|
||
"""log2(a/b) in Q16: the ratio's exponent, and the table interpolated on the
|
||
mantissa's top four fractional bits with the next twelve as the weight."""
|
||
b = max(b, 1)
|
||
if a <= b:
|
||
return 0
|
||
x = (a << 16) // b
|
||
e = x.bit_length() - 1 - 16
|
||
f = (x >> e) - 65536
|
||
i, r = f >> 12, f & 4095
|
||
if MUTATE == "ilog2-no-interp":
|
||
r = 0
|
||
return e * 65536 + LOG2_TABLE[i] + (((LOG2_TABLE[i + 1] - LOG2_TABLE[i]) * r) >> 12)
|
||
|
||
|
||
def throughput_bps(samples, n):
|
||
k = min(max(n, 1), len(samples))
|
||
tail = samples[len(samples) - k:]
|
||
t = sum(max(s[2], 1) for s in tail)
|
||
return sum(s[1] for s in tail) * 8000 // t if t else 0
|
||
|
||
|
||
def option_bps(nbytes, seg):
|
||
return -(-(nbytes * 8000) // max(seg, 1))
|
||
|
||
|
||
def highest_under(st, budget):
|
||
for k in range(len(st["option_bytes"]) - 1, -1, -1):
|
||
if option_bps(st["option_bytes"][k], st["segment_ms"]) <= budget:
|
||
return k
|
||
return 0
|
||
|
||
|
||
def utilities(st):
|
||
s0 = max(st["option_bytes"][0] if st["option_bytes"] else 1, 1)
|
||
return [ilog2_q16(max(b, 1), s0) for b in st["option_bytes"]]
|
||
|
||
|
||
def choose(p, st):
|
||
opts, seg, k = st["option_bytes"], st["segment_ms"], len(st["option_bytes"])
|
||
if k == 0:
|
||
return 0
|
||
if p["id"] == "throughput":
|
||
return highest_under(st, throughput_bps(st["samples"], p["last_n"]) * p["safety_permille"] // 1000)
|
||
if p["id"] == "buffer_based":
|
||
b, r, c = st["buffer_ms"], p["reservoir_ms"], p["cushion_ms"]
|
||
if b <= r:
|
||
return 0
|
||
if c == 0 or b >= r + c:
|
||
return k - 1
|
||
lo, hi = option_bps(opts[0], seg), option_bps(opts[-1], seg)
|
||
return highest_under(st, lo + max(hi - lo, 0) * (b - r) // c)
|
||
if p["id"] == "bola_like":
|
||
sg = max(seg, 1)
|
||
q, qmax = st["buffer_ms"] * 65536 // sg, p["max_buffer_ms"] * 65536 // sg
|
||
u = utilities(st)
|
||
vn, vd = max(qmax - 65536, 0), max(u) + 65536
|
||
score = [vn * p["v_milli"] * (x + 65536) // (vd * 1000) - q for x in u]
|
||
size = [max(b, 1) for b in opts]
|
||
best = None
|
||
if any(x > 0 for x in score):
|
||
for i in range(k):
|
||
if score[i] > 0 and (best is None or score[i] * size[best] > score[best] * size[i]):
|
||
best = i
|
||
elif MUTATE == "bola-waits":
|
||
return 0
|
||
else:
|
||
for i in range(k):
|
||
if best is None or score[i] > score[best]:
|
||
best = i
|
||
return best
|
||
if p["id"] == "mpc":
|
||
pred = throughput_bps(st["samples"], 5) * p["safety_permille"] // 1000
|
||
if pred == 0:
|
||
return 0
|
||
h = min(max(p["horizon"], 1), 5)
|
||
while h > 1 and k ** h > 65536:
|
||
h -= 1
|
||
u = utilities(st)
|
||
u_top = max(u)
|
||
best = None
|
||
for code in range(k ** h):
|
||
seq, c = [0] * h, code
|
||
for t in range(h - 1, -1, -1):
|
||
seq[t], c = c % k, c // k
|
||
b, rebuf, qoe = st["buffer_ms"], 0, 0
|
||
last = st["last"]
|
||
prev = u[last] if last is not None and last < k else None
|
||
for o in seq:
|
||
dl = -(-(opts[o] * 8000) // pred)
|
||
if dl > b:
|
||
rebuf, b = rebuf + dl - b, 0
|
||
else:
|
||
b -= dl
|
||
b += seg
|
||
qoe += 1000 * u[o]
|
||
if prev is not None:
|
||
qoe -= 1000 * abs(u[o] - prev)
|
||
prev = u[o]
|
||
qoe -= rebuf * (u_top + 65536)
|
||
if best is None or qoe > best[0]:
|
||
best = (qoe, seq[0])
|
||
return best[1]
|
||
raise SystemExit("http_verify: an unknown policy " + str(p["id"]))
|
||
|
||
|
||
def clamp_to_budget(st, choice):
|
||
"""Appendix F's budget: it never raises the choice; the highest option at or below it
|
||
whose labelled figure is within it, else option 0 unmet; not applied only when no
|
||
option has a labelled figure."""
|
||
budget = st["joule_budget_micro"]
|
||
if budget is None:
|
||
return choice, "not_set"
|
||
figs = st["option_joules_micro"]
|
||
n = max(len(st["option_bytes"]), choice + 1)
|
||
fig = lambda j: figs[j] if j < len(figs) and j < n else None
|
||
if MUTATE == "unlabelled-as-zero":
|
||
fig = lambda j: (figs[j] if j < len(figs) and j < n else None) or 0
|
||
labelled = [(j, fig(j)) for j in range(n) if fig(j) is not None]
|
||
if not labelled or (MUTATE == "unlabelled-escapes" and fig(choice) is None):
|
||
return choice, "not_applied"
|
||
if fig(choice) is not None and fig(choice) <= budget:
|
||
return choice, "within"
|
||
below = [j for j, f in labelled if j < choice and f <= budget]
|
||
if below:
|
||
return below[-1], "lowered"
|
||
if MUTATE == "budget-raises":
|
||
above = [j for j, f in labelled if j > choice and f <= budget]
|
||
if above:
|
||
return above[0], "unmet"
|
||
return 0, "unmet"
|
||
|
||
|
||
MAX_OPTIONS = 256
|
||
OUT_OF_BOUNDS = "abr_state_out_of_bounds"
|
||
|
||
|
||
def bounded(v, top):
|
||
"""An unsigned integer of at most top; out of bounds for any other number."""
|
||
if isinstance(v, bool) or not isinstance(v, (int, float)):
|
||
raise SystemExit("http_verify: an adaptation value is not a number")
|
||
if not isinstance(v, int) or not 0 <= v <= top:
|
||
raise Refusal(OUT_OF_BOUNDS)
|
||
return v
|
||
|
||
|
||
def check_policy(p):
|
||
"""Appendix F: each parameter within its type."""
|
||
tops = {"throughput": {"last_n": 255, "safety_permille": 65535}, "buffer_based": {"reservoir_ms": U32, "cushion_ms": U32},
|
||
"bola_like": {"v_milli": U32, "max_buffer_ms": U32}, "mpc": {"horizon": 255, "safety_permille": 65535}}
|
||
if p.get("id") not in tops:
|
||
raise SystemExit("http_verify: an unknown policy " + str(p.get("id")))
|
||
for k, top in tops[p["id"]].items():
|
||
bounded(p[k], top)
|
||
|
||
|
||
def check_state(st):
|
||
if (len(st["option_bytes"]) > MAX_OPTIONS or len(st["option_joules_micro"]) > MAX_OPTIONS) and MUTATE != "abr-unbounded":
|
||
raise Refusal(OUT_OF_BOUNDS)
|
||
|
||
|
||
def decide(p, st):
|
||
check_policy(p)
|
||
check_state(st)
|
||
return clamp_to_budget(st, choose(p, st))
|
||
|
||
|
||
def abr_state(v):
|
||
"""A state from its JSON form, each value within its field's type (Appendix F)."""
|
||
opt = lambda x, top: None if x is None else bounded(x, top)
|
||
return {"buffer_ms": bounded(v["buffer_ms"], U32), "samples": [tuple(bounded(x, U64) for x in s[:3]) for s in v.get("samples") or []],
|
||
"last": opt(v.get("last"), U64), "option_bytes": [bounded(x, U64) for x in v.get("option_bytes") or []],
|
||
"option_joules_micro": [opt(x, U64) for x in v.get("option_joules_micro") or []],
|
||
"joule_budget_micro": opt(v.get("joule_budget_micro"), U64), "segment_ms": bounded(v["segment_ms"], U32)}
|
||
|
||
|
||
def depth(layers_n, p, st):
|
||
check_policy(p)
|
||
check_state(st)
|
||
if len(layers_n) > MAX_OPTIONS:
|
||
raise Refusal(OUT_OF_BOUNDS)
|
||
st = dict(st)
|
||
acc, cum, jcum, fsum = 0, [], [], 0
|
||
figs = st["option_joules_micro"]
|
||
for k, n in enumerate(layers_n):
|
||
acc += n
|
||
if acc > U64 and MUTATE != "abr-unbounded":
|
||
raise Refusal(OUT_OF_BOUNDS)
|
||
cum.append(acc)
|
||
f = figs[k] if k < len(figs) else None
|
||
fsum = None if fsum is None or f is None else fsum + f
|
||
if fsum is not None and fsum > U64 and MUTATE != "abr-unbounded":
|
||
raise Refusal(OUT_OF_BOUNDS)
|
||
jcum.append(fsum)
|
||
st["option_bytes"], st["option_joules_micro"] = cum, jcum
|
||
if not layers_n:
|
||
return 0, clamp_to_budget(st, 0)[1]
|
||
return clamp_to_budget(st, choose(p, st))
|
||
|
||
|
||
def order_entries(head, adv, p, st):
|
||
"""SPEC §7.1 step 2's order under a policy: each advertised selectable capability's
|
||
entries by component length, the choice first, then those below it nearest first,
|
||
then those above; every other entry after, in table order."""
|
||
check_policy(p)
|
||
check_state(st)
|
||
if head["version"] != "WAI2":
|
||
return [0]
|
||
entries = head["manifest"]["renditions"]
|
||
ln = lambda i: head["components"][i][1]
|
||
groups = []
|
||
for i, e in enumerate(entries):
|
||
if not selectable(e) or e["capability"] not in adv:
|
||
continue
|
||
g = next((g for g in groups if g[0] == e["capability"]), None)
|
||
if g is None:
|
||
groups.append((e["capability"], [i]))
|
||
else:
|
||
g[1].append(i)
|
||
order = []
|
||
for _, g in groups:
|
||
g = sorted(g, key=lambda i: (ln(i), i))
|
||
s2 = dict(st)
|
||
figs = st["option_joules_micro"]
|
||
s2["option_bytes"] = [ln(i) for i in g]
|
||
s2["option_joules_micro"] = [figs[i] if i < len(figs) else None for i in g]
|
||
c = min(decide(p, s2)[0], len(g) - 1)
|
||
order += [g[c]] + list(reversed(g[:c])) + g[c + 1:]
|
||
return order + [i for i in range(len(entries)) if i not in order]
|
||
|
||
|
||
# ------------------------------------------------------ playlists (§10.4, §11)
|
||
|
||
U64 = 2 ** 64 - 1
|
||
U32 = 2 ** 32 - 1
|
||
DIGITS3 = re.compile(r"[0-9]+(\.[0-9]+)?", re.ASCII)
|
||
|
||
|
||
def decimal_ms(s):
|
||
"""A decimal number of seconds, exactly, in milliseconds rounded half up."""
|
||
if not DIGITS3.fullmatch(s):
|
||
return None
|
||
whole, _, frac = s.partition(".")
|
||
f = (frac + "0000")[:4]
|
||
ms = int(whole) * 1000 + int(f[:3]) + (1 if int(f[3]) >= (6 if MUTATE == "round-half-down" else 5) else 0)
|
||
return ms if ms <= U64 else None
|
||
|
||
|
||
def decimal_int(s):
|
||
if not re.fullmatch(r"[0-9]+", s, re.ASCII):
|
||
return None
|
||
v = int(s)
|
||
return v if v <= U64 else None
|
||
|
||
|
||
def attributes(s):
|
||
"""An HLS attribute list (RFC 8216 §4.2) as [(name, quoted, value)]."""
|
||
bad = Refusal("playlist_malformed")
|
||
out, rest = [], s
|
||
while True:
|
||
if "=" not in rest:
|
||
raise bad
|
||
name, after = rest.split("=", 1)
|
||
if not re.fullmatch(r"[A-Z0-9-]+", name, re.ASCII) or any(n == name for n, _, _ in out):
|
||
raise bad
|
||
if after.startswith('"'):
|
||
close = after.find('"', 1)
|
||
if close < 0:
|
||
raise bad
|
||
v = after[1:close]
|
||
if "\r" in v or "\n" in v:
|
||
raise bad
|
||
out.append((name, True, v))
|
||
after = after[close + 1:]
|
||
else:
|
||
end = after.find(",")
|
||
end = len(after) if end < 0 else end
|
||
v = after[:end]
|
||
if not v or '"' in v or any(ch in " \t\n\r\x0c" for ch in v):
|
||
raise bad
|
||
out.append((name, False, v))
|
||
after = after[end:]
|
||
if not after:
|
||
return out
|
||
if not after.startswith(","):
|
||
raise bad
|
||
rest = after[1:]
|
||
|
||
|
||
def attr(a, name):
|
||
return next(((q, v) for n, q, v in a if n == name), None)
|
||
|
||
|
||
def attr_ms(a, name):
|
||
x = attr(a, name)
|
||
if x is None:
|
||
return None
|
||
if x[0] or decimal_ms(x[1]) is None:
|
||
raise Refusal("playlist_malformed")
|
||
return decimal_ms(x[1])
|
||
|
||
|
||
def attr_quoted(a, name):
|
||
x = attr(a, name)
|
||
if x is None:
|
||
return None
|
||
if not x[0]:
|
||
raise Refusal("playlist_malformed")
|
||
return x[1]
|
||
|
||
|
||
def attr_yes(a, name):
|
||
x = attr(a, name)
|
||
if x is None:
|
||
return False
|
||
if x[0] or x[1] not in ("YES", "NO"):
|
||
raise Refusal("playlist_malformed")
|
||
return x[1] == "YES"
|
||
|
||
|
||
def new_extra():
|
||
return {"discontinuity": False, "gap": False, "pdt": None, "parts": [], "layers": [], "hint": None}
|
||
|
||
|
||
def push_layer(layers, l):
|
||
if layers and layers[-1]["n"] >= l["n"]:
|
||
raise Refusal("playlist_malformed")
|
||
layers.append(l)
|
||
|
||
|
||
def parse_media_playlist(text):
|
||
"""A media playlist, classical or WAI, read as §11 needs it."""
|
||
bad = Refusal("playlist_malformed")
|
||
lines = [l[:-1] if l.endswith("\r") else l for l in text.split("\n")]
|
||
lines = [l for l in lines if l]
|
||
if not lines or lines[0] != "#EXTM3U":
|
||
raise bad
|
||
p = {"version": None, "target": None, "seq": 0, "dseq": 0, "part_target": None, "hold_back": None, "block": False,
|
||
"segments": [], "open": None, "hint": None, "end": False, "ptype": None, "indep": False, "map": None}
|
||
pending, extinf, anchor, started = new_extra(), None, None, False
|
||
for line in lines[1:]:
|
||
if line.startswith("#WAI-LAYER:"):
|
||
l = parse_layer_tag(line)
|
||
if anchor == "extinf":
|
||
push_layer(pending["layers"], l)
|
||
elif anchor == "part":
|
||
push_layer(pending["parts"][-1]["layers"], l)
|
||
else:
|
||
raise bad
|
||
continue
|
||
anchor = None
|
||
if not line.startswith("#"):
|
||
if extinf is None:
|
||
raise bad
|
||
p["segments"].append({"ms": extinf, "uri": line, "extra": pending})
|
||
pending, extinf, started = new_extra(), None, True
|
||
continue
|
||
if not line.startswith("#EXT"):
|
||
continue
|
||
tag, _, value = line.partition(":")
|
||
if tag == "#EXTM3U":
|
||
raise bad
|
||
elif tag == "#EXT-X-VERSION":
|
||
v = decimal_int(value)
|
||
if v is None or v > 255:
|
||
raise bad
|
||
p["version"] = v
|
||
elif tag == "#EXT-X-TARGETDURATION":
|
||
v = decimal_int(value)
|
||
if v is None:
|
||
raise bad
|
||
p["target"] = v
|
||
elif tag in ("#EXT-X-MEDIA-SEQUENCE", "#EXT-X-DISCONTINUITY-SEQUENCE"):
|
||
if started or extinf is not None or pending["parts"]:
|
||
raise bad
|
||
v = decimal_int(value)
|
||
if v is None:
|
||
raise bad
|
||
p["seq" if tag == "#EXT-X-MEDIA-SEQUENCE" else "dseq"] = v
|
||
elif tag == "#EXT-X-PART-INF":
|
||
a = attributes(value)
|
||
pt = attr_ms(a, "PART-TARGET")
|
||
if pt is None:
|
||
raise bad
|
||
p["part_target"] = pt
|
||
elif tag == "#EXT-X-SERVER-CONTROL":
|
||
a = attributes(value)
|
||
p["block"] = attr_yes(a, "CAN-BLOCK-RELOAD")
|
||
p["hold_back"] = attr_ms(a, "PART-HOLD-BACK")
|
||
elif tag == "#EXT-X-DISCONTINUITY":
|
||
pending["discontinuity"] = True
|
||
elif tag == "#EXT-X-GAP":
|
||
pending["gap"] = True
|
||
elif tag == "#EXT-X-INDEPENDENT-SEGMENTS":
|
||
p["indep"] = True
|
||
elif tag == "#EXT-X-PLAYLIST-TYPE":
|
||
if value not in ("VOD", "EVENT"):
|
||
raise bad
|
||
p["ptype"] = value
|
||
elif tag == "#EXT-X-MAP":
|
||
uri = attr_quoted(attributes(value), "URI")
|
||
if uri is None:
|
||
raise bad
|
||
if not started and extinf is None and pending == new_extra() and p["map"] is None:
|
||
p["map"] = uri
|
||
elif tag == "#EXT-X-PROGRAM-DATE-TIME":
|
||
pending["pdt"] = value
|
||
elif tag == "#EXT-X-PART":
|
||
a = attributes(value)
|
||
d, uri = attr_ms(a, "DURATION"), attr_quoted(a, "URI")
|
||
if d is None or uri is None:
|
||
raise bad
|
||
pending["parts"].append({"ms": d, "independent": attr_yes(a, "INDEPENDENT"), "uri": uri, "layers": []})
|
||
anchor = "part"
|
||
elif tag == "#EXTINF":
|
||
if extinf is not None:
|
||
raise bad
|
||
extinf = decimal_ms(value.split(",", 1)[0])
|
||
if extinf is None:
|
||
raise bad
|
||
anchor = "extinf"
|
||
elif tag == "#EXT-X-PRELOAD-HINT":
|
||
a = attributes(value)
|
||
t = attr(a, "TYPE")
|
||
if t is None or t[0]:
|
||
raise bad
|
||
if t[1] == "PART":
|
||
if p["hint"] is not None:
|
||
raise bad
|
||
h = attr_quoted(a, "URI")
|
||
if h is None:
|
||
raise bad
|
||
p["hint"] = h
|
||
elif tag == "#EXT-X-ENDLIST":
|
||
p["end"] = True
|
||
elif tag in ("#EXT-X-SKIP", "#EXT-X-STREAM-INF", "#EXT-X-I-FRAME-STREAM-INF", "#EXT-X-MEDIA"):
|
||
raise bad
|
||
if extinf is not None:
|
||
raise bad
|
||
if pending != new_extra():
|
||
p["open"] = pending
|
||
return p
|
||
|
||
|
||
def secs(ms):
|
||
return "%d.%03d" % (ms // 1000, ms % 1000)
|
||
|
||
|
||
def render_hls(segs, r, extras):
|
||
"""The rendering of §8.6: header, then each segment (discontinuity, date-time, parts
|
||
with their layer lines, EXTINF, its layer lines, URI), the open segment, the hint,
|
||
the end."""
|
||
target = r["target"] if r["target"] is not None else max([-(-s["ms"] // 1000) for s in segs], default=0)
|
||
out = ["#EXTM3U", "#EXT-X-VERSION:%d" % r["version"], "#EXT-X-TARGETDURATION:%d" % target]
|
||
if r["part_target"] is not None:
|
||
out.append("#EXT-X-PART-INF:PART-TARGET=" + secs(r["part_target"]))
|
||
ctl = (["CAN-BLOCK-RELOAD=YES"] if r["block"] else []) + (["PART-HOLD-BACK=" + secs(r["hold_back"])] if r["hold_back"] is not None else [])
|
||
if ctl:
|
||
out.append("#EXT-X-SERVER-CONTROL:" + ",".join(ctl))
|
||
out.append("#EXT-X-MEDIA-SEQUENCE:%d" % r["seq"])
|
||
if r["dseq"]:
|
||
out.append("#EXT-X-DISCONTINUITY-SEQUENCE:%d" % r["dseq"])
|
||
if r.get("ptype"):
|
||
out.append("#EXT-X-PLAYLIST-TYPE:" + r["ptype"])
|
||
if r.get("indep"):
|
||
out.append("#EXT-X-INDEPENDENT-SEGMENTS")
|
||
if r.get("map") is not None:
|
||
out.append('#EXT-X-MAP:URI="%s"' % r["map"])
|
||
|
||
def parts(ps):
|
||
for q in ps:
|
||
out.append("#EXT-X-PART:DURATION=%s%s,URI=\"%s\"" % (secs(q["ms"]), ",INDEPENDENT=YES" if q["independent"] else "", q["uri"]))
|
||
out.extend("#WAI-LAYER:N=%d,URI=\"%s\"" % (l["n"], l["uri"]) for l in q["layers"])
|
||
|
||
for i, s in enumerate(segs):
|
||
e = extras[i] if i < len(extras) else new_extra()
|
||
if s.get("ad") or e["discontinuity"]:
|
||
out.append("#EXT-X-DISCONTINUITY")
|
||
if e["pdt"] is not None:
|
||
out.append("#EXT-X-PROGRAM-DATE-TIME:" + e["pdt"])
|
||
if MUTATE == "parts-after-extinf":
|
||
out.append("#EXTINF:%s," % secs(s["ms"]))
|
||
parts(e["parts"])
|
||
else:
|
||
parts(e["parts"])
|
||
if e.get("gap"):
|
||
out.append("#EXT-X-GAP")
|
||
out.append("#EXTINF:%s," % secs(s["ms"]))
|
||
out.extend("#WAI-LAYER:N=%d,URI=\"%s\"" % (l["n"], l["uri"]) for l in e["layers"])
|
||
out.append(s["uri"])
|
||
if len(extras) > len(segs):
|
||
o = extras[len(segs)]
|
||
if o["discontinuity"]:
|
||
out.append("#EXT-X-DISCONTINUITY")
|
||
if o["pdt"] is not None:
|
||
out.append("#EXT-X-PROGRAM-DATE-TIME:" + o["pdt"])
|
||
parts(o["parts"])
|
||
if extras and extras[-1]["hint"] is not None:
|
||
out.append("#EXT-X-PRELOAD-HINT:TYPE=PART,URI=\"%s\"" % extras[-1]["hint"])
|
||
if r["end"]:
|
||
out.append("#EXT-X-ENDLIST")
|
||
return "\n".join(out) + "\n"
|
||
|
||
|
||
def uri_ok(u, quoted):
|
||
return bool(u) and not u.startswith("#") and "\r" not in u and "\n" not in u and not (quoted and '"' in u)
|
||
|
||
|
||
def check_render(segs, r, extras):
|
||
bad = Refusal("playlist_malformed")
|
||
n = len(segs)
|
||
if len(extras) not in (0, n, n + 1) or r["version"] == 0:
|
||
raise bad
|
||
if r["part_target"] is not None and (r["hold_back"] is None or r["part_target"] == 0 or r["hold_back"] < 2 * r["part_target"]):
|
||
raise bad
|
||
if any(not uri_ok(s["uri"], False) for s in segs):
|
||
raise bad
|
||
if r.get("map") is not None and not uri_ok(r["map"], True):
|
||
raise bad
|
||
|
||
def layers_ok(ls):
|
||
last = 0
|
||
for l in ls:
|
||
if l["n"] == 0 or l["n"] <= last or not uri_ok(l["uri"], True):
|
||
raise bad
|
||
last = l["n"]
|
||
|
||
for i, e in enumerate(extras):
|
||
if e["pdt"] is not None and (not e["pdt"] or not re.fullmatch(r"[A-Za-z0-9:.+-]+", e["pdt"], re.ASCII)):
|
||
raise bad
|
||
if e["parts"] and r["part_target"] is None:
|
||
raise bad
|
||
for q in e["parts"]:
|
||
if not uri_ok(q["uri"], True):
|
||
raise bad
|
||
layers_ok(q["layers"])
|
||
layers_ok(e["layers"])
|
||
if i < n and e.get("gap") and (e["parts"] or e["layers"]) and MUTATE != "gap-carries":
|
||
raise bad
|
||
if e["hint"] is not None and (i + 1 != len(extras) or r["end"] or not uri_ok(e["hint"], True)):
|
||
raise bad
|
||
if i == n:
|
||
if r["end"] or e["layers"] or e.get("gap") or (not e["parts"] and (e["discontinuity"] or e["pdt"] is not None)) or (not e["parts"] and e["hint"] is None):
|
||
raise bad
|
||
|
||
|
||
def render_inputs(p):
|
||
segs = [{"ms": s["ms"], "uri": s["uri"]} for s in p["segments"]]
|
||
r = {"version": p["version"] if p["version"] is not None else 1, "seq": p["seq"], "dseq": p["dseq"], "target": p["target"],
|
||
"part_target": p["part_target"], "hold_back": p["hold_back"], "block": p["block"], "end": p["end"],
|
||
"ptype": p["ptype"], "indep": p["indep"], "map": p["map"]}
|
||
extras = [json.loads(json.dumps(s["extra"])) for s in p["segments"]]
|
||
if p["open"] is not None:
|
||
extras.append(json.loads(json.dumps(p["open"])))
|
||
if p["hint"] is not None:
|
||
if not extras:
|
||
extras.append(new_extra())
|
||
extras[-1]["hint"] = p["hint"]
|
||
if all(e == new_extra() for e in extras):
|
||
extras = []
|
||
return segs, r, extras
|
||
|
||
|
||
def is_rendered_form(p, text):
|
||
segs, r, extras = render_inputs(p)
|
||
try:
|
||
check_render(segs, r, extras)
|
||
except Refusal:
|
||
return False
|
||
return render_hls(segs, r, extras) == text
|
||
|
||
|
||
def misaligned():
|
||
return Refusal("adjacency_misaligned")
|
||
|
||
|
||
def timelines_adjacent(w, c):
|
||
if w["seq"] != c["seq"] or (w["dseq"] != c["dseq"] and MUTATE != "dseq-ignored") or len(w["segments"]) != len(c["segments"]):
|
||
raise misaligned()
|
||
both = w["part_target"] is not None and c["part_target"] is not None
|
||
if both and w["part_target"] != c["part_target"]:
|
||
raise misaligned()
|
||
for a, b in zip(w["segments"], c["segments"]):
|
||
if a["ms"] != b["ms"] or a["extra"]["discontinuity"] != b["extra"]["discontinuity"]:
|
||
raise misaligned()
|
||
if a["extra"]["pdt"] != b["extra"]["pdt"] and MUTATE != "pdt-ignored":
|
||
raise misaligned()
|
||
pa, pb = [q["ms"] for q in a["extra"]["parts"]], [q["ms"] for q in b["extra"]["parts"]]
|
||
if both and pa and pb and pa != pb:
|
||
raise misaligned()
|
||
if w["open"] is not None and c["open"] is not None:
|
||
a, b = w["open"], c["open"]
|
||
if a["discontinuity"] != b["discontinuity"] or a["pdt"] != b["pdt"]:
|
||
raise misaligned()
|
||
if both and [q["ms"] for q in a["parts"]] != [q["ms"] for q in b["parts"]]:
|
||
raise misaligned()
|
||
|
||
|
||
def check_adjacent(wai, classical):
|
||
w = parse_media_playlist(wai)
|
||
c = parse_media_playlist(classical)
|
||
timelines_adjacent(w, c)
|
||
|
||
|
||
def wai_media_playlist(c, segs, open_, prefix, end):
|
||
"""The WAI media playlist on classical timeline `c` (§11.1-§11.4)."""
|
||
if len(segs) != len(c["segments"]):
|
||
raise misaligned()
|
||
has_parts = any(s["parts"] for s in segs) or (open_ is not None and open_["parts"])
|
||
if has_parts and c["part_target"] is None:
|
||
raise misaligned()
|
||
lay = lambda ls: [{"n": l["n"], "uri": "%s%s.wai#n=%d" % (prefix, l["cid"].hex(), l["len"])} for l in ls]
|
||
prt = lambda ps: [{"ms": q["ms"], "independent": q["independent"], "uri": "%s#cid=%s&n=%d" % (q["alias"], q["cid"].hex(), q["n"]), "layers": lay(q["layers"])} for q in ps]
|
||
out_segs, extras = [], []
|
||
for i, (w, k) in enumerate(zip(segs, c["segments"])):
|
||
if w["ms"] != k["ms"]:
|
||
raise misaligned()
|
||
if w["parts"] and k["extra"]["parts"] and [q["ms"] for q in w["parts"]] != [q["ms"] for q in k["extra"]["parts"]]:
|
||
raise misaligned()
|
||
h = "&h=%d" % w["h"] if w["h"] is not None else ""
|
||
out_segs.append({"ms": w["ms"], "uri": "%s%s.wai#n=%d%s" % (prefix, w["cid"].hex(), w["n"], h)})
|
||
e = new_extra()
|
||
e.update(discontinuity=k["extra"]["discontinuity"], gap=w.get("gap", False), pdt=k["extra"]["pdt"], parts=prt(w["parts"]), layers=lay(w["layers"]))
|
||
extras.append(e)
|
||
hint = None
|
||
if open_ is not None:
|
||
if not open_["parts"]:
|
||
hint = open_["hint"]
|
||
else:
|
||
ko = c["open"]
|
||
if ko is None:
|
||
raise misaligned()
|
||
if ko["parts"] and [q["ms"] for q in open_["parts"]] != [q["ms"] for q in ko["parts"]]:
|
||
raise misaligned()
|
||
e = new_extra()
|
||
e.update(discontinuity=ko["discontinuity"], pdt=ko["pdt"], parts=prt(open_["parts"]), hint=open_["hint"])
|
||
extras.append(e)
|
||
if all(e == new_extra() for e in extras):
|
||
extras = []
|
||
if hint is not None:
|
||
if not extras:
|
||
extras = [new_extra() for _ in range(max(len(segs), 1))]
|
||
extras[-1]["hint"] = hint
|
||
pt = c["part_target"]
|
||
r = {"version": c["version"] if c["version"] is not None else 6, "seq": c["seq"], "dseq": c["dseq"], "target": c["target"],
|
||
"part_target": pt, "hold_back": c["hold_back"] if c["hold_back"] is not None else (pt * 3 if pt is not None else None),
|
||
"block": c["block"], "end": end, "ptype": c["ptype"], "indep": c["indep"], "map": None}
|
||
check_render(out_segs, r, extras)
|
||
text = render_hls(out_segs, r, extras)
|
||
timelines_adjacent(parse_media_playlist(text), c)
|
||
return text
|
||
|
||
|
||
def bandwidths(objs):
|
||
if not objs or any(d == 0 for _, d in objs):
|
||
return None
|
||
rate = lambda n, d: min(-(-(8000 * n) // d), U64)
|
||
peak = max(rate(n, d) for n, d in objs)
|
||
sn, sd = sum(n for n, _ in objs), sum(d for _, d in objs)
|
||
return peak, (rate(sn, sd) if MUTATE != "average-floor" else 8000 * sn // sd)
|
||
|
||
|
||
SESSION_DATA = '#EXT-X-SESSION-DATA:DATA-ID="science.transaction.wai.multivariant",VALUE="'
|
||
|
||
|
||
def session_data_position(original):
|
||
"""§11.1: the offset after the first line and the EXT-X-VERSION and
|
||
EXT-X-INDEPENDENT-SEGMENTS lines that directly follow it; None when there is none."""
|
||
at = 0
|
||
lines = original.split("\n")
|
||
lines = [l + "\n" for l in lines[:-1]] + ([lines[-1]] if lines[-1] else [])
|
||
for i, l in enumerate(lines):
|
||
bare = l[:-1] if l.endswith("\n") else l
|
||
bare = bare[:-1] if l.endswith("\n") and bare.endswith("\r") else bare
|
||
if i > 0 and not (bare.startswith("#EXT-X-VERSION:") or bare == "#EXT-X-INDEPENDENT-SEGMENTS") and MUTATE != "session-data-anywhere":
|
||
break
|
||
if not l.endswith("\n"):
|
||
return None
|
||
at += len(l.encode("utf-8"))
|
||
if MUTATE == "session-data-anywhere":
|
||
break
|
||
return at if at > 0 else None
|
||
|
||
|
||
def classical_unchanged(original, published):
|
||
if original == published:
|
||
return
|
||
o, p = original.encode("utf-8"), published.encode("utf-8")
|
||
at = session_data_position(original)
|
||
if at is None or not (p.startswith(o[:at]) and p.endswith(o[at:]) and len(p) > len(o)):
|
||
raise Refusal("classical_changed")
|
||
inserted = p[at:len(p) - (len(o) - at)].decode("utf-8")
|
||
if not inserted.endswith("\n"):
|
||
raise Refusal("classical_changed")
|
||
bare = inserted[:-1]
|
||
bare = bare[:-1] if bare.endswith("\r") else bare
|
||
if "\n" in bare or not (bare.startswith(SESSION_DATA) and bare.endswith('"')):
|
||
raise Refusal("classical_changed")
|
||
uri = bare[len(SESSION_DATA):-1]
|
||
if not uri or any(ch in uri for ch in '"\r\n'):
|
||
raise Refusal("classical_changed")
|
||
|
||
|
||
def listing_lines(p):
|
||
"""Every part, layer and segment line of a media playlist, in the order a client
|
||
reads them for a cid (§5.2, §11.4)."""
|
||
lines = []
|
||
for s in p["segments"]:
|
||
for q in s["extra"]["parts"]:
|
||
lines.append(q["uri"])
|
||
lines += [l["uri"] for l in q["layers"]]
|
||
lines += [l["uri"] for l in s["extra"]["layers"]]
|
||
lines.append(s["uri"])
|
||
if p["open"] is not None:
|
||
for q in p["open"]["parts"]:
|
||
lines.append(q["uri"])
|
||
lines += [l["uri"] for l in q["layers"]]
|
||
return lines
|
||
|
||
|
||
def uri_cid(line):
|
||
"""The cid a listed URI names: its cid fragment, or the <cid>.wai its path ends in."""
|
||
base, _, frag = line.partition("#")
|
||
for kv in frag.split("&") if frag else []:
|
||
k, eq, v = kv.partition("=")
|
||
if eq and k == "cid" and re.fullmatch(r"[0-9a-f]{64}", v):
|
||
return bytes.fromhex(v)
|
||
name = base.rsplit("/", 1)[-1]
|
||
if name.endswith(".wai") and re.fullmatch(r"[0-9a-f]{64}", name[:-4]):
|
||
return bytes.fromhex(name[:-4])
|
||
return None
|
||
|
||
|
||
def listed_cid(p, uri):
|
||
line = next((l for l in listing_lines(p) if l.split("#", 1)[0] == uri), None)
|
||
return None if line is None else uri_cid(line)
|
||
|
||
|
||
RETENTION = 256
|
||
URI_PARTS = re.compile(r"^(([^:/?#]+):)?(//([^/?#]*))?([^?#]*)(\?([^#]*))?(#(.*))?$", re.S)
|
||
|
||
|
||
def remove_dot_segments(path):
|
||
"""RFC 3986 §5.2.4."""
|
||
inp, out = path, ""
|
||
while inp:
|
||
if inp.startswith("../"):
|
||
inp = inp[3:]
|
||
elif inp.startswith("./"):
|
||
inp = inp[2:]
|
||
elif inp.startswith("/./"):
|
||
inp = inp[2:]
|
||
elif inp == "/.":
|
||
inp = "/"
|
||
elif inp.startswith("/../") or inp == "/..":
|
||
inp = "/" + inp[4:] if inp.startswith("/../") else "/"
|
||
out = out[:out.rfind("/")] if "/" in out else ""
|
||
elif inp in (".", ".."):
|
||
inp = ""
|
||
else:
|
||
j = inp.find("/", 1 if inp.startswith("/") else 0)
|
||
j = len(inp) if j < 0 else j
|
||
out, inp = out + inp[:j], inp[j:]
|
||
return out
|
||
|
||
|
||
def pct_normal(s):
|
||
"""RFC 3986 §6.2.2.2: upper-case hexadecimal digits, unreserved characters decoded."""
|
||
def one(m):
|
||
c = chr(int(m.group(1), 16))
|
||
return c if re.fullmatch(r"[A-Za-z0-9._~-]", c) else "%" + m.group(1).upper()
|
||
return re.sub(r"%([0-9A-Fa-f]{2})", one, s)
|
||
|
||
|
||
def rfc3986_resolve(base, ref):
|
||
"""ref resolved against base (RFC 3986 §5.2.2, §5.2.3) and normalized (§6.2.2), its
|
||
fragment dropped."""
|
||
def parts(u):
|
||
m = URI_PARTS.match(u)
|
||
return m.group(2), m.group(4) if m.group(3) is not None else None, m.group(5), m.group(7) if m.group(6) is not None else None
|
||
bs, ba, bp, bq = parts(base)
|
||
rs, ra, rp, rq = parts(ref)
|
||
if MUTATE == "uri-unresolved":
|
||
return ref.split("#", 1)[0]
|
||
if rs is not None:
|
||
ts, ta, tp, tq = rs, ra, remove_dot_segments(rp), rq
|
||
elif ra is not None:
|
||
ts, ta, tp, tq = bs, ra, remove_dot_segments(rp), rq
|
||
elif rp == "":
|
||
ts, ta, tp, tq = bs, ba, bp, rq if rq is not None else bq
|
||
elif rp.startswith("/"):
|
||
ts, ta, tp, tq = bs, ba, remove_dot_segments(rp), rq
|
||
else:
|
||
if ba is not None and bp == "":
|
||
merged = "/" + rp
|
||
else:
|
||
merged = bp[:bp.rfind("/") + 1] + rp
|
||
ts, ta, tp, tq = bs, ba, remove_dot_segments(merged), rq
|
||
out = ""
|
||
if ts is not None:
|
||
out += ts.lower() + ":"
|
||
if ta is not None:
|
||
a = pct_normal(ta)
|
||
at = a.rfind("@")
|
||
out += "//" + (a[:at + 1] + a[at + 1:].lower() if at >= 0 else a.lower())
|
||
out += remove_dot_segments(pct_normal(tp))
|
||
if tq is not None:
|
||
out += "?" + pct_normal(tq)
|
||
return out
|
||
|
||
|
||
def new_bindings(url):
|
||
return {"base": url, "uris": {}, "seqs": {}, "rev": 0}
|
||
|
||
|
||
def bind(b, p):
|
||
"""§5.2, §11.5: the bindings b with playlist p's added as one more revision: each URL
|
||
a line names (resolved against the playlist's URL) and each media sequence number to
|
||
its cid. Refusal when p lists another cid for one bound, or a gap on a bound sequence
|
||
number; then what none of the last RETENTION revisions listed is forgotten."""
|
||
rev = b["rev"] + 1
|
||
uris, seqs = dict(b["uris"]), dict(b["seqs"])
|
||
for line in listing_lines(p):
|
||
cid = uri_cid(line)
|
||
if cid is None:
|
||
continue
|
||
key = rfc3986_resolve(b["base"], line)
|
||
if key in uris and uris[key][0] != cid and MUTATE != "rebind-uri":
|
||
raise Refusal("playlist_listing_rebound")
|
||
uris[key] = (cid, rev)
|
||
for i, sg in enumerate(p["segments"]):
|
||
n = p["seq"] + i
|
||
if n > U64:
|
||
raise Refusal("playlist_malformed")
|
||
if sg["extra"]["gap"]:
|
||
if n in seqs and MUTATE != "gap-withdraws":
|
||
raise Refusal("playlist_listing_rebound")
|
||
continue
|
||
cid = uri_cid(sg["uri"])
|
||
if cid is None:
|
||
continue
|
||
if n in seqs and seqs[n][0] != cid and MUTATE != "rebind-sequence":
|
||
raise Refusal("playlist_listing_rebound")
|
||
seqs[n] = (cid, rev)
|
||
keep = (lambda last: True) if MUTATE == "bindings-unbounded" else (lambda last: last + RETENTION > rev)
|
||
return {"base": b["base"], "uris": {k: v for k, v in uris.items() if keep(v[1])}, "seqs": {k: v for k, v in seqs.items() if keep(v[1])}, "rev": rev}
|
||
|
||
|
||
def cid_of(b, uri):
|
||
v = b["uris"].get(rfc3986_resolve(b["base"], uri))
|
||
return None if v is None else v[0]
|
||
|
||
|
||
def xml_char(c):
|
||
o = ord(c)
|
||
return o in (9, 10, 13) or 0x20 <= o <= 0xD7FF or 0xE000 <= o <= 0xFFFD or 0x10000 <= o <= 0x10FFFF
|
||
|
||
|
||
def esc_xml(s):
|
||
"""An attribute value: & < > " as entities, tab, LF and CR as character references;
|
||
Refusal for a character that is not an XML Char."""
|
||
out = []
|
||
for c in s:
|
||
if c in '&<>"\t\n\r':
|
||
out.append({"&": "&", "<": "<", ">": ">", '"': """, "\t": "	", "\n": " ", "\r": " "}[c])
|
||
elif xml_char(c) or MUTATE == "xml-char-unchecked":
|
||
out.append(c)
|
||
else:
|
||
raise Refusal("playlist_malformed")
|
||
return "".join(out)
|
||
|
||
|
||
# The characters of Unicode's White_Space property: an MPD @id holds none.
|
||
WHITE_SPACE = set("\t\n\x0b\x0c\r \x85\xa0\u1680\u2028\u2029\u202f\u205f\u3000") | {chr(c) for c in range(0x2000, 0x200B)}
|
||
|
||
|
||
def id_ok(s):
|
||
return s != "" and not any(c in WHITE_SPACE for c in s)
|
||
|
||
|
||
def xs_seconds(ms):
|
||
return "PT%dS" % (ms // 1000) if ms % 1000 == 0 else "PT%sS" % secs(ms)
|
||
|
||
|
||
def wai_mpd(t, set_id, union, floor, rep_id, bandwidth, media):
|
||
"""wai.mpd (§11.1) on the timeline t = (period id or None, start ms, timescale,
|
||
[(t, d)]); media[i] None leaves segment i out (a gap)."""
|
||
period_id, start_ms, ts, segs = t
|
||
if not segs or ts == 0 or len(media) != len(segs) or union[0] != floor[0]:
|
||
raise Refusal("playlist_malformed")
|
||
if not id_ok(rep_id) or (period_id is not None and not id_ok(period_id)):
|
||
raise Refusal("playlist_malformed")
|
||
offered = [(sg, m) for sg, m in zip(segs, media) if m is not None]
|
||
if not offered:
|
||
raise Refusal("playlist_malformed")
|
||
to_ms = lambda d: min(d * 1000 // ts, U64)
|
||
total = min(sum(to_ms(d) for _, d in segs), U64)
|
||
longest = max(to_ms(d) for _, d in segs)
|
||
period = (' <Period id="%s" start="%s">' % (esc_xml(period_id), xs_seconds(start_ms))) if period_id is not None else (' <Period start="%s">' % xs_seconds(start_ms))
|
||
out = ['<?xml version="1.0" encoding="UTF-8"?>',
|
||
'<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" profiles="urn:mpeg:dash:profile:full:2011" type="static" mediaPresentationDuration="PT%sS" minBufferTime="PT%dS">' % (secs(total), -(-longest // 1000)),
|
||
period,
|
||
' <AdaptationSet id="%d" mimeType="application/wai" codecs="%s" segmentAlignment="true">' % (set_id, esc_xml(render(union))),
|
||
' <EssentialProperty schemeIdUri="tag:transaction.science,2026:wai/http-delivery"/>',
|
||
' <SupplementalProperty schemeIdUri="tag:transaction.science,2026:wai/floor" value="%s"/>' % esc_xml(render(floor)),
|
||
' <Representation id="%s" bandwidth="%d">' % (esc_xml(rep_id), bandwidth),
|
||
' <SegmentList timescale="%d">' % ts, ' <SegmentTimeline>']
|
||
out += [' <S t="%d" d="%d"/>' % sg for sg, _ in offered]
|
||
out += [' </SegmentTimeline>'] + [' <SegmentURL media="%s"/>' % esc_xml(m) for _, m in offered]
|
||
out += [' </SegmentList>', ' </Representation>', ' </AdaptationSet>', ' </Period>', '</MPD>']
|
||
return "\n".join(out) + "\n"
|
||
|
||
|
||
# ---------------------------------------------- reading an MPD (Appendix I)
|
||
|
||
MAX_TIMELINES = 256
|
||
MAX_DEPTH = 64
|
||
NAME = re.compile(r"[A-Za-z_:][A-Za-z0-9_:.\-]*", re.ASCII)
|
||
|
||
|
||
class Xml:
|
||
"""A reader of the XML subset of Appendix I over a normalised text."""
|
||
|
||
def __init__(self, s):
|
||
self.s, self.at = s, 0
|
||
|
||
def bad(self):
|
||
return Refusal("playlist_malformed")
|
||
|
||
def rest(self):
|
||
return self.s[self.at:]
|
||
|
||
def eat(self, lit):
|
||
if self.s.startswith(lit, self.at):
|
||
self.at += len(lit)
|
||
return True
|
||
return False
|
||
|
||
def space(self):
|
||
n = 0
|
||
while self.at + n < len(self.s) and self.s[self.at + n] in " \t\n":
|
||
n += 1
|
||
self.at += n
|
||
return n > 0
|
||
|
||
def name(self):
|
||
m = NAME.match(self.s, self.at)
|
||
if m is None:
|
||
raise self.bad()
|
||
self.at = m.end()
|
||
return m.group(0)
|
||
|
||
def reference(self):
|
||
end = self.s.find(";", self.at)
|
||
if end < 0:
|
||
raise self.bad()
|
||
body = self.s[self.at:end]
|
||
named = {"amp": "&", "lt": "<", "gt": ">", "quot": '"', "apos": "'"}
|
||
if body in named:
|
||
c = named[body]
|
||
else:
|
||
if body.startswith("#x") and re.fullmatch(r"[0-9A-Fa-f]{1,8}", body[2:], re.ASCII):
|
||
n = int(body[2:], 16)
|
||
elif body.startswith("#") and re.fullmatch(r"[0-9]{1,10}", body[1:], re.ASCII) and int(body[1:]) <= 0xFFFFFFFF:
|
||
n = int(body[1:])
|
||
else:
|
||
raise self.bad()
|
||
if 0xD800 <= n <= 0xDFFF or n > 0x10FFFF or not xml_char(chr(n)):
|
||
raise self.bad()
|
||
c = chr(n)
|
||
self.at = end + 1
|
||
return c
|
||
|
||
def attr_value(self):
|
||
if self.at >= len(self.s) or self.s[self.at] not in "\"'":
|
||
raise self.bad()
|
||
q = self.s[self.at]
|
||
self.at += 1
|
||
out = []
|
||
while True:
|
||
if self.at >= len(self.s):
|
||
raise self.bad()
|
||
c = self.s[self.at]
|
||
self.at += 1
|
||
if c == q:
|
||
return "".join(out)
|
||
if c == "<":
|
||
raise self.bad()
|
||
if c == "&":
|
||
out.append(self.reference())
|
||
elif c in "\t\n":
|
||
out.append(" ")
|
||
else:
|
||
out.append(c)
|
||
|
||
def comment(self):
|
||
self.at += 4
|
||
end = self.s.find("--", self.at)
|
||
if end < 0 or not self.s.startswith("-->", end):
|
||
raise self.bad()
|
||
self.at = end + 3
|
||
|
||
def misc(self):
|
||
while True:
|
||
if self.space():
|
||
continue
|
||
if self.s.startswith("<!--", self.at):
|
||
self.comment()
|
||
continue
|
||
return
|
||
|
||
def start_tag(self):
|
||
name = self.name()
|
||
attrs, names = [], set()
|
||
while True:
|
||
spaced = self.space()
|
||
if self.eat("/>"):
|
||
return {"name": name, "attrs": attrs, "children": []}, True
|
||
if self.eat(">"):
|
||
return {"name": name, "attrs": attrs, "children": []}, False
|
||
if not spaced:
|
||
raise self.bad()
|
||
k = self.name()
|
||
self.space()
|
||
if not self.eat("="):
|
||
raise self.bad()
|
||
self.space()
|
||
v = self.attr_value()
|
||
seen = any(a == k for a, _ in attrs) if MUTATE == "xml-attributes-scanned" else k in names
|
||
if seen and MUTATE != "xml-attribute-twice":
|
||
raise self.bad()
|
||
names.add(k)
|
||
attrs.append((k, v))
|
||
|
||
|
||
def read_xml(text):
|
||
"""The root element of a document in the XML subset of Appendix I:
|
||
{"name", "attrs", "children"}; Refusal("playlist_malformed") otherwise."""
|
||
if text.startswith("\ufeff"):
|
||
text = text[1:]
|
||
text = text.replace("\r\n", "\n").replace("\r", "\n")
|
||
if not all(xml_char(c) for c in text):
|
||
raise Refusal("playlist_malformed")
|
||
r = Xml(text)
|
||
if text.startswith("<?xml") and text[5:6] in (" ", "\t", "\n", "?") and text[5:6] != "":
|
||
end = text.find("?>")
|
||
if end < 0:
|
||
raise r.bad()
|
||
r.at = end + 2
|
||
r.misc()
|
||
if not r.eat("<"):
|
||
raise r.bad()
|
||
root, closed = r.start_tag()
|
||
stack, done = ([], root) if closed else ([root], None)
|
||
while done is None:
|
||
if r.at >= len(text):
|
||
raise r.bad()
|
||
c = text[r.at]
|
||
if c == "<":
|
||
if text.startswith("<!--", r.at):
|
||
r.comment()
|
||
elif r.eat("</"):
|
||
name = r.name()
|
||
r.space()
|
||
if not r.eat(">"):
|
||
raise r.bad()
|
||
e = stack.pop()
|
||
if e["name"] != name:
|
||
raise r.bad()
|
||
if stack:
|
||
stack[-1]["children"].append(e)
|
||
else:
|
||
done = e
|
||
elif text.startswith("<!", r.at) or text.startswith("<?", r.at):
|
||
raise r.bad()
|
||
else:
|
||
r.at += 1
|
||
e, closed = r.start_tag()
|
||
if closed:
|
||
stack[-1]["children"].append(e)
|
||
else:
|
||
if len(stack) >= MAX_DEPTH:
|
||
raise r.bad()
|
||
stack.append(e)
|
||
elif c == "&":
|
||
r.at += 1
|
||
r.reference()
|
||
else:
|
||
if text.startswith("]]>", r.at):
|
||
raise r.bad()
|
||
r.at += 1
|
||
r.misc()
|
||
if r.at != len(text):
|
||
raise r.bad()
|
||
return done
|
||
|
||
|
||
def local(e):
|
||
return e["name"].rsplit(":", 1)[-1]
|
||
|
||
|
||
def xattr(e, k):
|
||
return next((v for a, v in e["attrs"] if a == k), None)
|
||
|
||
|
||
def duration_ms(s):
|
||
"""An xs:duration of days, hours, minutes and seconds in milliseconds; None otherwise."""
|
||
if not s.startswith("P"):
|
||
return None
|
||
rest = s[1:]
|
||
date, sep, time = rest.partition("T")
|
||
time = time if sep else None
|
||
integer = lambda v: int(v) if re.fullmatch(r"[0-9]+", v, re.ASCII) else None
|
||
ms, anything = 0, False
|
||
if date:
|
||
if not date.endswith("D") or integer(date[:-1]) is None:
|
||
return None
|
||
ms, anything = integer(date[:-1]) * 86400000, True
|
||
if time is not None:
|
||
if time == "":
|
||
return None
|
||
for unit, scale in (("H", 3600000), ("M", 60000)):
|
||
i = time.find(unit)
|
||
if i >= 0:
|
||
v = integer(time[:i])
|
||
if v is None:
|
||
return None
|
||
ms += v * scale
|
||
time, anything = time[i + 1:], True
|
||
if time:
|
||
if not time.endswith("S"):
|
||
return None
|
||
v = decimal_ms(time[:-1])
|
||
if v is None:
|
||
return None
|
||
ms, anything = ms + v, True
|
||
return ms if anything and ms <= U64 else None
|
||
|
||
|
||
def unsigned(v):
|
||
if v is None or not re.fullmatch(r"[0-9]+", v, re.ASCII) or int(v) > U64:
|
||
raise Refusal("playlist_malformed")
|
||
return int(v)
|
||
|
||
|
||
def timeline_of(e, inherited):
|
||
"""A timeline as written: {"timescale", "runs": [(t, d, count)]}, one run per S, never
|
||
expanded; None without a SegmentTimeline."""
|
||
st = next((c for c in e["children"] if local(c) == "SegmentTimeline"), None)
|
||
if st is None:
|
||
return None
|
||
ts = next((xattr(x, "timescale") for x in [e] + inherited if xattr(x, "timescale") is not None), None)
|
||
timescale = unsigned(ts) if ts is not None else 1
|
||
if timescale == 0:
|
||
raise Refusal("playlist_malformed")
|
||
runs, nxt = [], 0
|
||
for sel in (c for c in st["children"] if local(c) == "S"):
|
||
d = unsigned(xattr(sel, "d"))
|
||
if d == 0:
|
||
raise Refusal("playlist_malformed")
|
||
t = unsigned(xattr(sel, "t")) if xattr(sel, "t") is not None else nxt
|
||
r = unsigned(xattr(sel, "r")) if xattr(sel, "r") is not None else 0
|
||
end = t + d * (r + 1)
|
||
if end > U64:
|
||
raise Refusal("playlist_malformed")
|
||
if MUTATE == "dash-r-expanded" and r < 1 << 20:
|
||
runs += [(t + i * d, d, 1) for i in range(r + 1)]
|
||
else:
|
||
runs.append((t, d, r + 1))
|
||
nxt = end
|
||
return {"timescale": timescale, "runs": runs}
|
||
|
||
|
||
SEGMENT_INFO = ("SegmentTemplate", "SegmentList")
|
||
|
||
|
||
def read_mpd(text):
|
||
"""An MPD's periods: [{"id", "start_ms", "sets": [{"id", "timelines"}]}]."""
|
||
root = read_xml(text)
|
||
if local(root) != "MPD":
|
||
raise Refusal("playlist_malformed")
|
||
periods, held = [], 0
|
||
for i, p in enumerate(c for c in root["children"] if local(c) == "Period"):
|
||
st = xattr(p, "start")
|
||
if st is not None:
|
||
start = duration_ms(st)
|
||
if start is None:
|
||
raise Refusal("playlist_malformed")
|
||
elif i == 0:
|
||
start = 0
|
||
else:
|
||
raise Refusal("playlist_malformed")
|
||
p_info = [c for c in p["children"] if local(c) in SEGMENT_INFO]
|
||
sets = []
|
||
for a in (c for c in p["children"] if local(c) == "AdaptationSet"):
|
||
tls = []
|
||
a_info = [c for c in a["children"] if local(c) in SEGMENT_INFO]
|
||
for e in a_info:
|
||
tl = timeline_of(e, [x for x in p_info if local(x) == local(e)])
|
||
if tl is not None:
|
||
tls.append(tl)
|
||
for rep in (c for c in a["children"] if local(c) == "Representation"):
|
||
for e in (c for c in rep["children"] if local(c) in SEGMENT_INFO):
|
||
tl = timeline_of(e, [x for x in a_info + p_info if local(x) == local(e)])
|
||
if tl is not None:
|
||
tls.append(tl)
|
||
held += len(tls)
|
||
if held > MAX_TIMELINES and MUTATE != "dash-timelines-unbounded":
|
||
raise Refusal("playlist_malformed")
|
||
sets.append({"id": xattr(a, "id"), "timelines": tls})
|
||
periods.append({"id": xattr(p, "id"), "start_ms": start, "sets": sets})
|
||
if not periods:
|
||
raise Refusal("playlist_malformed")
|
||
return periods
|
||
|
||
|
||
def within(wai, classical):
|
||
"""Whether the WAI runs list only segments of the classical runs, each with its t and
|
||
d, in order: for each WAI segment, the first classical segment at or after the one
|
||
matched before with its t has its d too. Run by run, never expanded."""
|
||
ci, co = 0, 0
|
||
for wt, wd, wn in wai:
|
||
t, left = wt, wn
|
||
while left > 0:
|
||
while True:
|
||
if ci == len(classical):
|
||
return False
|
||
ct, cd, cn = classical[ci]
|
||
if t >= ct + co * cd and (t - ct) % cd == 0 and (t - ct) // cd < cn:
|
||
co = (t - ct) // cd
|
||
break
|
||
ci, co = ci + 1, 0
|
||
ct, cd, cn = classical[ci]
|
||
if cd != wd:
|
||
return False
|
||
take = min(left, cn - co)
|
||
left, co, t = left - take, co + take, t + take * wd
|
||
if co == cn:
|
||
ci, co = ci + 1, 0
|
||
return True
|
||
|
||
|
||
def check_dash_adjacent(wai, classical):
|
||
"""§11.2 in DASH: Refusal("playlist_malformed") when either does not read,
|
||
Refusal("adjacency_misaligned") when not adjacent."""
|
||
w, k = read_mpd(wai), read_mpd(classical)
|
||
mis = Refusal("adjacency_misaligned")
|
||
for wp in w:
|
||
kp = next((p for p in k if p["id"] == wp["id"]), None)
|
||
if kp is None or (kp["start_ms"] != wp["start_ms"] and MUTATE != "dash-start-ignored"):
|
||
raise mis
|
||
if not wp["sets"]:
|
||
raise mis
|
||
for st in wp["sets"]:
|
||
if not st["timelines"]:
|
||
raise mis
|
||
for tl in st["timelines"]:
|
||
if not tl["runs"] or not any((c["timescale"] == tl["timescale"] or MUTATE == "dash-timescale-ignored") and within(tl["runs"], c["runs"])
|
||
for cs in kp["sets"] for c in cs["timelines"]):
|
||
raise mis
|
||
|
||
|
||
def mpd_timeline(text, period_id, set_id, max_segments):
|
||
"""The classical MPD's timeline (§11.1): (period id, start ms, timescale, segments),
|
||
expanded only when it lists at most max_segments segments."""
|
||
periods = read_mpd(text)
|
||
p = next((x for x in periods if x["id"] == period_id), None) if period_id is not None else periods[0]
|
||
if p is None:
|
||
raise Refusal("playlist_malformed")
|
||
st = next((x for x in p["sets"] if x["id"] == set_id), None) if set_id is not None else next((x for x in p["sets"] if x["timelines"]), None)
|
||
if st is None or not st["timelines"]:
|
||
raise Refusal("playlist_malformed")
|
||
tl = st["timelines"][0]
|
||
if sum(n for _, _, n in tl["runs"]) > max_segments:
|
||
raise Refusal("playlist_malformed")
|
||
return (p["id"], p["start_ms"], tl["timescale"], [(t + i * d, d) for t, d, n in tl["runs"] for i in range(n)])
|
||
|
||
|
||
# ------------------------------------------------ playlist receipts (§8.6)
|
||
|
||
def jcs(v):
|
||
"""RFC 8785 over the subset the render digest uses."""
|
||
if v is None:
|
||
return "null"
|
||
if v is True or v is False:
|
||
return "true" if v else "false"
|
||
if isinstance(v, int):
|
||
if not 0 <= v <= 2 ** 53 - 1:
|
||
raise ValueError("outside the canonical subset")
|
||
return str(v)
|
||
if isinstance(v, str):
|
||
return json.dumps(v, ensure_ascii=False)
|
||
if isinstance(v, list):
|
||
return "[" + ",".join(jcs(x) for x in v) + "]"
|
||
keys = sorted(v, key=lambda k: k.encode("utf-16-be"))
|
||
return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + jcs(v[k]) for k in keys) + "}"
|
||
|
||
|
||
# Ed25519, strictly as WAI verifies every signature (SPEC §7 *Signatures*): the key and R
|
||
# decoded with y < p (and no sign bit on x = 0), neither of small order, S < L, and
|
||
# [S]B = R + [k]A checked by encoding [S]B − [k]A and comparing it with R byte for byte.
|
||
# `_ed25519.py` writes it from RFC 8032's definitions, with the standard library only;
|
||
# --mutate small-order-key and small-order-r leave out its two order checks.
|
||
import _ed25519 as ED # noqa: E402
|
||
|
||
|
||
def ed25519_verify(pk, msg, sig):
|
||
return ED.verifies(pk, msg, sig, skip=(MUTATE,))
|
||
|
||
|
||
RECEIPT_KNOWN = {"kind", "manifest_hash", "session_class_hash", "class", "segments", "merkle_root", "work_compose_ops", "viewers_served",
|
||
"joules_micro", "grant", "parent_receipt_hash", "signer_pubkey", "signer_id", "sig", "render", "extras",
|
||
"energy_provenance", "energy_uncertainty", "energy_counter_filtering", "energy_calibration_ref"}
|
||
LEGACY_CLASSES = {"HwShunt": 1, "ModelBased": 2, "Estimator": 3}
|
||
|
||
|
||
MAX_FILE_NUMBER = 2 ** 53 - 1
|
||
|
||
|
||
class ReceiptMalformed(Exception):
|
||
pass
|
||
|
||
|
||
class NotUnsigned:
|
||
"""A JSON number with a sign, a fraction or an exponent: never an unsigned integer of
|
||
the receipt file, whatever its value (so "-0" and "1.0" are not 0 and 1)."""
|
||
|
||
|
||
def read_receipt(text):
|
||
"""The receipt file (§8.6), read strictly; ReceiptMalformed otherwise. A number is an
|
||
unsigned integer written as digits alone."""
|
||
def pairs(kv):
|
||
# A member name twice in one object (I-JSON, RFC 7493 §2.3), names compared as read.
|
||
if len({k for k, _ in kv}) != len(kv) and MUTATE != "receipt-duplicates":
|
||
raise ReceiptMalformed()
|
||
return dict(kv)
|
||
|
||
try:
|
||
# A number is an unsigned integer of at most 2^53 - 1, written as digits alone.
|
||
o = json.loads(text, object_pairs_hook=pairs, parse_float=lambda t: NotUnsigned(),
|
||
parse_int=lambda t: NotUnsigned() if t.startswith("-") or (int(t) > MAX_FILE_NUMBER and MUTATE != "file-numbers-unbounded") else int(t))
|
||
except (ValueError, RecursionError):
|
||
raise ReceiptMalformed()
|
||
bad = ReceiptMalformed
|
||
|
||
def texts(v):
|
||
if isinstance(v, str):
|
||
yield v
|
||
elif isinstance(v, dict):
|
||
for k, x in v.items():
|
||
yield k
|
||
yield from texts(x)
|
||
elif isinstance(v, list):
|
||
for x in v:
|
||
yield from texts(x)
|
||
|
||
# A \u escape that leaves a surrogate unpaired denotes no Unicode text (RFC 8259 §8.2).
|
||
if any(0xD800 <= ord(c) <= 0xDFFF for t in texts(o) for c in t) and MUTATE != "receipt-surrogates":
|
||
raise bad()
|
||
|
||
def obj(v, n):
|
||
if not isinstance(v, dict) or len(v) != n:
|
||
raise bad()
|
||
return v
|
||
|
||
def u64(v):
|
||
if isinstance(v, bool) or not isinstance(v, int) or not 0 <= v <= U64:
|
||
raise bad()
|
||
return v
|
||
|
||
def s(v):
|
||
if not isinstance(v, str):
|
||
raise bad()
|
||
return v
|
||
|
||
def hx(v, n):
|
||
if not isinstance(v, str) or not re.fullmatch("[0-9a-f]{%d}" % (2 * n), v):
|
||
raise bad()
|
||
return bytes.fromhex(v)
|
||
|
||
def opt(v, f):
|
||
return None if v is None else f(v)
|
||
|
||
if not isinstance(o, dict) or not set(o) <= RECEIPT_KNOWN or o.get("kind") != "wai.video.manifest":
|
||
raise bad()
|
||
for k in ("manifest_hash", "session_class_hash", "class", "segments", "merkle_root", "work_compose_ops", "viewers_served",
|
||
"joules_micro", "grant", "parent_receipt_hash", "signer_pubkey", "signer_id", "sig"):
|
||
if k not in o:
|
||
raise bad()
|
||
cl = obj(o["class"], 6)
|
||
if not isinstance(cl.get("blackouts"), list):
|
||
raise bad()
|
||
klass = {"content_id": s(cl.get("content_id")), "entitlement": s(cl.get("entitlement")), "geo": s(cl.get("geo")),
|
||
"blackouts": [s(b) for b in cl["blackouts"]], "ad_pod_id": s(cl.get("ad_pod_id")), "window": u64(cl.get("window"))}
|
||
segs = []
|
||
if not isinstance(o["segments"], list):
|
||
raise bad()
|
||
for x in o["segments"]:
|
||
x = obj(x, 5)
|
||
if x.get("kind") not in ("content", "ad") or not (x.get("ad_id") is None or isinstance(x.get("ad_id"), str)) or "ad_id" not in x:
|
||
raise bad()
|
||
segs.append({"index": u64(x.get("index")), "ad": x["kind"] == "ad", "uri": s(x.get("uri")), "ms": u64(x.get("duration_ms")), "ad_id": x["ad_id"]})
|
||
g = obj(o["grant"], 4)
|
||
if "funds_ceiling" not in g or "joule_ceiling_micro" not in g:
|
||
raise bad()
|
||
# No joule ceiling is written null and signs as 2^64 - 1.
|
||
grant = {"hash": hx(g.get("grant_hash"), 32), "capability": s(g.get("capability")),
|
||
"ceiling": U64 if g["joule_ceiling_micro"] is None else u64(g["joule_ceiling_micro"]), "funds": opt(g["funds_ceiling"], u64)}
|
||
label = o.get("energy_provenance")
|
||
if label is None:
|
||
if any(o.get(k) is not None for k in ("energy_uncertainty", "energy_counter_filtering", "energy_calibration_ref")):
|
||
raise bad()
|
||
prov = None
|
||
elif label in LEGACY_CLASSES and all(o.get(k) is None for k in ("energy_uncertainty", "energy_counter_filtering", "energy_calibration_ref")):
|
||
prov = label
|
||
else:
|
||
raise SystemExit("http_verify: an energy class this verifier does not read")
|
||
render_, extras = None, []
|
||
if ("render" in o) != ("extras" in o):
|
||
raise bad()
|
||
if "render" in o:
|
||
r = obj(o["render"], 11)
|
||
if not isinstance(r.get("can_block_reload"), bool) or not isinstance(r.get("end_list"), bool) or not isinstance(r.get("independent_segments"), bool):
|
||
raise bad()
|
||
if "playlist_type" not in r or r["playlist_type"] not in (None, "VOD", "EVENT") or "map_uri" not in r:
|
||
raise bad()
|
||
render_ = {"version": u64(r.get("version")), "seq": u64(r.get("media_sequence")), "dseq": u64(r.get("discontinuity_sequence")),
|
||
"target": opt(r.get("target_duration_s"), u64), "part_target": opt(r.get("part_target_ms"), u64),
|
||
"hold_back": opt(r.get("part_hold_back_ms"), u64), "block": r["can_block_reload"], "end": r["end_list"],
|
||
"ptype": r["playlist_type"], "indep": r["independent_segments"], "map": opt(r["map_uri"], s)}
|
||
for k in ("target_duration_s", "part_target_ms", "part_hold_back_ms"):
|
||
if k not in r:
|
||
raise bad()
|
||
if render_["version"] > 255 or not isinstance(o["extras"], list):
|
||
raise bad()
|
||
|
||
def layers(v):
|
||
if not isinstance(v, list):
|
||
raise bad()
|
||
out = []
|
||
for l in v:
|
||
l = obj(l, 2)
|
||
n = u64(l.get("n"))
|
||
if n > 255:
|
||
raise bad()
|
||
out.append({"n": n, "uri": s(l.get("uri"))})
|
||
return out
|
||
|
||
for e in o["extras"]:
|
||
e = obj(e, 6)
|
||
if not isinstance(e.get("gap"), bool) or not isinstance(e.get("discontinuity"), bool) or not isinstance(e.get("parts"), list) or "program_date_time" not in e or "preload_hint" not in e:
|
||
raise bad()
|
||
parts = []
|
||
for q in e["parts"]:
|
||
q = obj(q, 4)
|
||
if not isinstance(q.get("independent"), bool):
|
||
raise bad()
|
||
parts.append({"ms": u64(q.get("duration_ms")), "independent": q["independent"], "uri": s(q.get("uri")), "layers": layers(q.get("layers"))})
|
||
extras.append({"discontinuity": e["discontinuity"], "gap": e["gap"], "pdt": opt(e["program_date_time"], s), "parts": parts,
|
||
"layers": layers(e.get("layers")), "hint": opt(e["preload_hint"], s)})
|
||
return {"manifest_hash": hx(o["manifest_hash"], 32), "class_hash": hx(o["session_class_hash"], 32), "class": klass, "segments": segs,
|
||
"merkle_root": hx(o["merkle_root"], 32), "work": u64(o["work_compose_ops"]), "viewers": u64(o["viewers_served"]),
|
||
"joules": u64(o["joules_micro"]), "prov": prov, "grant": grant, "parent": opt(o["parent_receipt_hash"], lambda v: hx(v, 32)),
|
||
"pubkey": hx(o["signer_pubkey"], 32), "signer_id": s(o["signer_id"]), "sig": hx(o["sig"], 64), "render": render_, "extras": extras,
|
||
"render_json": (o.get("render"), o.get("extras"))}
|
||
|
||
|
||
def b3(*parts):
|
||
return blake3(b"".join(parts))
|
||
|
||
|
||
def be(n, width):
|
||
return n.to_bytes(width, "big")
|
||
|
||
|
||
def lp(s):
|
||
b = s.encode("utf-8")
|
||
return be(len(b), 4) + b
|
||
|
||
|
||
def segment_leaf(sg):
|
||
ad = (b"\x01" + lp(sg["ad_id"])) if sg["ad_id"] is not None else b"\x00"
|
||
return b3(b"wai:video-manifest-seg\x01", be(sg["index"], 8), bytes([2 if sg["ad"] else 1]), be(sg["ms"], 8), lp(sg["uri"]), ad)
|
||
|
||
|
||
def merkle_root(hashes):
|
||
if not hashes:
|
||
return bytes(32)
|
||
level = [b3(b"moq-jwp:merkle-leaf\x01", h) for h in hashes]
|
||
while len(level) > 1:
|
||
if len(level) % 2:
|
||
level.append(level[-1])
|
||
level = [b3(b"moq-jwp:merkle-node\x01", level[i], level[i + 1]) for i in range(0, len(level), 2)]
|
||
return level[0]
|
||
|
||
|
||
def class_hash(c):
|
||
bs = sorted(set(c["blackouts"]), key=lambda x: x.encode("utf-8"))
|
||
out = [b"wai:video-manifest-class\x01"] + [lp(c[k]) for k in ("content_id", "entitlement", "geo", "ad_pod_id")]
|
||
out.append(be(len(bs), 4))
|
||
out += [lp(b) for b in bs]
|
||
out.append(be(c["window"], 8))
|
||
return b3(*out)
|
||
|
||
|
||
def render_digest(r):
|
||
"""BLAKE3 of the render domain and the JCS of {"render", "extras"}, as the receipt
|
||
file carries them; None without a rendering or past the canonical subset."""
|
||
if r["render"] is None:
|
||
return None
|
||
try:
|
||
c = jcs({"render": r["render_json"][0], "extras": r["render_json"][1]})
|
||
except ValueError:
|
||
return None
|
||
return b3(b"wai:video-manifest-render\x01", c.encode("utf-8"))
|
||
|
||
|
||
def signing_payload(r):
|
||
o = [b"wai:video-manifest-receipt\x01", r["manifest_hash"], r["class_hash"], r["merkle_root"], be(r["work"], 8), be(r["viewers"], 8), be(r["joules"], 8),
|
||
bytes([LEGACY_CLASSES[r["prov"]] if r["prov"] else 0]), r["grant"]["hash"], lp(r["grant"]["capability"]), be(r["grant"]["ceiling"], 8),
|
||
(b"\x01" + be(r["grant"]["funds"], 8)) if r["grant"]["funds"] is not None else b"\x00",
|
||
(b"\x01" + r["parent"]) if r["parent"] is not None else b"\x00", be(len(r["segments"]), 8)]
|
||
o += [segment_leaf(sg) for sg in r["segments"]]
|
||
o += [r["pubkey"], lp(r["signer_id"])]
|
||
if r["render"] is not None and MUTATE != "render-digest-off":
|
||
d = render_digest(r)
|
||
o.append(b"\x01" + d if d is not None else b"\x02")
|
||
return b"".join(o)
|
||
|
||
|
||
def receipt_hash(r):
|
||
return b3(b"wai:video-manifest-receipt-id\x01", signing_payload(r), r["sig"])
|
||
|
||
|
||
def rendered(r):
|
||
rr = r["render"] or {"version": 6, "seq": 0, "dseq": 0, "target": None, "part_target": None, "hold_back": None, "block": False, "end": True,
|
||
"ptype": None, "indep": False, "map": None}
|
||
return render_hls(r["segments"], rr, r["extras"])
|
||
|
||
|
||
def receipt_verifies(r):
|
||
if r["render"] is None and r["extras"]:
|
||
return False
|
||
if r["render"] is not None:
|
||
try:
|
||
check_render(r["segments"], r["render"], r["extras"])
|
||
except Refusal:
|
||
return False
|
||
if render_digest(r) is None:
|
||
return False
|
||
if blake3(rendered(r).encode("utf-8")) != r["manifest_hash"]:
|
||
return False
|
||
if merkle_root([segment_leaf(sg) for sg in r["segments"]]) != r["merkle_root"] or class_hash(r["class"]) != r["class_hash"]:
|
||
return False
|
||
if r["work"] != len(r["segments"]) or r["viewers"] == 0:
|
||
return False
|
||
if any(sg["index"] != i or sg["ad"] != (sg["ad_id"] is not None) for i, sg in enumerate(r["segments"])):
|
||
return False
|
||
if (r["prov"] is None) != (r["joules"] == 0):
|
||
return False
|
||
return ed25519_verify(r["pubkey"], signing_payload(r), r["sig"])
|
||
|
||
|
||
def check_playlist(playlist, receipt_bytes, name):
|
||
"""§8.6, §11.6, in order: the file reads; its name; the bytes; the receipt."""
|
||
try:
|
||
r = read_receipt(receipt_bytes.decode("utf-8"))
|
||
except (ReceiptMalformed, UnicodeDecodeError):
|
||
raise Refusal("playlist_receipt_malformed")
|
||
if name is not None and name != r["manifest_hash"].hex() + ".json" and MUTATE != "name-unchecked":
|
||
raise Refusal("receipt_name_mismatch")
|
||
if blake3(playlist) != r["manifest_hash"] or rendered(r).encode("utf-8") != playlist:
|
||
raise Refusal("playlist_receipt_mismatch")
|
||
if not receipt_verifies(r):
|
||
raise Refusal("playlist_receipt_invalid")
|
||
return r
|
||
|
||
|
||
WALK_BOUND = 64
|
||
LIVE_URL = "https://origin.example/live/wai-720.m3u8"
|
||
|
||
|
||
def advance(chain, cur, fetch):
|
||
"""§8.6: take the receipt cur (checked against its playlist) into the chain a client
|
||
holds, {"head", "bindings", "bound"}; fetch(h) gives the bytes at mr/<h>.json or None.
|
||
Returns ("first" | "unchanged" | "chained" | "reanchored", walked) and updates chain,
|
||
or raises Refusal and leaves it as it was."""
|
||
if chain["head"] is None:
|
||
b = bind_receipt(chain["bindings"], cur)
|
||
chain["head"], chain["bindings"] = cur, b
|
||
return "first", 0
|
||
target = receipt_hash(chain["head"])
|
||
if receipt_hash(cur) == target and MUTATE != "reload-walked":
|
||
return "unchanged", 0
|
||
if cur["pubkey"] != chain["head"]["pubkey"] and MUTATE != "key-unchecked":
|
||
raise Refusal("playlist_receipt_unchained")
|
||
walked, link = [], cur["parent"]
|
||
while MUTATE != "chain-unchecked":
|
||
if link is None:
|
||
raise Refusal("playlist_receipt_unchained")
|
||
if link == target:
|
||
break
|
||
if len(walked) == chain["bound"] and MUTATE != "walk-unbounded":
|
||
# Past the bound: a revision ahead of the head re-anchors the chain, its
|
||
# bindings kept; one that is not ahead is refused.
|
||
if last_segment(cur) <= last_segment(chain["head"]) and MUTATE != "reanchor-backwards":
|
||
raise Refusal("playlist_receipt_walk_exceeded")
|
||
if MUTATE == "walk-exceeded-wedges":
|
||
raise Refusal("playlist_receipt_walk_exceeded")
|
||
chain["bindings"] = bind_receipt(chain["bindings"], cur)
|
||
chain["head"] = cur
|
||
return "reanchored", len(walked)
|
||
raw = fetch(link)
|
||
if raw is None:
|
||
raise Refusal("playlist_receipt_parent_missing")
|
||
try:
|
||
r = read_receipt(raw.decode("utf-8"))
|
||
except (ReceiptMalformed, UnicodeDecodeError):
|
||
raise Refusal("playlist_receipt_malformed")
|
||
if receipt_hash(r) != link and MUTATE != "walk-name-unchecked":
|
||
raise Refusal("receipt_name_mismatch")
|
||
if not receipt_verifies(r) and MUTATE != "walk-verify-unchecked":
|
||
raise Refusal("playlist_receipt_invalid")
|
||
if r["pubkey"] != cur["pubkey"] and MUTATE != "key-unchecked":
|
||
raise Refusal("playlist_receipt_unchained")
|
||
link = r["parent"]
|
||
walked.append(r)
|
||
b = chain["bindings"]
|
||
for r in list(reversed(walked)) + [cur]:
|
||
b = bind_receipt(b, r)
|
||
chain["head"], chain["bindings"] = cur, b
|
||
return "chained", len(walked)
|
||
|
||
|
||
def last_segment(r):
|
||
"""The media sequence number after the last complete segment r lists."""
|
||
return (r["render"]["seq"] if r["render"] is not None else 0) + len(r["segments"])
|
||
|
||
|
||
def bind_receipt(b, r):
|
||
try:
|
||
p = parse_media_playlist(rendered(r))
|
||
except Refusal:
|
||
raise Refusal("playlist_receipt_malformed")
|
||
try:
|
||
return bind(b, p)
|
||
except Refusal as e:
|
||
raise Refusal("playlist_listing_rebound" if e.code == "playlist_listing_rebound" else "playlist_receipt_malformed")
|
||
|
||
|
||
# ---------------------------------------------------- the playlist-side cases
|
||
|
||
def edit_text(t, edits):
|
||
for e in edits or []:
|
||
if e[0] not in t:
|
||
raise SystemExit("http_verify: a case edits text its file does not hold")
|
||
t = t.replace(e[0], e[1], 1)
|
||
return t
|
||
|
||
|
||
def read_text(d, path):
|
||
return open(os.path.join(d, path), "rb").read().decode("utf-8")
|
||
|
||
|
||
def layer_of(env):
|
||
h = whole_head(env)
|
||
if not is_layer(h):
|
||
raise Refusal("not_a_layer")
|
||
try:
|
||
r = SV.parse_refines(h["manifest"].get("refines"))
|
||
except SV.Malformed:
|
||
raise Refusal("not_a_layer")
|
||
return {"n": r["layer"], "cid": content_hash(env), "len": len(env)}
|
||
|
||
|
||
def parts_of(objs, ps, cms):
|
||
out = []
|
||
for j, p in enumerate(ps or []):
|
||
env = objs[p["object"]]
|
||
ms = cms[j] if j < len(cms) else p.get("duration_ms")
|
||
if not isinstance(ms, int):
|
||
raise SystemExit("http_verify: a part with no duration")
|
||
whole_head(env)
|
||
out.append({"alias": p["alias"], "cid": content_hash(env), "n": len(env), "ms": ms, "independent": p.get("independent") is True,
|
||
"layers": [layer_of(objs[x]) for x in p.get("layers") or []]})
|
||
return out
|
||
|
||
|
||
def render_media_case(d, objs, inp):
|
||
c = parse_media_playlist(read_text(d, inp["classical"]))
|
||
segs = []
|
||
for i, s in enumerate(inp["segments"]):
|
||
k = c["segments"][i] if i < len(c["segments"]) else None
|
||
env = objs[s["object"]]
|
||
h = whole_head(env)
|
||
w = {"cid": content_hash(env), "n": len(env), "h": covered_len(h["head_len"], len(env)) if h["version"] == "WAI2" else None,
|
||
"ms": s["duration_ms"] if isinstance(s.get("duration_ms"), int) else (k["ms"] if k else 0), "gap": s.get("gap") is True}
|
||
w["layers"] = [layer_of(objs[x]) for x in s.get("layers") or []]
|
||
w["parts"] = parts_of(objs, s.get("parts"), [q["ms"] for q in k["extra"]["parts"]] if k else [])
|
||
segs.append(w)
|
||
open_ = None
|
||
if inp.get("open") is not None:
|
||
o = inp["open"]
|
||
cms = [q["ms"] for q in c["open"]["parts"]] if c["open"] else []
|
||
open_ = {"parts": parts_of(objs, o.get("parts"), cms), "hint": o.get("preload_hint")}
|
||
end = inp["end_list"] if isinstance(inp.get("end_list"), bool) else c["end"]
|
||
return wai_media_playlist(c, segs, open_, inp.get("uri_prefix") or "", end)
|
||
|
||
|
||
def resolve_uri(objs, uri):
|
||
base, _, frag = uri.partition("#")
|
||
cid = None
|
||
for kv in frag.split("&") if frag else []:
|
||
k, eq, v = kv.partition("=")
|
||
if eq and k == "cid" and re.fullmatch(r"[0-9a-f]{64}", v):
|
||
cid = v
|
||
if cid is None:
|
||
name = base.rsplit("/", 1)[-1]
|
||
if name.endswith(".wai") and re.fullmatch(r"[0-9a-f]{64}", name[:-4]):
|
||
cid = name[:-4]
|
||
b = next((b for b in objs.values() if cid is not None and content_hash(b).hex() == cid), None)
|
||
if b is None:
|
||
raise SystemExit("http_verify: %s names no corpus object" % uri)
|
||
return b
|
||
|
||
|
||
def listed_objects(objs, p):
|
||
segs, parts, every = [], [], []
|
||
|
||
def extra(e):
|
||
for q in e["parts"]:
|
||
b = resolve_uri(objs, q["uri"])
|
||
parts.append((len(b), q["ms"]))
|
||
every.append(b)
|
||
every.extend(resolve_uri(objs, l["uri"]) for l in q["layers"])
|
||
every.extend(resolve_uri(objs, l["uri"]) for l in e["layers"])
|
||
|
||
for s in p["segments"]:
|
||
if s["extra"]["gap"]:
|
||
continue
|
||
b = resolve_uri(objs, s["uri"])
|
||
every.append(b)
|
||
segs.append((len(b) + sum(len(resolve_uri(objs, l["uri"])) for l in s["extra"]["layers"]), s["ms"]))
|
||
extra(s["extra"])
|
||
if p["open"] is not None:
|
||
extra(p["open"])
|
||
return segs, parts, every
|
||
|
||
|
||
def union_floor_of(every):
|
||
so = [seq_object(e) for e in every]
|
||
return union_of(so), floor_of(so)
|
||
|
||
|
||
def multivariant_case(d, objs, inp):
|
||
out = ["#EXTM3U"]
|
||
for v in inp["variants"]:
|
||
p = parse_media_playlist(read_text(d, v["playlist"]))
|
||
segs, parts, every = listed_objects(objs, p)
|
||
union, floor = union_floor_of(every)
|
||
bw = bandwidths(segs if segs else parts)
|
||
if bw is None:
|
||
raise SystemExit("http_verify: a variant with no duration")
|
||
uri = v["uri"]
|
||
if not uri_ok(uri, False) or union[0] != floor[0]:
|
||
raise Refusal("playlist_malformed")
|
||
line = '#EXT-X-STREAM-INF:BANDWIDTH=%d,AVERAGE-BANDWIDTH=%d,CODECS="%s",X-WAI-FLOOR="%s"' % (bw[0], bw[1], render(union), render(floor))
|
||
if v.get("resolution"):
|
||
line += ",RESOLUTION=%dx%d" % tuple(v["resolution"])
|
||
if v.get("frame_rate_milli") is not None:
|
||
if v["frame_rate_milli"] == 0:
|
||
raise Refusal("playlist_malformed")
|
||
line += ",FRAME-RATE=" + secs(v["frame_rate_milli"])
|
||
out += [line, uri]
|
||
return "\n".join(out) + "\n"
|
||
|
||
|
||
def mpd_case(d, objs, inp):
|
||
p = parse_media_playlist(read_text(d, inp["playlist"]))
|
||
segs, _, every = listed_objects(objs, p)
|
||
union, floor = union_floor_of(every)
|
||
peak = bandwidths(segs)[0]
|
||
if isinstance(inp.get("classical_mpd"), str):
|
||
t = mpd_timeline(read_text(d, inp["classical_mpd"]), inp.get("period_id"), inp.get("classical_set"), len(p["segments"]))
|
||
else:
|
||
c = parse_media_playlist(read_text(d, inp["classical"]))
|
||
at, segs_t = 0, []
|
||
for s in c["segments"]:
|
||
segs_t.append((at, s["ms"]))
|
||
at += s["ms"]
|
||
if at > U64:
|
||
raise Refusal("playlist_malformed")
|
||
t = (inp.get("period_id") or "p0", 0, 1000, segs_t)
|
||
media = [None if s["extra"]["gap"] else s["uri"] for s in p["segments"]]
|
||
return wai_mpd(t, inp.get("set_id") or 1, union, floor, inp["rep_id"], peak, media)
|
||
|
||
|
||
def rendered_case(d, inp, t):
|
||
b = t.encode("utf-8")
|
||
o = {"blake3": blake3(b).hex(), "bytes": len(b)}
|
||
if "file" in inp:
|
||
o["same_as_file"] = os.path.exists(os.path.join(d, inp["file"])) and open(os.path.join(d, inp["file"]), "rb").read() == b
|
||
return o
|
||
|
||
|
||
def parsed_case(p, t):
|
||
parts = lambda ps: [[q["ms"], q["independent"], [l["n"] for l in q["layers"]]] for q in ps]
|
||
return {"version": p["version"], "target_duration_s": p["target"], "media_sequence": p["seq"], "discontinuity_sequence": p["dseq"],
|
||
"part_target_ms": p["part_target"], "part_hold_back_ms": p["hold_back"], "can_block_reload": p["block"], "end_list": p["end"],
|
||
"playlist_type": p["ptype"], "independent_segments": p["indep"], "map_uri": p["map"],
|
||
"segments": [[s["ms"], s["extra"]["discontinuity"], s["extra"]["pdt"], parts(s["extra"]["parts"]), [l["n"] for l in s["extra"]["layers"]], s["uri"], s["extra"]["gap"]] for s in p["segments"]],
|
||
"open": None if p["open"] is None else [p["open"]["discontinuity"], p["open"]["pdt"], parts(p["open"]["parts"])],
|
||
"preload_hint": p["hint"], "rendered_form": is_rendered_form(p, t)}
|
||
|
||
|
||
def live_case(d, objs, inp):
|
||
"""§8.6, §11.4, §11.6 over a live playlist's revisions in the order a client fetched
|
||
them, and parts fetched by alias."""
|
||
revisions = inp["revisions"]
|
||
parents = inp.get("parents")
|
||
if parents is None:
|
||
parents = (revisions[0]["wai"].rsplit("/", 1)[0] if "/" in revisions[0]["wai"] else "") + "/mr" if revisions else "/mr"
|
||
|
||
def fetch(h):
|
||
f = os.path.join(d, parents, h.hex() + ".json")
|
||
return open(f, "rb").read() if os.path.isfile(f) else None
|
||
|
||
url = inp["url"] if isinstance(inp.get("url"), str) else LIVE_URL
|
||
chain = {"head": None, "bindings": new_bindings(url), "bound": inp["walk_bound"] if isinstance(inp.get("walk_bound"), int) else WALK_BOUND}
|
||
listed_b = new_bindings(url)
|
||
revs, listed = [], []
|
||
for r in revisions:
|
||
w = edit_text(read_text(d, r["wai"]), r.get("edit_wai"))
|
||
k = read_text(d, r["classical"])
|
||
try:
|
||
check_adjacent(w, k)
|
||
adjacent = True
|
||
except Refusal as e:
|
||
adjacent = e.code
|
||
p = parse_media_playlist(w)
|
||
try:
|
||
listed_b = bind(listed_b, p)
|
||
except Refusal as e:
|
||
revs.append({"adjacent": adjacent, "listing": e.code, "receipt": None, "chained": None, "walked": None})
|
||
listed.append(None)
|
||
continue
|
||
listed.append(p)
|
||
rpath = r["receipt"]
|
||
name = None if rpath.startswith("cases/") else rpath.rsplit("/", 1)[-1]
|
||
try:
|
||
rr = check_playlist(w.encode("utf-8"), open(os.path.join(d, rpath), "rb").read(), name)
|
||
except Refusal as e:
|
||
revs.append({"adjacent": adjacent, "listing": True, "receipt": e.code, "chained": None, "walked": None})
|
||
continue
|
||
try:
|
||
step, walked = advance(chain, rr, fetch)
|
||
ch = step if step in ("unchanged", "reanchored") else True
|
||
except Refusal as e:
|
||
ch, walked = e.code, None
|
||
revs.append({"adjacent": adjacent, "listing": True, "receipt": "signed", "chained": ch, "walked": walked})
|
||
fetches = []
|
||
for f in inp.get("fetches") or []:
|
||
served = apply_patch(objs[f["object"]], f.get("patch_object"))
|
||
first = next(((k, c) for k, p in enumerate(listed) if p is not None for c in [listed_cid(p, f["alias"])] if c is not None), None)
|
||
if first is None:
|
||
fetches.append({"alias": f["alias"], "listed_in": None, "outcome": "not_listed", "held": None})
|
||
elif content_hash(served) == first[1]:
|
||
held = "signed" if cid_of(chain["bindings"], f["alias"]) == first[1] else "listed"
|
||
fetches.append({"alias": f["alias"], "listed_in": first[0], "outcome": "accepted", "held": held})
|
||
else:
|
||
fetches.append({"alias": f["alias"], "listed_in": first[0], "outcome": "object_digest_mismatch", "held": None})
|
||
return {"revisions": revs, "fetches": fetches}
|
||
|
||
|
||
def playlist_evaluate(d, objs, op, inp, obj):
|
||
"""The playlist and adaptation ops; None for any other op."""
|
||
if op == "media_playlist":
|
||
return rendered_case(d, inp, render_media_case(d, objs, inp))
|
||
if op == "multivariant":
|
||
return rendered_case(d, inp, multivariant_case(d, objs, inp))
|
||
if op == "mpd":
|
||
return rendered_case(d, inp, mpd_case(d, objs, inp))
|
||
if op == "playlist_parse":
|
||
t = inp["text"] if "text" in inp else read_text(d, inp["file"])
|
||
return parsed_case(parse_media_playlist(t), t)
|
||
inline = lambda k: inp[k + "_text"] if isinstance(inp.get(k + "_text"), str) else read_text(d, inp[k])
|
||
if op == "uri_resolve":
|
||
return {"url": rfc3986_resolve(inp["base"], inp["reference"])}
|
||
if op == "dash_adjacency":
|
||
check_dash_adjacent(edit_text(inline("wai"), inp.get("edit_wai")), edit_text(inline("classical"), inp.get("edit_classical")))
|
||
return {"adjacent": True}
|
||
if op == "adjacency":
|
||
check_adjacent(edit_text(inline("wai"), inp.get("edit_wai")), edit_text(inline("classical"), inp.get("edit_classical")))
|
||
return {"adjacent": True}
|
||
if op == "unchanged":
|
||
classical_unchanged(inline("original"), edit_text(inline("published"), inp.get("edit_published")))
|
||
return {"unchanged": True}
|
||
if op == "render":
|
||
segs = [{"ms": x[0], "uri": x[1], "ad": x[2] is True} for x in inp["segments"]]
|
||
r = inp["render"]
|
||
rr = {"version": r["version"] % 256, "seq": r["media_sequence"], "dseq": r["discontinuity_sequence"], "target": r["target_duration_s"],
|
||
"part_target": r["part_target_ms"], "hold_back": r["part_hold_back_ms"], "block": r["can_block_reload"] is True, "end": r["end_list"] is True,
|
||
"ptype": r.get("playlist_type") if r.get("playlist_type") in ("VOD", "EVENT") else None, "indep": r.get("independent_segments") is True,
|
||
"map": r.get("map_uri") if isinstance(r.get("map_uri"), str) else None}
|
||
txt = lambda v: v if isinstance(v, str) else None
|
||
lays = lambda ls: [{"n": l["n"] % 256, "uri": l["uri"]} for l in ls or []]
|
||
extras = [{"discontinuity": e.get("discontinuity") is True, "gap": e.get("gap") is True, "pdt": txt(e.get("program_date_time")),
|
||
"parts": [{"ms": q["duration_ms"], "independent": q.get("independent") is True, "uri": q["uri"], "layers": lays(q.get("layers"))} for q in e.get("parts") or []],
|
||
"layers": lays(e.get("layers")), "hint": txt(e.get("preload_hint"))} for e in inp.get("extras") or []]
|
||
check_render(segs, rr, extras)
|
||
t = render_hls(segs, rr, extras).encode("utf-8")
|
||
return {"blake3": blake3(t).hex(), "bytes": len(t)}
|
||
if op == "ilog2":
|
||
return {"log2_q16": ilog2_q16(inp["a"], inp["b"])}
|
||
if op == "playlist_receipt":
|
||
pl = edit_text(read_text(d, inp["playlist"]), inp.get("edit_playlist")).encode("utf-8")
|
||
rc = edit_text(read_text(d, inp["receipt"]), inp.get("edit_receipt")).encode("utf-8")
|
||
name = inp["name"] if isinstance(inp.get("name"), str) else (inp["receipt"].rsplit("/", 1)[-1] if inp.get("name") is None else None)
|
||
r = check_playlist(pl, rc, name)
|
||
return {"signed": receipt_hash(r).hex(), "parent": r["parent"].hex() if r["parent"] is not None else None}
|
||
if op == "live":
|
||
return live_case(d, objs, inp)
|
||
if op == "abr":
|
||
k, b = decide(inp["policy"], abr_state(inp["state"]))
|
||
return {"choice": k, "budget": b}
|
||
if op == "abr_order":
|
||
return {"order": order_entries(whole_head(obj(inp)), inp.get("advertises") or [], inp["policy"], abr_state(inp["state"]))}
|
||
if op == "depth":
|
||
layers = [len(objs[i]) for i in inp["objects"]] if "objects" in inp else inp["layers_n"]
|
||
k, b = depth(layers, inp["policy"], abr_state(inp["state"]))
|
||
return {"depth": k, "budget": b}
|
||
return None
|
||
|
||
|
||
def refusal(e):
|
||
out = {"error": e.code}
|
||
if e.skipped is not None:
|
||
out["skipped"] = e.skipped
|
||
return out
|
||
|
||
|
||
def evaluate(d, objs, op, inp):
|
||
def obj(i, patch=True):
|
||
if "envelope_hex" in i:
|
||
b = bytes.fromhex(i["envelope_hex"])
|
||
elif "object" in i:
|
||
b = objs[i["object"]]
|
||
else:
|
||
b = open(os.path.join(d, i["file"]), "rb").read()
|
||
return apply_patch(b, i.get("patch_object")) if patch else b
|
||
|
||
try:
|
||
if op == "parse_head":
|
||
b = obj(inp)
|
||
n = inp.get("envelope_len", len(b))
|
||
p = min(inp.get("prefix", len(b)), len(b))
|
||
h = parse_head(b[:p], n)
|
||
if isinstance(h, tuple):
|
||
return {"need": h[1]}
|
||
return {"version": h["version"], "head_len": h["head_len"], "envelope_len": h["envelope_len"],
|
||
"covered_len": covered_len(h["head_len"], n), "components": [list(c) for c in h["components"]]}
|
||
if op == "head_proof":
|
||
b = obj(inp)
|
||
original = obj(inp, patch=False)
|
||
cid = content_hash(original)
|
||
pf = open(os.path.join(d, inp["proof"]), "rb").read() if "proof" in inp else make_proof_file(original)
|
||
p = read_proof(pf)
|
||
verify_proof(p, cid, b)
|
||
return {"covered_len": p["covered"], "n_cv": len(p["cvs"]), "envelope_len": p["n"]}
|
||
if op == "plan":
|
||
b = obj(inp)
|
||
h = parse_head(b, len(b))
|
||
pl = plan(h, [tuple(w) for w in inp["wanted"]], [tuple(r) for r in inp.get("held", [])], inp.get("threshold_permille", 900))
|
||
if pl[0] == "rest":
|
||
return {"whole_rest": list(pl[1]), "why": pl[2]}
|
||
return {"requests": [list(r) for r in pl[1]], "dropped": pl[2]}
|
||
if op == "response":
|
||
r = check_response(inp.get("request"), inp["status"], inp.get("content_range"), inp.get("content_encoding"), inp["body_len"], inp["envelope_len"])
|
||
return {"ok": list(r)}
|
||
if op == "select":
|
||
b = obj(inp)
|
||
h = whole_head(b)
|
||
pay = [b[s:s + ln] for s, ln in h["components"]]
|
||
adv, _, held, order, _, _ = sink_of(d, inp, objs)
|
||
order = order(h) if callable(order) else (order if order is not None else list(range(len(pay))))
|
||
try:
|
||
i, sk = select(h["manifest"], order, adv, held, lambda i, e: payload_ok(e, pay[i]))
|
||
return {"pick": i, "skipped": sk}
|
||
except Refusal as e:
|
||
return {"inert": e.skipped}
|
||
if op in ("fetch", "same_pick"):
|
||
env = obj(inp, patch=False)
|
||
t = inp.get("tamper") or {}
|
||
served = apply_patch(env, t.get("patch_object"))
|
||
r = fetch(env, served, frag_of(inp.get("fragment")), sink_of(d, inp, objs), inp.get("threshold_permille", 900), t,
|
||
inp.get("listed_as_layer", False))
|
||
if op == "fetch":
|
||
return r
|
||
whole_pick = evaluate(d, objs, "select", inp)
|
||
if "pick" not in whole_pick or whole_pick["pick"] != r["pick"]:
|
||
return {"paths_differ": [whole_pick, r]}
|
||
return {"pick": r["pick"]}
|
||
if op == "fetch_timed":
|
||
env = obj(inp, patch=False)
|
||
sk = sink_of(d, inp, objs)
|
||
return fetch_timed(env, frag_of(inp.get("fragment")), sk, inp.get("threshold_permille", 900), inp.get("deadline_ms"),
|
||
inp["script"], inp.get("flip"))
|
||
if op == "codecs":
|
||
return {"elements": parse_codecs(inp["value"])}
|
||
if op == "union_floor":
|
||
envs = [bytes.fromhex(x) for x in inp["envelopes_hex"]] if "envelopes_hex" in inp else [objs[i] for i in inp["objects"]]
|
||
os_ = [seq_object(e) for e in envs]
|
||
return {"union": render(union_of(os_)), "floor": render(floor_of(os_))}
|
||
if op == "supports":
|
||
return {"supports": supports(parse_element(inp["floor"]), inp["containers"], inp["caps"])}
|
||
if op == "layer_tag":
|
||
return parse_layer_tag(inp["line"])
|
||
r = playlist_evaluate(d, objs, op, inp, obj)
|
||
if r is not None:
|
||
return r
|
||
except Refusal as e:
|
||
if op == "same_pick" and e.code == "inert":
|
||
try:
|
||
evaluate(d, objs, "select", inp)["inert"]
|
||
return {"inert": True}
|
||
except KeyError:
|
||
return {"paths_differ": "range inert, whole picks"}
|
||
return refusal(e)
|
||
raise SystemExit("unknown op " + op)
|
||
|
||
|
||
# ------------------------------------------------------------------ self-test
|
||
|
||
def self_test():
|
||
v = json.load(open(os.path.join(HERE, "blake3_vectors.json")))
|
||
for c in v["cases"]:
|
||
data = bytes(i % 251 for i in range(c["input_len"]))
|
||
got = blake3(data)
|
||
if MUTATE == "blake3-iv":
|
||
got = bytes([got[0] ^ 1]) + got[1:]
|
||
if got.hex() != c["hash"]:
|
||
print("FAIL self-test: BLAKE3 of %d bytes" % c["input_len"])
|
||
return False
|
||
if c["input_len"] > 1024:
|
||
n = c["input_len"]
|
||
l = 1024 * 2 ** ((((n - 1) // 1024)).bit_length() - 1)
|
||
if B3.parent_cv(B3.subtree_cv(data[:l], 0), B3.subtree_cv(data[l:], l), True) != got:
|
||
print("FAIL self-test: the tree functions over %d bytes" % n)
|
||
return False
|
||
return mpd_bounds_hold()
|
||
|
||
|
||
def mpd_bounds_hold():
|
||
"""Appendix I's reader holds what a document writes, not what its @r says, and reads
|
||
an element's attributes in linear time: 8 timelines of 2^16 segments each in a
|
||
kilobyte, read in a footprint bounded by its size (tracemalloc's peak); and one
|
||
element of 40000 attributes, read in bounded CPU time of this thread (never wall
|
||
time)."""
|
||
import tracemalloc
|
||
bomb = "<MPD><Period id=\"p\"><AdaptationSet>%s</AdaptationSet></Period></MPD>" % "".join(
|
||
'<Representation id="r%d"><SegmentList timescale="1"><SegmentTimeline><S t="0" d="1" r="65535"/></SegmentTimeline></SegmentList></Representation>' % i
|
||
for i in range(8))
|
||
tracemalloc.start()
|
||
try:
|
||
periods = read_mpd(bomb)
|
||
peak = tracemalloc.get_traced_memory()[1]
|
||
finally:
|
||
tracemalloc.stop()
|
||
if sum(n for tl in periods[0]["sets"][0]["timelines"] for _, _, n in tl["runs"]) != 8 << 16 or peak > 64 * len(bomb) + (1 << 20):
|
||
print("FAIL self-test: an MPD of %d bytes held %d bytes" % (len(bomb), peak))
|
||
return False
|
||
wide = "<MPD %s/>" % " ".join('a%d=""' % i for i in range(40000))
|
||
t0 = time.thread_time()
|
||
try:
|
||
read_mpd(wide)
|
||
except Refusal:
|
||
pass
|
||
spent = time.thread_time() - t0
|
||
if spent > 2.0:
|
||
print("FAIL self-test: one element of 40000 attributes read in %.2f s of CPU" % spent)
|
||
return False
|
||
return bindings_retention_holds()
|
||
|
||
|
||
def bindings_retention_holds():
|
||
"""§11.5: a binding is kept while one of the last RETENTION revisions lists it, then
|
||
forgotten, so a rebinding is refused at the span's last revision and not after it."""
|
||
cid = lambda b: bytes([b]) * 32
|
||
first = parse_media_playlist('#EXTM3U\n#EXT-X-TARGETDURATION:1\n#EXT-X-PART-INF:PART-TARGET=0.5\n#EXT-X-PART:DURATION=0.5,URI="p.wai#cid=%s"\n' % cid(1).hex())
|
||
again = parse_media_playlist('#EXTM3U\n#EXT-X-TARGETDURATION:1\n#EXT-X-PART-INF:PART-TARGET=0.5\n#EXT-X-PART:DURATION=0.5,URI="./p.wai#cid=%s"\n' % cid(2).hex())
|
||
other = parse_media_playlist("#EXTM3U\n#EXT-X-TARGETDURATION:1\n#EXTINF:1,\n%s.wai\n" % cid(3).hex())
|
||
b = bind(new_bindings("https://origin.example/live/p.m3u8"), first)
|
||
for _ in range(RETENTION - 1):
|
||
b = bind(b, other)
|
||
try:
|
||
bind(b, again)
|
||
print("FAIL self-test: a rebinding within the retention was taken")
|
||
return False
|
||
except Refusal:
|
||
pass
|
||
b = bind(b, other)
|
||
try:
|
||
bind(b, again)
|
||
except Refusal:
|
||
print("FAIL self-test: a binding outlived the retention")
|
||
return False
|
||
return True
|
||
|
||
|
||
def run_cases(d, path):
|
||
"""Evaluate every case of a case file (the corpus's format) against the corpus at d,
|
||
and require each committed outcome."""
|
||
load_registry()
|
||
objs = {i: open(os.path.join(d, o["path"]), "rb").read() for i, o in json.load(open(os.path.join(d, "objects.json"))).items()}
|
||
cases = json.load(open(path))
|
||
bad = 0
|
||
for c in cases:
|
||
got = evaluate(d, objs, c["op"], c["input"])
|
||
if got != c["expect"]:
|
||
bad += 1
|
||
if bad <= 20:
|
||
print("DIFF %s: rust %s, python %s" % (c["name"], json.dumps(c["expect"]), json.dumps(got)))
|
||
print("%d/%d cases agree" % (len(cases) - bad, len(cases)))
|
||
return bad == 0
|
||
|
||
|
||
def run(d):
|
||
load_registry()
|
||
ok = self_test()
|
||
listing = json.load(open(os.path.join(d, "objects.json")))
|
||
objs = {}
|
||
for i, o in listing.items():
|
||
b = open(os.path.join(d, o["path"]), "rb").read()
|
||
if content_hash(b).hex() != o["cid"]:
|
||
print("FAIL object %s: its bytes do not hash to its cid" % i)
|
||
ok = False
|
||
if open(os.path.join(d, o["path"] + ".hp"), "rb").read() != make_proof_file(b):
|
||
print("FAIL object %s: its head proof is not the one the text gives" % i)
|
||
ok = False
|
||
objs[i] = b
|
||
cases = json.load(open(os.path.join(d, "cases.json")))
|
||
passed = 0
|
||
for c in cases:
|
||
got = evaluate(d, objs, c["op"], c["input"])
|
||
if got == c["expect"]:
|
||
passed += 1
|
||
else:
|
||
print("FAIL %s: expected %s, got %s" % (c["name"], json.dumps(c["expect"]), json.dumps(got)))
|
||
ok = False
|
||
print("%d/%d cases give their outcome%s" % (passed, len(cases), " (mutation %s)" % MUTATE if MUTATE else ""))
|
||
return ok
|
||
|
||
|
||
def diff(path, flip, seed):
|
||
load_registry()
|
||
lines = [json.loads(l) for l in open(path) if l.strip()]
|
||
rng = random.Random(seed)
|
||
candidates = [i for i, l in enumerate(lines) if len(bytes.fromhex(l["proof"])) > 53]
|
||
flipped = set(rng.sample(candidates, flip)) if flip else set()
|
||
refused = set()
|
||
for i, l in enumerate(lines):
|
||
proof = bytearray(bytes.fromhex(l["proof"]))
|
||
if i in flipped:
|
||
k = proof[52]
|
||
at = 53 + 32 * rng.randrange(k) + rng.randrange(32)
|
||
proof[at] ^= 1 << rng.randrange(8)
|
||
try:
|
||
p = read_proof(bytes(proof))
|
||
if p["n"] != l["envelope_len"]:
|
||
raise Refusal("head_proof_malformed")
|
||
verify_proof(p, bytes.fromhex(l["cid"]), bytes.fromhex(l["prefix"]))
|
||
except Refusal:
|
||
refused.add(i)
|
||
print("%d proofs; %d flipped; %d refused" % (len(lines), len(flipped), len(refused)))
|
||
if refused != flipped:
|
||
print("FAIL: refused %s, flipped %s" % (sorted(refused - flipped)[:10], sorted(flipped - refused)[:10]))
|
||
return False
|
||
return True
|
||
|
||
|
||
def main(argv):
|
||
global MUTATE
|
||
if "--mutate" in argv:
|
||
i = argv.index("--mutate")
|
||
MUTATE = argv[i + 1]
|
||
argv = argv[:i] + argv[i + 2:]
|
||
if argv and argv[0] == "cases":
|
||
return 0 if run_cases(argv[2] if len(argv) > 2 else HERE, argv[1]) else 1
|
||
if argv and argv[0] == "diff":
|
||
flip = int(argv[argv.index("--flip") + 1]) if "--flip" in argv else 0
|
||
seed = int(argv[argv.index("--seed") + 1]) if "--seed" in argv else 1
|
||
return 0 if diff(argv[1], flip, seed) else 1
|
||
d = argv[0] if argv else HERE
|
||
return 0 if run(d) else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv[1:]))
|