diff --git a/docs/gcp-leoclean-nosend-service-package.md b/docs/gcp-leoclean-nosend-service-package.md index 0f35764..d5f7fee 100644 --- a/docs/gcp-leoclean-nosend-service-package.md +++ b/docs/gcp-leoclean-nosend-service-package.md @@ -344,6 +344,33 @@ contact GCP or Cloud SQL, change Telegram traffic, touch the VPS, approve a proposal, or promote production. Those remain later, separately authorized boundaries. +## IAP result and failure receipts + +The fixed-target IAP runner treats the remote `result.json`, the remote process +exit status/stdout, the IAP transport, and the private local result receipt as +one bounded result path. A successful operation is accepted only when the +structured result matches the exact request, revision, artifact, image, config +digest, service, database proof, and no-send authority contract. + +Post-install failures are returned with allowlisted `failure_stage` and +`failure_reason` values. If explicit rollback passes, its exact restored +posture is recorded. If rollback cannot be proven, the remote helper stops the +staging service only after exact-binding the currently loaded unit, `ExecStop`, +supervising process, and container to the reviewed release. It then proves +`inactive/dead`, `MainPID=0`, and absence of the named candidate container +before reporting a fail-closed posture. Drift blocks the stop; a failure to +prove that posture is reported as unknown and is never converted into success. + +If IAP returns no output, malformed output, a contract mismatch, or a transport +interruption, the local runner writes a private self-hashed failure receipt +with the bounded stage/reason, transfer-cleanup result, and `remote_state: +unknown`. Command stdout, stderr, tokens, passwords, and secret values are not +copied into a receipt or exception. That local-only receipt records requested +authority, not unproven remote outcomes. Operators must not infer a successful +or inert remote service from a local transport-failure receipt. Remote +operation failures and local transport failures use distinct outer receipt +schemas; the successful result-receipt v1 contract is unchanged. + ## Offline evidence and live gate Offline tests prove strict receipt/descriptor/inspection schemas, exact diff --git a/ops/gcp_leoclean_nosend_iap_remote.py b/ops/gcp_leoclean_nosend_iap_remote.py index 0bfde9f..dcb4902 100644 --- a/ops/gcp_leoclean_nosend_iap_remote.py +++ b/ops/gcp_leoclean_nosend_iap_remote.py @@ -16,6 +16,7 @@ import subprocess import sys import tarfile import tempfile +import time import zipfile from collections.abc import Callable, Mapping from pathlib import Path @@ -24,6 +25,7 @@ from typing import Any SCHEMA = "livingip.leocleanNoSendIapRequest.v1" RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v1" FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapFailure.v1" +OPERATION_FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v1" PROJECT = "teleo-501523" ZONE = "europe-west6-a" INSTANCE = "teleo-staging-1" @@ -48,6 +50,17 @@ MAX_MEMBER_BYTES = 2 * 1024 * 1024 MAX_MANIFEST_BYTES = 128 * 1024 MAX_COMMAND_OUTPUT_BYTES = 1024 * 1024 MAX_TOKEN_BYTES = 64 * 1024 +POST_INSTALL_FAILURES = frozenset( + { + ("install_receipt", "install_receipt_invalid"), + ("service_before_restart", "service_verification_failed"), + ("database_before_restart", "database_verification_failed"), + ("service_restart", "service_restart_failed"), + ("service_after_restart", "service_verification_failed"), + ("restart_identity", "restart_identity_mismatch"), + ("database_after_restart", "database_verification_failed"), + } +) EXPECTED_MEMBERS = { "request-manifest.json", "release-artifact.zip", @@ -852,6 +865,236 @@ def _is_verification_failure(installer_module: Any, failure: Exception) -> bool: return isinstance(failure_type, type) and isinstance(failure, failure_type) +def _post_install_failure_coordinates( + installer_module: Any, + failure: Exception, + *, + fallback_stage: str, + fallback_reason: str, +) -> tuple[str, str]: + if _is_verification_failure(installer_module, failure): + failure_stage = getattr(failure, "failure_stage", None) + failure_reason = getattr(failure, "failure_reason", None) + _require( + isinstance(failure_stage, str) + and isinstance(failure_reason, str) + and (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES, + "verification failure was not allowlisted", + ) + return failure_stage, failure_reason + _require( + (fallback_stage, fallback_reason) in POST_INSTALL_FAILURES, + "post-install failure was not allowlisted", + ) + return fallback_stage, fallback_reason + + +def _rollback_proven_service(posture: str) -> dict[str, Any]: + _require(posture in {"active", "inactive", "absent"}, "rollback posture was invalid") + return { + "status": "not_required", + "service_posture": posture, + "active_state": "not_observed", + "sub_state": "not_observed", + "main_pid": None, + "n_restarts": None, + "candidate_container": "not_observed", + } + + +def _fail_closed_inert_service( + *, + installer_module: Any, + release: Mapping[str, Any], + paths: Any, + runner: Runner, +) -> dict[str, Any]: + def contract_runner(argv: list[str]) -> subprocess.CompletedProcess[bytes]: + return _run( + argv, + runner=runner, + label="fail-closed service command", + require_success=False, + ) + + try: + systemctl = getattr(installer_module, "_systemctl", None) + stable_service_posture = getattr(installer_module, "_stable_service_posture", None) + verify_running_release = getattr(installer_module, "verify_running_release", None) + require_candidate_absent = getattr(installer_module, "_require_candidate_absent", None) + _require( + callable(systemctl) + and callable(stable_service_posture) + and callable(verify_running_release) + and callable(require_candidate_absent), + "reviewed installer lacked fail-closed service controls", + ) + before = stable_service_posture( + release, + runner=contract_runner, + paths=paths, + sleeper=time.sleep, + ) + _require(before.kind in {"active", "inactive"}, "managed service posture was not trusted") + if before.kind == "active": + verification = verify_running_release( + release, + runner=contract_runner, + paths=paths, + sleeper=time.sleep, + ) + _require( + verification.get("main_pid") == before.state.get("MainPID") + and verification.get("invocation_id") == before.state.get("InvocationID") + and verification.get("n_restarts") == before.state.get("NRestarts"), + "running service identity changed before fail-closed stop", + ) + systemctl("stop", contract_runner) + after = stable_service_posture( + release, + runner=contract_runner, + paths=paths, + sleeper=time.sleep, + ) + _require( + after.kind == "inactive" + and after.state.get("ActiveState") == "inactive" + and after.state.get("SubState") == "dead" + and after.state.get("MainPID") == 0 + and isinstance(after.state.get("NRestarts"), int) + and not isinstance(after.state.get("NRestarts"), bool) + and after.state["NRestarts"] >= 0, + "service did not stop into a stable inert posture", + ) + require_candidate_absent( + runner=contract_runner, + paths=paths, + sleeper=time.sleep, + ) + return { + "status": "pass", + "service_posture": "inactive", + "active_state": "inactive", + "sub_state": "dead", + "main_pid": 0, + "n_restarts": after.state["NRestarts"], + "candidate_container": "absent", + } + except Exception: + return { + "status": "fail", + "service_posture": "unknown", + "active_state": "unknown", + "sub_state": "unknown", + "main_pid": None, + "n_restarts": None, + "candidate_container": "unknown", + } + + +def _operational_failure_result( + *, + manifest: Mapping[str, Any], + release: Mapping[str, Any], + failure_stage: str, + failure_reason: str, + rollback: Mapping[str, str], + fail_closed: Mapping[str, Any], + archive_sha256: str, + helper_sha256: str, + installer_module: Any, +) -> dict[str, Any]: + _require( + manifest.get("operation") == "install" + and set(rollback) == {"status", "mode", "restored_posture"} + and ( + (failure_stage, failure_reason) in POST_INSTALL_FAILURES + or (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES + ), + "post-install failure was not allowlisted", + ) + rollback_passed = ( + rollback.get("status") == "pass" + and rollback.get("mode") == "explicit" + and rollback.get("restored_posture") in {"active", "inactive", "absent"} + ) + rollback_failed = rollback == { + "status": "fail", + "mode": "explicit", + "restored_posture": "unknown", + } + verification_pair = (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES + _require( + (rollback_passed and not verification_pair) or rollback_failed, + "post-install rollback summary was invalid", + ) + if rollback_passed: + _require( + fail_closed == _rollback_proven_service(str(rollback["restored_posture"])), + "post-install restored posture summary was invalid", + ) + else: + _require( + fail_closed + in ( + { + "status": "fail", + "service_posture": "unknown", + "active_state": "unknown", + "sub_state": "unknown", + "main_pid": None, + "n_restarts": None, + "candidate_container": "unknown", + }, + { + "status": "pass", + "service_posture": "inactive", + "active_state": "inactive", + "sub_state": "dead", + "main_pid": 0, + "n_restarts": fail_closed.get("n_restarts"), + "candidate_container": "absent", + }, + ) + and ( + fail_closed.get("status") != "pass" + or ( + isinstance(fail_closed.get("n_restarts"), int) + and not isinstance(fail_closed.get("n_restarts"), bool) + and fail_closed["n_restarts"] >= 0 + ) + ), + "post-install fail-closed summary was invalid", + ) + return { + "schema": OPERATION_FAILURE_RESULT_SCHEMA, + "status": "fail", + "operation": manifest["operation"], + "request_id": manifest["request_id"], + "runner_revision": manifest["runner_revision"], + "release_revision": manifest["release_revision"], + "release_sha256": release["release_sha256"], + "image_digest": release["image"]["digest"], + "image_config_digest": release["image"]["config_digest"], + "release_artifact_sha256": manifest["release_artifact"]["archive_sha256"], + "request_binding": { + "archive_sha256": archive_sha256, + "helper_sha256": helper_sha256, + "manifest_sha256": manifest["manifest_sha256"], + }, + "failure_stage": failure_stage, + "failure_reason": failure_reason, + "rollback": dict(rollback), + "fail_closed": dict(fail_closed), + "authority": { + "sending_enabled": False, + "production_traffic_changed": False, + "vps_touched": False, + "canonical_proposal_approved": False, + }, + } + + def execute_request( manifest: Mapping[str, Any], payload_root: Path, @@ -919,7 +1162,6 @@ def execute_request( installer_module=installer_module, ) ) from exc - _require(install_receipt.get("status") == "pass", "installation did not pass") elif operation == "rollback": rollback = _execute_explicit_rollback( installer_module=installer_module, @@ -945,12 +1187,19 @@ def execute_request( ) if operation == "install": + failure_stage = "install_receipt" + failure_reason = "install_receipt_invalid" try: + _require(install_receipt.get("status") == "pass", "installation did not pass") + failure_stage = "service_before_restart" + failure_reason = "service_verification_failed" before = installer_module.verify_running_release( release, runner=contract_runner, paths=paths, ) + failure_stage = "database_before_restart" + failure_reason = "database_verification_failed" before_database = _verify_database( container_id=before["container_id"], run_id=f"{manifest['request_id']}-pre", @@ -959,16 +1208,22 @@ def execute_request( package_module=package_module, runner=runner, ) + failure_stage = "service_restart" + failure_reason = "service_restart_failed" _run( [installer_module.SYSTEMCTL, "restart", SERVICE], runner=runner, label="fixed service restart", ) + failure_stage = "service_after_restart" + failure_reason = "service_verification_failed" after = installer_module.verify_running_release( release, runner=contract_runner, paths=paths, ) + failure_stage = "restart_identity" + failure_reason = "restart_identity_mismatch" _require( before["main_pid"] != after["main_pid"] and before["invocation_id"] != after["invocation_id"] @@ -976,6 +1231,8 @@ def execute_request( and after["n_restarts"] >= before["n_restarts"], "post-install restart did not create a new exact service/container identity", ) + failure_stage = "database_after_restart" + failure_reason = "database_verification_failed" after_database = _verify_database( container_id=after["container_id"], run_id=f"{manifest['request_id']}-post", @@ -985,6 +1242,12 @@ def execute_request( runner=runner, ) except Exception as exc: + bounded_stage, bounded_reason = _post_install_failure_coordinates( + installer_module, + exc, + fallback_stage=failure_stage, + fallback_reason=failure_reason, + ) try: recovery = _execute_explicit_rollback( installer_module=installer_module, @@ -1000,8 +1263,30 @@ def execute_request( archive_sha256=archive_sha256, helper_sha256=helper_sha256, ) - except Exception as rollback_exc: - raise RemoteError("install_failed_rollback_incomplete") from rollback_exc + except Exception: + fail_closed = _fail_closed_inert_service( + installer_module=installer_module, + release=release, + paths=paths, + runner=runner, + ) + raise RemoteOperationFailure( + _operational_failure_result( + manifest=manifest, + release=release, + failure_stage=bounded_stage, + failure_reason=bounded_reason, + rollback={ + "status": "fail", + "mode": "explicit", + "restored_posture": "unknown", + }, + fail_closed=fail_closed, + archive_sha256=archive_sha256, + helper_sha256=helper_sha256, + installer_module=installer_module, + ) + ) from exc if _is_verification_failure(installer_module, exc): raise RemoteOperationFailure( _verification_failure_result( @@ -1018,7 +1303,24 @@ def execute_request( installer_module=installer_module, ) ) from exc - raise RemoteError("install_failed_rolled_back") from exc + restored_posture = str(rollback_result["service"]["restored_posture"]) + raise RemoteOperationFailure( + _operational_failure_result( + manifest=manifest, + release=release, + failure_stage=bounded_stage, + failure_reason=bounded_reason, + rollback={ + "status": "pass", + "mode": "explicit", + "restored_posture": restored_posture, + }, + fail_closed=_rollback_proven_service(restored_posture), + archive_sha256=archive_sha256, + helper_sha256=helper_sha256, + installer_module=installer_module, + ) + ) from exc service = { "name": SERVICE, "before_restart": before, @@ -1089,6 +1391,35 @@ def execute_request( } +def _execute_and_persist_result( + manifest: Mapping[str, Any], + payload_root: Path, + request_dir: Path, + archive_sha256: str, + helper_sha256: str, + *, + runner: Runner, +) -> dict[str, Any]: + try: + result = execute_request( + manifest, + payload_root, + request_dir, + archive_sha256, + helper_sha256, + runner=runner, + ) + except RemoteOperationFailure as exc: + result = exc.result + result_path = request_dir / "result.json" + _copy_into_request( + result_path, + (json.dumps(result, allow_nan=False, indent=2, sort_keys=True) + "\n").encode("utf-8"), + 0o600, + ) + return result + + def receive_request( operation: str, request_id: str, @@ -1154,24 +1485,14 @@ def receive_request( release_revision=release_revision, ) verify_service_account(runner=runner) - try: - result = execute_request( - manifest, - payload_root, - request_dir, - archive_sha256, - helper_sha256, - runner=runner, - ) - except RemoteOperationFailure as exc: - result = exc.result - result_path = request_dir / "result.json" - _copy_into_request( - result_path, - (json.dumps(result, allow_nan=False, indent=2, sort_keys=True) + "\n").encode("utf-8"), - 0o600, + return _execute_and_persist_result( + manifest, + payload_root, + request_dir, + archive_sha256, + helper_sha256, + runner=runner, ) - return result finally: os.close(lock_descriptor) finally: diff --git a/ops/run_gcp_leoclean_nosend_iap.py b/ops/run_gcp_leoclean_nosend_iap.py index 3461c5a..f616677 100644 --- a/ops/run_gcp_leoclean_nosend_iap.py +++ b/ops/run_gcp_leoclean_nosend_iap.py @@ -33,7 +33,10 @@ SCHEMA = "livingip.leocleanNoSendIapRequest.v1" PLAN_SCHEMA = "livingip.leocleanNoSendIapPlan.v1" REMOTE_RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v1" REMOTE_FAILURE_SCHEMA = "livingip.leocleanNoSendIapFailure.v1" +REMOTE_OPERATION_FAILURE_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v1" RESULT_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapResultReceipt.v1" +FAILURE_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapFailureReceipt.v1" +LOCAL_FAILURE_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapLocalFailureReceipt.v1" EXECUTION_SCHEMA = "livingip.leocleanNoSendIapExecution.v1" PROJECT = package.PROJECT ZONE = package.ZONE @@ -78,6 +81,26 @@ FORBIDDEN_OUTPUT_MARKERS = ( "77.42.65.182", "teleo-prod-1", ) +POST_INSTALL_FAILURES = frozenset( + { + ("install_receipt", "install_receipt_invalid"), + ("service_before_restart", "service_verification_failed"), + ("database_before_restart", "database_verification_failed"), + ("service_restart", "service_restart_failed"), + ("service_after_restart", "service_verification_failed"), + ("restart_identity", "restart_identity_mismatch"), + ("database_after_restart", "database_verification_failed"), + } +) +LOCAL_FAILURES = frozenset( + { + ("iap_transport", "upload_failed"), + ("iap_transport", "remote_command_interrupted"), + ("remote_receipt", "absent"), + ("remote_receipt", "malformed"), + ("remote_receipt", "contract_mismatch"), + } +) BOOTSTRAP_CODE = textwrap.dedent( r""" import fcntl, hashlib, io, json, os, pwd, re, stat, sys, tarfile @@ -203,7 +226,7 @@ def _strict_json(raw: bytes, label: str) -> dict[str, Any]: object_pairs_hook=reject_duplicates, parse_constant=lambda _value: (_ for _ in ()).throw(ValueError("constant")), ) - except (UnicodeError, ValueError, json.JSONDecodeError) as exc: + except (UnicodeError, ValueError, json.JSONDecodeError, RecursionError) as exc: raise IapRequestError(f"{label} was invalid JSON") from exc _require(isinstance(value, dict), f"{label} was not a JSON object") return value @@ -717,6 +740,17 @@ def _action_receipts(commands: tuple[list[str], list[str], list[str]]) -> list[d ] +def _expected_authority(operation: str) -> dict[str, bool]: + _require(operation in OPERATIONS, "operation was invalid") + return { + "gcp_mutation": operation in {"install", "rollback"}, + "service_restart": operation in {"install", "rollback"}, + "database_or_secret_provisioning": False, + "production_or_transport": False, + "vps": False, + } + + def prepare_request( operation: str, request_id: str, @@ -833,13 +867,7 @@ def prepare_request( "ssh_key": str(ssh_key), }, "actions": actions, - "authority": { - "gcp_mutation": operation in {"install", "rollback"}, - "service_restart": operation in {"install", "rollback"}, - "database_or_secret_provisioning": False, - "production_or_transport": False, - "vps": False, - }, + "authority": _expected_authority(operation), } @@ -1028,33 +1056,114 @@ def validate_remote_result(plan: Mapping[str, Any], result: Mapping[str, Any]) - def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any]) -> None: - _require( - set(result) - == { - "schema", - "status", - "operation", - "request_id", - "runner_revision", - "release_revision", - "release_sha256", - "image_digest", - "image_config_digest", - "release_artifact_sha256", - "request_binding", - "failure_stage", - "failure_reason", - "rollback", - "authority", - }, - "remote failure fields were not exact", - ) request = plan["request"] release = plan["release"] rollback = result.get("rollback") + common_fields = { + "schema", + "status", + "operation", + "request_id", + "runner_revision", + "release_revision", + "release_sha256", + "image_digest", + "image_config_digest", + "release_artifact_sha256", + "request_binding", + "failure_stage", + "failure_reason", + "rollback", + "authority", + } + schema = result.get("schema") + if schema == REMOTE_FAILURE_SCHEMA: + _require(set(result) == common_fields, "remote failure fields were not exact") + failure_contract = ( + (result.get("failure_stage"), result.get("failure_reason")) in installer.VERIFICATION_FAILURES + and rollback + in ( + {"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"}, + {"status": "pass", "mode": "explicit", "restored_posture": "active"}, + {"status": "pass", "mode": "explicit", "restored_posture": "inactive"}, + {"status": "pass", "mode": "explicit", "restored_posture": "absent"}, + {"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"}, + ) + ) + else: + _require( + schema == REMOTE_OPERATION_FAILURE_SCHEMA and set(result) == common_fields | {"fail_closed"}, + "remote failure fields were not exact", + ) + failure_pair = (result.get("failure_stage"), result.get("failure_reason")) + rollback_passed = ( + isinstance(rollback, dict) + and set(rollback) == {"status", "mode", "restored_posture"} + and rollback.get("status") == "pass" + and rollback.get("mode") == "explicit" + and rollback.get("restored_posture") in {"active", "inactive", "absent"} + ) + rollback_failed = rollback == { + "status": "fail", + "mode": "explicit", + "restored_posture": "unknown", + } + fail_closed = result.get("fail_closed") + if rollback_passed: + fail_closed_contract = fail_closed == { + "status": "not_required", + "service_posture": rollback["restored_posture"], + "active_state": "not_observed", + "sub_state": "not_observed", + "main_pid": None, + "n_restarts": None, + "candidate_container": "not_observed", + } + else: + fail_closed_contract = ( + fail_closed + == { + "status": "fail", + "service_posture": "unknown", + "active_state": "unknown", + "sub_state": "unknown", + "main_pid": None, + "n_restarts": None, + "candidate_container": "unknown", + } + or ( + isinstance(fail_closed, dict) + and set(fail_closed) + == { + "status", + "service_posture", + "active_state", + "sub_state", + "main_pid", + "n_restarts", + "candidate_container", + } + and fail_closed.get("status") == "pass" + and fail_closed.get("service_posture") == "inactive" + and fail_closed.get("active_state") == "inactive" + and fail_closed.get("sub_state") == "dead" + and fail_closed.get("main_pid") == 0 + and isinstance(fail_closed.get("n_restarts"), int) + and not isinstance(fail_closed.get("n_restarts"), bool) + and fail_closed["n_restarts"] >= 0 + and fail_closed.get("candidate_container") == "absent" + ) + ) + failure_contract = ( + plan["operation"] == "install" + and ( + (failure_pair in POST_INSTALL_FAILURES and (rollback_passed or rollback_failed)) + or (failure_pair in installer.VERIFICATION_FAILURES and rollback_failed) + ) + and fail_closed_contract + ) _require( - result.get("schema") == REMOTE_FAILURE_SCHEMA - and result.get("status") == "fail" + result.get("status") == "fail" and result.get("operation") == plan["operation"] and result.get("request_id") == plan["request_id"] and result.get("runner_revision") == plan["runner_revision"] @@ -1069,15 +1178,7 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any]) "helper_sha256": request["helper_sha256"], "manifest_sha256": request["manifest_sha256"], } - and (result.get("failure_stage"), result.get("failure_reason")) in installer.VERIFICATION_FAILURES - and rollback - in ( - {"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"}, - {"status": "pass", "mode": "explicit", "restored_posture": "active"}, - {"status": "pass", "mode": "explicit", "restored_posture": "inactive"}, - {"status": "pass", "mode": "explicit", "restored_posture": "absent"}, - {"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"}, - ) + and failure_contract and result.get("authority") == { "sending_enabled": False, @@ -1089,33 +1190,8 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any]) ) -def _write_result_receipt( - plan: Mapping[str, Any], - result: Mapping[str, Any], - *, - status: str = "pass", -) -> tuple[Path, dict[str, Any]]: - _require(status in {"pass", "fail"} and result.get("status") == status, "local result status was invalid") - archive = Path(str(plan["request"]["archive"])) - output = archive.parent / f"livingip-leoclean-nosend-result-{plan['request_id']}.json" +def _publish_local_receipt(output: Path, stable: Mapping[str, Any]) -> dict[str, Any]: _require(not output.exists() and not output.is_symlink(), "local result receipt already existed") - stable = { - "schema": RESULT_RECEIPT_SCHEMA, - "status": status, - "operation": plan["operation"], - "request_id": plan["request_id"], - "runner_revision": plan["runner_revision"], - "release_revision": plan["release_revision"], - "request": { - "archive_sha256": plan["request"]["archive_sha256"], - "helper_sha256": plan["request"]["helper_sha256"], - "manifest_sha256": plan["request"]["manifest_sha256"], - }, - "release": dict(plan["release"]), - "release_artifact": dict(plan["release_artifact"]), - "expected": dict(plan["expected"]), - "remote_result": dict(result), - } receipt = {**stable, "receipt_sha256": package.canonical_sha256(stable)} content = _canonical_json(receipt) descriptor, temporary_name = tempfile.mkstemp(prefix=".iap-result-", dir=output.parent) @@ -1142,6 +1218,74 @@ def _write_result_receipt( temporary.unlink() except FileNotFoundError: pass + return receipt + + +def _write_result_receipt( + plan: Mapping[str, Any], + result: Mapping[str, Any], + *, + status: str = "pass", +) -> tuple[Path, dict[str, Any]]: + _require(status in {"pass", "fail"} and result.get("status") == status, "local result status was invalid") + archive = Path(str(plan["request"]["archive"])) + output = archive.parent / f"livingip-leoclean-nosend-result-{plan['request_id']}.json" + stable = { + "schema": RESULT_RECEIPT_SCHEMA if status == "pass" else FAILURE_RECEIPT_SCHEMA, + "status": status, + "operation": plan["operation"], + "request_id": plan["request_id"], + "runner_revision": plan["runner_revision"], + "release_revision": plan["release_revision"], + "request": { + "archive_sha256": plan["request"]["archive_sha256"], + "helper_sha256": plan["request"]["helper_sha256"], + "manifest_sha256": plan["request"]["manifest_sha256"], + }, + "release": dict(plan["release"]), + "release_artifact": dict(plan["release_artifact"]), + "expected": dict(plan["expected"]), + "remote_result": dict(result), + } + receipt = _publish_local_receipt(output, stable) + return output, receipt + + +def _write_local_failure_receipt( + plan: Mapping[str, Any], + *, + failure_stage: str, + failure_reason: str, + cleanup_status: str, +) -> tuple[Path, dict[str, Any]]: + _require( + (failure_stage, failure_reason) in LOCAL_FAILURES and cleanup_status in {"pass", "fail"}, + "local failure classification was invalid", + ) + archive = Path(str(plan["request"]["archive"])) + output = archive.parent / f"livingip-leoclean-nosend-result-{plan['request_id']}.json" + stable = { + "schema": LOCAL_FAILURE_RECEIPT_SCHEMA, + "status": "fail", + "operation": plan["operation"], + "request_id": plan["request_id"], + "runner_revision": plan["runner_revision"], + "release_revision": plan["release_revision"], + "request": { + "archive_sha256": plan["request"]["archive_sha256"], + "helper_sha256": plan["request"]["helper_sha256"], + "manifest_sha256": plan["request"]["manifest_sha256"], + }, + "release": dict(plan["release"]), + "release_artifact": dict(plan["release_artifact"]), + "expected": dict(plan["expected"]), + "failure_stage": failure_stage, + "failure_reason": failure_reason, + "cleanup": {"status": cleanup_status, "mode": "failed_transfer"}, + "remote_state": "unknown", + "requested_authority": dict(plan["authority"]), + } + receipt = _publish_local_receipt(output, stable) return output, receipt @@ -1177,6 +1321,7 @@ def execute_plan( ) operation = plan.get("operation") _require(operation in {"install", "verify", "rollback"}, "operation was not executable") + _require(plan.get("authority") == _expected_authority(operation), "request authority was not exact") request_id = plan.get("request_id") runner_revision = plan.get("runner_revision") release_revision = plan.get("release_revision") @@ -1261,9 +1406,12 @@ def execute_plan( scp, ssh, cleanup = expected_commands validate_remote_main(runner_revision, runner=git_runner) environment = dict(gcloud_env or gcloud_environment()) + failure_stage = "iap_transport" + failure_reason = "upload_failed" try: try: _run_bounded(list(scp), runner=runner, label="IAP upload", environment=environment) + failure_reason = "remote_command_interrupted" completed = _run_bounded( list(ssh), runner=runner, @@ -1271,8 +1419,17 @@ def execute_plan( environment=environment, accepted_returncodes=(0, 65), ) + failure_stage = "remote_receipt" + if not completed.stdout: + failure_reason = "absent" + raise IapRequestError("remote result was absent") + failure_reason = "malformed" result = _strict_json(completed.stdout, "remote result") - encoded = json.dumps(result, allow_nan=False, sort_keys=True).casefold() + try: + encoded = json.dumps(result, allow_nan=False, sort_keys=True).casefold() + except (TypeError, ValueError, RecursionError) as exc: + raise IapRequestError("remote result could not be normalized") from exc + failure_reason = "contract_mismatch" _require( not any(marker in encoded for marker in FORBIDDEN_OUTPUT_MARKERS), "remote result was not sanitized", @@ -1282,6 +1439,7 @@ def execute_plan( else: validate_remote_result(plan, result) except IapRequestError as exc: + cleanup_status = "pass" try: _run_bounded( list(cleanup), @@ -1289,9 +1447,19 @@ def execute_plan( label="IAP failed-transfer cleanup", environment=environment, ) - except IapRequestError as cleanup_exc: - raise IapRequestError("IAP operation failed and remote upload cleanup was unproven") from cleanup_exc - raise exc + except IapRequestError: + cleanup_status = "fail" + receipt_path, _receipt = _write_local_failure_receipt( + plan, + failure_stage=failure_stage, + failure_reason=failure_reason, + cleanup_status=cleanup_status, + ) + raise IapRequestError( + "IAP operation failed closed " + f"at {failure_stage}/{failure_reason}; " + f"sanitized receipt: {receipt_path}" + ) from exc finally: _cleanup_local_ssh_key(ssh_key) if completed.returncode == 65: diff --git a/tests/test_gcp_leoclean_nosend_iap_runner.py b/tests/test_gcp_leoclean_nosend_iap_runner.py index dcd16c2..897b72a 100644 --- a/tests/test_gcp_leoclean_nosend_iap_runner.py +++ b/tests/test_gcp_leoclean_nosend_iap_runner.py @@ -159,6 +159,106 @@ def verifier_failure_result( } +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", + "restored_posture": "unknown", + } + result["fail_closed"] = { + "status": "pass", + "service_posture": "inactive", + "active_state": "inactive", + "sub_state": "dead", + "main_pid": 0, + "n_restarts": 0, + "candidate_container": "absent", + } + 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, + "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) @@ -183,6 +283,10 @@ 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( @@ -499,6 +603,125 @@ def test_local_result_receipt_is_private_and_self_hashed(tmp_path: Path) -> None 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, @@ -532,6 +755,94 @@ def test_execute_records_sanitized_verifier_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) @@ -541,6 +852,15 @@ def test_remote_failure_rejects_unallowlisted_reason(tmp_path: Path) -> None: 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="did not match"): + local.validate_remote_failure(plan, failure) + + def test_remote_main_emits_structured_failure_on_stdout( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, @@ -570,8 +890,16 @@ def test_execute_failure_does_not_forward_command_output(monkeypatch: pytest.Mon 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: @@ -595,7 +923,16 @@ def test_execute_rejects_unsanitized_remote_result(monkeypatch: pytest.MonkeyPat 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: @@ -606,6 +943,22 @@ def test_execute_rederives_commands_and_rejects_plan_tampering(tmp_path: Path) - 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]] = [] @@ -838,6 +1191,7 @@ def test_post_install_proof_failure_executes_explicit_rollback( installer_module = SimpleNamespace( SYSTEMCTL="/usr/bin/systemctl", + VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES, InstallPaths=Paths, validate_release_bundle=lambda _bundle: (release, b"unit"), _validate_source_binding=lambda _binding: None, @@ -881,7 +1235,7 @@ def test_post_install_proof_failure_executes_explicit_rollback( return completed(argv, returncode=1) return completed(argv) - with pytest.raises(remote.RemoteError, match="install_failed_rolled_back"): + with pytest.raises(remote.RemoteOperationFailure) as caught: remote.execute_request( manifest, tmp_path, @@ -891,6 +1245,180 @@ def test_post_install_proof_failure_executes_explicit_rollback( 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", + "restored_posture": "absent", + } + 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", + } + + +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": "", + }, + ), + ] + ) + installer_module = SimpleNamespace( + SYSTEMCTL="/usr/bin/systemctl", + InstallPaths=Paths, + validate_release_bundle=lambda _bundle: (release, b"unit"), + _validate_source_binding=lambda _binding: None, + install_release=lambda *_args, **_kwargs: {"status": "pass"}, + 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: (_ for _ in ()).throw(RuntimeError("rollback failed")), + _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", + "restored_posture": "unknown", + } + assert result["fail_closed"] == { + "status": "pass", + "service_posture": "inactive", + "active_state": "inactive", + "sub_state": "dead", + "main_pid": 0, + "n_restarts": 0, + "candidate_container": "absent", + } + 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", + } @pytest.mark.parametrize(