from __future__ import annotations import io import json import os import stat import subprocess import tarfile from pathlib import Path from types import SimpleNamespace from typing import Any import pytest from ops import gcp_leoclean_nosend_iap_remote as remote from ops import run_gcp_leoclean_nosend_iap as local def completed( argv: list[str], *, returncode: int = 0, stdout: bytes = b"", stderr: bytes = b"", ) -> subprocess.CompletedProcess[bytes]: return subprocess.CompletedProcess(argv, returncode, stdout, stderr) def install_execution_result( release: dict[str, Any], transition: dict[str, Any], ) -> dict[str, Any]: return { "schema": local.installer.INSTALL_EXECUTION_RESULT_SCHEMA, "status": "pass", "outcome": transition.get("outcome"), "release_sha256": release["release_sha256"], "persisted_receipt": { "schema": local.installer.INSTALL_RECEIPT_SCHEMA, "sha256": "9" * 64, }, "transition": transition, "verification": {}, } def private_directory(path: Path) -> Path: path.mkdir(mode=0o700) path.chmod(0o700) return path def fixed_commands(tmp_path: Path, operation: str = "install") -> tuple[list[str], list[str], list[str]]: request_id = "iap-abcdefghijkl" archive_name, key_name = local.remote_names(request_id) archive = tmp_path / archive_name ssh_key = tmp_path / key_name archive.write_bytes(b"archive") archive.chmod(0o600) return local.build_gcloud_argv( operation=operation, request_id=request_id, runner_revision="a" * 40, release_revision="b" * 40, archive=archive, archive_sha256="c" * 64, ssh_key=ssh_key, helper_sha256="d" * 64, ) def executable_plan(tmp_path: Path, operation: str) -> dict[str, Any]: request_id = "iap-abcdefghijkl" archive_name, key_name = local.remote_names(request_id) archive = tmp_path / archive_name ssh_key = tmp_path / key_name archive.write_bytes(b"archive") archive.chmod(0o600) archive_sha256 = local._sha256_bytes(archive.read_bytes()) helper_sha256 = "c" * 64 commands = local.build_gcloud_argv( operation=operation, request_id=request_id, runner_revision="a" * 40, release_revision="b" * 40, archive=archive, archive_sha256=archive_sha256, ssh_key=ssh_key, helper_sha256=helper_sha256, ) expected = { "release_artifact_sha256": "d" * 64, "release_sha256": "e" * 64, "image_digest": f"sha256:{'f' * 64}", "image_config_digest": f"sha256:{'1' * 64}", } return { "schema": local.PLAN_SCHEMA, "status": "pass", "mode": "dry_run", "operation": operation, "request_id": "iap-abcdefghijkl", "runner_revision": "a" * 40, "release_revision": "b" * 40, "target": { "project": local.PROJECT, "zone": local.ZONE, "instance": local.INSTANCE, "service": local.SERVICE, "service_account": local.SERVICE_ACCOUNT, }, "release": { "release_sha256": expected["release_sha256"], "image_reference": f"{local.package.IMAGE_REPOSITORY}@{expected['image_digest']}", "image_digest": expected["image_digest"], "image_config_digest": expected["image_config_digest"], }, "release_artifact": { "archive_sha256": expected["release_artifact_sha256"], "archive_size": 1, "members": [], }, "expected": expected, "request": { "archive": str(archive), "archive_sha256": archive_sha256, "helper_sha256": helper_sha256, "manifest_sha256": "d" * 64, "ssh_key": str(ssh_key), }, "actions": local._action_receipts(commands), "authority": { "gcp_mutation": operation in {"install", "rollback"}, "service_restart": operation in {"install", "rollback"}, "database_or_secret_provisioning": False, "production_or_transport": False, "vps": False, }, } def verifier_failure_result( plan: dict[str, Any], *, failure_stage: str = "process_environment", failure_reason: str = "process_environment_mismatch", ) -> dict[str, Any]: return { "schema": local.REMOTE_FAILURE_SCHEMA, "status": "fail", "operation": plan["operation"], "request_id": plan["request_id"], "runner_revision": plan["runner_revision"], "release_revision": plan["release_revision"], "release_sha256": plan["release"]["release_sha256"], "image_digest": plan["release"]["image_digest"], "image_config_digest": plan["release"]["image_config_digest"], "release_artifact_sha256": plan["release_artifact"]["archive_sha256"], "request_binding": { "archive_sha256": plan["request"]["archive_sha256"], "helper_sha256": plan["request"]["helper_sha256"], "manifest_sha256": plan["request"]["manifest_sha256"], }, "failure_stage": failure_stage, "failure_reason": failure_reason, "rollback": { "status": "pass", "mode": "installer_automatic", "executed": True, "restored_posture": "absent", "prior_posture_preserved": True, }, "authority": { "sending_enabled": False, "production_traffic_changed": False, "vps_touched": False, "canonical_proposal_approved": False, }, } def operational_failure_result(plan: dict[str, Any]) -> dict[str, Any]: result = verifier_failure_result( plan, failure_stage="database_before_restart", failure_reason="database_verification_failed", ) result["schema"] = local.REMOTE_OPERATION_FAILURE_SCHEMA result["rollback"] = { "status": "fail", "mode": "explicit", "executed": True, "restored_posture": "unknown", "prior_posture_preserved": False, } result["fail_closed"] = { "status": "pass", "service_posture": "inactive", "active_state": "inactive", "sub_state": "dead", "main_pid": 0, "n_restarts": 0, "candidate_container": "absent", "service_action_executed": True, } return result def successful_remote_result(plan: dict[str, Any]) -> dict[str, Any]: def verification(main_pid: int, invocation: str, container: str) -> dict[str, Any]: return { "active_state": "active", "sub_state": "running", "main_pid": main_pid, "n_restarts": 0, "invocation_id": invocation * 32, "supervising_process": {"status": "pass"}, "container_id": container * 64, "container_pid": main_pid + 1, "health": "healthy", "validated_process_environment_fields": ["PATH"], "dropins": "absent", "environment_files": "absent", } def database(name: str, receipt_hash: str) -> dict[str, Any]: return { "schema": "livingip.leocleanNoSendContainerDatabaseVerification.v1", "status": "pass", "receipt_sha256": receipt_hash * 64, "receipt_path": f"/var/lib/livingip/leoclean-nosend-iap/requests/{plan['request_id']}/{name}", "proof": { "proposal_staging": "function_only", "proposal_transaction": "rolled_back", "durable_proposal_created": False, "canonical_direct_writes": "denied", "stage_direct_writes": "denied", "role_escalation": "denied", }, } before = verification(101, "2", "3") if plan["operation"] == "install": after = verification(103, "5", "6") service_result = { "name": local.SERVICE, "installer_transition": { "outcome": "installed", "prior_posture": "absent", "prior_n_restarts": 0, "final_posture": "active", "service_action": "restart", "managed_files_replaced": True, }, "before_restart": before, "after_restart": after, } database_result = { "status": "pass", "before_restart": database("database-before-restart.json", "4"), "after_restart": database("database-after-restart.json", "7"), } else: service_result = {"name": local.SERVICE, "verification": before} database_result = database("database-verify.json", "4") return { "schema": local.REMOTE_RESULT_SCHEMA, "status": "pass", "operation": plan["operation"], "request_id": plan["request_id"], "runner_revision": plan["runner_revision"], "release_revision": plan["release_revision"], "release_sha256": plan["release"]["release_sha256"], "image_digest": plan["release"]["image_digest"], "image_config_digest": plan["release"]["image_config_digest"], "release_artifact_sha256": plan["release_artifact"]["archive_sha256"], "request_binding": { "archive_sha256": plan["request"]["archive_sha256"], "helper_sha256": plan["request"]["helper_sha256"], "manifest_sha256": plan["request"]["manifest_sha256"], }, "service": service_result, "database": database_result, "authority": { "sending_enabled": False, "production_traffic_changed": False, "vps_touched": False, "canonical_proposal_approved": False, }, } def test_gcloud_commands_are_fixed_iap_only_and_expire_keys(tmp_path: Path) -> None: scp, ssh, cleanup = fixed_commands(tmp_path) assert scp[:3] == [local.GCLOUD, "compute", "scp"] assert ssh[:4] == [local.GCLOUD, "compute", "ssh", local.INSTANCE] for command in (scp, ssh, cleanup): assert f"--project={local.PROJECT}" in command assert f"--zone={local.ZONE}" in command assert "--tunnel-through-iap" in command assert "--ssh-key-expire-after=5m" in command encoded = json.dumps({"scp": scp, "ssh": ssh, "cleanup": cleanup}) assert local.INSTANCE in encoded assert local.SERVICE not in encoded assert "teleo-prod-1" not in encoded assert "77.42.65.182" not in encoded assert "telegram" not in encoded.casefold() assert "sudo -- /usr/bin/python3 -I -c" in ssh[-1] assert "/tmp/livingip-leoclean-nosend-remote" not in ssh[-1] def test_embedded_root_bootstrap_compiles() -> None: compile(local.BOOTSTRAP_CODE, "", "exec") def test_remote_and_local_post_install_failure_contracts_match() -> None: assert remote.POST_INSTALL_FAILURES == local.POST_INSTALL_FAILURES def test_live_origin_main_readback_is_exact() -> None: revision = "a" * 40 responses = iter( [ f"{local.ORIGIN_URL}\n".encode(), f"{revision}\trefs/heads/main\n".encode(), ] ) def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: return completed(argv, stdout=next(responses)) local.validate_remote_main(revision, runner=runner) @pytest.mark.parametrize( "responses", [ [b"git@github.com:someone-else/teleo-infrastructure.git\n"], [ f"{local.ORIGIN_URL}\n".encode(), f"{'b' * 40}\trefs/heads/main\n".encode(), ], ], ) def test_live_origin_main_readback_rejects_different_authority(responses: list[bytes]) -> None: output = iter(responses) def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: return completed(argv, stdout=next(output)) with pytest.raises(local.IapRequestError): local.validate_remote_main("a" * 40, runner=runner) @pytest.mark.parametrize("value", ["main", "a" * 39, "A" * 40, "a" * 41, "a" * 39 + ";"]) def test_revision_input_is_exact(value: str, tmp_path: Path) -> None: archive_name, key_name = local.remote_names("iap-abcdefghijkl") with pytest.raises(local.IapRequestError, match="revision"): local.build_gcloud_argv( operation="install", request_id="iap-abcdefghijkl", runner_revision=value, release_revision="b" * 40, archive=tmp_path / archive_name, archive_sha256="b" * 64, ssh_key=tmp_path / key_name, helper_sha256="c" * 64, ) @pytest.mark.parametrize("value", ["iap-short", "iap-abcdefghijkl;id", "../iap-abcdefghijkl", "iap-ABCDEFGHIJKL"]) def test_request_id_rejects_shell_and_path_fragments(value: str) -> None: with pytest.raises(local.IapRequestError, match="request ID"): local.remote_names(value) def test_mutable_release_reference_is_rejected(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: release = { "image": {"reference": f"{local.package.IMAGE_REPOSITORY}:latest"}, "image_input": {"runtime": {"teleo_git_head": "a" * 40}}, } monkeypatch.setattr(local.installer, "validate_release_bundle", lambda _bundle: (release, b"unit")) with pytest.raises(local.IapRequestError, match="immutable"): local.validate_release( tmp_path, "a" * 40, expected_release_sha256="b" * 64, expected_image_digest=f"sha256:{'c' * 64}", expected_image_config_digest=f"sha256:{'d' * 64}", ) def test_stable_reader_rejects_symlink(tmp_path: Path) -> None: root = private_directory(tmp_path / "root") target = root / "target" target.write_text("safe", encoding="utf-8") alias = root / "alias" alias.symlink_to(target) with pytest.raises(local.IapRequestError, match="unsafe"): local._read_stable_file(alias.absolute(), root=root.absolute()) def archive_manifest(payloads: dict[str, tuple[bytes, int]]) -> dict[str, Any]: files = [ {"path": path, "sha256": local._sha256_bytes(content), "size": len(content), "mode": mode} for path, (content, mode) in sorted(payloads.items()) ] stable = { "schema": remote.SCHEMA, "operation": "verify", "request_id": "iap-abcdefghijkl", "runner_revision": "a" * 40, "release_revision": "b" * 40, "target": { "project": remote.PROJECT, "zone": remote.ZONE, "instance": remote.INSTANCE, "service": remote.SERVICE, "service_account": remote.SERVICE_ACCOUNT, }, "release": { "release_sha256": "b" * 64, "image_reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'c' * 64}", "image_digest": f"sha256:{'c' * 64}", "image_config_digest": f"sha256:{'d' * 64}", }, "installer_source": {"revision": "a" * 40, "files": [], "sha256": "e" * 64}, "release_artifact": {"archive_sha256": "f" * 64, "archive_size": 1, "members": []}, "files": files, } return {**stable, "manifest_sha256": remote._canonical_sha256(stable)} def exact_payloads() -> dict[str, tuple[bytes, int]]: return {name: (f"payload:{name}\n".encode(), 0o644) for name in remote.EXPECTED_MEMBERS - {"request-manifest.json"}} def test_archive_is_deterministic_and_extracts_exact_members(tmp_path: Path) -> None: first_root = private_directory(tmp_path / "first") second_root = private_directory(tmp_path / "second") extracted = private_directory(tmp_path / "extracted") payloads = exact_payloads() manifest = archive_manifest(payloads) first_hash = local.build_archive(first_root / "request.tar.gz", manifest=manifest, payloads=payloads) second_hash = local.build_archive(second_root / "request.tar.gz", manifest=manifest, payloads=payloads) observed = remote.extract_archive(first_root / "request.tar.gz", extracted) assert first_hash == second_hash assert (first_root / "request.tar.gz").read_bytes() == (second_root / "request.tar.gz").read_bytes() assert observed == manifest assert {str(path.relative_to(extracted)) for path in extracted.rglob("*") if path.is_file()} == ( remote.EXPECTED_MEMBERS - {"request-manifest.json"} ) def test_archive_extraction_preserves_manifest_modes_under_restrictive_umask(tmp_path: Path) -> None: archive_root = private_directory(tmp_path / "archive") extracted = private_directory(tmp_path / "extracted") allowed_modes = (0o600, 0o644, 0o755) payloads = { name: (f"payload:{name}\n".encode(), allowed_modes[index % len(allowed_modes)]) for index, name in enumerate(sorted(remote.EXPECTED_MEMBERS - {"request-manifest.json"})) } manifest = archive_manifest(payloads) archive = archive_root / "request.tar.gz" local.build_archive(archive, manifest=manifest, payloads=payloads) previous_umask = os.umask(0o077) try: remote.extract_archive(archive, extracted) finally: os.umask(previous_umask) assert { str(path.relative_to(extracted)): stat.S_IMODE(path.stat().st_mode) for path in extracted.rglob("*") if path.is_file() } == {name: mode for name, (_content, mode) in payloads.items()} def write_unsafe_archive(path: Path, *, kind: str) -> None: with tarfile.open(path, "w:gz") as archive: for index, name in enumerate(sorted(remote.EXPECTED_MEMBERS)): info = tarfile.TarInfo(name) info.uid = 0 info.gid = 0 info.uname = "root" info.gname = "root" info.mtime = 0 info.mode = 0o600 content = b"{}" if kind == "link" and index == 0: info.type = tarfile.SYMTYPE info.linkname = "/etc/passwd" info.size = 0 archive.addfile(info) continue if kind == "oversize" and index == 0: content = b"x" * (remote.MAX_MEMBER_BYTES + 1) info.size = len(content) archive.addfile(info, io.BytesIO(content)) if kind == "duplicate": info = tarfile.TarInfo("request-manifest.json") info.size = 2 info.mode = 0o600 info.uid = info.gid = info.mtime = 0 info.uname = info.gname = "root" archive.addfile(info, io.BytesIO(b"{}")) @pytest.mark.parametrize("kind", ["link", "oversize", "duplicate"]) def test_archive_rejects_links_oversize_and_duplicate_members(tmp_path: Path, kind: str) -> None: path = tmp_path / f"{kind}.tar.gz" destination = private_directory(tmp_path / kind) write_unsafe_archive(path, kind=kind) with pytest.raises(remote.RemoteError): remote.extract_archive(path, destination) def test_archive_output_requires_private_directory(tmp_path: Path) -> None: output = tmp_path / "public" output.mkdir(mode=0o755) output.chmod(0o755) with pytest.raises(local.IapRequestError, match="private"): local.build_archive(output / "request.tar.gz", manifest={}, payloads={}) @pytest.mark.parametrize( ("field", "value"), [ ("expected_artifact_sha256", "not-a-hash"), ("expected_release_sha256", "a" * 63), ("expected_image_digest", "latest"), ("expected_image_config_digest", "sha256:" + "A" * 64), ], ) def test_expected_release_inputs_are_exact(field: str, value: str) -> None: arguments = { "expected_artifact_sha256": "a" * 64, "expected_release_sha256": "b" * 64, "expected_image_digest": f"sha256:{'c' * 64}", "expected_image_config_digest": f"sha256:{'d' * 64}", } arguments[field] = value with pytest.raises(local.IapRequestError, match="expected"): local._validate_expected_bindings(**arguments) def test_release_and_original_artifact_require_independent_expected_values( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: release = { "release_sha256": "a" * 64, "image": { "reference": f"{local.package.IMAGE_REPOSITORY}@sha256:{'b' * 64}", "digest": f"sha256:{'b' * 64}", "config_digest": f"sha256:{'c' * 64}", }, "image_input": {"runtime": {"teleo_git_head": "d" * 40}}, } monkeypatch.setattr(local.installer, "validate_release_bundle", lambda _bundle: (release, b"unit")) with pytest.raises(local.IapRequestError, match="independently supplied"): local.validate_release( tmp_path, "d" * 40, expected_release_sha256="e" * 64, expected_image_digest=f"sha256:{'b' * 64}", expected_image_config_digest=f"sha256:{'c' * 64}", ) artifact = tmp_path / "artifact.zip" artifact.write_bytes(b"not-the-approved-artifact") artifact.chmod(0o600) with pytest.raises(local.IapRequestError, match="independently supplied"): local.validate_release_artifact( artifact, bundle=tmp_path, release=release, expected_artifact_sha256="f" * 64, ) def test_dry_run_plan_records_only_bounded_action_hashes(tmp_path: Path) -> None: plan = executable_plan(tmp_path, "install") encoded = json.dumps(plan, sort_keys=True) assert [item["label"] for item in plan["actions"]] == [ "upload_request", "execute_remote_request", "cleanup_failed_transfer", ] assert all(set(item) == {"label", "argv_sha256"} for item in plan["actions"]) assert local.BOOTSTRAP_CODE not in encoded assert "--command=" not in encoded assert "sudo --" not in encoded def test_gcloud_environment_selects_absolute_compatible_python( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: home = private_directory(tmp_path / "home") config = private_directory(home / "gcloud") python = tmp_path / "python311" python.write_text("placeholder", encoding="utf-8") python.chmod(0o700) monkeypatch.setenv("HOME", str(home)) monkeypatch.setenv("CLOUDSDK_CONFIG", str(config)) monkeypatch.setenv("CLOUDSDK_PYTHON", str(python)) environment = local.gcloud_environment( probe_runner=lambda argv, **_kwargs: completed(argv, stdout=b"Python 3.11.9\n") ) assert environment["CLOUDSDK_PYTHON"] == str(python.resolve()) assert environment["CLOUDSDK_CONFIG"] == str(config.resolve()) def test_local_result_receipt_is_private_and_self_hashed(tmp_path: Path) -> None: plan = executable_plan(tmp_path, "verify") result = {"status": "pass", "bounded": "remote-result"} path, receipt = local._write_result_receipt(plan, result) stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"} assert stat.S_IMODE(path.stat().st_mode) == 0o600 assert receipt["receipt_sha256"] == local.package.canonical_sha256(stable) assert json.loads(path.read_text(encoding="utf-8")) == receipt def test_execute_success_returns_private_receipt( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: plan = executable_plan(tmp_path, "install") remote_result = successful_remote_result(plan) monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None) calls = 0 def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: nonlocal calls calls += 1 if calls == 1: return completed(argv) return completed(argv, stdout=json.dumps(remote_result).encode()) execution = local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"}) receipt_path = Path(execution["receipt_path"]) receipt = json.loads(receipt_path.read_text(encoding="utf-8")) assert calls == 2 assert execution["status"] == "pass" assert stat.S_IMODE(receipt_path.stat().st_mode) == 0o600 assert receipt["remote_result"] == remote_result assert receipt["receipt_sha256"] == execution["receipt_sha256"] @pytest.mark.parametrize("remote_returncode", [0, 65], ids=["success", "failure"]) def test_execute_remote_without_output_records_absent_receipt( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, remote_returncode: int, ) -> None: plan = executable_plan(tmp_path, "verify") monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None) calls = 0 def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: nonlocal calls calls += 1 return completed(argv, returncode=remote_returncode if calls == 2 else 0) with pytest.raises(local.IapRequestError, match="remote_receipt/absent"): local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"}) receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json" receipt = json.loads(receipt_path.read_text(encoding="utf-8")) assert calls == 3 assert receipt["schema"] == local.LOCAL_FAILURE_RECEIPT_SCHEMA assert (receipt["failure_stage"], receipt["failure_reason"]) == ("remote_receipt", "absent") assert receipt["cleanup"] == {"status": "pass", "mode": "failed_transfer"} assert receipt["remote_state"] == "unknown" def test_execute_transport_interruption_records_bounded_failure( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: plan = executable_plan(tmp_path, "verify") monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None) calls = 0 def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: nonlocal calls calls += 1 if calls == 2: raise subprocess.TimeoutExpired(argv, 300, output=b"access_token=hidden") return completed(argv) with pytest.raises(local.IapRequestError, match="iap_transport/remote_command_interrupted") as caught: local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"}) receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json" receipt_text = receipt_path.read_text(encoding="utf-8") receipt = json.loads(receipt_text) assert calls == 3 assert (receipt["failure_stage"], receipt["failure_reason"]) == ( "iap_transport", "remote_command_interrupted", ) assert "hidden" not in receipt_text assert "access_token" not in receipt_text assert "hidden" not in str(caught.value) @pytest.mark.parametrize( "payload", [ b"{not-json", (b"[" * 1100) + b"0" + (b"]" * 1100), ], ids=["invalid-json", "recursive-json"], ) def test_execute_malformed_remote_receipt_fails_closed( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, payload: bytes, ) -> None: plan = executable_plan(tmp_path, "verify") monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None) calls = 0 def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: nonlocal calls calls += 1 if calls == 2: return completed(argv, stdout=payload) return completed(argv) with pytest.raises(local.IapRequestError, match="remote_receipt/malformed"): local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"}) receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json" receipt = json.loads(receipt_path.read_text(encoding="utf-8")) assert calls == 3 assert (receipt["failure_stage"], receipt["failure_reason"]) == ("remote_receipt", "malformed") assert receipt["remote_state"] == "unknown" def test_execute_records_sanitized_verifier_failure( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: plan = executable_plan(tmp_path, "install") failure = verifier_failure_result(plan) monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None) calls = 0 def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: nonlocal calls calls += 1 if calls == 1: return completed(argv) return completed( argv, returncode=65, stdout=json.dumps(failure).encode(), stderr=b'{"error":"remote_operation_failed","status":"fail"}', ) with pytest.raises(local.IapRequestError, match="process_environment/process_environment_mismatch") as caught: local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"}) receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json" receipt = json.loads(receipt_path.read_text(encoding="utf-8")) assert calls == 2 assert stat.S_IMODE(receipt_path.stat().st_mode) == 0o600 assert receipt["status"] == "fail" assert receipt["remote_result"] == failure assert "remote_operation_failed" not in str(caught.value) def test_execute_records_operational_failure_and_inert_state( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: plan = executable_plan(tmp_path, "install") failure = operational_failure_result(plan) monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None) calls = 0 def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: nonlocal calls calls += 1 if calls == 1: return completed(argv) return completed(argv, returncode=65, stdout=json.dumps(failure).encode()) with pytest.raises( local.IapRequestError, match="database_before_restart/database_verification_failed", ): local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"}) receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json" receipt = json.loads(receipt_path.read_text(encoding="utf-8")) assert calls == 2 assert receipt["remote_result"] == failure assert receipt["remote_result"]["fail_closed"]["service_posture"] == "inactive" def test_operational_failure_persists_and_propagates_to_local_receipt( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str], ) -> None: plan = executable_plan(tmp_path, "install") failure = operational_failure_result(plan) request_dir = private_directory(tmp_path / "remote-request") payload_root = private_directory(tmp_path / "remote-payload") def fail_execute(*_args: Any, **_kwargs: Any) -> dict[str, Any]: raise remote.RemoteOperationFailure(failure) monkeypatch.setattr(remote, "execute_request", fail_execute) persisted = remote._execute_and_persist_result( plan, payload_root, request_dir, plan["request"]["archive_sha256"], plan["request"]["helper_sha256"], runner=lambda argv, **_kwargs: completed(argv), ) remote_receipt_path = request_dir / "result.json" assert persisted == failure assert stat.S_IMODE(remote_receipt_path.stat().st_mode) == 0o600 assert json.loads(remote_receipt_path.read_text(encoding="utf-8")) == failure monkeypatch.setattr(remote, "receive_request", lambda *_args: persisted) previous_umask = os.umask(0o077) try: returncode = remote.main(["install", "request", "runner", "release", "archive", "helper"]) finally: os.umask(previous_umask) captured = capsys.readouterr() assert returncode == 65 assert captured.err == "" assert json.loads(captured.out) == failure monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None) calls = 0 def local_runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: nonlocal calls calls += 1 if calls == 1: return completed(argv) return completed(argv, returncode=65, stdout=captured.out.encode()) with pytest.raises(local.IapRequestError, match="database_before_restart/database_verification_failed"): local.execute_plan(plan, runner=local_runner, gcloud_env={"PATH": "/usr/bin"}) local_receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json" local_receipt = json.loads(local_receipt_path.read_text(encoding="utf-8")) stable = {key: value for key, value in local_receipt.items() if key != "receipt_sha256"} assert local_receipt["schema"] == local.FAILURE_RECEIPT_SCHEMA assert local_receipt["remote_result"] == failure assert local_receipt["receipt_sha256"] == local.package.canonical_sha256(stable) def test_remote_failure_rejects_unallowlisted_reason(tmp_path: Path) -> None: plan = executable_plan(tmp_path, "install") failure = verifier_failure_result(plan) failure["failure_reason"] = "password=not-allowlisted" with pytest.raises(local.IapRequestError, match="did not match"): local.validate_remote_failure(plan, failure) def test_operational_failure_rejects_extra_rollback_fields(tmp_path: Path) -> None: plan = executable_plan(tmp_path, "install") failure = operational_failure_result(plan) failure["rollback"]["credential"] = "not-allowed" with pytest.raises(local.IapRequestError, match="rollback summary was invalid"): local.validate_remote_failure(plan, failure) @pytest.mark.parametrize( "rollback", [ { "status": "not_run", "mode": "not_applicable", "executed": False, "restored_posture": "unchanged", "prior_posture_preserved": True, }, { "status": "pass", "mode": "installer_automatic", "executed": True, "restored_posture": "inactive", "prior_posture_preserved": True, }, { "status": "pass", "mode": "posture_recovery", "executed": False, "restored_posture": "absent", "prior_posture_preserved": True, }, ], ) def test_operational_failure_rejects_impossible_rollback_modes( tmp_path: Path, rollback: dict[str, Any], ) -> None: plan = executable_plan(tmp_path, "install") failure = operational_failure_result(plan) failure["rollback"] = rollback with pytest.raises(local.IapRequestError, match=r"remote failure did not match|rollback summary was invalid"): local.validate_remote_failure(plan, failure) def test_verify_failure_rejects_mutating_rollback_summary(tmp_path: Path) -> None: plan = executable_plan(tmp_path, "verify") failure = verifier_failure_result(plan) with pytest.raises(local.IapRequestError, match="remote failure did not match"): local.validate_remote_failure(plan, failure) failure["rollback"] = { "status": "not_run", "mode": "not_applicable", "executed": False, "restored_posture": "unchanged", "prior_posture_preserved": True, } local.validate_remote_failure(plan, failure) def test_remote_main_emits_structured_failure_on_stdout( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str], ) -> None: failure = verifier_failure_result(executable_plan(tmp_path, "install")) monkeypatch.setattr(remote, "receive_request", lambda *_args: failure) previous_umask = os.umask(0o077) try: returncode = remote.main(["install", "request", "runner", "release", "archive", "helper"]) finally: os.umask(previous_umask) captured = capsys.readouterr() assert returncode == 65 assert json.loads(captured.out) == failure assert captured.err == "" def test_execute_failure_does_not_forward_command_output(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: plan = executable_plan(tmp_path, "install") monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None) def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: return completed(argv, returncode=1, stderr=b"password=extremely-secret") with pytest.raises(local.IapRequestError) as caught: local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"}) receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json" receipt_text = receipt_path.read_text(encoding="utf-8") receipt = json.loads(receipt_text) assert "extremely-secret" not in str(caught.value) assert "password" not in str(caught.value).casefold() assert "extremely-secret" not in receipt_text assert "password" not in receipt_text.casefold() assert receipt["cleanup"] == {"status": "fail", "mode": "failed_transfer"} assert "authority" not in receipt assert receipt["requested_authority"] == plan["authority"] def test_execute_rejects_unsanitized_remote_result(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: plan = executable_plan(tmp_path, "verify") monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None) calls = 0 def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: nonlocal calls calls += 1 if calls == 1: return completed(argv) payload = { "schema": local.REMOTE_RESULT_SCHEMA, "status": "pass", "operation": "verify", "request_id": "iap-abcdefghijkl", "access_token": "not-allowed", } return completed(argv, stdout=json.dumps(payload).encode()) with pytest.raises(local.IapRequestError) as caught: local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"}) receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json" receipt_text = receipt_path.read_text(encoding="utf-8") receipt = json.loads(receipt_text) assert "not-allowed" not in str(caught.value) assert "not-allowed" not in receipt_text assert (receipt["failure_stage"], receipt["failure_reason"]) == ( "remote_receipt", "contract_mismatch", ) assert receipt["cleanup"] == {"status": "pass", "mode": "failed_transfer"} def test_execute_rederives_commands_and_rejects_plan_tampering(tmp_path: Path) -> None: plan = executable_plan(tmp_path, "verify") plan["actions"][1]["argv_sha256"] = "0" * 64 with pytest.raises(local.IapRequestError, match="fixed IAP plan"): local.execute_plan(plan, runner=lambda argv, **_kwargs: completed(argv)) def test_execute_rejects_authority_tampering_before_dispatch(tmp_path: Path) -> None: plan = executable_plan(tmp_path, "verify") plan["authority"]["credential"] = "not-allowed" calls: list[list[str]] = [] with pytest.raises(local.IapRequestError, match="request authority was not exact") as caught: local.execute_plan( plan, runner=lambda argv, **_kwargs: (calls.append(argv) or completed(argv)), ) assert calls == [] assert "not-allowed" not in str(caught.value) assert not (tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json").exists() @pytest.mark.parametrize("fail_pull", [False, True]) def test_ephemeral_registry_credentials_are_always_removed(tmp_path: Path, fail_pull: bool) -> None: calls: list[tuple[list[str], bytes | None]] = [] def runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[bytes]: input_bytes = kwargs.get("input") calls.append((argv, input_bytes)) if argv[0] == "/usr/bin/curl": return completed(argv, stdout=b'{"access_token":"abcdefghijklmnop","token_type":"Bearer"}') if "pull" in argv and fail_pull: return completed(argv, returncode=1, stderr=b"registry said secret") return completed(argv) reference = f"{remote.IMAGE_REPOSITORY}@sha256:{'a' * 64}" if fail_pull: with pytest.raises(remote.RemoteError, match="immutable image pull"): remote.pull_immutable_image(reference, tmp_path, runner=runner) else: remote.pull_immutable_image(reference, tmp_path, runner=runner) assert not any(path.name.startswith("docker-auth-") for path in tmp_path.iterdir()) assert all("abcdefghijklmnop" not in " ".join(argv) for argv, _input in calls) assert any(input_bytes == b"abcdefghijklmnop\n" for _argv, input_bytes in calls) assert any("logout" in argv for argv, _input in calls) def test_metadata_target_binds_project_zone_instance_and_service_account() -> None: values = { "instance/service-accounts/default/email": remote.SERVICE_ACCOUNT, "project/project-id": remote.PROJECT, "instance/name": remote.INSTANCE, "instance/zone": f"projects/123/zones/{remote.ZONE}", } def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: path = argv[-1].removeprefix(f"{remote.METADATA}/") return completed(argv, stdout=values[path].encode()) remote.verify_service_account(runner=runner) values["instance/name"] = "wrong-instance" with pytest.raises(remote.RemoteError, match="target or service account"): remote.verify_service_account(runner=runner) def test_outer_lock_and_reused_request_id_fail_closed(tmp_path: Path) -> None: lock = tmp_path / "runner.lock" first = remote._acquire_lock(lock, expected_uid=os.geteuid()) try: with pytest.raises(remote.RemoteError, match="another request"): remote._acquire_lock(lock, expected_uid=os.geteuid()) finally: os.close(first) root = tmp_path / "requests" claimed = remote._claim_request_directory( root, "iap-abcdefghijkl", expected_uid=os.geteuid(), trusted_parent=tmp_path, ) assert stat.S_IMODE(claimed.stat().st_mode) == 0o700 with pytest.raises(remote.RemoteError, match="already used"): remote._claim_request_directory( root, "iap-abcdefghijkl", expected_uid=os.geteuid(), trusted_parent=tmp_path, ) def test_rollback_uses_only_explicit_receipt_bound_interface(tmp_path: Path) -> None: observed: list[dict[str, Any]] = [] def rollback(release_sha256: str, **kwargs: Any) -> dict[str, Any]: observed.append({"release_sha256": release_sha256, **kwargs}) if kwargs["execute_rollback"] is False: return { "schema": "rollback", "status": "pass", "mode": "dry_run", "expected_release_sha256": release_sha256, } return {"schema": "rollback", "status": "pass"} installer = SimpleNamespace(rollback_release=rollback) result = remote._execute_explicit_rollback( installer_module=installer, release_sha256="a" * 64, paths=object(), source_binding={"revision": "b" * 40}, runner=lambda argv, **_kwargs: completed(argv), ) assert result["status"] == "pass" assert [call["release_sha256"] for call in observed] == ["a" * 64, "a" * 64] assert [call["execute_rollback"] for call in observed] == [False, True] assert all("install_release" not in call for call in observed) def test_rollback_fails_when_explicit_interface_is_absent(tmp_path: Path) -> None: with pytest.raises(remote.RemoteError, match="lacked the explicit rollback"): remote._execute_explicit_rollback( installer_module=SimpleNamespace(), release_sha256="a" * 64, paths=object(), source_binding={}, runner=lambda argv, **_kwargs: completed(argv), ) def test_explicit_rollback_preflight_failure_reports_no_execution() -> None: calls: list[bool] = [] def rollback(_release_sha256: str, **kwargs: Any) -> dict[str, Any]: calls.append(kwargs["execute_rollback"]) return {"schema": "rollback", "status": "fail", "mode": "dry_run"} with pytest.raises(remote.ExplicitRollbackFailure) as caught: remote._execute_explicit_rollback( installer_module=SimpleNamespace(rollback_release=rollback), release_sha256="a" * 64, paths=object(), source_binding={"revision": "b" * 40}, runner=lambda argv, **_kwargs: completed(argv), ) assert caught.value.executed is False assert calls == [False] def test_install_verifier_failure_returns_bounded_automatic_rollback_result( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: release = { "release_sha256": "a" * 64, "image": { "reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'b' * 64}", "digest": f"sha256:{'b' * 64}", "config_digest": f"sha256:{'c' * 64}", }, "image_input": {"runtime": {"teleo_git_head": "d" * 40}}, } manifest = { "operation": "install", "request_id": "iap-abcdefghijkl", "runner_revision": "e" * 40, "release_revision": "d" * 40, "manifest_sha256": "f" * 64, "release": { "release_sha256": release["release_sha256"], "image_reference": release["image"]["reference"], "image_digest": release["image"]["digest"], "image_config_digest": release["image"]["config_digest"], }, "release_artifact": {"archive_sha256": "1" * 64}, "installer_source": {"revision": "e" * 40}, } class Paths: @staticmethod def under(_root: Path) -> object: return object() def fail_install(*_args: Any, **_kwargs: Any) -> dict[str, Any]: failure = local.installer.VerificationError( "process_environment", "process_environment_mismatch", ) failure.rollback_executed = True failure.restored_posture = "absent" failure.prior_posture_preserved = True raise failure installer_module = SimpleNamespace( VerificationError=local.installer.VerificationError, VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES, InstallPaths=Paths, validate_release_bundle=lambda _bundle: (release, b"unit"), _validate_source_binding=lambda _binding: None, install_release=fail_install, ) package_module = SimpleNamespace(render_systemd_unit=lambda _release: "unit") monkeypatch.setattr(remote, "validate_original_release_artifact", lambda *_args, **_kwargs: None) monkeypatch.setattr(remote, "_load_contract_modules", lambda _root: (package_module, installer_module)) monkeypatch.setattr(remote, "pull_immutable_image", lambda *_args, **_kwargs: None) with pytest.raises(remote.RemoteOperationFailure) as caught: remote.execute_request( manifest, tmp_path, tmp_path, "6" * 64, "7" * 64, runner=lambda argv, **_kwargs: completed(argv), ) result = caught.value.result assert result["failure_stage"] == "process_environment" assert result["failure_reason"] == "process_environment_mismatch" assert result["rollback"] == { "status": "pass", "mode": "installer_automatic", "executed": True, "restored_posture": "absent", "prior_posture_preserved": True, } assert result["authority"]["sending_enabled"] is False @pytest.mark.parametrize("failure_point", ["first_database", "restart", "second_database"]) def test_post_install_proof_failure_executes_explicit_rollback( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, failure_point: str, ) -> None: release = { "release_sha256": "a" * 64, "image": { "reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'b' * 64}", "digest": f"sha256:{'b' * 64}", "config_digest": f"sha256:{'c' * 64}", }, "image_input": {"runtime": {"teleo_git_head": "d" * 40}}, } manifest = { "operation": "install", "request_id": "iap-abcdefghijkl", "runner_revision": "e" * 40, "release_revision": "d" * 40, "manifest_sha256": "f" * 64, "release": { "release_sha256": release["release_sha256"], "image_reference": release["image"]["reference"], "image_digest": release["image"]["digest"], "image_config_digest": release["image"]["config_digest"], }, "release_artifact": {"archive_sha256": "1" * 64}, "installer_source": {"revision": "e" * 40}, } verifications = iter( [ { "main_pid": 10, "invocation_id": "2" * 32, "container_id": "3" * 64, "n_restarts": 0, }, { "main_pid": 11, "invocation_id": "4" * 32, "container_id": "5" * 64, "n_restarts": 0, }, ] ) rollbacks: list[str] = [] class Paths: @staticmethod def under(_root: Path) -> object: return object() def rollback_release(release_sha256: str, **kwargs: Any) -> dict[str, Any]: if kwargs["execute_rollback"] is False: return { "schema": "livingip.leocleanNoSendRollbackResult.v1", "status": "pass", "mode": "dry_run", "expected_release_sha256": release_sha256, } rollbacks.append(release_sha256) return { "schema": "livingip.leocleanNoSendRollbackResult.v1", "status": "pass", "mode": "executed", "rolled_back_release_sha256": release_sha256, "restored_release_sha256": None, "restored_service_posture": "absent", "verification": None, } installer_module = SimpleNamespace( SYSTEMCTL="/usr/bin/systemctl", VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES, INSTALL_EXECUTION_RESULT_SCHEMA=local.installer.INSTALL_EXECUTION_RESULT_SCHEMA, LEGACY_INSTALL_RECEIPT_SCHEMA=local.installer.LEGACY_INSTALL_RECEIPT_SCHEMA, INSTALL_RECEIPT_SCHEMA=local.installer.INSTALL_RECEIPT_SCHEMA, InstallPaths=Paths, validate_release_bundle=lambda _bundle: (release, b"unit"), _validate_source_binding=lambda _binding: None, install_release=lambda *_args, **_kwargs: install_execution_result( release, { "outcome": "installed", "prior_posture": "absent", "prior_n_restarts": 0, "final_posture": "active", "service_action": "restart", "managed_files_replaced": True, }, ), verify_running_release=lambda *_args, **_kwargs: next(verifications), rollback_release=rollback_release, ) package_module = SimpleNamespace( render_systemd_unit=lambda _release: "unit", canonical_sha256=local.package.canonical_sha256, ) monkeypatch.setattr(remote, "validate_original_release_artifact", lambda *_args, **_kwargs: None) monkeypatch.setattr(remote, "_load_contract_modules", lambda _root: (package_module, installer_module)) monkeypatch.setattr(remote, "pull_immutable_image", lambda *_args, **_kwargs: None) database_calls = 0 def verify_database(**_kwargs: Any) -> dict[str, Any]: nonlocal database_calls database_calls += 1 if failure_point == "first_database" and database_calls == 1: raise remote.RemoteError("database failed") if failure_point == "second_database" and database_calls == 2: raise remote.RemoteError("database failed") return {"status": "pass"} monkeypatch.setattr(remote, "_verify_database", verify_database) def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: if failure_point == "restart" and argv[:2] == ["/usr/bin/systemctl", "restart"]: return completed(argv, returncode=1) return completed(argv) with pytest.raises(remote.RemoteOperationFailure) as caught: remote.execute_request( manifest, tmp_path, tmp_path, "6" * 64, "7" * 64, runner=runner, ) assert rollbacks == [release["release_sha256"]] expected_failure = { "first_database": ("database_before_restart", "database_verification_failed"), "restart": ("service_restart", "service_restart_failed"), "second_database": ("database_after_restart", "database_verification_failed"), }[failure_point] result = caught.value.result assert result["schema"] == remote.OPERATION_FAILURE_RESULT_SCHEMA assert (result["failure_stage"], result["failure_reason"]) == expected_failure assert result["rollback"] == { "status": "pass", "mode": "explicit", "executed": True, "restored_posture": "absent", "prior_posture_preserved": True, } assert result["fail_closed"] == { "status": "not_required", "service_posture": "absent", "active_state": "not_observed", "sub_state": "not_observed", "main_pid": None, "n_restarts": None, "candidate_container": "not_observed", "service_action_executed": False, } @pytest.mark.parametrize( ("outcome", "prior_posture", "service_action", "expected_actions", "expected_executed"), [ ("verified_noop", "active", "none", [], False), ("controlled_start", "inactive", "start", ["stop"], True), ], ) def test_post_install_failure_preserves_exact_release_posture_without_consuming_history( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, outcome: str, prior_posture: str, service_action: str, expected_actions: list[str], expected_executed: bool, ) -> None: release = { "release_sha256": "a" * 64, "image": { "reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'b' * 64}", "digest": f"sha256:{'b' * 64}", "config_digest": f"sha256:{'c' * 64}", }, "image_input": {"runtime": {"teleo_git_head": "d" * 40}}, } manifest = { "operation": "install", "request_id": "iap-abcdefghijkl", "runner_revision": "e" * 40, "release_revision": "d" * 40, "manifest_sha256": "f" * 64, "release": { "release_sha256": release["release_sha256"], "image_reference": release["image"]["reference"], "image_digest": release["image"]["digest"], "image_config_digest": release["image"]["config_digest"], }, "release_artifact": {"archive_sha256": "1" * 64}, "installer_source": {"revision": "e" * 40}, } verification = { "main_pid": 10, "invocation_id": "2" * 32, "container_id": "3" * 64, "n_restarts": 0, } actions: list[str] = [] postures = [ SimpleNamespace( kind="active", state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 0}, ) ] if prior_posture == "active": postures.append( SimpleNamespace( kind="active", state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 0}, ) ) else: postures.append( SimpleNamespace( kind="inactive", state={"MainPID": 0, "InvocationID": "", "NRestarts": 0}, ) ) posture_iter = iter(postures) class Paths: @staticmethod def under(_root: Path) -> object: return object() installer_module = SimpleNamespace( SYSTEMCTL="/usr/bin/systemctl", VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES, INSTALL_EXECUTION_RESULT_SCHEMA=local.installer.INSTALL_EXECUTION_RESULT_SCHEMA, LEGACY_INSTALL_RECEIPT_SCHEMA=local.installer.LEGACY_INSTALL_RECEIPT_SCHEMA, INSTALL_RECEIPT_SCHEMA=local.installer.INSTALL_RECEIPT_SCHEMA, InstallPaths=Paths, validate_release_bundle=lambda _bundle: (release, b"unit"), _validate_source_binding=lambda _binding: None, install_release=lambda *_args, **_kwargs: install_execution_result( release, { "outcome": outcome, "prior_posture": prior_posture, "prior_n_restarts": 0, "final_posture": "active", "service_action": service_action, "managed_files_replaced": False, }, ), verify_running_release=lambda *_args, **_kwargs: verification, rollback_release=lambda *_args, **_kwargs: (_ for _ in ()).throw( AssertionError("historical rollback must not be consumed") ), _systemctl=lambda action, _runner: actions.append(action), _stable_service_posture=lambda *_args, **_kwargs: next(posture_iter), _require_candidate_absent=lambda **_kwargs: None, ) package_module = SimpleNamespace( render_systemd_unit=lambda _release: "unit", canonical_sha256=local.package.canonical_sha256, ) monkeypatch.setattr(remote, "validate_original_release_artifact", lambda *_args, **_kwargs: None) monkeypatch.setattr(remote, "_load_contract_modules", lambda _root: (package_module, installer_module)) monkeypatch.setattr(remote, "pull_immutable_image", lambda *_args, **_kwargs: None) monkeypatch.setattr( remote, "_verify_database", lambda **_kwargs: (_ for _ in ()).throw(remote.RemoteError("database failed")), ) with pytest.raises(remote.RemoteOperationFailure) as caught: remote.execute_request( manifest, tmp_path, tmp_path, "6" * 64, "7" * 64, runner=lambda argv, **_kwargs: completed(argv), ) result = caught.value.result assert actions == expected_actions assert result["rollback"] == { "status": "pass", "mode": "posture_recovery", "executed": expected_executed, "restored_posture": prior_posture, "prior_posture_preserved": True, } def test_invalid_installer_transition_fails_closed_with_structured_receipt( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: release = { "release_sha256": "a" * 64, "image": { "reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'b' * 64}", "digest": f"sha256:{'b' * 64}", "config_digest": f"sha256:{'c' * 64}", }, "image_input": {"runtime": {"teleo_git_head": "d" * 40}}, } manifest = { "operation": "install", "request_id": "iap-abcdefghijkl", "runner_revision": "e" * 40, "release_revision": "d" * 40, "manifest_sha256": "f" * 64, "release": { "release_sha256": release["release_sha256"], "image_reference": release["image"]["reference"], "image_digest": release["image"]["digest"], "image_config_digest": release["image"]["config_digest"], }, "release_artifact": {"archive_sha256": "1" * 64}, "installer_source": {"revision": "e" * 40}, } actions: list[str] = [] historical_rollbacks: list[bool] = [] postures = iter( [ SimpleNamespace( kind="active", state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 0}, ), SimpleNamespace( kind="inactive", state={ "ActiveState": "inactive", "SubState": "dead", "MainPID": 0, "InvocationID": "", "NRestarts": 0, }, ), ] ) class Paths: @staticmethod def under(_root: Path) -> object: return object() installer_module = SimpleNamespace( SYSTEMCTL="/usr/bin/systemctl", VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES, INSTALL_EXECUTION_RESULT_SCHEMA=local.installer.INSTALL_EXECUTION_RESULT_SCHEMA, LEGACY_INSTALL_RECEIPT_SCHEMA=local.installer.LEGACY_INSTALL_RECEIPT_SCHEMA, INSTALL_RECEIPT_SCHEMA=local.installer.INSTALL_RECEIPT_SCHEMA, InstallPaths=Paths, validate_release_bundle=lambda _bundle: (release, b"unit"), _validate_source_binding=lambda _binding: None, install_release=lambda *_args, **_kwargs: install_execution_result( release, {"outcome": "controlled_start"}, ), verify_running_release=lambda *_args, **_kwargs: { "main_pid": 10, "invocation_id": "2" * 32, "container_id": "3" * 64, "n_restarts": 0, }, rollback_release=lambda *_args, **_kwargs: historical_rollbacks.append(True), _systemctl=lambda action, _runner: actions.append(action), _stable_service_posture=lambda *_args, **_kwargs: next(postures), _require_candidate_absent=lambda **_kwargs: None, ) package_module = SimpleNamespace( render_systemd_unit=lambda _release: "unit", canonical_sha256=local.package.canonical_sha256, ) monkeypatch.setattr(remote, "validate_original_release_artifact", lambda *_args, **_kwargs: None) monkeypatch.setattr(remote, "_load_contract_modules", lambda _root: (package_module, installer_module)) monkeypatch.setattr(remote, "pull_immutable_image", lambda *_args, **_kwargs: None) with pytest.raises(remote.RemoteOperationFailure) as caught: remote.execute_request( manifest, tmp_path, tmp_path, "6" * 64, "7" * 64, runner=lambda argv, **_kwargs: completed(argv), ) result = caught.value.result assert historical_rollbacks == [] assert actions == ["stop"] assert (result["failure_stage"], result["failure_reason"]) == ( "install_receipt", "install_receipt_invalid", ) assert result["rollback"] == { "status": "fail", "mode": "posture_recovery", "executed": True, "restored_posture": "unknown", "prior_posture_preserved": False, } assert result["fail_closed"]["status"] == "pass" assert result["fail_closed"]["service_action_executed"] is True def test_posture_recovery_refuses_identity_drift_before_stop( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: release = { "release_sha256": "a" * 64, "image": { "reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'b' * 64}", "digest": f"sha256:{'b' * 64}", "config_digest": f"sha256:{'c' * 64}", }, "image_input": {"runtime": {"teleo_git_head": "d" * 40}}, } manifest = { "operation": "install", "request_id": "iap-abcdefghijkl", "runner_revision": "e" * 40, "release_revision": "d" * 40, "manifest_sha256": "f" * 64, "release": { "release_sha256": release["release_sha256"], "image_reference": release["image"]["reference"], "image_digest": release["image"]["digest"], "image_config_digest": release["image"]["config_digest"], }, "release_artifact": {"archive_sha256": "1" * 64}, "installer_source": {"revision": "e" * 40}, } actions: list[str] = [] postures = iter( [ SimpleNamespace(kind="active", state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 0}), SimpleNamespace(kind="active", state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 0}), ] ) verifications = iter( [ {"main_pid": 10, "invocation_id": "2" * 32, "container_id": "3" * 64, "n_restarts": 0}, {"main_pid": 99, "invocation_id": "9" * 32, "container_id": "8" * 64, "n_restarts": 0}, {"main_pid": 99, "invocation_id": "9" * 32, "container_id": "8" * 64, "n_restarts": 0}, ] ) class Paths: @staticmethod def under(_root: Path) -> object: return object() installer_module = SimpleNamespace( SYSTEMCTL="/usr/bin/systemctl", VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES, INSTALL_EXECUTION_RESULT_SCHEMA=local.installer.INSTALL_EXECUTION_RESULT_SCHEMA, LEGACY_INSTALL_RECEIPT_SCHEMA=local.installer.LEGACY_INSTALL_RECEIPT_SCHEMA, INSTALL_RECEIPT_SCHEMA=local.installer.INSTALL_RECEIPT_SCHEMA, InstallPaths=Paths, validate_release_bundle=lambda _bundle: (release, b"unit"), _validate_source_binding=lambda _binding: None, install_release=lambda *_args, **_kwargs: install_execution_result( release, { "outcome": "controlled_start", "prior_posture": "inactive", "prior_n_restarts": 0, "final_posture": "active", "service_action": "start", "managed_files_replaced": False, }, ), verify_running_release=lambda *_args, **_kwargs: next(verifications), rollback_release=lambda *_args, **_kwargs: (_ for _ in ()).throw( AssertionError("historical rollback must not be consumed") ), _systemctl=lambda action, _runner: actions.append(action), _stable_service_posture=lambda *_args, **_kwargs: next(postures), _require_candidate_absent=lambda **_kwargs: None, ) package_module = SimpleNamespace( render_systemd_unit=lambda _release: "unit", canonical_sha256=local.package.canonical_sha256, ) monkeypatch.setattr(remote, "validate_original_release_artifact", lambda *_args, **_kwargs: None) monkeypatch.setattr(remote, "_load_contract_modules", lambda _root: (package_module, installer_module)) monkeypatch.setattr(remote, "pull_immutable_image", lambda *_args, **_kwargs: None) monkeypatch.setattr( remote, "_verify_database", lambda **_kwargs: (_ for _ in ()).throw(remote.RemoteError("database failed")), ) with pytest.raises(remote.RemoteOperationFailure) as caught: remote.execute_request( manifest, tmp_path, tmp_path, "6" * 64, "7" * 64, runner=lambda argv, **_kwargs: completed(argv), ) result = caught.value.result assert actions == [] assert result["rollback"]["executed"] is False assert result["rollback"]["mode"] == "posture_recovery" assert result["fail_closed"]["status"] == "fail" assert result["fail_closed"]["service_action_executed"] is False def test_posture_recovery_rejects_restart_counter_drift_after_fixed_restart() -> None: actions: list[str] = [] postures = iter( [ SimpleNamespace( kind="active", state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 1}, ), SimpleNamespace( kind="inactive", state={"MainPID": 0, "InvocationID": "", "NRestarts": 1}, ), ] ) installer_module = SimpleNamespace( _systemctl=lambda action, _runner: actions.append(action), _stable_service_posture=lambda *_args, **_kwargs: next(postures), verify_running_release=lambda *_args, **_kwargs: { "main_pid": 10, "invocation_id": "2" * 32, "container_id": "3" * 64, "n_restarts": 1, }, _require_candidate_absent=lambda **_kwargs: None, ) with pytest.raises(remote.PostureRecoveryFailure) as caught: remote._restore_transition_posture( installer_module=installer_module, release={"release_sha256": "a" * 64}, transition={ "outcome": "controlled_start", "prior_posture": "inactive", "prior_n_restarts": 0, "final_posture": "active", "service_action": "start", "managed_files_replaced": False, }, paths=object(), runner=lambda argv, **_kwargs: completed(argv), ) assert caught.value.executed is True assert actions == ["stop"] def test_incomplete_rollback_stops_service_and_reports_inert_posture( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: release = { "release_sha256": "a" * 64, "image": { "reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'b' * 64}", "digest": f"sha256:{'b' * 64}", "config_digest": f"sha256:{'c' * 64}", }, "image_input": {"runtime": {"teleo_git_head": "d" * 40}}, } manifest = { "operation": "install", "request_id": "iap-abcdefghijkl", "runner_revision": "e" * 40, "release_revision": "d" * 40, "manifest_sha256": "f" * 64, "release": { "release_sha256": release["release_sha256"], "image_reference": release["image"]["reference"], "image_digest": release["image"]["digest"], "image_config_digest": release["image"]["config_digest"], }, "release_artifact": {"archive_sha256": "1" * 64}, "installer_source": {"revision": "e" * 40}, } stops: list[str] = [] absence_checks: list[bool] = [] class Paths: @staticmethod def under(_root: Path) -> object: return object() postures = iter( [ SimpleNamespace( kind="active", state={ "ActiveState": "active", "SubState": "running", "MainPID": 10, "NRestarts": 0, "InvocationID": "2" * 32, }, ), SimpleNamespace( kind="inactive", state={ "ActiveState": "inactive", "SubState": "dead", "MainPID": 0, "NRestarts": 0, "InvocationID": "", }, ), ] ) def fail_rollback(release_sha256: str, **kwargs: Any) -> dict[str, Any]: if kwargs["execute_rollback"] is False: return { "schema": "livingip.leocleanNoSendRollbackResult.v1", "status": "pass", "mode": "dry_run", "expected_release_sha256": release_sha256, } raise RuntimeError("rollback failed") installer_module = SimpleNamespace( SYSTEMCTL="/usr/bin/systemctl", INSTALL_EXECUTION_RESULT_SCHEMA=local.installer.INSTALL_EXECUTION_RESULT_SCHEMA, LEGACY_INSTALL_RECEIPT_SCHEMA=local.installer.LEGACY_INSTALL_RECEIPT_SCHEMA, INSTALL_RECEIPT_SCHEMA=local.installer.INSTALL_RECEIPT_SCHEMA, InstallPaths=Paths, validate_release_bundle=lambda _bundle: (release, b"unit"), _validate_source_binding=lambda _binding: None, install_release=lambda *_args, **_kwargs: install_execution_result( release, { "outcome": "installed", "prior_posture": "absent", "prior_n_restarts": 0, "final_posture": "active", "service_action": "restart", "managed_files_replaced": True, }, ), verify_running_release=lambda *_args, **_kwargs: { "main_pid": 10, "invocation_id": "2" * 32, "container_id": "3" * 64, "n_restarts": 0, }, rollback_release=fail_rollback, _systemctl=lambda action, _runner: stops.append(action), _stable_service_posture=lambda *_args, **_kwargs: next(postures), _require_candidate_absent=lambda **_kwargs: absence_checks.append(True), VERIFICATION_FAILURES=frozenset(), ) package_module = SimpleNamespace( render_systemd_unit=lambda _release: "unit", canonical_sha256=local.package.canonical_sha256, ) monkeypatch.setattr(remote, "validate_original_release_artifact", lambda *_args, **_kwargs: None) monkeypatch.setattr(remote, "_load_contract_modules", lambda _root: (package_module, installer_module)) monkeypatch.setattr(remote, "pull_immutable_image", lambda *_args, **_kwargs: None) monkeypatch.setattr( remote, "_verify_database", lambda **_kwargs: (_ for _ in ()).throw(remote.RemoteError("database failed")), ) monkeypatch.setattr(remote.time, "sleep", lambda _seconds: None) with pytest.raises(remote.RemoteOperationFailure) as caught: remote.execute_request( manifest, tmp_path, tmp_path, "6" * 64, "7" * 64, runner=lambda argv, **_kwargs: completed(argv), ) result = caught.value.result assert stops == ["stop"] assert absence_checks == [True] assert result["rollback"] == { "status": "fail", "mode": "explicit", "executed": True, "restored_posture": "unknown", "prior_posture_preserved": False, } assert result["fail_closed"] == { "status": "pass", "service_posture": "inactive", "active_state": "inactive", "sub_state": "dead", "main_pid": 0, "n_restarts": 0, "candidate_container": "absent", "service_action_executed": True, } assert result["authority"]["sending_enabled"] is False def test_fail_closed_stop_refuses_untrusted_unit_or_dropin() -> None: stop_calls: list[str] = [] installer_module = SimpleNamespace( _systemctl=lambda action, _runner: stop_calls.append(action), _stable_service_posture=lambda *_args, **_kwargs: (_ for _ in ()).throw( local.installer.InstallError("unit or drop-in drift") ), verify_running_release=lambda *_args, **_kwargs: {"status": "unexpected"}, _require_candidate_absent=lambda **_kwargs: None, ) result = remote._fail_closed_inert_service( installer_module=installer_module, release={"release_sha256": "a" * 64}, paths=object(), runner=lambda argv, **_kwargs: completed(argv), ) assert stop_calls == [] assert result == { "status": "fail", "service_posture": "unknown", "active_state": "unknown", "sub_state": "unknown", "main_pid": None, "n_restarts": None, "candidate_container": "unknown", "service_action_executed": False, } @pytest.mark.parametrize( ("posture", "restored_release", "verification"), [ ( "active", "e" * 64, { "active_state": "active", "sub_state": "running", "container_id": "f" * 64, }, ), ("inactive", "e" * 64, None), ("absent", None, None), ], ) def test_rollback_result_accepts_exact_recorded_postures( posture: str, restored_release: str | None, verification: dict[str, Any] | None, ) -> None: manifest = { "request_id": "iap-abcdefghijkl", "runner_revision": "a" * 40, "release_revision": "d" * 40, "manifest_sha256": "1" * 64, "release_artifact": {"archive_sha256": "2" * 64}, } release = { "release_sha256": "b" * 64, "image": {"digest": f"sha256:{'c' * 64}", "config_digest": f"sha256:{'d' * 64}"}, } rollback = { "schema": "livingip.leocleanNoSendRollbackResult.v1", "status": "pass", "mode": "executed", "rolled_back_release_sha256": "b" * 64, "restored_release_sha256": restored_release, "restored_service_posture": posture, "verification": verification, } result = remote._rollback_result( manifest=manifest, release=release, rollback=rollback, archive_sha256="3" * 64, helper_sha256="4" * 64, ) assert result["service"]["restored_posture"] == posture assert result["database"] == {"status": "not_run_after_explicit_rollback"} def test_absent_rollback_rejects_invented_database_or_runtime_verification() -> None: manifest = { "request_id": "iap-abcdefghijkl", "runner_revision": "a" * 40, "release_revision": "d" * 40, "manifest_sha256": "1" * 64, "release_artifact": {"archive_sha256": "2" * 64}, } release = { "release_sha256": "b" * 64, "image": {"digest": f"sha256:{'c' * 64}", "config_digest": f"sha256:{'d' * 64}"}, } rollback = { "schema": "livingip.leocleanNoSendRollbackResult.v1", "status": "pass", "mode": "executed", "rolled_back_release_sha256": "b" * 64, "restored_release_sha256": None, "restored_service_posture": "absent", "verification": {"active_state": "active"}, } with pytest.raises(remote.RemoteError, match="absent rollback"): remote._rollback_result( manifest=manifest, release=release, rollback=rollback, archive_sha256="3" * 64, helper_sha256="4" * 64, )