beam: deterministic directory manifest tool (make/check)

wizard-work/beam/manifest.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""Deterministic directory manifest: hash every regular file, verify integrity.
+
+Usage:
+  python3 manifest.py make <root> <out.json>
+  python3 manifest.py check <root> <manifest.json>
+"""
+import hashlib, json, os, sys
+
+
+def _walk(root):
+    entries = []
+    for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
+        dirnames.sort()
+        for name in sorted(filenames):
+            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)))
+                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)
+            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": []}
+    for e in entries:
+        if e[1] == "symlink":
+            manifest["entries"].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")
+    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]}
+
+    errors = []
+    for p, e in expected.items():
+        if p not in actual:
+            errors.append(f"MISSING: {p}")
+            continue
+        a = actual[p]
+        if a["type"] != e["type"]:
+            errors.append(f"TYPE_CHANGED: {p}")
+            continue
+        if e["type"] == "file":
+            if a["sha256"] != e["sha256"]:
+                errors.append(f"HASH_MISMATCH: {p}")
+            if a["size"] != e["size"]:
+                errors.append(f"SIZE_MISMATCH: {p}")
+        else:
+            if a["target"] != e["target"]:
+                errors.append(f"SYMLINK_TARGET_CHANGED: {p}")
+    for p in actual:
+        if p not in expected:
+            errors.append(f"EXTRA: {p}")
+    return errors
+
+
+def main(argv):
+    if len(argv) < 4:
+        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
+    print("unknown command", file=sys.stderr)
+    return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))