86d8d6896d
A WKL1 list of version 2 carries not_before after issued, signed by the key holder; a list without one keeps version 1 and its bytes. §7.2 check 5 refuses a request decided before the release time with its own decision, DeniedNotReleased (code 14), whatever the entitlement grants: after the entitlement's validity, freshness, replay and the keys, before its class, window and allowance. Later checks renumber and decide as before; no earlier code changes. The corpus gains a version-2 list derived from the film's CPIX document, the log release_time and two negative controls; key_release_verify.py reads both versions and decides check 5. Earlier vectors and lists are unchanged. Co-Authored-By: Claude Opus 5.5 <[email protected]>
894 lines
40 KiB
Python
894 lines
40 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 of version 1 or 2 (with the content's release time),
|
||
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:
|
||
raise ValueError("magic")
|
||
version = r.int(1)
|
||
if version not in (1, 2):
|
||
raise ValueError("version")
|
||
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 = r.int(8)
|
||
not_before = r.int(8) if version == 2 else None # §4.1: version 2 only
|
||
n = 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},
|
||
"not_before": not_before, "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 lst["not_before"] is not None and issued < lst["not_before"]:
|
||
return denied("DeniedNotReleased") # check 5
|
||
if e["class"] != lst["required"]:
|
||
return denied("DeniedNoEntitlement") # check 6
|
||
if issued < e["nb"]:
|
||
return denied("DeniedNotYetValid") # check 7
|
||
if issued > e["na"]:
|
||
return denied("DeniedExpired") # check 8
|
||
for k in r["keys"]: # check 9
|
||
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 10
|
||
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 11
|
||
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 12
|
||
if prior["licence_expiry"] is None or issued > prior["licence_expiry"]:
|
||
return denied("DeniedRenewalLapsed") # check 13
|
||
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")),
|
||
mutate("the release time of a list moved a second earlier, its signature kept",
|
||
lambda d, vs, fs: d["lists"].update(film_release=release_moved(d["lists"]["film_release"], -1))),
|
||
mutate("the list with a release time read as version 1",
|
||
lambda d, vs, fs: d["lists"].update(film_release=d["lists"]["film_release"][:8] + "01" + d["lists"]["film_release"][10:])),
|
||
mutate("the release-time log given the list without one", lambda d, vs, fs: vs[vn["release_time"]].update(lists=["film"])),
|
||
mutate("a refusal before the release time stated as an entitlement's",
|
||
lambda d, vs, fs: vs[vn["release_time"]]["receipts"][0].update(decision="DeniedNoEntitlement")),
|
||
mutate("the release-time negative control stated as accepted", lambda d, vs, fs: vs[vn["n_refused_at_release_time"]].update(
|
||
outcome="accept", receipts=vs[vn["release_time"]]["receipts"])),
|
||
]
|
||
|
||
|
||
def release_moved(h, by):
|
||
"""A list of version 2 with its `not_before` moved by `by` seconds and nothing else."""
|
||
b = bytes.fromhex(h)
|
||
r = Reader(b)
|
||
r.take(5)
|
||
r.string()
|
||
r.string()
|
||
r.take(32 if r.int(1) else 0)
|
||
at = r.at + 8
|
||
t = int.from_bytes(b[at:at + 8], "big") + by
|
||
return (b[:at] + t.to_bytes(8, "big") + b[at + 8:]).hex()
|
||
|
||
|
||
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, 14, 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]))
|