#!/usr/bin/env python3 """Materialize and seal a postfreeze Task26 delivery without rerunning qualification.""" from __future__ import annotations import hashlib import json import os import shutil import stat import sys from pathlib import Path from typing import cast POSTFREEZE_SEAL = "postfreeze-seal.json" DELIVERED_SEAL = "delivered-bundle-seal.json" RECEIPT_PASS = "PASS.json" RECEIPT_SEAL = "SEAL.json" def canonical(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _relative_sibling(source: Path, target: Path) -> str: relative = os.path.relpath(target, source) if Path(relative).is_absolute(): raise ValueError("sealed root binding must be relative") return Path(relative).as_posix() def _private(path: Path, *, directory: bool = False) -> None: info = path.lstat() expected = stat.S_ISDIR if directory else stat.S_ISREG modes = {0o500, 0o700} if directory else {0o400, 0o600} if ( stat.S_ISLNK(info.st_mode) or not expected(info.st_mode) or info.st_uid != os.geteuid() or (not directory and info.st_nlink != 1) or stat.S_IMODE(info.st_mode) not in modes ): raise ValueError(f"unsafe sealed path: {path}") def inventory(root: Path, *, excluded: frozenset[str] = frozenset()) -> list[dict[str, object]]: _private(root, directory=True) rows: list[dict[str, object]] = [] for path in sorted(root.rglob("*")): relative = path.relative_to(root).as_posix() if relative in excluded: continue if path.is_symlink(): raise ValueError("sealed root contains a symlink") _private(path, directory=path.is_dir()) row: dict[str, object] = { "relative_path": relative, "kind": "directory" if path.is_dir() else "file", "mode": stat.S_IMODE(path.stat().st_mode), } if path.is_file(): row.update(sha256=sha256_file(path), size=path.stat().st_size) rows.append(row) return rows def file_inventory(root: Path, *, excluded: frozenset[str]) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] for path in sorted(root.rglob("*")): relative = path.relative_to(root).as_posix() if relative in excluded or not path.is_file(): continue _private(path) rows.append({ "path": relative, "sha256": sha256_file(path), "size": path.stat().st_size, }) return rows def _object(value: object, label: str) -> dict[str, object]: if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): raise ValueError(f"{label} is invalid") return cast(dict[str, object], value) def _write(path: Path, document: object) -> bytes: data = canonical(document) + b"\n" path.write_bytes(data) path.chmod(0o400) return data def materialize_delivered_bundle( prefreeze_root: Path, receipt_root: Path, delivered_root: Path, ) -> dict[str, object]: """Copy prefreeze bytes, preserve the exact postfreeze seal, and seal both roots.""" prefreeze_root = prefreeze_root.resolve(strict=True) receipt_root = receipt_root.resolve(strict=True) delivered_root = delivered_root.absolute() _private(prefreeze_root, directory=True) _private(receipt_root, directory=True) if delivered_root.exists(): raise ValueError("delivered root already exists") postfreeze_path = receipt_root / POSTFREEZE_SEAL pass_path = receipt_root / RECEIPT_PASS _private(postfreeze_path) _private(pass_path) postfreeze_bytes = postfreeze_path.read_bytes() postfreeze = _object(json.loads(postfreeze_bytes), "postfreeze seal") postfreeze_unsigned = { key: value for key, value in postfreeze.items() if key != "seal_sha256" } if ( postfreeze.get("schema") != "task26-postfreeze-seal-v1" or postfreeze.get("seal_sha256") != hashlib.sha256(canonical(postfreeze_unsigned)).hexdigest() ): raise ValueError("postfreeze seal hash differs") receipt_root.chmod(0o700) shutil.copytree(prefreeze_root, delivered_root, copy_function=shutil.copy2) delivered_root.chmod(0o700) try: before = inventory(delivered_root) if ( postfreeze.get("inventory") != before or postfreeze.get("inventory_sha256") != hashlib.sha256(canonical(before)).hexdigest() ): raise ValueError("postfreeze inventory differs from prefreeze root") delivered_postfreeze = delivered_root / POSTFREEZE_SEAL delivered_postfreeze.write_bytes(postfreeze_bytes) delivered_postfreeze.chmod(0o400) final_inventory = inventory( delivered_root, excluded=frozenset({DELIVERED_SEAL}) ) final_seal: dict[str, object] = { "schema": "task26-delivered-bundle-seal-v1", "candidate_digest": postfreeze.get("candidate_digest"), "prefreeze_seal_sha256": sha256_file(prefreeze_root / "SEAL.json"), "postfreeze_seal_sha256": hashlib.sha256(postfreeze_bytes).hexdigest(), "permitted_postfreeze_additions": [POSTFREEZE_SEAL, DELIVERED_SEAL], "receipt_root_relative_path": _relative_sibling( delivered_root, receipt_root ), "inventory": final_inventory, "inventory_sha256": hashlib.sha256( canonical(final_inventory) ).hexdigest(), } final_seal["seal_sha256"] = hashlib.sha256(canonical(final_seal)).hexdigest() final_bytes = _write(delivered_root / DELIVERED_SEAL, final_seal) pass_path.chmod(0o600) receipt_pass = _object(json.loads(pass_path.read_bytes()), "PASS receipt") receipt_pass["postfreeze_seal"] = { "receipt_relative_path": POSTFREEZE_SEAL, "delivered_relative_path": POSTFREEZE_SEAL, "sha256": hashlib.sha256(postfreeze_bytes).hexdigest(), } receipt_pass["delivered_bundle"] = { "relative_path": _relative_sibling(receipt_root, delivered_root), "seal_relative_path": DELIVERED_SEAL, "seal_sha256": hashlib.sha256(final_bytes).hexdigest(), } receipt_pass.pop("receipt_sha256", None) receipt_pass["receipt_sha256"] = hashlib.sha256( canonical(receipt_pass) ).hexdigest() _write(pass_path, receipt_pass) old_receipt_seal = receipt_root / RECEIPT_SEAL if old_receipt_seal.exists(): old_receipt_seal.chmod(0o600) old_receipt_seal.unlink() receipt_files = file_inventory( receipt_root, excluded=frozenset({RECEIPT_SEAL}) ) receipt_seal: dict[str, object] = { "schema": "task26-postfreeze-receipt-seal-v2", "candidate_digest": postfreeze.get("candidate_digest"), "status": "PASS", "files": receipt_files, "file_count": len(receipt_files), "inventory_sha256": hashlib.sha256( canonical(receipt_files) ).hexdigest(), } receipt_seal["document_sha256"] = hashlib.sha256( canonical(receipt_seal) ).hexdigest() _write(old_receipt_seal, receipt_seal) for path in sorted(delivered_root.rglob("*"), reverse=True): path.chmod(0o500 if path.is_dir() else 0o400) delivered_root.chmod(0o500) for path in sorted(receipt_root.rglob("*"), reverse=True): path.chmod(0o500 if path.is_dir() else 0o400) receipt_root.chmod(0o500) return final_seal except BaseException: if delivered_root.exists(): for path in delivered_root.rglob("*"): if not path.is_symlink(): path.chmod(0o700 if path.is_dir() else 0o600) shutil.rmtree(delivered_root) raise def main() -> int: if len(sys.argv) != 4 or not sys.flags.isolated: print("usage: python -I task26_seal_delivered_bundle.py PREFREEZE RECEIPTS DELIVERED", file=sys.stderr) return 2 try: seal = materialize_delivered_bundle( Path(sys.argv[1]), Path(sys.argv[2]), Path(sys.argv[3]) ) print(json.dumps(seal, sort_keys=True, separators=(",", ":"))) return 0 except Exception as exc: print(json.dumps({"status": "TASK26_DELIVERY_SEAL_FAIL", "reason": str(exc)}, sort_keys=True), file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())