#!/usr/bin/env python3 """Finalize outer isolation and cleanup evidence for the clone canary.""" from __future__ import annotations import hashlib import json import os import sys from collections.abc import Mapping from pathlib import Path from typing import Any def _service(raw: str) -> dict[str, str]: return dict(line.split("=", 1) for line in raw.splitlines() if "=" in line) def _table_deltas(before: dict[str, int], after: dict[str, int]) -> dict[str, int]: if set(before) != set(after): raise ValueError( f"live source count keys changed during the canary: before={sorted(before)} after={sorted(after)}" ) return {key: after[key] - before[key] for key in sorted(before)} def _verify_source_binding(root: Path, records: list[dict[str, object]]) -> dict[str, object]: readbacks = [] for record in records: relative_text = str(record.get("repo_relative_path", "")) relative = Path(relative_text) if not relative_text or relative.is_absolute() or ".." in relative.parts: raise ValueError(f"unsafe repo-relative source path: {relative_text!r}") candidate = (root / relative).resolve() try: candidate.relative_to(root) except ValueError as exc: raise ValueError(f"source path escaped repository root: {relative_text}") from exc if not candidate.is_file(): readbacks.append( { "repo_relative_path": relative.as_posix(), "expected_sha256": record.get("sha256"), "actual_sha256": None, "matches": False, } ) continue content = candidate.read_bytes() actual_sha256 = hashlib.sha256(content).hexdigest() readbacks.append( { "repo_relative_path": relative.as_posix(), "expected_sha256": record.get("sha256"), "actual_sha256": actual_sha256, "size_bytes": len(content), "matches": (actual_sha256 == record.get("sha256") and len(content) == record.get("size_bytes")), } ) return { "files": readbacks, "all_match": bool(readbacks) and all(bool(row["matches"]) for row in readbacks), } def _contains_ephemeral_path(value: object) -> bool: if isinstance(value, str): return "/tmp/" in value if isinstance(value, dict): return any(_contains_ephemeral_path(item) for item in value.values()) if isinstance(value, list): return any(_contains_ephemeral_path(item) for item in value) return False def finalize(path: Path, env: Mapping[str, str]) -> dict[str, Any]: if not path.is_file(): raise ValueError("inner canary did not produce a report") data = json.loads(path.read_text(encoding="utf-8")) if not isinstance(data, dict): raise ValueError("inner canary report must be a JSON object") before_counts = json.loads(env["SOURCE_COUNTS_BEFORE"]) after_counts = json.loads(env["SOURCE_COUNTS_AFTER"]) live_table_deltas = _table_deltas(before_counts, after_counts) before_service = _service(env["SERVICE_BEFORE"]) after_service = _service(env["SERVICE_AFTER"]) source_tier = env["SOURCE_TIER"] source_network_mode = env["SOURCE_NETWORK_MODE"] source_canary_label = env["SOURCE_CANARY_LABEL"] container_network_mode = env["CANARY_NETWORK_MODE"] container_mounts = json.loads(env["CANARY_MOUNTS_JSON"]) container_canary_label = env["CANARY_LABEL"] container_instance_label = env["CANARY_INSTANCE_LABEL"] container_volume_mounts = [mount for mount in container_mounts if mount.get("Type") == "volume"] container_isolated = container_network_mode == "none" and not container_volume_mounts container_labeled = container_canary_label == "proposal-apply-lifecycle" and container_instance_label.startswith( "working-leo-approve-claim-" ) source_tier_enforced = source_tier != "isolated-local" or ( source_network_mode == "none" and source_canary_label == "working-leo-approved-source" ) service_observable = before_service.get("Available") == "true" and after_service.get("Available") == "true" service_stable = ( all( before_service.get(key) == after_service.get(key) for key in ("ActiveState", "SubState", "MainPID", "NRestarts") ) if service_observable else None ) service_gate_passes = service_stable is True if source_tier == "live-readonly" else service_stable is not False container_absent = int(env["CONTAINER_NAME_READBACK_RC"]) == 0 and not env["CONTAINER_NAME_READBACK"].strip() label_scoped_orphan_count = int(env["CONTAINER_LABEL_ORPHAN_COUNT"]) label_scoped_container_absent = ( int(env["CONTAINER_LABEL_READBACK_RC"]) == 0 and label_scoped_orphan_count == 0 and not env["CONTAINER_LABEL_READBACK"].strip() ) workdir_absent = env["OUTER_WORKDIR_LEXISTS"] == "false" source_binding = _verify_source_binding( Path(env["REPO_ROOT"]).resolve(), data.get("source_binding", {}).get("files", []), ) data["source_binding"]["independent_post_cleanup_verification"] = source_binding source_binding_ok = ( data["source_binding"].get("unchanged_during_run") is True and source_binding["all_match"] is True ) stale_ephemeral_paths_absent = not _contains_ephemeral_path(data) inner_runtime_pass = data.get("status") == "pass" and int(env["CANARY_RC"]) == 0 outer_ok = ( before_counts == after_counts and all(delta == 0 for delta in live_table_deltas.values()) and service_gate_passes and container_absent and label_scoped_container_absent and workdir_absent and source_binding_ok and stale_ephemeral_paths_absent and container_isolated and container_labeled and source_tier_enforced ) if source_tier == "isolated-local": data["required_tier"] = "T2_runtime" data["required_tiers"] = ["T2_runtime"] data["runtime_scope"] = "isolated_postgres_container_from_readonly_local_fixture" else: data["required_tier"] = "T3_live_readonly" data["required_tiers"] = ["T2_runtime", "T3_live_readonly"] data["runtime_scope"] = "isolated_postgres_container_from_readonly_live_schema" data["outer_isolation"] = { "source_container": env.get("SOURCE_CONTAINER", "teleo-pg"), "source_database": env.get("SOURCE_DB", "teleo"), "source_counts_before": before_counts, "source_counts_after": after_counts, "canonical_table_deltas": live_table_deltas, "source_counts_unchanged": before_counts == after_counts, "service_before": before_service, "service_after": after_service, "service_required": source_tier == "live-readonly", "service_observable": service_observable, "service_stable": service_stable, "service_gate_passes": service_gate_passes, "source_boundary": { "source_tier": source_tier, "network_mode": source_network_mode, "canary_label": source_canary_label, "tier_enforced": source_tier_enforced, }, "container_isolation": { "network_mode": container_network_mode, "volume_mounts": container_volume_mounts, "canary_label": container_canary_label, "instance_label": container_instance_label, "lifecycle_labeled": container_labeled, "network_disabled": container_network_mode == "none", "no_docker_volumes": not container_volume_mounts, }, "cleanup_readback": { "container_remove_returncode": int(env["OUTER_CONTAINER_REMOVE_RC"]), "container_name_query_returncode": int(env["CONTAINER_NAME_READBACK_RC"]), "container_name_query_stdout": env["CONTAINER_NAME_READBACK"], "temporary_container_absent": container_absent, "label_scoped_query_returncode": int(env["CONTAINER_LABEL_READBACK_RC"]), "label_scoped_query_stdout": env["CONTAINER_LABEL_READBACK"], "label_scoped_orphan_count": label_scoped_orphan_count, "label_scoped_temporary_container_absent": label_scoped_container_absent, "workdir_remove_returncode": int(env["OUTER_WORKDIR_REMOVE_RC"]), "workdir_lexists_after_cleanup": not workdir_absent, "temporary_workdir_absent": workdir_absent, }, } checks = data["checks"] checks["outer_live_counts_exactly_unchanged"] = before_counts == after_counts checks["outer_live_table_deltas_exactly_zero"] = all(delta == 0 for delta in live_table_deltas.values()) checks["outer_service_observed_if_required"] = source_tier != "live-readonly" or service_observable checks["outer_service_stable_if_observed"] = service_gate_passes checks["outer_container_absent_independent_readback"] = container_absent checks["outer_labeled_container_absent_independent_readback"] = label_scoped_container_absent checks["outer_workdir_absent_independent_readback"] = workdir_absent checks["source_binding_independently_verified"] = source_binding_ok checks["stale_ephemeral_paths_absent"] = stale_ephemeral_paths_absent checks["outer_source_tier_enforced"] = source_tier_enforced checks["outer_container_network_disabled"] = container_network_mode == "none" checks["outer_container_lifecycle_labeled"] = container_labeled checks["outer_container_has_no_docker_volumes"] = not container_volume_mounts checks["outer_isolation_complete"] = outer_ok if outer_ok and inner_runtime_pass: data["current_tier"] = "T2_runtime" if source_tier == "isolated-local" else "T3_live_readonly" else: data["status"] = "fail" data["current_tier"] = "T2_runtime" if inner_runtime_pass else "T1_model" path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8") return data def main(argv: list[str] | None = None) -> int: arguments = sys.argv[1:] if argv is None else argv if len(arguments) != 1: raise SystemExit("usage: finalize_approve_claim_isolated_canary.py REPORT.json") try: data = finalize(Path(arguments[0]), os.environ) except (KeyError, OSError, ValueError, json.JSONDecodeError) as exc: raise SystemExit(str(exc)) from exc return 0 if data["status"] == "pass" else 1 if __name__ == "__main__": raise SystemExit(main())