stackchain/troupe-standup
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
- 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
This commit is contained in:
volt committed 2026-10-26 17:58:37 +00:00
1 parent 65be4da7a0
commit f7194284d2
1 file changed
+142 -47
Download Patch File Download Diff File Expand all files Collapse all files
wizard-work/beam/manifest.py
+142 -47
View File
@@ -4,81 +4,172 @@
Usage:
python3 manifest.py make <root> <out.json>
python3 manifest.py check <root> <manifest.json>
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 <root> <file>", 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