27d76115e0
Written from extensions/key-release.md, sharing no code with wai-rs: it reads the scheme, delivery, decision and reason tables from the extension, derives each list from its CPIX document with the standard library's XML parser, recomputes every decision of every log, and requires every mutated copy of the corpus to fail. Co-Authored-By: Claude Opus 5.5 <[email protected]>
865 lines
38 KiB
Python
865 lines
38 KiB
Python
#!/usr/bin/env python3
|
||
"""Independent verifier of the key-release corpus (extensions/key-release.md).
|
||
|
||
Written from the extension's text. It shares no code with wai-rs, which generates the corpus,
|
||
and never runs it. It checks Ed25519 signatures with ../claim-conformance/_ed25519.py, WAI's
|
||
strict rule (SPEC §7 *Signatures*) written from RFC 8032 with the standard library, and
|
||
computes BLAKE3 with ../claim-conformance/_blake3.py, written from its specification. It reads
|
||
CPIX documents with the standard library's XML parser. Everything else is here:
|
||
|
||
- the scheme, delivery and decision codes, the deliveries' schemes and the reasons, read
|
||
from the extension's tables (§3.1, §3.2, §3.3, §7.2, §11);
|
||
- the commitment list (§4): its layout, rules, signature and identity, and its derivation from
|
||
a CPIX document (§4.4);
|
||
- the entitlement (§5) and the device's proof over a request (§6.1);
|
||
- the release decision and the licence it permits (§7);
|
||
- the proxy form of the wai.video.keys receipt (§8): its JSON carriage, signing payload,
|
||
identity and own rules;
|
||
- the release ticket (§9.3) and verification of a log (§10).
|
||
|
||
For every vector it requires the outcome, reason and place, and every fact the corpus states;
|
||
of the corpus, that every reason of §11 and every decision of §7.2 appears in it and that no
|
||
other file is there. Then it mutates copies of the corpus, one defect each, and requires every
|
||
copy to fail.
|
||
|
||
python3 key_release_verify.py <live-conformance/keys>
|
||
|
||
Exit status 0 only if every vector verifies and every mutated copy fails. Python 3.8 or later;
|
||
standard library only.
|
||
"""
|
||
|
||
import base64
|
||
import calendar
|
||
import copy
|
||
import datetime
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import struct
|
||
import sys
|
||
import xml.etree.ElementTree as ET
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
||
import _ed25519 as ED # noqa: E402
|
||
import _blake3 as B3 # noqa: E402
|
||
|
||
EXTENSION = os.path.join(HERE, "..", "extensions", "key-release.md")
|
||
|
||
LIST_MAGIC, ENT_MAGIC, TICKET_MAGIC = b"WKL1", b"WKE1", b"WKT1"
|
||
LIST_DOMAIN = b"wai:video-keys-list\x01" # §4.2
|
||
LIST_ID_DOMAIN = b"wai:video-keys-list-id\x01"
|
||
ENT_DOMAIN = b"wai:video-keys-entitlement\x01" # §5.2
|
||
REQUEST_DOMAIN = b"wai:video-keys-request\x01" # §6.1
|
||
RECEIPT_DOMAIN = b"wai:video-keys-proxy-receipt\x01" # §8.3
|
||
ID_DOMAIN = b"wai:video-keys-receipt-id\x01"
|
||
TICKET_DOMAIN = b"wai:video-keys-ticket\x01" # §9.3
|
||
COMMIT_DOMAIN = b"wai:video-keys-commit\x01" # §4.1, form 1's commitment
|
||
CPIX = "urn:dashif:org:cpix" # §4.4
|
||
PSKC = "urn:ietf:params:xml:ns:keyprov:pskc"
|
||
FRESH = 300 # §7.2 check 2
|
||
MAX_KEYS, MAX_LIST, MAX_TICKET = 64, 4096, 300
|
||
U64, U32, U16 = 2 ** 64 - 1, 2 ** 32 - 1, 2 ** 16 - 1
|
||
FLOOR = 288676 # §8.4 rule 10
|
||
CLASS_TAG = {"ModelBased": 2, "Estimator": 3, "OnChipCounter": 4, "CalibratedInstrument": 5} # energy-measurement §2.7
|
||
FILTERING = {"Unknown": 0, "Off": 1, "On": 2}
|
||
KINDS = {"streaming": 1, "offline": 2, "renewal": 3} # §6.1
|
||
WHITE_SPACE = set(map(chr, [0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x20, 0x85, 0xA0, 0x1680] + list(range(0x2000, 0x200B))
|
||
+ [0x2028, 0x2029, 0x202F, 0x205F, 0x3000]))
|
||
|
||
|
||
# ---------------------------------------------------------------- the extension's tables
|
||
|
||
def _text():
|
||
with open(EXTENSION, encoding="utf-8") as f:
|
||
return f.read()
|
||
|
||
|
||
def table(heading):
|
||
"""The body rows of the first table under `heading`, each a list of its cells."""
|
||
text = _text()
|
||
at = text.index("\n" + heading + "\n") + len(heading) + 2
|
||
end = text.find("\n#", at)
|
||
rows = []
|
||
for line in text[at:end if end >= 0 else len(text)].splitlines():
|
||
if line.startswith("|") and not set(line) <= set("|- "):
|
||
rows.append([c.strip() for c in line.strip("|").split("|")])
|
||
return rows[1:]
|
||
|
||
|
||
def code(cell):
|
||
return cell.strip().strip("`")
|
||
|
||
|
||
SCHEMES = {code(r[1]): (int(r[0]), int(r[4])) for r in table("### 3.1 Schemes")} # label: (code, octets)
|
||
DELIVERIES = {code(r[1]): int(r[0]) for r in table("### 3.2 Key deliveries")} # kind: code
|
||
CARRIES = {}
|
||
for _r in table("### 3.3 Which deliveries carry which schemes"):
|
||
_named = re.findall(r"`([^`]+)`", _r[1])
|
||
CARRIES[code(_r[0])] = set(SCHEMES) if _r[1].startswith("all") else {s for s in _named if s in SCHEMES}
|
||
DECISIONS = {code(r[1]): int(r[2]) for r in table("### 7.2 The checks, in order")} # label: code
|
||
REASONS = [code(r[0]) for r in table("## 11. Reasons")]
|
||
SCHEME_OF = {c: label for label, (c, _) in SCHEMES.items()}
|
||
DELIVERY_OF = {c: label for label, c in DELIVERIES.items()}
|
||
|
||
|
||
class Refuse(Exception):
|
||
def __init__(self, reason, at=None):
|
||
super().__init__(reason)
|
||
self.reason, self.at = reason, at
|
||
|
||
|
||
def sha(*parts):
|
||
h = hashlib.sha256()
|
||
for p in parts:
|
||
h.update(p)
|
||
return h.digest()
|
||
|
||
|
||
def b3(*parts):
|
||
return B3.blake3(b"".join(parts))
|
||
|
||
|
||
def be(n, v):
|
||
return v.to_bytes(n, "big")
|
||
|
||
|
||
def string(s):
|
||
b = s.encode("utf-8")
|
||
return be(4, len(b)) + b
|
||
|
||
|
||
def opt(b):
|
||
return b"\x00" if b is None else b"\x01" + b
|
||
|
||
|
||
def commitment(key):
|
||
"""§4.1: form 1's key commitment."""
|
||
return b3(COMMIT_DOMAIN, be(4, len(key)), key)
|
||
|
||
|
||
class Reader:
|
||
def __init__(self, b):
|
||
self.b, self.at = b, 0
|
||
|
||
def take(self, n):
|
||
if self.at + n > len(self.b):
|
||
raise ValueError("short")
|
||
s = self.b[self.at:self.at + n]
|
||
self.at += n
|
||
return s
|
||
|
||
def int(self, n):
|
||
return int.from_bytes(self.take(n), "big")
|
||
|
||
def string(self):
|
||
return self.take(self.int(4)).decode("utf-8")
|
||
|
||
def done(self):
|
||
return self.at == len(self.b)
|
||
|
||
|
||
def period_bytes(p):
|
||
"""§4.1: a period's tag and fields; None is tag 0."""
|
||
if p is None:
|
||
return b"\x00"
|
||
tag = (1 if p["index"] is not None else 0) + (2 if p["start"] is not None else 0)
|
||
out = bytes([tag])
|
||
if p["index"] is not None:
|
||
out += be(4, p["index"])
|
||
if p["start"] is not None:
|
||
out += be(8, p["start"]) + be(8, p["end"])
|
||
return out
|
||
|
||
|
||
def read_period(r):
|
||
tag = r.int(1)
|
||
if tag == 0:
|
||
return None
|
||
if tag > 3:
|
||
raise ValueError("tag")
|
||
index = r.int(4) if tag & 1 else None
|
||
start, end = (r.int(8), r.int(8)) if tag & 2 else (None, None)
|
||
return {"index": index, "start": start, "end": end}
|
||
|
||
|
||
# ---------------------------------------------------------------- §4 the commitment list
|
||
|
||
def read_list(b):
|
||
"""§4.1 and §4.3: the list, or Refuse."""
|
||
try:
|
||
r = Reader(b)
|
||
if r.take(4) != LIST_MAGIC or r.int(1) != 1:
|
||
raise ValueError("magic")
|
||
content, required = r.string(), r.string()
|
||
flag = r.int(1)
|
||
if flag > 1:
|
||
raise ValueError("flag")
|
||
cpix = r.take(32) if flag else None
|
||
issued, n = r.int(8), r.int(4)
|
||
if not 1 <= n <= MAX_LIST or not content or not required:
|
||
raise ValueError("count")
|
||
keys = []
|
||
for _ in range(n):
|
||
kid, scheme, commit = r.take(16), r.int(1), r.take(32)
|
||
p = read_period(r)
|
||
if scheme not in SCHEME_OF or commit == bytes(32) or (keys and keys[-1]["kid"] >= kid):
|
||
raise ValueError("key")
|
||
if p is not None and p["start"] is not None and p["start"] >= p["end"]:
|
||
raise ValueError("window")
|
||
keys.append({"kid": kid, "scheme": SCHEME_OF[scheme], "commit": commit, "period": p})
|
||
pk = r.take(32)
|
||
sid = r.string()
|
||
signed_end = r.at
|
||
sig = r.take(64)
|
||
if not r.done():
|
||
raise ValueError("trailing")
|
||
except (ValueError, UnicodeDecodeError):
|
||
raise Refuse("list_malformed")
|
||
if ED.verify(pk, LIST_DOMAIN + b[:signed_end], sig):
|
||
raise Refuse("list_bad_signature")
|
||
return {"content": content, "required": required, "cpix": cpix, "keys": {k["kid"]: k for k in keys},
|
||
"identity": b3(LIST_ID_DOMAIN, b), "signer": pk, "signer_id": sid}
|
||
|
||
|
||
# ---------------------------------------------------------------- §4.4 a CPIX document
|
||
|
||
UUID = re.compile(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$")
|
||
TIME = re.compile(r"^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})Z$")
|
||
|
||
|
||
def q(name, ns=CPIX):
|
||
return "{%s}%s" % (ns, name)
|
||
|
||
|
||
def uuid(s):
|
||
return bytes.fromhex(s.replace("-", "")) if s is not None and UUID.match(s) else None
|
||
|
||
|
||
def seconds(s):
|
||
m = TIME.match(s or "")
|
||
if not m:
|
||
return None
|
||
try:
|
||
t = calendar.timegm(datetime.datetime(*map(int, m.groups())).timetuple())
|
||
except ValueError:
|
||
return None
|
||
return t if t >= 0 else None
|
||
|
||
|
||
def b64(text):
|
||
s = "".join(c for c in (text or "") if c not in " \t\n\r")
|
||
try:
|
||
v = base64.b64decode(s, validate=True)
|
||
except ValueError:
|
||
return None
|
||
return v if base64.b64encode(v).decode() == s else None
|
||
|
||
|
||
def derive(data):
|
||
"""§4.4 steps 1 to 5: (content_id, digest, keys by KID), or Refuse."""
|
||
# Step 1.
|
||
head = data[:200].decode("utf-8", "replace")
|
||
m = re.match(r"^?<\?xml[^>]*encoding\s*=\s*[\"']([^\"']+)[\"']", head)
|
||
if b"<!DOCTYPE" in data or (m and m.group(1).lower() != "utf-8"):
|
||
raise Refuse("cpix_malformed")
|
||
try:
|
||
data.decode("utf-8")
|
||
root = ET.fromstring(data)
|
||
except (ET.ParseError, UnicodeDecodeError):
|
||
raise Refuse("cpix_malformed")
|
||
if root.tag != q("CPIX"):
|
||
raise Refuse("cpix_malformed")
|
||
digest = sha(data)
|
||
# Step 2.
|
||
content = root.get("contentId")
|
||
if content is None:
|
||
raise Refuse("cpix_unsupported")
|
||
# Step 3.
|
||
keys = {}
|
||
lists = root.findall(q("ContentKeyList"))
|
||
for ck in lists[0].findall(q("ContentKey")) if lists else []:
|
||
kid = uuid(ck.get("kid"))
|
||
if kid is None or kid in keys:
|
||
raise Refuse("cpix_malformed")
|
||
if ck.get("contentId") is not None or ck.get("dependsOnKey") is not None or ck.get("commonEncryptionScheme") not in SCHEMES:
|
||
raise Refuse("cpix_unsupported")
|
||
scheme = ck.get("commonEncryptionScheme")
|
||
data_el = ck.find(q("Data"))
|
||
secret = data_el.find(q("Secret", PSKC)) if data_el is not None else None
|
||
plain = secret.find(q("PlainValue", PSKC)) if secret is not None else None
|
||
if plain is None or secret.find(q("EncryptedValue", PSKC)) is not None:
|
||
raise Refuse("cpix_unsupported")
|
||
key = b64(plain.text)
|
||
if key is None or len(key) != SCHEMES[scheme][1]:
|
||
raise Refuse("cpix_malformed")
|
||
keys[kid] = {"kid": kid, "scheme": scheme, "commit": commitment(key), "period": None}
|
||
if not keys:
|
||
raise Refuse("cpix_malformed")
|
||
# Step 4.
|
||
pl = root.find(q("ContentKeyPeriodList"))
|
||
periods = pl.findall(q("ContentKeyPeriod")) if pl is not None else []
|
||
named = {}
|
||
ul = root.find(q("ContentKeyUsageRuleList"))
|
||
for rule in ul.findall(q("ContentKeyUsageRule")) if ul is not None else []:
|
||
kid = uuid(rule.get("kid"))
|
||
if kid not in keys:
|
||
raise Refuse("cpix_malformed")
|
||
for f in rule.findall(q("KeyPeriodFilter")):
|
||
pid = f.get("periodId")
|
||
hits = [i for i, p in enumerate(periods) if p.get("id") == pid and pid is not None]
|
||
if not hits:
|
||
raise Refuse("cpix_malformed")
|
||
if kid in named and named[kid] != hits[0]:
|
||
raise Refuse("cpix_unsupported")
|
||
named[kid] = hits[0]
|
||
# Step 5.
|
||
for kid in sorted(keys):
|
||
if kid in named:
|
||
keys[kid]["period"] = cpix_period(periods[named[kid]])
|
||
return content, digest, keys
|
||
|
||
|
||
def cpix_period(p):
|
||
if any(p.get(a) is not None for a in ("startOffset", "endOffset", "duration")):
|
||
raise Refuse("cpix_unsupported")
|
||
index = p.get("index")
|
||
if index is not None:
|
||
if not re.match(r"^[0-9]+$", index) or int(index) > U32:
|
||
raise Refuse("cpix_unsupported")
|
||
index = int(index)
|
||
start, end = p.get("start"), p.get("end")
|
||
if (start is None) != (end is None):
|
||
raise Refuse("cpix_unsupported")
|
||
if start is not None:
|
||
start, end = seconds(start), seconds(end)
|
||
if start is None or end is None:
|
||
raise Refuse("cpix_unsupported")
|
||
if start >= end:
|
||
raise Refuse("cpix_malformed")
|
||
if index is None and start is None:
|
||
raise Refuse("cpix_unsupported")
|
||
return {"index": index, "start": start, "end": end}
|
||
|
||
|
||
# ---------------------------------------------------------------- §5 the entitlement, §6 the proof
|
||
|
||
def read_entitlement(b):
|
||
"""§5.1's layout, or None."""
|
||
try:
|
||
r = Reader(b)
|
||
if r.take(4) != ENT_MAGIC or r.int(1) != 1:
|
||
return None
|
||
e = {"session": r.take(16), "holder": r.take(32), "class": r.string(), "nb": r.int(8), "na": r.int(8),
|
||
"offline": r.int(8), "renewals": r.int(4), "authority": r.take(32), "authority_id": r.string()}
|
||
end = r.at
|
||
sig = r.take(64)
|
||
if not r.done():
|
||
return None
|
||
except (ValueError, UnicodeDecodeError):
|
||
return None
|
||
e["holds"] = (bool(e["class"]) and e["nb"] <= e["na"] and (e["offline"] > 0 or e["renewals"] == 0)
|
||
and ED.verifies(e["authority"], ENT_DOMAIN + b[:end], sig))
|
||
return e
|
||
|
||
|
||
def proof_payload(ent, r):
|
||
"""§6.1."""
|
||
d = r["delivery"]
|
||
out = [REQUEST_DOMAIN, sha(ent), string(r["content_id"]), bytes([DELIVERIES[d["kind"]]]), opt(d["system_id"]),
|
||
be(4, len(r["keys"]))]
|
||
out += [k["kid"] for k in r["keys"]]
|
||
out += [bytes([KINDS[r["kind"]]]), opt(r["renews"]), r["request_sha256"], be(8, r["request_t"])]
|
||
return b"".join(out)
|
||
|
||
|
||
# ---------------------------------------------------------------- §8 the receipt
|
||
|
||
MEMBERS = ["form", "content_id", "list", "cpix_sha256", "required_entitlement", "delivery", "keys", "kind", "renews",
|
||
"entitlement", "request_sha256", "request_t", "proof", "issued", "decision", "licence_expiry", "renewal",
|
||
"upstream_status", "response_sha256", "released", "joules_micro", "energy_provenance", "signer_pubkey",
|
||
"signer_id", "sig"]
|
||
HEX = re.compile(r"^(?:[0-9a-f]{2})*$")
|
||
|
||
|
||
def is_int(v, top=U64):
|
||
return isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= top
|
||
|
||
|
||
def hexb(v, n=None, null=False):
|
||
if v is None and null:
|
||
return None
|
||
if not isinstance(v, str) or not HEX.match(v) or (n is not None and len(v) != 2 * n):
|
||
raise ValueError("hex")
|
||
return bytes.fromhex(v)
|
||
|
||
|
||
def nint(v, top=U64):
|
||
if v is None:
|
||
return None
|
||
if not is_int(v, top):
|
||
raise ValueError("int")
|
||
return v
|
||
|
||
|
||
def nstr(v):
|
||
if v is None or isinstance(v, str):
|
||
return v
|
||
raise ValueError("str")
|
||
|
||
|
||
def read_receipt(o):
|
||
"""§8.2: the receipt, or Refuse receipt_malformed."""
|
||
try:
|
||
if not isinstance(o, dict):
|
||
raise ValueError("object")
|
||
label = o.get("energy_provenance")
|
||
want = list(MEMBERS)
|
||
if label == "OnChipCounter":
|
||
want += ["energy_uncertainty", "energy_counter_filtering"]
|
||
elif label == "CalibratedInstrument":
|
||
want += ["energy_uncertainty", "energy_calibration_ref"]
|
||
elif label is not None and label not in CLASS_TAG:
|
||
raise ValueError("class")
|
||
if set(o) != set(want) or o["form"] != 2 or not is_int(o["form"]):
|
||
raise ValueError("members")
|
||
cls = None
|
||
if label is not None:
|
||
cls = {"label": label}
|
||
if label in ("OnChipCounter", "CalibratedInstrument"):
|
||
u = o["energy_uncertainty"]
|
||
if not isinstance(u, dict) or set(u) != {"relative_ppm", "window_us"} or not is_int(u["relative_ppm"], U32) \
|
||
or not is_int(u["window_us"]):
|
||
raise ValueError("uncertainty")
|
||
cls.update(ppm=u["relative_ppm"], window=u["window_us"])
|
||
if label == "OnChipCounter":
|
||
if o["energy_counter_filtering"] not in FILTERING:
|
||
raise ValueError("filtering")
|
||
cls["filtering"] = FILTERING[o["energy_counter_filtering"]]
|
||
if label == "CalibratedInstrument":
|
||
if not isinstance(o["energy_calibration_ref"], str):
|
||
raise ValueError("ref")
|
||
cls["ref"] = o["energy_calibration_ref"]
|
||
d = o["delivery"]
|
||
if not isinstance(d, dict) or set(d) != {"kind", "system_id"} or d["kind"] not in DELIVERIES:
|
||
raise ValueError("delivery")
|
||
keys = []
|
||
if not isinstance(o["keys"], list):
|
||
raise ValueError("keys")
|
||
for k in o["keys"]:
|
||
if not isinstance(k, dict) or set(k) != {"kid", "scheme", "period", "key_commit"}:
|
||
raise ValueError("key")
|
||
if k["scheme"] is not None and k["scheme"] not in SCHEMES:
|
||
raise ValueError("scheme")
|
||
p = k["period"]
|
||
if p is not None:
|
||
if not isinstance(p, dict) or set(p) != {"index", "start", "end"}:
|
||
raise ValueError("period")
|
||
p = {"index": nint(p["index"], U32), "start": nint(p["start"]), "end": nint(p["end"])}
|
||
if (p["start"] is None) != (p["end"] is None) or (p["index"] is None and p["start"] is None):
|
||
raise ValueError("period")
|
||
keys.append({"kid": hexb(k["kid"], 16), "scheme": k["scheme"], "period": p,
|
||
"key_commit": hexb(k["key_commit"], 32, True)})
|
||
if o["kind"] not in KINDS or o["decision"] not in DECISIONS or not isinstance(o["released"], bool):
|
||
raise ValueError("labels")
|
||
for k in ("content_id", "signer_id"):
|
||
if not isinstance(o[k], str):
|
||
raise ValueError(k)
|
||
r = {
|
||
"content_id": o["content_id"], "list": hexb(o["list"], 32, True), "cpix_sha256": hexb(o["cpix_sha256"], 32, True),
|
||
"required": nstr(o["required_entitlement"]),
|
||
"delivery": {"kind": d["kind"], "system_id": hexb(d["system_id"], 16, True)}, "keys": keys, "kind": o["kind"],
|
||
"renews": hexb(o["renews"], 32, True), "entitlement": hexb(o["entitlement"], None, True),
|
||
"request_sha256": hexb(o["request_sha256"], 32), "request_t": nint(o["request_t"]), "proof": hexb(o["proof"], 64, True),
|
||
"issued": nint(o["issued"]), "decision": o["decision"], "licence_expiry": nint(o["licence_expiry"]),
|
||
"renewal": nint(o["renewal"], U32), "upstream_status": nint(o["upstream_status"], U16),
|
||
"response_sha256": hexb(o["response_sha256"], 32, True), "released": o["released"],
|
||
"joules_micro": nint(o["joules_micro"]), "class": cls, "signer_pubkey": hexb(o["signer_pubkey"], 32),
|
||
"signer_id": o["signer_id"], "sig": hexb(o["sig"], 64),
|
||
}
|
||
if r["request_t"] is None or r["issued"] is None or r["joules_micro"] is None:
|
||
raise ValueError("int")
|
||
except (ValueError, KeyError, TypeError):
|
||
raise Refuse("receipt_malformed")
|
||
return r
|
||
|
||
|
||
def class_bytes(c):
|
||
"""energy-measurement §2.7's signed encoding."""
|
||
if c is None:
|
||
return b"\x00"
|
||
out = bytes([CLASS_TAG[c["label"]]])
|
||
if c["label"] == "OnChipCounter":
|
||
out += struct.pack(">IQ", c["ppm"], c["window"]) + bytes([c["filtering"]])
|
||
elif c["label"] == "CalibratedInstrument":
|
||
ref = c["ref"].encode()
|
||
out += struct.pack(">IQ", c["ppm"], c["window"]) + struct.pack(">I", len(ref)) + ref
|
||
return out
|
||
|
||
|
||
def payload(r):
|
||
"""§8.3."""
|
||
d = r["delivery"]
|
||
out = [RECEIPT_DOMAIN, string(r["content_id"]), opt(r["list"]), opt(r["cpix_sha256"]),
|
||
opt(None if r["required"] is None else string(r["required"])),
|
||
bytes([DELIVERIES[d["kind"]]]), opt(d["system_id"]), be(4, len(r["keys"]))]
|
||
for k in r["keys"]:
|
||
out += [k["kid"], bytes([0 if k["scheme"] is None else SCHEMES[k["scheme"]][0]]), period_bytes(k["period"]),
|
||
opt(k["key_commit"])]
|
||
e = r["entitlement"]
|
||
out += [bytes([KINDS[r["kind"]]]), opt(r["renews"]), opt(None if e is None else be(4, len(e)) + e),
|
||
r["request_sha256"], be(8, r["request_t"]), opt(r["proof"]),
|
||
be(8, r["issued"]), bytes([DECISIONS[r["decision"]]]),
|
||
opt(None if r["licence_expiry"] is None else be(8, r["licence_expiry"])),
|
||
opt(None if r["renewal"] is None else be(4, r["renewal"])),
|
||
opt(None if r["upstream_status"] is None else be(2, r["upstream_status"])), opt(r["response_sha256"]),
|
||
bytes([1 if r["released"] else 0]), be(8, r["joules_micro"]), class_bytes(r["class"]),
|
||
r["signer_pubkey"], string(r["signer_id"])]
|
||
return b"".join(out)
|
||
|
||
|
||
def identity(r):
|
||
return b3(ID_DOMAIN, payload(r), r["sig"])
|
||
|
||
|
||
def own_rules(r):
|
||
"""§8.4, rules in order: None, or the refusal."""
|
||
bad = ED.verify(r["signer_pubkey"], payload(r), r["sig"]) # rule 1
|
||
if bad:
|
||
return bad
|
||
keys, kind, d = r["keys"], r["kind"], r["delivery"]
|
||
kids = [k["kid"] for k in keys]
|
||
if not 1 <= len(keys) <= MAX_KEYS or any(a >= b for a, b in zip(kids, kids[1:])): # rule 2
|
||
return "receipt_invalid"
|
||
drm = d["kind"] == "drm"
|
||
if (d["system_id"] is not None) != drm or d["system_id"] == bytes(16) or (d["kind"] == "hls-key-uri" and len(keys) != 1):
|
||
return "receipt_invalid" # rule 3
|
||
if any(k["scheme"] is not None and k["scheme"] not in CARRIES[d["kind"]] for k in keys):
|
||
return "scheme_delivery_mismatch" # rule 4
|
||
if r["list"] is None: # rule 5
|
||
if r["cpix_sha256"] is not None or r["required"] is not None or any(
|
||
k["scheme"] is not None or k["period"] is not None or k["key_commit"] is not None for k in keys):
|
||
return "receipt_invalid"
|
||
elif r["required"] is None:
|
||
return "receipt_invalid"
|
||
if (r["entitlement"] is None) != (r["proof"] is None): # rule 6
|
||
return "receipt_invalid"
|
||
if r["entitlement"] is None and r["decision"] != "DeniedEntitlementInvalid":
|
||
return "receipt_invalid"
|
||
if (r["renews"] is not None) != (kind == "renewal"):
|
||
return "receipt_invalid"
|
||
granted, rel = r["decision"] == "Granted", r["released"]
|
||
if (rel and not granted) or any((k["key_commit"] is not None) != rel for k in keys): # rule 7
|
||
return "receipt_invalid"
|
||
st = r["upstream_status"] # rule 8
|
||
if drm:
|
||
if (st is not None) != granted or (r["response_sha256"] is not None) != granted:
|
||
return "receipt_invalid"
|
||
if rel != (st is not None and 200 <= st <= 299) or (st is not None and not 100 <= st <= 599):
|
||
return "receipt_invalid"
|
||
elif st is not None or r["response_sha256"] is not None or rel != granted:
|
||
return "receipt_invalid"
|
||
licence = rel and kind != "streaming" # rule 9
|
||
if (r["licence_expiry"] is not None) != licence or (r["renewal"] is not None) != licence:
|
||
return "receipt_invalid"
|
||
if licence and (r["licence_expiry"] < r["issued"] or (r["renewal"] == 0) != (kind == "offline")):
|
||
return "receipt_invalid"
|
||
c, j = r["class"], r["joules_micro"] # rule 10
|
||
if c is None:
|
||
return None if j == 0 else "receipt_invalid"
|
||
if j == 0:
|
||
return "receipt_invalid"
|
||
if c["label"] in ("OnChipCounter", "CalibratedInstrument"):
|
||
if c["ppm"] == 0 or c["window"] == 0 or j * c["ppm"] < FLOOR:
|
||
return "receipt_invalid"
|
||
if c["label"] == "CalibratedInstrument" and all(ch in WHITE_SPACE for ch in c["ref"]):
|
||
return "receipt_invalid"
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------- §7 the decision
|
||
|
||
def decide(authorities, r, lst, earlier):
|
||
"""§7.2 and §7.3: (decision, licence_expiry, renewal) of receipt r's request."""
|
||
def denied(d):
|
||
return d, None, None
|
||
eb, proof, issued = r["entitlement"], r["proof"], r["issued"]
|
||
e = read_entitlement(eb) if eb is not None else None
|
||
if (e is None or proof is None or not e["holds"] or e["authority"] not in authorities
|
||
or not ED.verifies(e["holder"], proof_payload(eb, r), proof)):
|
||
return denied("DeniedEntitlementInvalid") # check 1
|
||
if abs(issued - r["request_t"]) > FRESH:
|
||
return denied("DeniedStale") # check 2
|
||
if any(p["proof"] == proof for p in earlier):
|
||
return denied("DeniedReplayed") # check 3
|
||
if lst is None or any(k["kid"] not in lst["keys"] for k in r["keys"]):
|
||
return denied("DeniedUnknownKey") # check 4
|
||
if e["class"] != lst["required"]:
|
||
return denied("DeniedNoEntitlement") # check 5
|
||
if issued < e["nb"]:
|
||
return denied("DeniedNotYetValid") # check 6
|
||
if issued > e["na"]:
|
||
return denied("DeniedExpired") # check 7
|
||
for k in r["keys"]: # check 8
|
||
p = lst["keys"][k["kid"]]["period"]
|
||
if p is not None and p["start"] is not None and (p["start"] > e["na"] or p["end"] <= e["nb"]):
|
||
return denied("DeniedPeriodOutside")
|
||
if r["kind"] != "streaming" and e["offline"] == 0:
|
||
return denied("DeniedOfflineNotEntitled") # check 9
|
||
expiry = min(issued + e["offline"], e["na"])
|
||
if r["kind"] == "streaming":
|
||
return "Granted", None, None
|
||
if r["kind"] == "offline":
|
||
return "Granted", expiry, 0
|
||
kids = [k["kid"] for k in r["keys"]]
|
||
prior = None
|
||
for p in earlier: # check 10
|
||
if p["identity"] != r["renews"]:
|
||
continue
|
||
pe = read_entitlement(p["entitlement"]) if p["entitlement"] is not None else None
|
||
if (p["released"] and p["kind"] != "streaming" and p["content_id"] == r["content_id"]
|
||
and [k["kid"] for k in p["keys"]] == kids and pe is not None
|
||
and pe["session"] == e["session"] and pe["holder"] == e["holder"]):
|
||
prior = p
|
||
break
|
||
if prior is None:
|
||
return denied("DeniedRenewalUnknown")
|
||
if (prior["renewal"] or 0) + 1 > e["renewals"]:
|
||
return denied("DeniedRenewalsExhausted") # check 11
|
||
if prior["licence_expiry"] is None or issued > prior["licence_expiry"]:
|
||
return denied("DeniedRenewalLapsed") # check 12
|
||
return "Granted", expiry, (prior["renewal"] or 0) + 1
|
||
|
||
|
||
# ---------------------------------------------------------------- §9.3 the ticket
|
||
|
||
def read_ticket(b):
|
||
try:
|
||
r = Reader(b)
|
||
if r.take(4) != TICKET_MAGIC or r.int(1) != 1:
|
||
return None
|
||
t = {"receipt": r.take(32), "delivery": r.int(1), "session_type": r.int(1)}
|
||
n = r.int(4)
|
||
if t["delivery"] not in (1, 2) or t["session_type"] not in (1, 2) or not 1 <= n <= MAX_KEYS:
|
||
return None
|
||
t["kids"] = [r.take(16) for _ in range(n)]
|
||
if any(a >= b for a, b in zip(t["kids"], t["kids"][1:])):
|
||
return None
|
||
t["expires"], t["pk"] = r.int(8), r.take(32)
|
||
end = r.at
|
||
sig = r.take(64)
|
||
if not r.done() or not ED.verifies(t["pk"], TICKET_DOMAIN + b[:end], sig):
|
||
return None
|
||
except ValueError:
|
||
return None
|
||
return t
|
||
|
||
|
||
def ticket_follows(t, r):
|
||
return (r["released"] and r["delivery"]["kind"] != "drm" and t["receipt"] == r["identity"]
|
||
and DELIVERY_OF.get(t["delivery"]) == r["delivery"]["kind"]
|
||
and t["session_type"] == (1 if r["kind"] == "streaming" else 2)
|
||
and t["kids"] == [k["kid"] for k in r["keys"]] and t["pk"] == r["signer_pubkey"]
|
||
and 1 <= t["expires"] - r["issued"] <= MAX_TICKET)
|
||
|
||
|
||
# ---------------------------------------------------------------- §10 verification
|
||
|
||
def verify(v, doc, files):
|
||
"""§10.2 over vector v: what the corpus states for it."""
|
||
try:
|
||
return {"outcome": "accept", "receipts": verify_log(v, doc, files)}
|
||
except Refuse as s:
|
||
return {"outcome": "refuse", "reason": s.reason, "at": s.at}
|
||
|
||
|
||
def verify_log(v, doc, files):
|
||
authorities = [bytes.fromhex(a) for a in v["authorities"]]
|
||
proxies = [bytes.fromhex(p) for p in v["proxies"]]
|
||
lists, by_content = [], {}
|
||
for i, name in enumerate(v["lists"]): # step 1
|
||
try:
|
||
lst = read_list(bytes.fromhex(doc["lists"][name]))
|
||
except Refuse as s:
|
||
raise Refuse(s.reason, ["list", i])
|
||
if lst["content"] in by_content:
|
||
raise Refuse("list_malformed", ["list", i])
|
||
by_content[lst["content"]] = lst
|
||
lists.append(lst)
|
||
by_id = {lst["identity"]: lst for lst in lists}
|
||
for j, c in enumerate(v["cpix"]): # step 2
|
||
try:
|
||
content, digest, keys = derive(files[c["file"]])
|
||
except Refuse as s:
|
||
raise Refuse(s.reason, ["cpix", j])
|
||
lst = lists[c["list"]] if c["list"] < len(lists) else None
|
||
if lst is None or lst["content"] != content or lst["cpix"] != digest or lst["keys"] != keys:
|
||
raise Refuse("cpix_mismatch", ["cpix", j])
|
||
log, facts = [], []
|
||
for i, o in enumerate(v["log"]): # step 3
|
||
at = ["receipt", i]
|
||
try:
|
||
r = read_receipt(o)
|
||
except Refuse as s:
|
||
raise Refuse(s.reason, at)
|
||
bad = own_rules(r)
|
||
if bad:
|
||
raise Refuse(bad, at)
|
||
if r["signer_pubkey"] not in proxies:
|
||
raise Refuse("signer_unknown", at)
|
||
r["identity"] = identity(r)
|
||
if any(p["identity"] == r["identity"] for p in log):
|
||
raise Refuse("receipt_duplicate", at)
|
||
lst = None
|
||
if r["list"] is not None:
|
||
lst = by_id.get(r["list"])
|
||
if lst is None:
|
||
raise Refuse("list_unknown", at)
|
||
if lst["content"] != r["content_id"]:
|
||
raise Refuse("list_mismatch", at)
|
||
elif r["content_id"] in by_content:
|
||
raise Refuse("list_mismatch", at)
|
||
if lst is not None:
|
||
if r["cpix_sha256"] != lst["cpix"] or r["required"] != lst["required"]:
|
||
raise Refuse("list_mismatch", at)
|
||
for k in r["keys"]:
|
||
lk = lst["keys"].get(k["kid"])
|
||
if lk is None:
|
||
ok = k["scheme"] is None and k["period"] is None and k["key_commit"] is None
|
||
else:
|
||
ok = (k["scheme"] == lk["scheme"] and k["period"] == lk["period"]
|
||
and k["key_commit"] in (None, lk["commit"]))
|
||
if not ok:
|
||
raise Refuse("list_mismatch", at)
|
||
d, expiry, renewal = decide(authorities, r, lst, log)
|
||
if d != r["decision"] or (r["released"] and r["kind"] != "streaming"
|
||
and (r["licence_expiry"], r["renewal"]) != (expiry, renewal)):
|
||
raise Refuse("decision_mismatch", at)
|
||
facts.append({"identity": r["identity"].hex(), "decision": r["decision"], "released": r["released"],
|
||
"keys": [{"kid": k["kid"].hex(), "key_commit": k["key_commit"].hex()} for k in r["keys"]
|
||
if k["key_commit"] is not None],
|
||
"licence_expiry": r["licence_expiry"], "renewal": r["renewal"]})
|
||
log.append(r)
|
||
for j, h in enumerate(v["tickets"]): # step 4
|
||
t = read_ticket(bytes.fromhex(h))
|
||
r = next((p for p in log if t is not None and p["identity"] == t["receipt"]), None)
|
||
if t is None or r is None or not ticket_follows(t, r):
|
||
raise Refuse("ticket_invalid", ["ticket", j])
|
||
return facts
|
||
|
||
|
||
def check(doc, files, quiet=False):
|
||
"""Every vector against its statement: the failures."""
|
||
fails = []
|
||
for v in doc.get("vectors", []):
|
||
try:
|
||
got = verify(v, doc, files)
|
||
except Exception as e: # a crash on a mutated copy is a failure to read it
|
||
got, bad = None, ["the vector does not verify: %s: %s" % (type(e).__name__, e)]
|
||
else:
|
||
bad = [k for k in got if v.get(k) != got[k]]
|
||
if not quiet:
|
||
what = "?" if got is None else got["outcome"] + (" %s at %s" % (got["reason"], got["at"]) if got["outcome"] == "refuse" else "")
|
||
print(("PASS" if not bad else "FAIL") + " log %s: %s" % (v.get("name"), what))
|
||
fails += ["log %s: %s differs from the corpus" % (v.get("name"), k) for k in bad]
|
||
return fails
|
||
|
||
|
||
def flip(h, at):
|
||
b = bytearray(bytes.fromhex(h))
|
||
b[at] ^= 0x01
|
||
return b.hex()
|
||
|
||
|
||
def mutations(doc, files):
|
||
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
||
vn = {v["name"]: k for k, v in enumerate(doc["vectors"])}
|
||
|
||
def mutate(label, f):
|
||
d, fs = copy.deepcopy(doc), dict(files)
|
||
f(d, d["vectors"], fs)
|
||
return label, d, fs
|
||
|
||
def receipt(name, i):
|
||
return lambda d, vs, fs: vs[vn[name]]["log"][i]
|
||
|
||
def edit_receipt(name, i, key, value):
|
||
return lambda d, vs, fs: vs[vn[name]]["log"][i].update({key: value})
|
||
|
||
def ent_byte(d, vs, fs):
|
||
r = vs[vn["refusals"]]["log"][4]
|
||
r["entitlement"] = flip(r["entitlement"], 60)
|
||
|
||
return [
|
||
mutate("a byte of a release's signature", lambda d, vs, fs: edit_receipt("release_drm", 0, "sig", flip(vs[vn["release_drm"]]["log"][0]["sig"], 3))(d, vs, fs)),
|
||
mutate("a release's issued time moved", lambda d, vs, fs: edit_receipt("release_drm", 1, "issued", vs[vn["release_drm"]]["log"][1]["issued"] + 1)(d, vs, fs)),
|
||
mutate("a stated decision", lambda d, vs, fs: vs[vn["refusals"]]["receipts"][3].update(decision="DeniedExpired")),
|
||
mutate("a stated licence expiry", lambda d, vs, fs: vs[vn["offline"]]["receipts"][0].update(licence_expiry=1)),
|
||
mutate("a stated key commitment",
|
||
lambda d, vs, fs: vs[vn["release_drm"]]["receipts"][0]["keys"][0].update(key_commit="00" * 32)),
|
||
mutate("the granted request removed from before its replay", lambda d, vs, fs: (vs[vn["refusals"]]["log"].pop(4),
|
||
vs[vn["refusals"]]["receipts"].pop(4))),
|
||
mutate("a renewal moved before the licence it renews", lambda d, vs, fs: vs[vn["offline"]]["log"].insert(0, vs[vn["offline"]]["log"].pop(1))),
|
||
mutate("the untrusted authority trusted", lambda d, vs, fs: vs[vn["refusals"]]["authorities"].append(
|
||
d["principals"]["rogue_authority"])),
|
||
mutate("a byte of an entitlement inside a release", ent_byte),
|
||
mutate("a key of the film document changed",
|
||
lambda d, vs, fs: fs.update({"film.cpix.xml": fs["film.cpix.xml"].replace(b"<pskc:PlainValue>", b"<pskc:PlainValue>AAAA", 1)})),
|
||
mutate("the forged list replaced by the genuine one", lambda d, vs, fs: vs[vn["n_list_forged"]].update(lists=["film"])),
|
||
mutate("the stated reason of a negative control", lambda d, vs, fs: vs[vn["n_misreported_grant"]].update(reason="list_mismatch")),
|
||
mutate("the stated place of a negative control", lambda d, vs, fs: vs[vn["n_duplicate"]].update(at=["receipt", 0])),
|
||
mutate("a byte of a ticket's expiry", lambda d, vs, fs: vs[vn["release_hls_key_uri"]].update(
|
||
tickets=[vs[vn["release_hls_key_uri"]]["tickets"][0][:-200] + "ff" + vs[vn["release_hls_key_uri"]]["tickets"][0][-198:]])),
|
||
mutate("a period of the channel list stated otherwise", lambda d, vs, fs: vs[vn["rotation"]]["log"][0]["keys"][0].update(
|
||
period={"index": 2, "start": None, "end": None})),
|
||
mutate("a refusal's scheme stated for a key not in the list", lambda d, vs, fs: next(
|
||
k for k in vs[vn["refusals"]]["log"][6]["keys"] if k["scheme"] is None).update(scheme="cenc")),
|
||
mutate("a CPIX document removed", lambda d, vs, fs: fs.pop("channel.cpix.xml")),
|
||
]
|
||
|
||
|
||
def main(corpus_dir):
|
||
fails = []
|
||
st = ED._self_test()
|
||
print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
|
||
fails += ["ed25519 self-test"] if st else []
|
||
ok = (len(SCHEMES), len(DELIVERIES), len(DECISIONS), len(REASONS)) == (7, 3, 13, 16) and len(CARRIES) == 3
|
||
print(("PASS" if ok else "FAIL") + " the extension's tables read: %d schemes, %d deliveries, %d decisions, %d reasons"
|
||
% (len(SCHEMES), len(DELIVERIES), len(DECISIONS), len(REASONS)))
|
||
fails += [] if ok else ["the extension's tables do not read"]
|
||
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
||
doc = json.load(f)
|
||
named = sorted({c["file"] for v in doc["vectors"] for c in v["cpix"]})
|
||
files = {}
|
||
for name in named:
|
||
with open(os.path.join(corpus_dir, name), "rb") as f:
|
||
files[name] = f.read()
|
||
fails += check(doc, files)
|
||
extra = sorted(set(os.listdir(corpus_dir)) - set(named) - {"vectors.json", "README.md"})
|
||
print(("PASS" if not extra else "FAIL") + " no file but vectors.json, README.md and the CPIX documents the vectors name")
|
||
fails += ["an extra file: %s" % n for n in extra]
|
||
seen = {v.get("reason") for v in doc["vectors"]}
|
||
missing = [r for r in REASONS if r not in seen]
|
||
print(("PASS" if not missing else "FAIL") + " every reason of §11 appears in the corpus")
|
||
fails += ["no case gives %s" % r for r in missing]
|
||
decided = {r["decision"] for v in doc["vectors"] for r in v.get("receipts", [])}
|
||
undecided = [d for d in DECISIONS if d not in decided]
|
||
print(("PASS" if not undecided else "FAIL") + " every decision of §7.2 appears in an accepted log")
|
||
fails += ["no accepted receipt is %s" % d for d in undecided]
|
||
for label, d, fs in mutations(doc, files):
|
||
caught = check(d, fs, quiet=True)
|
||
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:100] if caught else ""))
|
||
if not caught:
|
||
fails.append("a mutated copy verifies: %s" % label)
|
||
for f in fails:
|
||
print("FAIL " + f)
|
||
print(("PASS" if not fails else "FAIL") + " key-release corpus: %d logs, %d CPIX documents" % (len(doc["vectors"]), len(files)))
|
||
return 1 if fails else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) != 2:
|
||
print("usage: key_release_verify.py <live-conformance/keys>", file=sys.stderr)
|
||
sys.exit(2)
|
||
sys.exit(main(sys.argv[1]))
|