beam: add manifest_check tool with tests
This commit is contained in:
1 parent
6b5fc526de
commit
50cc5bba37
2 files changed
+93
No files matched your search
+78
@@ -0,0 +1,78 @@
|
||||
#!/usr/bin/env python3
|
||||
"""manifest_check: build and verify a sha256 file manifest for a directory."""
|
||||
from __future__ import annotations
|
||||
import argparse, hashlib, json, os, sys
|
||||
from pathlib import Path
|
||||
|
||||
def sha256_file(p: Path) -> str:
|
||||
h = hashlib.sha256()
|
||||
with p.open("rb") as f:
|
||||
for chunk in iter(lambda: f.read(65536), b""):
|
||||
h.update(chunk)
|
||||
return h.hexdigest()
|
||||
|
||||
def build(root: Path) -> dict:
|
||||
entries = []
|
||||
for p in sorted(root.rglob("*")):
|
||||
if p.is_file():
|
||||
rel = p.relative_to(root).as_posix()
|
||||
entries.append({"path": rel, "sha256": sha256_file(p), "size": p.stat().st_size})
|
||||
return {"root": str(root), "count": len(entries), "entries": entries}
|
||||
|
||||
def verify(manifest: dict, root: Path) -> tuple[bool, list[str]]:
|
||||
errors = []
|
||||
# expected files
|
||||
expected = {e["path"]: e for e in manifest["entries"]}
|
||||
actual_files = set()
|
||||
for p in root.rglob("*"):
|
||||
if p.is_file():
|
||||
actual_files.add(p.relative_to(root).as_posix())
|
||||
# missing
|
||||
for path in expected:
|
||||
if path not in actual_files:
|
||||
errors.append(f"missing: {path}")
|
||||
# extra
|
||||
for path in actual_files:
|
||||
if path not in expected:
|
||||
errors.append(f"extra: {path}")
|
||||
# content
|
||||
for path, e in expected.items():
|
||||
if path in actual_files:
|
||||
p = root / path
|
||||
actual_hash = sha256_file(p)
|
||||
if actual_hash != e["sha256"]:
|
||||
errors.append(f"changed: {path} (expected {e['sha256']}, got {actual_hash})")
|
||||
if p.stat().st_size != e["size"]:
|
||||
errors.append(f"size mismatch: {path}")
|
||||
if manifest.get("count") != len(expected):
|
||||
errors.append(f"count field {manifest['count']} != {len(expected)}")
|
||||
return (len(errors) == 0, errors)
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||||
b = sub.add_parser("build")
|
||||
b.add_argument("root")
|
||||
b.add_argument("-o", "--output", required=True)
|
||||
v = sub.add_parser("verify")
|
||||
v.add_argument("manifest")
|
||||
v.add_argument("root")
|
||||
a = ap.parse_args()
|
||||
if a.cmd == "build":
|
||||
root = Path(a.root)
|
||||
m = build(root)
|
||||
Path(a.output).write_text(json.dumps(m, indent=2) + "\n")
|
||||
print(f"built manifest: {m['count']} files -> {a.output}")
|
||||
return 0
|
||||
else:
|
||||
manifest = json.loads(Path(a.manifest).read_text())
|
||||
ok, errors = verify(manifest, Path(a.root))
|
||||
if ok:
|
||||
print(f"OK: {manifest['count']} files verified")
|
||||
return 0
|
||||
for e in errors:
|
||||
print("FAIL:", e)
|
||||
return 1
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,15 @@
|
||||
import subprocess, json, tempfile, os, pathlib, sys
|
||||
base = pathlib.Path("/root/troupe-standup/wizard-work/beam/manifest-check/manifest_check.py")
|
||||
d = pathlib.Path(tempfile.mkdtemp())
|
||||
(d/"a.txt").write_text("hello")
|
||||
(d/"sub").mkdir()
|
||||
(d/"sub/b.txt").write_text("world")
|
||||
m = pathlib.Path(tempfile.mkdtemp())/"manifest.json"
|
||||
subprocess.run([sys.executable, str(base), "build", str(d), "-o", str(m)], check=True)
|
||||
r = subprocess.run([sys.executable, str(base), "verify", str(m), str(d)])
|
||||
assert r.returncode == 0
|
||||
# tamper
|
||||
(d/"a.txt").write_text("tampered")
|
||||
r2 = subprocess.run([sys.executable, str(base), "verify", str(m), str(d)])
|
||||
assert r2.returncode == 1
|
||||
print("all tests passed")
|
||||
Reference in new issue
Block a user