aafbf8e1d6
live-conformance/watermark/vectors.json from wai_live_watermark_vectors, and live_watermark_verify.py, written from the extension's text, which agrees on every vector, fails every mutated copy and catches a verifier that leaves out any one of its checks. Co-Authored-By: Claude Opus 5.5 <[email protected]>
793 lines
32 KiB
Python
793 lines
32 KiB
Python
#!/usr/bin/env python3
|
||
"""Independent verifier of the forensic-watermark corpus (extensions/forensic-watermark.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 reads
|
||
descriptors, grants and key-log heads with the live-rooms corpus's own second implementation,
|
||
live_rooms_verify.py (live-rooms §3, §4.3, §5). Everything else is here:
|
||
|
||
- the token (§4): a reader of deterministically encoded CBOR written from RFC 8949 §3 and §4.2.1,
|
||
the COSE_Mac0 and COSE_Sign1 structures of RFC 9052, and the claims of the TS's §5.4, in the
|
||
order of §4.2's steps;
|
||
- the session record (§5), its rules, its signature, who may sign it, its content, its keys,
|
||
its region and retention (§6.1), and its token beside it (§5.7);
|
||
- the recovered sequence (§7.1, §7.2), candidates and agreement (§7.3) and the outcome (§7.4);
|
||
- the disclosure (§8) and its checks (§9.3).
|
||
|
||
For every vector it requires the outcome, the reason, the place and everything the corpus
|
||
states; of the corpus, that every reason of §11 appears in it, which it reads from the
|
||
extension's own table. Then it mutates copies of the corpus, one defect each, and requires
|
||
every copy to fail; and it drops each of its own checks in turn and requires the corpus to
|
||
catch the verifier that lacks it.
|
||
|
||
python3 live_watermark_verify.py <live-conformance/watermark>
|
||
|
||
Exit status 0 only if every case verifies and every mutated copy and every weakened verifier
|
||
fails. Python 3.8 or later; standard library only.
|
||
"""
|
||
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import struct
|
||
import sys
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, HERE)
|
||
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
||
import live_events_verify as LEV # noqa: E402
|
||
import live_rooms_verify as LRV # noqa: E402
|
||
import _ed25519 as ED # noqa: E402
|
||
|
||
Refuse = LEV.Refuse
|
||
EXTENSION = os.path.join(HERE, "..", "extensions", "forensic-watermark.md")
|
||
|
||
SESSION_MAGIC, SEQUENCE_MAGIC, DISCLOSURE_MAGIC = b"WWS1", b"WWQ1", b"WWD1"
|
||
SESSION_DOMAIN = b"wai:live-watermark-session\x01"
|
||
SEQUENCE_DOMAIN = b"wai:live-watermark-sequence\x01"
|
||
DISCLOSURE_DOMAIN = b"wai:live-watermark-disclosure\x01"
|
||
BINDING_DOMAIN = b"wai:live-watermark-binding\x01"
|
||
EDGE_AUTHORITY = "live/watermark"
|
||
WRE1, WRE1_DOMAIN, WRE1_LEN = b"WRE1", b"wai:live-room-e2ee\x01", 105
|
||
MAX_DEPTH, MAX_KIDS = 16, 8
|
||
CWT, COSE = 61, (16, 17, 18, 96, 97, 98)
|
||
MAC0, SIGN1 = 17, 18
|
||
HMAC, ES256 = 5, -7
|
||
EXP, NBF, IAT, ISS, SUB, AUD, CTI = 4, 5, 6, 1, 2, 3, 7
|
||
WMVER, WMVND, WMPATLEN, WMSEGDURATION, WMPATTERN, WMID, WMOPID, WMKEYVER = range(300, 308)
|
||
CLEAR, DECRYPTED, DERIVED = 1, 2, 3
|
||
|
||
# A check named here is left out: the verifier weakened on purpose, to show the corpus catches it.
|
||
SKIP = set()
|
||
|
||
|
||
def sha(*parts):
|
||
h = hashlib.sha256()
|
||
for p in parts:
|
||
h.update(p)
|
||
return h.digest()
|
||
|
||
|
||
def reasons_of_extension():
|
||
"""§11's codes, read from the text."""
|
||
with open(EXTENSION, encoding="utf-8") as f:
|
||
text = f.read()
|
||
at = text.index("\n## 11. Reasons")
|
||
end = text.index("\n## ", at + 1)
|
||
return [line.split("`")[1] for line in text[at:end].splitlines() if line.startswith("| `")]
|
||
|
||
|
||
REASONS = reasons_of_extension()
|
||
|
||
|
||
# ---------------------------------------------------------------- bitmaps (§2)
|
||
|
||
def nbytes(patlen):
|
||
return (patlen + 7) // 8
|
||
|
||
|
||
def bit(m, p):
|
||
return p // 8 < len(m) and (m[p // 8] >> (7 - p % 8)) & 1 == 1
|
||
|
||
|
||
def is_bitmap(m, patlen):
|
||
return len(m) == nbytes(patlen) and not any(bit(m, p) for p in range(patlen, len(m) * 8))
|
||
|
||
|
||
def within(a, b):
|
||
return len(a) == len(b) and all(x & ~y & 0xFF == 0 for x, y in zip(a, b))
|
||
|
||
|
||
def count(m):
|
||
return sum(bin(x).count("1") for x in m)
|
||
|
||
|
||
# ---------------------------------------------------------------- §4.2 step 1: CBOR
|
||
|
||
class Bad(Exception):
|
||
"""A CBOR item that does not read: "malformed" or "unsupported"."""
|
||
|
||
def __init__(self, kind):
|
||
super().__init__(kind)
|
||
self.kind = kind
|
||
|
||
|
||
class Cbor:
|
||
"""RFC 8949 §3, held to §4.2.1's core deterministic encoding, as §4.2 step 1 reads."""
|
||
|
||
def __init__(self, b):
|
||
self.b, self.p = b, 0
|
||
|
||
def take(self, n):
|
||
if len(self.b) - self.p < n:
|
||
raise Bad("malformed")
|
||
s = self.b[self.p:self.p + n]
|
||
self.p += n
|
||
return s
|
||
|
||
def arg(self, ai):
|
||
if ai < 24:
|
||
return ai
|
||
n, least = {24: (1, 24), 25: (2, 256), 26: (4, 65536), 27: (8, 1 << 32)}[ai]
|
||
v = int.from_bytes(self.take(n), "big")
|
||
if v < least:
|
||
raise Bad("malformed")
|
||
return v
|
||
|
||
def item(self, depth):
|
||
if depth > MAX_DEPTH and "depth" not in SKIP:
|
||
raise Bad("unsupported")
|
||
ib = self.take(1)[0]
|
||
major, ai = ib >> 5, ib & 31
|
||
if ai >= 28:
|
||
raise Bad("malformed")
|
||
if major == 7:
|
||
if ai == 20:
|
||
return ("false",)
|
||
if ai == 21:
|
||
return ("true",)
|
||
if ai == 22:
|
||
return ("null",)
|
||
if ai == 24:
|
||
raise Bad("malformed" if self.take(1)[0] < 32 else "unsupported")
|
||
raise Bad("unsupported")
|
||
v = self.arg(ai)
|
||
if major == 0:
|
||
return ("uint", v)
|
||
if major == 1:
|
||
return ("nint", v)
|
||
if major == 2:
|
||
return ("bytes", self.take(v))
|
||
if major == 3:
|
||
try:
|
||
return ("text", self.take(v).decode("utf-8"))
|
||
except UnicodeDecodeError:
|
||
raise Bad("malformed")
|
||
if major == 4:
|
||
return ("array", [self.item(depth + 1) for _ in range(v)])
|
||
if major == 5:
|
||
out, prev = [], None
|
||
for _ in range(v):
|
||
s = self.p
|
||
k = self.item(depth + 1)
|
||
kb = self.b[s:self.p]
|
||
if prev is not None and prev >= kb and "map-order" not in SKIP:
|
||
raise Bad("malformed")
|
||
prev = kb
|
||
out.append((k, self.item(depth + 1)))
|
||
return ("map", out)
|
||
return ("tag", v, self.item(depth + 1))
|
||
|
||
|
||
def read_one(b):
|
||
r = Cbor(b)
|
||
item = r.item(1)
|
||
if r.p != len(b):
|
||
raise Bad("malformed")
|
||
return item
|
||
|
||
|
||
def as_int(i):
|
||
if i[0] == "uint":
|
||
return i[1]
|
||
if i[0] == "nint":
|
||
return -1 - i[1]
|
||
return None
|
||
|
||
|
||
def get(m, k):
|
||
for key, v in m[1]:
|
||
if key == ("uint", k):
|
||
return v
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------- §4.2 the token
|
||
|
||
def token_refuse(kind):
|
||
return Refuse("token_" + kind)
|
||
|
||
|
||
def cbor(b):
|
||
try:
|
||
return read_one(b)
|
||
except Bad as e:
|
||
raise token_refuse(e.kind)
|
||
|
||
|
||
def is_cose_tag(i):
|
||
return i[0] == "tag" and i[1] in COSE
|
||
|
||
|
||
def read_token(b):
|
||
"""§4.2: what the token gives (§4.3), or Refuse."""
|
||
item = cbor(b) # step 1
|
||
cwt_tag = False
|
||
if item[0] == "tag" and item[1] == CWT: # step 2
|
||
if not is_cose_tag(item[2]):
|
||
raise token_refuse("malformed")
|
||
cwt_tag, item = True, item[2]
|
||
if not (item[0] == "tag" and item[1] in (MAC0, SIGN1)): # step 3
|
||
raise token_refuse("unsupported")
|
||
tag, body = item[1], item[2]
|
||
if not (body[0] == "array" and len(body[1]) == 4): # step 4
|
||
raise token_refuse("malformed")
|
||
prot, unprot, payload, mac = body[1]
|
||
if prot[0] != "bytes" or unprot[0] != "map" or payload[0] not in ("bytes", "null") or mac[0] != "bytes":
|
||
raise token_refuse("malformed")
|
||
if payload[0] == "null":
|
||
raise token_refuse("unsupported")
|
||
if prot[1] == b"": # step 5
|
||
pmap = []
|
||
else:
|
||
pm = cbor(prot[1])
|
||
if pm[0] != "map":
|
||
raise token_refuse("malformed")
|
||
pmap = pm[1]
|
||
umap = unprot[1]
|
||
if any(as_int(k) not in (1, 4) for k, _ in pmap + umap):
|
||
raise token_refuse("unsupported")
|
||
if any(k == u for k, _ in pmap for u, _ in umap):
|
||
raise token_refuse("malformed")
|
||
alg = next((v for k, v in pmap if as_int(k) == 1), None)
|
||
if alg is None:
|
||
raise token_refuse("malformed")
|
||
alg = as_int(alg)
|
||
if alg not in (HMAC, ES256):
|
||
raise token_refuse("unsupported")
|
||
if not ((alg == HMAC and tag == MAC0 and len(mac[1]) == 32) or (alg == ES256 and tag == SIGN1 and len(mac[1]) == 64)):
|
||
raise token_refuse("malformed")
|
||
if any(as_int(k) == 4 and v[0] != "bytes" for k, v in pmap + umap):
|
||
raise token_refuse("malformed")
|
||
claims = cbor(payload[1]) # step 6
|
||
if claims[0] == "tag" and (claims[1] == CWT or claims[1] in COSE):
|
||
raise token_refuse("unsupported")
|
||
if claims[0] != "map" or any(k[0] not in ("uint", "nint") for k, _ in claims[1]):
|
||
raise token_refuse("malformed")
|
||
|
||
def u(k):
|
||
v = get(claims, k)
|
||
return v[1] if v is not None and v[0] == "uint" else None
|
||
|
||
def typed(k, kind):
|
||
v = get(claims, k)
|
||
if v is not None and v[0] != kind:
|
||
raise token_refuse("malformed")
|
||
|
||
exp, iat = u(EXP), u(IAT) # step 7.1
|
||
if exp is None or iat is None:
|
||
raise token_refuse("malformed")
|
||
typed(NBF, "uint")
|
||
for k in (ISS, SUB, AUD): # 7.2
|
||
typed(k, "text")
|
||
typed(CTI, "bytes")
|
||
if u(WMVER) is None: # 7.3
|
||
raise token_refuse("malformed")
|
||
if u(WMVER) != 1:
|
||
raise token_refuse("unsupported")
|
||
vendor, patlen = u(WMVND), u(WMPATLEN) # 7.4
|
||
if vendor is None or vendor > 255 or patlen is None or patlen > 65535:
|
||
raise token_refuse("malformed")
|
||
seg = get(claims, WMSEGDURATION) # 7.5
|
||
if seg is not None and not (seg[0] == "array" and len(seg[1]) == 2 and all(x[0] == "uint" for x in seg[1])):
|
||
raise token_refuse("malformed")
|
||
typed(WMID, "text")
|
||
typed(WMOPID, "uint")
|
||
typed(WMKEYVER, "uint")
|
||
pat = get(claims, WMPATTERN) # 7.6
|
||
pattern = None
|
||
if pat is None:
|
||
if get(claims, WMID) is None or get(claims, WMOPID) is None or get(claims, WMKEYVER) is None:
|
||
raise token_refuse("malformed")
|
||
source = DERIVED
|
||
elif pat[0] == "bytes" and len(pat[1]) == nbytes(patlen):
|
||
source, pattern = CLEAR, pat[1]
|
||
elif pat[0] == "array" and encrypted_shape(pat[1]):
|
||
source = DECRYPTED
|
||
else:
|
||
raise token_refuse("malformed")
|
||
nbf = get(claims, NBF)
|
||
return {"cwt_tag": cwt_tag, "alg": alg, "vendor": vendor, "patlen": patlen, "source": source,
|
||
"pattern": pattern.hex() if pattern is not None else None, "iat": iat, "exp": exp,
|
||
"nbf": nbf[1] if nbf is not None else None}
|
||
|
||
|
||
def encrypted_shape(a):
|
||
"""COSE_Encrypt0 (RFC 9052 §5.2) or COSE_Encrypt (§5.1), untagged."""
|
||
if len(a) not in (3, 4):
|
||
return False
|
||
if a[0][0] != "bytes" or a[1][0] != "map" or a[2][0] not in ("bytes", "null"):
|
||
return False
|
||
return len(a) == 3 or (a[3][0] == "array" and len(a[3][1]) > 0)
|
||
|
||
|
||
def good_at(t, u):
|
||
"""§4.3: max(iat, nbf) × 10^6 ≤ u < exp × 10^6."""
|
||
return max(t["iat"], t["nbf"] or 0) * 1000000 <= u < t["exp"] * 1000000
|
||
|
||
|
||
# ---------------------------------------------------------------- §5 the session record
|
||
|
||
def read_head(b, magic):
|
||
if len(b) < 4:
|
||
raise Refuse("truncated")
|
||
if b[:4] != magic:
|
||
raise Refuse("malformed")
|
||
if len(b) < 5:
|
||
raise Refuse("truncated")
|
||
if b[4] != 1:
|
||
raise Refuse("unknown_version")
|
||
|
||
|
||
HEX_KEY = set(b"0123456789abcdef")
|
||
SLOT_CHARS = set(b"abcdefghijklmnopqrstuvwxyz0123456789_")
|
||
|
||
|
||
def content_form_ok(owner, media):
|
||
hex_key = len(owner) == 64 and all(c in HEX_KEY for c in owner)
|
||
slot = owner[:1] == b"_" and 1 <= len(owner) - 1 <= 32 and all(c in SLOT_CHARS for c in owner[1:])
|
||
return (hex_key or slot) and media in (1, 2, 3)
|
||
|
||
|
||
def read_session(b):
|
||
"""§5.5: the record, or Refuse."""
|
||
read_head(b, SESSION_MAGIC)
|
||
r = LRV.Reader(b, 5)
|
||
s = {"room": r.take(32), "edge": r.take(32)}
|
||
s["owner"] = r.take(r.u8())
|
||
s["media"] = r.u8()
|
||
s["kids"] = [r.take(16) for _ in range(r.u8())]
|
||
s["start"], s["end"] = r.u64(), r.u64()
|
||
s["token"], s["vendor"], s["source"], s["patlen"] = r.take(32), r.u8(), r.u8(), r.u16()
|
||
s["served"], s["bits"] = r.take(nbytes(s["patlen"])), r.take(nbytes(s["patlen"]))
|
||
s["segments"], s["unmarked"] = r.u32(), r.u32()
|
||
s["binding"], s["region"], s["expires"] = r.take(32), r.take(2), r.u64()
|
||
s["body"] = b[:r.p]
|
||
s["sig"] = r.take(64)
|
||
if r.p != len(b):
|
||
raise Refuse("malformed")
|
||
kids = s["kids"]
|
||
ok = (content_form_ok(s["owner"], s["media"])
|
||
and len(kids) <= MAX_KIDS and all(a < c for a, c in zip(kids, kids[1:])) and bytes(16) not in kids
|
||
and s["start"] < s["end"] and s["source"] in (1, 2, 3) and s["patlen"] >= 1
|
||
and is_bitmap(s["served"], s["patlen"]) and is_bitmap(s["bits"], s["patlen"])
|
||
and within(s["bits"], s["served"]) and s["segments"] >= count(s["served"])
|
||
and all(65 <= c <= 90 for c in s["region"]))
|
||
if not ok:
|
||
raise Refuse("record_invalid")
|
||
s["digest"] = sha(s["body"])
|
||
return s
|
||
|
||
|
||
def binding(salt, grant, principal):
|
||
return sha(BINDING_DOMAIN, salt, grant, principal)
|
||
|
||
|
||
# ---------------------------------------------------------------- §7 the recovered sequence
|
||
|
||
def read_sequence(b):
|
||
read_head(b, SEQUENCE_MAGIC)
|
||
r = LRV.Reader(b, 5)
|
||
q = {"room": r.take(32)}
|
||
q["owner"] = r.take(r.u8())
|
||
q["media"], q["vendor"], q["patlen"] = r.u8(), r.u8(), r.u16()
|
||
q["start"], q["end"] = r.u64(), r.u64()
|
||
q["known"], q["bits"] = r.take(nbytes(q["patlen"])), r.take(nbytes(q["patlen"]))
|
||
q["copy"], q["claimant"] = r.take(32), r.take(32)
|
||
q["body"] = b[:r.p]
|
||
q["sig"] = r.take(64)
|
||
if r.p != len(b):
|
||
raise Refuse("malformed")
|
||
ok = (content_form_ok(q["owner"], q["media"]) and q["patlen"] >= 1 and q["start"] < q["end"]
|
||
and is_bitmap(q["known"], q["patlen"]) and is_bitmap(q["bits"], q["patlen"])
|
||
and within(q["bits"], q["known"]) and count(q["known"]) > 0)
|
||
if not ok:
|
||
raise Refuse("sequence_invalid")
|
||
q["digest"] = sha(q["body"])
|
||
return q
|
||
|
||
|
||
def candidate(s, q):
|
||
"""§7.3."""
|
||
return (s["owner"] == q["owner"] and s["media"] == q["media"] and s["vendor"] == q["vendor"]
|
||
and s["patlen"] == q["patlen"] and s["start"] < q["end"] and q["start"] < s["end"])
|
||
|
||
|
||
def disagreement(s, q):
|
||
"""§7.3: None when s agrees with q, else (position, how)."""
|
||
for p in range(q["patlen"]):
|
||
if not bit(q["known"], p):
|
||
continue
|
||
if not bit(s["served"], p):
|
||
return (p, "unserved")
|
||
if bit(s["bits"], p) != bit(q["bits"], p) and "compare-bits" not in SKIP:
|
||
return (p, "differs")
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------- §9 verification
|
||
|
||
class Stop(Exception):
|
||
def __init__(self, reason, at):
|
||
super().__init__(reason)
|
||
self.reason, self.at = reason, at
|
||
|
||
|
||
def check_room(room):
|
||
"""§9.1 steps 1 and 2."""
|
||
try:
|
||
d = LRV.read_descriptor(room["descriptor"])
|
||
except Refuse as e:
|
||
raise Stop(e.reason, {"place": "descriptor"})
|
||
try:
|
||
LRV.check_key_log(d, room["key_log"])
|
||
except Refuse as e:
|
||
raise Stop(e.reason, {"place": "key_log"})
|
||
entries = None
|
||
if d["origin"] != b"":
|
||
entries = json.loads(room["key_log"])["keys"]
|
||
held, err = LRV.check_grants(d, room["grants"])
|
||
if err is not None:
|
||
raise Stop(err[1], {"place": "grant", "index": err[0]})
|
||
c = room["confidentiality"]
|
||
if c is not None:
|
||
at = {"place": "confidentiality"}
|
||
if len(c) < 4:
|
||
raise Stop("truncated", at)
|
||
if c[:4] != WRE1:
|
||
raise Stop("malformed", at)
|
||
if len(c) < 5:
|
||
raise Stop("truncated", at)
|
||
if c[4] != 1:
|
||
raise Stop("unknown_version", at)
|
||
if len(c) < WRE1_LEN:
|
||
raise Stop("truncated", at)
|
||
if len(c) > WRE1_LEN:
|
||
raise Stop("malformed", at)
|
||
if c[5:37] != d["room"]:
|
||
raise Stop("wrong_room", at)
|
||
bad = ED.verify(d["authority"], WRE1_DOMAIN + c[:41], c[41:])
|
||
if bad:
|
||
raise Stop(bad, at)
|
||
if "e2ee" not in SKIP:
|
||
raise Stop("e2ee_room", at)
|
||
return {"d": d, "entries": entries, "grants": held}
|
||
|
||
|
||
def check_session(room, record, token, earlier):
|
||
"""§9.1 step 3 for one record: its facts, or Refuse."""
|
||
d = room["d"]
|
||
s = read_session(record) # 1
|
||
if s["room"] != d["room"]: # 2
|
||
raise Refuse("wrong_room")
|
||
bad = ED.verify(s["edge"], SESSION_DOMAIN + s["body"], s["sig"]) # 3
|
||
if bad:
|
||
raise Refuse(bad)
|
||
if "edge" not in SKIP: # 4
|
||
if room["entries"] is None:
|
||
if s["edge"] not in (d["authority"], d["clock"]):
|
||
raise Refuse("edge_not_authorised")
|
||
else:
|
||
e = next((e for e in room["entries"] if e["key"] == s["edge"].hex() and EDGE_AUTHORITY in e["may_sign"]), None)
|
||
if e is None:
|
||
raise Refuse("edge_not_authorised")
|
||
if e["status"] == "revoked":
|
||
raise Refuse("edge_revoked")
|
||
if s["owner"][:1] == b"_" and "content" not in SKIP: # 5
|
||
if not any(x["name"] == s["owner"][1:] and x["media"] == s["media"] for x in d["slots"]):
|
||
raise Refuse("content_unknown")
|
||
if not s["kids"] and "kids" not in SKIP: # 6
|
||
raise Refuse("not_protected")
|
||
if d["regions"] and s["region"] not in d["regions"] and "region" not in SKIP: # 7
|
||
raise Refuse("region_not_pinned")
|
||
end_us = d["t0"] + s["end"]
|
||
if s["expires"] * 1000000 < end_us and "expires" not in SKIP:
|
||
raise Refuse("record_invalid")
|
||
if d["retention"] and s["expires"] * 1000000 > end_us + d["retention"] * 86400 * 1000000 and "retention" not in SKIP:
|
||
raise Refuse("retention_exceeded")
|
||
if s["token"] in earlier and "reuse" not in SKIP: # 8
|
||
raise Refuse("token_reused")
|
||
t = None
|
||
if token is not None: # 9, §5.7
|
||
if sha(token) != s["token"]:
|
||
raise Refuse("token_mismatch")
|
||
t = read_token(token)
|
||
if (t["vendor"], t["patlen"], t["source"]) != (s["vendor"], s["patlen"], s["source"]):
|
||
raise Refuse("token_mismatch")
|
||
if t["pattern"] is not None and "pattern" not in SKIP:
|
||
p = bytes.fromhex(t["pattern"])
|
||
if bytes(a & b for a, b in zip(p, s["served"])) != s["bits"]:
|
||
raise Refuse("token_mismatch")
|
||
if not (good_at(t, d["t0"] + s["start"]) and good_at(t, d["t0"] + s["end"] - 1)) and "window" not in SKIP:
|
||
raise Refuse("outside_token_window")
|
||
s["token_facts"] = t
|
||
return s
|
||
|
||
|
||
def session_facts(s):
|
||
return {"digest": s["digest"].hex(), "edge": s["edge"].hex(), "owner": s["owner"].decode(), "media": s["media"],
|
||
"kids": [k.hex() for k in s["kids"]], "start": s["start"], "end": s["end"], "vendor": s["vendor"],
|
||
"source": s["source"], "patlen": s["patlen"], "served": count(s["served"]), "segments": s["segments"],
|
||
"unmarked": s["unmarked"], "region": s["region"].decode(), "expires": s["expires"],
|
||
"token": s["token_facts"]}
|
||
|
||
|
||
def verify_sessions(room_in, records):
|
||
room = check_room(room_in)
|
||
out = []
|
||
for i, (record, token) in enumerate(records):
|
||
try:
|
||
out.append(check_session(room, record, token, [s["token"] for s in out]))
|
||
except Refuse as e:
|
||
raise Stop(e.reason, {"place": "session", "index": i})
|
||
return room, out
|
||
|
||
|
||
def check_sequence(room, b):
|
||
at = {"place": "sequence"}
|
||
try:
|
||
q = read_sequence(b)
|
||
except Refuse as e:
|
||
raise Stop(e.reason, at)
|
||
if q["room"] != room["d"]["room"]:
|
||
raise Stop("wrong_room", at)
|
||
bad = ED.verify(q["claimant"], SEQUENCE_DOMAIN + q["body"], q["sig"])
|
||
if bad:
|
||
raise Stop(bad, at)
|
||
return q
|
||
|
||
|
||
def verify_match(room_in, records, seq):
|
||
"""§9.2: (outcome, reason, facts)."""
|
||
room, sessions = verify_sessions(room_in, records)
|
||
q = check_sequence(room, seq)
|
||
cands = []
|
||
for i, s in enumerate(sessions):
|
||
if candidate(s, q):
|
||
dis = disagreement(s, q)
|
||
cands.append({"index": i, "digest": s["digest"].hex(), "agrees": dis is None,
|
||
"first": None if dis is None else {"position": dis[0], "how": dis[1]}})
|
||
agreeing = [c["digest"] for c in cands if c["agrees"]]
|
||
if len(agreeing) == 1:
|
||
outcome, reason, matched = "matched", None, agreeing[0]
|
||
else:
|
||
outcome, matched = "unmatched", None
|
||
reason = "no_match" if not agreeing else "ambiguous_match"
|
||
return outcome, reason, {"sequence": q["digest"].hex(), "stated": count(q["known"]), "candidates": cands, "session": matched}
|
||
|
||
|
||
def verify_disclosure(room_in, record, token, seq, disclosure):
|
||
"""§9.3: the facts, or Stop."""
|
||
room, sessions = verify_sessions(room_in, [(record, token)])
|
||
s = sessions[0]
|
||
q = check_sequence(room, seq)
|
||
d = room["d"]
|
||
at = {"place": "disclosure"}
|
||
|
||
def stop(r):
|
||
return Stop(r, at)
|
||
|
||
try:
|
||
read_head(disclosure, DISCLOSURE_MAGIC)
|
||
except Refuse as e:
|
||
raise stop(e.reason)
|
||
if len(disclosure) < 269:
|
||
raise stop("truncated")
|
||
if len(disclosure) > 269:
|
||
raise stop("malformed")
|
||
f = {name: disclosure[5 + 32 * i:37 + 32 * i] for i, name in enumerate(("room", "session", "sequence", "grant", "principal", "salt"))}
|
||
t = struct.unpack("<Q", disclosure[197:205])[0]
|
||
if f["room"] != d["room"]:
|
||
raise stop("wrong_room")
|
||
bad = ED.verify(d["authority"], DISCLOSURE_DOMAIN + disclosure[:205], disclosure[205:])
|
||
if bad:
|
||
raise stop(bad)
|
||
if (f["session"] != s["digest"] or f["sequence"] != q["digest"]) and "names" not in SKIP:
|
||
raise stop("disclosure_mismatch")
|
||
if not candidate(s, q) or disagreement(s, q) is not None:
|
||
raise stop("sequence_disagrees")
|
||
if binding(f["salt"], f["grant"], f["principal"]) != s["binding"] and "opening" not in SKIP:
|
||
raise stop("opening_mismatch")
|
||
g = next((g for g in room["grants"] if g["digest"] == f["grant"]), None)
|
||
if g is None:
|
||
raise stop("grant_unknown")
|
||
if g["principal"] != f["principal"]:
|
||
raise stop("grant_wrong_principal")
|
||
if (g["nbf"] * 1000000 > d["t0"] + s["start"] or d["t0"] + s["end"] > g["exp"] * 1000000) and "grant-window" not in SKIP:
|
||
raise stop("outside_grant_window")
|
||
if t > s["expires"] and "disclosure-expires" not in SKIP:
|
||
raise stop("disclosure_expired")
|
||
return {"digest": sha(disclosure[:205]).hex(), "session": s["digest"].hex(), "sequence": q["digest"].hex(),
|
||
"grant": f["grant"].hex(), "principal": f["principal"].hex(), "stated": count(q["known"]),
|
||
"token_given": token is not None}
|
||
|
||
|
||
# ---------------------------------------------------------------- the corpus
|
||
|
||
def unhex(x):
|
||
return None if x is None else bytes.fromhex(x)
|
||
|
||
|
||
def room_input(doc, v):
|
||
r = dict(doc["rooms"][v["room"]])
|
||
r.update(v.get("room_override", {}))
|
||
return {"descriptor": unhex(r["descriptor"]), "grants": [unhex(g) for g in r["grants"]],
|
||
"key_log": r["key_log"], "confidentiality": unhex(r["confidentiality"])}
|
||
|
||
|
||
def records_of(v):
|
||
return [(unhex(x["record"]), unhex(x["token"])) for x in v["records"]]
|
||
|
||
|
||
def check(doc, quiet=False):
|
||
"""Every vector's outcome against the corpus: a list of failures."""
|
||
fails = []
|
||
|
||
def note(kind, v, got, want):
|
||
if got != want:
|
||
fails.append("%s %s: got %s, the corpus states %s" % (kind, v["name"], got, want))
|
||
if not quiet:
|
||
print("FAIL %s %s" % (kind, v["name"]))
|
||
elif not quiet:
|
||
print("PASS %s %s: %s %s" % (kind, v["name"], want[0], want[1] or ""))
|
||
|
||
for v in doc["tokens"]:
|
||
try:
|
||
got = ("read", None, read_token(unhex(v["token"])))
|
||
except Refuse as e:
|
||
got = ("refuse", e.reason, None)
|
||
note("token", v, got, (v["outcome"], v["reason"], v["facts"]))
|
||
for v in doc["sessions"]:
|
||
try:
|
||
_, ss = verify_sessions(room_input(doc, v), records_of(v))
|
||
got = ("accept", None, None, [session_facts(s) for s in ss])
|
||
except Stop as e:
|
||
got = ("refuse", e.reason, e.at, None)
|
||
note("sessions", v, got, (v["outcome"], v["reason"], v["at"], v["sessions"]))
|
||
for v in doc["matches"]:
|
||
try:
|
||
got = verify_match(room_input(doc, v), records_of(v), unhex(v["sequence"])) + (None,)
|
||
except Stop as e:
|
||
got = ("refuse", e.reason, None, e.at)
|
||
note("match", v, got, (v["outcome"], v["reason"], v["match"], v["at"]))
|
||
for v in doc["disclosures"]:
|
||
rec = v["record"]
|
||
try:
|
||
got = ("accept", None, None, verify_disclosure(room_input(doc, v), unhex(rec["record"]), unhex(rec["token"]),
|
||
unhex(v["sequence"]), unhex(v["disclosure"])))
|
||
except Stop as e:
|
||
got = ("refuse", e.reason, e.at, None)
|
||
note("disclosure", v, got, (v["outcome"], v["reason"], v["at"], v["facts"]))
|
||
return fails
|
||
|
||
|
||
# ---------------------------------------------------------------- mutations
|
||
|
||
def by_name(vs, name):
|
||
return next(v for v in vs if v["name"] == name)
|
||
|
||
|
||
def flip(hexstr, at):
|
||
b = bytearray(bytes.fromhex(hexstr))
|
||
b[at] ^= 0x01
|
||
return b.hex()
|
||
|
||
|
||
def mutations(doc):
|
||
"""(label, mutated copy) for each defect the corpus must catch."""
|
||
out = []
|
||
|
||
def mutate(label, f):
|
||
d = copy.deepcopy(doc)
|
||
f(d)
|
||
out.append((label, d))
|
||
|
||
lane = lambda d: by_name(d["sessions"], "lane_a_sessions") # noqa: E731
|
||
mutate("a byte of a record's signature flipped", lambda d: lane(d)["records"][0].update(record=flip(lane(d)["records"][0]["record"], -1)))
|
||
mutate("a byte of a record's served bitmap flipped",
|
||
lambda d: lane(d)["records"][1].update(record=flip(lane(d)["records"][1]["record"], 165)))
|
||
mutate("two records' tokens swapped", lambda d: (lambda r: (r[0].update(token=r[2]["token"]), r[2].update(token=lane(doc)["records"][0]["token"])))(lane(d)["records"]))
|
||
mutate("a record's stated served count", lambda d: lane(d)["sessions"][0].update(served=63))
|
||
mutate("a record's stated token facts", lambda d: lane(d)["sessions"][2]["token"].update(source=1))
|
||
mutate("a token's stated reason", lambda d: by_name(d["tokens"], "untagged").update(reason="token_malformed"))
|
||
mutate("a token's pattern byte flipped", lambda d: by_name(d["tokens"], "mac0_clear").update(token=flip(by_name(d["tokens"], "mac0_clear")["token"], -40)))
|
||
mutate("the end-to-end encrypted room's confidentiality record removed", lambda d: d["rooms"]["e2ee"].update(confidentiality=None))
|
||
mutate("the pin's stated outcome", lambda d: by_name(d["sessions"], "region_not_pinned").update(outcome="accept", reason=None, at=None))
|
||
mutate("a byte of the matched sequence flipped", lambda d: by_name(d["matches"], "recovered_from_session_1").update(sequence=flip(by_name(d["matches"], "recovered_from_session_1")["sequence"], 120)))
|
||
mutate("the matched session's stated digest", lambda d: by_name(d["matches"], "recovered_from_session_1")["match"].update(session="00" * 32))
|
||
mutate("an ambiguous match stated as matched", lambda d: by_name(d["matches"], "agrees_with_two").update(outcome="matched", reason=None))
|
||
mutate("a candidate's first disagreement", lambda d: by_name(d["matches"], "agrees_with_two")["match"]["candidates"][1]["first"].update(how="differs"))
|
||
mutate("the disclosure's principal stated as another's", lambda d: by_name(d["disclosures"], "disclosure_of_session_1")["facts"].update(principal="00" * 32))
|
||
mutate("a byte of the disclosure's salt flipped", lambda d: by_name(d["disclosures"], "disclosure_of_session_1").update(disclosure=flip(by_name(d["disclosures"], "disclosure_of_session_1")["disclosure"], 170)))
|
||
mutate("the disclosure checked against another session's record", lambda d: by_name(d["disclosures"], "disclosure_without_token").update(record=by_name(d["disclosures"], "sequence_disagrees")["record"]))
|
||
mutate("the expired disclosure stated as accepted", lambda d: by_name(d["disclosures"], "disclosure_expired").update(outcome="accept", reason=None, at=None))
|
||
mutate("a refusal's stated place", lambda d: by_name(d["sessions"], "token_reused")["at"].update(index=0))
|
||
return out
|
||
|
||
|
||
# Each of the verifier's own checks, by its SKIP name: the corpus must catch a verifier without it.
|
||
CHECKS = {
|
||
"depth": "§4.2 step 1: the nesting bound",
|
||
"map-order": "§4.2 step 1: map keys in deterministic order",
|
||
"edge": "§5.6: who may sign",
|
||
"content": "§9.1 step 3.5: the content is the room's",
|
||
"kids": "§3.1: a record names a content key",
|
||
"region": "§6.1: the region pin",
|
||
"expires": "§6.1: expires at or after the session's end",
|
||
"retention": "§6.1: the retention bound",
|
||
"reuse": "§9.1 step 3.8: one token, one session",
|
||
"pattern": "§5.7 step 3: the bits are the clear pattern's",
|
||
"window": "§5.7 step 4: the token's window",
|
||
"e2ee": "§3.2: an end-to-end encrypted room",
|
||
"compare-bits": "§7.3: a served position's variant is compared",
|
||
"names": "§9.3 step 4: the disclosure names the record and the sequence",
|
||
"opening": "§9.3 step 6: the opening opens the binding",
|
||
"grant-window": "§9.3 step 8: the grant's window",
|
||
"disclosure-expires": "§9.3 step 9: before expires",
|
||
}
|
||
|
||
|
||
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 []
|
||
print(("PASS" if len(REASONS) == 20 else "FAIL") + " §11 read from the extension: %d reasons" % len(REASONS))
|
||
fails += [] if len(REASONS) == 20 else ["§11 does not read"]
|
||
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
||
doc = json.load(f)
|
||
fails += check(doc)
|
||
extra = sorted(set(os.listdir(corpus_dir)) - {"vectors.json", "README.md"})
|
||
print(("PASS" if not extra else "FAIL") + " no file but vectors.json and README.md")
|
||
fails += ["an extra file: %s" % n for n in extra]
|
||
seen = {v["reason"] for k in ("tokens", "sessions", "matches", "disclosures") for v in doc[k]}
|
||
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]
|
||
for label, d in mutations(doc):
|
||
caught = check(d, quiet=True)
|
||
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:90] if caught else ""))
|
||
if not caught:
|
||
fails.append("a mutated copy verifies: %s" % label)
|
||
for name, what in CHECKS.items():
|
||
SKIP.clear()
|
||
SKIP.add(name)
|
||
caught = check(doc, quiet=True)
|
||
SKIP.clear()
|
||
print(("PASS" if caught else "FAIL") + " a verifier without %s fails the corpus%s" % (what, " (%s)" % caught[0][:70] if caught else ""))
|
||
if not caught:
|
||
fails.append("the corpus does not catch a verifier without %s" % what)
|
||
for f in fails:
|
||
print("FAIL " + f)
|
||
print(("PASS" if not fails else "FAIL") + " forensic-watermark corpus: %d tokens, %d session lists, %d matches, %d disclosures"
|
||
% (len(doc["tokens"]), len(doc["sessions"]), len(doc["matches"]), len(doc["disclosures"])))
|
||
return 1 if fails else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) != 2:
|
||
print("usage: live_watermark_verify.py <live-conformance/watermark>", file=sys.stderr)
|
||
sys.exit(2)
|
||
sys.exit(main(sys.argv[1]))
|