troupe-standup/wizard-work/spark/json_sha256_validator.py

#!/usr/bin/env python3
"""Validate JSON documents and print the SHA-256 of the exact bytes parsed.

One read serves both halves of the verdict: the bytes that are hashed are the
bytes that were parsed, so a file rewritten between the parse and the digest
cannot be described as "valid" while a different document's hash is printed
(Q80 measured the two-open form: two reads, and the printed digest belonged to
the second one).

Contract -- the repo's tools/ exit codes:

  0  every input is a strict RFC 8259 JSON text; one ``VALID <path> SHA256=``
     line per input.
  1  every input was readable and at least one is not a JSON text: a syntax
     error, a bare ``NaN`` / ``Infinity`` / ``-Infinity`` (RFC 8259 s6: "Numeric
     values that cannot be represented in the grammar below (such as Infinity
     and NaN) are not permitted" -- rejected here through ``parse_constant``),
     a duplicate object member name (s4 says names SHOULD be unique; refused
     unless ``--allow-duplicates``), or bytes that are not decodable JSON text.
  2  could not grade: a usage error, or an input that could not be read
     (missing, a directory, permission denied, a non-directory in the path).
     Nothing is printed for *any* input when any input is ungradable -- a
     substrate failure never wears a verdict, and no digest is emitted.

A leading UTF-8 byte order mark is tolerated: the parsed text is the bytes
decoded as ``utf-8-sig`` (RFC 8259 s8.1 lets a parser ignore a BOM, and the
lane's ``jsonl_validate.py`` already reads ``utf-8-sig``). Any other encoding --
UTF-16, UTF-32, Latin-1, binary -- is not a JSON *text* in the interchange
encoding RFC 8259 s8.1 requires, and grades rc 1 as undecodable. Both are stated
so neither is filed as a bug. ``--results PATH`` writes the run as JSON: the
exact invocation, and each input's path, size, blob sha256 and verdict.
"""
import argparse
import hashlib
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

RC_OK = 0
RC_INVALID = 1
RC_UNGRADABLE = 2

VALID = "valid"
INVALID = "invalid"
UNGRADABLE = "ungradable"


class _NonStandardConstant(ValueError):
    """NaN / Infinity / -Infinity: not JSON (RFC 8259 s6)."""


class _DuplicateKeys(ValueError):
    """An object repeats a member name (RFC 8259 s4: names SHOULD be unique)."""


def _reject_constant(name):
    raise _NonStandardConstant(
        f"non-standard constant {name!r}: RFC 8259 s6 forbids NaN and Infinity"
    )


def _pairs_hook(pairs):
    seen = set()
    for name, _value in pairs:
        if name in seen:
            raise _DuplicateKeys(
                f"duplicate object member name {name!r}: RFC 8259 s4 says names "
                "SHOULD be unique (pass --allow-duplicates to accept them)"
            )
        seen.add(name)
    return dict(pairs)


def read_bytes(path):
    """Return (bytes, None) or (None, message). One read, no second open."""
    try:
        return Path(path).read_bytes(), None
    except OSError as exc:
        detail = exc.strerror or str(exc) or type(exc).__name__
        return None, f"{type(exc).__name__}: {detail}"


def parse_bytes(data, allow_duplicates=False):
    """Return (parsed, None) or (None, message). Parses the bytes it is given.

    One decode (``utf-8-sig``: a BOM is tolerated, no other encoding is), then
    one parse of that text, so the digest taken beside this call covers exactly
    the bytes this function read.
    """
    try:
        text = data.decode("utf-8-sig")
    except UnicodeDecodeError as exc:
        return None, f"not decodable JSON text: {exc.reason} at byte {exc.start}"
    try:
        if allow_duplicates:
            return json.loads(text, parse_constant=_reject_constant), None
        return json.loads(text, parse_constant=_reject_constant,
                          object_pairs_hook=_pairs_hook), None
    except ValueError as exc:
        return None, str(exc)


def grade(path, allow_duplicates=False):
    """Grade one input. Returns a record; the digest covers the parsed bytes."""
    record = {"path": str(path), "bytes": None, "sha256": None,
              "verdict": None, "error": None}
    data, read_error = read_bytes(path)
    if data is None:
        record["verdict"] = UNGRADABLE
        record["error"] = read_error
        return record
    record["bytes"] = len(data)
    record["sha256"] = hashlib.sha256(data).hexdigest()
    _, parse_error = parse_bytes(data, allow_duplicates=allow_duplicates)
    if parse_error is None:
        record["verdict"] = VALID
    else:
        record["verdict"] = INVALID
        record["error"] = parse_error
    return record


def _write_results(path, argv, records):
    payload = {
        "tool": Path(__file__).name,
        "invocation": [sys.executable, str(Path(__file__).resolve())] + list(argv),
        "argv": list(argv),
        "cwd": os.getcwd(),
        "generated_at_utc": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "inputs": records,
    }
    try:
        Path(path).write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n",
                              encoding="utf-8")
    except OSError as exc:
        return f"{type(exc).__name__}: {exc.strerror or exc}"
    return None


def main(argv=None):
    ap = argparse.ArgumentParser(
        prog="json_sha256_validator.py",
        description="Validate JSON and print the SHA-256 of the parsed bytes.",
    )
    ap.add_argument("inputs", nargs="+", metavar="FILE",
                    help="JSON document(s) to validate")
    ap.add_argument("--allow-duplicates", action="store_true",
                    help="accept repeated object member names (last wins)")
    ap.add_argument("--results", metavar="PATH",
                    help="write the run -- invocation, per-input blob sha, verdict -- as JSON")
    try:
        args = ap.parse_args(argv)
    except SystemExit as exc:
        return RC_UNGRADABLE if exc.code not in (0, None) else RC_OK

    records = [grade(p, allow_duplicates=args.allow_duplicates) for p in args.inputs]

    ungradable = [r for r in records if r["verdict"] == UNGRADABLE]
    if ungradable:
        for r in ungradable:
            print(f"UNGRADABLE {r['path']} ERROR={r['error']}")
        print(f"cannot grade: {len(ungradable)} of {len(records)} input(s) unreadable; "
              "no verdict and no digest printed")
        return RC_UNGRADABLE

    if args.results:
        error = _write_results(args.results, argv if argv is not None else sys.argv[1:],
                               records)
        if error is not None:
            print(f"UNGRADABLE --results ERROR={error}")
            return RC_UNGRADABLE

    for r in records:
        if r["verdict"] == VALID:
            print(f"VALID  {r['path']} SHA256={r['sha256']}")
        else:
            print(f"INVALID {r['path']} ERROR={r['error']}")

    return RC_OK if all(r["verdict"] == VALID for r in records) else RC_INVALID


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