429 lines
19 KiB
Python
429 lines
19 KiB
Python
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()
|