398 lines
15 KiB
Python
398 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""artifact_seal -- deterministic SHA-256 seal manifests over a file tree.
|
|
|
|
Lane: Codeforge. stdlib only. No third-party dependencies.
|
|
|
|
Why this exists
|
|
---------------
|
|
AI-generated artifacts drift silently: an editor strips a trailing newline, a
|
|
manually edited file changes by one character, a `git checkout` restores an
|
|
older revision. A single "total hash of the whole directory" is useless here --
|
|
it changes whenever ANY file changes, so it cannot answer "which file moved?".
|
|
|
|
This tool answers that question with a per-file digest plus a whole-set seal.
|
|
|
|
Properties that are load-bearing (see selftest.py):
|
|
* deterministic -- same bytes, same seal, on any platform, no timestamps
|
|
* tamper-evident -- flipping one bit in one file changes that file's digest
|
|
AND the whole-set seal; nothing else changes
|
|
* order-invariant -- reordering the entries inside the manifest does not
|
|
change the whole-set seal
|
|
* self-describing -- the manifest records the algorithm and the entry count,
|
|
so a manifest from a different tool version is rejected
|
|
* no false positives -- distinct byte streams never share a digest here; the
|
|
whole-set digest is not a plain concatenation digest
|
|
(naive concatenation is ambiguous across path lengths:
|
|
["ab","c"] and ["a","bc"] would collide)
|
|
|
|
Public API
|
|
----------
|
|
seal(paths, *, root=".", prefix=None) -> dict # seal a set of paths
|
|
seal_dir(dirpath, *, prefix=None) -> dict # seal a directory tree
|
|
write_manifest(manifest, out) -> str # canonical text form
|
|
parse_manifest(text) -> dict
|
|
verify_manifest(manifest, *, root=".") -> (ok, problems)
|
|
main(argv) -> int # CLI
|
|
|
|
CLI
|
|
---
|
|
python3 artifact_seal.py seal DIR [--out FILE] [--prefix P]
|
|
python3 artifact_seal.py verify --manifest FILE [--target DIR]
|
|
python3 artifact_seal.py seal FILE ... --out FILE
|
|
|
|
`verify` reads the tree from --target (default: the current directory).
|
|
Passing the directory as a bare positional is NOT accepted: it is ambiguous
|
|
with the manifest's own path and was a real source of silent wrong-tree
|
|
verification.
|
|
|
|
Exit code 0 iff verification succeeds (or a seal was written).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
ALGO = "sha256"
|
|
FORMAT = "artifact-seal/1"
|
|
# Canonical ordering: by relative POSIX path, then by name, so the seal does not
|
|
# depend on filesystem readdir order or on the order paths were passed in.
|
|
_KEY = lambda e: (e["path"], e["name"])
|
|
|
|
|
|
class SealError(Exception):
|
|
"""Raised for malformed manifests, missing files, or digest mismatch."""
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# hashing
|
|
# --------------------------------------------------------------------------
|
|
|
|
def hash_bytes(data: bytes) -> str:
|
|
"""Hex digest of raw bytes."""
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def hash_text(text: str) -> str:
|
|
"""Hex digest of TEXT as bytes, with the trailing-newline question settled.
|
|
|
|
A file ending in '\\n' and the same file without it are DIFFERENT content --
|
|
the digest differs. (Do not "normalise" this away; a stripped trailing
|
|
newline is exactly the kind of silent drift this tool exists to catch.)
|
|
"""
|
|
return hash_bytes(text.encode("utf-8"))
|
|
|
|
|
|
def _read(path: str) -> bytes:
|
|
try:
|
|
with open(path, "rb") as fh:
|
|
return fh.read()
|
|
except OSError as exc:
|
|
raise SealError(f"cannot read {path!r}: {exc}") from exc
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# manifest construction
|
|
# --------------------------------------------------------------------------
|
|
|
|
def _entry(root: str, path: str, prefix: str | None) -> dict:
|
|
"""Build one manifest entry.
|
|
|
|
The digest covers the path AS WRITTEN IN THE MANIFEST, so a rename is a
|
|
detectable event (new path -> no matching entry -> 'missing'/'extra').
|
|
"""
|
|
data = _read(path)
|
|
rel = os.path.relpath(path, root)
|
|
if prefix is not None:
|
|
rel = rel[len(prefix):].lstrip("/\\") if rel.startswith(prefix) else rel
|
|
rel = rel.replace(os.sep, "/")
|
|
return {
|
|
"path": rel,
|
|
"name": os.path.basename(rel),
|
|
"size": len(data),
|
|
"sha256": hash_bytes(data),
|
|
}
|
|
|
|
|
|
def seal(paths, *, root: str = ".", prefix: str | None = None) -> dict:
|
|
"""Seal an explicit iterable of paths (relative to CWD or to `root`)."""
|
|
entries = []
|
|
for p in paths:
|
|
entries.append(_entry(root, p, prefix))
|
|
return _assemble(entries)
|
|
|
|
|
|
def seal_dir(dirpath: str, *, prefix: str | None = None) -> dict:
|
|
"""Seal every regular file under `dirpath`, recursing into subdirectories.
|
|
|
|
Symlinks are skipped (they are not content), as are FIFOs and other
|
|
non-regular files: hashing those is either infinite or meaningless.
|
|
Hidden files (dot-prefixed) ARE included -- .gitignore etc. are content too.
|
|
"""
|
|
root = os.path.abspath(dirpath)
|
|
found: list[str] = []
|
|
for dirpath_, dirnames, filenames in os.walk(root):
|
|
dirnames.sort()
|
|
for fn in sorted(filenames):
|
|
full = os.path.join(dirpath_, fn)
|
|
if os.path.islink(full) or not os.path.isfile(full):
|
|
continue
|
|
found.append(full)
|
|
return seal(found, root=root, prefix=prefix)
|
|
|
|
|
|
def _assemble(entries: list[dict]) -> dict:
|
|
"""Canonicalise entries and compute the whole-set seal.
|
|
|
|
The whole-set digest is sha256 over the canonical byte encoding of the
|
|
sorted entries -- NOT a concatenation of the digests. A plain
|
|
concatenation is ambiguous (entries ["ab","c"] vs ["a","bc"] both hash the
|
|
same joined string); length-prefixing each field removes that ambiguity.
|
|
"""
|
|
entries = sorted(entries, key=_KEY)
|
|
if len({e["path"] for e in entries}) != len(entries):
|
|
dupes = sorted({e["path"] for e in entries
|
|
if sum(1 for x in entries if x["path"] == e["path"]) > 1})
|
|
raise SealError(f"duplicate paths in manifest: {dupes}")
|
|
lines = []
|
|
for e in entries:
|
|
for field in (e["path"], e["name"], str(e["size"]), e["sha256"]):
|
|
lines.append(f"{len(field)}:{field}")
|
|
canonical = "\n".join(lines) + ("\n" if lines else "")
|
|
return {
|
|
"format": FORMAT,
|
|
"algo": ALGO,
|
|
"count": len(entries),
|
|
"entries": entries,
|
|
"seal": hash_text(canonical),
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# serialisation
|
|
# --------------------------------------------------------------------------
|
|
|
|
def write_manifest(manifest: dict, out: str) -> str:
|
|
"""Write the canonical manifest text; returns the text written.
|
|
|
|
Canonical form: sorted keys, no incidental whitespace, UTF-8, one trailing
|
|
newline. A manifest that has merely been re-serialised must verify.
|
|
"""
|
|
text = json.dumps(manifest, sort_keys=True, indent=2, ensure_ascii=False)
|
|
text = text + "\n"
|
|
with open(out, "w", encoding="utf-8") as fh:
|
|
fh.write(text)
|
|
return text
|
|
|
|
|
|
def parse_manifest(text: str) -> dict:
|
|
"""Parse manifest text, rejecting anything that is not our format."""
|
|
try:
|
|
m = json.loads(text)
|
|
except json.JSONDecodeError as exc:
|
|
raise SealError(f"manifest is not valid JSON: {exc}") from exc
|
|
if not isinstance(m, dict):
|
|
raise SealError("manifest root is not an object")
|
|
if m.get("format") != FORMAT:
|
|
raise SealError(f"unsupported manifest format: {m.get('format')!r}")
|
|
if m.get("algo") != ALGO:
|
|
raise SealError(f"unsupported algorithm: {m.get('algo')!r}")
|
|
entries = m.get("entries")
|
|
if not isinstance(entries, list):
|
|
raise SealError("manifest has no entries list")
|
|
for e in entries:
|
|
if not isinstance(e, dict) or not isinstance(e.get("path"), str):
|
|
raise SealError(f"malformed entry: {e!r}")
|
|
if not isinstance(e.get("sha256"), str) or len(e["sha256"]) != 64:
|
|
raise SealError(f"entry {e.get('path')!r} has a bad digest")
|
|
if not all(c in "0123456789abcdef" for c in e["sha256"]):
|
|
raise SealError(f"entry {e.get('path')!r} digest is not lowercase hex")
|
|
if m.get("count") != len(entries):
|
|
raise SealError(
|
|
f"manifest count {m.get('count')!r} != actual entries {len(entries)}")
|
|
return m
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# verification
|
|
# --------------------------------------------------------------------------
|
|
|
|
def _recompute_seal(entries: list[dict]) -> str:
|
|
lines = []
|
|
for e in sorted(entries, key=_KEY):
|
|
for field in (e["path"], e["name"], str(e["size"]), e["sha256"]):
|
|
lines.append(f"{len(field)}:{field}")
|
|
canonical = "\n".join(lines) + ("\n" if lines else "")
|
|
return hash_text(canonical)
|
|
|
|
|
|
def verify_manifest(manifest: dict, *, root: str = "."):
|
|
"""Verify a manifest against the filesystem rooted at `root`.
|
|
|
|
Returns (ok: bool, problems: list[str]). Never raises for content
|
|
problems -- a list of problems is the result. A structural problem with
|
|
the manifest itself (bad JSON, wrong format, wrong count) raises SealError
|
|
because there is nothing sensible to compare against.
|
|
|
|
Checks, in order:
|
|
1. the manifest's own whole-set seal matches its entries
|
|
(catches hand-edits to the manifest file itself)
|
|
2. every entry exists, has the recorded size, and hashes to the recorded
|
|
digest
|
|
3. no extra files exist on disk that the manifest does not list
|
|
4. no duplicate paths
|
|
"""
|
|
problems: list[str] = []
|
|
entries = manifest["entries"]
|
|
paths = [e["path"] for e in entries]
|
|
if len(set(paths)) != len(paths):
|
|
problems.append("duplicate paths in manifest")
|
|
|
|
actual_seal = _recompute_seal(entries)
|
|
if actual_seal != manifest.get("seal"):
|
|
problems.append(
|
|
f"manifest seal mismatch: recorded {manifest.get('seal')!r}, "
|
|
f"recomputed {actual_seal!r}")
|
|
|
|
for e in sorted(entries, key=_KEY):
|
|
p = os.path.join(root, e["path"])
|
|
if not os.path.exists(p):
|
|
problems.append(f"missing: {e['path']}")
|
|
continue
|
|
if os.path.islink(p):
|
|
problems.append(f"is a symlink (not content): {e['path']}")
|
|
continue
|
|
if not os.path.isfile(p):
|
|
problems.append(f"not a regular file: {e['path']}")
|
|
continue
|
|
data = _read(p)
|
|
if len(data) != e["size"]:
|
|
# A size mismatch is evidence, not a verdict: report it AND still
|
|
# hash. Reporting only the size would silently swallow a tamper
|
|
# that happens to preserve length (e.g. appending to a 4-byte file
|
|
# whose digest no longer matches) -- the caller would be told
|
|
# "size" when the real finding is "content changed".
|
|
problems.append(
|
|
f"size mismatch: {e['path']} (manifest {e['size']}, "
|
|
f"actual {len(data)})")
|
|
got = hash_bytes(data)
|
|
if got != e["sha256"]:
|
|
problems.append(
|
|
f"digest mismatch: {e['path']} (manifest {e['sha256']}, "
|
|
f"actual {got})")
|
|
|
|
listed = set(paths)
|
|
disk = set()
|
|
for dirpath_, dirnames, filenames in os.walk(root):
|
|
dirnames.sort()
|
|
for fn in sorted(filenames):
|
|
full = os.path.join(dirpath_, fn)
|
|
if os.path.islink(full) or not os.path.isfile(full):
|
|
continue
|
|
disk.add(os.path.relpath(full, root).replace(os.sep, "/"))
|
|
for extra in sorted(disk - listed):
|
|
problems.append(f"extra file not in manifest: {extra}")
|
|
|
|
return (not problems), problems
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# CLI
|
|
# --------------------------------------------------------------------------
|
|
|
|
def _usage(stream) -> None:
|
|
stream.write(__doc__ or "")
|
|
|
|
|
|
def _positional(args: list[str], options=("--prefix", "--manifest", "--out",
|
|
"--target")) -> list[str]:
|
|
"""Args that are not an option and not the value of a preceding option."""
|
|
skip_next = False
|
|
out = []
|
|
for a in args:
|
|
if skip_next:
|
|
skip_next = False
|
|
continue
|
|
if a in options:
|
|
skip_next = True
|
|
continue
|
|
if a.startswith("--"):
|
|
continue
|
|
out.append(a)
|
|
return out
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
args = list(argv[1:])
|
|
if not args or args[0] in ("-h", "--help"):
|
|
_usage(sys.stdout)
|
|
return 0 if args else 2
|
|
|
|
cmd = args[0]
|
|
rest = args[1:]
|
|
|
|
def opt(flag: str) -> str | None:
|
|
if flag in rest:
|
|
i = rest.index(flag)
|
|
if i + 1 >= len(rest):
|
|
sys.stderr.write(f"error: {flag} needs a value\n")
|
|
raise SystemExit(2)
|
|
return rest[i + 1]
|
|
return None
|
|
|
|
prefix = opt("--prefix")
|
|
manifest_path = opt("--manifest") or opt("--out")
|
|
target_opt = opt("--target")
|
|
pos = _positional(rest)
|
|
|
|
if cmd == "seal":
|
|
if not pos:
|
|
sys.stderr.write("error: seal needs at least one path\n")
|
|
return 2
|
|
first = pos[0]
|
|
if len(pos) == 1 and os.path.isdir(first):
|
|
manifest = seal_dir(first, prefix=prefix)
|
|
else:
|
|
manifest = seal(pos, prefix=prefix)
|
|
if not manifest_path:
|
|
sys.stdout.write(json.dumps(manifest, sort_keys=True, indent=2) + "\n")
|
|
return 0
|
|
write_manifest(manifest, manifest_path)
|
|
print(f"sealed {manifest['count']} file(s) -> {manifest_path} "
|
|
f"[{manifest['seal'][:16]}...]")
|
|
return 0
|
|
|
|
if cmd == "verify":
|
|
if not manifest_path:
|
|
sys.stderr.write("error: verify needs --manifest FILE\n")
|
|
return 2
|
|
target = target_opt or (pos[0] if pos else ".")
|
|
try:
|
|
text = _read(manifest_path).decode("utf-8")
|
|
except (OSError, UnicodeDecodeError) as exc:
|
|
sys.stderr.write(f"error: cannot read manifest {manifest_path!r}: {exc}\n")
|
|
return 1
|
|
try:
|
|
manifest = parse_manifest(text)
|
|
except SealError as exc:
|
|
sys.stderr.write(f"error: {exc}\n")
|
|
return 1
|
|
ok, problems = verify_manifest(manifest, root=target)
|
|
for p in problems:
|
|
print(f"FAIL {p}")
|
|
if ok:
|
|
print(f"OK {manifest['count']} file(s) verified, "
|
|
f"seal {manifest['seal'][:16]}...")
|
|
return 0
|
|
return 1
|
|
|
|
sys.stderr.write(f"error: unknown command {cmd!r}\n")
|
|
_usage(sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv))
|