#!/usr/bin/env python3 """Create and verify the local immutable Task26 strict-rerun retention object.""" from __future__ import annotations import argparse import fcntl import hashlib import json import os import shutil import stat import subprocess import sys import tempfile from pathlib import Path from typing import Any SCHEMA = "task26-local-immutable-retention-v1" MECHANISM = "root-owned-ext4-immutable-content-addressed-copy-v1" FS_IOC_GETFLAGS = 0x80086601 FS_IMMUTABLE_FL = 0x00000010 EXPECTED = { "archive_root_sha256": "9f5d72840b4cc1fcb57720a61ebc53529fa00202d9201615e4ebdbb3e047a635", "archive_entry_count": 172, "candidate_digest": "4a6c7ee54cf9526a30de8bb576c1d71b411938beba33a04914738f6e1b6ed1cb", "lifecycle_root_sha256": "fdcfb1efd1f2d4a2b95f22f7311827d197898a6d480cb572956e87bd66d20361", "lifecycle_seal_sha256": "50e7556e58a876d8136fbfb2023ede8bcb46ac816d97d4b85e1d0559f463d008", "v3_seal_sha256": "84d937ef67c1517bfdbca4cabce9fb4bfbe39aa1165458bb3aedd14decc4bff0", "v4_seal_sha256": "188e9e630b53e9329ae185fc8c007aecb7988c16d174d325d68e8e004edb38e9", "run_id": "task26-current-4a6c7ee5-20260817", } class RetentionError(RuntimeError): pass def canon(value: Any) -> bytes: return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode() def sha(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def file_sha(path: Path) -> str: return sha(path.read_bytes()) def immutable(path: Path) -> bool: fd = os.open(path, os.O_RDONLY | os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0)) try: flags = bytearray(4) fcntl.ioctl(fd, FS_IOC_GETFLAGS, flags, True) return bool(int.from_bytes(flags, "little") & FS_IMMUTABLE_FL) finally: os.close(fd) def source_inventory(root: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for base, dirs, files in os.walk(root, followlinks=False): dirs.sort() files.sort() for name in dirs: path = Path(base) / name st = path.lstat() if not stat.S_ISDIR(st.st_mode): raise RetentionError(f"non-directory archive entry: {path}") rows.append({"path": path.relative_to(root).as_posix(), "type": "D", "mode": f"{stat.S_IMODE(st.st_mode):04o}"}) for name in files: path = Path(base) / name st = path.lstat() if not stat.S_ISREG(st.st_mode) or st.st_nlink != 1: raise RetentionError(f"unsafe archive file: {path}") raw = path.read_bytes() after = path.lstat() if (st.st_dev, st.st_ino, st.st_size, st.st_mtime_ns, st.st_ctime_ns) != ( after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns ): raise RetentionError(f"archive changed while reading: {path}") rows.append({ "path": path.relative_to(root).as_posix(), "type": "F", "mode": f"{stat.S_IMODE(st.st_mode):04o}", "size": len(raw), "sha256": sha(raw), }) return rows def inventory_root(rows: list[dict[str, Any]]) -> str: sealed_rows = [ {key: row[key] for key in ("path", "type", "mode", "sha256") if key in row} for row in rows ] return sha(canon(sealed_rows)) def validate_source(source: Path, v3_seal: Path, v4_seal: Path) -> list[dict[str, Any]]: if source.is_symlink() or not source.is_dir(): raise RetentionError("source archive is not a real directory") if file_sha(v3_seal) != EXPECTED["v3_seal_sha256"]: raise RetentionError("v3 seal hash mismatch") if file_sha(v4_seal) != EXPECTED["v4_seal_sha256"]: raise RetentionError("v4 seal hash mismatch") seal = json.loads(v3_seal.read_bytes()) for key in ("archive_root_sha256", "archive_entry_count", "candidate_digest"): if seal.get(key) != EXPECTED[key]: raise RetentionError(f"v3 seal {key} mismatch") rows = source_inventory(source) if len(rows) != EXPECTED["archive_entry_count"] or inventory_root(rows) != EXPECTED["archive_root_sha256"]: raise RetentionError("source archive inventory mismatch") manifest = json.loads((source / "manifest.json").read_bytes()) bindings = manifest.get("bindings", {}) for key in ("candidate_digest", "lifecycle_root_sha256", "lifecycle_seal_sha256"): if bindings.get(key) != EXPECTED[key]: raise RetentionError(f"archive manifest {key} mismatch") if manifest.get("run_id") != EXPECTED["run_id"]: raise RetentionError("archive run_id mismatch") return rows def seal_with_docker(object_dir: Path, image: str) -> None: object_dir = object_dir.resolve() command = [ "/usr/bin/docker", "run", "--rm", "--pull", "never", "--network", "none", "--privileged", "-v", f"{object_dir}:/retention", image, "sh", "-ceu", "find /retention -xdev -type f -exec chmod 0444 {} +; " "find /retention -xdev -depth -type d -exec chmod 0555 {} +; " "chown -R 0:0 /retention; " "find /retention -xdev -type f -exec chattr +i {} +; " "find /retention -xdev -depth -type d -exec chattr +i {} +", ] result = subprocess.run(command, text=True, capture_output=True, check=False) if result.returncode != 0: raise RetentionError(f"immutable seal failed ({result.returncode}): {result.stderr.strip()}") def create(args: argparse.Namespace) -> Path: rows = validate_source(args.source, args.v3_seal, args.v4_seal) objects = args.store / "objects" final = objects / EXPECTED["archive_root_sha256"] if final.exists(): verify(final / "retention-proof.json") return final / "retention-proof.json" objects.mkdir(parents=True, exist_ok=True, mode=0o700) stage = Path(tempfile.mkdtemp(prefix=".pending-retention-", dir=objects)) try: archive = stage / "archive" shutil.copytree(args.source, archive, symlinks=False, copy_function=shutil.copyfile) for row in rows: os.chmod(archive / row["path"], int(row["mode"], 8)) copied = source_inventory(archive) if copied != rows: raise RetentionError("copied archive is not byte/mode exact before sealing") metadata = { "schema": SCHEMA, "status": "RETAIN_UNTIL", "mechanism": MECHANISM, "object_id": EXPECTED["archive_root_sha256"], "source_archive_root_sha256": EXPECTED["archive_root_sha256"], "source_archive_entry_count": EXPECTED["archive_entry_count"], "candidate_digest": EXPECTED["candidate_digest"], "lifecycle_root_sha256": EXPECTED["lifecycle_root_sha256"], "lifecycle_seal_sha256": EXPECTED["lifecycle_seal_sha256"], "v3_seal_sha256": EXPECTED["v3_seal_sha256"], "v4_seal_sha256": EXPECTED["v4_seal_sha256"], "run_id": EXPECTED["run_id"], "retention_started_utc": "2026-08-17T00:00:00Z", "retain_until_utc": "2033-08-17T00:00:00Z", "retention_basis": "strict-rerun-audit-and-incident-reconstruction", "archive_objects": rows, } metadata_raw = canon(metadata) (stage / "retention-metadata.json").write_bytes(metadata_raw) proof = { "schema": "task26-local-immutable-retention-proof-v1", "status": "SEALED", "mechanism": MECHANISM, "object_id": EXPECTED["archive_root_sha256"], "metadata_sha256": sha(metadata_raw), "candidate_digest": EXPECTED["candidate_digest"], "lifecycle_root_sha256": EXPECTED["lifecycle_root_sha256"], "lifecycle_seal_sha256": EXPECTED["lifecycle_seal_sha256"], "retain_until_utc": metadata["retain_until_utc"], } (stage / "retention-proof.json").write_bytes(canon(proof)) os.rename(stage, final) seal_with_docker(final, args.docker_image) verify(final / "retention-proof.json") return final / "retention-proof.json" except Exception: if stage.exists(): shutil.rmtree(stage) raise def verify(proof_path: Path) -> dict[str, Any]: object_dir = proof_path.parent proof = json.loads(proof_path.read_bytes()) metadata_raw = (object_dir / "retention-metadata.json").read_bytes() metadata = json.loads(metadata_raw) expected_proof = { "schema": "task26-local-immutable-retention-proof-v1", "status": "SEALED", "mechanism": MECHANISM, "object_id": EXPECTED["archive_root_sha256"], "metadata_sha256": sha(metadata_raw), "candidate_digest": EXPECTED["candidate_digest"], "lifecycle_root_sha256": EXPECTED["lifecycle_root_sha256"], "lifecycle_seal_sha256": EXPECTED["lifecycle_seal_sha256"], "retain_until_utc": "2033-08-17T00:00:00Z", } if proof != expected_proof: raise RetentionError("retention proof mismatch") if metadata.get("schema") != SCHEMA or metadata.get("mechanism") != MECHANISM: raise RetentionError("retention metadata contract mismatch") for key in ("candidate_digest", "lifecycle_root_sha256", "lifecycle_seal_sha256", "v3_seal_sha256", "v4_seal_sha256", "run_id"): if metadata.get(key) != EXPECTED[key]: raise RetentionError(f"retention metadata {key} mismatch") rows = metadata.get("archive_objects") if not isinstance(rows, list) or len(rows) != EXPECTED["archive_entry_count"] or inventory_root(rows) != EXPECTED["archive_root_sha256"]: raise RetentionError("retention object inventory binding mismatch") expected_paths = {row["path"] for row in rows} actual_paths = {row["path"] for row in source_inventory(object_dir / "archive")} if actual_paths != expected_paths: raise RetentionError("retention object path set mismatch") for row in rows: path = object_dir / "archive" / row["path"] if row["type"] == "F" and (path.stat().st_size != row["size"] or file_sha(path) != row["sha256"]): raise RetentionError(f"retention object hash mismatch: {row['path']}") checked = 0 for base, dirs, files in os.walk(object_dir): for name in dirs + files: path = Path(base) / name st = path.lstat() required_mode = 0o555 if stat.S_ISDIR(st.st_mode) else 0o444 if st.st_uid != 0 or st.st_gid != 0 or stat.S_IMODE(st.st_mode) != required_mode or not immutable(path): raise RetentionError(f"retention prevention control missing: {path}") checked += 1 root_st = object_dir.lstat() if root_st.st_uid != 0 or stat.S_IMODE(root_st.st_mode) != 0o555 or not immutable(object_dir): raise RetentionError("retention object root is not root-owned immutable") return {"schema": "task26-local-immutable-retention-verification-v1", "status": "PASS", "object_id": EXPECTED["archive_root_sha256"], "checked_inodes": checked + 1, "verified_reads": len([r for r in rows if r["type"] == "F"])} def parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser() sub = p.add_subparsers(dest="command", required=True) create_p = sub.add_parser("create") create_p.add_argument("--source", type=Path, required=True) create_p.add_argument("--v3-seal", type=Path, required=True) create_p.add_argument("--v4-seal", type=Path, required=True) create_p.add_argument("--store", type=Path, required=True) create_p.add_argument("--docker-image", default="ubuntu:24.04") verify_p = sub.add_parser("verify") verify_p.add_argument("--proof", type=Path, required=True) return p def main() -> int: args = parser().parse_args() try: if args.command == "create": proof = create(args) result = verify(proof) result["proof"] = str(proof) else: result = verify(args.proof) print(canon(result).decode(), end="") return 0 except (RetentionError, OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as exc: print(f"BLOCKED: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())