b19e078cc3
live-ingest §9.5: a WAI-aware WHEP player writes, per media section of its offer, the WAI capabilities its sink dispatches, in its dispatch order, as SDP media capabilities (RFC 6871) under capability negotiation (RFC 5939): a=csup:med-v0, then for each capability §5.1 carries over WebRTC (wai.video.avc, vp8, vp9, av1, wai.audio.opus.packet) that the section offers, an a=rmcap, an a=mfcap of its fmtp, and an a=pcfg whose number gives the order. The egress gateway reads them back, refuses with 422 an offer whose capability attributes do not parse or do not hold together (caps_malformed, caps_number_reused, caps_config_reused, caps_unknown_capability, caps_payload_type, caps_option_missing, caps_unsupported, caps_too_many; §9.5 and §14), forwards for each section the rendition the first advertised capability names at or below the grant's max_layer, answered with a=acfg, and the floor where nothing advertised is published. §17 records that joule-moq-ingest does not read them yet. wai::sdp_caps: read, Offer::advertised, pick and Pick::acfg, write. Its tests replay live-conformance/sdp-caps/vectors.json (offers read and refused with every reason, picks, offers written byte for byte and read back), write and read back every order of the video capabilities, and hold §9.5's table and reasons and §14's to the module. live-conformance/sdp_caps_verify.py, written from §9.5 and the RFCs' grammars and sharing no code with wai-rs, recomputes every case and requires every mutated copy to fail. Co-Authored-By: Claude Opus 5.5 <[email protected]>
492 lines
19 KiB
Python
492 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the SDP capability attributes of a WHEP offer
|
|
(extensions/live-ingest.md §9.5).
|
|
|
|
Written from §9.5's text and the grammars it cites (RFC 5939 §3.3 and §3.5, RFC 6871 §3.3
|
|
and §3.4); it shares no code with wai-rs's sdp_caps, which the corpus is checked against too.
|
|
It reads from §9.5 the table of the capabilities a gateway carries over WebRTC, and from §14
|
|
the reasons. For every case of sdp-caps/vectors.json it recomputes:
|
|
|
|
- `reads`: the offer read or refused, with the refusal's reason; for an offer read, each media
|
|
section's advertised WAI capabilities in order, each with its configuration, media
|
|
capability and payload type;
|
|
- `picks`: the rendition a gateway forwards for a section, or the floor, and the answer's
|
|
`a=acfg` line;
|
|
- `writes`: the offer with a sink's capabilities written into it, byte for byte, the
|
|
capabilities left out, and that reading it back gives the capabilities written, in order.
|
|
|
|
It requires every reason of §9.5 to appear in the corpus and §14 to list each. Then it
|
|
mutates copies of the corpus, one defect each, and requires every copy to fail.
|
|
|
|
python3 -I sdp_caps_verify.py <live-conformance/sdp-caps>
|
|
|
|
Exit status 0 only if every check holds and every mutated copy fails. Python 3.8 or later;
|
|
standard library only.
|
|
"""
|
|
import copy
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
WAI = os.path.normpath(os.path.join(HERE, ".."))
|
|
|
|
|
|
def read_text(p):
|
|
with open(p, encoding="utf-8") as f:
|
|
return f.read()
|
|
|
|
|
|
LI = read_text(os.path.join(WAI, "extensions", "live-ingest.md"))
|
|
|
|
|
|
def section(text, heading, stop):
|
|
i = text.index(heading)
|
|
return text[i:text.index(stop, i + len(heading))]
|
|
|
|
|
|
def ticks(s):
|
|
return re.findall(r"`([^`]*)`", s)
|
|
|
|
|
|
def rtp_rows():
|
|
"""§9.5's table: capability, encoding name, clock rate, encoding parameters."""
|
|
rows = []
|
|
for line in section(LI, "### 9.5 ", "## 10.").splitlines():
|
|
cells = [c.strip() for c in line.strip().strip("|").split("|")]
|
|
if line.startswith("| `wai.") and len(cells) == 4:
|
|
params = ticks(cells[3])
|
|
rows.append((ticks(cells[0])[0], ticks(cells[1])[0], int(ticks(cells[2])[0]), params[0] if params else None))
|
|
assert rows, "§9.5 has no table"
|
|
return rows
|
|
|
|
|
|
def reasons_95():
|
|
out = []
|
|
for line in section(LI, "### 9.5 ", "## 10.").splitlines():
|
|
if line.startswith("| `caps_"):
|
|
out.append(ticks(line)[0])
|
|
return out
|
|
|
|
|
|
def reasons_14():
|
|
out = set()
|
|
for line in section(LI, "## 14. Reasons", "## 15.").splitlines():
|
|
if line.startswith("| `"):
|
|
out.update(ticks(line.split("|")[1]))
|
|
return out
|
|
|
|
|
|
ROWS = rtp_rows()
|
|
REASONS = reasons_95()
|
|
MAX_CAPS = 1024
|
|
|
|
|
|
class Refusal(Exception):
|
|
def __init__(self, reason, detail):
|
|
Exception.__init__(self, "%s: %s" % (reason, detail))
|
|
self.reason = reason
|
|
|
|
|
|
TOKEN = set(b"!#$%&'*+-.^_`{|}~") | set(range(0x30, 0x3A)) | set(range(0x41, 0x5B)) | set(range(0x61, 0x7B))
|
|
|
|
|
|
def is_token(s):
|
|
return bool(s) and all(c in TOKEN for c in s.encode("latin-1", "replace"))
|
|
|
|
|
|
def num(s, nonzero=False):
|
|
if not s or len(s) > 10 or not s.isdigit() or not all(c in "0123456789" for c in s):
|
|
return None
|
|
if nonzero and s[0] == "0":
|
|
return None
|
|
v = int(s)
|
|
return v if 1 <= v <= 2 ** 31 - 1 else None
|
|
|
|
|
|
def cap_list(s):
|
|
out = []
|
|
for el in s.split(","):
|
|
if "-" in el:
|
|
a, b = el.split("-", 1)
|
|
a, b = num(a, True), num(b, True)
|
|
if a is None or b is None or a >= b:
|
|
raise Refusal("caps_malformed", "range " + el)
|
|
if b - a >= MAX_CAPS:
|
|
raise Refusal("caps_too_many", "range " + el)
|
|
out.extend(range(a, b + 1))
|
|
else:
|
|
n = num(el, True)
|
|
if n is None:
|
|
raise Refusal("caps_malformed", "number " + el)
|
|
out.append(n)
|
|
return out
|
|
|
|
|
|
def wsp(s):
|
|
m = re.match(r"([^ \t]*)[ \t]+(.+)$", s, re.S)
|
|
return (m.group(1), m.group(2)) if m and m.group(1) else None
|
|
|
|
|
|
def att_config(val):
|
|
"""RFC 5939 attribute-config-list after its "a="; None when it is off the grammar."""
|
|
dels = ("-m", "-s", "-ms")
|
|
if ":" in val:
|
|
d, lists = val.split(":", 1)
|
|
if d not in dels:
|
|
return None
|
|
elif val in dels:
|
|
return []
|
|
else:
|
|
lists = val
|
|
out = []
|
|
for alt in lists.split("|"):
|
|
m = re.fullmatch(r"(\d+(?:,\d+)*)?(?:(?(1),)\[(\d+(?:,\d+)*)\])?", alt)
|
|
if not m or not alt:
|
|
return None
|
|
for g in (m.group(1), m.group(2)):
|
|
for n in (g.split(",") if g else []):
|
|
if num(n) is None:
|
|
return None
|
|
out.append(num(n))
|
|
return out
|
|
|
|
|
|
def pcfg(number, rest):
|
|
cfg = {"number": number, "media": [], "pt": [], "skip": False}
|
|
a_refs, t_refs, seen = [], [], set()
|
|
for item in [x for x in re.split(r"[ \t]+", rest) if x]:
|
|
plus = item.startswith("+")
|
|
body = item[1:] if plus else item
|
|
if "=" not in body:
|
|
raise Refusal("caps_malformed", item)
|
|
key, val = body.split("=", 1)
|
|
if key in ("a", "t", "m", "pt"):
|
|
if key in seen:
|
|
raise Refusal("caps_malformed", "a second " + key)
|
|
seen.add(key)
|
|
if key == "a" and not plus:
|
|
got = att_config(val)
|
|
if got is None:
|
|
raise Refusal("caps_malformed", item)
|
|
a_refs += got
|
|
cfg["skip"] = True
|
|
elif key == "t" and not plus:
|
|
for n in val.split("|"):
|
|
if num(n) is None:
|
|
raise Refusal("caps_malformed", item)
|
|
t_refs.append(num(n))
|
|
cfg["skip"] = True
|
|
elif key == "m":
|
|
cfg["media"] = [cap_list(alt) for alt in val.split("|")]
|
|
elif key == "pt":
|
|
for mp in val.split(","):
|
|
if ":" not in mp:
|
|
raise Refusal("caps_malformed", item)
|
|
m, p = mp.split(":", 1)
|
|
m, pv = num(m, True), num(p, True)
|
|
if m is None or pv is None or len(p) > 3:
|
|
raise Refusal("caps_malformed", item)
|
|
if pv > 127:
|
|
raise Refusal("caps_payload_type", item)
|
|
if any(q == pv or c == m for c, q in cfg["pt"]):
|
|
raise Refusal("caps_payload_type", item)
|
|
cfg["pt"].append((m, pv))
|
|
else:
|
|
if not key or not all(c.isalnum() and c.isascii() for c in key) or not val or not all(0x21 <= ord(c) <= 0x7E for c in val):
|
|
raise Refusal("caps_malformed", item)
|
|
if plus:
|
|
cfg["skip"] = True
|
|
return cfg, a_refs, t_refs
|
|
|
|
|
|
def sdp_lines(sdp):
|
|
return [l[:-1] if l.endswith("\r") else l for l in sdp.split("\n") if l not in ("", "\r")]
|
|
|
|
|
|
def read(sdp):
|
|
options, required, caps, mf, acap, tcap, sections, cfgnums, pending, creq_at = set(), [], {}, [], {}, {}, [], set(), [], set()
|
|
for line in sdp_lines(sdp):
|
|
if line.startswith("m="):
|
|
sections.append({"media": line[2:].split(" ")[0], "configs": []})
|
|
continue
|
|
if not line.startswith("a="):
|
|
continue
|
|
attr = line[2:]
|
|
name, value = attr.split(":", 1) if ":" in attr else (attr, "")
|
|
here = len(sections) - 1 if sections else None
|
|
if name in ("csup", "creq"):
|
|
tags = value.split(",")
|
|
if not all(is_token(t) for t in tags):
|
|
raise Refusal("caps_malformed", value)
|
|
if name == "creq":
|
|
if here in creq_at:
|
|
raise Refusal("caps_malformed", "a second creq at one level")
|
|
creq_at.add(here)
|
|
required += tags
|
|
options.update(tags)
|
|
elif name in ("rmcap", "omcap"):
|
|
sp = wsp(value)
|
|
if not sp:
|
|
raise Refusal("caps_malformed", attr)
|
|
nums = cap_list(sp[0])
|
|
if name == "rmcap":
|
|
parts = sp[1].split("/")
|
|
if len(parts) not in (2, 3) or not is_token(parts[0]) or (len(parts) == 3 and not is_token(parts[2])):
|
|
raise Refusal("caps_malformed", attr)
|
|
rate = num(parts[1], True)
|
|
if rate is None:
|
|
raise Refusal("caps_malformed", attr)
|
|
entry = (True, parts[0], rate, parts[2] if len(parts) == 3 else None)
|
|
else:
|
|
if not is_token(sp[1]):
|
|
raise Refusal("caps_malformed", attr)
|
|
entry = (False, sp[1], None, None)
|
|
for n in nums:
|
|
if n in caps:
|
|
raise Refusal("caps_number_reused", str(n))
|
|
caps[n] = {"rtp": entry[0], "name": entry[1], "rate": entry[2], "params": entry[3], "level": here, "fmtp": []}
|
|
if len(caps) > MAX_CAPS:
|
|
raise Refusal("caps_too_many", "count")
|
|
elif name == "mfcap":
|
|
sp = wsp(value)
|
|
if not sp:
|
|
raise Refusal("caps_malformed", attr)
|
|
mf.append((cap_list(sp[0]), sp[1]))
|
|
elif name in ("acap", "tcap"):
|
|
sp = wsp(value)
|
|
if not sp or num(sp[0]) is None:
|
|
raise Refusal("caps_malformed", attr)
|
|
ok = is_token(sp[1].split(":")[0]) if name == "acap" else all(is_token(t) for p in re.split(r"[ \t]+", sp[1]) if p for t in p.split("/"))
|
|
if not ok:
|
|
raise Refusal("caps_malformed", attr)
|
|
table = acap if name == "acap" else tcap
|
|
if num(sp[0]) in table:
|
|
raise Refusal("caps_number_reused", attr)
|
|
table[num(sp[0])] = here
|
|
elif name == "pcfg":
|
|
if here is None:
|
|
raise Refusal("caps_malformed", "pcfg at the session level")
|
|
sp = wsp(value)
|
|
n_s, rest = sp if sp else (value, "")
|
|
n = num(n_s)
|
|
if n is None:
|
|
raise Refusal("caps_malformed", attr)
|
|
if n in cfgnums:
|
|
raise Refusal("caps_config_reused", str(n))
|
|
cfgnums.add(n)
|
|
cfg, ar, tr = pcfg(n, rest)
|
|
pending.append((here, cfg, ar, tr))
|
|
elif name in ("sescap", "lcfg"):
|
|
raise Refusal("caps_unsupported", name)
|
|
if caps and "med-v0" not in options:
|
|
raise Refusal("caps_option_missing", "no med-v0")
|
|
for t in required:
|
|
if t not in ("cap-v0", "med-v0"):
|
|
raise Refusal("caps_unsupported", t)
|
|
|
|
def vis(level, sec):
|
|
return level is None or level == sec
|
|
|
|
for nums, params in mf:
|
|
for n in nums:
|
|
if n not in caps:
|
|
raise Refusal("caps_unknown_capability", "mfcap %d" % n)
|
|
caps[n]["fmtp"].append(params)
|
|
for sec, cfg, ar, tr in pending:
|
|
for alt in cfg["media"]:
|
|
for n in alt:
|
|
if n not in caps or not vis(caps[n]["level"], sec):
|
|
raise Refusal("caps_unknown_capability", "m %d" % n)
|
|
if caps[n]["rtp"] and n not in [c for c, _ in cfg["pt"]]:
|
|
raise Refusal("caps_payload_type", "no pt for %d" % n)
|
|
for c, _ in cfg["pt"]:
|
|
if c not in caps or not vis(caps[c]["level"], sec):
|
|
raise Refusal("caps_unknown_capability", "pt %d" % c)
|
|
for n in ar:
|
|
if n not in acap or not vis(acap[n], sec):
|
|
raise Refusal("caps_unknown_capability", "a %d" % n)
|
|
for n in tr:
|
|
if n not in tcap or not vis(tcap[n], sec):
|
|
raise Refusal("caps_unknown_capability", "t %d" % n)
|
|
sections[sec]["configs"].append(cfg)
|
|
return {"caps": caps, "sections": sections}
|
|
|
|
|
|
def wai_cap(c):
|
|
if not c["rtp"]:
|
|
return None
|
|
for cap, enc, rate, params in ROWS:
|
|
if enc.lower() == c["name"].lower() and rate == c["rate"] and params == c["params"]:
|
|
return cap
|
|
return None
|
|
|
|
|
|
def advertised(offer, sec):
|
|
out = []
|
|
for cfg in sorted((c for c in offer["sections"][sec]["configs"] if not c["skip"]), key=lambda c: c["number"]):
|
|
for alt in cfg["media"]:
|
|
for n in alt:
|
|
cap = wai_cap(offer["caps"][n])
|
|
if cap is None or any(a["capability"] == cap for a in out):
|
|
continue
|
|
pt = [p for c, p in cfg["pt"] if c == n][0]
|
|
out.append({"capability": cap, "config": cfg["number"], "cap": n, "payload_type": pt})
|
|
return out
|
|
|
|
|
|
def pick(adv, renditions, max_layer, floor):
|
|
for a in adv:
|
|
rs = [r for r in renditions if r["layer"] <= max_layer and r["capability"] == a["capability"]]
|
|
if rs:
|
|
r = min(rs, key=lambda r: r["layer"])
|
|
return {"layer": r["layer"], "capability": r["capability"], "config": a["config"], "cap": a["cap"], "payload_type": a["payload_type"],
|
|
"acfg": "a=acfg:%d m=%d pt=%d:%d" % (a["config"], a["cap"], a["cap"], a["payload_type"])}
|
|
return {"layer": floor, "floor": True, "acfg": None}
|
|
|
|
|
|
CAPS_ATTRS = ("a=csup:", "a=creq:", "a=rmcap:", "a=omcap:", "a=mfcap:", "a=pcfg:", "a=acap:", "a=tcap:")
|
|
|
|
|
|
def kind(cap):
|
|
return "audio" if cap.startswith("wai.audio.") else "video"
|
|
|
|
|
|
def write(offer, wanted):
|
|
eol = "\r\n" if "\r\n" in offer else "\n"
|
|
ls = sdp_lines(offer)
|
|
if any(l.startswith(CAPS_ATTRS) for l in ls):
|
|
return None, "caps_present"
|
|
ms = [i for i, l in enumerate(ls) if l.startswith("m=")]
|
|
if not ms:
|
|
return None, "no_media"
|
|
out = ls[:ms[0]] + ["a=csup:med-v0"]
|
|
used, n, c = set(), 1, 1
|
|
for k, st in enumerate(ms):
|
|
sec = ls[st:ms[k + 1] if k + 1 < len(ms) else len(ls)]
|
|
out += sec
|
|
parts = sec[0][2:].split(" ")
|
|
for cap in wanted:
|
|
row = [r for r in ROWS if r[0] == cap]
|
|
if not row or kind(cap) != parts[0]:
|
|
continue
|
|
_, enc, rate, params = row[0]
|
|
hit = None
|
|
for pt in parts[3:]:
|
|
rm = [l[len("a=rtpmap:%s " % pt):] for l in sec if l.startswith("a=rtpmap:%s " % pt)]
|
|
if rm:
|
|
p = rm[0].split("/")
|
|
if len(p) >= 2 and p[0].lower() == enc.lower() and p[1] == str(rate) and (p[2] if len(p) > 2 else None) == params:
|
|
hit = pt
|
|
break
|
|
if hit is None:
|
|
continue
|
|
used.add(cap)
|
|
out.append("a=rmcap:%d %s/%d%s" % (n, enc, rate, "/" + params if params else ""))
|
|
fm = [l[len("a=fmtp:%s " % hit):] for l in sec if l.startswith("a=fmtp:%s " % hit)]
|
|
if fm:
|
|
out.append("a=mfcap:%d %s" % (n, fm[0]))
|
|
out.append("a=pcfg:%d m=%d pt=%d:%s" % (c, n, n, hit))
|
|
n += 1
|
|
c += 1
|
|
return eol.join(out) + eol, [w for w in wanted if w not in used]
|
|
|
|
|
|
def verify(doc):
|
|
fails, seen = [], set()
|
|
for c in doc["reads"]:
|
|
try:
|
|
o = read(c["offer"])
|
|
got = {"outcome": "read", "sections": [{"media": s["media"], "advertised": advertised(o, i)} for i, s in enumerate(o["sections"])]}
|
|
except Refusal as e:
|
|
got = {"outcome": "refused", "reason": e.reason}
|
|
seen.add(e.reason)
|
|
want = {k: c[k] for k in ("outcome", "sections", "reason") if k in c}
|
|
if got != want:
|
|
fails.append("read %s: %s" % (c["name"], json.dumps(got)[:300]))
|
|
for c in doc["picks"]:
|
|
try:
|
|
o = read(c["offer"])
|
|
except Refusal as e:
|
|
fails.append("pick %s: the offer is refused (%s)" % (c["name"], e))
|
|
continue
|
|
got = pick(advertised(o, c["section"]), c["renditions"], c["max_layer"], c["floor_layer"])
|
|
if got != c["pick"]:
|
|
fails.append("pick %s: %s" % (c["name"], json.dumps(got)))
|
|
for c in doc["writes"]:
|
|
out, left = write(c["offer"], c["capabilities"])
|
|
if out is None:
|
|
if c.get("refused") != left:
|
|
fails.append("write %s: refused %s" % (c["name"], left))
|
|
continue
|
|
if out != c["written"] or left != c["left_out"]:
|
|
fails.append("write %s: %s" % (c["name"], json.dumps([out, left])[:300]))
|
|
continue
|
|
o = read(out)
|
|
for i, s in enumerate(o["sections"]):
|
|
back = [a["capability"] for a in advertised(o, i)]
|
|
if back != [w for w in c["capabilities"] if w not in left and kind(w) == s["media"]]:
|
|
fails.append("write %s: section %d read back as %s" % (c["name"], i, back))
|
|
for r in REASONS:
|
|
if r not in seen:
|
|
fails.append("nothing in the corpus gives %s" % r)
|
|
return fails
|
|
|
|
|
|
def mutations(doc):
|
|
def find(d, fam, name):
|
|
return next(c for c in d[fam] if c["name"] == name)
|
|
|
|
out = []
|
|
for c in doc["reads"]:
|
|
if c["outcome"] == "refused":
|
|
out.append(("%s read as another refusal" % c["name"], lambda d, n=c["name"]: find(d, "reads", n).update(reason="caps_other")))
|
|
else:
|
|
out.append(("%s's sections reordered" % c["name"], lambda d, n=c["name"]: find(d, "reads", n)["sections"].reverse() if len(find(d, "reads", n)["sections"]) > 1 else find(d, "reads", n).update(outcome="refused", reason="caps_malformed")))
|
|
if any(s["advertised"] for s in c["sections"]):
|
|
out.append(("%s's first advertised capability dropped" % c["name"], lambda d, n=c["name"]: [s["advertised"].pop(0) for s in find(d, "reads", n)["sections"] if s["advertised"]][:1]))
|
|
for c in doc["picks"]:
|
|
out.append(("%s's layer moved" % c["name"], lambda d, n=c["name"]: find(d, "picks", n)["pick"].update(layer=find(d, "picks", n)["pick"]["layer"] + 1)))
|
|
for c in doc["writes"]:
|
|
if c.get("written"):
|
|
out.append(("%s's text changed" % c["name"], lambda d, n=c["name"]: find(d, "writes", n).update(written=find(d, "writes", n)["written"].replace("a=pcfg:1 ", "a=pcfg:2 "))))
|
|
kept = [w for w in c["capabilities"] if w not in c["left_out"]]
|
|
if any(sum(1 for w in kept if kind(w) == k) > 1 for k in ("audio", "video")):
|
|
out.append(("%s's order swapped" % c["name"], lambda d, n=c["name"]: find(d, "writes", n)["capabilities"].reverse()))
|
|
return out
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) != 2:
|
|
print(__doc__)
|
|
return 2
|
|
with open(os.path.join(sys.argv[1], "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
bad = 0
|
|
missing = [r for r in REASONS if r not in reasons_14()]
|
|
if missing or not REASONS:
|
|
print("FAIL §14 does not list %s" % missing)
|
|
bad += 1
|
|
fails = verify(doc)
|
|
for f in fails:
|
|
print("FAIL " + f)
|
|
n = sum(len(doc[k]) for k in ("reads", "picks", "writes"))
|
|
print("%s every case gets the outcome the corpus states (%d cases; every reason of §9.5 given)" % ("PASS" if not fails else "FAIL", n))
|
|
bad += bool(fails)
|
|
muts = mutations(doc)
|
|
caught = 0
|
|
for name, change in muts:
|
|
d2 = copy.deepcopy(doc)
|
|
change(d2)
|
|
if verify(d2):
|
|
caught += 1
|
|
else:
|
|
print("FAIL the mutation '%s' is not caught" % name)
|
|
print("%s %d of %d mutated copies fail" % ("PASS" if caught == len(muts) else "FAIL", caught, len(muts)))
|
|
bad += caught != len(muts)
|
|
return 1 if bad else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|