#!/usr/bin/env python3 """Filesystem and failure-path tests for immutable staging deployment.""" from __future__ import annotations import hashlib import importlib.util import io import json from pathlib import Path import subprocess import sys import tarfile import tempfile import unittest from unittest import mock ROOT = Path(__file__).resolve().parents[1] SCRIPT = ROOT / "scripts" / "deploy_staging.py" COMMIT_A = "a" * 40 COMMIT_B = "b" * 40 def load_deploy(): spec = importlib.util.spec_from_file_location("deploy_staging", SCRIPT) module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def make_archive(path: Path, members: list[tuple[str, bytes, str]]) -> str: with tarfile.open(path, "w:gz") as archive: for name, body, kind in members: info = tarfile.TarInfo(name) if kind == "file": info.size = len(body) archive.addfile(info, io.BytesIO(body)) elif kind == "dir": info.type = tarfile.DIRTYPE archive.addfile(info) elif kind == "symlink": info.type = tarfile.SYMTYPE info.linkname = "server.mjs" archive.addfile(info) elif kind == "hardlink": info.type = tarfile.LNKTYPE info.linkname = "server.mjs" archive.addfile(info) elif kind == "fifo": info.type = tarfile.FIFOTYPE archive.addfile(info) elif kind == "device": info.type = tarfile.CHRTYPE archive.addfile(info) return hashlib.sha256(path.read_bytes()).hexdigest() class DeployTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) / "opt" self.root.mkdir() self.deploy = load_deploy() self.config = self.deploy.DeploymentConfig( root=self.root, restart_command=("fixture-restart",), stop_command=("fixture-stop",), smoke_command=("fixture-smoke",), health_url="http://127.0.0.1:4174/api/healthz", command_timeout=1.0, health_timeout=1.0, ) self.commands = [] def tearDown(self): self.tmp.cleanup() def runner(self, argv, **kwargs): self.commands.append((tuple(argv), kwargs)) return subprocess.CompletedProcess(argv, 0, "", "") def healthy(self, url, commit, timeout): self.assertEqual(url, self.config.health_url) self.assertEqual(timeout, self.config.health_timeout) return {"ok": True, "commit": commit} def archive(self, name="release.tar.gz", members=None): path = Path(self.tmp.name) / name digest = make_archive(path, members or [ ("timmy-release/", b"", "dir"), ("timmy-release/server.mjs", b"console.log('ok')\n", "file"), ("timmy-release/package.json", b"{}\n", "file"), ]) return path, digest def seed_release(self, commit): release = self.root / "releases" / commit release.mkdir(parents=True) (release / "server.mjs").write_text("ok", encoding="utf-8") (release / ".timmy-release.json").write_text(json.dumps({"commit": commit, "tag": "old"}), encoding="utf-8") return release def point_current(self, commit): (self.root / "current").symlink_to(Path("releases") / commit) def promote(self, archive, digest, commit=COMMIT_B, **kwargs): return self.deploy.promote( config=self.config, tag="daily-test", archive=archive, expected_sha256=digest, commit=commit, run_command=kwargs.get("run_command", self.runner), health_check=kwargs.get("health_check", self.healthy), ) def test_checksum_mismatch_fails_before_tar_is_opened(self): archive = Path(self.tmp.name) / "not-even-a-tar" archive.write_bytes(b"untrusted") with self.assertRaisesRegex(self.deploy.DeploymentError, "SHA-256 mismatch"): self.promote(archive, "0" * 64) self.assertFalse((self.root / "releases").exists()) def test_verified_private_archive_copy_is_used_after_caller_archive_mutates(self): archive, digest = self.archive("mutable.tar.gz") evil = Path(self.tmp.name) / "evil.tar.gz" make_archive(evil, [ ("timmy-release/", b"", "dir"), ("timmy-release/server.mjs", b"EVIL\n", "file"), ]) real_inspect = self.deploy.inspect_archive inspected_paths = [] def mutate_then_inspect(path, config): inspected_paths.append(Path(path)) archive.write_bytes(evil.read_bytes()) return real_inspect(path, config) with mock.patch.object(self.deploy, "inspect_archive", side_effect=mutate_then_inspect): self.promote(archive, digest) release = self.root / "releases" / COMMIT_B self.assertNotEqual(inspected_paths, [archive]) self.assertEqual((release / "server.mjs").read_text(), "console.log('ok')\n") self.assertTrue(all(not path.exists() for path in inspected_paths), "verified temp archive must be cleaned") def test_archive_source_must_be_a_nonsymlink_regular_file(self): archive, digest = self.archive("regular.tar.gz") symlink = Path(self.tmp.name) / "archive-link.tar.gz" symlink.symlink_to(archive) for source in (symlink, Path(self.tmp.name)): with self.subTest(source=source), self.assertRaisesRegex(self.deploy.DeploymentError, "regular file|stage archive"): self.promote(source, digest) def test_absolute_and_traversal_paths_are_rejected(self): for index, unsafe in enumerate(("/etc/passwd", "root/../../escape", "../escape", "root//double")): archive, digest = self.archive(f"unsafe-{index}.tar.gz", [(unsafe, b"bad", "file")]) with self.subTest(unsafe=unsafe), self.assertRaisesRegex(self.deploy.DeploymentError, "unsafe archive path"): self.promote(archive, digest) self.assertFalse((self.root / "releases" / COMMIT_B).exists()) def test_links_devices_fifos_and_oversized_archives_are_rejected(self): for index, kind in enumerate(("symlink", "hardlink", "fifo", "device")): archive, digest = self.archive(f"special-{index}.tar.gz", [(f"root/bad-{kind}", b"", kind)]) with self.subTest(kind=kind), self.assertRaisesRegex(self.deploy.DeploymentError, "unsupported archive member"): self.promote(archive, digest) archive, digest = self.archive("large.tar.gz", [("root/large", b"x" * 17, "file")]) tiny = self.deploy.DeploymentConfig(**{**self.config.__dict__, "max_member_bytes": 16}) with self.assertRaisesRegex(self.deploy.DeploymentError, "member size"): self.deploy.promote(config=tiny, tag="daily-test", archive=archive, expected_sha256=digest, commit=COMMIT_B, run_command=self.runner, health_check=self.healthy) def test_member_count_limit_is_enforced(self): archive, digest = self.archive("many.tar.gz", [(f"root/{i}", b"x", "file") for i in range(3)]) tiny = self.deploy.DeploymentConfig(**{**self.config.__dict__, "max_members": 2}) with self.assertRaisesRegex(self.deploy.DeploymentError, "member count"): self.deploy.promote(config=tiny, tag="daily-test", archive=archive, expected_sha256=digest, commit=COMMIT_B, run_command=self.runner, health_check=self.healthy) def test_secrets_env_and_forbidden_artifacts_are_rejected(self): forbidden = ("root/.env", "root/secrets/token.txt", "root/private.pem", "root/model.gguf", "root/.git/config", "root/video/raw.webm") for index, name in enumerate(forbidden): archive, digest = self.archive(f"secret-{index}.tar.gz", [(name, b"secret", "file")]) with self.subTest(name=name), self.assertRaisesRegex(self.deploy.DeploymentError, "forbidden archive artifact"): self.promote(archive, digest) def test_forbidden_common_roots_are_rejected_case_insensitively_but_release_wrapper_is_allowed(self): for index, root in enumerate((".git", "ViDeO", "ARTIFACTS", "Credentials")): archive, _ = self.archive(f"forbidden-root-{index}.tar.gz", [ (f"{root}/", b"", "dir"), (f"{root}/server.mjs", b"evil", "file"), ]) with self.subTest(root=root), self.assertRaisesRegex(self.deploy.DeploymentError, "forbidden archive artifact"): self.deploy.inspect_archive(archive, self.config) archive, _ = self.archive("normal-wrapper.tar.gz") _, wrapper = self.deploy.inspect_archive(archive, self.config) self.assertEqual(wrapper, "timmy-release") def test_release_root_symlink_is_rejected(self): outside = Path(self.tmp.name) / "outside" outside.mkdir() (self.root / "releases").symlink_to(outside, target_is_directory=True) archive, digest = self.archive() with self.assertRaisesRegex(self.deploy.DeploymentError, "releases directory"): self.promote(archive, digest) self.assertEqual(list(outside.iterdir()), []) def test_deployment_root_symlink_is_rejected_before_writing_outside(self): outside = Path(self.tmp.name) / "outside-root" outside.mkdir() self.root.rmdir() self.root.symlink_to(outside, target_is_directory=True) archive, digest = self.archive() with self.assertRaisesRegex(self.deploy.DeploymentError, "deployment root"): self.promote(archive, digest) self.assertEqual(list(outside.iterdir()), []) def test_deployment_root_non_directory_is_rejected(self): self.root.rmdir() self.root.write_text("not a directory") archive, digest = self.archive() with self.assertRaisesRegex(self.deploy.DeploymentError, "deployment root is not a directory"): self.promote(archive, digest) def test_current_and_pending_symlink_boundaries_are_rejected_before_extraction(self): outside = Path(self.tmp.name) / "outside-boundary" outside.mkdir() (self.root / "releases").mkdir() (self.root / "current").symlink_to(outside, target_is_directory=True) archive, digest = self.archive("current-boundary.tar.gz") with self.assertRaisesRegex(self.deploy.DeploymentError, "current symlink escapes"): self.promote(archive, digest) self.assertEqual(list(outside.iterdir()), []) (self.root / "current").unlink() pending = self.root / "releases" / f".pending-{COMMIT_B}-fixed" pending.symlink_to(outside, target_is_directory=True) with mock.patch.object(self.deploy.secrets, "token_hex", return_value="fixed"): with self.assertRaisesRegex(self.deploy.DeploymentError, "pending release boundary"): self.promote(archive, digest) self.assertEqual(list(outside.iterdir()), []) def test_valid_archive_extracts_to_commit_release_and_is_never_overwritten(self): archive, digest = self.archive() result = self.promote(archive, digest) release = self.root / "releases" / COMMIT_B self.assertEqual(result["commit"], COMMIT_B) self.assertEqual((release / "server.mjs").read_text(), "console.log('ok')\n") self.assertEqual(json.loads((release / ".timmy-release.json").read_text())["sha256"], digest) with self.assertRaisesRegex(self.deploy.DeploymentError, "already exists"): self.promote(archive, digest) def test_success_atomically_swaps_current_and_uses_bounded_argv_commands(self): self.seed_release(COMMIT_A) self.point_current(COMMIT_A) archive, digest = self.archive() self.promote(archive, digest) self.assertEqual((self.root / "current").resolve(), self.root / "releases" / COMMIT_B) self.assertEqual([command for command, _ in self.commands], [("fixture-restart",), ("fixture-smoke",)]) self.assertTrue(all(options["shell"] is False and options["timeout"] == 1.0 for _, options in self.commands)) self.assertFalse(any(path.name.startswith(".current-") for path in self.root.iterdir())) def test_restart_health_and_smoke_failures_automatically_restore_prior_release(self): phases = ("restart", "health", "smoke") for phase in phases: with self.subTest(phase=phase): root = Path(self.tmp.name) / phase config = self.deploy.DeploymentConfig(**{**self.config.__dict__, "root": root}) release = root / "releases" / COMMIT_A release.mkdir(parents=True) (release / "server.mjs").write_text("ok") (root / "current").symlink_to(Path("releases") / COMMIT_A) archive, digest = self.archive(f"{phase}.tar.gz") calls = [] def run(argv, **kwargs): calls.append(tuple(argv)) if (phase == "restart" and len(calls) == 1) or (phase == "smoke" and tuple(argv) == ("fixture-smoke",)): raise subprocess.CalledProcessError(1, argv) return subprocess.CompletedProcess(argv, 0, "", "") def health(url, commit, timeout): if phase == "health" and commit == COMMIT_B: raise self.deploy.DeploymentError("health failed") return {"ok": True, "commit": commit} with self.assertRaisesRegex(self.deploy.DeploymentError, "promotion verification failed"): self.deploy.promote(config=config, tag="daily-test", archive=archive, expected_sha256=digest, commit=COMMIT_B, run_command=run, health_check=health) self.assertEqual((root / "current").resolve(), root / "releases" / COMMIT_A) self.assertGreaterEqual(calls.count(("fixture-restart",)), 2) def test_first_promotion_failure_removes_current_and_stops_service_without_restart(self): archive, digest = self.archive("first-failure.tar.gz") calls = [] def run(argv, **kwargs): calls.append(tuple(argv)) if tuple(argv) == ("fixture-smoke",): raise subprocess.CalledProcessError(1, argv) return subprocess.CompletedProcess(argv, 0, "", "") with self.assertRaisesRegex(self.deploy.DeploymentError, "no prior release; service stopped"): self.promote(archive, digest, run_command=run) self.assertFalse((self.root / "current").exists()) self.assertFalse((self.root / "current").is_symlink()) self.assertEqual(calls, [("fixture-restart",), ("fixture-smoke",), ("fixture-stop",)]) self.assertTrue((self.root / "releases" / COMMIT_B).is_dir(), "failed release remains inert evidence") def test_rollback_requires_valid_immutable_release_and_restarts_and_checks_health(self): self.seed_release(COMMIT_A) self.seed_release(COMMIT_B) self.point_current(COMMIT_B) result = self.deploy.rollback(config=self.config, commit=COMMIT_A, run_command=self.runner, health_check=self.healthy) self.assertEqual(result["commit"], COMMIT_A) self.assertEqual((self.root / "current").resolve(), self.root / "releases" / COMMIT_A) with self.assertRaisesRegex(self.deploy.DeploymentError, "does not exist"): self.deploy.rollback(config=self.config, commit="c" * 40, run_command=self.runner, health_check=self.healthy) tampered = self.seed_release("d" * 40) (tampered / ".timmy-release.json").write_text(json.dumps({"commit": COMMIT_A, "tag": "wrong"})) with self.assertRaisesRegex(self.deploy.DeploymentError, "metadata"): self.deploy.rollback(config=self.config, commit="d" * 40, run_command=self.runner, health_check=self.healthy) def test_default_command_adapter_accepts_verifier_kwargs_and_stays_bounded(self): result = self.deploy._run_argv( (sys.executable, "-c", "print('ok')"), check=True, text=True, capture_output=True, shell=False, timeout=1.0, ) self.assertEqual(result.stdout.strip(), "ok") def test_status_and_cli_dry_run_are_rootless_and_machine_readable(self): self.seed_release(COMMIT_A) self.point_current(COMMIT_A) status = self.deploy.status(self.config) self.assertEqual(status["commit"], COMMIT_A) run = subprocess.run([ sys.executable, str(SCRIPT), "--root", str(self.root), "--dry-run", "status" ], text=True, capture_output=True, check=False) self.assertEqual(run.returncode, 0, run.stderr) self.assertEqual(json.loads(run.stdout)["commit"], COMMIT_A) def test_commit_tag_and_checksum_arguments_are_strictly_validated(self): archive, digest = self.archive() for commit in ("abc", "A" * 40, "a" * 41, "../" + "a" * 40): with self.subTest(commit=commit), self.assertRaisesRegex(self.deploy.DeploymentError, "commit"): self.promote(archive, digest, commit=commit) with self.assertRaisesRegex(self.deploy.DeploymentError, "tag"): self.deploy.promote(config=self.config, tag="../bad", archive=archive, expected_sha256=digest, commit=COMMIT_B, run_command=self.runner, health_check=self.healthy) if __name__ == "__main__": unittest.main()