vignette/scripts/test_serve_nas_preview_rollback_executor.py

from __future__ import annotations

import hashlib
import importlib.util
import io
import json
import os
import re
import stat
import subprocess
import sys
import tempfile
import threading
import unittest
import urllib.error
import urllib.request
from pathlib import Path
from types import SimpleNamespace
from unittest import mock


SCRIPT_PATH = Path(__file__).with_name("serve-nas-preview-rollback-executor.py")
SPEC = importlib.util.spec_from_file_location(
    "serve_nas_preview_rollback_executor", SCRIPT_PATH
)
assert SPEC is not None and SPEC.loader is not None
MODULE = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = MODULE
SPEC.loader.exec_module(MODULE)


OLD_IMAGES = MODULE.ImagePair("sha256:" + "1" * 64, "sha256:" + "2" * 64)
DESIRED_IMAGES = MODULE.ImagePair("sha256:" + "3" * 64, "sha256:" + "4" * 64)
JOURNAL_KEY = b"j" * 32


def _resolved_config_payload(
    images: MODULE.ImagePair = DESIRED_IMAGES,
) -> dict[str, object]:
    return {
        "name": MODULE.TARGET_PROJECT,
        "networks": {
            "vignette": {
                "driver": "bridge",
                "name": f"{MODULE.TARGET_PROJECT}_vignette",
            }
        },
        "services": {
            "api": {
                "image": images.api_image,
                "networks": {"vignette": None},
                "volumes": [
                    {
                        "type": "volume",
                        "source": "apiuploads",
                        "target": "/app/uploads",
                    }
                ],
            },
            "db": {},
            "proxy": {},
            "web": {
                "image": images.web_image,
                "networks": {"vignette": None},
            },
        },
        "volumes": {
            "apiuploads": {},
            "caddydata": {},
            "pgdata": {},
        },
    }


def _artifact_payload(**changes: object) -> dict[str, object]:
    payload: dict[str, object] = {
        "idempotency_key": "10000000-0000-0000-0000-000000000001",
        "approval_event_id": "20000000-0000-0000-0000-000000000001",
        "authorization_evidence_refs": ["audit://g8/human-approval/1"],
        "artifact_id": "rollback-preview-release-v1",
        "artifact_sha256": "a" * 64,
        "artifact_record_id": "40000000-0000-0000-0000-000000000001",
        "artifact_provenance_uri": "repo://vignette/releases/rollback-v1",
        "target_id": "30000000-0000-0000-0000-000000000001",
        "subject_id": "candidate-release-v2",
        "rollback_target_id": "stable-release-v1",
        "compose_file": (
            "/volume1/docker/vignette-preview-20260807/.rollback-executor/"
            "runtime/rollback-preview-release-v1/docker-compose.yml"
        ),
        "compose_sha256": "b" * 64,
        "env_file": (
            "/volume1/docker/vignette-preview-20260807/.rollback-executor/"
            "runtime/rollback-preview-release-v1/.env"
        ),
        "env_sha256": "c" * 64,
        "resolved_config_sha256": MODULE._resolved_policy_sha256(
            _resolved_config_payload()
        ),
        "api_image": DESIRED_IMAGES.api_image,
        "web_image": DESIRED_IMAGES.web_image,
    }
    payload.update(changes)
    return payload


def _manifest_payload(**artifact_changes: object) -> dict[str, object]:
    return {
        "schema_version": MODULE.MANIFEST_SCHEMA,
        "target": {
            "compose_project": MODULE.TARGET_PROJECT,
            "remote_root": str(MODULE.TARGET_ROOT),
            "services": list(MODULE.TARGET_SERVICES),
            "health_url": MODULE.TARGET_HEALTH_URL,
        },
        "artifacts": [_artifact_payload(**artifact_changes)],
    }


def _request_payload(**changes: object) -> dict[str, object]:
    payload: dict[str, object] = {
        "schema_version": MODULE.REQUEST_SCHEMA,
        "idempotency_key": "10000000-0000-0000-0000-000000000001",
        "approval_event_id": "20000000-0000-0000-0000-000000000001",
        "rollback_scope": "runtime",
        "target_kind": "release_gate",
        "target_id": "30000000-0000-0000-0000-000000000001",
        "subject_id": "candidate-release-v2",
        "rollback_target_id": "stable-release-v1",
        "artifact_record_id": "40000000-0000-0000-0000-000000000001",
        "artifact_id": "rollback-preview-release-v1",
        "artifact_sha256": "a" * 64,
        "artifact_provenance_uri": "repo://vignette/releases/rollback-v1",
        "authorization_evidence_refs": ["audit://g8/human-approval/1"],
    }
    payload.update(changes)
    return payload


class FakeRuntime:
    def __init__(
        self,
        *,
        fail_first_apply: bool = False,
        fail_restore_verify: bool = False,
    ) -> None:
        self.current = OLD_IMAGES
        self.fail_first_apply = fail_first_apply
        self.fail_restore_verify = fail_restore_verify
        self.apply_calls: list[MODULE.ImagePair] = []
        self.snapshot_calls = 0
        self.ensure_calls: list[MODULE.ImagePair] = []

    def snapshot(self) -> MODULE.ImagePair:
        self.snapshot_calls += 1
        return self.current

    def ensure_images_present(self, images: MODULE.ImagePair) -> None:
        self.ensure_calls.append(images)

    def apply(
        self,
        artifact: MODULE.RollbackArtifact,
        images: MODULE.ImagePair,
        execution_id: str,
    ) -> None:
        del artifact, execution_id
        self.apply_calls.append(images)
        self.current = images
        if self.fail_first_apply and len(self.apply_calls) == 1:
            raise MODULE.CommandFailure("synthetic_partial_failure")

    def verify(self, images: MODULE.ImagePair) -> None:
        if self.fail_restore_verify and len(self.apply_calls) >= 2:
            raise MODULE.CommandFailure("synthetic_restore_mismatch")
        if self.current != images:
            raise MODULE.CommandFailure("synthetic_image_mismatch")


class FakeHealth:
    def __init__(self, *, failures: int = 0) -> None:
        self.failures = failures
        self.calls = 0

    def wait_ready(self) -> None:
        self.calls += 1
        if self.calls <= self.failures:
            raise MODULE.CommandFailure("synthetic_health_failure")


class CaptureCommandRunner:
    def __init__(self) -> None:
        self.calls: list[tuple[str, list[str], float]] = []

    def run(
        self,
        operation: str,
        argv: list[str],
        *,
        timeout: float,
    ) -> MODULE.ProcessResult:
        self.calls.append((operation, argv, timeout))
        if operation == "inspect_preview_network":
            return MODULE.ProcessResult(
                0,
                f"{MODULE.TARGET_PROJECT}|vignette\n",
            )
        if operation == "inspect_preview_volume":
            return MODULE.ProcessResult(
                0,
                f"{MODULE.TARGET_PROJECT}|apiuploads\n",
            )
        if operation == "compose_policy":
            return MODULE.ProcessResult(
                0,
                json.dumps(_resolved_config_payload()),
            )
        return MODULE.ProcessResult(0, "")


class StatefulDockerRunner(CaptureCommandRunner):
    def __init__(self) -> None:
        super().__init__()
        self.current = OLD_IMAGES
        self.recreate_calls = 0

    @staticmethod
    def _override_images(argv: list[str]) -> MODULE.ImagePair:
        last_flag = len(argv) - 1 - argv[::-1].index("-f")
        override = Path(argv[last_flag + 1])
        values = re.findall(
            r"sha256:[a-f0-9]{64}",
            override.read_text(encoding="utf-8"),
        )
        if len(values) != 2:
            raise AssertionError("override image pair missing")
        return MODULE.ImagePair(values[0], values[1])

    def run(
        self,
        operation: str,
        argv: list[str],
        *,
        timeout: float,
    ) -> MODULE.ProcessResult:
        self.calls.append((operation, argv, timeout))
        if operation.endswith("_project"):
            return MODULE.ProcessResult(0, MODULE.TARGET_PROJECT + "\n")
        if operation.endswith("_service"):
            service = "api" if "api" in operation else "web"
            return MODULE.ProcessResult(0, service + "\n")
        if operation == "inspect_api_image":
            return MODULE.ProcessResult(0, self.current.api_image + "\n")
        if operation == "inspect_web_image":
            return MODULE.ProcessResult(0, self.current.web_image + "\n")
        if operation.startswith("verify_") and operation.endswith("_image_present"):
            return MODULE.ProcessResult(0, argv[-1] + "\n")
        if operation == "inspect_preview_network":
            return MODULE.ProcessResult(0, f"{MODULE.TARGET_PROJECT}|vignette\n")
        if operation == "inspect_preview_volume":
            return MODULE.ProcessResult(0, f"{MODULE.TARGET_PROJECT}|apiuploads\n")
        if operation == "compose_policy":
            images = self._override_images(argv)
            return MODULE.ProcessResult(0, json.dumps(_resolved_config_payload(images)))
        if operation == "compose_recreate":
            self.current = self._override_images(argv)
            self.recreate_calls += 1
            return MODULE.ProcessResult(0, "")
        raise AssertionError(f"unexpected operation: {operation}")


class FailSuccessJournal(MODULE.DurableJournal):
    def __init__(self, state_dir: Path) -> None:
        super().__init__(state_dir, integrity_key=JOURNAL_KEY)
        self.failed_once = False

    def append(self, event: dict[str, object]) -> dict[str, object]:
        if event.get("event") == "executed" and not self.failed_once:
            self.failed_once = True
            raise MODULE.ExecutorError(
                "synthetic_journal_commit_failure", http_status=500
            )
        return super().append(event)


class NasPreviewRollbackExecutorTest(unittest.TestCase):
    def setUp(self) -> None:
        self.temp = tempfile.TemporaryDirectory(prefix="vignette-g8-executor-test-")
        self.root = Path(self.temp.name)
        self.manifest_path = self.root / "rollback-manifest.json"
        self.state_dir = self.root / "state"
        self.state_dir.mkdir()

    def tearDown(self) -> None:
        self.temp.cleanup()

    def write_manifest(
        self, payload: dict[str, object] | None = None
    ) -> tuple[MODULE.RollbackManifest, str]:
        raw = json.dumps(
            payload or _manifest_payload(),
            ensure_ascii=False,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("utf-8")
        self.manifest_path.write_bytes(raw)
        if os.name != "nt":
            self.manifest_path.chmod(0o600)
        digest = hashlib.sha256(raw).hexdigest()
        return MODULE.load_manifest(self.manifest_path, digest), digest

    def make_executor(
        self,
        *,
        runtime: FakeRuntime | None = None,
        health: FakeHealth | None = None,
    ) -> tuple[MODULE.NasPreviewRollbackExecutor, FakeRuntime, FakeHealth]:
        manifest, _ = self.write_manifest()
        actual_runtime = runtime or FakeRuntime()
        actual_health = health or FakeHealth()
        executor = MODULE.NasPreviewRollbackExecutor(
            manifest=manifest,
            runtime=actual_runtime,
            health=actual_health,
            journal=MODULE.DurableJournal(
                self.state_dir,
                integrity_key=JOURNAL_KEY,
            ),
        )
        return executor, actual_runtime, actual_health

    def test_default_serve_mode_is_disabled_twice(self) -> None:
        _, digest = self.write_manifest()
        with mock.patch.dict(os.environ, {MODULE.ENABLED_ENV: "false"}, clear=False):
            with self.assertRaisesRegex(SystemExit, "explicit CLI and environment"):
                MODULE.main(
                    [
                        "--manifest",
                        str(self.manifest_path),
                        "--manifest-sha256",
                        digest,
                    ]
                )
        with mock.patch.dict(os.environ, {MODULE.ENABLED_ENV: "true"}, clear=False):
            with self.assertRaisesRegex(SystemExit, "explicit CLI and environment"):
                MODULE.main(
                    [
                        "--manifest",
                        str(self.manifest_path),
                        "--manifest-sha256",
                        digest,
                    ]
                )

    def test_check_config_is_non_mutating_and_needs_no_token(self) -> None:
        _, digest = self.write_manifest()
        stdout = io.StringIO()
        with (
            mock.patch.dict(os.environ, {}, clear=True),
            mock.patch("sys.stdout", stdout),
        ):
            result = MODULE.main(
                [
                    "--manifest",
                    str(self.manifest_path),
                    "--manifest-sha256",
                    digest,
                    "--check-config",
                ]
            )
        self.assertEqual(0, result)
        self.assertFalse(json.loads(stdout.getvalue())["enabled"])
        self.assertEqual([], list(self.state_dir.iterdir()))

    def test_authentication_uses_constant_time_byte_comparison(self) -> None:
        with mock.patch.object(
            MODULE.hmac, "compare_digest", return_value=False
        ) as compare:
            self.assertFalse(MODULE.authenticate("wrong", "configured-secret"))
        compare.assert_called_once_with(b"wrong", b"configured-secret")

    def test_authenticated_health_check_is_read_only(self) -> None:
        executor, runtime, health = self.make_executor()
        token = "x" * 32
        server = MODULE.QuietThreadingHTTPServer(
            ("127.0.0.1", 0),
            MODULE._handler(token=token, executor=executor),
        )
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        endpoint = f"http://127.0.0.1:{server.server_port}/healthz"
        try:
            with self.assertRaises(urllib.error.HTTPError) as unauthorized:
                urllib.request.urlopen(endpoint, timeout=2)
            self.assertEqual(403, unauthorized.exception.code)
            request = urllib.request.Request(
                endpoint,
                headers={MODULE.TOKEN_HEADER: token},
            )
            with urllib.request.urlopen(request, timeout=2) as response:
                payload = json.load(response)
            self.assertEqual("ok", payload["status"])
        finally:
            server.shutdown()
            server.server_close()
            thread.join(timeout=2)
        self.assertEqual(0, runtime.snapshot_calls)
        self.assertEqual(0, health.calls)

    def test_only_runtime_release_gate_requests_are_accepted(self) -> None:
        with self.assertRaisesRegex(MODULE.ExecutorError, "unsupported_rollback_scope"):
            MODULE.RollbackRequest.parse(_request_payload(rollback_scope="model"))
        with self.assertRaisesRegex(MODULE.ExecutorError, "unsupported_target_kind"):
            MODULE.RollbackRequest.parse(
                _request_payload(target_kind="model_change_gate")
            )

    def test_malicious_artifact_identifier_is_rejected_before_lookup(self) -> None:
        for artifact_id in ("../rollback", "/tmp/rollback", "rollback\\artifact"):
            with self.subTest(artifact_id=artifact_id):
                with self.assertRaisesRegex(
                    MODULE.ExecutorError, "invalid_artifact_id"
                ):
                    MODULE.RollbackRequest.parse(
                        _request_payload(artifact_id=artifact_id)
                    )

    def test_manifest_rejects_path_escape_and_unapproved_service_target(self) -> None:
        payload = _manifest_payload(
            compose_file=(
                "/volume1/docker/vignette-preview-20260807/"
                "releases/../prod/infra/docker-compose.yml"
            )
        )
        with self.assertRaisesRegex(MODULE.ExecutorError, "invalid_compose_file"):
            self.write_manifest(payload)

        payload = _manifest_payload()
        assert isinstance(payload["target"], dict)
        payload["target"]["services"] = ["api", "web", "proxy", "db"]
        with self.assertRaisesRegex(MODULE.ExecutorError, "manifest_target_invalid"):
            self.write_manifest(payload)

    def test_manifest_sha_is_pinned_and_permissions_are_checked(self) -> None:
        self.write_manifest()
        with self.assertRaisesRegex(MODULE.ExecutorError, "manifest_sha256_mismatch"):
            MODULE.load_manifest(self.manifest_path, "f" * 64)
        if os.name != "nt":
            self.manifest_path.chmod(0o622)
            digest = hashlib.sha256(self.manifest_path.read_bytes()).hexdigest()
            with self.assertRaisesRegex(MODULE.ExecutorError, "permissions_unsafe"):
                MODULE.load_manifest(self.manifest_path, digest)

    def test_posix_host_chain_requires_current_owner_and_nonwritable_ancestors(
        self,
    ) -> None:
        root = Path("/volume1/docker/vignette-preview-20260807")
        state = root / ".rollback-executor"
        runtime = state / "runtime"
        artifact_dir = runtime / "rollback-preview-release-v1"
        manifest = state / "rollback-manifest.json"
        compose = artifact_dir / "docker-compose.yml"
        env_file = artifact_dir / ".env"
        directories = {root, state, runtime, artifact_dir}
        expected_paths = {*directories, manifest, compose, env_file}
        owner = 1001
        metadata = {
            path: SimpleNamespace(
                st_uid=owner,
                st_mode=(stat.S_IFDIR | 0o755)
                if path in directories
                else (stat.S_IFREG | 0o400),
            )
            for path in expected_paths
        }

        MODULE._validate_posix_host_path_chain(
            root=root,
            state_dir=state,
            manifest_path=manifest,
            runtime_paths=(compose, env_file),
            current_euid=owner,
            lstat_func=metadata.__getitem__,
        )

        for unsafe_path in expected_paths:
            with self.subTest(unsafe_path=unsafe_path):
                original = metadata[unsafe_path]
                metadata[unsafe_path] = SimpleNamespace(
                    st_uid=owner,
                    st_mode=original.st_mode | stat.S_IWGRP,
                )
                with self.assertRaisesRegex(
                    MODULE.ExecutorError,
                    "host_path_permissions_unsafe",
                ):
                    MODULE._validate_posix_host_path_chain(
                        root=root,
                        state_dir=state,
                        manifest_path=manifest,
                        runtime_paths=(compose, env_file),
                        current_euid=owner,
                        lstat_func=metadata.__getitem__,
                    )
                metadata[unsafe_path] = original

    def test_posix_host_chain_rejects_root_0777_foreign_owner_and_ancestor_swap(
        self,
    ) -> None:
        root = Path("/volume1/docker/vignette-preview-20260807")
        state = root / ".rollback-executor"
        runtime = state / "runtime"
        artifact_dir = runtime / "rollback-preview-release-v1"
        manifest = state / "rollback-manifest.json"
        compose = artifact_dir / "docker-compose.yml"
        env_file = artifact_dir / ".env"
        owner = 1001

        def metadata() -> dict[Path, SimpleNamespace]:
            directories = {root, state, runtime, artifact_dir}
            return {
                path: SimpleNamespace(
                    st_uid=owner,
                    st_mode=(stat.S_IFDIR | 0o755)
                    if path in directories
                    else (stat.S_IFREG | 0o400),
                )
                for path in {*directories, manifest, compose, env_file}
            }

        cases = (
            (
                "permissions_unsafe",
                root,
                SimpleNamespace(st_uid=owner, st_mode=stat.S_IFDIR | 0o777),
            ),
            (
                "owner_unsafe",
                runtime,
                SimpleNamespace(st_uid=2002, st_mode=stat.S_IFDIR | 0o755),
            ),
            (
                "symlink_unsafe",
                artifact_dir,
                SimpleNamespace(st_uid=owner, st_mode=stat.S_IFLNK | 0o777),
            ),
        )
        for error, unsafe_path, replacement in cases:
            with self.subTest(error=error):
                current = metadata()
                current[unsafe_path] = replacement
                with self.assertRaisesRegex(MODULE.ExecutorError, error):
                    MODULE._validate_posix_host_path_chain(
                        root=root,
                        state_dir=state,
                        manifest_path=manifest,
                        runtime_paths=(compose, env_file),
                        current_euid=owner,
                        lstat_func=current.__getitem__,
                    )

    def test_manifest_cannot_reuse_one_approval_for_a_second_command(self) -> None:
        payload = _manifest_payload()
        assert isinstance(payload["artifacts"], list)
        payload["artifacts"].append(
            _artifact_payload(
                artifact_id="rollback-preview-release-v2",
                artifact_sha256="d" * 64,
                artifact_record_id="40000000-0000-0000-0000-000000000002",
                idempotency_key="10000000-0000-0000-0000-000000000002",
                compose_file=(
                    "/volume1/docker/vignette-preview-20260807/.rollback-executor/"
                    "runtime/rollback-preview-release-v2/docker-compose.yml"
                ),
                env_file=(
                    "/volume1/docker/vignette-preview-20260807/.rollback-executor/"
                    "runtime/rollback-preview-release-v2/.env"
                ),
            )
        )
        with self.assertRaisesRegex(MODULE.ExecutorError, "artifact_duplicate"):
            self.write_manifest(payload)

    def test_artifact_hash_and_every_request_binding_are_exact(self) -> None:
        manifest, _ = self.write_manifest()
        hash_mismatch = MODULE.RollbackRequest.parse(
            _request_payload(artifact_sha256="b" * 64)
        )
        with self.assertRaisesRegex(MODULE.ExecutorError, "artifact_not_allowlisted"):
            manifest.bind(hash_mismatch)

        record_mismatch = MODULE.RollbackRequest.parse(
            _request_payload(artifact_record_id="40000000-0000-0000-0000-000000000002")
        )
        with self.assertRaisesRegex(MODULE.ExecutorError, "binding_mismatch"):
            manifest.bind(record_mismatch)

        for changes in (
            {"idempotency_key": "10000000-0000-0000-0000-000000000002"},
            {"approval_event_id": "20000000-0000-0000-0000-000000000002"},
            {"authorization_evidence_refs": ["audit://g8/forged-approval"]},
        ):
            with self.subTest(changes=changes):
                mismatch = MODULE.RollbackRequest.parse(_request_payload(**changes))
                with self.assertRaisesRegex(MODULE.ExecutorError, "binding_mismatch"):
                    manifest.bind(mismatch)

    def test_success_receipt_is_durable_and_replay_runs_no_docker(self) -> None:
        executor, runtime, health = self.make_executor()
        request = MODULE.RollbackRequest.parse(_request_payload())

        first = executor.execute(request)
        second = executor.execute(request)

        self.assertEqual(first, second)
        self.assertEqual("executed", first["status"])
        self.assertEqual(1, runtime.snapshot_calls)
        self.assertEqual([DESIRED_IMAGES], runtime.apply_calls)
        self.assertEqual(1, health.calls)
        self.assertEqual(
            ["started", "prepared", "executed"],
            [record["event"] for record in executor.journal.read()],
        )
        self.assertTrue(first["evidence_refs"][0].startswith("audit://"))

    def test_idempotency_key_cannot_be_rebound_to_changed_request(self) -> None:
        executor, runtime, _ = self.make_executor()
        request = MODULE.RollbackRequest.parse(_request_payload())
        executor.journal.append(
            {
                "event": "started",
                "execution_id": "nas-g8-" + "f" * 24,
                "idempotency_key": request.idempotency_key,
                "request_hash": "f" * 64,
                "manifest_sha256": executor.manifest.content_sha256,
                "artifact_binding_sha256": "e" * 64,
            }
        )
        with self.assertRaisesRegex(
            MODULE.ExecutorError, "idempotency_binding_conflict"
        ):
            executor.execute(request)
        self.assertEqual([], runtime.apply_calls)

    def test_incomplete_started_record_is_closed_without_runtime_mutation(self) -> None:
        executor, runtime, _ = self.make_executor()
        request = MODULE.RollbackRequest.parse(_request_payload())
        executor.journal.append(
            {
                "event": "started",
                "execution_id": "nas-g8-" + "a" * 24,
                "idempotency_key": request.idempotency_key,
                "request_hash": MODULE._sha256(request.canonical_payload()),
                "manifest_sha256": executor.manifest.content_sha256,
                "artifact_binding_sha256": "e" * 64,
            }
        )

        executor.recover_incomplete()

        self.assertEqual([], runtime.apply_calls)
        terminal = executor.journal.read()[-1]
        self.assertEqual("failed", terminal["event"])
        self.assertEqual("not_started", terminal["rollback_status"])

    def test_incomplete_prepared_record_restores_previous_images_on_restart(
        self,
    ) -> None:
        runtime = FakeRuntime()
        runtime.current = DESIRED_IMAGES
        executor, runtime, health = self.make_executor(runtime=runtime)
        request = MODULE.RollbackRequest.parse(_request_payload())
        artifact = next(iter(executor.manifest.artifacts.values()))
        request_hash = MODULE._sha256(request.canonical_payload())
        execution_id = "nas-g8-" + "b" * 24
        executor.journal.append(
            {
                "event": "started",
                "execution_id": execution_id,
                "idempotency_key": request.idempotency_key,
                "request_hash": request_hash,
                "manifest_sha256": executor.manifest.content_sha256,
                "artifact_binding_sha256": MODULE._sha256(MODULE.asdict(artifact)),
            }
        )
        executor.journal.append(
            {
                "event": "prepared",
                "execution_id": execution_id,
                "idempotency_key": request.idempotency_key,
                "request_hash": request_hash,
                "artifact_id": artifact.artifact_id,
                "artifact_sha256": artifact.artifact_sha256,
                "artifact_binding_sha256": MODULE._sha256(MODULE.asdict(artifact)),
                "previous_api_image": OLD_IMAGES.api_image,
                "previous_web_image": OLD_IMAGES.web_image,
            }
        )

        executor.recover_incomplete()

        self.assertEqual([OLD_IMAGES], runtime.apply_calls)
        self.assertEqual(OLD_IMAGES, runtime.current)
        self.assertEqual(1, health.calls)
        self.assertEqual("verified", executor.journal.read()[-1]["rollback_status"])

    def test_partial_compose_failure_restores_previous_exact_images(self) -> None:
        runtime = FakeRuntime(fail_first_apply=True)
        executor, runtime, health = self.make_executor(runtime=runtime)
        request = MODULE.RollbackRequest.parse(_request_payload())

        with self.assertRaisesRegex(MODULE.ExecutorError, "execution_failed"):
            executor.execute(request)

        self.assertEqual([DESIRED_IMAGES, OLD_IMAGES], runtime.apply_calls)
        self.assertEqual(OLD_IMAGES, runtime.current)
        self.assertEqual(1, health.calls)
        terminal = executor.journal.read()[-1]
        self.assertEqual("verified", terminal["rollback_status"])

    def test_failed_postdeploy_health_restores_and_rechecks_previous_runtime(
        self,
    ) -> None:
        health = FakeHealth(failures=1)
        executor, runtime, health = self.make_executor(health=health)

        with self.assertRaisesRegex(MODULE.ExecutorError, "execution_failed"):
            executor.execute(MODULE.RollbackRequest.parse(_request_payload()))

        self.assertEqual([DESIRED_IMAGES, OLD_IMAGES], runtime.apply_calls)
        self.assertEqual(OLD_IMAGES, runtime.current)
        self.assertEqual(2, health.calls)

    def test_real_docker_controller_policy_allows_exact_previous_image_restore(
        self,
    ) -> None:
        manifest, _ = self.write_manifest()
        artifact = next(iter(manifest.artifacts.values()))
        runner = StatefulDockerRunner()
        controller = MODULE.DockerRuntimeController(runner, state_dir=self.state_dir)
        executor = MODULE.NasPreviewRollbackExecutor(
            manifest=manifest,
            runtime=controller,
            health=FakeHealth(failures=1),
            journal=MODULE.DurableJournal(
                self.state_dir,
                integrity_key=JOURNAL_KEY,
            ),
        )

        def runtime_hash(path: Path) -> str:
            if path.name == "docker-compose.yml":
                return artifact.compose_sha256
            return artifact.env_sha256

        with mock.patch.object(
            MODULE,
            "_sha256_regular_file",
            side_effect=runtime_hash,
        ):
            with self.assertRaisesRegex(MODULE.ExecutorError, "execution_failed"):
                executor.execute(MODULE.RollbackRequest.parse(_request_payload()))

        self.assertEqual(2, runner.recreate_calls)
        self.assertEqual(OLD_IMAGES, runner.current)
        self.assertEqual("verified", executor.journal.read()[-1]["rollback_status"])

    def test_unverified_restore_never_emits_success_receipt(self) -> None:
        runtime = FakeRuntime(fail_first_apply=True, fail_restore_verify=True)
        executor, runtime, _ = self.make_executor(runtime=runtime)

        with self.assertRaisesRegex(MODULE.ExecutorError, "rollback_unverified"):
            executor.execute(MODULE.RollbackRequest.parse(_request_payload()))

        terminal = executor.journal.read()[-1]
        self.assertEqual("failed", terminal["event"])
        self.assertEqual("unverified", terminal["rollback_status"])
        self.assertNotIn("receipt", terminal)
        self.assertFalse(executor.ready)
        with self.assertRaisesRegex(MODULE.ExecutorError, "rollback_unverified"):
            executor.execute(MODULE.RollbackRequest.parse(_request_payload()))

        blocked_runtime = FakeRuntime(fail_restore_verify=True)
        blocked_runtime.apply_calls.append(DESIRED_IMAGES)
        restarted = MODULE.NasPreviewRollbackExecutor(
            manifest=executor.manifest,
            runtime=blocked_runtime,
            health=FakeHealth(),
            journal=executor.journal,
        )
        with self.assertRaisesRegex(MODULE.ExecutorError, "rollback_unverified"):
            restarted.recover_incomplete()
        self.assertFalse(restarted.ready)

    def test_receipt_journal_failure_restores_runtime_before_failing(self) -> None:
        manifest, _ = self.write_manifest()
        runtime = FakeRuntime()
        executor = MODULE.NasPreviewRollbackExecutor(
            manifest=manifest,
            runtime=runtime,
            health=FakeHealth(),
            journal=FailSuccessJournal(self.state_dir),
        )

        with self.assertRaisesRegex(MODULE.ExecutorError, "execution_failed"):
            executor.execute(MODULE.RollbackRequest.parse(_request_payload()))

        self.assertEqual([DESIRED_IMAGES, OLD_IMAGES], runtime.apply_calls)
        self.assertEqual(OLD_IMAGES, runtime.current)
        terminal = executor.journal.read()[-1]
        self.assertEqual("failed", terminal["event"])
        self.assertEqual("verified", terminal["rollback_status"])

    def test_docker_apply_uses_exact_argv_allowlist_without_destructive_verbs(
        self,
    ) -> None:
        runner = CaptureCommandRunner()
        artifact = MODULE.RollbackArtifact.parse(_artifact_payload())
        controller = MODULE.DockerRuntimeController(runner, state_dir=self.state_dir)

        with mock.patch.object(
            MODULE,
            "_sha256_regular_file",
            side_effect=[
                artifact.compose_sha256,
                artifact.env_sha256,
                artifact.compose_sha256,
                artifact.env_sha256,
            ],
        ):
            controller.apply(artifact, DESIRED_IMAGES, "nas-g8-" + "a" * 24)

        operation, argv, timeout = next(
            call for call in runner.calls if call[0] == "compose_recreate"
        )
        self.assertEqual("compose_recreate", operation)
        self.assertEqual(300, timeout)
        self.assertIsInstance(argv, list)
        self.assertEqual(["api", "web"], argv[-2:])
        self.assertEqual(MODULE.TARGET_PROJECT, argv[argv.index("-p") + 1])
        self.assertIn("--no-build", argv)
        self.assertIn("--no-deps", argv)
        self.assertEqual("never", argv[argv.index("--pull") + 1])
        self.assertFalse(
            {"down", "stop", "rm", "prune", "volume", "network"} & set(argv)
        )
        self.assertFalse(
            any(";" in item or "&&" in item or "||" in item for item in argv)
        )
        self.assertEqual([], list(self.state_dir.glob("*.override.yml")))

    def test_resolved_compose_policy_rejects_privilege_and_bind_mounts(self) -> None:
        privileged = _resolved_config_payload()
        privileged["services"]["api"]["privileged"] = True
        with self.assertRaisesRegex(MODULE.CommandFailure, "compose_privilege_policy"):
            MODULE.DockerRuntimeController._validate_resolved_config(
                privileged,
                DESIRED_IMAGES,
            )

        bind_mount = _resolved_config_payload()
        bind_mount["services"]["api"]["volumes"] = [
            {
                "type": "bind",
                "source": "/var/run/docker.sock",
                "target": "/var/run/docker.sock",
            }
        ]
        with self.assertRaisesRegex(MODULE.CommandFailure, "compose_volume_policy"):
            MODULE.DockerRuntimeController._validate_resolved_config(
                bind_mount,
                DESIRED_IMAGES,
            )

    def test_resolved_compose_hash_mismatch_blocks_before_recreate(self) -> None:
        runner = CaptureCommandRunner()
        artifact = MODULE.RollbackArtifact.parse(
            _artifact_payload(resolved_config_sha256="f" * 64)
        )
        controller = MODULE.DockerRuntimeController(runner, state_dir=self.state_dir)
        with mock.patch.object(
            MODULE,
            "_sha256_regular_file",
            side_effect=[artifact.compose_sha256, artifact.env_sha256],
        ):
            with self.assertRaisesRegex(
                MODULE.CommandFailure,
                "resolved_config_sha256_mismatch",
            ):
                controller.apply(artifact, DESIRED_IMAGES, "nas-g8-" + "c" * 24)
        self.assertNotIn("compose_recreate", [call[0] for call in runner.calls])

    def test_subprocess_runner_explicitly_disables_shell(self) -> None:
        completed = subprocess.CompletedProcess(["docker", "version"], 0, "ok\n", "")
        with mock.patch.object(MODULE.subprocess, "run", return_value=completed) as run:
            result = MODULE.SubprocessRunner().run(
                "probe",
                ["docker", "version"],
                timeout=3,
            )
        self.assertEqual("ok\n", result.stdout)
        _, kwargs = run.call_args
        self.assertIs(kwargs["shell"], False)
        self.assertIsInstance(run.call_args.args[0], list)

    def test_journal_tampering_fails_closed(self) -> None:
        executor, _, _ = self.make_executor()
        executor.execute(MODULE.RollbackRequest.parse(_request_payload()))
        path = executor.journal.path
        raw = path.read_text(encoding="utf-8").replace(
            '"event":"executed"', '"event":"failed"'
        )
        path.write_text(raw, encoding="utf-8")
        with self.assertRaisesRegex(MODULE.ExecutorError, "journal_integrity_failure"):
            executor.journal.read()

    def test_journal_chain_is_authenticated_by_executor_secret(self) -> None:
        executor, _, _ = self.make_executor()
        executor.execute(MODULE.RollbackRequest.parse(_request_payload()))
        wrong_key_journal = MODULE.DurableJournal(
            self.state_dir,
            integrity_key=b"k" * 32,
        )
        with self.assertRaisesRegex(MODULE.ExecutorError, "journal_integrity_failure"):
            wrong_key_journal.read()

    def test_new_approval_manifest_reads_existing_journal_with_stable_key(self) -> None:
        executor, _, _ = self.make_executor()
        executor.execute(MODULE.RollbackRequest.parse(_request_payload()))
        second_payload = _manifest_payload(
            artifact_id="rollback-preview-release-v2",
            artifact_sha256="d" * 64,
            artifact_record_id="40000000-0000-0000-0000-000000000002",
            idempotency_key="10000000-0000-0000-0000-000000000002",
            approval_event_id="20000000-0000-0000-0000-000000000002",
            target_id="30000000-0000-0000-0000-000000000002",
            compose_file=(
                "/volume1/docker/vignette-preview-20260807/.rollback-executor/"
                "runtime/rollback-preview-release-v2/docker-compose.yml"
            ),
            env_file=(
                "/volume1/docker/vignette-preview-20260807/.rollback-executor/"
                "runtime/rollback-preview-release-v2/.env"
            ),
        )
        second_manifest, _ = self.write_manifest(second_payload)
        restarted = MODULE.NasPreviewRollbackExecutor(
            manifest=second_manifest,
            runtime=FakeRuntime(),
            health=FakeHealth(),
            journal=MODULE.DurableJournal(
                self.state_dir,
                integrity_key=JOURNAL_KEY,
            ),
        )

        restarted.recover_incomplete()

        self.assertEqual(3, len(restarted.journal.read()))


if __name__ == "__main__":
    unittest.main()