399ba16b25
C2PA 2.4 §18.16.12.4.3 requires validationResults in a v3 ingredient with an activeManifest, and an independent validator refuses one without it; the writer records the success codes a valid parent gives and the verifier recomputes them. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1691 lines
65 KiB
Python
1691 lines
65 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the live-provenance corpus (extensions/live-provenance.md).
|
|
|
|
Written from the extension's text and the texts it reads (C2PA 2.4 §19 and what it relies on,
|
|
RFC 9052, RFC 9053, RFC 9360, RFC 5280, RFC 8410, jwp-receipts §2 and §3). It shares no code
|
|
with wai-rs, which generates the corpus, and never runs it. It reads, with the standard library
|
|
alone:
|
|
|
|
- CBOR in RFC 8949 §4.2.1's core deterministic encoding;
|
|
- ISO BMFF boxes: a segment's layout (§3.2), its chunks, the initialization segment's `moov`,
|
|
the C2PA Manifest Box and the VSI `emsg` box, and the BMFF hash over whole-box exclusions
|
|
(§4.5);
|
|
- the JUMBF manifest store (§4.1), the claim (§4.2) and its `COSE_Sign1_Tagged` (§4.3);
|
|
- X.509 DER: the claim signer's certificates, their profile (§4.4), their chain to the trust
|
|
anchor and their validity (§11.2);
|
|
- the session keys, their `signerBinding` and rotation (§7), and each segment-info map;
|
|
- the JWP group and object receipts of each group (§8; jwp-receipts §2.1 to §2.4 and §3).
|
|
|
|
Signatures are verified with ../claim-conformance/_ed25519.py, WAI's strict rule written from
|
|
RFC 8032, and BLAKE3 is ../claim-conformance/_blake3.py. The codes a refusal reports are read
|
|
from the extension's §12 table.
|
|
|
|
For every vector it requires the outcome, the reason, the place and the two C2PA codes, and for
|
|
an accepted stream what §11.6 states; of the corpus, that every reason of §12 refuses a vector
|
|
and that no other file is there. Then it mutates copies of the good vectors, one defect each,
|
|
and requires every copy to be refused.
|
|
|
|
python3 c2pa_live_verify.py <live-conformance/c2pa-live>
|
|
|
|
Exit status 0 only if every vector verifies and every mutated copy is refused. Python 3.8 or
|
|
later; standard library only.
|
|
"""
|
|
|
|
import base64
|
|
import calendar
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import struct
|
|
import sys
|
|
import time
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
|
import _blake3 as B3 # noqa: E402
|
|
import _ed25519 as ED # noqa: E402
|
|
|
|
EXTENSION = os.path.join(HERE, "..", "extensions", "live-provenance.md")
|
|
|
|
C2PA_UUID = bytes.fromhex("D8FEC3D61B0E483C92975828877EC481") # Appendix A.5.1
|
|
VSI_SCHEME = "urn:c2pa:verifiable-segment-info" # §19.4.2
|
|
STREAM_DOMAIN = b"wai:live-provenance:stream\x01" # §3.4
|
|
MANIFEST_DOMAIN = b"wai:live-provenance:manifest\x01" # §4.2
|
|
ROLE_INIT, ROLE_SEGMENT, ROLE_ROTATION, ROLE_UPDATE = 0, 1, 2, 3 # §4.2
|
|
L_INGREDIENT, L_REPOSITORY = "c2pa.ingredient.v3", "c2pa.repository-receipt" # §4.7
|
|
DST_PREFIX = "http://cv.iptc.org/newscodes/digitalsourcetype/"
|
|
MODEL_URI = "tag:transaction.science,2026:wai/model/" # timed-text §10.3
|
|
L_HASH, L_ACTIONS, L_LIVE, L_KEYS, L_AI = (
|
|
"c2pa.hash.bmff.v3", "c2pa.actions.v2", "c2pa.livevideo.segment", "c2pa.session-keys", "c2pa.ai-disclosure")
|
|
L_STREAM, L_GROUP, L_TAKE = "science.transaction.wai.stream", "science.transaction.wai.group", "science.transaction.wai.take"
|
|
|
|
|
|
def jtype(prefix):
|
|
"""A JUMBF type UUID: four characters, then 0011-0010-8000-00AA00389B71 (§4.1)."""
|
|
return prefix + bytes.fromhex("00110010800000AA00389B71")
|
|
|
|
|
|
T_STORE, T_MANIFEST, T_UPDATE, T_ASSERTIONS, T_CBOR, T_JSON, T_CLAIM, T_SIG = (
|
|
jtype(b"c2pa"), jtype(b"c2ma"), jtype(b"c2um"), jtype(b"c2as"), jtype(b"cbor"), jtype(b"json"), jtype(b"c2cl"),
|
|
jtype(b"c2cs"))
|
|
|
|
# X.509 object identifiers, as content octets.
|
|
def oid(dotted):
|
|
arcs = [int(x) for x in dotted.split(".")]
|
|
out = bytearray()
|
|
for v in [arcs[0] * 40 + arcs[1]] + arcs[2:]:
|
|
tmp = [v & 0x7F]
|
|
v >>= 7
|
|
while v:
|
|
tmp.append(0x80 | (v & 0x7F))
|
|
v >>= 7
|
|
out += bytes(reversed(tmp))
|
|
return bytes(out)
|
|
|
|
|
|
OID_ED25519 = oid("1.3.101.112")
|
|
OID_BC, OID_KU, OID_EKU, OID_SKI, OID_AKI = (oid("2.5.29.19"), oid("2.5.29.15"), oid("2.5.29.37"),
|
|
oid("2.5.29.14"), oid("2.5.29.35"))
|
|
EKU_CLAIM, EKU_TSA, EKU_OCSP, EKU_ANY = (oid("1.3.6.1.4.1.62558.2.1"), oid("1.3.6.1.5.5.7.3.8"),
|
|
oid("1.3.6.1.5.5.7.3.9"), oid("2.5.29.37.0"))
|
|
ED25519_ALG = b"\x30\x05\x06\x03" + OID_ED25519
|
|
|
|
|
|
class Refuse(Exception):
|
|
def __init__(self, reason, at):
|
|
super().__init__(reason)
|
|
self.reason, self.at = reason, at
|
|
|
|
|
|
def reason_table():
|
|
"""§12: each reason and its two C2PA codes, from the extension's own table."""
|
|
text = open(EXTENSION, encoding="utf-8").read()
|
|
sec = text[text.index("\n## 12. Reasons"):]
|
|
sec = sec[:sec.index("\n## 13.")]
|
|
out = {}
|
|
for line in sec.splitlines():
|
|
if line.startswith("| `"):
|
|
cells = [c.strip().strip("`") for c in line.split("|")]
|
|
out[cells[1]] = (None if cells[3] == "—" else cells[3], None if cells[4] == "—" else cells[4])
|
|
return out
|
|
|
|
|
|
REASONS = reason_table()
|
|
|
|
# ───────────────────────────── CBOR (RFC 8949 §4.2.1) ─────────────────────────────
|
|
|
|
|
|
class Tag:
|
|
def __init__(self, tag, value):
|
|
self.tag, self.value = tag, value
|
|
|
|
def __eq__(self, o):
|
|
return isinstance(o, Tag) and (self.tag, self.value) == (o.tag, o.value)
|
|
|
|
|
|
class CborError(Exception):
|
|
pass
|
|
|
|
|
|
def cbor_read(b):
|
|
"""One data item in deterministic encoding, with nothing after it."""
|
|
pos = [0]
|
|
|
|
def take(n):
|
|
if len(b) - pos[0] < n:
|
|
raise CborError("short")
|
|
s = b[pos[0]:pos[0] + n]
|
|
pos[0] += n
|
|
return s
|
|
|
|
def arg(ai):
|
|
if ai < 24:
|
|
return ai
|
|
if ai > 27:
|
|
raise CborError("indefinite or reserved")
|
|
n = {24: 1, 25: 2, 26: 4, 27: 8}[ai]
|
|
v = int.from_bytes(take(n), "big")
|
|
if v < {1: 24, 2: 256, 4: 65536, 8: 1 << 32}[n]:
|
|
raise CborError("not the shortest head")
|
|
return v
|
|
|
|
def item(depth):
|
|
if depth > 16:
|
|
raise CborError("too deep")
|
|
ib = take(1)[0]
|
|
major, ai = ib >> 5, ib & 31
|
|
if major == 7:
|
|
if ai == 20:
|
|
return False
|
|
if ai == 21:
|
|
return True
|
|
if ai == 22:
|
|
return None
|
|
raise CborError("unsupported simple or float")
|
|
v = arg(ai)
|
|
if major == 0:
|
|
return v
|
|
if major == 1:
|
|
return -1 - v
|
|
if major == 2:
|
|
return bytes(take(v))
|
|
if major == 3:
|
|
try:
|
|
return take(v).decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
raise CborError("text")
|
|
if major == 4:
|
|
return [item(depth + 1) for _ in range(v)]
|
|
if major == 5:
|
|
m, prev = {}, None
|
|
for _ in range(v):
|
|
s = pos[0]
|
|
k = item(depth + 1)
|
|
kb = b[s:pos[0]]
|
|
if prev is not None and not prev < kb:
|
|
raise CborError("keys out of order")
|
|
prev = kb
|
|
if isinstance(k, (list, dict)):
|
|
raise CborError("key type")
|
|
m[k] = item(depth + 1)
|
|
return m
|
|
return Tag(v, item(depth + 1))
|
|
|
|
v = item(1)
|
|
if pos[0] != len(b):
|
|
raise CborError("trailing bytes")
|
|
return v
|
|
|
|
|
|
def head(major, v):
|
|
if v < 24:
|
|
return bytes([major << 5 | v])
|
|
for n, ai in ((1, 24), (2, 25), (4, 26), (8, 27)):
|
|
if v < 1 << (8 * n):
|
|
return bytes([major << 5 | ai]) + v.to_bytes(n, "big")
|
|
|
|
|
|
def sig_structure(protected, payload):
|
|
"""RFC 9052 §4.4: ["Signature1", protected, h'', payload], encoded."""
|
|
return (b"\x84" + head(3, 10) + b"Signature1" + head(2, len(protected)) + protected + head(2, 0)
|
|
+ head(2, len(payload)) + payload)
|
|
|
|
|
|
def is_map(m, required, optional=()):
|
|
return (isinstance(m, dict) and all(isinstance(k, str) and (k in required or k in optional) for k in m)
|
|
and all(k in m for k in required))
|
|
|
|
|
|
def uint(v):
|
|
return isinstance(v, int) and not isinstance(v, bool) and v >= 0
|
|
|
|
|
|
def b32(v):
|
|
return isinstance(v, bytes) and len(v) == 32
|
|
|
|
|
|
# ───────────────────────────── BMFF ─────────────────────────────
|
|
|
|
|
|
def boxes(b, start, end):
|
|
out, p = [], start
|
|
while p < end:
|
|
if end - p < 8:
|
|
return None
|
|
size = struct.unpack(">I", b[p:p + 4])[0]
|
|
if size < 8 or end - p < size:
|
|
return None
|
|
out.append((b[p + 4:p + 8], p, p + size))
|
|
p += size
|
|
return out
|
|
|
|
|
|
def full(b, r):
|
|
"""(version, flags) of a full box."""
|
|
if r[2] - r[1] < 12:
|
|
return None
|
|
v = struct.unpack(">I", b[r[1] + 8:r[1] + 12])[0]
|
|
return v >> 24, v & 0xFFFFFF
|
|
|
|
|
|
def is_c2pa_uuid(b, r):
|
|
return r[0] == b"uuid" and b[r[1] + 8:r[1] + 24] == C2PA_UUID
|
|
|
|
|
|
def manifest_store_bytes(b, r, purpose=b"manifest"):
|
|
"""The store a C2PA uuid box of `purpose` holds (§4.1, §4.7; Appendix A.5.3)."""
|
|
body = b[r[1] + 8:r[2]]
|
|
pre = C2PA_UUID + bytes(4) + purpose + b"\x00" + bytes(8)
|
|
return body[len(pre):] if body.startswith(pre) else None
|
|
|
|
|
|
VSI_MARK = VSI_SCHEME.encode() + b"\x00"
|
|
|
|
|
|
def is_vsi(b, r):
|
|
return r[0] == b"emsg" and b[r[1] + 12:r[1] + 12 + len(VSI_MARK)] == VSI_MARK
|
|
|
|
|
|
def read_emsg(b, r):
|
|
if r[0] != b"emsg" or full(b, r) != (0, 0):
|
|
return None
|
|
body = b[r[1] + 12:r[2]]
|
|
z1 = body.find(b"\x00")
|
|
if z1 < 0:
|
|
return None
|
|
z2 = body.find(b"\x00", z1 + 1)
|
|
if z2 < 0:
|
|
return None
|
|
rest = body[z2 + 1:]
|
|
if len(rest) < 16:
|
|
return None
|
|
try:
|
|
scheme, value = body[:z1].decode(), body[z1 + 1:z2].decode()
|
|
except UnicodeDecodeError:
|
|
return None
|
|
ts, ptd, dur, eid = struct.unpack(">IIII", rest[:16])
|
|
return dict(scheme=scheme, value=value, timescale=ts, ptd=ptd, duration=dur, id=eid, data=rest[16:])
|
|
|
|
|
|
PROFILE_EXCLUSIONS = [
|
|
{"xpath": "/uuid", "data": [{"offset": 8, "value": C2PA_UUID}]},
|
|
{"xpath": "/ftyp"},
|
|
{"xpath": "/mfra"},
|
|
{"xpath": "/emsg", "data": [{"offset": 12, "value": VSI_MARK}]},
|
|
]
|
|
|
|
|
|
def bmff_hash(b, top):
|
|
"""§4.5: SHA-256 over each top-level box no exclusion matches, as its u64 offset, then its bytes."""
|
|
h = hashlib.sha256()
|
|
for ty, s, e in top:
|
|
excluded = False
|
|
for x in PROFILE_EXCLUSIONS:
|
|
if x["xpath"] != "/" + ty.decode("latin-1"):
|
|
continue
|
|
if all(s + d["offset"] + len(d["value"]) <= e and b[s + d["offset"]:s + d["offset"] + len(d["value"])] == d["value"]
|
|
for d in x.get("data", [])):
|
|
excluded = True
|
|
if not excluded:
|
|
h.update(struct.pack(">Q", s))
|
|
h.update(b[s:e])
|
|
return h.digest()
|
|
|
|
|
|
def children(b, r, skip=0):
|
|
if r[2] - r[1] < 8 + skip:
|
|
return None
|
|
return boxes(b, r[1] + 8 + skip, r[2])
|
|
|
|
|
|
def read_chunk(b, moof, mdat):
|
|
"""§3.2's chunk: moof (mfhd, traf: tfhd, tfdt, trun) then mdat; the trun's data at the mdat's payload."""
|
|
if moof[0] != b"moof" or mdat[0] != b"mdat" or mdat[1] != moof[2]:
|
|
return None
|
|
mc = children(b, moof)
|
|
if mc is None or [c[0] for c in mc] != [b"mfhd", b"traf"]:
|
|
return None
|
|
mfhd, traf = mc
|
|
if full(b, mfhd) != (0, 0) or mfhd[2] - mfhd[1] != 16:
|
|
return None
|
|
seq = struct.unpack(">I", b[mfhd[1] + 12:mfhd[1] + 16])[0]
|
|
tc = children(b, traf)
|
|
if tc is None or [c[0] for c in tc] != [b"tfhd", b"tfdt", b"trun"]:
|
|
return None
|
|
tfhd, tfdt, trun = tc
|
|
if full(b, tfhd) != (0, 0x020000) or tfhd[2] - tfhd[1] != 16:
|
|
return None
|
|
track_id = struct.unpack(">I", b[tfhd[1] + 12:tfhd[1] + 16])[0]
|
|
if full(b, tfdt) != (1, 0) or tfdt[2] - tfdt[1] != 20:
|
|
return None
|
|
bmdt = struct.unpack(">Q", b[tfdt[1] + 12:tfdt[1] + 20])[0]
|
|
if full(b, trun) != (0, 0x000701) or trun[2] - trun[1] < 20:
|
|
return None
|
|
count = struct.unpack(">I", b[trun[1] + 12:trun[1] + 16])[0]
|
|
if trun[2] - trun[1] != 20 + 12 * count:
|
|
return None
|
|
offset = struct.unpack(">i", b[trun[1] + 16:trun[1] + 20])[0]
|
|
if offset != moof[2] - moof[1] + 8:
|
|
return None
|
|
dur = size = 0
|
|
for i in range(count):
|
|
d, s, _ = struct.unpack(">III", b[trun[1] + 20 + 12 * i:trun[1] + 32 + 12 * i])
|
|
dur, size = dur + d, size + s
|
|
if size != mdat[2] - mdat[1] - 8:
|
|
return None
|
|
return dict(seq=seq, track_id=track_id, bmdt=bmdt, duration=dur)
|
|
|
|
|
|
def read_moov(b, moov):
|
|
mc = children(b, moov)
|
|
if mc is None:
|
|
return None
|
|
traks = [c for c in mc if c[0] == b"trak"]
|
|
mvex = [c for c in mc if c[0] == b"mvex"]
|
|
if len(traks) != 1 or not mvex:
|
|
return None
|
|
mv = children(b, mvex[0])
|
|
if mv is None or not any(c[0] == b"trex" for c in mv):
|
|
return None
|
|
tc = children(b, traks[0])
|
|
if tc is None:
|
|
return None
|
|
tkhd = [c for c in tc if c[0] == b"tkhd"]
|
|
mdia = [c for c in tc if c[0] == b"mdia"]
|
|
if not tkhd or not mdia:
|
|
return None
|
|
md = children(b, mdia[0])
|
|
mdhd = [c for c in (md or []) if c[0] == b"mdhd"]
|
|
if not mdhd:
|
|
return None
|
|
|
|
def field(r):
|
|
vf = full(b, r)
|
|
if vf is None or vf[0] not in (0, 1):
|
|
return None
|
|
at = 20 if vf[0] == 0 else 28
|
|
if r[2] - r[1] < at + 4:
|
|
return None
|
|
return struct.unpack(">I", b[r[1] + at:r[1] + at + 4])[0]
|
|
|
|
track_id, timescale = field(tkhd[0]), field(mdhd[0])
|
|
if track_id is None or not timescale:
|
|
return None
|
|
return dict(track_id=track_id, timescale=timescale)
|
|
|
|
|
|
# ───────────────────────────── JUMBF (§4.1) ─────────────────────────────
|
|
|
|
|
|
def read_super(b, r):
|
|
if r[0] != b"jumb":
|
|
return None
|
|
inner = boxes(b, r[1] + 8, r[2])
|
|
if not inner or inner[0][0] != b"jumd":
|
|
return None
|
|
d = b[inner[0][1] + 8:inner[0][2]]
|
|
if len(d) < 18 or d[16] != 0x03 or d[-1] != 0:
|
|
return None
|
|
label = d[17:-1]
|
|
if not label or 0 in label:
|
|
return None
|
|
try:
|
|
label = label.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return None
|
|
return dict(type=d[:16], label=label, content=inner[1:])
|
|
|
|
|
|
def one_cbor(b, s, ty=b"cbor"):
|
|
if len(s["content"]) != 1 or s["content"][0][0] != ty:
|
|
return None
|
|
c = s["content"][0]
|
|
return b[c[1] + 8:c[2]]
|
|
|
|
|
|
def read_store(b):
|
|
top = boxes(b, 0, len(b))
|
|
if not top or len(top) != 1:
|
|
return None
|
|
store = read_super(b, top[0])
|
|
if not store or store["type"] != T_STORE or store["label"] != "c2pa" or len(store["content"]) != 1:
|
|
return None
|
|
mbox = store["content"][0]
|
|
man = read_super(b, mbox)
|
|
if not man or man["type"] not in (T_MANIFEST, T_UPDATE) or len(man["content"]) != 3:
|
|
return None
|
|
parts = [read_super(b, r) for r in man["content"]]
|
|
want = [(T_ASSERTIONS, "c2pa.assertions"), (T_CLAIM, "c2pa.claim.v2"), (T_SIG, "c2pa.signature")]
|
|
if any(p is None or (p["type"], p["label"]) != w for p, w in zip(parts, want)):
|
|
return None
|
|
claim, sig = one_cbor(b, parts[1]), one_cbor(b, parts[2])
|
|
if claim is None or sig is None:
|
|
return None
|
|
assertions = []
|
|
for r in parts[0]["content"]:
|
|
a = read_super(b, r)
|
|
if not a or a["type"] not in (T_CBOR, T_JSON):
|
|
return None
|
|
form = "cbor" if a["type"] == T_CBOR else "json"
|
|
c = one_cbor(b, a, form.encode())
|
|
if c is None:
|
|
return None
|
|
assertions.append(dict(label=a["label"], form=form, cbor=c, hash=hashlib.sha256(b[r[1] + 8:r[2]]).digest()))
|
|
sbox = man["content"][2]
|
|
return dict(type=man["type"], id=man["label"], assertions=assertions, claim=claim, signature=sig,
|
|
manifest_hash=hashlib.sha256(b[mbox[1] + 8:mbox[2]]).digest(),
|
|
signature_hash=hashlib.sha256(b[sbox[1] + 8:sbox[2]]).digest())
|
|
|
|
|
|
# ───────────────────────────── COSE (§4.3, §7) ─────────────────────────────
|
|
|
|
|
|
def read_sign1(item):
|
|
"""A COSE_Sign1_Tagged item: protected map with alg EdDSA (-8), unprotected map, payload or nil, 64-byte signature."""
|
|
if not isinstance(item, Tag) or item.tag != 18 or not isinstance(item.value, list) or len(item.value) != 4:
|
|
return None
|
|
p, u, payload, sig = item.value
|
|
if not isinstance(p, bytes) or not isinstance(u, dict) or not isinstance(sig, bytes) or len(sig) != 64:
|
|
return None
|
|
if not (payload is None or isinstance(payload, bytes)):
|
|
return None
|
|
try:
|
|
ph = cbor_read(p)
|
|
except CborError:
|
|
return None
|
|
if not isinstance(ph, dict) or ph.get(1) != -8 or 1 in u:
|
|
return None
|
|
return dict(pbytes=p, protected=ph, unprotected=u, payload=payload, sig=sig)
|
|
|
|
|
|
def cose_verifies(s, key, signed):
|
|
return ED.verify(key, sig_structure(s["pbytes"], signed), s["sig"]) is None
|
|
|
|
|
|
def read_okp(k):
|
|
"""§7.1's OKP COSE_Key: {1: 1, 2: kid, -1: 6, -2: x}, kid equal to x."""
|
|
if not isinstance(k, dict) or len(k) != 4 or set(k) != {1, 2, -1, -2}:
|
|
return None
|
|
if k[1] != 1 or k[-1] != 6 or not b32(k[-2]) or k[2] != k[-2]:
|
|
return None
|
|
return k[-2]
|
|
|
|
|
|
# ───────────────────────────── X.509 (§4.4, §11.2) ─────────────────────────────
|
|
|
|
|
|
def der(b):
|
|
"""(tag, content, whole, rest) of the first element, DER lengths only."""
|
|
if len(b) < 2 or b[0] & 0x1F == 0x1F:
|
|
return None
|
|
l0 = b[1]
|
|
if l0 < 0x80:
|
|
n, hdr = l0, 2
|
|
else:
|
|
k = l0 & 0x7F
|
|
if k == 0 or k > 4 or len(b) < 2 + k or b[2] == 0:
|
|
return None
|
|
n = int.from_bytes(b[2:2 + k], "big")
|
|
if n < 0x80:
|
|
return None
|
|
hdr = 2 + k
|
|
if len(b) < hdr + n:
|
|
return None
|
|
return b[0], b[hdr:hdr + n], b[:hdr + n], b[hdr + n:]
|
|
|
|
|
|
def der_items(b):
|
|
out = []
|
|
while b:
|
|
e = der(b)
|
|
if e is None:
|
|
return None
|
|
out.append(e)
|
|
b = e[3]
|
|
return out
|
|
|
|
|
|
def der_only(b, tag):
|
|
e = der(b)
|
|
if e is None or e[3] or e[0] != tag:
|
|
return None
|
|
return e
|
|
|
|
|
|
def read_time(e):
|
|
try:
|
|
s = e[1].decode("ascii")
|
|
except UnicodeDecodeError:
|
|
return None
|
|
if e[0] == 0x17 and len(s) == 13:
|
|
yy = int(s[:2]) if s[:2].isdigit() else -1
|
|
if yy < 0:
|
|
return None
|
|
y, rest = (1900 + yy if yy >= 50 else 2000 + yy), s[2:]
|
|
elif e[0] == 0x18 and len(s) == 15:
|
|
if not s[:4].isdigit():
|
|
return None
|
|
y, rest = int(s[:4]), s[4:]
|
|
else:
|
|
return None
|
|
if not rest.endswith("Z") or not rest[:-1].isdigit():
|
|
return None
|
|
mo, d, h, mi, se = (int(rest[i:i + 2]) for i in range(0, 10, 2))
|
|
if not (1 <= mo <= 12 and 1 <= d <= 31 and h <= 23 and mi <= 59 and se <= 59):
|
|
return None
|
|
return unix(y, mo, d, h, mi, se)
|
|
|
|
|
|
def unix(y, mo, d, h, mi, se):
|
|
# Days from the civil date, proleptic Gregorian, as calendar.timegm counts them without range limits.
|
|
yy = y - 1 if mo <= 2 else y
|
|
era = yy // 400
|
|
yoe = yy - era * 400
|
|
doy = (153 * (mo - 3 if mo > 2 else mo + 9) + 2) // 5 + d - 1
|
|
doe = yoe * 365 + yoe // 4 - yoe // 100 + doy
|
|
return (era * 146097 + doe - 719468) * 86400 + h * 3600 + mi * 60 + se
|
|
|
|
|
|
def read_cert(b):
|
|
top = der_only(b, 0x30)
|
|
if top is None:
|
|
return None
|
|
parts = der_items(top[1])
|
|
if parts is None or len(parts) != 3:
|
|
return None
|
|
tbs, alg, bits = parts
|
|
if tbs[0] != 0x30 or alg[0] != 0x30 or bits[0] != 0x03:
|
|
return None
|
|
t = der_items(tbs[1])
|
|
if not t:
|
|
return None
|
|
version, i = 0, 0
|
|
if t[0][0] == 0xA0:
|
|
v = der_only(t[0][1], 0x02)
|
|
if v is None:
|
|
return None
|
|
version, i = int.from_bytes(v[1], "big"), 1
|
|
if len(t) < i + 6:
|
|
return None
|
|
serial, tbs_alg, issuer, validity, subject, spki = t[i:i + 6]
|
|
if [x[0] for x in (serial, tbs_alg, issuer, validity, subject, spki)] != [0x02, 0x30, 0x30, 0x30, 0x30, 0x30]:
|
|
return None
|
|
v = der_items(validity[1])
|
|
sp = der_items(spki[1])
|
|
if not v or len(v) != 2 or not sp or len(sp) != 2:
|
|
return None
|
|
if sp[0][2] != ED25519_ALG or sp[1][0] != 0x03 or len(sp[1][1]) != 33 or sp[1][1][0] != 0:
|
|
return None
|
|
unique, exts = False, []
|
|
for e in t[i + 6:]:
|
|
if e[0] in (0x81, 0x82):
|
|
unique = True
|
|
elif e[0] == 0xA3:
|
|
s = der_only(e[1], 0x30)
|
|
xs = der_items(s[1]) if s else None
|
|
if xs is None:
|
|
return None
|
|
for x in xs:
|
|
if x[0] != 0x30:
|
|
return None
|
|
f = der_items(x[1])
|
|
if f and len(f) == 2:
|
|
o, crit, val = f[0], False, f[1]
|
|
elif f and len(f) == 3 and f[1][0] == 0x01 and f[1][1] == b"\xff":
|
|
o, crit, val = f[0], True, f[2]
|
|
else:
|
|
return None
|
|
if o[0] != 0x06 or val[0] != 0x04:
|
|
return None
|
|
exts.append((o[1], crit, val[1]))
|
|
else:
|
|
return None
|
|
if len(bits[1]) != 65 or bits[1][0] != 0:
|
|
return None
|
|
nb, na = read_time(v[0]), read_time(v[1])
|
|
if nb is None or na is None:
|
|
return None
|
|
return dict(tbs=tbs[2], version=version, issuer=issuer[2], subject=subject[2], nb=nb, na=na,
|
|
key=sp[1][1][1:], unique=unique, ed=tbs_alg[2] == ED25519_ALG and alg[2] == ED25519_ALG,
|
|
exts=exts, sig=bits[1][1:])
|
|
|
|
|
|
def ext(c, o):
|
|
for x in c["exts"]:
|
|
if x[0] == o:
|
|
return x
|
|
return None
|
|
|
|
|
|
def ski(c):
|
|
x = ext(c, OID_SKI)
|
|
e = der_only(x[2], 0x04) if x else None
|
|
return e[1] if e else None
|
|
|
|
|
|
def aki(c):
|
|
x = ext(c, OID_AKI)
|
|
s = der_only(x[2], 0x30) if x else None
|
|
f = der_items(s[1]) if s else None
|
|
if f is None:
|
|
return None
|
|
for e in f:
|
|
if e[0] == 0x80:
|
|
return e[1]
|
|
return None
|
|
|
|
|
|
def profile_ok(c, ca):
|
|
"""§4.4 and C2PA 2.4 §14.5.1.1, for the end entity (ca False) or an intermediate."""
|
|
known = {OID_BC, OID_KU, OID_EKU, OID_SKI, OID_AKI}
|
|
seen = set()
|
|
for o, crit, _ in c["exts"]:
|
|
if o in seen or (crit and o not in known):
|
|
return False
|
|
seen.add(o)
|
|
if c["version"] != 2 or not c["ed"] or c["unique"] or aki(c) is None:
|
|
return False
|
|
bc = ext(c, OID_BC)
|
|
is_ca = False
|
|
if bc:
|
|
s = der_only(bc[2], 0x30)
|
|
if s is None:
|
|
return False
|
|
f = der_items(s[1])
|
|
if f is None:
|
|
return False
|
|
is_ca = bool(f) and f[0][0] == 0x01 and f[0][1] == b"\xff"
|
|
ku = ext(c, OID_KU)
|
|
kb = der_only(ku[2], 0x03) if ku else None
|
|
if kb is None or not kb[1] or len(kb[1]) > 3 or kb[1][0] > 7:
|
|
return False
|
|
first = kb[1][1] if len(kb[1]) > 1 else 0
|
|
digital_signature, key_cert_sign = bool(first & 0x80), bool(first & 0x04)
|
|
if ca:
|
|
return is_ca and key_cert_sign and ski(c) is not None
|
|
if is_ca or not digital_signature or key_cert_sign:
|
|
return False
|
|
e = ext(c, OID_EKU)
|
|
s = der_only(e[2], 0x30) if e else None
|
|
f = der_items(s[1]) if s else None
|
|
if f is None or any(x[0] != 0x06 for x in f):
|
|
return False
|
|
ekus = {x[1] for x in f}
|
|
return bool(ekus) and EKU_ANY not in ekus and EKU_CLAIM in ekus and EKU_TSA not in ekus and EKU_OCSP not in ekus
|
|
|
|
|
|
def issued_by(c, issuer):
|
|
a = aki(c)
|
|
s = ski(issuer)
|
|
return (c["issuer"] == issuer["subject"] and a is not None and (s is None or s == a)
|
|
and ED.verify(issuer["key"], c["tbs"], c["sig"]) is None)
|
|
|
|
|
|
# ───────────────────────────── receipts (jwp-receipts §2, §3) ─────────────────────────────
|
|
|
|
|
|
def blake3(*parts):
|
|
return B3.blake3(b"".join(parts))
|
|
|
|
|
|
GROUP_FIELDS = ["kind", "track_alias", "group_id", "root_hash", "leaf_count", "joules_total", "group_start",
|
|
"group_end", "signer_pubkey", "signer_id", "sig", "parent_receipt_hash"]
|
|
|
|
|
|
def group_payload(g):
|
|
"""jwp-receipts §2.3's legacy signing payload."""
|
|
p = b"moq-jwp:group-sign\x01" + struct.pack(">QQ", g["track_alias"], g["group_id"]) + bytes.fromhex(g["root_hash"])
|
|
p += struct.pack(">QQ", g["leaf_count"], g["joules_total"])
|
|
for s in (g["group_start"], g["group_end"]):
|
|
s = s.encode()
|
|
p += struct.pack(">I", len(s)) + s
|
|
sid = g["signer_id"].encode()
|
|
return p + bytes.fromhex(g["signer_pubkey"]) + struct.pack(">I", len(sid)) + sid
|
|
|
|
|
|
def link_hash(g):
|
|
return blake3(b"moq-jwp:receipt-link\x01", group_payload(g), bytes.fromhex(g["sig"]))
|
|
|
|
|
|
def group_ok(g):
|
|
"""§3 steps 1 and 2 for a group receipt of the legacy form, which this profile uses."""
|
|
if set(g) != set(GROUP_FIELDS) or g["kind"] != "group":
|
|
return False
|
|
return ED.verify(bytes.fromhex(g["signer_pubkey"]), group_payload(g), bytes.fromhex(g["sig"])) is None
|
|
|
|
|
|
def leaf(content):
|
|
return blake3(b"moq-jwp:merkle-leaf\x01", content)
|
|
|
|
|
|
def node(left, right):
|
|
return blake3(b"moq-jwp:merkle-node\x01", left, right)
|
|
|
|
|
|
def merkle_root(leaves):
|
|
if not leaves:
|
|
return bytes(32)
|
|
level = list(leaves)
|
|
while len(level) > 1:
|
|
if len(level) % 2:
|
|
level.append(level[-1])
|
|
level = [node(level[i], level[i + 1]) for i in range(0, len(level), 2)]
|
|
return level[0]
|
|
|
|
|
|
def object_accepts(o, g, chunk):
|
|
"""jwp-receipts §3 steps 1 to 5: the group verifies, the chunk hashes to the content hash, the
|
|
object stands at its own position, and its leaf walks to the root."""
|
|
if not group_ok(g) or o.get("kind") != "object" or "leaf_version" in o:
|
|
return False
|
|
if bytes.fromhex(o["content_hash"]) != blake3(b"moq-jwp:object\x01", chunk):
|
|
return False
|
|
p = o["merkle_proof"]
|
|
n = g["leaf_count"]
|
|
depth = 0 if n <= 1 else (n - 1).bit_length()
|
|
if (o["track_alias"], o["group_id"]) != (g["track_alias"], g["group_id"]) or o["object_id"] != p["leaf_index"]:
|
|
return False
|
|
if not (p["leaf_index"] < n and len(p["siblings"]) == depth and p["direction_bits"] == p["leaf_index"]):
|
|
return False
|
|
cur = leaf(bytes.fromhex(o["content_hash"]))
|
|
for i, s in enumerate(p["siblings"]):
|
|
s = bytes.fromhex(s)
|
|
cur = node(s, cur) if (p["direction_bits"] >> i) & 1 else node(cur, s)
|
|
return cur == bytes.fromhex(g["root_hash"])
|
|
|
|
|
|
# ───────────────────────────── the assertions ─────────────────────────────
|
|
|
|
|
|
def hexs(b):
|
|
return b.hex()
|
|
|
|
|
|
def uuid_text(b):
|
|
h = b.hex()
|
|
return "%s-%s-%s-%s-%s" % (h[:8], h[8:12], h[12:16], h[16:20], h[20:])
|
|
|
|
|
|
def uuid_v8(b16):
|
|
b = bytearray(b16)
|
|
b[6] = (b[6] & 0x0F) | 0x80
|
|
b[8] = (b[8] & 0x3F) | 0x80
|
|
return uuid_text(bytes(b))
|
|
|
|
|
|
def manifest_uuid(s):
|
|
if not isinstance(s, str) or not s.startswith("urn:c2pa:"):
|
|
return None
|
|
u = s[len("urn:c2pa:"):]
|
|
if not re.fullmatch(r"[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}", u):
|
|
return None
|
|
return u
|
|
|
|
|
|
def track_kind(track, room):
|
|
"""The media kind of `<room>-<owner>-<kind>--<layer>` (live-rooms §6.1) whose room is `room`."""
|
|
if not isinstance(track, str) or "--" not in track:
|
|
return None
|
|
ns, layer = track.split("--", 1)
|
|
if not re.fullmatch(r"0|[1-9][0-9]*", layer):
|
|
return None
|
|
f = ns.split("-")
|
|
if len(f) != 3 or f[0] != room.hex():
|
|
return None
|
|
if not (re.fullmatch(r"[0-9a-f]{64}", f[1]) or re.fullmatch(r"_[a-z0-9_]+", f[1])):
|
|
return None
|
|
return f[2] if f[2] in ("audio", "video", "screen") else None
|
|
|
|
|
|
def read_origin(m):
|
|
o = m.get("origin")
|
|
if not uint(o):
|
|
return None
|
|
model = m.get("model")
|
|
if "model" in m and not b32(model):
|
|
return None
|
|
if o == 1 and model is None:
|
|
return (1, None)
|
|
if o in (2, 3) and model is not None:
|
|
return (o, model)
|
|
return None
|
|
|
|
|
|
def read_label(m):
|
|
if is_map(m, ["inputs"]):
|
|
v = m["inputs"]
|
|
if not isinstance(v, list) or not v:
|
|
return None
|
|
out = []
|
|
for i in v:
|
|
if not is_map(i, ["streamId", "track", "origin"], ["model"]):
|
|
return None
|
|
om = read_origin(i)
|
|
if om is None or not isinstance(i["streamId"], str) or not isinstance(i["track"], str):
|
|
return None
|
|
out.append(dict(streamId=i["streamId"], track=i["track"], origin=om[0], model=om[1]))
|
|
return ("inputs", out)
|
|
if not is_map(m, ["origin"], ["model"]):
|
|
return None
|
|
om = read_origin(m)
|
|
return None if om is None else ("own", om)
|
|
|
|
|
|
def read_stream(m):
|
|
"""§5.1's stream assertion."""
|
|
if not is_map(m, ["room", "track", "trackAlias", "epoch", "firstGroup", "label"]):
|
|
return None
|
|
if not (b32(m["room"]) and isinstance(m["track"], str) and uint(m["trackAlias"]) and uint(m["epoch"])
|
|
and uint(m["firstGroup"])):
|
|
return None
|
|
label = read_label(m["label"])
|
|
if label is None or m["trackAlias"] >= 1 << 53 or track_kind(m["track"], m["room"]) is None:
|
|
return None
|
|
if label[0] == "inputs" and any(track_kind(i["track"], m["room"]) is None for i in label[1]):
|
|
return None
|
|
return dict(m, label=label)
|
|
|
|
|
|
def stream_id_of(sa):
|
|
"""§3.4."""
|
|
t = sa["track"].encode()
|
|
h = hashlib.sha256(STREAM_DOMAIN + sa["room"] + struct.pack(">I", len(t)) + t + struct.pack(">Q", sa["epoch"])).digest()
|
|
return uuid_v8(h[:16])
|
|
|
|
|
|
def read_take(m, init):
|
|
"""§5.3's take assertion: epoch and no commit in the initialization manifest, commit and no epoch later."""
|
|
need = ["take", "room", "device", "epoch"] if init else ["take", "room", "device", "commit"]
|
|
if not is_map(m, need):
|
|
return None
|
|
if not all(b32(m[k]) for k in ("take", "room", "device")):
|
|
return None
|
|
if init and not uint(m["epoch"]):
|
|
return None
|
|
if not init and not b32(m["commit"]):
|
|
return None
|
|
return m
|
|
|
|
|
|
def disclosure(kind, ident):
|
|
"""§9.1 and §9.2: the IPTC term and the AI disclosures (model, fully autonomous)."""
|
|
if kind == "take":
|
|
return "digitalCapture", []
|
|
|
|
def models(pairs):
|
|
out = []
|
|
for o, m in pairs:
|
|
if m is None:
|
|
continue
|
|
for x in out:
|
|
if x[0] == m:
|
|
x[1] = x[1] or o == 2
|
|
break
|
|
else:
|
|
out.append([m, o == 2])
|
|
return [tuple(x) for x in out]
|
|
|
|
label = ident["label"]
|
|
if label[0] == "own":
|
|
o, m = label[1]
|
|
k = track_kind(ident["track"], ident["room"])
|
|
term = ({1: "screenCapture" if k == "screen" else "digitalCapture", 2: "trainedAlgorithmicMedia"}
|
|
.get(o, "compositeWithTrainedAlgorithmicMedia"))
|
|
return term, models([(o, m)])
|
|
inputs = label[1]
|
|
if all(i["origin"] == 1 and track_kind(i["track"], ident["room"]) in ("audio", "video") for i in inputs):
|
|
term = "compositeCapture"
|
|
elif any(i["origin"] != 1 for i in inputs):
|
|
term = "compositeSynthetic"
|
|
else:
|
|
term = "composite"
|
|
return term, models([(i["origin"], i["model"]) for i in inputs])
|
|
|
|
|
|
def ai_label(i):
|
|
return L_AI if i == 0 else "%s__%d" % (L_AI, i)
|
|
|
|
|
|
def read_actions(m):
|
|
if not is_map(m, ["actions"]) or not isinstance(m["actions"], list) or len(m["actions"]) != 1:
|
|
return None
|
|
a = m["actions"][0]
|
|
if not is_map(a, ["action", "digitalSourceType"]) or a["action"] != "c2pa.created":
|
|
return None
|
|
d = a["digitalSourceType"]
|
|
if not isinstance(d, str) or not d.startswith(DST_PREFIX):
|
|
return None
|
|
return d[len(DST_PREFIX):]
|
|
|
|
|
|
def read_ai(m):
|
|
if not is_map(m, ["modelType", "modelIdentifier", "contentProfile"]) or m["modelType"] != "c2pa.types.model":
|
|
return None
|
|
cp = m["contentProfile"]
|
|
if not is_map(cp, ["humanOversightLevel"]):
|
|
return None
|
|
if not isinstance(m["modelIdentifier"], str) or not isinstance(cp["humanOversightLevel"], str):
|
|
return None
|
|
return (m["modelIdentifier"], cp["humanOversightLevel"])
|
|
|
|
|
|
def read_hash_item(m, segment):
|
|
keys = ["exclusions", "alg", "hash"] + (["sequenceNumber"] if segment else [])
|
|
if not is_map(m, keys) or m["alg"] != "sha256" or m["exclusions"] != PROFILE_EXCLUSIONS or not b32(m["hash"]):
|
|
return None
|
|
if segment and not uint(m["sequenceNumber"]):
|
|
return None
|
|
return m["hash"], (m["sequenceNumber"] if segment else None)
|
|
|
|
|
|
def read_live(m):
|
|
if not is_map(m, ["sequenceNumber", "streamId"], ["continuityMethod", "previousManifestId"]):
|
|
return None
|
|
if not uint(m["sequenceNumber"]) or not isinstance(m["streamId"], str):
|
|
return None
|
|
for k in ("continuityMethod", "previousManifestId"):
|
|
if k in m and not isinstance(m[k], str):
|
|
return None
|
|
return m
|
|
|
|
|
|
def read_group(m):
|
|
if not is_map(m, ["groupId", "receipt", "root", "objects"]):
|
|
return None
|
|
if not (uint(m["groupId"]) and b32(m["receipt"]) and b32(m["root"]) and uint(m["objects"])):
|
|
return None
|
|
return m
|
|
|
|
|
|
def shape_of(labels, init, kind):
|
|
"""§11.2 step 7: which list a manifest is held to."""
|
|
if len(labels) < 3 or labels[0] != L_HASH or labels[1] != L_ACTIONS:
|
|
return None
|
|
rest = labels[2:]
|
|
if init:
|
|
shape = "init"
|
|
if rest[0] == L_KEYS:
|
|
rest = rest[1:]
|
|
elif rest[0] == L_LIVE:
|
|
shape, rest = "per-segment", rest[1:]
|
|
elif rest[0] == L_KEYS:
|
|
shape, rest = "rotation", rest[1:]
|
|
else:
|
|
return None
|
|
if not rest:
|
|
return None
|
|
ai, last = rest[:-1], rest[-1]
|
|
if any(l != ai_label(i) for i, l in enumerate(ai)):
|
|
return None
|
|
want = L_GROUP if (shape == "per-segment" and kind == "track") else (L_TAKE if kind == "take" else L_STREAM)
|
|
return shape if last == want else None
|
|
|
|
|
|
# ───────────────────────────── §11.2: a manifest ─────────────────────────────
|
|
|
|
|
|
def verify_claim(store, anchors, t_us):
|
|
"""§11.2 steps 1 (the claim and its signature's form) to 6. The labels in the claim's order
|
|
and the end-entity certificate; raises the reason as a string."""
|
|
mid = store["id"]
|
|
u = manifest_uuid(mid)
|
|
if u is None:
|
|
raise ValueError("manifest_malformed")
|
|
try:
|
|
claim = cbor_read(store["claim"])
|
|
except CborError:
|
|
raise ValueError("manifest_malformed")
|
|
if (not is_map(claim, ["instanceID", "claim_generator_info", "signature", "created_assertions", "alg"])
|
|
or claim["instanceID"] != "xmp:iid:" + u or claim["signature"] != "self#jumbf=/c2pa/%s/c2pa.signature" % mid
|
|
or claim["alg"] != "sha256"):
|
|
raise ValueError("manifest_malformed")
|
|
cgi = claim["claim_generator_info"]
|
|
if not is_map(cgi, ["name", "specVersion"]) or not isinstance(cgi["name"], str) or not isinstance(cgi["specVersion"], str):
|
|
raise ValueError("manifest_malformed")
|
|
refs = claim["created_assertions"]
|
|
if not isinstance(refs, list):
|
|
raise ValueError("manifest_malformed")
|
|
try:
|
|
s = read_sign1(cbor_read(store["signature"]))
|
|
except CborError:
|
|
s = None
|
|
if s is None or s["payload"] is not None or s["unprotected"] != {} or len(s["protected"]) != 2:
|
|
raise ValueError("manifest_malformed")
|
|
x5 = s["protected"].get(33)
|
|
if isinstance(x5, list) and x5 and all(isinstance(c, bytes) for c in x5):
|
|
chain_der = x5
|
|
elif isinstance(x5, bytes):
|
|
chain_der = [x5]
|
|
else:
|
|
raise ValueError("manifest_malformed")
|
|
chain = [read_cert(d) for d in chain_der]
|
|
if any(c is None for c in chain) or not profile_ok(chain[0], False) or not all(profile_ok(c, True) for c in chain[1:]):
|
|
raise ValueError("credential_invalid")
|
|
if any(not issued_by(chain[i], chain[i + 1]) for i in range(len(chain) - 1)):
|
|
raise ValueError("credential_untrusted")
|
|
anchor = next((a for a in anchors if a is not None and issued_by(chain[-1], a)), None)
|
|
if anchor is None:
|
|
raise ValueError("credential_untrusted")
|
|
if not cose_verifies(s, chain[0]["key"], store["claim"]):
|
|
raise ValueError("claim_signature_mismatch")
|
|
if not all(c["nb"] * 1000000 <= t_us <= c["na"] * 1000000 for c in chain + [anchor]):
|
|
raise ValueError("credential_expired")
|
|
prefix = "self#jumbf=/c2pa/%s/c2pa.assertions/" % mid
|
|
labels = []
|
|
for i, rf in enumerate(refs):
|
|
if not is_map(rf, ["url", "hash"]) or not isinstance(rf["url"], str) or not rf["url"].startswith(prefix) or not b32(rf["hash"]):
|
|
raise ValueError("manifest_malformed")
|
|
label = rf["url"][len(prefix):]
|
|
held = [a for a in store["assertions"] if a["label"] == label]
|
|
if len(held) != 1 or i >= len(store["assertions"]) or store["assertions"][i]["label"] != label:
|
|
raise ValueError("manifest_malformed")
|
|
if held[0]["hash"] != rf["hash"]:
|
|
raise ValueError("assertion_hash_mismatch")
|
|
labels.append(label)
|
|
if len(labels) != len(store["assertions"]):
|
|
raise ValueError("manifest_malformed")
|
|
return labels, chain_der[0]
|
|
|
|
|
|
def verify_manifest(asset, top, store_bytes, init, kind, anchors, t_us):
|
|
"""§11.2 for a standard manifest. Raises the reason as a string; returns the manifest's
|
|
identifier, shape, assertions, end-entity cert and store."""
|
|
store = read_store(store_bytes or b"")
|
|
if store is None or store["type"] != T_MANIFEST or any(a["form"] != "cbor" for a in store["assertions"]):
|
|
raise ValueError("manifest_malformed")
|
|
labels, ee = verify_claim(store, anchors, t_us)
|
|
shape = shape_of(labels, init, kind)
|
|
if shape is None:
|
|
raise ValueError("manifest_malformed")
|
|
items = []
|
|
for a in store["assertions"]:
|
|
try:
|
|
items.append((a["label"], cbor_read(a["cbor"])))
|
|
except CborError:
|
|
raise ValueError("manifest_malformed")
|
|
hb = read_hash_item(items[0][1], not init)
|
|
if hb is None or read_actions(items[1][1]) is None:
|
|
raise ValueError("manifest_malformed")
|
|
for label, item in items[2:]:
|
|
if label == L_LIVE:
|
|
ok = read_live(item) is not None
|
|
elif label == L_KEYS:
|
|
ok = is_map(item, ["keys"]) and isinstance(item["keys"], list) and bool(item["keys"])
|
|
elif label == L_GROUP:
|
|
ok = read_group(item) is not None
|
|
elif label in (L_STREAM, L_TAKE):
|
|
ok = isinstance(item, dict)
|
|
else:
|
|
ok = read_ai(item) is not None
|
|
if not ok:
|
|
raise ValueError("manifest_malformed")
|
|
if bmff_hash(asset, top) != hb[0]:
|
|
raise ValueError("hard_binding_mismatch")
|
|
return dict(id=store["id"], shape=shape, items=dict(items), order=items, ee=ee, seq=hb[1], store=store)
|
|
|
|
|
|
def derived_id(stream_id, role, seq):
|
|
"""§4.2: the identifier of the manifest of `role` for sequence number `seq`."""
|
|
sid = stream_id.encode()
|
|
h = bytearray(hashlib.sha256(MANIFEST_DOMAIN + struct.pack(">I", len(sid)) + sid + bytes([role])
|
|
+ struct.pack(">Q", seq)).digest()[:16])
|
|
h[6] = (h[6] & 0x0F) | 0x40
|
|
h[8] = (h[8] & 0x3F) | 0x80
|
|
return "urn:c2pa:" + uuid_text(bytes(h))
|
|
|
|
|
|
# ───────────────────────────── the repository receipt (§4.7) ─────────────────────────────
|
|
|
|
|
|
def log_leaf(entry):
|
|
return hashlib.sha256(b"\x00" + entry).digest()
|
|
|
|
|
|
def log_node(left, right):
|
|
return hashlib.sha256(b"\x01" + left + right).digest()
|
|
|
|
|
|
def log_root(entry, index, size, path):
|
|
"""RFC 9162 §2.1.3.2."""
|
|
if index >= size:
|
|
return None
|
|
fn, sn, r = index, size - 1, log_leaf(entry)
|
|
for p in path:
|
|
if sn == 0:
|
|
return None
|
|
if fn & 1 or fn == sn:
|
|
r = log_node(p, r)
|
|
while not fn & 1 and fn != 0:
|
|
fn >>= 1
|
|
sn >>= 1
|
|
else:
|
|
r = log_node(r, p)
|
|
fn >>= 1
|
|
sn >>= 1
|
|
return r if sn == 0 else None
|
|
|
|
|
|
def verify_inclusion_receipt(receipt, entry, key):
|
|
"""RFC 9942 §5.2.1: protected {1: -8, 395: 1}; unprotected {396: {-1: [proof]}}; detached
|
|
payload; the proof leads from the entry to a root under which the key's signature
|
|
verifies. The tree size and the leaf index, or None."""
|
|
try:
|
|
s = read_sign1(cbor_read(receipt))
|
|
except CborError:
|
|
return None
|
|
if s is None or s["protected"] != {1: -8, 395: 1} or s["payload"] is not None or set(s["unprotected"]) != {396}:
|
|
return None
|
|
vdp = s["unprotected"][396]
|
|
if not isinstance(vdp, dict) or set(vdp) != {-1} or not isinstance(vdp[-1], list) or len(vdp[-1]) != 1:
|
|
return None
|
|
if not isinstance(vdp[-1][0], bytes):
|
|
return None
|
|
try:
|
|
c = cbor_read(vdp[-1][0])
|
|
except CborError:
|
|
return None
|
|
if not isinstance(c, list) or len(c) != 3 or not uint(c[0]) or not uint(c[1]) or not isinstance(c[2], list) or not c[2]:
|
|
return None
|
|
if not all(b32(h) for h in c[2]):
|
|
return None
|
|
root = log_root(entry, c[1], c[0], c[2])
|
|
if root is None or not cose_verifies(s, key, root):
|
|
return None
|
|
return c[0], c[1]
|
|
|
|
|
|
def verify_update(store_bytes, original, stream_id, anchors, repositories, t_us):
|
|
"""§11.3 step 6: the update manifest over `original` and its repository receipt."""
|
|
store = read_store(store_bytes or b"")
|
|
if store is None or store["type"] != T_UPDATE:
|
|
raise ValueError("update_manifest_invalid")
|
|
try:
|
|
labels, _ = verify_claim(store, anchors, t_us)
|
|
except ValueError as e:
|
|
raise ValueError("update_manifest_invalid" if str(e) == "manifest_malformed" else str(e))
|
|
if store["id"] != derived_id(stream_id, ROLE_UPDATE, 0):
|
|
raise ValueError("manifest_id_mismatch")
|
|
if labels != [L_INGREDIENT, L_REPOSITORY] or [a["form"] for a in store["assertions"]] != ["cbor", "json"]:
|
|
raise ValueError("update_manifest_invalid")
|
|
sig = "self#jumbf=/c2pa/%s/c2pa.signature" % original["id"]
|
|
labels0 = [a["label"] for a in original["assertions"]]
|
|
success = [{"code": c, "url": sig} for c in ("claimSignature.validated", "signingCredential.trusted", "claimSignature.insideValidity")]
|
|
success += [{"code": "assertion.hashedURI.match", "url": "self#jumbf=/c2pa/%s/c2pa.assertions/%s" % (original["id"], l)} for l in labels0]
|
|
success += [{"code": "assertion.bmffHash.match", "url": "self#jumbf=/c2pa/%s/c2pa.assertions/%s" % (original["id"], labels0[0])}]
|
|
want = {
|
|
"relationship": "parentOf",
|
|
"validationResults": {"activeManifest": {"success": success, "informational": [], "failure": []},
|
|
"ingredientDeltas": [], "specVersion": "2.4.0"},
|
|
"activeManifest": {"url": "self#jumbf=/c2pa/" + original["id"], "hash": original["manifest_hash"]},
|
|
"claimSignature": {"url": "self#jumbf=/c2pa/%s/c2pa.signature" % original["id"], "hash": original["signature_hash"]},
|
|
}
|
|
try:
|
|
if cbor_read(store["assertions"][0]["cbor"]) != want:
|
|
raise ValueError("update_manifest_invalid")
|
|
v = json.loads(store["assertions"][1]["cbor"])
|
|
except (CborError, ValueError):
|
|
raise ValueError("update_manifest_invalid")
|
|
|
|
def jkeys(o, required, optional=()):
|
|
return isinstance(o, dict) and all(k in required or k in optional for k in o) and all(k in o for k in required)
|
|
|
|
ok = (jkeys(v, ["@context", "@type", "repository", "anchor"]) and isinstance(v["@context"], (dict, str))
|
|
and v["@type"] == "org.c2pa.repository-receipt" and jkeys(v["repository"], ["uri", "manifestId"])
|
|
and jkeys(v["anchor"], ["uri", "proof"], ["parameters"]) and jkeys(v["anchor"]["proof"], ["profile", "entry", "receipt"])
|
|
and isinstance(v["anchor"]["uri"], str) and v["repository"]["manifestId"] == original["id"]
|
|
and v["anchor"]["proof"]["profile"] == "urn:ietf:rfc:9942"
|
|
and v["anchor"]["proof"]["entry"] == original["manifest_hash"].hex()
|
|
and isinstance(v["repository"]["uri"], str) and isinstance(v["anchor"]["proof"]["receipt"], str))
|
|
if not ok:
|
|
raise ValueError("update_manifest_invalid")
|
|
r64 = v["anchor"]["proof"]["receipt"]
|
|
if not re.fullmatch(r"[A-Za-z0-9_-]*", r64) or len(r64) % 4 == 1:
|
|
raise ValueError("update_manifest_invalid")
|
|
receipt = base64.urlsafe_b64decode(r64 + "=" * (-len(r64) % 4))
|
|
key = repositories.get(v["repository"]["uri"])
|
|
if key is None:
|
|
raise ValueError("repository_receipt_invalid")
|
|
got = verify_inclusion_receipt(receipt, original["manifest_hash"], key)
|
|
if got is None:
|
|
raise ValueError("repository_receipt_invalid")
|
|
return {"repository": v["repository"]["uri"], "treeSize": got[0], "leafIndex": got[1]}
|
|
|
|
|
|
def check_label(m, kind, ident):
|
|
"""§9.3."""
|
|
term, models = disclosure(kind, ident)
|
|
if read_actions(m["items"][L_ACTIONS]) != term:
|
|
raise ValueError("source_type_mismatch")
|
|
held = [read_ai(i) for l, i in m["order"] if l.startswith(L_AI)]
|
|
want = [(MODEL_URI + mm.hex(), "fully_autonomous" if auto else "human_validated") for mm, auto in models]
|
|
if held != want:
|
|
raise ValueError("ai_disclosure_missing")
|
|
|
|
|
|
def rfc3339(s):
|
|
if not isinstance(s, str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z", s):
|
|
return None
|
|
y, mo, d, h, mi, se = int(s[0:4]), int(s[5:7]), int(s[8:10]), int(s[11:13]), int(s[14:16]), int(s[17:19])
|
|
if not (1 <= mo <= 12 and 1 <= d <= 31 and h <= 23 and mi <= 59 and se <= 59):
|
|
return None
|
|
try:
|
|
t = calendar.timegm((y, mo, d, h, mi, se, 0, 0, 0))
|
|
except (OverflowError, ValueError):
|
|
return None
|
|
back = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t))
|
|
return t if back == s and t >= 0 else None
|
|
|
|
|
|
def read_keys(m, device):
|
|
"""§11.3 step 5: the session keys a manifest delivers."""
|
|
item = m["items"].get(L_KEYS)
|
|
if item is None:
|
|
return []
|
|
out = []
|
|
for k in item["keys"]:
|
|
if not is_map(k, ["key", "minSequenceNumber", "createdAt", "validityPeriod", "signerBinding"]):
|
|
raise ValueError("session_key_invalid")
|
|
x = read_okp(k["key"])
|
|
ca = k["createdAt"]
|
|
created = rfc3339(ca.value) if isinstance(ca, Tag) and ca.tag == 0 else None
|
|
if x is None or not uint(k["minSequenceNumber"]) or created is None or not uint(k["validityPeriod"]):
|
|
raise ValueError("session_key_invalid")
|
|
b = read_sign1(k["signerBinding"])
|
|
if (b is None or len(b["protected"]) != 1 or b["payload"] is not None or b["unprotected"] != {}
|
|
or not cose_verifies(b, x, m["ee"])):
|
|
raise ValueError("session_key_invalid")
|
|
if (device is not None and device != x) or any(h["x"] == x for h in out):
|
|
raise ValueError("session_key_invalid")
|
|
out.append(dict(x=x, min=k["minSequenceNumber"], frm=created * 1000000,
|
|
to=(created + k["validityPeriod"]) * 1000000))
|
|
return out
|
|
|
|
|
|
# ───────────────────────────── §11: a stream ─────────────────────────────
|
|
|
|
|
|
def verify_stream(kind, init, segments, groups, objects, anchors, repositories, t_us):
|
|
"""Returns §11.6's statement, or raises Refuse."""
|
|
def refuse(reason, at):
|
|
raise Refuse(reason, at)
|
|
|
|
top = boxes(init, 0, len(init))
|
|
if top is None:
|
|
refuse("segment_malformed", "init")
|
|
if any(r[0] == b"mdat" for r in top):
|
|
refuse("init_invalid", "init")
|
|
if (len(top) not in (3, 4) or top[0][0] != b"ftyp" or not is_c2pa_uuid(init, top[1]) or top[2][0] != b"moov"
|
|
or (len(top) == 4 and not is_c2pa_uuid(init, top[3]))):
|
|
refuse("init_invalid", "init")
|
|
update = top[3] if len(top) == 4 else None
|
|
track = read_moov(init, top[2])
|
|
if track is None:
|
|
refuse("init_invalid", "init")
|
|
store0 = manifest_store_bytes(init, top[1], b"original" if update else b"manifest")
|
|
if store0 is None:
|
|
refuse("manifest_malformed", "init")
|
|
try:
|
|
m0 = verify_manifest(init, top, store0, True, kind, anchors, t_us)
|
|
except ValueError as e:
|
|
refuse(str(e), "init")
|
|
if kind == "track":
|
|
ident = read_stream(m0["items"][L_STREAM])
|
|
if ident is None:
|
|
refuse("stream_assertion_invalid", "init")
|
|
stream_id, epoch, first_group, device = stream_id_of(ident), ident["epoch"], ident["firstGroup"], None
|
|
else:
|
|
ident = read_take(m0["items"][L_TAKE], True)
|
|
if ident is None:
|
|
refuse("stream_assertion_invalid", "init")
|
|
stream_id, epoch, first_group, device = uuid_v8(ident["take"][:16]), ident["epoch"], 0, ident["device"]
|
|
if m0["id"] != derived_id(stream_id, ROLE_INIT, 0):
|
|
refuse("manifest_id_mismatch", "init")
|
|
registration = None
|
|
try:
|
|
check_label(m0, kind, ident)
|
|
held_keys = read_keys(m0, device)
|
|
if update:
|
|
ustore = manifest_store_bytes(init, update, b"update")
|
|
if ustore is None:
|
|
raise ValueError("update_manifest_invalid")
|
|
registration = verify_update(ustore, m0["store"], stream_id, anchors, repositories, t_us)
|
|
except ValueError as e:
|
|
refuse(str(e), "init")
|
|
|
|
prev_seq = prev_group = None
|
|
prev_last = 0
|
|
prev_manifest = active = m0["id"]
|
|
seen, out = [], []
|
|
for k, (b, g) in enumerate(segments):
|
|
at = "segment %d" % k
|
|
top = boxes(b, 0, len(b))
|
|
if top is None:
|
|
refuse("segment_malformed", at)
|
|
i = 0
|
|
mbox = top[0] if top and is_c2pa_uuid(b, top[0]) else None
|
|
if mbox:
|
|
i += 1
|
|
vbox = top[i] if len(top) > i and is_vsi(b, top[i]) else None
|
|
if vbox:
|
|
i += 1
|
|
rest = top[i:]
|
|
if not rest or len(rest) % 2:
|
|
refuse("segment_malformed", at)
|
|
chunks, infos = [], []
|
|
for j in range(0, len(rest), 2):
|
|
info = read_chunk(b, rest[j], rest[j + 1])
|
|
if info is None:
|
|
refuse("segment_malformed", at)
|
|
infos.append(info)
|
|
chunks.append(b[rest[j][1]:rest[j + 1][2]])
|
|
if i == 0:
|
|
refuse("provenance_missing", at)
|
|
if any(x["track_id"] != track["track_id"] for x in infos) or any(
|
|
infos[j + 1]["seq"] != infos[j]["seq"] + 1 for j in range(len(infos) - 1)):
|
|
refuse("segment_malformed", at)
|
|
seq, last, n = infos[0]["seq"], infos[-1]["seq"], len(chunks)
|
|
if (prev_seq is not None and seq <= prev_seq) or (
|
|
kind == "track" and prev_group is not None and (g is None or g <= prev_group)):
|
|
refuse("sequence_not_increasing", at)
|
|
# Step 3.
|
|
per = None
|
|
if mbox:
|
|
try:
|
|
m = verify_manifest(b, top, manifest_store_bytes(b, mbox), False, kind, anchors, t_us)
|
|
except ValueError as e:
|
|
refuse(str(e), at)
|
|
role = ROLE_SEGMENT if m["shape"] == "per-segment" else ROLE_ROTATION
|
|
if m["id"] != derived_id(stream_id, role, seq):
|
|
refuse("manifest_id_mismatch", at)
|
|
if m["seq"] != seq:
|
|
refuse("sequence_mismatch", at)
|
|
if m["shape"] == "per-segment":
|
|
lv = m["items"][L_LIVE]
|
|
if lv["sequenceNumber"] != seq:
|
|
refuse("sequence_mismatch", at)
|
|
if lv["streamId"] != stream_id:
|
|
refuse("stream_id_mismatch", at)
|
|
if lv.get("continuityMethod") != "c2pa.manifestId" or "previousManifestId" not in lv:
|
|
refuse("continuity_invalid", at)
|
|
if lv["previousManifestId"] != prev_manifest:
|
|
refuse("continuity_broken", at)
|
|
if kind == "take":
|
|
t = read_take(m["items"][L_TAKE], False)
|
|
if t is None or (t["take"], t["room"], t["device"]) != (ident["take"], ident["room"], ident["device"]):
|
|
refuse("stream_assertion_invalid", at)
|
|
else:
|
|
if vbox is None:
|
|
refuse("provenance_missing", at)
|
|
if kind == "track":
|
|
same = read_stream(m["items"][L_STREAM]) == ident
|
|
else:
|
|
same = read_take(m["items"][L_TAKE], True) == ident
|
|
if not same:
|
|
refuse("stream_assertion_invalid", at)
|
|
try:
|
|
new = read_keys(m, device)
|
|
except ValueError as e:
|
|
refuse(str(e), at)
|
|
if any(h["min"] < seq or any(o["x"] == h["x"] for o in held_keys) for h in new):
|
|
refuse("session_key_invalid", at)
|
|
held_keys = held_keys + new
|
|
try:
|
|
check_label(m, kind, ident)
|
|
except ValueError as e:
|
|
refuse(str(e), at)
|
|
prev_manifest = active = m["id"]
|
|
if m["shape"] == "per-segment":
|
|
per = m
|
|
methods = {(True, True): "manifest+vsi", (True, False): "manifest"}.get((per is not None, vbox is not None), "vsi")
|
|
# Step 4.
|
|
signer = None
|
|
if vbox:
|
|
e = read_emsg(b, vbox)
|
|
dur = sum(x["duration"] for x in infos)
|
|
if (e is None or e["scheme"] != VSI_SCHEME or e["value"] != "fseg" or e["timescale"] != track["timescale"]
|
|
or e["ptd"] != 0 or e["duration"] != dur or e["id"] != seq):
|
|
refuse("segment_info_invalid", at)
|
|
try:
|
|
cs = read_sign1(cbor_read(e["data"]))
|
|
except CborError:
|
|
cs = None
|
|
if (cs is None or len(cs["protected"]) != 1 or len(cs["unprotected"]) != 1
|
|
or not isinstance(cs["unprotected"].get(4), bytes) or cs["payload"] is None):
|
|
refuse("segment_info_invalid", at)
|
|
try:
|
|
info = cbor_read(cs["payload"])
|
|
except CborError:
|
|
refuse("segment_info_invalid", at)
|
|
if not is_map(info, ["sequenceNumber", "bmffHash", "manifestId"]):
|
|
refuse("segment_info_invalid", at)
|
|
vh = read_hash_item(info["bmffHash"], True)
|
|
if vh is None or not uint(info["sequenceNumber"]) or not isinstance(info["manifestId"], str):
|
|
refuse("segment_info_invalid", at)
|
|
key = next((h for h in held_keys if h["x"] == cs["unprotected"][4]), None)
|
|
if key is None:
|
|
refuse("session_key_unknown", at)
|
|
if not cose_verifies(cs, key["x"], cs["payload"]):
|
|
refuse("segment_info_signature_failed", at)
|
|
pres = epoch + infos[0]["bmdt"] * 1000000 // track["timescale"]
|
|
superseded = any(key["min"] < o["min"] <= seq for o in held_keys)
|
|
if key["min"] > seq or superseded or not key["frm"] <= pres < key["to"]:
|
|
refuse("session_key_out_of_range", at)
|
|
if info["sequenceNumber"] != seq or vh[1] != seq:
|
|
refuse("sequence_mismatch", at)
|
|
if info["manifestId"] != active:
|
|
refuse("segment_info_invalid", at)
|
|
if bmff_hash(b, top) != vh[0]:
|
|
refuse("segment_hash_mismatch", at)
|
|
signer = key["x"]
|
|
# Steps 5 and 6.
|
|
if kind == "track":
|
|
if g is None:
|
|
refuse("segment_malformed", at)
|
|
expect = 1
|
|
for h in range(first_group, g):
|
|
rh = [r for r in groups if r["group_id"] == h]
|
|
if not rh:
|
|
refuse("receipt_missing", "group %d" % h)
|
|
expect += rh[0]["leaf_count"]
|
|
held = [r for r in groups if r["group_id"] == g]
|
|
if len(held) != 1:
|
|
refuse("receipt_missing" if not held else "receipt_invalid", "group %d" % g)
|
|
rg = held[0]
|
|
if seq != expect:
|
|
refuse("sequence_mismatch", at)
|
|
atg = "group %d" % g
|
|
prev = next((r for r in groups if r["group_id"] == g - 1), None) if g > first_group else None
|
|
parent = rg["parent_receipt_hash"]
|
|
if prev is not None and prev["signer_pubkey"] == rg["signer_pubkey"]:
|
|
parent_ok = parent == link_hash(prev).hex()
|
|
else:
|
|
parent_ok = parent is None
|
|
if g < first_group or not group_ok(rg) or rg["track_alias"] != ident["trackAlias"] or not parent_ok:
|
|
refuse("receipt_invalid", atg)
|
|
leaves = [leaf(blake3(b"moq-jwp:object\x01", c)) for c in chunks]
|
|
named = True
|
|
if per is not None:
|
|
ga = per["items"][L_GROUP]
|
|
named = (ga["groupId"] == g and ga["receipt"] == link_hash(rg) and ga["root"].hex() == rg["root_hash"]
|
|
and ga["objects"] == rg["leaf_count"])
|
|
if rg["leaf_count"] != n or rg["root_hash"] != merkle_root(leaves).hex() or not named:
|
|
refuse("receipt_mismatch", atg)
|
|
if signer is not None and signer.hex() != rg["signer_pubkey"]:
|
|
refuse("receipt_signer_mismatch", atg)
|
|
objs = [o for o in objects if o["group_id"] == g]
|
|
if len(objs) != n:
|
|
refuse("object_receipt_invalid", atg)
|
|
for j, c in enumerate(chunks):
|
|
mine = [o for o in objs if o["object_id"] == j]
|
|
if len(mine) != 1 or not object_accepts(mine[0], rg, c):
|
|
refuse("object_receipt_invalid", atg)
|
|
seen.append(g)
|
|
prev_group = g
|
|
index, digest, dkey, ikey = g, rg["root_hash"], "root", "group"
|
|
else:
|
|
if seq != prev_last + 1:
|
|
refuse("sequence_mismatch", at)
|
|
index = k
|
|
digest = hashlib.sha256((init + b) if k == 0 else b).hexdigest()
|
|
dkey, ikey = "chunkHash", "chunk"
|
|
prev_seq, prev_last = seq, last
|
|
out.append({"sequenceNumber": seq, ikey: index, "methods": methods, "manifestId": active, "chunks": n, dkey: digest})
|
|
if kind == "track":
|
|
gs = [r["group_id"] for r in groups if r["group_id"] >= first_group]
|
|
if gs:
|
|
for h in range(first_group, max(gs) + 1):
|
|
if h not in seen:
|
|
refuse("segment_missing", "group %d" % h)
|
|
return {"outcome": "accept", "streamId": stream_id, "sourceType": DST_PREFIX + disclosure(kind, ident)[0],
|
|
"registration": registration, "segments": out}
|
|
|
|
|
|
# ───────────────────────────── the corpus ─────────────────────────────
|
|
|
|
|
|
def run(v, doc, files):
|
|
anchors = [read_cert(files[a]) for a in doc["anchors"]]
|
|
t_us = rfc3339(doc["time"]) * 1000000
|
|
segs = [(files[s["file"]], s.get("group")) for s in v["segments"]]
|
|
groups, objects = [], []
|
|
if v["receipts"] is not None:
|
|
r = json.loads(files[v["receipts"]])
|
|
groups, objects = r["groups"], r["objects"]
|
|
repositories = {r["uri"]: bytes.fromhex(r["key"]) for r in doc["repositories"]}
|
|
try:
|
|
return verify_stream(v["kind"], files[v["init"]], segs, groups, objects, anchors, repositories, t_us)
|
|
except Refuse as e:
|
|
c2pa, live = REASONS[e.reason]
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at, "c2pa": c2pa, "livevideo": live}
|
|
|
|
|
|
def load(root):
|
|
files = {}
|
|
for dirpath, _, names in os.walk(root):
|
|
for n in names:
|
|
p = os.path.join(dirpath, n)
|
|
rel = os.path.relpath(p, root).replace(os.sep, "/")
|
|
if rel != "README.md":
|
|
files[rel] = open(p, "rb").read()
|
|
return files
|
|
|
|
|
|
def flip_last_mdat(b):
|
|
top = boxes(b, 0, len(b))
|
|
m = [r for r in top if r[0] == b"mdat"][-1]
|
|
return b[:m[2] - 1] + bytes([b[m[2] - 1] ^ 1]) + b[m[2]:]
|
|
|
|
|
|
def mutations(doc, files):
|
|
"""One defect each, on copies of the good vectors; every copy must be refused."""
|
|
out = []
|
|
for v in doc["vectors"]:
|
|
if v["expect"]["outcome"] != "accept":
|
|
continue
|
|
mid = len(v["segments"]) // 2
|
|
|
|
def segment_bytes(fs, vv, j, f):
|
|
path = "mutated/" + vv["segments"][j]["file"]
|
|
fs[path] = f(fs[vv["segments"][j]["file"]])
|
|
vv["segments"][j]["file"] = path
|
|
|
|
def m_mdat(vv, fs):
|
|
segment_bytes(fs, vv, mid, flip_last_mdat)
|
|
|
|
def m_drop(vv, fs):
|
|
del vv["segments"][mid]
|
|
|
|
def m_swap(vv, fs):
|
|
vv["segments"][0], vv["segments"][1] = vv["segments"][1], vv["segments"][0]
|
|
|
|
def m_moov(vv, fs):
|
|
b = fs[vv["init"]]
|
|
top = boxes(b, 0, len(b))
|
|
at = top[2][2] - 1
|
|
fs["mutated/" + vv["init"]] = b[:at] + bytes([b[at] ^ 1]) + b[at + 1:]
|
|
vv["init"] = "mutated/" + vv["init"]
|
|
|
|
def m_claim_sig(vv, fs):
|
|
b = fs[vv["init"]]
|
|
top = boxes(b, 0, len(b))
|
|
sig = read_store(manifest_store_bytes(b, top[1]) or manifest_store_bytes(b, top[1], b"original"))["signature"][-64:]
|
|
at = b.index(sig) + 10
|
|
fs["mutated/" + vv["init"]] = b[:at] + bytes([b[at] ^ 1]) + b[at + 1:]
|
|
vv["init"] = "mutated/" + vv["init"]
|
|
|
|
def m_last_byte_of_c2pa_box(vv, fs):
|
|
def f(b):
|
|
top = boxes(b, 0, len(b))
|
|
r = top[0]
|
|
return b[:r[2] - 1] + bytes([b[r[2] - 1] ^ 1]) + b[r[2]:]
|
|
segment_bytes(fs, vv, mid, f)
|
|
|
|
def m_receipt(vv, fs):
|
|
r = json.loads(fs[vv["receipts"]])
|
|
g = r["groups"][mid]
|
|
g["sig"] = ("0" if g["sig"][0] != "0" else "1") + g["sig"][1:]
|
|
fs["mutated/receipts.json"] = json.dumps(r).encode()
|
|
vv["receipts"] = "mutated/receipts.json"
|
|
|
|
def m_object(vv, fs):
|
|
r = json.loads(fs[vv["receipts"]])
|
|
o = [x for x in r["objects"] if x["group_id"] == vv["segments"][mid]["group"]][0]
|
|
o["content_hash"] = ("0" if o["content_hash"][0] != "0" else "1") + o["content_hash"][1:]
|
|
fs["mutated/receipts.json"] = json.dumps(r).encode()
|
|
vv["receipts"] = "mutated/receipts.json"
|
|
|
|
def m_time(vv, fs, d):
|
|
d["time"] = "2027-06-01T00:00:00Z"
|
|
|
|
ms = [("a byte of the middle segment's media", m_mdat), ("the middle segment dropped", m_drop),
|
|
("the first two segments swapped", m_swap), ("a byte of the initialization segment's moov", m_moov),
|
|
("a byte of the initialization manifest's claim signature", m_claim_sig),
|
|
("the last byte of the middle segment's first C2PA box", m_last_byte_of_c2pa_box)]
|
|
def m_update(vv, fs):
|
|
b = fs[vv["init"]]
|
|
at = len(b) - 1
|
|
fs["mutated/" + vv["init"]] = b[:at] + bytes([b[at] ^ 1])
|
|
vv["init"] = "mutated/" + vv["init"]
|
|
|
|
if len(boxes(files[v["init"]], 0, len(files[v["init"]]))) == 4:
|
|
ms.append(("the last byte of the update manifest's store", m_update))
|
|
if v["kind"] == "track":
|
|
ms += [("a byte of the middle group's receipt signature", m_receipt),
|
|
("a byte of an object receipt's content hash", m_object)]
|
|
for label, f in ms:
|
|
vv, fs = copy.deepcopy(v), dict(files)
|
|
f(vv, fs)
|
|
out.append(("%s: %s" % (v["name"], label), vv, doc, fs))
|
|
vv, fs, dd = copy.deepcopy(v), dict(files), copy.deepcopy(doc)
|
|
m_time(vv, fs, dd)
|
|
out.append(("%s: verified after the end-entity certificate expired" % v["name"], vv, dd, fs))
|
|
return out
|
|
|
|
|
|
def main(root):
|
|
files = load(root)
|
|
doc = json.loads(files["vectors.json"])
|
|
fails, reasons = [], set()
|
|
used = {"vectors.json"} | set(doc["anchors"])
|
|
for v in doc["vectors"]:
|
|
used |= {v["init"]} | {s["file"] for s in v["segments"]} | ({v["receipts"]} if v["receipts"] else set())
|
|
try:
|
|
got = run(v, doc, files)
|
|
except Exception as e: # a crash is a failure
|
|
got = {"error": repr(e)}
|
|
if got.get("outcome") == "refuse":
|
|
reasons.add(got["reason"])
|
|
if got != v["expect"]:
|
|
fails.append("%s: expected %s, got %s" % (v["name"], json.dumps(v["expect"]), json.dumps(got)))
|
|
else:
|
|
print("PASS %s: %s%s" % (v["name"], got["outcome"], (" " + got["reason"] + " at " + got["at"]) if got["outcome"] == "refuse" else ""))
|
|
for r in REASONS:
|
|
if r not in reasons:
|
|
fails.append("no vector is refused %s" % r)
|
|
extra = sorted(set(files) - used)
|
|
if extra:
|
|
fails.append("files no vector names: %s" % extra)
|
|
caught = 0
|
|
for label, vv, dd, fs in mutations(doc, files):
|
|
try:
|
|
got = run(vv, dd, fs)
|
|
except Exception as e: # a crash on a mutated copy is a failure to read it, which is a refusal
|
|
got = {"outcome": "refuse", "reason": "crash: %r" % e}
|
|
if got.get("outcome") == "accept":
|
|
fails.append("mutation not caught: " + label)
|
|
else:
|
|
caught += 1
|
|
print("%d mutated copies, %d refused" % (caught + sum(1 for f in fails if f.startswith("mutation")), caught))
|
|
for f in fails:
|
|
print("FAIL " + f)
|
|
print("PASS" if not fails else "FAIL", "every vector of", root)
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: c2pa_live_verify.py <live-conformance/c2pa-live>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|