From f7194284d243b72224b901c4409e10c1750bee6e Mon Sep 17 00:00:00 2001 From: volt Date: Mon, 26 Oct 2026 17:58:37 +0000 Subject: [PATCH] volt: beam: Q79 -- manifest.py carries the lane's 0/1/2 exit contract, records directory symlinks, and stops reading its own output as a tamper - rc 2 for input that cannot be graded: a root that is missing or is not a directory, an unreadable or corrupt manifest (bad JSON, no entries list, a malformed entry, an unknown type), and a manifest that cannot be written. rc 1 is reserved for a graded fault, rc 0 for a clean check. - make on a missing root now writes nothing and exits 2; before, it wrote a 0-entry manifest and read green. - _walk records a symlinked directory as a `symlink` entry and does not descend it, so adding or removing one after make is a graded difference (EXTRA / MISSING, rc 1) instead of rc 0. - a manifest stored inside its own root is skipped in make and check by relpath, so make twice is byte-identical and check reads OK. - a root it cannot read (OSError from os.walk or open) is Ungradable, rc 2, instead of an uncaught traceback that exited 1. CHECK: python3 wizard-work/beam/manifest.py make /root/troupe-standup/wizard-work/beam /tmp/q79-manifest.json && python3 wizard-work/beam/manifest.py check /root/troupe-standup/wizard-work/beam /tmp/q79-manifest.json --- wizard-work/beam/manifest.py | 189 ++++++++++++++++++++++++++--------- 1 file changed, 142 insertions(+), 47 deletions(-) diff --git a/wizard-work/beam/manifest.py b/wizard-work/beam/manifest.py index 297dcef..8717be5 100644 --- a/wizard-work/beam/manifest.py +++ b/wizard-work/beam/manifest.py @@ -4,81 +4,172 @@ Usage: python3 manifest.py make python3 manifest.py check + +Exit codes (the lane contract): + 0 clean -- make wrote the manifest; check found the tree matches it + 1 fault -- check ran and graded a mismatch (missing, extra, hash, size, + type change, symlink-target change) + 2 ungradable -- bad usage, a root that is missing or is not a directory, an + unreadable or corrupt manifest, or a manifest that could not + be written; no verdict exists, so no 0 and no 1 + +Canonical: this file owns the directory-manifest surface for the lane, including +the file hash and size it shares with `sha256_manifest.py`. `sha256_manifest.py` +is retained only for the sha256sum-compatible `digest path` line format; use it +when a consumer needs that text shape, and this file for anything else. """ -import hashlib, json, os, sys +import hashlib +import json +import os +import sys -def _walk(root): +class Ungradable(Exception): + """Input that cannot be graded: no verdict exists, so the exit code is 2.""" + + +def _require_root(root): + root = os.path.abspath(root) + if not os.path.isdir(root): + raise Ungradable("root does not exist or is not a directory: %s" % root) + return root + + +def _output_relpath(root, out): + """Relpath of a manifest that lives inside root, else None.""" + if not out: + return None + rel = os.path.relpath(os.path.abspath(out), root) + if rel == os.pardir or rel.startswith(os.pardir + os.sep): + return None + return rel + + +def _walk(root, skip=()): + """Sorted entries for every file and symlink under root, never following one. + + A symlinked directory is recorded as a `symlink` entry and is not descended + into, so adding or removing one after `make` is a graded difference. Paths in + `skip` (the manifest's own output, when it is stored inside root) are ignored. + """ entries = [] - for dirpath, dirnames, filenames in os.walk(root, followlinks=False): + skip = set(skip) + + def _raise(exc): + raise Ungradable("cannot read %s: %s" % (exc.filename, exc.strerror or exc)) + + for dirpath, dirnames, filenames in os.walk(root, onerror=_raise, followlinks=False): dirnames.sort() - for name in sorted(filenames): + keep = [] + for name in dirnames: full = os.path.join(dirpath, name) if os.path.islink(full): - # Record symlink target as a special entry; never follow. entries.append((os.path.relpath(full, root), "symlink", os.readlink(full))) + else: + keep.append(name) + dirnames[:] = keep + for name in sorted(filenames): + full = os.path.join(dirpath, name) + rel = os.path.relpath(full, root) + if rel in skip: + continue + if os.path.islink(full): + # Record the symlink target as a special entry; never follow. + entries.append((rel, "symlink", os.readlink(full))) continue if not os.path.isfile(full): continue - rel = os.path.relpath(full, root) h = hashlib.sha256() - with open(full, "rb") as f: - for chunk in iter(lambda: f.read(65536), b""): - h.update(chunk) + try: + with open(full, "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + h.update(chunk) + except OSError as exc: + raise Ungradable("cannot read %s: %s" % (rel, exc.strerror or exc)) st = os.stat(full, follow_symlinks=False) entries.append((rel, "file", h.hexdigest(), st.st_size)) entries.sort(key=lambda e: e[0]) return entries -def make(root, out): - root = os.path.abspath(root) - entries = _walk(root) - manifest = {"root": os.path.basename(root), "entries": []} +def _entries_to_dicts(entries): + out = [] for e in entries: if e[1] == "symlink": - manifest["entries"].append({"path": e[0], "type": "symlink", "target": e[2]}) + out.append({"path": e[0], "type": "symlink", "target": e[2]}) else: - manifest["entries"].append({"path": e[0], "type": "file", "sha256": e[2], "size": e[3]}) - with open(out, "w") as f: - json.dump(manifest, f, indent=2, sort_keys=True) - f.write("\n") + out.append({"path": e[0], "type": "file", "sha256": e[2], "size": e[3]}) + return out + + +def _load_manifest(path): + try: + with open(path, "rb") as f: + manifest = json.load(f) + except OSError as exc: + raise Ungradable("cannot read manifest %s: %s" % (path, exc.strerror or exc)) + except ValueError as exc: # json.JSONDecodeError and non-UTF-8 both land here + raise Ungradable("manifest is not valid JSON: %s: %s" % (path, exc)) + if not isinstance(manifest, dict): + raise Ungradable("manifest is not a JSON object: %s" % path) + rows = manifest.get("entries") + if not isinstance(rows, list): + raise Ungradable("manifest has no entries list: %s" % path) + for i, row in enumerate(rows): + if not isinstance(row, dict) or "path" not in row or "type" not in row: + raise Ungradable("manifest entry %d is malformed: %s" % (i, path)) + if row["type"] not in ("file", "symlink"): + raise Ungradable("manifest entry %d has unknown type %r: %s" + % (i, row["type"], path)) + if row["type"] == "file" and ("sha256" not in row or "size" not in row): + raise Ungradable("manifest entry %d lacks sha256/size: %s" % (i, path)) + if row["type"] == "symlink" and "target" not in row: + raise Ungradable("manifest entry %d lacks target: %s" % (i, path)) + return manifest + + +def make(root, out): + root = _require_root(root) + skip = _output_relpath(root, out) + entries = _walk(root, skip=[skip] if skip else ()) + manifest = {"root": os.path.basename(root), "entries": _entries_to_dicts(entries)} + try: + with open(out, "w") as f: + json.dump(manifest, f, indent=2, sort_keys=True) + f.write("\n") + except OSError as exc: + raise Ungradable("cannot write manifest %s: %s" % (out, exc.strerror or exc)) return len(entries) def check(root, manifest_path): - root = os.path.abspath(root) - with open(manifest_path) as f: - manifest = json.load(f) - expected = {e["path"]: e for e in manifest["entries"]} - actual_entries = _walk(root) - actual = {} - for e in actual_entries: - if e[1] == "symlink": - actual[e[0]] = {"path": e[0], "type": "symlink", "target": e[2]} - else: - actual[e[0]] = {"path": e[0], "type": "file", "sha256": e[2], "size": e[3]} + root = _require_root(root) + manifest = _load_manifest(manifest_path) + skip = _output_relpath(root, manifest_path) + expected = {row["path"]: row for row in manifest["entries"]} + actual = {row["path"]: row for row in + _entries_to_dicts(_walk(root, skip=[skip] if skip else ()))} errors = [] for p, e in expected.items(): if p not in actual: - errors.append(f"MISSING: {p}") + errors.append("MISSING: %s" % p) continue a = actual[p] if a["type"] != e["type"]: - errors.append(f"TYPE_CHANGED: {p}") + errors.append("TYPE_CHANGED: %s" % p) continue if e["type"] == "file": if a["sha256"] != e["sha256"]: - errors.append(f"HASH_MISMATCH: {p}") + errors.append("HASH_MISMATCH: %s" % p) if a["size"] != e["size"]: - errors.append(f"SIZE_MISMATCH: {p}") + errors.append("SIZE_MISMATCH: %s" % p) else: if a["target"] != e["target"]: - errors.append(f"SYMLINK_TARGET_CHANGED: {p}") + errors.append("SYMLINK_TARGET_CHANGED: %s" % p) for p in actual: if p not in expected: - errors.append(f"EXTRA: {p}") + errors.append("EXTRA: %s" % p) return errors @@ -87,18 +178,22 @@ def main(argv): print("usage: manifest.py make|check ", file=sys.stderr) return 2 cmd, root, mf = argv[1], argv[2], argv[3] - if cmd == "make": - n = make(root, mf) - print(f"manifest written: {n} entries -> {mf}") - return 0 - if cmd == "check": - errs = check(root, mf) - if errs: - for e in errs: - print(e) - return 1 - print("OK: manifest matches") - return 0 + try: + if cmd == "make": + n = make(root, mf) + print("manifest written: %d entries -> %s" % (n, mf)) + return 0 + if cmd == "check": + errs = check(root, mf) + if errs: + for e in errs: + print(e) + return 1 + print("OK: manifest matches") + return 0 + except Ungradable as exc: + print("ungradable: %s" % exc, file=sys.stderr) + return 2 print("unknown command", file=sys.stderr) return 2