from __future__ import annotations import copy import json import os import stat import subprocess import sys import tarfile import tempfile import unittest import zipfile from pathlib import Path from unittest import mock from common import ROOT as PACKAGE_ROOT from scripts import build_release from scripts.verify_release import _strict_json_loads as release_strict_json_loads VERSION = (PACKAGE_ROOT / "VERSION").read_text(encoding="utf-8").strip() PACKAGE_NAME = f"codex-multimodel-orchestrator-{VERSION}" class ReleaseIntegrityTests(unittest.TestCase): def test_release_manifest_json_rejects_ambiguous_members_and_invalid_unicode(self) -> None: for payload in ( '{"schema_version": 2, "schema_version": 1}', '{"path": "\\ud800"}', '{"size": 1e400}', ): with self.subTest(payload=payload), self.assertRaises(ValueError): release_strict_json_loads(payload) def test_archive_set_publication_rolls_back_as_a_unit(self) -> None: with tempfile.TemporaryDirectory(prefix="mmo-release-publish-") as temporary_raw: root = Path(temporary_raw) source_paths = tuple(root / f"source-{index}" for index in range(4)) final_paths = tuple(root / f"final-{index}" for index in range(4)) for index, path in enumerate(source_paths): path.write_bytes(f"new {index}".encode()) for index, path in enumerate(final_paths): path.write_bytes(f"old {index}".encode()) real_replace = os.replace def fail_metadata_publish( source: str | os.PathLike[str], destination: str | os.PathLike[str] ) -> None: source_path = Path(source) destination_path = Path(destination) if destination_path == final_paths[2] and ".publish-" in source_path.name: raise OSError("synthetic metadata publication failure") real_replace(source, destination) with mock.patch.object(build_release.os, "replace", side_effect=fail_metadata_publish): with self.assertRaisesRegex(OSError, "metadata publication"): build_release._publish_artifacts( tuple(zip(source_paths, final_paths, strict=True)) ) for index, path in enumerate(final_paths): self.assertEqual(path.read_bytes(), f"old {index}".encode()) self.assertFalse(any(path.name.startswith(".final") for path in root.iterdir())) symlink_target = root / "symlink-target" symlink_target.write_bytes(b"must remain untouched") symlink_destination = root / "symlink-destination" symlink_destination.symlink_to(symlink_target) with self.assertRaisesRegex(RuntimeError, "non-regular release artifact"): build_release._publish_artifacts(((source_paths[0], symlink_destination),)) self.assertTrue(symlink_destination.is_symlink()) self.assertEqual(symlink_target.read_bytes(), b"must remain untouched") def interrupt_final_verification() -> None: raise KeyboardInterrupt with self.assertRaises(KeyboardInterrupt): build_release._publish_artifacts( tuple(zip(source_paths, final_paths, strict=True)), verify=interrupt_final_verification, ) for index, path in enumerate(final_paths): self.assertEqual(path.read_bytes(), f"old {index}".encode()) def fail_publish_and_restore( source: str | os.PathLike[str], destination: str | os.PathLike[str] ) -> None: source_path = Path(source) destination_path = Path(destination) if destination_path == final_paths[2] and ".publish-" in source_path.name: raise OSError("synthetic publication failure") if destination_path == final_paths[0] and ".backup-" in source_path.name: raise OSError("synthetic restoration failure") real_replace(source, destination) with mock.patch.object( build_release.os, "replace", side_effect=fail_publish_and_restore, ): with self.assertRaisesRegex(RuntimeError, "rollback was incomplete"): build_release._publish_artifacts( tuple(zip(source_paths, final_paths, strict=True)) ) retained = list(root.glob(".final-0.backup-*")) self.assertEqual(len(retained), 1) self.assertEqual(retained[0].read_bytes(), b"old 0") final_paths[0].write_bytes(retained[0].read_bytes()) retained[0].unlink() def fail_final_verification() -> None: raise RuntimeError("synthetic final verification failure") with self.assertRaisesRegex(RuntimeError, "final verification"): build_release._publish_artifacts( tuple(zip(source_paths, final_paths, strict=True)), verify=fail_final_verification, ) for index, path in enumerate(final_paths): self.assertEqual(path.read_bytes(), f"old {index}".encode()) self.assertFalse(any(path.name.startswith(".final") for path in root.iterdir())) def test_release_builder_publishes_complete_verified_archives(self) -> None: with tempfile.TemporaryDirectory(prefix="mmo-release-test-") as temporary_raw: output = Path(temporary_raw) result = subprocess.run( [ sys.executable, str(PACKAGE_ROOT / "scripts" / "build_release.py"), "--output-dir", str(output), "--skip-validation", "--no-reproducibility-check", ], cwd=PACKAGE_ROOT, text=True, capture_output=True, check=False, timeout=120, ) self.assertEqual(result.returncode, 0, result.stderr or result.stdout) tar_path = output / f"{PACKAGE_NAME}-linux.tar.gz" zip_path = output / f"{PACKAGE_NAME}-linux.zip" integrity_path = output / f"{PACKAGE_NAME}-INTEGRITY.json" for path in (tar_path, zip_path, integrity_path): self.assertTrue(path.is_file(), path) self.assertGreater(path.stat().st_size, 1_000) verify = subprocess.run( [ sys.executable, str(PACKAGE_ROOT / "scripts" / "verify_release.py"), "--source-tree", str(PACKAGE_ROOT), str(tar_path), str(zip_path), ], cwd=PACKAGE_ROOT, text=True, capture_output=True, check=False, timeout=120, ) self.assertEqual(verify.returncode, 0, verify.stderr or verify.stdout) integrity = json.loads(integrity_path.read_text(encoding="utf-8")) self.assertGreaterEqual(integrity["source_files"], 100) self.assertEqual(len(integrity["archives"]), 2) self.assertEqual( {item["manifest_files"] for item in integrity["archives"]}, {integrity["source_files"]}, ) def test_builder_rejects_output_inside_source_tree(self) -> None: output = PACKAGE_ROOT / ".forbidden-release-output" self.assertFalse(output.exists()) result = subprocess.run( [ sys.executable, str(PACKAGE_ROOT / "scripts" / "build_release.py"), "--output-dir", str(output), "--skip-validation", "--no-reproducibility-check", ], cwd=PACKAGE_ROOT, text=True, capture_output=True, check=False, timeout=60, ) self.assertNotEqual(result.returncode, 0) self.assertIn("inside the source tree", result.stderr + result.stdout) self.assertFalse(output.exists()) def test_release_verifier_rejects_tampering_and_wrong_root(self) -> None: package = PACKAGE_NAME with tempfile.TemporaryDirectory(prefix="mmo-release-tamper-") as temporary_raw: output = Path(temporary_raw) result = subprocess.run( [ sys.executable, str(PACKAGE_ROOT / "scripts" / "build_release.py"), "--output-dir", str(output), "--skip-validation", "--no-reproducibility-check", ], cwd=PACKAGE_ROOT, text=True, capture_output=True, check=False, timeout=120, ) self.assertEqual(result.returncode, 0, result.stderr or result.stdout) original = output / f"{package}-linux.zip" tampered = output / "tampered.zip" with ( zipfile.ZipFile(original) as source, zipfile.ZipFile(tampered, "w", compression=zipfile.ZIP_DEFLATED) as destination, ): for info in source.infolist(): data = source.read(info.filename) if info.filename == f"{package}/README.md": data += b"\nTAMPERED\n" destination.writestr(copy.copy(info), data) wrong_root = output / "wrong-root.zip" with ( zipfile.ZipFile(original) as source, zipfile.ZipFile(wrong_root, "w", compression=zipfile.ZIP_DEFLATED) as destination, ): for info in source.infolist(): renamed = copy.copy(info) renamed.filename = info.filename.replace(package, "wrong-package-root", 1) destination.writestr(renamed, source.read(info.filename)) mode_drift = output / "mode-drift.zip" with ( zipfile.ZipFile(original) as source, zipfile.ZipFile(mode_drift, "w", compression=zipfile.ZIP_DEFLATED) as destination, ): for info in source.infolist(): changed = copy.copy(info) if info.filename == f"{package}/README.md": changed.external_attr = (stat.S_IFREG | 0o666) << 16 destination.writestr(changed, source.read(info.filename)) directory_mode_drift = output / "directory-mode-drift.zip" with ( zipfile.ZipFile(original) as source, zipfile.ZipFile( directory_mode_drift, "w", compression=zipfile.ZIP_DEFLATED ) as destination, ): for info in source.infolist(): changed = copy.copy(info) if info.filename == f"{package}/docs/": changed.external_attr = (stat.S_IFDIR | 0o777) << 16 destination.writestr(changed, source.read(info.filename)) missing_directory = output / "missing-directory-entry.zip" with ( zipfile.ZipFile(original) as source, zipfile.ZipFile( missing_directory, "w", compression=zipfile.ZIP_DEFLATED ) as destination, ): for info in source.infolist(): if info.filename != f"{package}/docs/": destination.writestr(copy.copy(info), source.read(info.filename)) extra_directory = output / "extra-empty-directory.zip" with ( zipfile.ZipFile(original) as source, zipfile.ZipFile( extra_directory, "w", compression=zipfile.ZIP_DEFLATED ) as destination, ): for info in source.infolist(): destination.writestr(copy.copy(info), source.read(info.filename)) empty = zipfile.ZipInfo(f"{package}/unexpected-empty/") empty.create_system = 3 empty.external_attr = (stat.S_IFDIR | 0o755) << 16 destination.writestr(empty, b"") original_tar = output / f"{package}-linux.tar.gz" tar_directory_mode_drift = output / "directory-mode-drift.tar.gz" with ( tarfile.open(original_tar, "r:gz") as source, tarfile.open(tar_directory_mode_drift, "w:gz") as destination, ): for tar_info in source: changed_tar = copy.copy(tar_info) if tar_info.name == f"{package}/docs": changed_tar.mode = 0o777 payload = source.extractfile(tar_info) if tar_info.isfile() else None if payload is None: destination.addfile(changed_tar) else: with payload: destination.addfile(changed_tar, payload) verify = subprocess.run( [ sys.executable, str(PACKAGE_ROOT / "scripts" / "verify_release.py"), str(tampered), str(wrong_root), str(mode_drift), str(directory_mode_drift), str(missing_directory), str(extra_directory), str(tar_directory_mode_drift), ], cwd=PACKAGE_ROOT, text=True, capture_output=True, check=False, timeout=120, ) self.assertNotEqual(verify.returncode, 0) report = json.loads(verify.stdout) self.assertEqual(len(report["errors"]), 7) self.assertTrue(any("sha256" in item for item in report["errors"])) self.assertTrue(any("top-level directory" in item for item in report["errors"])) self.assertTrue(any("mode" in item for item in report["errors"])) self.assertEqual(sum("invalid permission mode" in item for item in report["errors"]), 3) self.assertEqual(sum("directory set differs" in item for item in report["errors"]), 2) def test_release_verifier_rejects_path_traversal(self) -> None: with tempfile.TemporaryDirectory(prefix="mmo-release-traversal-") as temporary_raw: archive = Path(temporary_raw) / "traversal.zip" with zipfile.ZipFile(archive, "w") as handle: handle.writestr("../escape.txt", "bad") result = subprocess.run( [ sys.executable, str(PACKAGE_ROOT / "scripts" / "verify_release.py"), str(archive), ], cwd=PACKAGE_ROOT, text=True, capture_output=True, check=False, timeout=60, ) self.assertNotEqual(result.returncode, 0) self.assertIn("unsafe archive member path", result.stdout) def test_release_verifier_rejects_canonical_path_aliases(self) -> None: with tempfile.TemporaryDirectory(prefix="mmo-release-alias-") as temporary_raw: archive = Path(temporary_raw) / "alias.zip" with zipfile.ZipFile(archive, "w") as handle: handle.writestr(f"{PACKAGE_NAME}/nested/file.txt", "first") handle.writestr(f"{PACKAGE_NAME}/nested/./file.txt", "second") result = subprocess.run( [ sys.executable, str(PACKAGE_ROOT / "scripts" / "verify_release.py"), str(archive), ], cwd=PACKAGE_ROOT, text=True, capture_output=True, check=False, timeout=60, ) self.assertNotEqual(result.returncode, 0) self.assertIn("unsafe archive member path", result.stdout) def test_builder_rejects_invalid_source_epoch(self) -> None: with tempfile.TemporaryDirectory(prefix="mmo-release-epoch-") as temporary_raw: result = subprocess.run( [ sys.executable, str(PACKAGE_ROOT / "scripts" / "build_release.py"), "--output-dir", temporary_raw, "--source-date-epoch", "-1", "--skip-validation", "--no-reproducibility-check", ], cwd=PACKAGE_ROOT, text=True, capture_output=True, check=False, timeout=60, ) self.assertNotEqual(result.returncode, 0) self.assertIn("source-date-epoch", result.stderr + result.stdout) def test_one_file_archives_are_rejected(self) -> None: package = PACKAGE_NAME with tempfile.TemporaryDirectory(prefix="mmo-release-corrupt-") as temporary_raw: temporary = Path(temporary_raw) zip_path = temporary / "incomplete.zip" tar_path = temporary / "incomplete.tar.gz" payload = temporary / "PLAN-COVERAGE.md" payload.write_text("incomplete\n", encoding="utf-8") with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as archive: archive.write(payload, f"{package}/docs/PLAN-COVERAGE.md") with tarfile.open(tar_path, "w:gz") as archive: archive.add(payload, arcname=f"{package}/docs/PLAN-COVERAGE.md") result = subprocess.run( [ sys.executable, str(PACKAGE_ROOT / "scripts" / "verify_release.py"), str(zip_path), str(tar_path), ], cwd=PACKAGE_ROOT, text=True, capture_output=True, check=False, timeout=60, ) self.assertNotEqual(result.returncode, 0) report = json.loads(result.stdout) self.assertFalse(report["passed"]) self.assertEqual(len(report["errors"]), 2) self.assertTrue( all("archive directory set differs" in item for item in report["errors"]) ) if __name__ == "__main__": unittest.main()