#!/usr/bin/env python3 """Executable fail-closed and rollback regressions for tools/cutover.sh.""" from __future__ import annotations import hashlib import os import shutil import subprocess import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] CUTOVER = ROOT / "tools" / "cutover.sh" def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() class CutoverSafetyTests(unittest.TestCase): def setUp(self) -> None: self.tmp = Path(tempfile.mkdtemp(prefix="cosmos-cutover-test-")) self.addCleanup(self._cleanup) self.content = self.tmp / "content" self.live = self.content / "themes" / "clean" self.routes = self.content / "settings" / "routes.yaml" self.live.mkdir(parents=True) self.routes.parent.mkdir(parents=True) (self.live / "version.txt").write_text("old-theme\n", encoding="utf-8") self.routes.write_text("old-routes\n", encoding="utf-8") self.release = self.tmp / "release" candidate = self.release / "candidate" (candidate / "theme").mkdir(parents=True) (candidate / "theme" / "version.txt").write_text("new-theme\n", encoding="utf-8") (candidate / "routes.yaml").write_text("new-routes\n", encoding="utf-8") manifest_lines = [] for path in sorted(candidate.rglob("*")): if path.is_file(): manifest_lines.append(f"{sha256(path)} {path.relative_to(self.release)}") manifest = self.release / "PAYLOAD_SHA256SUMS" manifest.write_text("\n".join(manifest_lines) + "\n", encoding="utf-8") (self.release / "PAYLOAD_MANIFEST_SHA256").write_text( f"{sha256(manifest)} PAYLOAD_SHA256SUMS\n", encoding="utf-8" ) self.manifest_sha = sha256(manifest) self.seller = self.tmp / "seller.attestation" self.mx = self.tmp / "email-mx.attestation" self._write_attestation(self.seller) self._write_attestation(self.mx) for path in self.release.rglob("*"): path.chmod(0o555 if path.is_dir() else 0o444) self.release.chmod(0o555) def _cleanup(self) -> None: if self.tmp.exists(): for path in self.tmp.rglob("*"): if path.is_dir(): path.chmod(0o755) else: path.chmod(0o644) shutil.rmtree(self.tmp, ignore_errors=True) def _write_attestation(self, path: Path, *, status: str = "APPROVED", digest: str | None = None) -> None: path.write_text( f"status={status}\nrelease_manifest_sha256={digest or self.manifest_sha}\n", encoding="utf-8", ) path.chmod(0o600) def run_cutover(self, *, failpoint: str = "", seller: Path | None = None, mx: Path | None = None) -> subprocess.CompletedProcess[str]: env = os.environ.copy() env.update( { "CUTOVER_TEST_MODE": "1", "CUTOVER_TEST_CONTENT": str(self.content), "RELEASE": str(self.release), "STAMP": "exact-v2", "SELLER_EVIDENCE_FILE": str(seller or self.seller), "EMAIL_MX_EVIDENCE_FILE": str(mx or self.mx), } ) if failpoint: env["CUTOVER_FAILPOINT"] = failpoint return subprocess.run( ["bash", str(CUTOVER)], env=env, text=True, capture_output=True, timeout=20 ) def assert_live_old(self) -> None: self.assertEqual((self.live / "version.txt").read_text(), "old-theme\n") self.assertEqual(self.routes.read_text(), "old-routes\n") def test_missing_or_unapproved_external_gates_fail_before_mutation(self) -> None: missing = self.tmp / "missing.attestation" result = self.run_cutover(seller=missing) self.assertNotEqual(result.returncode, 0) self.assert_live_old() self._write_attestation(self.seller, status="REJECTED") result = self.run_cutover() self.assertNotEqual(result.returncode, 0) self.assert_live_old() def test_attestations_are_bound_to_exact_manifest(self) -> None: self._write_attestation(self.mx, digest="0" * 64) result = self.run_cutover() self.assertNotEqual(result.returncode, 0) self.assert_live_old() def test_manifest_tamper_fails_before_mutation(self) -> None: target = self.release / "candidate" / "theme" / "version.txt" target.chmod(0o644) target.write_text("tampered\n") result = self.run_cutover() self.assertNotEqual(result.returncode, 0) self.assert_live_old() def test_writable_release_fails_before_mutation(self) -> None: target = self.release / "candidate" / "routes.yaml" target.chmod(0o644) result = self.run_cutover() self.assertNotEqual(result.returncode, 0) self.assertIn("release is not immutable", result.stderr) self.assert_live_old() def test_preexisting_stage_fails_closed_without_deleting_it(self) -> None: stage = self.content / "themes" / ".clean-stage-exact-v2" stage.mkdir() marker = stage / "do-not-delete" marker.write_text("preserve\n") result = self.run_cutover() self.assertNotEqual(result.returncode, 0) self.assertEqual(marker.read_text(), "preserve\n") self.assert_live_old() def test_every_transaction_failpoint_restores_old_pair(self) -> None: for failpoint in ("after_live_backup", "after_routes_backup", "restart", "healthcheck", "after_restart"): with self.subTest(failpoint=failpoint): shutil.rmtree(self.content) self.live.mkdir(parents=True) self.routes.parent.mkdir(parents=True) (self.live / "version.txt").write_text("old-theme\n") self.routes.write_text("old-routes\n") result = self.run_cutover(failpoint=failpoint) self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr) self.assert_live_old() self.assertIn("ROLLBACK=VERIFIED", result.stderr) def test_success_installs_exact_candidate_and_preserves_rollback_pair(self) -> None: result = self.run_cutover() self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual((self.live / "version.txt").read_text(), "new-theme\n") self.assertEqual(self.routes.read_text(), "new-routes\n") self.assertEqual( (self.content / "themes" / ".clean-before-exact-v2" / "version.txt").read_text(), "old-theme\n", ) self.assertEqual( (self.content / "settings" / ".routes-exact-v2.before").read_text(), "old-routes\n", ) self.assertIn("CUTOVER=PASS", result.stdout) if __name__ == "__main__": unittest.main(verbosity=2)