volt
44b595e593
manifest_audit.py shipped into tools/ at 25e1cd6 with an unguarded --json
write, so write_probe.py graded it NO_PROBE and verify.py's write-axis leg
read FAULT (rc 1) on every run.
- tools/manifest_audit.py: wrap the --json write; an unwritable target now
prints one line naming the path and returns 2 (house contract: rc 2 outranks
the 0/1 the audit earned). Success path unchanged -- measured over docs/:
rc 0, payload keys cleared,new,no_manifest,root,rows, rows 0.
- tools/write_probe.py: WRITE_PROBES row (--root {tinyrepo} --json {target}).
Bracket, measured: same row run against HEAD's unguarded blob in a detach
worktree -> FAULT manifest_audit.py TRACEBACK rc 1 (FileNotFoundError);
against this tree -> json OK rc 2, naming the target. write_probe --selftest
15/15, manifest_audit --selftest 17/17, verify.py --repo . 8/8 legs pass,
selftest-sweep pin 22 pinned / 22 agree.
427 lines
18 KiB
Python
Executable File
427 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""manifest_audit.py -- does an artifact's `files.sha256` still describe the bytes beside it?
|
|
|
|
WHY THIS EXISTS
|
|
Directories under this lane's artifact root present a `files.sha256`, and no
|
|
instrument in `tools/` re-reads one against the tree it sits in.
|
|
`artifact_claim_check.py` grades WHICH base a manifest pins (SELF vs FOREIGN)
|
|
and was never shipped into `tools/`; `sha256sum -c` verifies a single manifest
|
|
when a reader remembers to run it. Neither answers the question a reader
|
|
actually has in front of an artifact: *is this manifest still true, today, for
|
|
this directory?*
|
|
|
|
This tool answers only that. It walks the artifact root named by --root, and
|
|
for every directory presenting a `files.sha256` it re-hashes each entry:
|
|
|
|
VERIFIED every entry resolves inside the directory and matches its hash
|
|
STALE an entry resolves inside the directory and no longer matches
|
|
MISSING_ENTRY an entry names a file the directory no longer holds
|
|
FOREIGN_BASE an entry resolves outside the directory (absolute, or climbing
|
|
`../`): such a manifest pins another tree, so it cannot verify
|
|
the artifact it sits in, whatever its hashes say
|
|
EMPTY a manifest with no parsable entry -- a claim that claims nothing
|
|
NO_MANIFEST a directory that never claimed a manifest: CENSUS, not a fault
|
|
|
|
A directory with no manifest is counted, never faulted: an artifact that never
|
|
claimed checksums has not broken a claim. That is the same rule this archive's
|
|
artifact-claims record already applies, and it is why the census figure (how
|
|
much of the lane is manifest-covered) is printed beside the verdict.
|
|
|
|
EXIT CODES (the house contract)
|
|
0 ran; every presented manifest graded; no NEW fault (or none at all)
|
|
1 --fail/--strict ran and at least one fault is not named by the ledger
|
|
2 could not grade: --root absent or not a directory, or the ledger is
|
|
unreadable. An audit that graded nothing did not pass.
|
|
|
|
THE LEDGER
|
|
`--baseline FILE` is a TSV of `dirname<TAB>class<TAB>note`, read on every run.
|
|
A fault whose (dir, class) is named prints `KNOWN` and does not raise rc 1; a
|
|
row that no longer reproduces prints `CLEARED`; `--strict` ignores the file
|
|
entirely. The ledger suppresses the exit code, never the finding -- every
|
|
known row is still printed with its class on every run.
|
|
|
|
WHY THIS IS IN `tools/` AND IN NO verify.py LEG
|
|
It grades an artifact tree (--root), not the repo it is invoked from, so a
|
|
fixed gate leg would grade whichever machine happened to be running it and
|
|
report a machine-local path as a repo verdict -- the "advice that names a
|
|
target which does not exist" defect `verify.py` opens with, one level down.
|
|
It is discovered by `selftest_sweep.py` from `tools/*.py` and graded by both
|
|
of that sweep's input axes and by the count pin (17 cells). The write end is
|
|
graded too, by `write_probe.py`'s json axis -- that probe runs this tool
|
|
against a throwaway fixture root, so the fixture, not the host, is what the
|
|
verdict describes. What no leg does is grade `--root` against this repository:
|
|
that stays a caller's explicit request, because the artifact tree a reader
|
|
cares about is theirs, not the checkout's.
|
|
|
|
The two axes it is graded on, and why the third does not apply to it:
|
|
read-refusal --root that does not exist, and --root naming a file, are
|
|
both rc 2 with a message (an audit that graded nothing did
|
|
not pass)
|
|
malformed a `files.sha256` whose lines parse as nothing is graded
|
|
EMPTY and is rc 1 under --fail: the tool grades a broken
|
|
claim rather than passing it silently
|
|
non-UTF-8 NOT a refusal here, and that is deliberate. The manifest is
|
|
read with `errors="replace"` -- it is ASCII hex by
|
|
construction -- so binary bytes degrade to junk lines and
|
|
are graded EMPTY (measured: rc 1 under --fail, no
|
|
traceback). A refusal would hide a manifest that exists and
|
|
claims nothing, which is the fault this tool is for.
|
|
|
|
USAGE
|
|
python3 manifest_audit.py --selftest # 17 cells, fixtures, ledger, both refusals
|
|
python3 manifest_audit.py --root docs # census only: no manifest, no fault
|
|
|
|
Against a real artifact root, the same `--root` with `--baseline FILE --fail`
|
|
exits 1 exactly when a fault is graded that the ledger does not name; without
|
|
`--fail` it reports and exits 0, and `--strict` ignores the ledger entirely.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
|
|
MANIFEST = "files.sha256"
|
|
ENTRY_RE = re.compile(r"^([0-9a-fA-F]{64})[ \t]+\*?(.+?)\s*$")
|
|
|
|
VERIFIED = "VERIFIED"
|
|
STALE = "STALE"
|
|
MISSING_ENTRY = "MISSING_ENTRY"
|
|
FOREIGN_BASE = "FOREIGN_BASE"
|
|
EMPTY = "EMPTY"
|
|
NO_MANIFEST = "NO_MANIFEST"
|
|
FAULT_CLASSES = (STALE, MISSING_ENTRY, FOREIGN_BASE, EMPTY)
|
|
|
|
|
|
def sha256_file(path: str) -> str:
|
|
h = hashlib.sha256()
|
|
with open(path, "rb") as fh:
|
|
for chunk in iter(lambda: fh.read(1 << 20), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def parse_manifest(path: str):
|
|
"""Return (entries, unparsable_line_count). Entry = (digest, raw_path)."""
|
|
entries, junk = [], 0
|
|
with open(path, "r", encoding="utf-8", errors="replace") as fh:
|
|
for line in fh:
|
|
line = line.rstrip("\n")
|
|
if not line.strip() or line.lstrip().startswith("#"):
|
|
continue
|
|
m = ENTRY_RE.match(line)
|
|
if not m:
|
|
junk += 1
|
|
continue
|
|
entries.append((m.group(1).lower(), m.group(2)))
|
|
return entries, junk
|
|
|
|
|
|
def inside(base: str, target: str) -> bool:
|
|
base = os.path.realpath(base)
|
|
target = os.path.realpath(target)
|
|
return target == base or target.startswith(base + os.sep)
|
|
|
|
|
|
def grade_dir(adir: str):
|
|
"""Grade one artifact directory. Returns dict, or None when no manifest."""
|
|
man = os.path.join(adir, MANIFEST)
|
|
if not os.path.isfile(man):
|
|
return None
|
|
row = {
|
|
"dir": os.path.basename(adir.rstrip(os.sep)),
|
|
"path": adir,
|
|
"entries": 0,
|
|
"self": 0,
|
|
"ok": 0,
|
|
"stale": [],
|
|
"missing": [],
|
|
"foreign": [],
|
|
"junk": 0,
|
|
"verdict": VERIFIED,
|
|
}
|
|
entries, junk = parse_manifest(man)
|
|
row["junk"] = junk
|
|
row["entries"] = len(entries)
|
|
for digest, raw in entries:
|
|
target = raw if os.path.isabs(raw) else os.path.join(adir, raw)
|
|
if not inside(adir, target):
|
|
row["foreign"].append(raw)
|
|
continue
|
|
row["self"] += 1
|
|
if not os.path.isfile(target):
|
|
row["missing"].append(raw)
|
|
continue
|
|
try:
|
|
got = sha256_file(target)
|
|
except OSError as exc: # unreadable file is a missing claim, not a pass
|
|
row["missing"].append("%s (%s)" % (raw, exc.strerror))
|
|
continue
|
|
if got == digest:
|
|
row["ok"] += 1
|
|
else:
|
|
row["stale"].append(raw)
|
|
if row["foreign"]:
|
|
row["verdict"] = FOREIGN_BASE
|
|
elif row["missing"]:
|
|
row["verdict"] = MISSING_ENTRY
|
|
elif row["stale"]:
|
|
row["verdict"] = STALE
|
|
elif not entries:
|
|
row["verdict"] = EMPTY
|
|
return row
|
|
|
|
|
|
def walk_root(root: str):
|
|
rows, bare = [], []
|
|
for name in sorted(os.listdir(root)):
|
|
adir = os.path.join(root, name)
|
|
if not os.path.isdir(adir):
|
|
continue
|
|
row = grade_dir(adir)
|
|
if row is None:
|
|
bare.append(name)
|
|
else:
|
|
rows.append(row)
|
|
return rows, bare
|
|
|
|
|
|
def load_baseline(path: str):
|
|
"""Return {(dir, class): note}. Raises OSError/ValueError on an unreadable ledger."""
|
|
known = {}
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
for n, line in enumerate(fh, 1):
|
|
line = line.rstrip("\n")
|
|
if not line.strip() or line.lstrip().startswith("#"):
|
|
continue
|
|
parts = line.split("\t")
|
|
if len(parts) < 2:
|
|
raise ValueError("baseline line %d: expected dir<TAB>class<TAB>note" % n)
|
|
known[(parts[0], parts[1])] = parts[2] if len(parts) > 2 else ""
|
|
return known
|
|
|
|
|
|
def apply_baseline(rows, known):
|
|
for row in rows:
|
|
key = (row["dir"], row["verdict"])
|
|
if row["verdict"] in FAULT_CLASSES and key in known:
|
|
row["state"] = "KNOWN"
|
|
elif row["verdict"] in FAULT_CLASSES:
|
|
row["state"] = "NEW"
|
|
else:
|
|
row["state"] = "-"
|
|
seen = {(r["dir"], r["verdict"]) for r in rows}
|
|
cleared = [k for k in known if k not in seen]
|
|
return [r for r in rows if r["state"] == "NEW"], cleared
|
|
|
|
|
|
def report(rows, bare, new, cleared, known_size, out=sys.stdout):
|
|
faults = [r for r in rows if r["verdict"] in FAULT_CLASSES]
|
|
counts = {}
|
|
for r in rows:
|
|
counts[r["verdict"]] = counts.get(r["verdict"], 0) + 1
|
|
total_entries = sum(r["entries"] for r in rows)
|
|
total_ok = sum(r["ok"] for r in rows)
|
|
w = out.write
|
|
if rows:
|
|
w("artifact root: %s\n" % os.path.dirname(rows[0]["path"]))
|
|
w("manifests: %d presented, %d without one (census, not a fault)\n"
|
|
% (len(rows), len(bare)))
|
|
order = (VERIFIED, FOREIGN_BASE, STALE, MISSING_ENTRY, EMPTY)
|
|
w("classes: " + ", ".join("%s %d" % (c, counts.get(c, 0)) for c in order) + "\n")
|
|
w("entries: %d parsed, %d hashed and matching, %d junk line(s)\n"
|
|
% (total_entries, total_ok, sum(r["junk"] for r in rows)))
|
|
for r in faults:
|
|
detail = []
|
|
if r["foreign"]:
|
|
detail.append("foreign %d" % len(r["foreign"]))
|
|
if r["stale"]:
|
|
detail.append("stale %d" % len(r["stale"]))
|
|
if r["missing"]:
|
|
detail.append("missing %d" % len(r["missing"]))
|
|
w(" %-40s %-14s %-5s %s\n" % (r["dir"], r["verdict"], r["state"],
|
|
", ".join(detail)))
|
|
for label, items in (("foreign", r["foreign"]), ("stale", r["stale"]),
|
|
("missing", r["missing"])):
|
|
for item in items[:4]:
|
|
w(" %s: %s\n" % (label, item))
|
|
for key in cleared:
|
|
w(" %-40s %-14s CLEARED (ledger row no longer reproduces: %s)\n"
|
|
% (key[0], key[1], key[1]))
|
|
w("ledger: %d row(s), %d known, %d new, %d cleared\n"
|
|
% (known_size, len(faults) - len(new), len(new), len(cleared)))
|
|
|
|
|
|
def run_audit(root, baseline_path, strict, quiet=False):
|
|
rows, bare = walk_root(root)
|
|
known = {}
|
|
if baseline_path and not strict:
|
|
known = load_baseline(baseline_path)
|
|
new, cleared = apply_baseline(rows, known)
|
|
if not quiet:
|
|
report(rows, bare, new, cleared, len(known))
|
|
return rows, bare, new, cleared
|
|
|
|
|
|
# ---------------------------------------------------------------- selftest
|
|
def _write(path, text):
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
fh.write(text)
|
|
|
|
|
|
def _manifest(adir, rels):
|
|
lines = []
|
|
for rel in rels:
|
|
lines.append("%s %s\n" % (sha256_file(os.path.join(adir, rel)), rel))
|
|
_write(os.path.join(adir, MANIFEST), "".join(lines))
|
|
|
|
|
|
def selftest() -> int:
|
|
cells, fails = [], []
|
|
|
|
def check(name, want, got):
|
|
cells.append(name)
|
|
if want != got:
|
|
fails.append("%s: want %r got %r" % (name, want, got))
|
|
print("FAIL %s: want %r got %r" % (name, want, got))
|
|
|
|
tmp = tempfile.mkdtemp(prefix="manifest-audit-selftest-")
|
|
try:
|
|
root = os.path.join(tmp, "root")
|
|
os.makedirs(root)
|
|
# clean artifact
|
|
_write(os.path.join(root, "clean", "REPORT.md"), "report\n")
|
|
_write(os.path.join(root, "clean", "check.sh"), "#!/bin/sh\nexit 0\n")
|
|
_manifest(os.path.join(root, "clean"), ["REPORT.md", "check.sh"])
|
|
# stale artifact: manifest written, then the file changes
|
|
_write(os.path.join(root, "stale", "REPORT.md"), "v1\n")
|
|
_manifest(os.path.join(root, "stale"), ["REPORT.md"])
|
|
_write(os.path.join(root, "stale", "REPORT.md"), "v1\nAPPENDED\n")
|
|
# missing entry: listed file deleted after the manifest
|
|
_write(os.path.join(root, "missing", "gone.txt"), "here\n")
|
|
_manifest(os.path.join(root, "missing"), ["gone.txt"])
|
|
os.unlink(os.path.join(root, "missing", "gone.txt"))
|
|
# foreign base: absolute path outside the artifact
|
|
outside = os.path.join(tmp, "outside.txt")
|
|
_write(outside, "elsewhere\n")
|
|
_write(os.path.join(root, "foreign", "REPORT.md"), "r\n")
|
|
_write(os.path.join(root, "foreign", MANIFEST),
|
|
"%s %s\n" % (sha256_file(outside), outside))
|
|
# empty manifest
|
|
_write(os.path.join(root, "empty", MANIFEST), "# no entries\n")
|
|
# no manifest at all: census only
|
|
_write(os.path.join(root, "bare", "REPORT.md"), "r\n")
|
|
|
|
rows, bare = walk_root(root)
|
|
by = {r["dir"]: r for r in rows}
|
|
check("c1 clean VERIFIED", VERIFIED, by["clean"]["verdict"])
|
|
check("c2 stale STALE", STALE, by["stale"]["verdict"])
|
|
check("c3 missing MISSING_ENTRY", MISSING_ENTRY, by["missing"]["verdict"])
|
|
check("c4 foreign FOREIGN_BASE", FOREIGN_BASE, by["foreign"]["verdict"])
|
|
check("c5 empty EMPTY", EMPTY, by["empty"]["verdict"])
|
|
check("c6 no-manifest is census", ["bare"], bare)
|
|
check("c7 hashed count", 2, sum(r["ok"] for r in rows))
|
|
|
|
known_none = apply_baseline(rows, {})
|
|
check("c8 no ledger: 4 new", 4, len(known_none[0]))
|
|
ledger = os.path.join(tmp, "baseline.tsv")
|
|
_write(ledger, "stale\tSTALE\ttest fixture\nmissing\tMISSING_ENTRY\tfixture\n")
|
|
known = load_baseline(ledger)
|
|
new, cleared = apply_baseline(rows, known)
|
|
check("c9 ledger suppresses 2", 2, len(new))
|
|
check("c10 known state", "KNOWN", {r["dir"]: r for r in rows}["stale"]["state"])
|
|
_write(ledger, "gone_dir\tSTALE\tno longer here\n")
|
|
new2, cleared2 = apply_baseline(rows, load_baseline(ledger))
|
|
check("c11 cleared row reported", [("gone_dir", STALE)], cleared2)
|
|
check("c12 cleared is not suppression", 4, len(new2))
|
|
|
|
# exit paths through the real CLI
|
|
rc = main(["--root", root, "--fail", "--baseline", os.path.join(tmp, "none.tsv")])
|
|
check("c13 unreadable ledger rc 2", 2, rc)
|
|
rc = main(["--root", os.path.join(tmp, "no-such-root")])
|
|
check("c14 missing root rc 2", 2, rc)
|
|
l2 = os.path.join(tmp, "b2.tsv")
|
|
_write(l2, "stale\tSTALE\tfixture\nmissing\tMISSING_ENTRY\tfixture\n"
|
|
"foreign\tFOREIGN_BASE\tfixture\nempty\tEMPTY\tfixture\n")
|
|
rc = main(["--root", root, "--fail", "--baseline", l2, "--quiet"])
|
|
check("c15 ledgered rc 0", 0, rc)
|
|
rc = main(["--root", root, "--strict", "--baseline", l2, "--quiet"])
|
|
check("c16 strict ignores ledger rc 1", 1, rc)
|
|
out = os.path.join(tmp, "rows.json")
|
|
rc = main(["--root", root, "--json", out, "--quiet"])
|
|
with open(out, encoding="utf-8") as fh:
|
|
blob = json.load(fh)
|
|
check("c17 json verdicts", ["empty", "foreign", "missing", "stale"],
|
|
sorted(r["dir"] for r in blob["rows"]
|
|
if r["verdict"] in FAULT_CLASSES))
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
print("selftest: %d cells, %d passed, %d failed" % (len(cells), len(cells) - len(fails), len(fails)))
|
|
for f in fails:
|
|
print(" " + f)
|
|
return 1 if fails else 0
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
ap = argparse.ArgumentParser(description="grade artifact files.sha256 manifests against the bytes beside them")
|
|
ap.add_argument("--root", default="/root/.hermes/wizard-outputs/volt",
|
|
help="artifact root: each subdirectory is one artifact (default: %(default)s)")
|
|
ap.add_argument("--baseline", default=None, help="ledger TSV: dir<TAB>class<TAB>note")
|
|
ap.add_argument("--strict", action="store_true", help="ignore the ledger entirely")
|
|
ap.add_argument("--fail", action="store_true", help="rc 1 when a NEW fault is graded")
|
|
ap.add_argument("--json", default=None, help="write the graded rows as JSON here")
|
|
ap.add_argument("--quiet", action="store_true", help="suppress the per-row report")
|
|
ap.add_argument("--selftest", action="store_true", help="grade the grader on fixtures")
|
|
args = ap.parse_args(argv)
|
|
|
|
if args.selftest:
|
|
return selftest()
|
|
if not os.path.isdir(args.root):
|
|
print("manifest_audit: cannot grade -- --root %s is not a directory" % args.root,
|
|
file=sys.stderr)
|
|
return 2
|
|
try:
|
|
rows, bare, new, cleared = run_audit(args.root, args.baseline, args.strict,
|
|
quiet=args.quiet)
|
|
except (OSError, ValueError) as exc:
|
|
print("manifest_audit: cannot grade -- %s" % exc, file=sys.stderr)
|
|
return 2
|
|
|
|
if args.quiet:
|
|
print("manifests: %d presented, %d without; new faults %d, cleared %d"
|
|
% (len(rows), len(bare), len(new), len(cleared)))
|
|
if args.json:
|
|
blob = {
|
|
"root": os.path.abspath(args.root),
|
|
"rows": rows,
|
|
"no_manifest": bare,
|
|
"new": [{"dir": r["dir"], "class": r["verdict"]} for r in new],
|
|
"cleared": [{"dir": d, "class": c} for d, c in cleared],
|
|
}
|
|
try:
|
|
with open(args.json, "w", encoding="utf-8") as fh:
|
|
json.dump(blob, fh, indent=2, sort_keys=True)
|
|
fh.write("\n")
|
|
except OSError as exc:
|
|
# House contract, graded by tools/write_probe.py: a target this host
|
|
# cannot write is rc 2 with one line naming it. rc 2 outranks the
|
|
# 0/1 the audit earned -- a caller who asked for a file and got a
|
|
# traceback cannot know a grade happened at all.
|
|
print("manifest_audit: cannot write --json %s: %s" % (args.json, exc),
|
|
file=sys.stderr)
|
|
return 2
|
|
if (args.fail or args.strict) and new:
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|