#!/usr/bin/env python3 """Seal the test-only Task26 inode-replacement successor.""" from __future__ import annotations import hashlib import json import os import shutil import stat import subprocess import sys import tempfile import zipfile from pathlib import Path PROJECT = Path("/home/cube/projects/richard/traning coach") OUTPUT = PROJECT / ".omo/evidence/task26" WORKSPACE = Path("/home/cube/.hermes/profiles/dualcoachtest/workspace/checkin_cli") BASELINE = OUTPUT / "task26-owner-customer-v1-candidate-v3/snapshot/profile-package" PREDECESSOR_DIGEST = "4a8627af8f9b931d537e17dbd3e357b7a0f2b2abffdbf28f6ad19a8d32893ec0" PREDECESSOR = OUTPUT / f"task26-token-rotation-successor-{PREDECESSOR_DIGEST}" WHEEL_SHA256 = "f75856d6d986b64d3d2f083aec2f865c7aea19f5950b84ff06e519f2f6505af6" TEST_PATH = "tests/test_adaptive_nutrition.py" OLD_TEST_SHA256 = "9d323da1fb01e3b05df6557795d87b6e26b8d0e7c8ca83f03b3287a237929369" NEW_TEST_SHA256 = "0831230dbebb27303a55d9ac2cb349adb217cd639d1db2981eab0974db527354" PREDECESSOR_DELTA = { "checkin_cli/activation_token_rotation_policy.py", "checkin_cli/customer_admin.py", "tests/test_customer_admin.py", } EXPECTED_TOTAL_DELTA = PREDECESSOR_DELTA | {TEST_PATH} IGNORED = {".ruff_cache", ".pytest_cache", ".venv", "__pycache__", "build", "dist"} VERIFIER = Path(__file__).with_name("verify_candidate.py") def canonical(value: object) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def json_bytes(value: object) -> bytes: return canonical(value) + b"\n" def sha256(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def sha256_file(path: Path) -> str: return sha256(path.read_bytes()) def write(path: Path, raw: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(raw) path.chmod(0o400) def run(command: list[str], *, cwd: Path, env: dict[str, str] | None = None, check: bool = True) -> subprocess.CompletedProcess[bytes]: result = subprocess.run(command, cwd=cwd, env=env, capture_output=True, timeout=1200) if check and result.returncode: raise AssertionError((result.stdout + result.stderr).decode("utf-8", "replace")[-12000:]) return result def run_umask(mask: str, command: list[str], *, cwd: Path, env: dict[str, str]) -> subprocess.CompletedProcess[bytes]: return run(["bash", "-c", f"umask {mask}; exec \"$@\"", "umask-run", *command], cwd=cwd, env=env) def source_inventory(root: Path) -> dict[str, str]: result: dict[str, str] = {} for path in sorted(root.rglob("*")): relative = path.relative_to(root) if not path.is_file() or path.is_symlink() or any(part in IGNORED for part in relative.parts): continue if path.name.endswith((".pyc", ".pyo", ".egg-info")): continue result[relative.as_posix()] = sha256_file(path) return result def clean_copy(source: Path, destination: Path) -> None: def ignore(_directory: str, names: list[str]) -> set[str]: return {name for name in names if name in IGNORED or name.endswith((".pyc", ".pyo", ".egg-info"))} shutil.copytree(source, destination, symlinks=False, ignore=ignore) destination.chmod(0o700) for path in destination.rglob("*"): path.chmod(0o700 if path.is_dir() else 0o600) def copy_sealed_tree(source: Path, destination: Path) -> None: shutil.copytree(source, destination, symlinks=False) for path in destination.rglob("*"): path.chmod(0o500 if path.is_dir() else 0o400) destination.chmod(0o500) def tree_inventory(root: Path) -> dict[str, tuple[str, int, int]]: return { path.relative_to(root).as_posix(): (sha256_file(path), path.stat().st_size, stat.S_IMODE(path.stat().st_mode)) for path in root.rglob("*") if path.is_file() } def build_wheel(source: Path, output: Path, mask: str, env: dict[str, str]) -> tuple[Path, bytes]: result = run_umask( mask, ["uv", "build", "--wheel", "--offline", "--no-build-isolation", "--out-dir", str(output), str(source)], cwd=output.parent, env={**env, "SOURCE_DATE_EPOCH": "946684800", "TZ": "UTC", "UV_OFFLINE": "1"}, ) wheels = list(output.glob("*.whl")) assert len(wheels) == 1 return wheels[0], result.stdout + result.stderr def main() -> int: assert PREDECESSOR.is_dir() and not PREDECESSOR.is_symlink() predecessor_manifest = json.loads((PREDECESSOR / "candidate-manifest.json").read_text()) assert predecessor_manifest["full_candidate_digest"] == PREDECESSOR_DIGEST predecessor_wheel = PREDECESSOR / "artifacts/physique_checkin_cli-0.1.0-py3-none-any.whl" assert sha256_file(predecessor_wheel) == WHEEL_SHA256 run([sys.executable, "-B", str(PREDECESSOR / "verify_candidate.py"), str(PREDECESSOR)], cwd=PROJECT) assert sha256_file(BASELINE / TEST_PATH) == OLD_TEST_SHA256 assert sha256_file(WORKSPACE / TEST_PATH) == NEW_TEST_SHA256 live_before = source_inventory(WORKSPACE) baseline = source_inventory(BASELINE) total_delta = {path for path in set(baseline) | set(live_before) if baseline.get(path) != live_before.get(path)} assert total_delta == EXPECTED_TOTAL_DELTA predecessor_delta = json.loads((PREDECESSOR / "bindings/source-delta.json").read_text()) assert {row["path"] for row in predecessor_delta["files"]} == PREDECESSOR_DELTA for row in predecessor_delta["files"]: assert live_before[row["path"]] == row["after_sha256"] test_text = (WORKSPACE / TEST_PATH).read_text() assert "path.chmod(0o600)" in test_text and "replacement.chmod(0o600)" in test_text assert 'assert path.read_bytes() == b"replacement"' in test_text assert "os.read(original_fd, len(complete) + 8) == complete + b'{\"torn\":'" in test_text with tempfile.TemporaryDirectory(prefix=".task26-inode-test-fix-", dir=OUTPUT) as temporary_name: temporary = Path(temporary_name) suite, source_a, source_b = (temporary / name for name in ("suite", "source-a", "source-b")) clean_copy(WORKSPACE, suite) clean_copy(WORKSPACE, source_a) clean_copy(WORKSPACE, source_b) env = dict(os.environ) env.update({"PYTHONDONTWRITEBYTECODE": "1", "PYTHONPYCACHEPREFIX": str(temporary / "pycache")}) focus_target = f"{TEST_PATH}::test_overlay_recovery_rejects_symlink_and_data_inode_replacement" focus_077 = run_umask("077", [sys.executable, "-B", "-m", "pytest", "-p", "no:cacheprovider", focus_target, "-q"], cwd=suite, env=env) focus_022 = run_umask("022", [sys.executable, "-B", "-m", "pytest", "-p", "no:cacheprovider", focus_target, "-q"], cwd=suite, env=env) assert b"1 passed" in focus_077.stdout and b"1 passed" in focus_022.stdout full = run_umask("077", [sys.executable, "-B", "-m", "pytest", "-p", "no:cacheprovider", "-q"], cwd=suite, env=env) assert b"679 passed" in full.stdout compile_result = run_umask("077", [sys.executable, "-B", "-m", "compileall", "-q", "-f", TEST_PATH], cwd=suite, env=env) ruff = run_umask( "077", ["uvx", "--offline", "ruff", "check", "--isolated", "--no-cache", "--select", "E9,F63,F7,F82", TEST_PATH], cwd=suite, env=env, ) wheel_a, build_a = build_wheel(source_a, temporary / "wheel-a", "002", env) wheel_b, build_b = build_wheel(source_b, temporary / "wheel-b", "002", env) assert wheel_a.read_bytes() == wheel_b.read_bytes() == predecessor_wheel.read_bytes() assert sha256_file(wheel_a) == sha256_file(wheel_b) == WHEEL_SHA256 predecessor_source = PREDECESSOR / "snapshot/package-runtime" with zipfile.ZipFile(wheel_a) as archive: infos = archive.infolist() assert len(infos) == 50 and {item.date_time for item in infos} == {(2000, 1, 1, 0, 0, 0)} package_members = sorted(item.filename for item in infos if item.filename.startswith("checkin_cli/")) source_members = sorted( path.relative_to(source_a).as_posix() for path in (source_a / "checkin_cli").rglob("*") if path.is_file() ) assert package_members == source_members and len(package_members) == 46 for member in package_members: assert archive.read(member) == (source_a / member).read_bytes() == (predecessor_source / member).read_bytes() assert not any(item.filename.startswith("tests/") for item in infos) assert source_inventory(WORKSPACE) == live_before staging = temporary / "candidate" for relative in ("artifacts", "bindings", "test-delta/before/tests", "test-delta/after/tests", "snapshot/tests", "historical"): (staging / relative).mkdir(parents=True, exist_ok=True) copy_sealed_tree(PREDECESSOR, staging / "historical/predecessor-candidate") original_tree = tree_inventory(PREDECESSOR) copied_tree = tree_inventory(staging / "historical/predecessor-candidate") assert original_tree == copied_tree write(staging / "test-delta/before" / TEST_PATH, (BASELINE / TEST_PATH).read_bytes()) write(staging / "test-delta/after" / TEST_PATH, (WORKSPACE / TEST_PATH).read_bytes()) write(staging / "snapshot/tests/test_adaptive_nutrition.py", (WORKSPACE / TEST_PATH).read_bytes()) write(staging / "artifacts/physique_checkin_cli-0.1.0-py3-none-any.whl", wheel_a.read_bytes()) write(staging / "artifacts/focused-overlay-umask-077.txt", focus_077.stdout + focus_077.stderr) write(staging / "artifacts/focused-overlay-umask-022.txt", focus_022.stdout + focus_022.stderr) write(staging / "artifacts/full-package-umask-077.txt", full.stdout + full.stderr) write(staging / "artifacts/compileall.txt", compile_result.stdout + compile_result.stderr + b"PASS\n") write(staging / "artifacts/ruff-critical.txt", ruff.stdout + ruff.stderr) write(staging / "artifacts/wheel-build-1-umask-002.txt", build_a) write(staging / "artifacts/wheel-build-2-umask-002.txt", build_b) write(staging / "verify_candidate.py", VERIFIER.read_bytes()) delta_files = [{ "path": TEST_PATH, "change": "modified", "before_sha256": OLD_TEST_SHA256, "after_sha256": NEW_TEST_SHA256, }] delta = { "schema": "task26-inode-test-fix-delta-v1", "predecessor_full_digest": PREDECESSOR_DIGEST, "files": delta_files, "file_count": 1, } write(staging / "bindings/test-delta.json", json_bytes(delta)) quality = { "schema": "task26-inode-test-fix-quality-v1", "status": "PASS", "focused_overlay_umask_077": {"passed": 1, "failed": 0}, "focused_overlay_umask_022": {"passed": 1, "failed": 0}, "full_package_umask_077": {"passed": 679, "failed": 0}, "strict_umask": "077", "writable_copy_corrected": True, "fixture_mode_assertions": True, "named_replacement_bytes_assertion": True, "held_fd_original_bytes_assertion": True, "compileall": "PASS", "ruff_critical_errors": 0, "reproducible_wheel_builds": 2, "wheel_build_umasks": ["002", "002"], "wheel_builds_byte_identical": True, "wheel_unchanged_from_predecessor": True, "wheel_source_parity": True, "wheel_contains_tests": False, "predecessor_byte_and_mode_drift": 0, "unindexed_count": 0, "deployment_actions": 0, "live_authority_mutations": 0, "network_actions": 0, "git_actions": 0, } write(staging / "artifacts/quality-summary.json", json_bytes(quality)) indexed_paths = sorted(path.relative_to(staging).as_posix() for path in staging.rglob("*") if path.is_file()) entries = [ {"path": relative, "bytes": (staging / relative).stat().st_size, "sha256": sha256_file(staging / relative)} for relative in indexed_paths ] inventory = { "schema": "task26-inode-test-fix-hash-inventory-v1", "entries": entries, "entry_count": len(entries), "entries_sha256": sha256(canonical(entries)), } write(staging / "hash-inventory.json", json_bytes(inventory)) core_material = { "schema": "task26-inode-test-fix-core-v1", "predecessor_full_digest": PREDECESSOR_DIGEST, "profile_wheel_sha256": WHEEL_SHA256, "test_delta": delta_files, } manifest_core: dict[str, object] = { "schema": "task26-inode-test-fix-successor-candidate-v1", "state": "PASS_SEALED_UNDEPLOYED", "authority": "IMMUTABLE_TEST_FIX_SUCCESSOR_NOT_DEPLOYED", "predecessor": { "full_digest": PREDECESSOR_DIGEST, "manifest_sha256": sha256_file(PREDECESSOR / "candidate-manifest.json"), "file_count": len(original_tree), "preserved_path": "historical/predecessor-candidate", "byte_and_mode_drift": 0, "verified": True, }, "core_material": core_material, "core_candidate_digest": sha256(canonical(core_material)), "profile_wheel": { "path": "artifacts/physique_checkin_cli-0.1.0-py3-none-any.whl", "sha256": WHEEL_SHA256, "bytes": wheel_a.stat().st_size, "member_count": 50, "package_member_count": 46, "tests_packaged": False, "reproducible_rounds": 2, "byte_identical_to_predecessor": True, }, "test_delta_path": "bindings/test-delta.json", "test_delta_sha256": sha256_file(staging / "bindings/test-delta.json"), "quality_summary_path": "artifacts/quality-summary.json", "quality_summary_sha256": sha256_file(staging / "artifacts/quality-summary.json"), "inventory_sha256": sha256_file(staging / "hash-inventory.json"), "inventory_count": len(entries), "verifier_sha256": sha256_file(staging / "verify_candidate.py"), "deployment": "NOT_PERFORMED", } full_digest = sha256(canonical(manifest_core)) manifest = {**manifest_core, "full_candidate_digest": full_digest} write(staging / "candidate-manifest.json", json_bytes(manifest)) seal = { "schema": "task26-inode-test-fix-successor-seal-v1", "status": "PASS_SEALED_UNDEPLOYED", "full_candidate_digest": full_digest, "core_candidate_digest": manifest["core_candidate_digest"], "manifest_sha256": sha256_file(staging / "candidate-manifest.json"), "inventory_sha256": manifest["inventory_sha256"], "wheel_sha256": WHEEL_SHA256, "predecessor_full_digest": PREDECESSOR_DIGEST, "verifier_sha256": manifest["verifier_sha256"], } write(staging / "candidate-seal.json", json_bytes(seal)) verifier_input = { "schema": "task26-inode-test-fix-independent-verifier-input-v1", "full_candidate_digest": full_digest, "manifest_sha256": seal["manifest_sha256"], "seal_sha256": sha256_file(staging / "candidate-seal.json"), } write(staging / "verifier-input.json", json_bytes(verifier_input)) for directory in sorted((path for path in staging.rglob("*") if path.is_dir()), key=lambda path: len(path.parts), reverse=True): directory.chmod(0o500) staging.chmod(0o700) final = OUTPUT / f"task26-inode-test-fix-successor-{full_digest}" assert not final.exists() staging.rename(final) final.chmod(0o500) first = run([sys.executable, "-B", str(final / "verify_candidate.py"), str(final)], cwd=temporary) verification = json.loads(first.stdout) receipt = { "schema": "task26-inode-test-fix-independent-verification-receipt-v1", "status": "PASS", "state": "PASS_SEALED_UNDEPLOYED", "full_candidate_digest": full_digest, "seal_sha256": sha256_file(final / "candidate-seal.json"), "verifier_sha256": sha256_file(final / "verify_candidate.py"), "verification": verification, "mutations": {"deployment": 0, "live_authority": 0, "service": 0, "network": 0, "git": 0, "plan": 0, "todo": 0, "ledger": 0}, } final.chmod(0o700) write(final / "independent-verification.json", json_bytes(receipt)) final.chmod(0o500) second = run([sys.executable, "-B", str(final / "verify_candidate.py"), str(final)], cwd=temporary) print(json.dumps({ **json.loads(second.stdout), "candidate_root": str(final), "manifest_sha256": sha256_file(final / "candidate-manifest.json"), "seal_sha256": sha256_file(final / "candidate-seal.json"), "inventory_sha256": sha256_file(final / "hash-inventory.json"), "quality_sha256": sha256_file(final / "artifacts/quality-summary.json"), "verification_sha256": sha256_file(final / "independent-verification.json"), }, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())