volt
f7194284d2
- 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
203 lines
7.3 KiB
Python
203 lines
7.3 KiB
Python
#!/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>
|
|
|
|
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
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
|
|
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 = []
|
|
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()
|
|
keep = []
|
|
for name in dirnames:
|
|
full = os.path.join(dirpath, name)
|
|
if os.path.islink(full):
|
|
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
|
|
h = hashlib.sha256()
|
|
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 _entries_to_dicts(entries):
|
|
out = []
|
|
for e in entries:
|
|
if e[1] == "symlink":
|
|
out.append({"path": e[0], "type": "symlink", "target": e[2]})
|
|
else:
|
|
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 = _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("MISSING: %s" % p)
|
|
continue
|
|
a = actual[p]
|
|
if a["type"] != e["type"]:
|
|
errors.append("TYPE_CHANGED: %s" % p)
|
|
continue
|
|
if e["type"] == "file":
|
|
if a["sha256"] != e["sha256"]:
|
|
errors.append("HASH_MISMATCH: %s" % p)
|
|
if a["size"] != e["size"]:
|
|
errors.append("SIZE_MISMATCH: %s" % p)
|
|
else:
|
|
if a["target"] != e["target"]:
|
|
errors.append("SYMLINK_TARGET_CHANGED: %s" % p)
|
|
for p in actual:
|
|
if p not in expected:
|
|
errors.append("EXTRA: %s" % 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]
|
|
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
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|