216 lines
7.6 KiB
Python
216 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
|
"""packfreeze: deterministic one-pass package manifest generator.
|
|
|
|
Walks a directory tree, reads every regular file's real content, and writes a
|
|
lockfile with blake2b digests plus per-package aggregate digests. Output is
|
|
byte-identical across runs and independent of filesystem ordering or mtimes.
|
|
|
|
Usage:
|
|
packfreeze make --root DIR --out LOCK.json [--include GLOB ...]
|
|
packfreeze verify --root DIR --lock LOCK.json
|
|
packfreeze check --root DIR --lock LOCK.json --output REPORT.json
|
|
|
|
Stdlib only. Digests are always computed from file bytes on disk.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Dict, List, Optional, Tuple
|
|
|
|
CHUNK = 1 << 16 # 64 KiB streaming read
|
|
|
|
EXCLUDE_DIRS = frozenset(
|
|
{".git", ".hg", ".svn", "__pycache__", ".mypy_cache", ".pytest_cache", ".tox", ".venv"}
|
|
)
|
|
|
|
|
|
def file_digest(path: str) -> str:
|
|
"""blake2b digest of the real bytes at `path`, streamed from disk."""
|
|
h = hashlib.blake2b()
|
|
with open(path, "rb") as fh:
|
|
for block in iter(lambda: fh.read(CHUNK), b""):
|
|
h.update(block)
|
|
return h.hexdigest()
|
|
|
|
|
|
def dir_digest(entries: List[Tuple[str, str]]) -> str:
|
|
"""Aggregate digest over (relative path, file digest) pairs, order-stable."""
|
|
h = hashlib.blake2b()
|
|
for rel, dig in sorted(entries):
|
|
h.update(rel.encode("utf-8"))
|
|
h.update(b"\0")
|
|
h.update(dig.encode("ascii"))
|
|
h.update(b"\n")
|
|
return h.hexdigest()
|
|
|
|
|
|
def _match(name: str, includes: Optional[List[str]]) -> bool:
|
|
"""Glob match on path or basename (fnmatch-style, case-sensitive)."""
|
|
import fnmatch
|
|
|
|
if not includes:
|
|
return True
|
|
return any(fnmatch.fnmatch(name, pat) for pat in includes)
|
|
|
|
|
|
def walk(root: str, includes: Optional[List[str]] = None) -> List[Tuple[str, str, int]]:
|
|
"""Return [(relative_path, digest, size)] for matching regular files.
|
|
|
|
Symlinks are skipped (never followed) so a link cannot smuggle content in or
|
|
out of the tree. The walk order is normalized by the caller.
|
|
"""
|
|
out: List[Tuple[str, str, int]] = []
|
|
root_abs = os.path.abspath(root)
|
|
for dirpath, dirnames, filenames in os.walk(root_abs, followlinks=False):
|
|
dirnames[:] = sorted(d for d in dirnames if d not in EXCLUDE_DIRS)
|
|
for fn in sorted(filenames):
|
|
full = os.path.join(dirpath, fn)
|
|
rel = os.path.relpath(full, root_abs).replace(os.sep, "/")
|
|
if not _match(rel, includes):
|
|
continue
|
|
if os.path.islink(full) or not os.path.isfile(full):
|
|
continue
|
|
out.append((rel, file_digest(full), os.path.getsize(full)))
|
|
out.sort()
|
|
return out
|
|
|
|
|
|
def build(root: str, includes: Optional[List[str]] = None) -> dict:
|
|
"""Build the lockfile dict from a real walk of `root`."""
|
|
entries = walk(root, includes)
|
|
files = {rel: dig for rel, dig, _ in entries}
|
|
|
|
# Derive packages: top-level directory, or the file itself at root.
|
|
packages: Dict[str, List[Tuple[str, str]]] = {}
|
|
for rel, dig, _ in entries:
|
|
pkg = rel.split("/", 1)[0] if "/" in rel else "(root)"
|
|
packages.setdefault(pkg, []).append((rel, dig))
|
|
|
|
pkg_digests = {name: dir_digest(items) for name, items in packages.items()}
|
|
return {
|
|
"format": 1,
|
|
"generator": "packfreeze 1.0.0",
|
|
"root": os.path.abspath(root),
|
|
"file_count": len(entries),
|
|
"total_bytes": sum(size for _, _, size in entries),
|
|
"digest_algorithm": "blake2b",
|
|
"packages": pkg_digests,
|
|
"tree_digest": dir_digest([(rel, dig) for rel, dig, _ in entries]),
|
|
"files": files,
|
|
}
|
|
|
|
|
|
def _cmp_lock(root: str, lock: dict) -> Tuple[List[str], List[str], List[str], List[str]]:
|
|
"""Compare a real walk against a lockfile.
|
|
|
|
Returns (missing, changed, added, extra) where extra = files named in the
|
|
lock that are no longer present, and added = files present but not locked.
|
|
"""
|
|
expected = lock.get("files") or {}
|
|
actual = {rel: dig for rel, dig, _ in walk(root)}
|
|
missing: List[str] = []
|
|
changed: List[str] = []
|
|
for rel, dig in sorted(expected.items()):
|
|
if rel not in actual:
|
|
missing.append(rel)
|
|
elif actual[rel] != dig:
|
|
changed.append(rel)
|
|
added = sorted(set(actual) - set(expected))
|
|
extra = sorted(set(expected) - set(actual))
|
|
return missing, changed, added, extra
|
|
|
|
|
|
def cmd_make(args: argparse.Namespace) -> int:
|
|
lock = build(args.root, args.include)
|
|
payload = json.dumps(lock, indent=2, sort_keys=True) + "\n"
|
|
with open(args.out, "w", encoding="utf-8") as fh:
|
|
fh.write(payload)
|
|
print(
|
|
"made {out}: {n} files, {b} bytes, tree_digest={d}".format(
|
|
out=args.out, n=lock["file_count"], b=lock["total_bytes"], d=lock["tree_digest"][:16]
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
def cmd_verify(args: argparse.Namespace) -> int:
|
|
with open(args.lock, "r", encoding="utf-8") as fh:
|
|
lock = json.load(fh)
|
|
missing, changed, added, extra = _cmp_lock(args.root, lock)
|
|
ok = not (missing or changed or added or extra)
|
|
if ok:
|
|
print("OK: {n} files match tree_digest {d}".format(n=lock["file_count"], d=lock["tree_digest"][:16]))
|
|
else:
|
|
for rel in missing:
|
|
print("MISSING " + rel)
|
|
for rel in changed:
|
|
print("CHANGED " + rel)
|
|
for rel in added:
|
|
print("ADDED " + rel)
|
|
for rel in extra:
|
|
print("STALE " + rel)
|
|
print("FAIL: {} missing, {} changed, {} added, {} stale".format(
|
|
len(missing), len(changed), len(added), len(extra)))
|
|
return 0 if ok else 1
|
|
|
|
|
|
def cmd_check(args: argparse.Namespace) -> int:
|
|
"""Rebuild in memory, diff against the lock, write a machine-readable report."""
|
|
with open(args.lock, "r", encoding="utf-8") as fh:
|
|
lock = json.load(fh)
|
|
rebuilt = build(args.root)
|
|
missing, changed, added, extra = _cmp_lock(args.root, lock)
|
|
report = {
|
|
"ok": not (missing or changed or added or extra),
|
|
"root": rebuilt["root"],
|
|
"locked_tree_digest": lock.get("tree_digest"),
|
|
"observed_tree_digest": rebuilt["tree_digest"],
|
|
"file_count": rebuilt["file_count"],
|
|
"missing": missing,
|
|
"changed": changed,
|
|
"added": added,
|
|
"stale": extra,
|
|
}
|
|
with open(args.output, "w", encoding="utf-8") as fh:
|
|
fh.write(json.dumps(report, indent=2, sort_keys=True) + "\n")
|
|
print("report {p}: ok={ok} files={n}".format(p=args.output, ok=report["ok"], n=report["file_count"]))
|
|
return 0 if report["ok"] else 1
|
|
|
|
|
|
def main(argv: Optional[List[str]] = None) -> int:
|
|
ap = argparse.ArgumentParser(prog="packfreeze", description=__doc__)
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
|
|
p_make = sub.add_parser("make", help="write a lockfile from a real directory walk")
|
|
p_make.add_argument("--root", required=True)
|
|
p_make.add_argument("--out", required=True)
|
|
p_make.add_argument("--include", action="append", default=[])
|
|
p_make.set_defaults(fn=cmd_make)
|
|
|
|
p_ver = sub.add_parser("verify", help="check a tree against a lockfile")
|
|
p_ver.add_argument("--root", required=True)
|
|
p_ver.add_argument("--lock", required=True)
|
|
p_ver.set_defaults(fn=cmd_verify)
|
|
|
|
p_chk = sub.add_parser("check", help="diff tree vs lock, write JSON report")
|
|
p_chk.add_argument("--root", required=True)
|
|
p_chk.add_argument("--lock", required=True)
|
|
p_chk.add_argument("--output", required=True)
|
|
p_chk.set_defaults(fn=cmd_check)
|
|
|
|
args = ap.parse_args(argv)
|
|
return int(args.fn(args))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|
|
|
|
|
|
|
|
|