2fd26925d3
`wai inspect` reads envelopes; most files a toolchain meets are not envelopes, and the first question about one is what it is. `wai identify <file>` (operation media.inspect, extensions/media-identify.md) answers it for any file: - the formats the bytes fit, each with the offsets of the bytes that say so, how far its structure was followed and where and why that stopped; - a container's tracks as it states them: codec, size, channels, rate, protection; - the registered capabilities they map to, and what a format probe reads of them; - signatures found inside the file; - where no format is recognised, measures that still hold: how the byte values are spread, uniform windows, and the lags at which the bytes repeat. The verdict is identified, ambiguous, signature_only or unknown, and there is no confidence figure: a candidate's strength says what was checked. Nothing is decoded, no protected track is opened, and unknown stays unknown. Walks cover ISO media, Matroska, transport streams, Ogg, FLAC, RIFF, MPEG audio, ADTS, PNG, JPEG, GIF, IVF, YUV4MPEG2, FLV, AV1 OBU and Annex B streams, and HLS and DASH text; 35 more formats are known by their signature. Every walk is bounded, reads a file where its lengths lead (a large file is never held whole), and has been given hostile bytes; it builds for wasm32. The agent door has it as media_inspect, its results marked untrusted_content: a file's author chooses the strings in it. identify-conformance/ holds files written by real muxers and encoders from test patterns and tones, what an independent media prober reads of them (prober.json), and generated cases sitting on each rule's edge. identify_verify.py is a second implementation written from the text; it reads the extension's tables, gives every file its whole result, and holds ext/ to the prober's reading. tests/identify.rs holds the reference's tables to the extension's. Co-Authored-By: Claude Opus 5.5 <[email protected]>
2155 lines
70 KiB
Python
2155 lines
70 KiB
Python
"""The second implementation of media identification: written from extensions/media-identify.md,
|
|
sharing no code with wai-rs (the probes are format_probe_verify.py's, itself written from
|
|
format-probes).
|
|
|
|
python3 -I identify_verify.py <corpus-dir>
|
|
|
|
It identifies every file of <corpus-dir>/vectors.json and requires the whole result stated there;
|
|
that every file under ext/ and gen/ is a vector's; that every reason, verdict and strength is
|
|
reached; and that what prober.json reads of ext/ agrees wherever both state a value. Then it mutates
|
|
copies of the corpus and requires each copy to fail.
|
|
|
|
Exit status 0 only if everything verifies and every mutated copy fails. Python 3.8 or later;
|
|
standard library only.
|
|
"""
|
|
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import re
|
|
import struct
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
HEAD = 1 << 20
|
|
SCAN = 16 << 20
|
|
MAX_STEPS = 4096
|
|
U64 = (1 << 64) - 1
|
|
|
|
_spec = importlib.util.spec_from_file_location("fpv", os.path.join(HERE, "..", "format-probe-conformance", "format_probe_verify.py"))
|
|
fpv = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(fpv)
|
|
|
|
|
|
def spec_text():
|
|
return open(os.path.join(HERE, "..", "extensions", "media-identify.md"), encoding="utf-8").read()
|
|
|
|
|
|
def section(text, heading):
|
|
at = text.index("\n" + heading)
|
|
end = text.find("\n## ", at + 1)
|
|
return text[at:] if end < 0 else text[at:end]
|
|
|
|
|
|
def table_rows(text):
|
|
rows = []
|
|
for line in text.split("\n"):
|
|
if line.startswith("| ") and not line.startswith("| id ") and not line.startswith("|-"):
|
|
rows.append([c.strip().strip("`") for c in line.strip().strip("|").split("|")])
|
|
return rows
|
|
|
|
|
|
def signature_rows(text):
|
|
out = []
|
|
for r in table_rows(section(text, "## 6.")):
|
|
parts = [(int(p.split(":")[0]), bytes.fromhex(p.split(":")[1])) for p in r[2].split()]
|
|
out.append((r[0], r[1], parts, r[3].split()))
|
|
return out
|
|
|
|
|
|
def embedded_rows(text):
|
|
return [(r[0], bytes.fromhex(r[1]), int(r[2])) for r in table_rows(section(text, "## 7."))]
|
|
|
|
|
|
def codec_capabilities(text):
|
|
s = section(text, "## 5.")
|
|
s = s[s.index("| codec | capability |"):]
|
|
out = {}
|
|
for line in s.split("\n")[2:]:
|
|
if not line.startswith("|"):
|
|
break
|
|
c = [x.strip().strip("`") for x in line.strip("|").split("|")]
|
|
out[c[0]] = c[1]
|
|
return out
|
|
|
|
|
|
def spec_table(text, header):
|
|
"""The rows of the table whose header line is `header`, each cell without its backticks."""
|
|
s = text[text.index("\n" + header + "\n") + 1:]
|
|
rows = []
|
|
for line in s.split("\n")[2:]:
|
|
if not line.startswith("|"):
|
|
break
|
|
rows.append([c.strip().strip("`") for c in line.strip().strip("|").split("|")])
|
|
return rows
|
|
|
|
|
|
def paired(rows, width):
|
|
"""A table that holds `width`-column records side by side, as records."""
|
|
out = []
|
|
for r in rows:
|
|
for k in range(0, len(r), width):
|
|
rec = r[k:k + width]
|
|
if rec and rec[0]:
|
|
out.append(rec)
|
|
return out
|
|
|
|
|
|
def iso_codecs(text):
|
|
out = {}
|
|
for names, codec in paired(spec_table(text, "| entry | codec | entry | codec |"), 2):
|
|
for n in names.replace("`", "").split():
|
|
out[n.encode()] = codec
|
|
return out
|
|
|
|
|
|
def mkv_codecs(text):
|
|
return {i.encode(): c for i, c in paired(spec_table(text, "| identifier | codec | identifier | codec |"), 2)}
|
|
|
|
|
|
def ts_types(text):
|
|
out = {}
|
|
for types, codec, media in paired(spec_table(text, "| type | codec | media | type | codec | media |"), 3):
|
|
for t in types.split():
|
|
out[int(t, 16)] = (None if codec == "null" else codec, media)
|
|
return out
|
|
|
|
|
|
def bitrates(text):
|
|
out = {}
|
|
for label, rates in spec_table(text, "| | indexes 1 to 14 |"):
|
|
version = 3 if label.startswith("version 1,") else 2
|
|
layers = [n for word, n in (("III", 1), ("II", 2), ("I", 3)) if re.search(r"\b%s\b" % word, label)]
|
|
for layer in layers:
|
|
out[(version, layer)] = [int(x) for x in rates.split()]
|
|
return out
|
|
|
|
|
|
def adts_rates(text):
|
|
m = re.search(r"rate index \(below 13: ([0-9,\s]+)\)", text)
|
|
return [int(x) for x in m.group(1).replace(",", " ").split()]
|
|
|
|
|
|
TEXT = spec_text()
|
|
SIGNATURES = signature_rows(TEXT)
|
|
EMBEDDED = embedded_rows(TEXT)
|
|
CAPS = codec_capabilities(TEXT)
|
|
ISO_CODEC = iso_codecs(TEXT)
|
|
MKV_CODEC = mkv_codecs(TEXT)
|
|
TS_TYPE = ts_types(TEXT)
|
|
BITRATES = bitrates(TEXT)
|
|
ADTS_RATES = adts_rates(TEXT)
|
|
AVC_PROFILES = [int(x) for x in re.search(r"a `profile_idc`\s+other than ([0-9,\sor]+?)\s+or a failed", TEXT).group(1).replace(",", " ").split() if x != "or"]
|
|
|
|
|
|
def clean(b):
|
|
out = []
|
|
for c in b[:64]:
|
|
out.append(chr(c) if 0x20 <= c < 0x7f and c != 0x5c else "\\x%02x" % c)
|
|
return "".join(out)
|
|
|
|
|
|
def num(b, i, n, order):
|
|
if i < 0 or i + n > len(b):
|
|
return None
|
|
return int.from_bytes(b[i:i + n], order)
|
|
|
|
|
|
def be16(b, i): return num(b, i, 2, "big")
|
|
def be24(b, i): return num(b, i, 3, "big")
|
|
def be32(b, i): return num(b, i, 4, "big")
|
|
def be64(b, i): return num(b, i, 8, "big")
|
|
def le16(b, i): return num(b, i, 2, "little")
|
|
def le32(b, i): return num(b, i, 4, "little")
|
|
|
|
|
|
class Cand:
|
|
def __init__(self, id_, kind):
|
|
self.id, self.kind, self.strength = id_, kind, "signature"
|
|
self.evidence, self.tracks, self.caps = [], [], set()
|
|
self.walked_to, self.stopped = None, None
|
|
self.facts = {}
|
|
|
|
def see(self, at, ln, is_):
|
|
if len(self.evidence) < 8:
|
|
self.evidence.append({"at": at, "len": ln, "is": is_})
|
|
|
|
def stop(self, at, why):
|
|
self.stopped = (at, why)
|
|
|
|
def track(self, media, stated, codec=None, track=None, width=None, height=None, channels=None, sample_rate=None, protected=None):
|
|
if codec in CAPS:
|
|
self.caps.add(CAPS[codec])
|
|
if len(self.tracks) < 64:
|
|
self.tracks.append({"track": track, "media": media, "stated": stated, "codec": codec, "width": width, "height": height,
|
|
"channels": channels, "sample_rate": sample_rate, "protected": protected})
|
|
|
|
def value(self, size):
|
|
return {
|
|
"id": self.id, "kind": self.kind, "strength": self.strength, "evidence": self.evidence,
|
|
"walked_to": None if self.walked_to is None else min(self.walked_to, size),
|
|
"stopped": None if self.stopped is None else {"at": min(self.stopped[0], size), "why": self.stopped[1]},
|
|
"facts": self.facts, "tracks": self.tracks, "capabilities": sorted(self.caps),
|
|
}
|
|
|
|
|
|
RANK = {"signature": 0, "structure": 1, "parameters": 2}
|
|
NOTHING = {"scheme": None, "original": None}
|
|
|
|
|
|
# ---- §5.2 ------------------------------------------------------------------------------
|
|
|
|
def wai_envelope(d, head):
|
|
if head[:4] not in (b"WAI1", b"WAI2"):
|
|
return None
|
|
version = head[3] - 0x30
|
|
c = Cand("wai", "wai")
|
|
c.see(0, 4, "magic:WAI%d" % version)
|
|
c.facts["version"] = version
|
|
|
|
def stopped(at, why):
|
|
c.walked_to = at
|
|
c.stop(at, why)
|
|
return c
|
|
ml = le32(head, 4)
|
|
if ml is None:
|
|
return stopped(0, "overrun")
|
|
if ml > HEAD:
|
|
return stopped(8, "limit")
|
|
if 8 + ml > len(d):
|
|
return stopped(8, "overrun")
|
|
try:
|
|
m = json.loads(d[8:8 + ml].decode("utf-8"))
|
|
except (ValueError, UnicodeDecodeError, RecursionError):
|
|
return stopped(8, "invalid")
|
|
c.see(8, ml, "manifest")
|
|
pos = 8 + ml
|
|
if version == 1:
|
|
cap = None
|
|
if isinstance(m, dict) and isinstance(m.get("model_requirement"), dict):
|
|
cap = m["model_requirement"].get("capability")
|
|
c.facts["capability"] = clean(cap.encode("utf-8")) if isinstance(cap, str) else None
|
|
pl = le32(d, pos)
|
|
if pl is None:
|
|
return stopped(pos, "overrun")
|
|
c.see(pos, 4, "payload_length")
|
|
if pos + 4 + pl > len(d):
|
|
return stopped(pos, "overrun")
|
|
pos += 4 + pl
|
|
else:
|
|
n = le16(d, pos)
|
|
if n is None:
|
|
return stopped(pos, "overrun")
|
|
c.see(pos, 2, "rendition_count")
|
|
c.facts["renditions"] = n
|
|
pos += 2
|
|
c.walked_to = pos
|
|
c.strength = "structure"
|
|
return c
|
|
|
|
|
|
# ---- §5.3 ------------------------------------------------------------------------------
|
|
|
|
ISO_FIRST = {b"ftyp", b"styp", b"moov", b"mdat", b"moof", b"free", b"skip", b"wide", b"sidx", b"pnot"}
|
|
|
|
|
|
def iso_children(b):
|
|
out, i = [], 0
|
|
while len(out) < MAX_STEPS:
|
|
s = be32(b, i)
|
|
if s is None or i + 8 > len(b):
|
|
break
|
|
ty = b[i + 4:i + 8]
|
|
if s == 0:
|
|
hdr, ln = 8, len(b) - i
|
|
elif s == 1:
|
|
ln = be64(b, i + 8)
|
|
if ln is None or ln < 16:
|
|
break
|
|
hdr = 16
|
|
elif s < 8:
|
|
break
|
|
else:
|
|
hdr, ln = 8, s
|
|
if i + ln > len(b):
|
|
break
|
|
out.append((ty, b[i + hdr:i + ln]))
|
|
i += ln
|
|
return out
|
|
|
|
|
|
def iso_child(b, ty):
|
|
for t, c in iso_children(b):
|
|
if t == ty:
|
|
return c
|
|
return None
|
|
|
|
|
|
def descr_size(b, i):
|
|
n = 0
|
|
for _ in range(4):
|
|
c = b[i]
|
|
i += 1
|
|
n = (n << 7) | (c & 0x7f)
|
|
if not c & 0x80:
|
|
return i
|
|
raise IndexError
|
|
|
|
|
|
def esds_codec(b):
|
|
try:
|
|
if b[4] != 3:
|
|
return None
|
|
j = descr_size(b, 5)
|
|
flags = b[j + 2]
|
|
j += 3
|
|
if flags & 0x80:
|
|
j += 2
|
|
if flags & 0x40:
|
|
j += 1 + b[j]
|
|
if flags & 0x20:
|
|
j += 2
|
|
if b[j] != 4:
|
|
return None
|
|
k = descr_size(b, j + 1)
|
|
o = b[k]
|
|
except (IndexError, TypeError):
|
|
return None
|
|
if o in (0x40, 0x66, 0x67, 0x68):
|
|
return "aac"
|
|
if o in (0x69, 0x6b):
|
|
return "mp3"
|
|
return None
|
|
|
|
|
|
def iso_track(c, trak):
|
|
track = None
|
|
tkhd = iso_child(trak, b"tkhd")
|
|
if tkhd is not None:
|
|
if len(tkhd) == 0:
|
|
return
|
|
track = be32(tkhd, 12) if tkhd[0] == 0 else be32(tkhd, 20)
|
|
mdia = iso_child(trak, b"mdia")
|
|
if mdia is None:
|
|
return
|
|
hdlr = iso_child(mdia, b"hdlr")
|
|
if hdlr is None or len(hdlr) < 12:
|
|
return
|
|
handler = hdlr[8:12]
|
|
media = {b"vide": "video", b"auxv": "video", b"pict": "video", b"soun": "audio", b"text": "text", b"subt": "text", b"sbtl": "text", b"clcp": "text"}.get(handler, "other")
|
|
minf = iso_child(mdia, b"minf")
|
|
stbl = None if minf is None else iso_child(minf, b"stbl")
|
|
stsd = None if stbl is None else iso_child(stbl, b"stsd")
|
|
if stsd is None or len(stsd) < 8:
|
|
return
|
|
entries = iso_children(stsd[8:])
|
|
if not entries:
|
|
return
|
|
ty, entry = entries[0]
|
|
named = ty
|
|
width = height = channels = rate = protected = None
|
|
inner = b""
|
|
if media == "video":
|
|
width, height = be16(entry, 24), be16(entry, 26)
|
|
inner = entry[78:]
|
|
elif media == "audio":
|
|
channels = be16(entry, 16)
|
|
r = be32(entry, 24)
|
|
rate = None if r is None else r >> 16
|
|
inner = entry[28:] if be16(entry, 8) == 0 else b""
|
|
if ty in (b"encv", b"enca", b"enct", b"encs"):
|
|
protected = dict(NOTHING)
|
|
sinf = iso_child(inner, b"sinf")
|
|
if sinf is not None:
|
|
frma = iso_child(sinf, b"frma")
|
|
schm = iso_child(sinf, b"schm")
|
|
original = None if frma is None else frma[:4]
|
|
if original is not None and len(original) < 4:
|
|
original = None
|
|
scheme = None if schm is None or len(schm) < 8 else clean(schm[4:8])
|
|
protected = {"scheme": scheme, "original": None if original is None else clean(original)}
|
|
if original is not None:
|
|
named = original
|
|
if named == b"mp4a":
|
|
esds = iso_child(inner, b"esds")
|
|
codec = None if esds is None else esds_codec(esds)
|
|
else:
|
|
codec = ISO_CODEC.get(named)
|
|
c.track(media, clean(ty), codec, track, width, height, channels, rate, protected)
|
|
|
|
|
|
def iso(d, head):
|
|
if head[4:8] not in ISO_FIRST:
|
|
return None
|
|
size = len(d)
|
|
c = Cand("isobmff", "container")
|
|
pos = headers = steps = fragments = 0
|
|
boxes = []
|
|
while pos != size:
|
|
if steps == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
h = d[pos:pos + 16]
|
|
if len(h) < 8:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
s, ty = be32(h, 0), h[4:8]
|
|
if not all(0x20 <= x < 0x7f or x == 0xa9 for x in ty):
|
|
c.stop(pos, "invalid")
|
|
break
|
|
if s == 0:
|
|
hdr, ln = 8, size - pos
|
|
elif s == 1:
|
|
ln = be64(h, 8)
|
|
if ln is None:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
if ln < 16:
|
|
c.stop(pos, "invalid")
|
|
break
|
|
hdr = 16
|
|
elif s < 8:
|
|
c.stop(pos, "invalid")
|
|
break
|
|
else:
|
|
hdr, ln = 8, s
|
|
headers += 1
|
|
name = clean(ty)
|
|
c.see(pos + 4, 4, "box:" + name)
|
|
if name not in boxes and len(boxes) < 16:
|
|
boxes.append(name)
|
|
if ty == b"moof":
|
|
fragments += 1
|
|
fits = pos + ln <= size
|
|
if fits and ty == b"ftyp":
|
|
b = d[pos + hdr:pos + hdr + min(ln - hdr, 4096)]
|
|
if len(b) < 8:
|
|
c.stop(pos, "invalid")
|
|
break
|
|
c.facts["major_brand"] = clean(b[:4])
|
|
brands = [b[i:i + 4] for i in range(8, len(b) - 3, 4)]
|
|
c.facts["compatible_brands"] = [clean(x) for x in brands[:16]]
|
|
if any(x in (b"avif", b"avis") for x in brands + [b[:4]]):
|
|
c.caps.add("wai.image.avif")
|
|
elif fits and ty == b"moov":
|
|
if ln > 64 << 20:
|
|
c.facts["moov_not_read"] = True
|
|
else:
|
|
for t, content in iso_children(d[pos + hdr:pos + ln]):
|
|
if t == b"trak":
|
|
iso_track(c, content)
|
|
if not fits:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
pos += ln
|
|
steps += 1
|
|
c.walked_to = pos
|
|
c.facts["boxes"] = boxes
|
|
if fragments:
|
|
c.facts["fragments"] = fragments
|
|
if headers >= 2:
|
|
c.strength = "structure"
|
|
return c
|
|
|
|
|
|
# ---- §5.4 ------------------------------------------------------------------------------
|
|
|
|
def leading(x):
|
|
return 8 - x.bit_length()
|
|
|
|
|
|
def element(b):
|
|
"""(id, size or None for unknown, header length), or None."""
|
|
if not b:
|
|
return None
|
|
idn = leading(b[0]) + 1
|
|
if idn > 4 or len(b) <= idn:
|
|
return None
|
|
id_ = int.from_bytes(b[:idn], "big")
|
|
sn = leading(b[idn]) + 1
|
|
if sn > 8 or len(b) < idn + sn:
|
|
return None
|
|
mask = 0xff >> sn
|
|
first = b[idn] & mask
|
|
rest = b[idn + 1:idn + sn]
|
|
size = (first << (8 * len(rest))) | int.from_bytes(rest, "big")
|
|
unknown = first == mask and all(x == 0xff for x in rest)
|
|
return id_, None if unknown else size, idn + sn
|
|
|
|
|
|
def ebml_children(b):
|
|
out, i = [], 0
|
|
while len(out) < MAX_STEPS:
|
|
e = element(b[i:i + 12])
|
|
if e is None or e[1] is None:
|
|
break
|
|
if i + e[2] + e[1] > len(b):
|
|
break
|
|
out.append((e[0], b[i + e[2]:i + e[2] + e[1]]))
|
|
i += e[2] + e[1]
|
|
return out
|
|
|
|
|
|
def uint(b):
|
|
return int.from_bytes(b, "big") if 1 <= len(b) <= 8 else None
|
|
|
|
|
|
def toward_zero(x):
|
|
if x != x or x <= 0:
|
|
return 0
|
|
return U64 if x >= U64 else int(x)
|
|
|
|
|
|
|
|
|
|
def mkv_track(c, entry):
|
|
t = dict(media="other", stated="", codec=None, track=None, width=None, height=None, channels=None, sample_rate=None, protected=None)
|
|
for id_, b in ebml_children(entry):
|
|
if id_ == 0xd7:
|
|
t["track"] = uint(b)
|
|
elif id_ == 0x83:
|
|
t["media"] = {1: "video", 2: "audio", 17: "text"}.get(uint(b), "other")
|
|
elif id_ == 0x86:
|
|
t["stated"] = clean(b)
|
|
t["codec"] = "aac" if b.startswith(b"A_AAC") else MKV_CODEC.get(b)
|
|
elif id_ == 0xe0:
|
|
for k, v in ebml_children(b):
|
|
if k == 0xb0:
|
|
t["width"] = uint(v)
|
|
elif k == 0xba:
|
|
t["height"] = uint(v)
|
|
elif id_ == 0xe1:
|
|
for k, v in ebml_children(b):
|
|
if k == 0xb5:
|
|
t["sample_rate"] = toward_zero(struct.unpack(">f", v)[0]) if len(v) == 4 else toward_zero(struct.unpack(">d", v)[0]) if len(v) == 8 else None
|
|
elif k == 0x9f:
|
|
t["channels"] = uint(v)
|
|
elif id_ == 0x6d80:
|
|
if any(k == 0x6240 and any(j == 0x5035 for j, _ in ebml_children(e)) for k, e in ebml_children(b)):
|
|
t["protected"] = dict(NOTHING)
|
|
c.track(t["media"], t["stated"], t["codec"], t["track"], t["width"], t["height"], t["channels"], t["sample_rate"], t["protected"])
|
|
|
|
|
|
def matroska(d, head):
|
|
if head[:4] != b"\x1a\x45\xdf\xa3":
|
|
return None
|
|
size = len(d)
|
|
c = Cand("matroska", "container")
|
|
c.see(0, 4, "ebml")
|
|
|
|
def stopped(at, why):
|
|
c.walked_to = at
|
|
c.stop(at, why)
|
|
return c
|
|
e = element(head)
|
|
if e is None or e[1] is None:
|
|
return stopped(0, "invalid")
|
|
_, hsize, hdr = e
|
|
if hdr + hsize > len(head):
|
|
return stopped(0, "overrun")
|
|
for id_, b in ebml_children(head[hdr:hdr + hsize]):
|
|
if id_ == 0x4282:
|
|
c.facts["doc_type"] = clean(b)
|
|
elif id_ == 0x4287:
|
|
c.facts["doc_type_version"] = uint(b)
|
|
pos = hdr + hsize
|
|
seg = d[pos:pos + 12]
|
|
e = element(seg)
|
|
if e is None or e[0] != 0x18538067:
|
|
return stopped(pos, "overrun" if len(seg) < 5 else "invalid")
|
|
c.see(pos, 4, "segment")
|
|
start = pos + e[2]
|
|
end = size if e[1] is None else start + e[1]
|
|
if end > size:
|
|
c.facts["segment_overruns_by"] = end - size
|
|
end = min(end, size)
|
|
pos = start
|
|
read = clusters = 0
|
|
while pos < end:
|
|
if read == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
h = d[pos:pos + 12]
|
|
e = element(h)
|
|
if e is None:
|
|
c.stop(pos, "overrun" if len(h) < 12 else "invalid")
|
|
break
|
|
id_, s, hl = e
|
|
if s is None:
|
|
c.stop(pos, "unknown_size")
|
|
break
|
|
nxt = pos + hl + s
|
|
if nxt > size:
|
|
if id_ == 0x1f43b675:
|
|
clusters += 1
|
|
c.stop(pos, "overrun")
|
|
break
|
|
if id_ == 0x1654ae6b and s <= HEAD:
|
|
c.see(pos, 4, "tracks")
|
|
for k, entry in ebml_children(d[pos + hl:pos + hl + s]):
|
|
if k == 0xae:
|
|
mkv_track(c, entry)
|
|
if id_ == 0x1f43b675:
|
|
clusters += 1
|
|
read += 1
|
|
pos = nxt
|
|
c.walked_to = pos
|
|
if clusters:
|
|
c.facts["clusters"] = clusters
|
|
if read:
|
|
c.strength = "structure"
|
|
return c
|
|
|
|
|
|
# ---- §5.5 ------------------------------------------------------------------------------
|
|
|
|
def crc_msb(b, init):
|
|
c = init
|
|
for x in b:
|
|
c ^= x << 24
|
|
for _ in range(8):
|
|
c = ((c << 1) ^ 0x04c11db7) & 0xffffffff if c & 0x80000000 else (c << 1) & 0xffffffff
|
|
return c
|
|
|
|
|
|
|
|
|
|
def has_ca(d):
|
|
j = 0
|
|
while j + 2 <= len(d):
|
|
if d[j] == 9:
|
|
return True
|
|
j += 2 + d[j + 1]
|
|
return False
|
|
|
|
|
|
def transport(head, size):
|
|
layout = None
|
|
for psize, sync in ((188, 0), (192, 4), (204, 0)):
|
|
for start in range(psize):
|
|
if all(start + k * psize + sync < len(head) and head[start + k * psize + sync] == 0x47 for k in range(5)):
|
|
layout = (psize, sync, start)
|
|
break
|
|
if layout:
|
|
break
|
|
if not layout:
|
|
return None
|
|
psize, sync, start = layout
|
|
c = Cand("mpeg-ts", "container")
|
|
c.see(start + sync, 1, "sync")
|
|
c.facts["packet_size"] = psize
|
|
packets = scrambled = 0
|
|
programs, gather, pmt_done, streams = {}, {}, {}, []
|
|
pat_done = False
|
|
pos = start
|
|
stopped = None
|
|
while True:
|
|
p = head[pos + sync:pos + sync + 188]
|
|
if len(p) < 188:
|
|
break
|
|
if p[0] != 0x47:
|
|
stopped = pos + sync
|
|
break
|
|
packets += 1
|
|
pid = ((p[1] & 0x1f) << 8) | p[2]
|
|
if p[3] & 0xc0:
|
|
scrambled += 1
|
|
afc = (p[3] >> 4) & 3
|
|
i = 4
|
|
if afc & 2:
|
|
i += 1 + p[4]
|
|
if afc & 1 and i < 188 and (pid == 0 or pid in programs.values()):
|
|
if p[1] & 0x40:
|
|
i += 1 + p[i]
|
|
if i < 188:
|
|
gather[pid] = [b"", pos + sync + i]
|
|
if pid in gather and i < 188:
|
|
g = gather[pid]
|
|
g[0] += p[i:]
|
|
if len(g[0]) >= 3:
|
|
n = 3 + (((g[0][1] & 0x0f) << 8) | g[0][2])
|
|
if len(g[0]) >= n:
|
|
sec, at = g[0][:n], g[1]
|
|
del gather[pid]
|
|
if n >= 12 and crc_msb(sec, 0xffffffff) == 0:
|
|
if pid == 0 and sec[0] == 0 and not pat_done:
|
|
pat_done = True
|
|
c.see(at, n, "pat")
|
|
entries = sec[8:n - 4]
|
|
for k in range(0, min(len(entries) // 4, 64) * 4, 4):
|
|
number = be16(entries, k)
|
|
mapped = ((entries[k + 2] & 0x1f) << 8) | entries[k + 3]
|
|
if number != 0:
|
|
programs[number] = mapped
|
|
pmt_done[mapped] = False
|
|
elif sec[0] == 2 and pmt_done.get(pid) is False:
|
|
pmt_done[pid] = True
|
|
c.see(at, n, "pmt")
|
|
j = 12 + (((sec[10] & 0x0f) << 8) | sec[11])
|
|
program_ca = j <= len(sec) and has_ca(sec[12:j])
|
|
while j + 5 <= n - 4:
|
|
dl = ((sec[j + 3] & 0x0f) << 8) | sec[j + 4]
|
|
ca = program_ca or (j + 5 + dl <= len(sec) and has_ca(sec[j + 5:j + 5 + dl]))
|
|
streams.append((((sec[j + 1] & 0x1f) << 8) | sec[j + 2], sec[j], ca))
|
|
j += 5 + dl
|
|
pos += psize
|
|
for es, ty, ca in streams:
|
|
codec, media = TS_TYPE.get(ty, (None, "other"))
|
|
c.track(media, "stream_type:0x%02x" % ty, codec, es, protected=dict(NOTHING) if ca else None)
|
|
c.facts["packets_read"] = packets
|
|
c.facts["scrambled_packets"] = scrambled
|
|
c.facts["programs"] = len(programs)
|
|
c.walked_to = pos
|
|
if stopped is not None:
|
|
c.stop(stopped, "invalid")
|
|
elif pos < size and len(head) == size:
|
|
c.stop(pos, "overrun")
|
|
elif len(head) < size:
|
|
c.stop(pos, "limit")
|
|
c.strength = "structure"
|
|
return c
|
|
|
|
|
|
# ---- §5.6 ------------------------------------------------------------------------------
|
|
|
|
def ogg(d, head):
|
|
if head[:5] != b"OggS\x00":
|
|
return None
|
|
size = len(d)
|
|
c = Cand("ogg", "container")
|
|
pos = pages = 0
|
|
while pos != size:
|
|
if pages == 256:
|
|
c.stop(pos, "limit")
|
|
break
|
|
h = d[pos:pos + 27]
|
|
if len(h) < 27:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
if h[:5] != b"OggS\x00":
|
|
c.stop(pos, "invalid")
|
|
break
|
|
nseg = h[26]
|
|
table = d[pos + 27:pos + 27 + nseg]
|
|
if len(table) < nseg:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
n = sum(table)
|
|
body = d[pos + 27 + nseg:pos + 27 + nseg + n]
|
|
if len(body) < n:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
if crc_msb(h[:22] + b"\0\0\0\0" + h[26:] + table + body, 0) != le32(h, 22):
|
|
c.stop(pos, "crc")
|
|
break
|
|
c.see(pos, 27 + nseg + n, "page")
|
|
if h[5] & 2:
|
|
serial = le32(h, 14)
|
|
if body.startswith(b"OpusHead"):
|
|
c.track("audio", "OpusHead", "opus", serial, channels=body[9] if len(body) > 9 else None, sample_rate=le32(body, 12))
|
|
elif body.startswith(b"\x01vorbis"):
|
|
c.track("audio", "vorbis", "vorbis", serial, channels=body[11] if len(body) > 11 else None, sample_rate=le32(body, 12))
|
|
elif body.startswith(b"\x7fFLAC"):
|
|
c.track("audio", "FLAC", "flac", serial)
|
|
elif body.startswith(b"\x80theora"):
|
|
c.track("video", "theora", "theora", serial)
|
|
elif body.startswith(b"Speex "):
|
|
c.track("audio", "Speex", "speex", serial)
|
|
else:
|
|
c.track("other", clean(body[:8]), None, serial)
|
|
pos += 27 + nseg + n
|
|
pages += 1
|
|
c.walked_to = pos
|
|
c.facts["pages_read"] = pages
|
|
if pages >= 2 or (pages == 1 and pos == size):
|
|
c.strength = "structure"
|
|
return c
|
|
|
|
|
|
# ---- §5.7 ------------------------------------------------------------------------------
|
|
|
|
def after_id3(head):
|
|
"""Where the bytes after an ID3v2 tag begin: 0 with none, None with a header that is not one."""
|
|
if head[:3] != b"ID3":
|
|
return 0
|
|
if len(head) < 10 or head[3] == 0xff or head[4] == 0xff or any(x & 0x80 for x in head[6:10]):
|
|
return None
|
|
n = 0
|
|
for x in head[6:10]:
|
|
n = (n << 7) | x
|
|
return 10 + n + (10 if head[5] & 0x10 else 0)
|
|
|
|
|
|
def flac(d, head):
|
|
start = after_id3(head)
|
|
if start is None or d[start:start + 4] != b"fLaC":
|
|
return None
|
|
size = len(d)
|
|
c = Cand("flac", "audio")
|
|
c.caps.add("wai.audio.flac")
|
|
c.see(start, 4, "fLaC")
|
|
pos = start + 4
|
|
blocks = 0
|
|
info = False
|
|
while True:
|
|
if blocks == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
h = d[pos:pos + 4]
|
|
if len(h) < 4:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
last, ty, ln = h[0] & 0x80, h[0] & 0x7f, be24(h, 1)
|
|
if blocks == 0:
|
|
if ty != 0 or ln != 34:
|
|
c.stop(pos, "invalid")
|
|
break
|
|
b = d[pos + 4:pos + 38]
|
|
if len(b) < 34:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
c.see(pos + 4, 34, "streaminfo")
|
|
c.facts["sample_rate"] = (b[10] << 12) | (b[11] << 4) | (b[12] >> 4)
|
|
c.facts["channels"] = ((b[12] >> 1) & 7) + 1
|
|
c.facts["bits_per_sample"] = (((b[12] & 1) << 4) | (b[13] >> 4)) + 1
|
|
c.facts["samples"] = ((b[13] & 0x0f) << 32) | be32(b, 14)
|
|
info = True
|
|
if pos + 4 + ln > size:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
pos += 4 + ln
|
|
blocks += 1
|
|
if last:
|
|
f = d[pos:pos + 2]
|
|
if len(f) == 2 and f[0] == 0xff and f[1] & 0xfe == 0xf8:
|
|
c.see(pos, 2, "frame_sync")
|
|
if info:
|
|
c.strength = "structure"
|
|
elif f:
|
|
c.stop(pos, "invalid")
|
|
else:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
c.walked_to = pos
|
|
return c
|
|
|
|
|
|
# ---- §5.8 ------------------------------------------------------------------------------
|
|
|
|
def riff(d, head):
|
|
if head[:4] != b"RIFF" or len(head) < 12:
|
|
return None
|
|
form = head[8:12]
|
|
id_, kind = {b"WAVE": ("wave", "audio"), b"AVI ": ("avi", "container"), b"WEBP": ("webp", "image")}.get(form, ("riff", "container"))
|
|
size = len(d)
|
|
c = Cand(id_, kind)
|
|
c.see(8, 4, "form:" + clean(form))
|
|
c.facts["form"] = clean(form)
|
|
declared = le32(head, 4) + 8
|
|
end = min(declared, size)
|
|
if declared != size:
|
|
c.facts["declared_size"] = declared
|
|
pos, chunks = 12, 0
|
|
names = []
|
|
while True:
|
|
if pos + 8 > end:
|
|
if pos < end:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
if chunks == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
ty, ln = d[pos:pos + 4], le32(d, pos + 4)
|
|
name = clean(ty)
|
|
c.see(pos, 4, "chunk:" + name)
|
|
if name not in names and len(names) < 16:
|
|
names.append(name)
|
|
b = d[pos + 8:pos + 8 + 64]
|
|
if id_ == "wave" and ty == b"fmt ":
|
|
b = b[:min(ln, 64)]
|
|
tag = le16(b, 0)
|
|
if tag == 0xfffe:
|
|
tag = le16(b, 24)
|
|
c.facts.update(format_tag=tag, channels=le16(b, 2), sample_rate=le32(b, 4), bits_per_sample=le16(b, 14))
|
|
elif id_ == "wave" and ty == b"data":
|
|
c.facts["data_bytes"] = ln
|
|
elif id_ == "webp" and ty == b"VP8 ":
|
|
b = b[:10]
|
|
if b[3:6] == b"\x9d\x01\x2a":
|
|
w, h = le16(b, 6), le16(b, 8)
|
|
c.facts.update(coding="lossy", width=None if w is None else w & 0x3fff, height=None if h is None else h & 0x3fff)
|
|
elif id_ == "webp" and ty == b"VP8L":
|
|
b = b[:5]
|
|
if b[:1] == b"\x2f" and len(b) == 5:
|
|
v = le32(b, 1)
|
|
c.facts.update(coding="lossless", width=(v & 0x3fff) + 1, height=((v >> 14) & 0x3fff) + 1)
|
|
elif id_ == "webp" and ty == b"VP8X":
|
|
b = b[:10]
|
|
if len(b) == 10:
|
|
c.facts.update(coding="extended", width=int.from_bytes(b[4:7], "little") + 1, height=int.from_bytes(b[7:10], "little") + 1)
|
|
if pos + 8 + ln > end:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
pos += 8 + ln + (ln & 1)
|
|
chunks += 1
|
|
c.walked_to = min(pos, end)
|
|
c.facts["chunks"] = names
|
|
if chunks >= 1:
|
|
c.strength = "structure"
|
|
return c
|
|
|
|
|
|
# ---- §5.9, §5.10 -----------------------------------------------------------------------
|
|
|
|
|
|
|
|
def mpeg_frame(b):
|
|
if len(b) < 4 or b[0] != 0xff or b[1] & 0xe0 != 0xe0:
|
|
return None
|
|
version, layer, bi, ri = (b[1] >> 3) & 3, (b[1] >> 1) & 3, b[2] >> 4, (b[2] >> 2) & 3
|
|
if version == 1 or layer == 0 or bi in (0, 15) or ri == 3:
|
|
return None
|
|
kbps = BITRATES[(3 if version == 3 else 2, layer)][bi - 1]
|
|
rate = (44100, 48000, 32000)[ri] >> {3: 0, 2: 1, 0: 2}[version]
|
|
pad = (b[2] >> 1) & 1
|
|
if layer == 3:
|
|
ln = (12000 * kbps // rate + pad) * 4
|
|
elif version != 3 and layer == 1:
|
|
ln = 72000 * kbps // rate + pad
|
|
else:
|
|
ln = 144000 * kbps // rate + pad
|
|
return version, layer, kbps, rate, b[3] >> 6, ln
|
|
|
|
|
|
def mpeg_audio(head, size):
|
|
start = after_id3(head)
|
|
if start is None or start >= len(head):
|
|
return None
|
|
first = mpeg_frame(head[start:start + 4])
|
|
if first is None:
|
|
return None
|
|
c = Cand("mpeg-audio", "audio")
|
|
if start:
|
|
c.see(0, 3, "id3v2")
|
|
c.see(start, 4, "frame_header")
|
|
c.facts.update(version={3: "1", 2: "2", 0: "2.5"}[first[0]], layer=4 - first[1], bitrate_kbps=first[2], sample_rate=first[3], channels=1 if first[4] == 3 else 2)
|
|
pos, frames = start, 0
|
|
while True:
|
|
if frames == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
left = len(head) - pos
|
|
if left <= 0:
|
|
if pos < size:
|
|
c.stop(pos, "limit")
|
|
break
|
|
if left == 128 and head[pos:pos + 3] == b"TAG":
|
|
c.see(pos, 3, "id3v1")
|
|
pos += 128
|
|
break
|
|
f = mpeg_frame(head[pos:pos + 4])
|
|
if f is None or f[:2] != first[:2] or f[3] != first[3]:
|
|
c.stop(pos, "overrun" if left < 4 else "invalid")
|
|
break
|
|
if pos + f[5] > size:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
if f[5] > left:
|
|
c.stop(pos, "limit")
|
|
break
|
|
pos += f[5]
|
|
frames += 1
|
|
c.walked_to = pos
|
|
c.facts["frames_read"] = frames
|
|
if frames >= 3:
|
|
c.strength = "structure"
|
|
return c if frames else None
|
|
|
|
|
|
|
|
|
|
def adts_frame(h):
|
|
if len(h) < 7 or h[0] != 0xff or h[1] & 0xf6 != 0xf0:
|
|
return None
|
|
ri = (h[2] >> 2) & 0x0f
|
|
ln = ((h[3] & 3) << 11) | (h[4] << 3) | (h[5] >> 5)
|
|
if ri >= len(ADTS_RATES) or ln < 7:
|
|
return None
|
|
return (h[2] >> 6) + 1, ADTS_RATES[ri], ((h[2] & 1) << 2) | (h[3] >> 6), ln
|
|
|
|
|
|
def adts(head, size):
|
|
first = adts_frame(head[:7])
|
|
if first is None:
|
|
return None
|
|
c = Cand("adts", "audio")
|
|
c.caps.add("wai.audio.aac")
|
|
c.see(0, 7, "frame_header")
|
|
c.facts.update(audio_object_type=first[0], sample_rate=first[1], channel_configuration=first[2])
|
|
pos, frames = 0, 0
|
|
while True:
|
|
if frames == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
left = len(head) - pos
|
|
if left == 0:
|
|
if pos < size:
|
|
c.stop(pos, "limit")
|
|
break
|
|
f = adts_frame(head[pos:pos + 7])
|
|
if f is None or f[:2] != first[:2]:
|
|
c.stop(pos, "overrun" if left < 7 else "invalid")
|
|
break
|
|
if pos + f[3] > size:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
if f[3] > left:
|
|
c.stop(pos, "limit")
|
|
break
|
|
pos += f[3]
|
|
frames += 1
|
|
c.walked_to = pos
|
|
c.facts["frames_read"] = frames
|
|
if frames >= 3:
|
|
c.strength = "structure"
|
|
return c if frames else None
|
|
|
|
|
|
# ---- §5.11 to §5.13 --------------------------------------------------------------------
|
|
|
|
def crc_png(b):
|
|
c = 0xffffffff
|
|
for x in b:
|
|
c ^= x
|
|
for _ in range(8):
|
|
c = (c >> 1) ^ 0xedb88320 if c & 1 else c >> 1
|
|
return c ^ 0xffffffff
|
|
|
|
|
|
def png(d, head):
|
|
if head[:8] != b"\x89PNG\r\n\x1a\n":
|
|
return None
|
|
size = len(d)
|
|
c = Cand("png", "image")
|
|
c.caps.add("wai.image.png")
|
|
c.see(0, 8, "signature")
|
|
pos, chunks = 8, 0
|
|
while True:
|
|
if chunks == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
if pos == size:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
h = d[pos:pos + 8]
|
|
if len(h) < 8:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
ln, ty = be32(h, 0), h[4:8]
|
|
if not all(0x41 <= x <= 0x5a or 0x61 <= x <= 0x7a for x in ty):
|
|
c.stop(pos, "invalid")
|
|
break
|
|
if chunks == 0:
|
|
if ty != b"IHDR" or ln != 13:
|
|
c.stop(pos, "invalid")
|
|
break
|
|
b = d[pos + 4:pos + 25]
|
|
if len(b) < 21:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
if crc_png(b[:17]) != be32(b, 17):
|
|
c.stop(pos, "crc")
|
|
break
|
|
c.see(pos, 25, "ihdr")
|
|
c.facts.update(width=be32(b, 4), height=be32(b, 8), bit_depth=b[12], colour_type=b[13], interlace=b[16])
|
|
c.strength = "structure"
|
|
if ty == b"acTL":
|
|
c.facts["animated"] = True
|
|
if ty == b"cICP":
|
|
b = d[pos + 8:pos + 12]
|
|
if len(b) == 4:
|
|
c.facts["cicp"] = list(b)
|
|
if pos + 12 + ln > size:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
pos += 12 + ln
|
|
chunks += 1
|
|
if ty == b"IEND":
|
|
c.see(pos - 12 - ln, 12, "iend")
|
|
break
|
|
c.walked_to = pos
|
|
c.facts["chunks_read"] = chunks
|
|
return c
|
|
|
|
|
|
APPS = [(b"JFIF\0", "jfif"), (b"Exif\0\0", "exif"), (b"ICC_PROFILE\0", "icc"), (b"MPF\0", "mpf"), (b"http://ns.adobe.com/xap/1.0/\0", "xmp"), (b"Adobe", "adobe")] # allow-name: the APP14 segment's identifier and the XMP namespace, byte signatures a JPEG carries
|
|
SOF = set(range(0xc0, 0xc4)) | set(range(0xc5, 0xc8)) | set(range(0xc9, 0xcc)) | set(range(0xcd, 0xd0))
|
|
|
|
|
|
def jpeg(d, head):
|
|
if head[:3] != b"\xff\xd8\xff":
|
|
return None
|
|
size = len(d)
|
|
c = Cand("jpeg", "image")
|
|
c.caps.add("wai.image.jpeg")
|
|
c.see(0, 2, "soi")
|
|
pos, markers = 2, 0
|
|
apps = []
|
|
while True:
|
|
if markers == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
h = d[pos:pos + 4]
|
|
if len(h) < 2:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
if h[0] != 0xff:
|
|
c.stop(pos, "invalid")
|
|
break
|
|
m = h[1]
|
|
if m == 0xff:
|
|
pos += 1
|
|
markers += 1
|
|
continue
|
|
if m == 0xda:
|
|
c.see(pos, 2, "sos")
|
|
break
|
|
if len(h) < 4:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
ln = be16(h, 2)
|
|
if ln < 2:
|
|
c.stop(pos, "invalid")
|
|
break
|
|
if 0xe0 <= m <= 0xef:
|
|
b = d[pos + 4:pos + 4 + min(ln - 2, 32)]
|
|
name = next((n for p, n in APPS if b.startswith(p)), "other")
|
|
if name not in apps and len(apps) < 8:
|
|
apps.append(name)
|
|
if m in SOF:
|
|
b = d[pos + 4:pos + 10]
|
|
if len(b) < 6:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
c.see(pos, 2 + ln, "sof")
|
|
process = {0xc0: "baseline", 0xc1: "extended", 0xc2: "progressive", 0xc3: "lossless", 0xc9: "arithmetic", 0xca: "arithmetic", 0xcb: "arithmetic"}.get(m, "hierarchical")
|
|
c.facts.update(process=process, precision=b[0], height=be16(b, 1), width=be16(b, 3), components=b[5])
|
|
c.strength = "structure"
|
|
if pos + 2 + ln > size:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
pos += 2 + ln
|
|
markers += 1
|
|
c.walked_to = pos
|
|
c.facts["applications"] = apps
|
|
return c
|
|
|
|
|
|
def gif(head, size):
|
|
if head[:6] not in (b"GIF87a", b"GIF89a"):
|
|
return None
|
|
c = Cand("gif", "image")
|
|
c.see(0, 6, "signature:" + clean(head[:6]))
|
|
|
|
def cut(at):
|
|
c.stop(at, "limit" if len(head) < size else "overrun")
|
|
if len(head) < 13:
|
|
c.walked_to = 6
|
|
cut(6)
|
|
return c
|
|
c.facts["width"], c.facts["height"] = le16(head, 6), le16(head, 8)
|
|
pos = 13 + (3 << ((head[10] & 7) + 1) if head[10] & 0x80 else 0)
|
|
|
|
def blocks(p):
|
|
while True:
|
|
if p >= len(head):
|
|
return None
|
|
n = head[p]
|
|
p += 1 + n
|
|
if n == 0:
|
|
return p
|
|
frames = steps = 0
|
|
ended = False
|
|
while True:
|
|
if steps == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
if pos >= len(head):
|
|
cut(pos)
|
|
break
|
|
b = head[pos]
|
|
if b == 0x3b:
|
|
c.see(pos, 1, "trailer")
|
|
pos += 1
|
|
ended = True
|
|
break
|
|
if b == 0x21:
|
|
p = blocks(pos + 2)
|
|
if p is None:
|
|
cut(pos)
|
|
break
|
|
pos = p
|
|
elif b == 0x2c:
|
|
desc = head[pos + 1:pos + 10]
|
|
if len(desc) < 9:
|
|
cut(pos)
|
|
break
|
|
table = 3 << ((desc[8] & 7) + 1) if desc[8] & 0x80 else 0
|
|
p = blocks(pos + 10 + table + 1)
|
|
if p is None:
|
|
cut(pos)
|
|
break
|
|
if frames == 0:
|
|
c.see(pos, 10, "image")
|
|
frames += 1
|
|
pos = p
|
|
else:
|
|
c.stop(pos, "invalid")
|
|
break
|
|
steps += 1
|
|
c.walked_to = pos
|
|
c.facts["frames_read"] = frames
|
|
if frames >= 1 or ended:
|
|
c.strength = "structure"
|
|
return c
|
|
|
|
|
|
# ---- §5.14 to §5.17 --------------------------------------------------------------------
|
|
|
|
def ivf(d, head):
|
|
if head[:4] != b"DKIF":
|
|
return None
|
|
size = len(d)
|
|
c = Cand("ivf", "container")
|
|
c.see(0, 4, "dkif")
|
|
if len(head) < 12:
|
|
c.walked_to = 0
|
|
c.stop(0, "overrun")
|
|
return c
|
|
hdr, fourcc = le16(head, 6), head[8:12]
|
|
if hdr < 32 or le16(head, 4) != 0:
|
|
c.walked_to = 0
|
|
c.stop(0, "invalid")
|
|
return c
|
|
c.track("video", clean(fourcc), {b"AV01": "av1", b"VP90": "vp9", b"VP80": "vp8"}.get(fourcc), width=le16(head, 12), height=le16(head, 14))
|
|
pos, frames = hdr, 0
|
|
while pos != size:
|
|
if frames == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
h = d[pos:pos + 12]
|
|
if len(h) < 12:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
n = le32(h, 0)
|
|
if pos + 12 + n > size:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
pos += 12 + n
|
|
frames += 1
|
|
c.walked_to = pos
|
|
c.facts["frames_read"] = frames
|
|
if frames >= 1:
|
|
c.strength = "structure"
|
|
return c
|
|
|
|
|
|
def decimal(t):
|
|
if not re.fullmatch(rb"\+?[0-9]+", t):
|
|
return None
|
|
v = int(t)
|
|
return v if v <= U64 else None
|
|
|
|
|
|
def y4m_frame(tag, w, h):
|
|
try:
|
|
tag = tag.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return None
|
|
base, depth = tag, 8
|
|
if "p" in tag:
|
|
b, _, dd = tag.partition("p")
|
|
if dd and re.fullmatch("[0-9]+", dd):
|
|
base, depth = b, int(dd)
|
|
if depth > U64:
|
|
return None
|
|
cw, ch = (w + 1) // 2, (h + 1) // 2
|
|
if base in ("420", "420jpeg", "420mpeg2", "420paldv"):
|
|
n = w * h + 2 * cw * ch
|
|
elif base == "422":
|
|
n = w * h + 2 * cw * h
|
|
elif base == "444":
|
|
n = 3 * w * h
|
|
elif base == "mono":
|
|
n = w * h
|
|
else:
|
|
return None
|
|
n *= 2 if depth > 8 else 1
|
|
return n if n <= U64 else None
|
|
|
|
|
|
def y4m(d, head):
|
|
if head[:10] != b"YUV4MPEG2 ":
|
|
return None
|
|
size = len(d)
|
|
c = Cand("y4m", "elementary")
|
|
c.see(0, 10, "signature")
|
|
nl = head[:256].find(b"\n")
|
|
if nl < 0:
|
|
c.walked_to = 0
|
|
c.stop(0, "overrun" if len(head) == size and len(head) < 256 else "invalid")
|
|
return c
|
|
w = h = None
|
|
colour = b"420"
|
|
for tok in head[10:nl].split(b" "):
|
|
if not tok:
|
|
continue
|
|
if tok[:1] == b"W":
|
|
w = decimal(tok[1:])
|
|
elif tok[:1] == b"H":
|
|
h = decimal(tok[1:])
|
|
elif tok[:1] == b"C":
|
|
colour = tok[1:]
|
|
elif tok[:1] == b"F":
|
|
c.facts["frame_rate"] = clean(tok[1:])
|
|
c.facts.update(width=w, height=h, colour=clean(colour))
|
|
pos = nl + 1
|
|
frame = None if w is None or h is None else y4m_frame(colour, w, h)
|
|
if frame is None:
|
|
c.walked_to = pos
|
|
c.stop(pos, "invalid")
|
|
return c
|
|
frames = 0
|
|
while pos != size:
|
|
if frames == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
line = d[pos:pos + 256]
|
|
if not line.startswith(b"FRAME"):
|
|
c.stop(pos, "overrun" if len(line) < 5 else "invalid")
|
|
break
|
|
e = line.find(b"\n")
|
|
if e < 0:
|
|
c.stop(pos, "invalid")
|
|
break
|
|
nxt = pos + e + 1 + frame
|
|
if nxt > size:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
pos = nxt
|
|
frames += 1
|
|
c.walked_to = pos
|
|
c.facts["frames_read"] = frames
|
|
if frames >= 1:
|
|
c.strength = "structure"
|
|
return c
|
|
|
|
|
|
def flv(d, head):
|
|
if head[:4] != b"FLV\x01":
|
|
return None
|
|
size = len(d)
|
|
c = Cand("flv", "container")
|
|
c.see(0, 4, "signature")
|
|
data = be32(head, 5)
|
|
if data is None or data < 9:
|
|
c.walked_to = 0
|
|
c.stop(0, "overrun" if len(head) < 9 else "invalid")
|
|
return c
|
|
pos = data + 4
|
|
tags = 0
|
|
audio = video = None
|
|
while pos != size:
|
|
if tags == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
h = d[pos:pos + 16]
|
|
if len(h) < 11:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
ty = h[0] & 0x1f
|
|
if ty not in (8, 9, 18):
|
|
c.stop(pos, "invalid")
|
|
break
|
|
n = be24(h, 1)
|
|
if ty == 8 and audio is None and n > 0 and len(h) > 11:
|
|
fmt = h[11] >> 4
|
|
audio = ("sound_format:%d" % fmt, {2: "mp3", 10: "aac"}.get(fmt))
|
|
if ty == 9 and video is None and n > 0 and len(h) > 11:
|
|
if h[11] & 0x80:
|
|
if len(h) == 16:
|
|
video = (clean(h[12:16]), {b"avc1": "avc", b"hvc1": "hevc", b"av01": "av1", b"vp09": "vp9"}.get(h[12:16]))
|
|
else:
|
|
video = ("codec_id:%d" % (h[11] & 0x0f), "avc" if h[11] & 0x0f == 7 else None)
|
|
p = be32(d, pos + 11 + n)
|
|
if p is None:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
if p != 11 + n:
|
|
c.stop(pos, "invalid")
|
|
break
|
|
pos += 15 + n
|
|
tags += 1
|
|
c.walked_to = pos
|
|
c.facts["tags_read"] = tags
|
|
if video:
|
|
c.track("video", video[0], video[1])
|
|
if audio:
|
|
c.track("audio", audio[0], audio[1])
|
|
if tags >= 1:
|
|
c.strength = "structure"
|
|
return c
|
|
|
|
|
|
class Bits:
|
|
def __init__(self, b):
|
|
self.b, self.at = b, 0
|
|
|
|
def u(self, n):
|
|
v = 0
|
|
for _ in range(n):
|
|
if self.at >= 8 * len(self.b):
|
|
raise IndexError
|
|
v = (v << 1) | ((self.b[self.at // 8] >> (7 - self.at % 8)) & 1)
|
|
self.at += 1
|
|
return v
|
|
|
|
def ue(self):
|
|
z = 0
|
|
while self.u(1) == 0:
|
|
z += 1
|
|
if z > 31:
|
|
raise IndexError
|
|
return 0 if z == 0 else (1 << z) - 1 + self.u(z)
|
|
|
|
|
|
def av1_sequence(b):
|
|
r = Bits(b)
|
|
try:
|
|
profile = r.u(3)
|
|
r.u(1)
|
|
if r.u(1):
|
|
level = r.u(5)
|
|
else:
|
|
model, delay_bits = False, 0
|
|
if r.u(1):
|
|
r.u(32)
|
|
r.u(32)
|
|
if r.u(1):
|
|
z = 0
|
|
while r.u(1) == 0:
|
|
z += 1
|
|
if z >= 32:
|
|
return None
|
|
r.u(z)
|
|
if r.u(1):
|
|
model = True
|
|
delay_bits = r.u(5) + 1
|
|
r.u(32)
|
|
r.u(5)
|
|
r.u(5)
|
|
delay = r.u(1)
|
|
level = None
|
|
for _ in range(r.u(5) + 1):
|
|
r.u(12)
|
|
k = r.u(5)
|
|
if level is None:
|
|
level = k
|
|
if k > 7:
|
|
r.u(1)
|
|
if model and r.u(1):
|
|
r.u(delay_bits)
|
|
r.u(delay_bits)
|
|
r.u(1)
|
|
if delay and r.u(1):
|
|
r.u(4)
|
|
wb = r.u(4) + 1
|
|
hb = r.u(4) + 1
|
|
return profile, level, r.u(wb) + 1, r.u(hb) + 1
|
|
except IndexError:
|
|
return None
|
|
|
|
|
|
def obu(b):
|
|
if not b:
|
|
return None
|
|
h = b[0]
|
|
if h & 0x81 or not h & 0x02:
|
|
return None
|
|
ty = (h >> 3) & 15
|
|
if not (1 <= ty <= 8 or ty == 15):
|
|
return None
|
|
i = 1 + ((h >> 2) & 1)
|
|
n = shift = 0
|
|
while True:
|
|
if i >= len(b):
|
|
return None
|
|
c = b[i]
|
|
i += 1
|
|
n |= (c & 0x7f) << shift
|
|
if not c & 0x80:
|
|
break
|
|
shift += 7
|
|
if shift > 49:
|
|
return None
|
|
return ty, i, n
|
|
|
|
|
|
def av1(head, size):
|
|
first = obu(head[:16])
|
|
if first is None or first[0] not in (1, 2):
|
|
return None
|
|
c = Cand("av1-obu", "elementary")
|
|
c.caps.add("wai.video.av1")
|
|
pos = count = 0
|
|
seen = False
|
|
while True:
|
|
if count == MAX_STEPS:
|
|
c.stop(pos, "limit")
|
|
break
|
|
left = len(head) - pos
|
|
if left == 0:
|
|
if pos < size:
|
|
c.stop(pos, "limit")
|
|
break
|
|
o = obu(head[pos:pos + 16])
|
|
if o is None:
|
|
c.stop(pos, "limit" if left < 8 and len(head) < size else "invalid")
|
|
break
|
|
ty, hdr, n = o
|
|
if pos + hdr + n > size:
|
|
c.stop(pos, "overrun")
|
|
break
|
|
if hdr + n > left:
|
|
c.stop(pos, "limit")
|
|
break
|
|
if ty == 1 and not seen:
|
|
seen = True
|
|
c.see(pos, hdr + n, "sequence_header")
|
|
q = av1_sequence(head[pos + hdr:pos + hdr + n])
|
|
if q:
|
|
c.facts.update(profile=q[0], level=q[1], width=q[2], height=q[3])
|
|
pos += hdr + n
|
|
count += 1
|
|
c.walked_to = pos
|
|
c.facts["units_read"] = count
|
|
if count and seen and count >= 2 and "width" in c.facts:
|
|
c.strength = "structure"
|
|
return c
|
|
return None
|
|
|
|
|
|
# ---- §5.18, §5.19 ----------------------------------------------------------------------
|
|
|
|
def hls(head):
|
|
b = head[3:] if head[:3] == b"\xef\xbb\xbf" else head
|
|
offset = len(head) - len(b)
|
|
if not b.startswith(b"#EXTM3U") or b[7:8] not in (b"\r", b"\n", b""):
|
|
return None
|
|
c = Cand("hls", "playlist")
|
|
c.see(offset, 7, "extm3u")
|
|
variants = segments = tags = maps = 0
|
|
key = version = codecs = None
|
|
ended = False
|
|
for line in b.split(b"\n"):
|
|
if line.endswith(b"\r"):
|
|
line = line[:-1]
|
|
if line.startswith(b"#EXT-X-"):
|
|
tags += 1
|
|
if line.startswith(b"#EXT-X-STREAM-INF:"):
|
|
variants += 1
|
|
if codecs is None and b'CODECS="' in line:
|
|
rest = line[line.index(b'CODECS="') + 8:]
|
|
if b'"' in rest:
|
|
codecs = clean(rest[:rest.index(b'"')])
|
|
elif line.startswith(b"#EXTINF:"):
|
|
segments += 1
|
|
elif line.startswith(b"#EXT-X-MAP:"):
|
|
maps += 1
|
|
elif line.startswith(b"#EXT-X-ENDLIST"):
|
|
ended = True
|
|
elif line.startswith(b"#EXT-X-VERSION:"):
|
|
version = decimal(line[15:])
|
|
elif line.startswith(b"#EXT-X-KEY:") or line.startswith(b"#EXT-X-SESSION-KEY:"):
|
|
if b"METHOD=" in line:
|
|
m = line[line.index(b"METHOD=") + 7:].split(b",")[0]
|
|
if m != b"NONE" and key is None:
|
|
key = clean(m)
|
|
c.facts.update(playlist="multivariant" if variants else "media" if segments else "other", variants=variants, segments=segments, maps=maps, ended=ended, version=version, key_method=key, codecs=codecs)
|
|
if tags:
|
|
c.strength = "structure"
|
|
return c
|
|
|
|
|
|
WS = b" \t\n\x0c\r"
|
|
|
|
|
|
def dash(head):
|
|
b = head[3:] if head[:3] == b"\xef\xbb\xbf" else head
|
|
offset = len(head) - len(b)
|
|
i = 0
|
|
|
|
def skip(i):
|
|
while i < len(b) and b[i] in WS:
|
|
i += 1
|
|
return i
|
|
i = skip(i)
|
|
while True:
|
|
if b[i:i + 2] == b"<?":
|
|
e = b.find(b"?>", i)
|
|
if e < 0:
|
|
return None
|
|
i = e + 2
|
|
elif b[i:i + 4] == b"<!--":
|
|
e = b.find(b"-->", i)
|
|
if e < 0:
|
|
return None
|
|
i = e + 3
|
|
else:
|
|
break
|
|
i = skip(i)
|
|
if b[i:i + 4] != b"<MPD" or i + 4 >= len(b) or (b[i + 4] not in WS and b[i + 4] != 0x3e):
|
|
return None
|
|
c = Cand("dash", "playlist")
|
|
c.see(offset + i, 4, "mpd")
|
|
n = b.find(b"urn:mpeg:dash:schema:mpd:2011")
|
|
if n >= 0:
|
|
c.see(offset + n, 29, "namespace")
|
|
c.strength = "structure"
|
|
rest = b[i + 4:]
|
|
root = rest.split(b">")[0]
|
|
kind = None
|
|
if b' type="' in root:
|
|
tail = root[root.index(b' type="') + 7:]
|
|
if b'"' in tail:
|
|
kind = clean(tail[:tail.index(b'"')])
|
|
c.facts.update(type=kind, periods=b.count(b"<Period"), adaptation_sets=b.count(b"<AdaptationSet"), representations=b.count(b"<Representation"), content_protection=b.count(b"<ContentProtection"))
|
|
codecs = []
|
|
j = 0
|
|
while True:
|
|
at = b.find(b'codecs="', j)
|
|
if at < 0:
|
|
break
|
|
s = at + 8
|
|
e = b.find(b'"', s)
|
|
if e < 0:
|
|
break
|
|
v = clean(b[s:e])
|
|
if v not in codecs and len(codecs) < 16:
|
|
codecs.append(v)
|
|
j = e
|
|
c.facts["codecs"] = codecs
|
|
return c
|
|
|
|
|
|
# ---- §5.20 -----------------------------------------------------------------------------
|
|
|
|
def nal_units(head):
|
|
lead = 0
|
|
while lead < len(head) and head[lead] == 0:
|
|
lead += 1
|
|
if lead >= len(head) or lead < 2 or head[lead] != 1:
|
|
return None
|
|
starts = []
|
|
i = head.find(b"\0\0\1")
|
|
while 0 <= i and i + 2 < len(head):
|
|
starts.append(i + 3)
|
|
if len(starts) > MAX_STEPS:
|
|
break
|
|
i = head.find(b"\0\0\1", i + 3)
|
|
units = []
|
|
for k, s in enumerate(starts):
|
|
if k == MAX_STEPS:
|
|
break
|
|
end = starts[k + 1] - 3 if k + 1 < len(starts) else len(head)
|
|
while end > s and head[end - 1] == 0:
|
|
end -= 1
|
|
units.append((s, head[s:end]))
|
|
return units
|
|
|
|
|
|
def unescape(n):
|
|
out = bytearray()
|
|
zeros = 0
|
|
for x in n:
|
|
if zeros >= 2 and x == 3:
|
|
zeros = 0
|
|
continue
|
|
zeros = zeros + 1 if x == 0 else 0
|
|
out.append(x)
|
|
return bytes(out)
|
|
|
|
|
|
def avc_sps(rbsp):
|
|
# The profiles whose header holds the chroma format: every one §5.20 lists but the first three.
|
|
high = set(AVC_PROFILES[3:])
|
|
r = Bits(rbsp)
|
|
try:
|
|
profile = r.u(8)
|
|
if profile not in AVC_PROFILES:
|
|
return None
|
|
flags, level = r.u(8), r.u(8)
|
|
r.ue()
|
|
chroma, separate = 1, False
|
|
if profile in high:
|
|
chroma = r.ue()
|
|
if chroma > 3:
|
|
return None
|
|
if chroma == 3:
|
|
separate = bool(r.u(1))
|
|
r.ue()
|
|
r.ue()
|
|
r.u(1)
|
|
if r.u(1):
|
|
for i in range(8 if chroma != 3 else 12):
|
|
if r.u(1):
|
|
last, nxt = 8, 8
|
|
for _ in range(16 if i < 6 else 64):
|
|
if nxt != 0:
|
|
code = r.ue()
|
|
delta = (code + 1) // 2 if code % 2 else -(code // 2)
|
|
nxt = (last + delta + 256) % 256
|
|
last = last if nxt == 0 else nxt
|
|
r.ue()
|
|
poc = r.ue()
|
|
if poc == 0:
|
|
r.ue()
|
|
elif poc == 1:
|
|
r.u(1)
|
|
r.ue()
|
|
r.ue()
|
|
n = r.ue()
|
|
if n > 255:
|
|
return None
|
|
for _ in range(n):
|
|
r.ue()
|
|
elif poc != 2:
|
|
return None
|
|
r.ue()
|
|
r.u(1)
|
|
w_mbs, h_units = r.ue() + 1, r.ue() + 1
|
|
frame_only = r.u(1)
|
|
if not frame_only:
|
|
r.u(1)
|
|
r.u(1)
|
|
left = right = top = bottom = 0
|
|
if r.u(1):
|
|
left, right, top, bottom = r.ue(), r.ue(), r.ue(), r.ue()
|
|
except IndexError:
|
|
return None
|
|
array = 0 if separate else chroma
|
|
sw, sh = {1: (2, 2), 2: (2, 1)}.get(array, (1, 1))
|
|
cx = 1 if array == 0 else sw
|
|
cy = (1 if array == 0 else sh) * (1 if frame_only else 2)
|
|
width = w_mbs * 16 - cx * (left + right)
|
|
height = h_units * (16 if frame_only else 32) - cy * (top + bottom)
|
|
if width < 0 or height < 0:
|
|
return None
|
|
return {"profile_idc": profile, "constraint_flags": flags, "level_idc": level, "chroma_format_idc": chroma, "width": width, "height": height}
|
|
|
|
|
|
def in_order(types, order, vcl):
|
|
first = next((k for k, t in enumerate(types) if vcl(t)), len(types))
|
|
last = 0
|
|
for want in order:
|
|
if want not in types:
|
|
return False
|
|
at = types.index(want)
|
|
if at >= first or at < last:
|
|
return False
|
|
last = at
|
|
return True
|
|
|
|
|
|
def annex_b(head, size):
|
|
units = nal_units(head)
|
|
if not units or not units[0][1] or units[0][1][0] in (0xba, 0xb3):
|
|
return []
|
|
out = []
|
|
|
|
def make(id_, cap):
|
|
c = Cand(id_, "elementary")
|
|
c.caps.add(cap)
|
|
c.strength = "structure"
|
|
c.facts["units_read"] = len(units)
|
|
return c
|
|
|
|
def finish(c):
|
|
if len(head) < size or len(units) == MAX_STEPS:
|
|
c.stop(units[-1][0], "limit")
|
|
out.append(c)
|
|
|
|
types = []
|
|
for _, u in units:
|
|
if not u:
|
|
types = None
|
|
break
|
|
b = u[0]
|
|
t = b & 31
|
|
if b & 0x80 or not 1 <= t <= 21 or (t in (5, 7, 8) and b >> 5 == 0):
|
|
types = None
|
|
break
|
|
types.append(t)
|
|
if types and in_order(types, [7, 8], lambda t: 1 <= t <= 5):
|
|
at, sps = next((a, u) for a, u in units if u[0] & 31 == 7)
|
|
v = avc_sps(unescape(sps)[1:])
|
|
if v:
|
|
c = make("avc-annexb", "wai.video.avc")
|
|
c.see(at, len(sps), "nal:sps")
|
|
c.facts.update(v)
|
|
finish(c)
|
|
|
|
types = []
|
|
for _, u in units:
|
|
if len(u) < 2:
|
|
types = None
|
|
break
|
|
t = (u[0] >> 1) & 63
|
|
if u[0] & 0x80 or not u[1] & 7 or not (t <= 9 or 16 <= t <= 21 or 32 <= t <= 40):
|
|
types = None
|
|
break
|
|
types.append(t)
|
|
if types and in_order(types, [32, 33, 34], lambda t: t <= 9 or 16 <= t <= 21):
|
|
c = make("hevc-annexb", "wai.video.hevc")
|
|
for at, u in units:
|
|
if (u[0] >> 1) & 63 == 33:
|
|
c.see(at, len(u), "nal:sps")
|
|
break
|
|
finish(c)
|
|
|
|
types = []
|
|
for _, u in units:
|
|
if len(u) < 2:
|
|
types = None
|
|
break
|
|
t = u[1] >> 3
|
|
if u[0] & 0xc0 or not u[1] & 7 or t > 25:
|
|
types = None
|
|
break
|
|
types.append(t)
|
|
if types and in_order(types, [15, 16], lambda t: t <= 10):
|
|
c = make("vvc-annexb", "wai.video.vvc")
|
|
for at, u in units:
|
|
if u[1] >> 3 == 15:
|
|
c.see(at, len(u), "nal:sps")
|
|
break
|
|
finish(c)
|
|
return out
|
|
|
|
|
|
# ---- §3.5, §6 --------------------------------------------------------------------------
|
|
|
|
IDENTIFYING = ["wai.image.jxs", "wai.video.jxs", "wai.video.hevc", "wai.video.mvhevc", "wai.video.vvc", "wai.video.lcevc", "wai.audio.mpegh", "wai.video.avs3", "wai.audio.avs3"]
|
|
|
|
|
|
def probes(head, found):
|
|
accepted = []
|
|
for cap in IDENTIFYING:
|
|
o = fpv.probe(cap, head)
|
|
if "accept" in o:
|
|
accepted.append(cap)
|
|
hit = next((c for c in found if c.kind == "elementary" and (cap in c.caps or (cap == "wai.video.mvhevc" and "wai.video.hevc" in c.caps))), None)
|
|
if hit is None:
|
|
hit = Cand(cap, "elementary")
|
|
hit.see(0, len(head), "probe")
|
|
found.append(hit)
|
|
hit.strength = "parameters"
|
|
hit.caps.add(cap)
|
|
hit.facts.setdefault("probe", {})[cap] = o["accept"]
|
|
elif o["refuse"] != "no_signature":
|
|
hit = next((c for c in found if c.kind == "elementary" and cap in c.caps), None)
|
|
if hit is not None:
|
|
hit.facts.setdefault("probe_refused", {})[cap] = o["refuse"]
|
|
return accepted
|
|
|
|
|
|
# ---- §7, §8 ----------------------------------------------------------------------------
|
|
|
|
def embedded(scan):
|
|
hits = []
|
|
for id_, b, back in EMBEDDED:
|
|
i = scan.find(b)
|
|
while i >= 0:
|
|
if i > back:
|
|
hits.append((i, len(hits), {"at": i - back, "id": id_}))
|
|
i = scan.find(b, i + 1)
|
|
# In order of the offset their bytes are found at, within one offset in the table's order.
|
|
hits.sort(key=lambda h: (h[0], h[1]))
|
|
return [h[2] for h in hits[:32]], len(hits) > 32
|
|
|
|
|
|
def counts(b):
|
|
return [b.count(bytes([v])) for v in range(256)]
|
|
|
|
|
|
def isqrt(n):
|
|
if n < 2:
|
|
return n
|
|
x = 1 << ((n.bit_length() + 1) // 2)
|
|
while True:
|
|
y = (x + n // x) // 2
|
|
if y >= x:
|
|
return x
|
|
x = y
|
|
|
|
|
|
def periods(s):
|
|
n = len(s)
|
|
longest = min(4096, n // 4)
|
|
if longest < 2:
|
|
return []
|
|
total = sum(c * c for c in counts(s))
|
|
whole = int.from_bytes(s, "big")
|
|
e = [0] * (longest + 2)
|
|
z = [0] * (longest + 2)
|
|
for lag in range(1, longest + 2):
|
|
m = n - lag
|
|
# The bytes at i and i + lag differ where the XOR of the two runs has a nonzero byte.
|
|
x = (whole >> (8 * lag)) ^ (whole & ((1 << (8 * m)) - 1))
|
|
matches = x.to_bytes(m, "big").count(0)
|
|
expected = max(m * total // (n * n), 1)
|
|
e[lag] = matches - expected
|
|
if e[lag] > 0:
|
|
z[lag] = isqrt(e[lag] * e[lag] // expected)
|
|
sizes = sorted(abs(v) for v in e[1:])
|
|
floor = 8 * sizes[len(sizes) // 2]
|
|
out = []
|
|
for lag in range(2, longest + 1):
|
|
if z[lag] < 8 or e[lag] < floor or e[lag] <= 2 * max(e[lag - 1], e[lag + 1], 0):
|
|
continue
|
|
if 2 * lag <= longest + 1 and z[2 * lag] < 8:
|
|
continue
|
|
if any(lag % p == 0 for p in out):
|
|
continue
|
|
out.append(lag)
|
|
if len(out) == 8:
|
|
break
|
|
return [{"lag": lag, "z": z[lag]} for lag in out]
|
|
|
|
|
|
def is_text(b):
|
|
if not b:
|
|
return False
|
|
try:
|
|
b.decode("utf-8")
|
|
except UnicodeDecodeError as err:
|
|
if err.end != len(b) or err.reason != "unexpected end of data":
|
|
return False
|
|
return not any((x < 0x20 and x not in (9, 10, 12, 13)) or x == 0x7f for x in set(b))
|
|
|
|
|
|
def measures(scan):
|
|
c = counts(scan)
|
|
n = len(scan)
|
|
windows = uniform = 0
|
|
for i in range(0, n - 4095, 4096):
|
|
windows += 1
|
|
w = scan[i:i + 4096]
|
|
if sum(x * x for x in counts(w)) <= 70496:
|
|
uniform += 1
|
|
return {
|
|
"scanned": n,
|
|
"distinct_bytes": sum(1 for x in c if x),
|
|
"zero_bytes": c[0],
|
|
"chi_square": 0 if n == 0 else max(256 * sum(x * x for x in c) // n - n, 0),
|
|
"windows": windows,
|
|
"uniform_windows": uniform,
|
|
"text": is_text(scan),
|
|
"periods": periods(scan[:256 << 10]),
|
|
}
|
|
|
|
|
|
def identify(d):
|
|
size = len(d)
|
|
head, scan = d[:HEAD], d[:SCAN]
|
|
found = [c for c in (
|
|
wai_envelope(d, head), iso(d, head), matroska(d, head), transport(head, size), ogg(d, head), flac(d, head), riff(d, head),
|
|
mpeg_audio(head, size), adts(head, size), png(d, head), jpeg(d, head), gif(head, size), ivf(d, head), y4m(d, head), flv(d, head),
|
|
av1(head, size), hls(head), dash(head)) if c is not None]
|
|
found += annex_b(head, size)
|
|
for id_, kind, parts, caps in SIGNATURES:
|
|
if all(head[a:a + len(b)] == b for a, b in parts):
|
|
c = Cand(id_, kind)
|
|
for a, b in parts:
|
|
c.see(a, len(b), "signature")
|
|
c.caps |= set(caps)
|
|
found.append(c)
|
|
accepted = probes(head, found)
|
|
found.sort(key=lambda c: -RANK[c.strength])
|
|
top = [c for c in found if c.strength == found[0].strength]
|
|
if not found:
|
|
verdict = "unknown"
|
|
elif len(top) > 1:
|
|
verdict = "ambiguous"
|
|
elif top[0].strength == "signature":
|
|
verdict = "signature_only"
|
|
else:
|
|
verdict = "identified"
|
|
em, capped = embedded(scan)
|
|
return {"format": "wai.identify/1", "size": size, "verdict": verdict, "candidates": [c.value(size) for c in found], "probes_accepted": accepted,
|
|
"embedded": em, "embedded_capped": capped, "measures": measures(scan)}
|
|
|
|
|
|
# ---- the corpus ------------------------------------------------------------------------
|
|
|
|
PROBER_FORMAT = {"isobmff": "mov", "matroska": "matroska", "mpeg-ts": "mpegts", "ogg": "ogg", "flac": "flac", "wave": "wav", "mpeg-audio": "mp3", "adts": "aac", "png": "png_pipe", "jpeg": "jpeg_pipe",
|
|
"gif": "gif", "webp": "webp_pipe", "ivf": "ivf", "y4m": "yuv4mpegpipe", "flv": "flv", "av1-obu": "obu", "avc-annexb": "h264", "hevc-annexb": "hevc", "wai.video.hevc": "hevc",
|
|
"jpegxl": "jpegxl_pipe", "jp2": "j2k_pipe", "j2k": "j2k_pipe", "tiff": "tiff_pipe", "aiff": "aiff", "caf": "caf", "mpeg-video": "mpegvideo", "mpeg-ps": "mpeg", "webvtt": "webvtt"}
|
|
|
|
|
|
def agrees(name, result, prober):
|
|
"""Where a result and the prober both state a value, what does not agree."""
|
|
p = prober.get(name)
|
|
if not p:
|
|
return []
|
|
out = []
|
|
best = result["candidates"][0] if result["candidates"] else None
|
|
if best is None:
|
|
return ["%s: the prober reads %s, and the result holds no candidate" % (name, p["format"])]
|
|
if not p["format"].startswith(PROBER_FORMAT.get(best["id"], "?")):
|
|
out.append("%s: %s against the prober's %s" % (name, best["id"], p["format"]))
|
|
if best["tracks"] and len(best["tracks"]) != len(p["streams"]):
|
|
out.append("%s: %d tracks against the prober's %d" % (name, len(best["tracks"]), len(p["streams"])))
|
|
for t in best["tracks"]:
|
|
s = next((x for x in p["streams"] if x["media"] == t["media"] and (t["codec"] is None or x["codec"] == t["codec"])), None)
|
|
if s is None:
|
|
out.append("%s: no stream of the prober's is %s %s" % (name, t["media"], t["codec"]))
|
|
continue
|
|
for k in ("width", "height", "channels", "sample_rate"):
|
|
if t[k] is not None and s.get(k) is not None and t[k] != s[k]:
|
|
out.append("%s: %s %s against the prober's %s" % (name, k, t[k], s[k]))
|
|
facts = best["facts"]
|
|
first = p["streams"][0] if p["streams"] else {}
|
|
for k in ("width", "height", "sample_rate", "channels"):
|
|
if facts.get(k) is not None and first.get(k) is not None and facts[k] != first[k]:
|
|
out.append("%s: %s %s against the prober's %s" % (name, k, facts[k], first[k]))
|
|
return out
|
|
|
|
|
|
def check(doc, files, prober):
|
|
fails = []
|
|
named = set()
|
|
results = []
|
|
for v in doc["vectors"]:
|
|
named.add(v["file"])
|
|
b = files.get(v["file"])
|
|
if b is None:
|
|
fails.append("%s is absent" % v["file"])
|
|
continue
|
|
if hashlib.sha256(b).hexdigest() != v["sha256"]:
|
|
fails.append("%s: its digest" % v["file"])
|
|
r = identify(b * v.get("repeat", 1))
|
|
results.append(r)
|
|
if r != v["result"]:
|
|
for k in sorted(set(r) | set(v["result"])):
|
|
if r.get(k) != v["result"].get(k):
|
|
fails.append("%s: %s is %s, stated %s" % (v["file"], k, json.dumps(r.get(k))[:400], json.dumps(v["result"].get(k))[:400]))
|
|
if v["origin"] == "ext":
|
|
fails += agrees(v["file"].split("/", 1)[1], r, prober)
|
|
for k in files:
|
|
if k not in named:
|
|
fails.append("%s is no vector's" % k)
|
|
stops = {c["stopped"]["why"] for r in results for c in r["candidates"] if c["stopped"]}
|
|
for why in ("overrun", "invalid", "crc", "limit", "unknown_size"):
|
|
if why not in stops:
|
|
fails.append("no vector reaches %s" % why)
|
|
verdicts = {r["verdict"] for r in results}
|
|
for k in ("identified", "ambiguous", "signature_only", "unknown"):
|
|
if k not in verdicts:
|
|
fails.append("no vector reaches %s" % k)
|
|
strengths = {c["strength"] for r in results for c in r["candidates"]}
|
|
for k in RANK:
|
|
if k not in strengths:
|
|
fails.append("no vector reaches %s" % k)
|
|
return fails
|
|
|
|
|
|
def load(corpus):
|
|
doc = json.load(open(os.path.join(corpus, "vectors.json"), encoding="utf-8"))
|
|
files = {}
|
|
for sub in ("ext", "gen"):
|
|
d = os.path.join(corpus, sub)
|
|
for n in sorted(os.listdir(d)):
|
|
files["%s/%s" % (sub, n)] = open(os.path.join(d, n), "rb").read()
|
|
prober = json.load(open(os.path.join(corpus, "prober.json"), encoding="utf-8"))
|
|
return doc, files, prober
|
|
|
|
|
|
def find(doc, name):
|
|
return next(v for v in doc["vectors"] if v["file"].endswith("/" + name))
|
|
|
|
|
|
def mutations(doc, files, prober):
|
|
"""Each yields a corpus that must fail, the fast vectors only: those past the windows are costly."""
|
|
fast = dict(doc)
|
|
fast["vectors"] = [v for v in doc["vectors"] if "repeat" not in v]
|
|
fast_files = {k: v for k, v in files.items() if not any(v2["file"] == k and "repeat" in v2 for v2 in doc["vectors"])}
|
|
|
|
def changed(name, edit):
|
|
d = copy.deepcopy(fast)
|
|
edit(find(d, name)["result"])
|
|
return d
|
|
|
|
def cand(r):
|
|
return r["candidates"][0]
|
|
yield "a result's verdict", changed("avc-aac.mp4", lambda r: r.update(verdict="ambiguous")), fast_files, prober
|
|
yield "a candidate's strength", changed("avc-aac.mp4", lambda r: cand(r).update(strength="signature")), fast_files, prober
|
|
yield "an evidence offset", changed("avc-aac.mp4", lambda r: cand(r)["evidence"][0].update(at=5)), fast_files, prober
|
|
yield "where a walk reached", changed("cut-mp4.bin", lambda r: cand(r).update(walked_to=41)), fast_files, prober
|
|
yield "why a walk stopped", changed("cut-mp4.bin", lambda r: cand(r)["stopped"].update(why="invalid")), fast_files, prober
|
|
yield "a track's codec", changed("avc-aac.mp4", lambda r: cand(r)["tracks"][0].update(codec="hevc")), fast_files, prober
|
|
yield "a track's width", changed("avc-aac.mp4", lambda r: cand(r)["tracks"][0].update(width=33)), fast_files, prober
|
|
yield "a track's protection", changed("avc-protected.mp4", lambda r: cand(r)["tracks"][0].update(protected=None)), fast_files, prober
|
|
yield "a capability", changed("avc-aac.mp4", lambda r: cand(r).update(capabilities=[])), fast_files, prober
|
|
yield "an embedded offset", changed("embedded.bin", lambda r: r["embedded"][0].update(at=1)), fast_files, prober
|
|
yield "the embedded cap", changed("embedded-capped.bin", lambda r: r.update(embedded_capped=False)), fast_files, prober
|
|
yield "a repeat", changed("records.bin", lambda r: r["measures"].update(periods=[])), fast_files, prober
|
|
yield "the chi-square", changed("noise.bin", lambda r: r["measures"].update(chi_square=1)), fast_files, prober
|
|
yield "the uniform windows", changed("noise.bin", lambda r: r["measures"].update(uniform_windows=0)), fast_files, prober
|
|
yield "whether it is text", changed("text.bin", lambda r: r["measures"].update(text=False)), fast_files, prober
|
|
yield "a probe's parameters", changed("hevc.h265", lambda r: r.update(probes_accepted=[])), fast_files, prober
|
|
yield "a stated fact", changed("a.flac", lambda r: cand(r)["facts"].update(sample_rate=44100)), fast_files, prober
|
|
|
|
d = copy.deepcopy(fast)
|
|
d["vectors"] = [v for v in d["vectors"] if not v["file"].endswith("/polyglot-annexb.bin") and not v["file"].endswith("/polyglot-signature.bin")]
|
|
yield "no vector reaching a verdict", d, {k: v for k, v in fast_files.items() if "polyglot" not in k}, prober
|
|
|
|
yield "a file not a vector's", fast, dict(fast_files, **{"gen/stray.bin": b"x"}), prober
|
|
f = dict(fast_files)
|
|
f["ext/avc-aac.mp4"] = f["ext/avc-aac.mp4"][:-1] + bytes([f["ext/avc-aac.mp4"][-1] ^ 1])
|
|
yield "a file's byte", fast, f, prober
|
|
p = copy.deepcopy(prober)
|
|
p["avc-aac.mp4"]["streams"][0]["width"] = 33
|
|
yield "the prober disagreeing on a width", fast, fast_files, p
|
|
p = copy.deepcopy(prober)
|
|
p["a.flac"]["format"] = "ogg"
|
|
yield "the prober disagreeing on the format", fast, fast_files, p
|
|
p = copy.deepcopy(prober)
|
|
p["avc-aac.mp4"]["streams"].pop()
|
|
yield "the prober seeing fewer tracks", fast, fast_files, p
|
|
|
|
|
|
def main(argv):
|
|
if len(argv) != 2:
|
|
print(__doc__)
|
|
return 2
|
|
doc, files, prober = load(argv[1])
|
|
fails = check(doc, files, prober)
|
|
for f in fails:
|
|
print("FAIL", f[:900])
|
|
print("%d vectors, %d files: %d failures" % (len(doc["vectors"]), len(files), len(fails)))
|
|
caught_all = True
|
|
for label, d, fs, p in mutations(doc, files, prober):
|
|
try:
|
|
caught = check(d, fs, p)
|
|
except Exception as e: # a crash on a mutated copy counts as a failure to read it
|
|
caught = ["raised %s" % e]
|
|
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:90] if caught else ""))
|
|
caught_all &= bool(caught)
|
|
return 0 if not fails and caught_all else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|