open-standards/wai/conform-result-conformance/conform_result_verify.py

#!/usr/bin/env python3
"""Second verifier for conform-result-conformance/: conformance results, the record
conform-result/1 (extensions/conformance-results.md §2-§5).

Written from the extension's text. It shares no code with wai-rs (`wai::conform_result`),
which generates the corpus. It takes only WAI's strict Ed25519 rule (SPEC §7 *Signatures*)
from ../claim-conformance/_ed25519.py, itself written from RFC 8032 and independent of
wai-rs; the JSON reading and RFC 8785 serialisation are written here. Standard library only.

    python3 -I conform_result_verify.py [corpus-dir]         every vector reaches its stated outcome
    python3 -I conform_result_verify.py verdicts <file>      `name<TAB>outcome` for every vector
    python3 -I conform_result_verify.py --skip <check> [dir] the same with one check of §4 left
                                                             out; succeeds only if the corpus then
                                                             fails, showing it guards that check

The checks --skip names: duplicates (step 1's member named twice), format (step 2), members
(step 3), signature (step 4), summary (step 5), verdict (step 6).

Exit status 0 only if every vector reaches its stated outcome (or, with --skip, only if one
does not).
"""
import hashlib
import json
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
import _ed25519 as ED  # noqa: E402

FORMAT = "conform-result/1"
DOMAIN = b"wai:conform-result" + b"\x01"
MAX_INT = 2 ** 53 - 1
MAX_DEPTH = 64
CHECKS = ("duplicates", "format", "members", "signature", "summary", "verdict")
SKIP = set()


class Refuse(Exception):
    def __init__(self, step, reason):
        super().__init__(f"step {step} {reason}")
        self.step, self.reason = step, reason


class Num:
    """A JSON number as written: the reader keeps its spelling."""

    def __init__(self, text):
        self.text = text

    def __eq__(self, other):
        return isinstance(other, Num) and other.text == self.text

    def uint(self):
        """The value of a plain decimal integer (no sign, fraction or exponent) up to 2^53 - 1."""
        t = self.text
        if re.fullmatch(r"0|[1-9][0-9]*", t) and int(t) <= MAX_INT:
            return int(t)
        return None


def _pairs(ps):
    d = {}
    for k, v in ps:
        if k in d and "duplicates" not in SKIP:
            raise ValueError("a member named twice: " + k)
        d[k] = v
    return d


def _no_constant(c):
    raise ValueError("not JSON: " + c)


def _depth_and_strings(v, depth):
    """§4 step 1: no value nested more than 64 levels deep, and no string that is not
    Unicode text (an escape that leaves a surrogate unpaired)."""
    if depth >= MAX_DEPTH:
        raise ValueError("nested too deep")
    if isinstance(v, str):
        v.encode("utf-8")
    elif isinstance(v, dict):
        for k, x in v.items():
            k.encode("utf-8")
            _depth_and_strings(x, depth + 1)
    elif isinstance(v, list):
        for x in v:
            _depth_and_strings(x, depth + 1)


def read(raw):
    """§4 step 1: one JSON text in UTF-8 whose value is an object, no member named twice."""
    try:
        text = raw.decode("utf-8")
        v = json.loads(text, object_pairs_hook=_pairs, parse_constant=_no_constant, parse_int=Num, parse_float=Num, strict=True)
        _depth_and_strings(v, 0)
    except (ValueError, UnicodeError, RecursionError):
        raise Refuse(1, "malformed")
    if not isinstance(v, dict):
        raise Refuse(1, "malformed")
    return v


def is_hex(s, n):
    return isinstance(s, str) and len(s) == n and re.fullmatch(r"[0-9a-f]*", s) is not None


def is_str(v):
    return isinstance(v, str)


def exactly(v, names):
    return isinstance(v, dict) and set(v) == set(names) and len(v) == len(names)


def is_uint(v):
    return isinstance(v, Num) and v.uint() is not None


def utc_second(s):
    m = isinstance(s, str) and re.fullmatch(r"([0-9]{4})-([0-9]{2})-([0-9]{2})T([0-9]{2}):([0-9]{2}):([0-9]{2})Z", s)
    if not m:
        return False
    mo, d, h, mi, se = (int(x) for x in m.groups()[1:])
    return 1 <= mo <= 12 and 1 <= d <= 31 and h <= 23 and mi <= 59 and se <= 60


TOP = ["wai", "kind", "suite", "standard", "toolchain", "corpus", "implementation", "started_at",
       "cases", "summary", "conformant", "key", "signer_pubkey", "sig"]


def members(r):
    """§4 step 3: exactly the members of §2, each of its type and form."""
    ok = exactly(r, TOP)
    ok = ok and all(is_str(r[k]) for k in ("kind", "suite", "standard"))
    t = r.get("toolchain")
    ok = ok and exactly(t, ["name", "version"]) and is_str(t["name"]) and is_str(t["version"])
    c = r.get("corpus")
    ok = ok and exactly(c, ["dir", "files", "digest"]) and is_str(c["dir"]) and is_uint(c["files"])
    ok = ok and is_str(c["digest"]) and c["digest"].startswith("sha256:") and is_hex(c["digest"][7:], 64)
    i = r.get("implementation")
    ok = ok and exactly(i, ["adapter", "impl", "name", "version", "files"]) and i["adapter"] in ("command", "http")
    ok = ok and is_str(i["impl"]) and all(x is None or is_str(x) for x in (i["name"], i["version"]))
    ok = ok and isinstance(i["files"], list) and all(
        exactly(f, ["path", "sha256"]) and is_str(f["path"]) and is_hex(f["sha256"], 64) for f in i["files"])
    ok = ok and utc_second(r["started_at"])
    ok = ok and isinstance(r["cases"], list) and all(
        exactly(x, ["name", "pass", "expected", "got"]) and is_str(x["name"]) and isinstance(x["pass"], bool)
        and is_str(x["expected"]) and is_str(x["got"]) for x in r["cases"])
    s = r.get("summary")
    ok = ok and exactly(s, ["cases", "passed", "failed"]) and all(is_uint(s[k]) for k in ("cases", "passed", "failed"))
    ok = ok and isinstance(r["conformant"], bool) and r["key"] in ("ephemeral", "supplied")
    ok = ok and is_hex(r["signer_pubkey"], 64) and is_hex(r["sig"], 128)
    return bool(ok)


def jcs(v):
    """RFC 8785 over the values a record holds: members sorted by their UTF-16 code units,
    strings escaped as ECMAScript's JSON.stringify escapes them, integers in decimal."""
    if v is None:
        return "null"
    if v is True:
        return "true"
    if v is False:
        return "false"
    if isinstance(v, Num):
        return str(v.uint())
    if isinstance(v, str):
        return json.dumps(v, ensure_ascii=False)
    if isinstance(v, list):
        return "[" + ",".join(jcs(x) for x in v) + "]"
    keys = sorted(v, key=lambda k: k.encode("utf-16-be"))
    return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + jcs(v[k]) for k in keys) + "}"


def check(raw):
    """§4, in order. Returns the verdict line."""
    try:
        r = read(raw)
        if "format" not in SKIP and r.get("wai") != FORMAT:
            raise Refuse(2, "unknown_format")
        if "members" not in SKIP and not members(r):
            raise Refuse(3, "malformed")
        if "signature" not in SKIP:
            body = {k: v for k, v in r.items() if k != "sig"}
            msg = DOMAIN + jcs(body).encode("utf-8")
            if ED.verify(bytes.fromhex(r["signer_pubkey"]), msg, bytes.fromhex(r["sig"])) is not None:
                raise Refuse(4, "signature_invalid")
        cases = r["cases"]
        passed = sum(1 for c in cases if c["pass"] is True)
        s = r["summary"]
        if "summary" not in SKIP and (s["cases"].uint(), s["passed"].uint(), s["failed"].uint()) != (len(cases), passed, len(cases) - passed):
            raise Refuse(5, "summary_mismatch")
        conformant = len(cases) > 0 and passed == len(cases)
        if "verdict" not in SKIP and r["conformant"] is not conformant:
            raise Refuse(6, "conformant_mismatch")
        return "accept conformant=" + ("true" if r["conformant"] else "false")
    except Refuse as e:
        return f"reject step={e.step} reason={e.reason}"
    except (KeyError, TypeError, AttributeError, ValueError):
        # Reached only when a check is skipped and a later one reads what it would refuse.
        return "error"


def reasons_table():
    """§5's reasons, read from the extension."""
    with open(os.path.join(HERE, "..", "extensions", "conformance-results.md"), encoding="utf-8") as f:
        text = f.read()
    section = text.split("\n## 5. Reasons", 1)[1].split("\n## ", 1)[0]
    return {m for m in re.findall(r"^\| `([a-z_]+)` \|", section, re.M)}


def vectors(path):
    with open(path, encoding="utf-8") as f:
        return json.load(f)["vectors"]


def run(path):
    fail = 0
    b = ED._self_test() if hasattr(ED, "_self_test") else None
    if b:
        print("FAIL Ed25519 self-test: " + str(b))
        fail += 1
    used = set()
    vs = vectors(path)
    for v in vs:
        got = check(v["record"].encode("utf-8"))
        ok = got == v["expected"]
        fail += not ok
        m = re.search(r"reason=([a-z_]+)$", v["expected"])
        if m:
            used.add(m.group(1))
        print(f"{'PASS' if ok else 'FAIL'} {v['name']}  expected={v['expected']}  got={got}")
    table = reasons_table()
    same = table == used
    print(("PASS" if same else "FAIL") + f" the corpus refuses with every reason of §5 and no other ({sorted(table ^ used)})")
    fail += not same
    print(f"{len(vs)} vectors, {fail} failure(s)")
    return fail


def main(argv):
    global SKIP
    if argv[1:2] == ["verdicts"] and len(argv) == 3:
        for v in vectors(argv[2]):
            print(f"{v['name']}\t{check(v['record'].encode('utf-8'))}")
        return 0
    args = argv[1:]
    if args[:1] == ["--skip"]:
        if len(args) < 2 or args[1] not in CHECKS:
            print("--skip takes one of " + ", ".join(CHECKS))
            return 2
        SKIP = {args[1]}
        d = args[2] if len(args) > 2 else HERE
        fail = run(os.path.join(d, "vectors.json"))
        print(f"with the {args[1]} check left out: {'caught' if fail else 'NOT caught'}")
        return 0 if fail else 1
    d = args[0] if args else HERE
    return 1 if run(os.path.join(d, "vectors.json")) else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))