5452b60cca
A host states a sealing key (an X25519 public key) as its screen statement's last field. A controller seals every key, text and clipboard action to it: a sealed action (type 8) keeps in the clear the type it seals, its dt_us, a clipboard text's length and a salted commitment binding its fields, its order and its author; the statement ends with one HPKE seal (RFC 9180 base mode, DHKEM(X25519, HKDF-SHA256), HKDF-SHA256, AES-128-GCM) of the actions' openings. Pointer, wheel and touch actions stay in the clear. §5.3 gains rule 8 (input_unsealed) and rule 9 (input_bad_seal), §12 a fourth table (seal_no_key, seal_not_opened, seal_unreadable, seal_commitment), §11.4 the two verification levels. wai-rs: Input::seal, Input::open, commitment, decide_opened, verify_opened; HPKE moves to a crate module behind its own feature (hpke), which telemetry_aggregate and remote_desktop both use. The corpus keeps its 42 vectors byte for byte and appends ten sealed ones, each stating the opened level; remote_desktop_verify.py checks both levels. Co-Authored-By: Claude Opus 5.5 <[email protected]>
927 lines
44 KiB
Python
927 lines
44 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the remote desktop corpus (extensions/remote-desktop.md).
|
|
|
|
Written from the extension's text. It shares no code with wai-rs, which generates the corpus,
|
|
and never runs it. It checks Ed25519 signatures with ../claim-conformance/_ed25519.py, WAI's
|
|
strict rule (SPEC §7 *Signatures*) written from RFC 8032 with the standard library. Each room
|
|
is verified by the live-rooms corpus's own second implementation, live_rooms_verify.py
|
|
(live-rooms §9), written from its text alone. Seals are opened with the HPKE base mode, X25519,
|
|
HKDF-SHA256 and AES-128-GCM of the aggregation corpus's second implementation,
|
|
telemetry_aggregate_verify.py, written from RFC 9180, RFC 7748, RFC 5869, FIPS 197 and SP
|
|
800-38D, itself checked against RFC 9180 A.1.1 here. Everything else is here:
|
|
|
|
- the grant attributes `remote_host` and `remote_control` (§3.1);
|
|
- the screen track's digest, the screen statement, its rules and its chain (§3.2 to §3.4);
|
|
- reading every statement and who signs it (§4);
|
|
- input statements and their actions (§5.1, §5.2), and the host's rules (§5.3);
|
|
- sealed actions (§5.5): the seal's form, its opening under the host's opening key, each
|
|
opening read and each commitment recomputed, at the opened level (§11.4);
|
|
- outcomes and their checks, the rules recomputed (§6);
|
|
- floor statements and the floor in effect (§7): a floor statement §7.2 refuses is stated with
|
|
its reason and changes nothing;
|
|
- latency records, their checks and their legs (§8);
|
|
- what a verification states (§11.3).
|
|
|
|
For every room it requires the outcome, reason and place, and everything else the corpus
|
|
states, at the commitment level and, under the corpus's opening key, at the opened level; of
|
|
the corpus, that every reason of the four tables of §12 appears in it and that no other file is
|
|
there. Then it mutates copies of the corpus, one defect each, and requires every copy to
|
|
fail; and it drops each of its own checks in turn and requires the corpus to catch the weakened
|
|
verifier.
|
|
|
|
python3 remote_desktop_verify.py <live-conformance/remote>
|
|
python3 remote_desktop_verify.py --room <room.json> [--opening-key <hex>]...
|
|
|
|
With `--room`, it verifies one room given in a vector's form (descriptor, grants, chains) and
|
|
prints what §11.3 states of it, as JSON: a session a deployment kept, checked from the text
|
|
alone; given opening keys (X25519 private keys, hexadecimal), at the opened level.
|
|
|
|
Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier
|
|
is caught. Python 3.8 or later; standard library only.
|
|
"""
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import struct
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
|
import live_events_verify as LEV # noqa: E402
|
|
import live_rooms_verify as LRV # noqa: E402
|
|
import telemetry_aggregate_verify as HPKE # noqa: E402
|
|
import _ed25519 as ED # noqa: E402
|
|
|
|
Refuse = LEV.Refuse
|
|
|
|
_verified = {}
|
|
_verify = ED.verify
|
|
|
|
|
|
def _memo_verify(pk, msg, sig, skip=()):
|
|
k = (bytes(pk), bytes(msg), bytes(sig), tuple(skip))
|
|
if k not in _verified:
|
|
_verified[k] = _verify(pk, msg, sig, skip)
|
|
return _verified[k]
|
|
|
|
|
|
ED.verify = _memo_verify
|
|
EXTENSION = os.path.join(HERE, "..", "extensions", "remote-desktop.md")
|
|
|
|
MAGICS = {b"WDS1": "screen", b"WDI1": "input", b"WDO1": "outcome", b"WDF1": "floor", b"WDL1": "latency"}
|
|
LATENCY_DOMAIN = b"wai:remote-latency\x01" # §8.1
|
|
SEAL_INFO = b"wai:remote-seal\x01" # §5.5
|
|
COMMIT_DOMAIN = b"wai:remote-commit\x01" # §5.5
|
|
SALT, TAG, TEXT_PAD = 16, 16, 64 # §5.5
|
|
ZERO = bytes(32)
|
|
NO_RTT = (1 << 32) - 1
|
|
META, JOIN, LEAVE, EXCHANGE = 0x01, 0x10, 0x11, 0x14
|
|
DIRECT = 2 # live-rooms §3.3, bit 1
|
|
SCREEN_PUBLISH = 4 # live-rooms §3.3, publish bit 2
|
|
TYPES = {1: "move", 2: "button", 3: "wheel", 4: "key", 5: "text", 6: "touch", 7: "clipboard"}
|
|
CLASS_OF = {1: 1, 2: 1, 3: 1, 4: 2, 5: 2, 6: 4, 7: 8} # §5.2: pointer, keys, touch, clipboard
|
|
SEALED = 8 # §5.2: a sealed action
|
|
SEALABLE = (4, 5, 7) # §5.5: key, text, clipboard
|
|
CLASS_NAMES = ((1, "pointer"), (2, "keys"), (4, "touch"), (8, "clipboard"))
|
|
REQUEST, GIVE, RELEASE, TAKE = 1, 2, 3, 4
|
|
|
|
DROP = set()
|
|
|
|
|
|
def on(name):
|
|
return name not in DROP
|
|
|
|
|
|
class Stop(Exception):
|
|
def __init__(self, reason, at):
|
|
super().__init__(reason)
|
|
self.reason, self.at = reason, at
|
|
|
|
|
|
def sha(*parts):
|
|
h = hashlib.sha256()
|
|
for p in parts:
|
|
h.update(p)
|
|
return h.digest()
|
|
|
|
|
|
def tables_of_extension():
|
|
"""§12: the codes of the first table (they refuse a room), the numbered reasons of the second
|
|
(a host states them), the codes of the third (a floor statement refused, which changes
|
|
nothing), and those of the fourth (why a seal is not opened, §5.5)."""
|
|
with open(EXTENSION, encoding="utf-8") as f:
|
|
text = f.read()
|
|
sec = text[text.index("\n## 12. Reasons"):]
|
|
sec = sec[:sec.index("\n## ", 1)]
|
|
tables = re.findall(r"((?:^\|.*\n)+)", sec, re.M)
|
|
if len(tables) != 4:
|
|
return [], {}, [], []
|
|
codes = re.findall(r"^\| `([a-z_0-9]+)` \|", tables[0], re.M)
|
|
numbered = {int(n): c for n, c in re.findall(r"^\| ([0-9]+) \| `([a-z_0-9]+)` \|", tables[1], re.M)}
|
|
floor_codes = re.findall(r"^\| `([a-z_0-9]+)` \|", tables[2], re.M)
|
|
seal_codes = re.findall(r"^\| `([a-z_0-9]+)` \|", tables[3], re.M)
|
|
return codes, numbered, floor_codes, seal_codes
|
|
|
|
|
|
REASONS, HOST_REASONS, FLOOR_REASONS, SEAL_REASONS = tables_of_extension()
|
|
ACTION_NAMES = {1: "request", 2: "give", 3: "release", 4: "take"}
|
|
|
|
|
|
class Rd:
|
|
def __init__(self, b):
|
|
self.b, self.p = b, 0
|
|
|
|
def take(self, n):
|
|
if self.p + n > len(self.b):
|
|
raise Refuse("rd_statement_malformed")
|
|
s = self.b[self.p:self.p + n]
|
|
self.p += n
|
|
return s
|
|
|
|
def u8(self):
|
|
return self.take(1)[0]
|
|
|
|
def u16(self):
|
|
return struct.unpack("<H", self.take(2))[0]
|
|
|
|
def i16(self):
|
|
return struct.unpack("<h", self.take(2))[0]
|
|
|
|
def u32(self):
|
|
return struct.unpack("<I", self.take(4))[0]
|
|
|
|
def u64(self):
|
|
return struct.unpack("<Q", self.take(8))[0]
|
|
|
|
def short(self):
|
|
return self.take(self.u8())
|
|
|
|
|
|
def bad(cond):
|
|
if cond:
|
|
raise Refuse("rd_statement_malformed")
|
|
|
|
|
|
def utf8(b):
|
|
"""RFC 3629: Python's strict decoder refuses overlong forms, surrogates and anything above
|
|
U+10FFFF."""
|
|
try:
|
|
return b.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
raise Refuse("rd_statement_malformed")
|
|
|
|
|
|
def token(b):
|
|
"""live-rooms §2: one or more bytes of 0x21 to 0x7E."""
|
|
return len(b) > 0 and all(0x21 <= c <= 0x7E for c in b)
|
|
|
|
|
|
# ---------------------------------------------------------------- §3
|
|
|
|
def attrs_of(g):
|
|
"""§3.1: (hosts, control classes); Refuse when either attribute is malformed."""
|
|
host, control = False, 0
|
|
for k, v in LRV.read_attrs(g["attrs"]) or []:
|
|
if k == "remote_host":
|
|
if v not in ("true", "false"):
|
|
raise Refuse("rd_attr_malformed")
|
|
host = v == "true"
|
|
elif k == "remote_control":
|
|
if not re.fullmatch(r"[1-9][0-9]?", v) or int(v) > 15:
|
|
raise Refuse("rd_attr_malformed")
|
|
control = int(v)
|
|
return host, control
|
|
|
|
|
|
def screen_track(room, host):
|
|
"""§3.2: SHA-256 over `<room>-<host>-screen`, a zero byte, and `0`."""
|
|
return sha(("%s-%s-screen" % (room.hex(), host.hex())).encode(), b"\x00", b"0")
|
|
|
|
|
|
def rect(r):
|
|
return {"x": r.u16(), "y": r.u16(), "w": r.u16(), "h": r.u16()}
|
|
|
|
|
|
def read_screen(r):
|
|
"""§3.3."""
|
|
s = {"prev": r.take(32), "state": r.u8(), "track": r.take(32), "width": r.u16(), "height": r.u16(), "fps": r.u16(),
|
|
"scope": r.u8(), "region": rect(r)}
|
|
s["redactions"] = [rect(r) for _ in range(r.u8())]
|
|
s.update(pointer=r.u8(), accepts=r.u8(), clip_max=r.u16(), max_age_ms=r.u32(), floor=r.u8())
|
|
s["capability"], s["codec"], s["encoder"] = r.short(), r.short(), r.short()
|
|
s["api"], s["hardware"] = r.u8(), r.u8()
|
|
# §3.3: the statement ends after `hardware`, or after a 32-byte sealing key; read_statement
|
|
# refuses any other byte after its last field.
|
|
s["seal"] = r.take(32) if len(r.b) - r.p == 32 else None
|
|
if on("screen_fields"):
|
|
reg = s["region"]
|
|
bad(s["state"] not in (1, 2) or s["width"] == 0 or s["height"] == 0 or s["fps"] == 0 or s["scope"] not in (1, 2, 3))
|
|
bad((reg["w"] == 0 or reg["h"] == 0) if s["scope"] == 3 else any(reg.values()))
|
|
bad(len(s["redactions"]) > 16)
|
|
for x in s["redactions"]:
|
|
bad(x["w"] == 0 or x["h"] == 0 or x["x"] + x["w"] > s["width"] or x["y"] + x["h"] > s["height"])
|
|
bad(s["pointer"] > 1 or s["accepts"] & ~0x7F or (s["clip_max"] > 0) != bool(s["accepts"] & 0x40))
|
|
bad(s["max_age_ms"] == 0 or s["floor"] not in (1, 2) or s["api"] not in range(1, 6) or s["hardware"] > 2)
|
|
bad(not token(s["capability"]) or not token(s["codec"]))
|
|
e = s["encoder"]
|
|
bad(not token(e) or e.count(b"@") != 1 or e[:1] == b"@" or e[-1:] == b"@")
|
|
return s
|
|
|
|
|
|
# ---------------------------------------------------------------- §5
|
|
|
|
def read_action(r):
|
|
"""§5.2: (type, fields)."""
|
|
start = r.p
|
|
ty, a = read_action_fields(r)
|
|
a["raw"] = r.b[start:r.p]
|
|
return ty, a
|
|
|
|
|
|
def read_action_fields(r):
|
|
ty = r.u8()
|
|
if ty == 1:
|
|
return ty, {"x": r.u16(), "y": r.u16()}
|
|
if ty == 2:
|
|
a = {"button": r.u8(), "down": r.u8(), "x": r.u16(), "y": r.u16()}
|
|
bad(a["button"] > 4 or a["down"] > 1)
|
|
return ty, a
|
|
if ty == 3:
|
|
a = {"dx": r.i16(), "dy": r.i16(), "x": r.u16(), "y": r.u16()}
|
|
bad(a["dx"] == 0 and a["dy"] == 0)
|
|
return ty, a
|
|
if ty == 4:
|
|
a = {"usage": r.u16(), "down": r.u8(), "modifiers": r.u8()}
|
|
bad(not 0x04 <= a["usage"] <= 0xE7 or a["down"] > 1 or a["modifiers"] & ~0x0F)
|
|
return ty, a
|
|
if ty == 5:
|
|
b = r.short()
|
|
bad(not 1 <= len(b) <= 64)
|
|
return ty, {"text": utf8(b)}
|
|
if ty == 6:
|
|
a = {"id": r.u8(), "phase": r.u8(), "x": r.u16(), "y": r.u16()}
|
|
bad(not 1 <= a["phase"] <= 4)
|
|
return ty, a
|
|
if ty == 7:
|
|
b = r.take(r.u16())
|
|
return ty, {"text": utf8(b), "bytes": len(b)}
|
|
if ty == SEALED:
|
|
a = {"of": r.u8(), "length": r.u16(), "commit": r.take(32)}
|
|
if on("sealed_of"):
|
|
bad(a["of"] not in SEALABLE or (a["of"] != 7 and a["length"] != 0))
|
|
return ty, a
|
|
raise Refuse("rd_statement_malformed")
|
|
|
|
|
|
def opening_len(ty, a):
|
|
"""§5.5: the length of a sealed action's opening: the salt, then 4 for a key, 1 + 64 for a
|
|
text, a clipboard text's `length`."""
|
|
if ty != SEALED:
|
|
return 0
|
|
return SALT + {4: 4, 5: 1 + TEXT_PAD}.get(a["of"], a["length"])
|
|
|
|
|
|
def acts_as(ty, a):
|
|
"""§5.2: a sealed action is of the type it seals."""
|
|
return a["of"] if ty == SEALED else ty
|
|
|
|
|
|
def read_input(r):
|
|
"""§5.1, and the seal of §5.5."""
|
|
i = {"host": r.take(32), "screen": r.take(32), "actions": [], "seal": None}
|
|
n = r.u8()
|
|
bad(not 1 <= n <= 32)
|
|
for _ in range(n):
|
|
dt = r.u32()
|
|
if on("dt_order"):
|
|
bad(i["actions"] and dt > i["actions"][-1][0])
|
|
ty, a = read_action(r)
|
|
i["actions"].append((dt, ty, a))
|
|
i["clear"] = r.b[:r.p]
|
|
if any(ty == SEALED for _, ty, _ in i["actions"]):
|
|
enc = r.take(32)
|
|
want = TAG + sum(opening_len(ty, a) for _, ty, a in i["actions"])
|
|
if on("seal_length"):
|
|
bad(len(r.b) - r.p != want)
|
|
i["seal"] = (enc, r.take(len(r.b) - r.p))
|
|
return i
|
|
|
|
|
|
def read_outcome(r):
|
|
"""§6.1."""
|
|
o = {"input": r.take(32), "outcome": r.u8(), "reason": r.u8(), "applied_at": r.u64(), "captured_at": r.u64(),
|
|
"content": r.take(32), "rtt_us": r.u32()}
|
|
bad(o["outcome"] not in (1, 2) or o["reason"] > 9)
|
|
return o
|
|
|
|
|
|
def read_floor(r):
|
|
"""§7.2."""
|
|
f = {"host": r.take(32), "action": r.u8(), "to": r.take(32)}
|
|
bad(f["action"] not in (1, 2, 3, 4) or (f["to"] != ZERO) != (f["action"] == GIVE))
|
|
return f
|
|
|
|
|
|
def read_latency(r):
|
|
"""§8.1."""
|
|
start = r.p
|
|
l = {"room": r.take(32), "viewer": r.take(32), "host": r.take(32), "input": r.take(32), "action": r.u8(),
|
|
"outcome": r.take(32), "track": r.take(32), "group": r.u64(), "object": r.u64(), "content": r.take(32),
|
|
"input_at": r.u64(), "presented_at": r.u64(), "source": r.u8(), "latency_us": r.u32(), "rtt_us": r.u32()}
|
|
l["body"] = r.b[:r.p]
|
|
l["sig"] = r.take(64)
|
|
bad(l["source"] not in (1, 2, 3))
|
|
del start
|
|
return l
|
|
|
|
|
|
def read_statement(payload):
|
|
"""§4.1, §4.2: None for a payload not this extension's; else (kind, fields)."""
|
|
kind = MAGICS.get(payload[:4])
|
|
if kind is None:
|
|
return None
|
|
r = Rd(payload)
|
|
r.take(4)
|
|
bad(r.u8() != 1)
|
|
f = {"screen": read_screen, "input": read_input, "outcome": read_outcome, "floor": read_floor, "latency": read_latency}[kind](r)
|
|
bad(r.p != len(payload))
|
|
return kind, f
|
|
|
|
|
|
def coords(ty, a):
|
|
return (a["x"], a["y"]) if ty in (1, 2, 3, 6) else None
|
|
|
|
|
|
def rules(inp, screen_digest, screen, control, holds_floor, me):
|
|
"""§5.3 steps 1 to 6 at the input's place: the first reason's number, or 0."""
|
|
if on("rule_screen") and (inp["host"] != me or screen is None or inp["screen"] != screen_digest):
|
|
return 1
|
|
if screen is None:
|
|
return 1
|
|
classes = 0
|
|
for _, ty, a in inp["actions"]:
|
|
classes |= CLASS_OF[acts_as(ty, a)]
|
|
if on("rule_power") and classes & ~control:
|
|
return 2
|
|
if on("rule_floor") and screen["floor"] == 2 and not holds_floor:
|
|
return 3
|
|
if on("rule_paused") and screen["state"] == 2:
|
|
return 4
|
|
if on("rule_accepts") and any(not screen["accepts"] >> (acts_as(ty, a) - 1) & 1 for _, ty, a in inp["actions"]):
|
|
return 5
|
|
for _, ty, a in inp["actions"]:
|
|
c = coords(ty, a)
|
|
clip = a["bytes"] if ty == 7 else a["length"] if ty == SEALED and a["of"] == 7 else 0
|
|
if on("rule_bounds") and ((c and (c[0] >= screen["width"] or c[1] >= screen["height"])) or clip > screen["clip_max"]):
|
|
return 6
|
|
return 0
|
|
|
|
|
|
def unsealed(inp, screen):
|
|
"""§5.3 step 8: a key, text or clipboard action in the clear to a screen with a sealing key."""
|
|
return on("rule_unsealed") and screen is not None and screen["seal"] is not None and any(ty in SEALABLE for _, ty, _ in inp["actions"])
|
|
|
|
|
|
# ---------------------------------------------------------------- §5.5
|
|
|
|
_opened = {}
|
|
|
|
|
|
def open_seal(inp, room, author, screen, keys):
|
|
"""§5.5, rule 9 of §5.3, against the screen statement the input names: ("opened", the opened
|
|
statement), (a reason of §12's fourth table, None), or ("unknown", None) when no opening key
|
|
for its sealing key is given (the commitment level gives none)."""
|
|
if screen is None or screen["seal"] is None:
|
|
return ("seal_no_key", None) if on("seal_no_key") else ("unknown", None)
|
|
sk = next((k for k in keys if HPKE.x25519(k, HPKE.BASE) == screen["seal"]), None)
|
|
if sk is None:
|
|
return "unknown", None
|
|
enc, ct = inp["seal"]
|
|
memo = (sk, enc, ct, room, author, inp["clear"])
|
|
if memo in _opened and not DROP:
|
|
return _opened[memo]
|
|
out = _open(inp, room, author, sk, enc, ct)
|
|
if not DROP:
|
|
_opened[memo] = out
|
|
return out
|
|
|
|
|
|
def _open(inp, room, author, sk, enc, ct):
|
|
info = SEAL_INFO + room + inp["host"]
|
|
aad = author + inp["clear"]
|
|
try:
|
|
pt = HPKE.hpke_open(enc, sk, info, aad, ct)
|
|
except ValueError:
|
|
if on("seal_open"):
|
|
return "seal_not_opened", None
|
|
return "opened", inp["clear"]
|
|
at, actions, commits = 0, [], []
|
|
for k, (dt, ty, a) in enumerate(inp["actions"]):
|
|
if ty != SEALED:
|
|
actions.append(a["raw"])
|
|
continue
|
|
n = opening_len(ty, a)
|
|
opening = pt[at:at + n]
|
|
at += n
|
|
f = opening[SALT:]
|
|
raw = read_opening(a["of"], f, len(opening) == n)
|
|
if raw is None:
|
|
if on("seal_read"):
|
|
return "seal_unreadable", None
|
|
raw = bytes([a["of"]])
|
|
actions.append(raw)
|
|
commits.append((k, dt, a, opening))
|
|
if on("seal_read") and at != len(pt):
|
|
return "seal_unreadable", None
|
|
for k, dt, a, opening in commits:
|
|
c = sha(COMMIT_DOMAIN, room, author, inp["host"], inp["screen"], bytes([k]), struct.pack("<I", dt), bytes([a["of"]]), opening)
|
|
if on("seal_commit") and c != a["commit"]:
|
|
return "seal_commitment", None
|
|
opened = inp["clear"][:70] + b"".join(struct.pack("<I", dt) + raw for (dt, _, _), raw in zip(inp["actions"], actions))
|
|
return "opened", opened
|
|
|
|
|
|
def read_opening(of, f, whole):
|
|
"""§5.5: the action an opening's fields (after the salt) read as, in its clear form (§5.2), or
|
|
None when they do not read."""
|
|
if not whole:
|
|
return None
|
|
if of == 4:
|
|
usage, down, mods = struct.unpack("<HBB", f)
|
|
if not 0x04 <= usage <= 0xE7 or down > 1 or mods & ~0x0F:
|
|
return None
|
|
return bytes([4]) + f
|
|
if of == 5:
|
|
n = f[0]
|
|
if not 1 <= n <= 64 or any(f[1 + n:]):
|
|
return None
|
|
try:
|
|
f[1:1 + n].decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return None
|
|
return bytes([5, n]) + f[1:1 + n]
|
|
try:
|
|
f.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return None
|
|
return bytes([7]) + struct.pack("<H", len(f)) + f
|
|
|
|
|
|
def stale(inp, t, screen, n):
|
|
"""§5.3 step 7."""
|
|
first = t - inp["actions"][0][0]
|
|
return on("rule_stale") and n - first > screen["max_age_ms"] * 1000
|
|
|
|
|
|
# ---------------------------------------------------------------- §11
|
|
|
|
def verify(descriptor, grants, chains, keys=None):
|
|
"""§11.2: a dict of what the corpus states (§11.3), at the commitment level, or with opening
|
|
keys (X25519 private keys) at the opened level (§11.4)."""
|
|
room = LRV.verify_room(descriptor, grants, None, chains) # step 1
|
|
if room["outcome"] == "refuse":
|
|
return {"outcome": "refuse", "reason": room["reason"], "at": ["room"] + room["at"]}
|
|
gap = room["outcome"] == "gap"
|
|
d = LRV.read_descriptor(descriptor)
|
|
room_id = d["room"]
|
|
held_grants, _ = LRV.check_grants(d, grants)
|
|
room_keys = (d["authority"], d["clock"])
|
|
try:
|
|
attrs = {}
|
|
for i, g in enumerate(held_grants): # step 2
|
|
try:
|
|
host, control = attrs_of(g)
|
|
except Refuse as e:
|
|
if on("attrs"):
|
|
raise Stop(e.reason, ["grant", i])
|
|
host, control = False, 0
|
|
attrs[g["digest"]] = {"host": host and g["publish"] & SCREEN_PUBLISH != 0, "control": control,
|
|
"direct": g["powers"] & DIRECT != 0}
|
|
records, seen = [], set()
|
|
for c, i, _ in room["order"]:
|
|
ev = LEV.read_record(chains[c][i])
|
|
if ev["digest"] in seen:
|
|
continue
|
|
seen.add(ev["digest"])
|
|
ev["chain"], ev["index"] = c, i
|
|
records.append(ev)
|
|
in_effect, hosts = {}, {}
|
|
screens, inputs, floor, floor_refused, later = [], [], [], [], []
|
|
named = {}
|
|
|
|
def attrs_now(p):
|
|
g = in_effect.get(p)
|
|
return attrs.get(g, {"host": False, "control": 0, "direct": False})
|
|
|
|
def lose(p, t):
|
|
for h, st in hosts.items():
|
|
if st["holder"] == p:
|
|
st["holder"] = None
|
|
floor.append({"host": h.hex(), "t": t, "holder": None})
|
|
|
|
for ev in records: # step 3
|
|
t, author = ev["t"], ev["principal"]
|
|
if author in room_keys and ev["kind"] in (JOIN, LEAVE, EXCHANGE):
|
|
pl = LEV.read_payload(ev["kind"], ev["payload"], 1 << 20)
|
|
x = LRV.read_detail(ev["kind"], pl["detail"])
|
|
who = pl["subject"]
|
|
if ev["kind"] == JOIN:
|
|
in_effect[who] = x["grant"]
|
|
elif ev["kind"] == EXCHANGE:
|
|
in_effect[who] = x["new"]
|
|
if on("floor_loss") and attrs_now(who)["control"] == 0:
|
|
lose(who, t)
|
|
else:
|
|
in_effect.pop(who, None)
|
|
if on("floor_loss"):
|
|
lose(who, t)
|
|
if who in hosts and hosts[who]["holder"] is not None:
|
|
hosts[who]["holder"] = None
|
|
floor.append({"host": who.hex(), "t": t, "holder": None})
|
|
continue
|
|
if ev["kind"] != META:
|
|
continue
|
|
at = ["record", ev["chain"], ev["index"]]
|
|
try:
|
|
s = read_statement(ev["payload"])
|
|
if s is None:
|
|
continue
|
|
kind, f = s
|
|
is_key = author in room_keys
|
|
me = attrs_now(author)
|
|
if kind == "screen":
|
|
if on("who_screen") and (is_key or not me["host"]):
|
|
raise Refuse("rd_not_permitted")
|
|
if on("track") and f["track"] != screen_track(room_id, author):
|
|
raise Refuse("screen_wrong_track")
|
|
st = hosts.setdefault(author, {"screen": None, "digest": ZERO, "holder": None})
|
|
if on("screen_link") and f["prev"] != st["digest"]:
|
|
raise Refuse("screen_unlinked")
|
|
st["screen"], st["digest"] = f, ev["digest"]
|
|
named[(author, ev["digest"])] = f
|
|
screens.append({"digest": ev["digest"].hex(), "host": author.hex(), "t": t,
|
|
"state": "live" if f["state"] == 1 else "paused", "width": f["width"], "height": f["height"],
|
|
"fps": f["fps"], "scope": f["scope"], "floor": "open" if f["floor"] == 1 else "floor",
|
|
"accepts": [TYPES[k] for k in range(1, 8) if f["accepts"] >> (k - 1) & 1],
|
|
"capability": f["capability"].decode(), "codec": f["codec"].decode(), "encoder": f["encoder"].decode(),
|
|
"api": f["api"], "hardware": f["hardware"]})
|
|
elif kind == "input":
|
|
if on("who_input") and is_key:
|
|
raise Refuse("rd_not_permitted")
|
|
st = hosts.get(f["host"])
|
|
scr = st["screen"] if st else None
|
|
first = rules(f, st["digest"] if st else None, scr, me["control"], bool(st) and st["holder"] == author, f["host"])
|
|
classes = 0
|
|
for _, ty, a in f["actions"]:
|
|
classes |= CLASS_OF[acts_as(ty, a)]
|
|
# Rule 9 against the screen statement the input names (§5.5).
|
|
seal = None
|
|
if f["seal"] is not None:
|
|
seal = open_seal(f, room_id, author, named.get((f["host"], f["screen"])), keys or [])
|
|
inputs.append({"digest": ev["digest"], "author": author, "t": t, "input": f, "first": first,
|
|
"unsealed": unsealed(f, scr), "seal": seal,
|
|
"sealed": sum(1 for _, ty, _ in f["actions"] if ty == SEALED),
|
|
"screen": scr, "classes": classes, "fate": {"outcome": "unanswered"}})
|
|
elif kind == "floor":
|
|
# §7.2: a statement its rules refuse is recorded, stated with its reason, and
|
|
# changes nothing; the floor is computed without it (§7.3, §11.3).
|
|
st = hosts.get(f["host"])
|
|
a = f["action"]
|
|
why = None
|
|
if on("floor_host") and (st is None or st["screen"] is None):
|
|
why = "floor_not_permitted"
|
|
else:
|
|
st = st or {"holder": None}
|
|
steward = is_key or author == f["host"] or me["direct"]
|
|
if a == REQUEST:
|
|
if on("floor_request") and (is_key or me["control"] == 0):
|
|
why = "floor_not_permitted"
|
|
elif a == GIVE:
|
|
if on("floor_give") and not (steward or st["holder"] == author):
|
|
why = "floor_not_permitted"
|
|
elif on("floor_eligible") and (f["to"] == f["host"] or attrs_now(f["to"])["control"] == 0):
|
|
why = "floor_not_eligible"
|
|
elif a == RELEASE:
|
|
if on("floor_release") and st["holder"] != author:
|
|
why = "floor_not_holder"
|
|
elif on("floor_take") and not steward:
|
|
why = "floor_not_permitted"
|
|
if why is not None:
|
|
floor_refused.append({"digest": ev["digest"].hex(), "author": author.hex(), "host": f["host"].hex(), "t": t,
|
|
"action": ACTION_NAMES[a], "reason": why})
|
|
elif a == GIVE:
|
|
st["holder"] = f["to"]
|
|
floor.append({"host": f["host"].hex(), "t": t, "holder": f["to"].hex()})
|
|
elif a == RELEASE:
|
|
st["holder"] = None
|
|
floor.append({"host": f["host"].hex(), "t": t, "holder": None})
|
|
elif a == TAKE and st["holder"] is not None:
|
|
st["holder"] = None
|
|
floor.append({"host": f["host"].hex(), "t": t, "holder": None})
|
|
elif kind == "outcome":
|
|
later.append(("outcome", at, ev, f))
|
|
else:
|
|
if on("who_latency") and is_key:
|
|
raise Refuse("rd_not_permitted")
|
|
if f["room"] != room_id:
|
|
raise Refuse("wrong_room")
|
|
if on("latency_key") and f["viewer"] != author:
|
|
raise Refuse("latency_wrong_key")
|
|
if on("latency_sig"):
|
|
why = ED.verify(f["viewer"], LATENCY_DOMAIN + f["body"], f["sig"])
|
|
if why:
|
|
raise Refuse(why)
|
|
later.append(("latency", at, ev, f))
|
|
except Refuse as e:
|
|
raise Stop(e.reason, at)
|
|
by_input = {x["digest"]: x for x in inputs}
|
|
outcomes = {ev["digest"]: (ev, f) for k, _, ev, f in later if k == "outcome"}
|
|
answered, latencies = set(), []
|
|
for k, at, ev, f in later: # §6.2, §8.2
|
|
try:
|
|
if k == "outcome":
|
|
x = by_input.get(f["input"])
|
|
if x is None:
|
|
if gap:
|
|
continue
|
|
raise Refuse("outcome_unknown_input")
|
|
if on("who_outcome") and ev["principal"] != x["input"]["host"]:
|
|
raise Refuse("rd_not_permitted")
|
|
if on("duplicate") and f["input"] in answered:
|
|
raise Refuse("outcome_duplicate")
|
|
answered.add(f["input"])
|
|
an = ev["anchor"]
|
|
if f["outcome"] == 1:
|
|
ok = (f["reason"] == 0 and an is not None and an[0] == screen_track(room_id, x["input"]["host"]) and an[2] == 0
|
|
and f["content"] != ZERO and x["t"] <= f["applied_at"] <= f["captured_at"] <= ev["t"])
|
|
else:
|
|
ok = (1 <= f["reason"] <= 9 and an is None and f["applied_at"] == 0 and f["captured_at"] == 0
|
|
and f["content"] == ZERO and ev["t"] >= x["t"])
|
|
if on("consistent") and not ok:
|
|
raise Refuse("outcome_invalid")
|
|
want = x["first"]
|
|
if want == 0 and stale(x["input"], x["t"], x["screen"], ev["t"]):
|
|
want = 7
|
|
if want == 0 and x["unsealed"]:
|
|
want = 8
|
|
# Rule 9: made where the seal's result is known, and otherwise either (§6.2 step 5).
|
|
allowed = {want}
|
|
if want == 0 and x["seal"] is not None:
|
|
allowed = {9} if x["seal"][0] not in ("opened", "unknown") else {0} if x["seal"][0] == "opened" else {0, 9}
|
|
stated = 0 if f["outcome"] == 1 else f["reason"]
|
|
if on("recompute") and stated not in allowed:
|
|
raise Refuse("outcome_mismatch")
|
|
x["fate"] = ({"outcome": "applied", "group": an[1], "object": an[2]} if f["outcome"] == 1 and an
|
|
else {"outcome": "refused", "reason": HOST_REASONS.get(f["reason"], "?")})
|
|
else:
|
|
x = by_input.get(f["input"])
|
|
if x is None and gap:
|
|
latencies.append({"digest": sha(f["body"]).hex(), "viewer": f["viewer"].hex(), "source": f["source"],
|
|
"latency_us": f["latency_us"], "legs": "unchecked"})
|
|
continue
|
|
if on("latency_input") and (x is None or x["author"] != f["viewer"] or x["input"]["host"] != f["host"]
|
|
or f["action"] >= len(x["input"]["actions"])):
|
|
raise Refuse("latency_unknown_input")
|
|
o = outcomes.get(f["outcome"])
|
|
if o is None and gap:
|
|
latencies.append({"digest": sha(f["body"]).hex(), "viewer": f["viewer"].hex(), "source": f["source"],
|
|
"latency_us": f["latency_us"], "legs": "unchecked"})
|
|
continue
|
|
dt = x["input"]["actions"][f["action"]][0]
|
|
if on("latency_frame") and not (o is not None and o[1]["input"] == f["input"] and o[1]["outcome"] == 1
|
|
and o[0]["anchor"] == (f["track"], f["group"], f["object"])
|
|
and o[1]["content"] == f["content"] and f["input_at"] == x["t"] - dt):
|
|
raise Refuse("latency_mismatch")
|
|
if on("latency_ahead") and f["presented_at"] > ev["t"]:
|
|
raise Refuse("latency_ahead")
|
|
if on("latency_sum") and (f["presented_at"] < f["input_at"] or f["presented_at"] - f["input_at"] != f["latency_us"]):
|
|
raise Refuse("latency_mismatch")
|
|
oo = o[1]
|
|
unc = None if f["rtt_us"] == NO_RTT or oo["rtt_us"] == NO_RTT else (f["rtt_us"] + oo["rtt_us"]) // 2
|
|
latencies.append({"digest": sha(f["body"]).hex(), "viewer": f["viewer"].hex(), "source": f["source"],
|
|
"latency_us": f["latency_us"],
|
|
"legs": {"up": oo["applied_at"] - f["input_at"], "host": oo["captured_at"] - oo["applied_at"],
|
|
"down": f["presented_at"] - oo["captured_at"], "uncertainty": unc}})
|
|
except Refuse as e:
|
|
raise Stop(e.reason, at)
|
|
except Stop as s:
|
|
return {"outcome": "refuse", "reason": s.reason, "at": s.at}
|
|
totals = {}
|
|
for x in inputs:
|
|
if gap and x["fate"]["outcome"] == "unanswered":
|
|
x["fate"] = {"outcome": "unchecked"}
|
|
h = x["input"]["host"].hex()
|
|
tt = totals.setdefault(h, {"host": h, "applied": 0, "refused": {HOST_REASONS[n]: 0 for n in sorted(HOST_REASONS)},
|
|
"unanswered": 0, "unchecked": 0})
|
|
fo = x["fate"]["outcome"]
|
|
if fo == "refused":
|
|
tt["refused"][x["fate"]["reason"]] += 1
|
|
else:
|
|
tt[fo] += 1
|
|
# §11.3: rules 1 to 7 always, rules 8 and 9 where they refused.
|
|
for tt in totals.values():
|
|
tt["refused"] = {k: v for k, v in tt["refused"].items() if v or k not in (HOST_REASONS.get(8), HOST_REASONS.get(9))}
|
|
out = {"outcome": "gap" if gap else "accept", "room": room_id.hex(),
|
|
"gaps": room.get("gaps", []) if gap else [],
|
|
"screens": screens,
|
|
"inputs": [input_fact(x, keys is not None) for x in inputs],
|
|
"floor": floor, "latencies": latencies, "totals": [totals[h] for h in sorted(totals)]}
|
|
if floor_refused:
|
|
out["floor_refused"] = floor_refused
|
|
return out
|
|
|
|
|
|
def input_fact(x, opened_level):
|
|
"""§11.3: an input as a verification states it; at the opened level, a sealed input's seal."""
|
|
o = {"digest": x["digest"].hex(), "author": x["author"].hex(), "host": x["input"]["host"].hex(), "t": x["t"],
|
|
"classes": [n for b, n in CLASS_NAMES if x["classes"] & b]}
|
|
if x["sealed"]:
|
|
o["sealed"] = x["sealed"]
|
|
o["fate"] = x["fate"]
|
|
if opened_level and x["seal"] is not None:
|
|
why, opened = x["seal"]
|
|
o["seal"] = "unopened" if why == "unknown" else why
|
|
if why == "opened":
|
|
o["opened"] = opened.hex()
|
|
return o
|
|
|
|
|
|
def run_vector(v, keys=None):
|
|
return verify(bytes.fromhex(v["descriptor"]), [bytes.fromhex(g) for g in v["grants"]], [[bytes.fromhex(r) for r in c] for c in v["chains"]], keys)
|
|
|
|
|
|
STATED = ("outcome", "reason", "at", "room", "gaps", "screens", "inputs", "floor", "floor_refused", "latencies", "totals")
|
|
|
|
|
|
def given(got, seen):
|
|
if got.get("outcome") == "refuse":
|
|
seen.add(got["reason"])
|
|
for x in got.get("inputs", []):
|
|
if x["fate"]["outcome"] == "refused":
|
|
seen.add(x["fate"]["reason"])
|
|
if x.get("seal", "opened") not in ("opened", "unopened"):
|
|
seen.add(x["seal"])
|
|
for x in got.get("floor_refused", []):
|
|
seen.add(x["reason"])
|
|
|
|
|
|
def check(doc, quiet=False, first=False):
|
|
"""Every vector of `doc` against this verifier, at both levels: the failures (with `first`, the
|
|
first alone), and every reason the corpus gives."""
|
|
fails, seen = [], set()
|
|
try:
|
|
keys = [bytes.fromhex(doc["opening_key"])]
|
|
except (KeyError, ValueError, TypeError):
|
|
return ["the corpus states no opening key"], seen
|
|
for v in doc["vectors"]:
|
|
if first and fails:
|
|
return fails, seen
|
|
try:
|
|
got = run_vector(v)
|
|
opened = run_vector(v, keys)
|
|
except Exception as e: # a copy that does not read is a failure to verify it
|
|
fails.append("%s: does not verify (%r)" % (v["name"], e))
|
|
continue
|
|
badk = [k for k in STATED if (k in got or k in v) and got.get(k) != v.get(k)]
|
|
# The opened level: what the vector states of it, or, where it states nothing of it, what
|
|
# the commitment level gives (§13).
|
|
want = v.get("opened", got)
|
|
bado = ["opened." + k for k in STATED if (k in opened or k in want) and opened.get(k) != (want.get(k) if isinstance(want, dict) else None)]
|
|
given(got, seen)
|
|
given(opened, seen)
|
|
if not quiet:
|
|
what = got["outcome"] + ("" if got["outcome"] != "refuse" else " " + got["reason"] + " at %s" % got["at"])
|
|
if "opened" in v:
|
|
what += "; opened: " + opened["outcome"] + ("" if opened["outcome"] != "refuse" else " " + opened["reason"] + " at %s" % opened["at"])
|
|
print(("PASS" if not badk and not bado else "FAIL") + " room %s: %s" % (v["name"], what))
|
|
for k in badk:
|
|
fails.append("%s: %s is %s, the corpus states %s" % (v["name"], k, json.dumps(got.get(k))[:200], json.dumps(v.get(k))[:200]))
|
|
for k in bado:
|
|
fails.append("%s: %s differs" % (v["name"], k))
|
|
return fails, seen
|
|
|
|
|
|
def mutations(doc):
|
|
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
|
vn = {v["name"]: k for k, v in enumerate(doc["vectors"])}
|
|
fs, op, gp = vn["floor_session"], vn["open_handover"], vn["floor_session_gap"]
|
|
|
|
def mutate(label, f):
|
|
d = copy.deepcopy(doc)
|
|
f(d)
|
|
return label, d
|
|
|
|
def vs(d):
|
|
return d["vectors"]
|
|
|
|
def flip(h, i):
|
|
b = bytearray(bytes.fromhex(h))
|
|
b[i] ^= 1
|
|
return b.hex()
|
|
|
|
def flip_record(d, k, chain, rec, i):
|
|
vs(d)[k]["chains"][chain][rec] = flip(vs(d)[k]["chains"][chain][rec], i)
|
|
|
|
host = doc["principals"]["host"]
|
|
ss = vn["sealed_session"]
|
|
return [
|
|
mutate("an applied input stated refused", lambda d: vs(d)[fs]["inputs"][0].update(fate={"outcome": "refused", "reason": "input_no_floor"})),
|
|
mutate("a refusal's stated reason", lambda d: vs(d)[fs]["inputs"][3]["fate"].update(reason="input_not_permitted")),
|
|
mutate("an unanswered input stated applied", lambda d: vs(d)[fs]["inputs"][10].update(fate={"outcome": "applied", "group": 9, "object": 0})),
|
|
mutate("a stated frame", lambda d: vs(d)[fs]["inputs"][1]["fate"].update(group=7)),
|
|
mutate("a stated floor holder", lambda d: vs(d)[fs]["floor"][1].update(holder=host)),
|
|
mutate("a floor change unstated", lambda d: vs(d)[fs]["floor"].pop()),
|
|
mutate("a stated leg", lambda d: vs(d)[fs]["latencies"][0]["legs"].update(up=vs(d)[fs]["latencies"][0]["legs"]["up"] + 1)),
|
|
mutate("a stated uncertainty", lambda d: vs(d)[fs]["latencies"][1]["legs"].update(uncertainty=0)),
|
|
mutate("a stated total", lambda d: vs(d)[fs]["totals"][0].update(applied=5)),
|
|
mutate("a stated encoder", lambda d: vs(d)[fs]["screens"][0].update(encoder="another@2")),
|
|
mutate("the handover's second applied input stated refused", lambda d: vs(d)[op]["inputs"][3].update(fate={"outcome": "refused", "reason": "input_not_permitted"})),
|
|
mutate("the gap's unchecked input stated unanswered", lambda d: vs(d)[gp]["inputs"][-3].update(fate={"outcome": "unanswered"})),
|
|
mutate("a byte of a latency record's own signature", lambda d: flip_record(d, fs, 2, 1, -70)),
|
|
mutate("a refusal's reason", lambda d: vs(d)[vn["applies_stale"]].update(reason="outcome_invalid")),
|
|
mutate("a refusal's place", lambda d: vs(d)[vn["screen_unlinked"]].update(at=["record", 1, 5])),
|
|
mutate("a refused floor statement's reason", lambda d: vs(d)[vn["floor_late_release"]]["floor_refused"][0].update(reason="floor_not_permitted")),
|
|
mutate("a refused floor statement unstated", lambda d: vs(d)[vn["floor_given_by_viewer"]].pop("floor_refused")),
|
|
mutate("a floor statement stated refused that the rules accept", lambda d: vs(d)[fs].update(floor_refused=vs(d)[vn["floor_give_by_non_holder"]]["floor_refused"])),
|
|
mutate("a refused give stated as passing the floor", lambda d: vs(d)[vn["floor_give_by_non_holder"]]["floor"].insert(2, {"host": host, "t": vs(d)[vn["floor_give_by_non_holder"]]["floor_refused"][0]["t"], "holder": doc["principals"]["alice"]})),
|
|
mutate("a sealed input's count of sealed actions", lambda d: vs(d)[ss]["inputs"][0].update(sealed=4)),
|
|
mutate("a text in the clear to a sealing key stated applied", lambda d: vs(d)[ss]["inputs"][2].update(fate={"outcome": "applied", "group": 9, "object": 0})),
|
|
mutate("a refusal by rule 9 stated as rule 2's", lambda d: vs(d)[ss]["inputs"][3]["fate"].update(reason="input_not_permitted")),
|
|
mutate("a host's count of refusals by rule 9", lambda d: vs(d)[ss]["totals"][0]["refused"].update(input_bad_seal=4)),
|
|
mutate("a host's count of refusals by rule 8 left out", lambda d: vs(d)[ss]["totals"][0]["refused"].pop("input_unsealed")),
|
|
mutate("an opened seal stated not opened", lambda d: vs(d)[ss]["opened"]["inputs"][0].update(seal="seal_not_opened")),
|
|
mutate("a seal to the wrong key stated a commitment that does not match", lambda d: vs(d)[ss]["opened"]["inputs"][3].update(seal="seal_commitment")),
|
|
mutate("a byte of an opened statement", lambda d: vs(d)[ss]["opened"]["inputs"][0].update(opened=flip(vs(d)[ss]["opened"]["inputs"][0]["opened"], -1))),
|
|
mutate("a seal the opened level refuses stated accepted there", lambda d: vs(d)[vn["sealed_applies_wrong_key"]].update(opened={k: v for k, v in vs(d)[vn["sealed_applies_wrong_key"]].items() if k in STATED})),
|
|
mutate("the opened level stated of a vector with no seal", lambda d: vs(d)[fs].update(opened={"outcome": "refuse", "reason": "outcome_mismatch", "at": ["record", 1, 3]})),
|
|
mutate("another opening key", lambda d: d.update(opening_key="11" * 32)),
|
|
]
|
|
|
|
|
|
CHECKS = ("attrs", "screen_fields", "dt_order", "who_screen", "track", "screen_link", "who_input", "floor_host", "floor_request",
|
|
"floor_give", "floor_eligible", "floor_release", "floor_take", "floor_loss", "who_latency", "latency_key", "latency_sig",
|
|
"who_outcome", "duplicate", "consistent", "recompute", "rule_screen", "rule_power", "rule_floor", "rule_paused",
|
|
"rule_accepts", "rule_bounds", "rule_stale", "latency_input", "latency_frame", "latency_ahead", "latency_sum",
|
|
"sealed_of", "seal_length", "rule_unsealed", "seal_no_key", "seal_open", "seal_read", "seal_commit")
|
|
|
|
|
|
def main(corpus_dir):
|
|
fails = []
|
|
st = ED._self_test()
|
|
print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
|
|
fails += ["ed25519 self-test"] if st else []
|
|
st = HPKE.crypto_self_test()
|
|
print(("PASS" if not st else "FAIL") + " X25519 (RFC 7748 §5.2) and HPKE base mode (RFC 9180 A.1.1) " + ",".join(st))
|
|
fails += ["HPKE self-test"] if st else []
|
|
ok = len(REASONS) == 13 and sorted(HOST_REASONS) == list(range(1, 10)) and len(FLOOR_REASONS) == 3 and len(SEAL_REASONS) == 4
|
|
print(("PASS" if ok else "FAIL") + " §12 read from the extension: %d reasons, %d host refusals, %d floor refusals, %d seal reasons" % (len(REASONS), len(HOST_REASONS), len(FLOOR_REASONS), len(SEAL_REASONS)))
|
|
fails += [] if ok else ["§12 does not read"]
|
|
names = sorted(n for n in os.listdir(corpus_dir) if n != "README.md")
|
|
ok = names == ["vectors.json"]
|
|
print(("PASS" if ok else "FAIL") + " the corpus holds vectors.json and README.md alone")
|
|
fails += [] if ok else ["other files: %s" % names]
|
|
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
f2, seen = check(doc)
|
|
fails += f2
|
|
missing = [r for r in REASONS + [HOST_REASONS[n] for n in sorted(HOST_REASONS)] + FLOOR_REASONS + SEAL_REASONS if r not in seen]
|
|
print(("PASS" if not missing else "FAIL") + " every reason of the four tables of §12 appears in the corpus")
|
|
fails += ["the corpus never gives %s" % ", ".join(missing)] if missing else []
|
|
for label, d in mutations(doc):
|
|
caught, _ = check(d, quiet=True, first=True)
|
|
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:90] if caught else ""))
|
|
if not caught:
|
|
fails.append("a mutated copy verifies: %s" % label)
|
|
for name in CHECKS:
|
|
DROP.add(name)
|
|
try:
|
|
caught, _ = check(doc, quiet=True, first=True)
|
|
finally:
|
|
DROP.discard(name)
|
|
print(("PASS" if caught else "FAIL") + " the corpus catches the check dropped: %s%s" % (name, " (%s)" % caught[0][:90] if caught else ""))
|
|
if not caught:
|
|
fails.append("the corpus does not catch the check dropped: %s" % name)
|
|
print("remote_desktop_verify: %s" % ("PASS" if not fails else "FAIL (%d)" % len(fails)))
|
|
for f in fails:
|
|
print(" " + f)
|
|
return 0 if not fails else 1
|
|
|
|
|
|
def room_main(path, keys):
|
|
with open(path, encoding="utf-8") as f:
|
|
v = json.load(f)
|
|
print(json.dumps(run_vector(v, [bytes.fromhex(k) for k in keys] if keys else None)))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) >= 3 and sys.argv[1] == "--room" and len(sys.argv) % 2 == 1 and all(a == "--opening-key" for a in sys.argv[3::2]):
|
|
sys.exit(room_main(sys.argv[2], sys.argv[4::2]))
|
|
if len(sys.argv) != 2:
|
|
print("usage: remote_desktop_verify.py <live-conformance/remote> | --room <room.json> [--opening-key <hex>]...", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|