#!/usr/bin/env python3 """Cyberdelia Archive Integrity Verifier. Verifies SHA-256, byte counts, public archive paths, and provenance sidecars for Cyberdelia-compatible repository-tier archive manifests. Exit codes: 0 = all verified objects passed integrity checks 1 = one or more integrity failures 2 = configuration / invocation error """ from __future__ import annotations import argparse import hashlib import json from pathlib import Path import sys VERSION = "0.1.0" def digest(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def local_path(root: Path, public_path: str, ftp_prefix: str) -> Path: rel = public_path.lstrip("/") target = (root / rel).resolve() root_resolved = root.resolve() if not rel.startswith(ftp_prefix.rstrip("/") + "/"): raise ValueError(f"invalid archive path outside {ftp_prefix}: {public_path}") if root_resolved not in target.parents: raise ValueError(f"archive path escaped root: {public_path}") return target def verify_object(root: Path, ftp_prefix: str, label: str, public_path: str, expected_hash: str, expected_bytes=None) -> list[str]: errors: list[str] = [] try: path = local_path(root, public_path, ftp_prefix) except Exception as exc: return [f"{label}: {exc}"] if not path.exists() or not path.is_file(): return [f"{label}: missing local bytes at {public_path}"] actual_hash = digest(path) if not expected_hash or len(expected_hash) != 64: errors.append(f"{label}: invalid/missing recorded SHA-256") elif actual_hash.lower() != expected_hash.lower(): errors.append(f"{label}: SHA-256 mismatch expected={expected_hash} actual={actual_hash}") if expected_bytes is not None: try: recorded = int(expected_bytes) actual_bytes = path.stat().st_size if actual_bytes != recorded: errors.append(f"{label}: byte-count mismatch expected={recorded} actual={actual_bytes}") except Exception: errors.append(f"{label}: invalid recorded byte count {expected_bytes!r}") sidecar = Path(str(path) + ".provenance.json") if not sidecar.exists(): errors.append(f"{label}: missing provenance sidecar {sidecar.relative_to(root)}") else: try: meta = json.loads(sidecar.read_text(encoding="utf-8")) if meta.get("sha256", "").lower() != actual_hash.lower(): errors.append(f"{label}: provenance sidecar hash does not match bytes") if meta.get("local_public_path") and meta.get("local_public_path") != public_path: errors.append(f"{label}: provenance local path disagrees with manifest") except Exception as exc: errors.append(f"{label}: unreadable provenance sidecar: {exc}") return errors def verify_case_manifest(root: Path, ftp_prefix: str, manifest: Path) -> tuple[int, list[str]]: if not manifest.exists(): return 0, [f"case manifest missing: {manifest}"] try: data = json.loads(manifest.read_text(encoding="utf-8")) except Exception as exc: return 0, [f"case manifest unreadable: {exc}"] count = 0 errors: list[str] = [] for case in data.get("cases", []): for source in case.get("sources", []): if source.get("local_mirror_status") != "verified": continue count += 1 path = source.get("local_mirror") or source.get("mock_ftp_path") errors.extend(verify_object( root, ftp_prefix, f"{case.get('id','case')} / {source.get('title','source')}", path or "", source.get("sha256", ""), source.get("retrieval_bytes"), )) return count, errors def verify_collection_manifests(root: Path, ftp_prefix: str, ftp_root: Path) -> tuple[int, list[str]]: count = 0 errors: list[str] = [] if not ftp_root.exists(): return count, errors for manifest in ftp_root.rglob("MANIFEST.json"): try: data = json.loads(manifest.read_text(encoding="utf-8")) except Exception as exc: errors.append(f"{manifest.relative_to(root)}: unreadable manifest: {exc}") continue objects = data.get("objects") if not isinstance(objects, list): continue for obj in objects: if obj.get("status") != "verified": continue public_path = obj.get("local_public_path") if not public_path: errors.append(f"{manifest.relative_to(root)}: verified object lacks local_public_path") continue count += 1 errors.extend(verify_object( root, ftp_prefix, f"{data.get('collection_id', manifest.parent.name)} / {obj.get('filename', public_path)}", public_path, obj.get("sha256", ""), obj.get("bytes"), )) return count, errors def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="Verify Cyberdelia-compatible archive custody manifests.") p.add_argument("--root", default=".", help="Archive repository root. Default: current directory.") p.add_argument("--case-manifest", default="data/case-files-manifest.json", help="Case-file manifest path relative to root.") p.add_argument("--ftp-root", default="ftp", help="FTP/archive tree path relative to root.") p.add_argument("--ftp-prefix", default="ftp", help="Required public-path prefix for mirrored objects. Default: ftp") p.add_argument("--version", action="version", version=f"%(prog)s {VERSION}") return p def main() -> int: args = build_parser().parse_args() try: root = Path(args.root).expanduser().resolve() case_manifest = (root / args.case_manifest).resolve() ftp_root = (root / args.ftp_root).resolve() except Exception as exc: print(f"CONFIGURATION ERROR: {exc}", file=sys.stderr) return 2 if not root.exists() or not root.is_dir(): print(f"CONFIGURATION ERROR: root is not a directory: {root}", file=sys.stderr) return 2 case_count, case_errors = verify_case_manifest(root, args.ftp_prefix, case_manifest) collection_count, collection_errors = verify_collection_manifests(root, args.ftp_prefix, ftp_root) errors = case_errors + collection_errors total = case_count + collection_count if errors: print(f"ARCHIVE INTEGRITY FAILED objects_checked={total} errors={len(errors)}", file=sys.stderr) for error in errors: print(f"ERROR {error}", file=sys.stderr) return 1 print(f"ARCHIVE INTEGRITY VERIFIED objects_checked={total}") return 0 if __name__ == "__main__": raise SystemExit(main())