From e5a5cc0ca2260161a84afd9c829ea8701a72b065 Mon Sep 17 00:00:00 2001 From: fwazb Date: Thu, 23 Jul 2026 16:25:17 -0700 Subject: [PATCH] Converge leoclean no-send installation state --- docs/gcp-leoclean-nosend-service-package.md | 26 +- ops/gcp_leoclean_nosend_iap_remote.py | 510 +++++++++++++-- ops/install_gcp_leoclean_nosend_release.py | 503 ++++++++++++++- ops/run_gcp_leoclean_nosend_iap.py | 162 ++++- tests/test_gcp_leoclean_nosend_iap_runner.py | 622 ++++++++++++++++++- tests/test_gcp_leoclean_nosend_installer.py | 584 ++++++++++++++++- 6 files changed, 2268 insertions(+), 139 deletions(-) diff --git a/docs/gcp-leoclean-nosend-service-package.md b/docs/gcp-leoclean-nosend-service-package.md index d5f7fee..f1a4286 100644 --- a/docs/gcp-leoclean-nosend-service-package.md +++ b/docs/gcp-leoclean-nosend-service-package.md @@ -324,9 +324,22 @@ disappears it rechecks that the canonical name is still unbound before removing the CID file. It never removes a container, uses a mutable name as stop authority, or invokes `--force`. -An already installed release is verified without a restart. During an upgrade, -the installer accepts only a stable two-readback active/running or inactive/dead -prior posture. Before trusting the old `ExecStop`, it exact-binds effective +Install is convergent: it means ensuring the requested exact release is +installed, active, and verified. An exact active release is verified as a +no-op. An exact inactive release receives one controlled `start`, while a +missing or different release uses the replacement transaction. The persisted +install history is not rewritten by exact-release convergence. New replacement +history uses `livingip.leocleanNoSendInstallReceipt.v3`; existing +`livingip.leocleanNoSendInstallReceipt.v2` history remains readable. Each +invocation returns a separate +`livingip.leocleanNoSendInstallExecution.v1` result whose transition records +`verified_noop`, `controlled_start`, `installed`, or `replaced`, the prior +posture and restart counter, the service action, and whether managed files were +replaced. + +During an upgrade, the installer accepts only a stable two-readback +active/running or inactive/dead prior posture. Before trusting the old +`ExecStop`, it exact-binds effective `ExecStartPre`, `ExecStart`, `ExecStop`, runtime-directory and systemd security properties to the release and verifies the supervising PID's executable, command line and cgroup through `/proc`. It then stops the prior service through @@ -339,6 +352,13 @@ two readbacks. On failure it stops the candidate, restores the exact prior files, modes and active/inactive/absent posture, and reloads systemd. An incomplete rollback fails closed and is reported as such. +The IAP result and bounded failure contracts use the v2 result schemas. A +post-install failure restores the transition's proven prior posture without +consuming historical rollback for an exact-release no-op or controlled start. +Replacement failures use only the receipt-bound explicit rollback. Malformed +or missing transition data forbids historical rollback and instead contains +the service fail closed with a sanitized failure stage and reason. + This installer does not enable the service, provision IAM, roles or secrets, contact GCP or Cloud SQL, change Telegram traffic, touch the VPS, approve a proposal, or promote production. Those remain later, separately authorized diff --git a/ops/gcp_leoclean_nosend_iap_remote.py b/ops/gcp_leoclean_nosend_iap_remote.py index dcb4902..47888d4 100644 --- a/ops/gcp_leoclean_nosend_iap_remote.py +++ b/ops/gcp_leoclean_nosend_iap_remote.py @@ -23,9 +23,9 @@ from pathlib import Path 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" +RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v2" +FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapFailure.v2" +OPERATION_FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v2" PROJECT = "teleo-501523" ZONE = "europe-west6-a" INSTANCE = "teleo-staging-1" @@ -99,6 +99,164 @@ class RemoteOperationFailure(RemoteError): self.result = dict(result) +class PostureRecoveryFailure(RemoteError): + """A posture recovery failed after a bounded amount of service action.""" + + def __init__(self, *, executed: bool) -> None: + super().__init__("posture_recovery_failed") + self.executed = executed + + +class ExplicitRollbackFailure(RemoteError): + """An explicit rollback failed before or after entering its mutation phase.""" + + def __init__(self, *, executed: bool) -> None: + super().__init__("explicit_rollback_failed") + self.executed = executed + + +def _validate_installer_transition(value: Any) -> dict[str, Any]: + _require( + isinstance(value, dict) + and set(value) + == { + "outcome", + "prior_posture", + "prior_n_restarts", + "final_posture", + "service_action", + "managed_files_replaced", + }, + "installer transition was invalid", + ) + outcome = value.get("outcome") + prior_posture = value.get("prior_posture") + prior_n_restarts = value.get("prior_n_restarts") + service_action = value.get("service_action") + replaced = value.get("managed_files_replaced") + _require( + isinstance(prior_n_restarts, int) + and not isinstance(prior_n_restarts, bool) + and prior_n_restarts >= 0 + and value.get("final_posture") == "active" + and ( + (outcome, prior_posture, service_action, replaced) + == ("verified_noop", "active", "none", False) + or (outcome, prior_posture, service_action, replaced) + == ("controlled_start", "inactive", "start", False) + or (outcome, prior_posture, service_action, replaced) + == ("installed", "absent", "restart", True) + or ( + outcome == "replaced" + and prior_posture in {"active", "inactive"} + and service_action == "restart" + and replaced is True + ) + ), + "installer transition was invalid", + ) + return dict(value) + + +def _validate_install_execution_result( + value: Any, + *, + release: Mapping[str, Any], + installer_module: Any, +) -> dict[str, Any]: + _require( + isinstance(value, dict) + and set(value) + == { + "schema", + "status", + "outcome", + "release_sha256", + "persisted_receipt", + "transition", + "verification", + }, + "install execution result was invalid", + ) + transition = _validate_installer_transition(value.get("transition")) + persisted_receipt = value.get("persisted_receipt") + outcome = value.get("outcome") + _require( + value.get("schema") == installer_module.INSTALL_EXECUTION_RESULT_SCHEMA + and value.get("status") == "pass" + and outcome == transition["outcome"] + and value.get("release_sha256") == release["release_sha256"] + and isinstance(persisted_receipt, dict) + and set(persisted_receipt) == {"schema", "sha256"} + and persisted_receipt.get("schema") + in { + installer_module.LEGACY_INSTALL_RECEIPT_SCHEMA, + installer_module.INSTALL_RECEIPT_SCHEMA, + } + and isinstance(persisted_receipt.get("sha256"), str) + and HEX_64.fullmatch(persisted_receipt["sha256"]) is not None + and isinstance(value.get("verification"), dict) + and ( + outcome in {"verified_noop", "controlled_start"} + or ( + outcome in {"installed", "replaced"} + and persisted_receipt["schema"] == installer_module.INSTALL_RECEIPT_SCHEMA + ) + ), + "install execution result was invalid", + ) + return transition + + +def _rollback_summary( + *, + status: str, + mode: str, + executed: bool, + restored_posture: str, + prior_posture_preserved: bool, +) -> dict[str, Any]: + value = { + "status": status, + "mode": mode, + "executed": executed, + "restored_posture": restored_posture, + "prior_posture_preserved": prior_posture_preserved, + } + _require( + ( + status == "pass" + and ( + ( + mode in {"installer_automatic", "explicit"} + and executed is True + and restored_posture in {"active", "inactive", "absent"} + ) + or ( + mode == "posture_recovery" + and restored_posture in {"active", "inactive"} + ) + ) + and prior_posture_preserved is True + ) + or ( + status == "not_run" + and mode == "not_applicable" + and executed is False + and restored_posture == "unchanged" + and prior_posture_preserved is True + ) + or ( + status == "fail" + and mode in {"explicit", "posture_recovery"} + and restored_posture == "unknown" + and prior_posture_preserved is False + ), + "rollback summary was invalid", + ) + return value + + def _require(condition: bool, message: str) -> None: if not condition: raise RemoteError(message) @@ -716,20 +874,43 @@ def _execute_explicit_rollback( rollback = getattr(installer_module, "rollback_release", None) _require(callable(rollback), "reviewed installer lacked the explicit rollback interface") - result = rollback( - release_sha256, - paths=paths, - source_binding=source_binding, - runner=lambda argv: _run( + executed = False + + def contract_runner(argv: list[str]) -> subprocess.CompletedProcess[bytes]: + return _run( argv, runner=runner, label="rollback command", require_success=False, - ), - execute_rollback=True, - ) - _require(isinstance(result, dict) and result.get("status") == "pass", "explicit rollback did not pass") - return result + ) + + try: + plan = rollback( + release_sha256, + paths=paths, + source_binding=source_binding, + runner=contract_runner, + execute_rollback=False, + ) + _require( + isinstance(plan, dict) + and plan.get("status") == "pass" + and plan.get("mode") == "dry_run" + and plan.get("expected_release_sha256") == release_sha256, + "explicit rollback preflight did not pass", + ) + executed = True + result = rollback( + release_sha256, + paths=paths, + source_binding=source_binding, + runner=contract_runner, + execute_rollback=True, + ) + _require(isinstance(result, dict) and result.get("status") == "pass", "explicit rollback did not pass") + return result + except Exception as exc: + raise ExplicitRollbackFailure(executed=executed) from exc def _rollback_result( @@ -808,7 +989,7 @@ def _verification_failure_result( manifest: Mapping[str, Any], release: Mapping[str, Any], failure: Exception, - rollback: Mapping[str, str], + rollback: Mapping[str, Any], archive_sha256: str, helper_sha256: str, installer_module: Any, @@ -822,13 +1003,13 @@ def _verification_failure_result( "verification failure was not allowlisted", ) _require( - 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"}, + dict(rollback) + == _rollback_summary( + status=str(rollback.get("status")), + mode=str(rollback.get("mode")), + executed=rollback.get("executed") is True, + restored_posture=str(rollback.get("restored_posture")), + prior_posture_preserved=rollback.get("prior_posture_preserved") is True, ), "verification rollback summary was invalid", ) @@ -860,6 +1041,25 @@ def _verification_failure_result( } +def _installer_failure_rollback(failure: Exception) -> dict[str, Any]: + rollback_executed = getattr(failure, "rollback_executed", None) + restored_posture = getattr(failure, "restored_posture", None) + prior_posture_preserved = getattr(failure, "prior_posture_preserved", None) + _require( + isinstance(rollback_executed, bool) + and restored_posture in {"active", "inactive", "absent"} + and prior_posture_preserved is True, + "installer failure lacked a proven recovery posture", + ) + return _rollback_summary( + status="pass" if rollback_executed else "not_run", + mode="installer_automatic" if rollback_executed else "not_applicable", + executed=rollback_executed, + restored_posture=str(restored_posture) if rollback_executed else "unchanged", + prior_posture_preserved=True, + ) + + def _is_verification_failure(installer_module: Any, failure: Exception) -> bool: failure_type = getattr(installer_module, "VerificationError", None) return isinstance(failure_type, type) and isinstance(failure, failure_type) @@ -899,6 +1099,7 @@ def _rollback_proven_service(posture: str) -> dict[str, Any]: "main_pid": None, "n_restarts": None, "candidate_container": "not_observed", + "service_action_executed": False, } @@ -909,6 +1110,8 @@ def _fail_closed_inert_service( paths: Any, runner: Runner, ) -> dict[str, Any]: + service_action_executed = False + def contract_runner(argv: list[str]) -> subprocess.CompletedProcess[bytes]: return _run( argv, @@ -949,6 +1152,7 @@ def _fail_closed_inert_service( and verification.get("n_restarts") == before.state.get("NRestarts"), "running service identity changed before fail-closed stop", ) + service_action_executed = True systemctl("stop", contract_runner) after = stable_service_posture( release, @@ -979,6 +1183,7 @@ def _fail_closed_inert_service( "main_pid": 0, "n_restarts": after.state["NRestarts"], "candidate_container": "absent", + "service_action_executed": service_action_executed, } except Exception: return { @@ -989,16 +1194,128 @@ def _fail_closed_inert_service( "main_pid": None, "n_restarts": None, "candidate_container": "unknown", + "service_action_executed": service_action_executed, } +def _restore_transition_posture( + *, + installer_module: Any, + release: Mapping[str, Any], + transition: Mapping[str, Any], + paths: Any, + runner: Runner, +) -> dict[str, Any]: + checked = _validate_installer_transition(transition) + desired = str(checked["prior_posture"]) + _require( + checked["outcome"] in {"verified_noop", "controlled_start"} + and desired in {"active", "inactive"}, + "installer transition required release rollback", + ) + + def contract_runner(argv: list[str]) -> subprocess.CompletedProcess[bytes]: + return _run( + argv, + runner=runner, + label="prior posture recovery command", + require_success=False, + ) + + 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 posture recovery controls", + ) + executed = False + try: + observed = stable_service_posture( + release, + runner=contract_runner, + paths=paths, + sleeper=time.sleep, + ) + _require(observed.kind in {"active", "inactive"}, "managed service posture was not trusted") + if desired == "inactive": + if observed.kind == "active": + verification = verify_running_release( + release, + runner=contract_runner, + paths=paths, + sleeper=time.sleep, + ) + _require( + verification.get("main_pid") == observed.state.get("MainPID") + and verification.get("invocation_id") == observed.state.get("InvocationID") + and verification.get("n_restarts") == observed.state.get("NRestarts"), + "running service identity changed before posture recovery stop", + ) + executed = True + systemctl("stop", contract_runner) + final = stable_service_posture( + release, + runner=contract_runner, + paths=paths, + sleeper=time.sleep, + ) + _require( + final.kind == "inactive" + and final.state.get("NRestarts") == checked["prior_n_restarts"], + "prior inactive posture was not restored", + ) + require_candidate_absent(runner=contract_runner, paths=paths, sleeper=time.sleep) + else: + if observed.kind == "inactive": + require_candidate_absent(runner=contract_runner, paths=paths, sleeper=time.sleep) + executed = True + systemctl("start", contract_runner) + final = stable_service_posture( + release, + runner=contract_runner, + paths=paths, + sleeper=time.sleep, + ) + _require( + final.kind == "active" + and final.state.get("NRestarts") == checked["prior_n_restarts"], + "prior active posture was not restored", + ) + verification = verify_running_release( + release, + runner=contract_runner, + paths=paths, + sleeper=time.sleep, + ) + _require( + verification.get("main_pid") == final.state.get("MainPID") + and verification.get("invocation_id") == final.state.get("InvocationID") + and verification.get("n_restarts") == checked["prior_n_restarts"], + "restored active service identity was unstable", + ) + except Exception as exc: + raise PostureRecoveryFailure(executed=executed) from exc + return _rollback_summary( + status="pass", + mode="posture_recovery", + executed=executed, + restored_posture=desired, + prior_posture_preserved=True, + ) + + def _operational_failure_result( *, manifest: Mapping[str, Any], release: Mapping[str, Any], failure_stage: str, failure_reason: str, - rollback: Mapping[str, str], + rollback: Mapping[str, Any], fail_closed: Mapping[str, Any], archive_sha256: str, helper_sha256: str, @@ -1006,7 +1323,8 @@ def _operational_failure_result( ) -> dict[str, Any]: _require( manifest.get("operation") == "install" - and set(rollback) == {"status", "mode", "restored_posture"} + and set(rollback) + == {"status", "mode", "executed", "restored_posture", "prior_posture_preserved"} and ( (failure_stage, failure_reason) in POST_INSTALL_FAILURES or (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES @@ -1015,14 +1333,22 @@ def _operational_failure_result( ) rollback_passed = ( rollback.get("status") == "pass" - and rollback.get("mode") == "explicit" + and rollback.get("mode") in {"explicit", "posture_recovery"} + and isinstance(rollback.get("executed"), bool) and rollback.get("restored_posture") in {"active", "inactive", "absent"} + and ( + rollback.get("mode") == "explicit" + or rollback.get("restored_posture") in {"active", "inactive"} + ) + and rollback.get("prior_posture_preserved") is True + ) + rollback_failed = ( + rollback.get("status") == "fail" + and rollback.get("mode") in {"explicit", "posture_recovery"} + and isinstance(rollback.get("executed"), bool) + and rollback.get("restored_posture") == "unknown" + and rollback.get("prior_posture_preserved") is False ) - 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, @@ -1045,6 +1371,7 @@ def _operational_failure_result( "main_pid": None, "n_restarts": None, "candidate_container": "unknown", + "service_action_executed": fail_closed.get("service_action_executed"), }, { "status": "pass", @@ -1054,6 +1381,7 @@ def _operational_failure_result( "main_pid": 0, "n_restarts": fail_closed.get("n_restarts"), "candidate_container": "absent", + "service_action_executed": fail_closed.get("service_action_executed"), }, ) and ( @@ -1063,7 +1391,8 @@ def _operational_failure_result( and not isinstance(fail_closed.get("n_restarts"), bool) and fail_closed["n_restarts"] >= 0 ) - ), + ) + and isinstance(fail_closed.get("service_action_executed"), bool), "post-install fail-closed summary was invalid", ) return { @@ -1156,7 +1485,7 @@ def execute_request( manifest=manifest, release=release, failure=exc, - rollback={"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"}, + rollback=_installer_failure_rollback(exc), archive_sha256=archive_sha256, helper_sha256=helper_sha256, installer_module=installer_module, @@ -1189,8 +1518,13 @@ def execute_request( if operation == "install": failure_stage = "install_receipt" failure_reason = "install_receipt_invalid" + installer_transition: dict[str, Any] | None = None try: - _require(install_receipt.get("status") == "pass", "installation did not pass") + installer_transition = _validate_install_execution_result( + install_receipt, + release=release, + installer_module=installer_module, + ) failure_stage = "service_before_restart" failure_reason = "service_verification_failed" before = installer_module.verify_running_release( @@ -1248,21 +1582,76 @@ def execute_request( fallback_stage=failure_stage, fallback_reason=failure_reason, ) - try: - recovery = _execute_explicit_rollback( + if installer_transition is None: + fail_closed = _fail_closed_inert_service( installer_module=installer_module, - release_sha256=release["release_sha256"], + release=release, paths=paths, - source_binding=source_binding, runner=runner, ) - rollback_result = _rollback_result( - manifest=manifest, - release=release, - rollback=recovery, - archive_sha256=archive_sha256, - helper_sha256=helper_sha256, - ) + raise RemoteOperationFailure( + _operational_failure_result( + manifest=manifest, + release=release, + failure_stage="install_receipt", + failure_reason="install_receipt_invalid", + rollback=_rollback_summary( + status="fail", + mode="posture_recovery", + executed=bool(fail_closed["service_action_executed"]), + restored_posture="unknown", + prior_posture_preserved=False, + ), + fail_closed=fail_closed, + archive_sha256=archive_sha256, + helper_sha256=helper_sha256, + installer_module=installer_module, + ) + ) from exc + release_changing = installer_transition["outcome"] in {"installed", "replaced"} + recovery_mode = "explicit" if release_changing else "posture_recovery" + recovery_executed = False + try: + if release_changing: + try: + recovery = _execute_explicit_rollback( + installer_module=installer_module, + release_sha256=release["release_sha256"], + paths=paths, + source_binding=source_binding, + runner=runner, + ) + except ExplicitRollbackFailure as rollback_exc: + recovery_executed = rollback_exc.executed + raise + recovery_executed = True + rollback_result = _rollback_result( + manifest=manifest, + release=release, + rollback=recovery, + archive_sha256=archive_sha256, + helper_sha256=helper_sha256, + ) + rollback_summary = _rollback_summary( + status="pass", + mode="explicit", + executed=True, + restored_posture=str(rollback_result["service"]["restored_posture"]), + prior_posture_preserved=True, + ) + else: + try: + rollback_summary = _restore_transition_posture( + installer_module=installer_module, + release=release, + transition=installer_transition, + paths=paths, + runner=runner, + ) + except PostureRecoveryFailure as posture_exc: + recovery_executed = posture_exc.executed + raise + recovery_executed = bool(rollback_summary["executed"]) except Exception: fail_closed = _fail_closed_inert_service( installer_module=installer_module, @@ -1276,11 +1665,13 @@ def execute_request( release=release, failure_stage=bounded_stage, failure_reason=bounded_reason, - rollback={ - "status": "fail", - "mode": "explicit", - "restored_posture": "unknown", - }, + rollback=_rollback_summary( + status="fail", + mode=recovery_mode, + executed=recovery_executed, + restored_posture="unknown", + prior_posture_preserved=False, + ), fail_closed=fail_closed, archive_sha256=archive_sha256, helper_sha256=helper_sha256, @@ -1293,28 +1684,20 @@ def execute_request( manifest=manifest, release=release, failure=exc, - rollback={ - "status": "pass", - "mode": "explicit", - "restored_posture": rollback_result["service"]["restored_posture"], - }, + rollback=rollback_summary, archive_sha256=archive_sha256, helper_sha256=helper_sha256, installer_module=installer_module, ) ) from exc - restored_posture = str(rollback_result["service"]["restored_posture"]) + restored_posture = str(rollback_summary["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, - }, + rollback=rollback_summary, fail_closed=_rollback_proven_service(restored_posture), archive_sha256=archive_sha256, helper_sha256=helper_sha256, @@ -1323,6 +1706,7 @@ def execute_request( ) from exc service = { "name": SERVICE, + "installer_transition": installer_transition, "before_restart": before, "after_restart": after, } @@ -1346,7 +1730,13 @@ def execute_request( manifest=manifest, release=release, failure=exc, - rollback={"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"}, + rollback=_rollback_summary( + status="not_run", + mode="not_applicable", + executed=False, + restored_posture="unchanged", + prior_posture_preserved=True, + ), archive_sha256=archive_sha256, helper_sha256=helper_sha256, installer_module=installer_module, diff --git a/ops/install_gcp_leoclean_nosend_release.py b/ops/install_gcp_leoclean_nosend_release.py index ee60069..a21707f 100644 --- a/ops/install_gcp_leoclean_nosend_release.py +++ b/ops/install_gcp_leoclean_nosend_release.py @@ -30,7 +30,9 @@ if str(CONTRACT_ROOT) not in sys.path: from ops import gcp_leoclean_nosend_package as package # noqa: E402 from ops import leoclean_nosend_service_control as service_control # noqa: E402 -INSTALL_RECEIPT_SCHEMA = "livingip.leocleanNoSendInstallReceipt.v2" +LEGACY_INSTALL_RECEIPT_SCHEMA = "livingip.leocleanNoSendInstallReceipt.v2" +INSTALL_RECEIPT_SCHEMA = "livingip.leocleanNoSendInstallReceipt.v3" +INSTALL_EXECUTION_RESULT_SCHEMA = "livingip.leocleanNoSendInstallExecution.v1" ROLLBACK_STATE_SCHEMA = "livingip.leocleanNoSendRollbackState.v1" ROLLBACK_RESULT_SCHEMA = "livingip.leocleanNoSendRollbackResult.v1" SERVICE = package.SERVICE @@ -171,12 +173,19 @@ class VerificationError(InstallError): failure_stage: str, failure_reason: str, message: str = "release verification failed", + *, + rollback_executed: bool | None = None, + restored_posture: str | None = None, + prior_posture_preserved: bool | None = None, ) -> None: if (failure_stage, failure_reason) not in VERIFICATION_FAILURES: raise ValueError("verification failure code was not allowlisted") super().__init__(message) self.failure_stage = failure_stage self.failure_reason = failure_reason + self.rollback_executed = rollback_executed + self.restored_posture = restored_posture + self.prior_posture_preserved = prior_posture_preserved class AtomicWriteError(InstallError): @@ -284,6 +293,93 @@ class ServicePosture: state: dict[str, Any] +def _install_transition( + outcome: str, + *, + prior_posture: str, + prior_n_restarts: int, + service_action: str, + managed_files_replaced: bool, +) -> dict[str, Any]: + _require( + outcome in {"verified_noop", "controlled_start", "installed", "replaced"} + and prior_posture in {"absent", "inactive", "active"} + and isinstance(prior_n_restarts, int) + and not isinstance(prior_n_restarts, bool) + and prior_n_restarts >= 0 + and service_action in {"none", "start", "restart"} + and ( + (outcome == "verified_noop" and prior_posture == "active" and service_action == "none") + or (outcome == "controlled_start" and prior_posture == "inactive" and service_action == "start") + or (outcome == "installed" and prior_posture == "absent" and service_action == "restart") + or (outcome == "replaced" and prior_posture in {"inactive", "active"} and service_action == "restart") + ) + and managed_files_replaced == (outcome in {"installed", "replaced"}), + "installer transition was invalid", + ) + return { + "outcome": outcome, + "prior_posture": prior_posture, + "prior_n_restarts": prior_n_restarts, + "final_posture": "active", + "service_action": service_action, + "managed_files_replaced": managed_files_replaced, + } + + +def _install_execution_result( + *, + outcome: str, + release: Mapping[str, Any], + receipt: Mapping[str, Any], + receipt_raw: bytes, + transition: Mapping[str, Any], + verification: Mapping[str, Any], +) -> dict[str, Any]: + _require( + outcome in {"verified_noop", "controlled_start", "installed", "replaced"} + and receipt.get("status") == "pass" + and receipt.get("release_sha256") == release["release_sha256"] + and ( + ( + outcome in {"verified_noop", "controlled_start"} + and receipt.get("schema") in {LEGACY_INSTALL_RECEIPT_SCHEMA, INSTALL_RECEIPT_SCHEMA} + ) + or ( + outcome in {"installed", "replaced"} + and receipt.get("schema") == INSTALL_RECEIPT_SCHEMA + and receipt.get("outcome") == outcome + ) + ), + "exact-release execution result was invalid", + ) + return { + "schema": INSTALL_EXECUTION_RESULT_SCHEMA, + "status": "pass", + "outcome": outcome, + "release_sha256": release["release_sha256"], + "persisted_receipt": { + "schema": receipt["schema"], + "sha256": _sha256_bytes(receipt_raw), + }, + "transition": dict(transition), + "verification": dict(verification), + } + + +def _attach_recovery( + failure: Exception, + *, + rollback_executed: bool, + restored_posture: str, + prior_posture_preserved: bool, +) -> None: + if isinstance(failure, VerificationError): + failure.rollback_executed = rollback_executed + failure.restored_posture = restored_posture + failure.prior_posture_preserved = prior_posture_preserved + + def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() @@ -1558,17 +1654,228 @@ def _read_rollback_state(paths: InstallPaths, release_sha256: str) -> tuple[dict return value, raw +def _validate_current_install_receipt_contract( + receipt: Mapping[str, Any], + *, + release: Mapping[str, Any], + rollback_state: Mapping[str, Any], +) -> None: + _require( + set(receipt) + == { + "schema", + "status", + "outcome", + "transition", + "installer_source", + "release_sha256", + "image", + "installed_files", + "service_before", + "verification", + "rollback", + "authority", + }, + "installed receipt fields were not exact", + ) + rollback = receipt.get("rollback") + _require( + isinstance(rollback, dict) + and set(rollback) + == { + "previous_release_present", + "previous_service_active", + "previous_service_posture", + "automatic_on_failure", + "state_path", + "state_sha256", + } + and isinstance(rollback.get("previous_release_present"), bool) + and isinstance(rollback.get("previous_service_active"), bool) + and isinstance(rollback.get("automatic_on_failure"), bool) + and rollback["automatic_on_failure"] is True, + "installed receipt rollback fields were invalid", + ) + previous_files = rollback_state["previous_files"] + previous_service = rollback_state["previous_service"] + previous_release_present = all( + _snapshot_from_json(previous_files[role], label=f"rollback {role} snapshot").existed + for role in sorted(previous_files) + ) + prior_posture = previous_service["posture"] + prior_n_restarts = previous_service["n_restarts"] + outcome = receipt.get("outcome") + _require( + ( + outcome == "installed" + and prior_posture == "absent" + and previous_release_present is False + ) + or ( + outcome == "replaced" + and prior_posture in {"active", "inactive"} + and previous_release_present is True + ), + "installed receipt outcome differed from rollback state", + ) + _require( + rollback["previous_release_present"] is previous_release_present + and rollback["previous_service_active"] is (prior_posture == "active") + and rollback["previous_service_posture"] == prior_posture, + "installed receipt prior posture differed from rollback state", + ) + service_before = receipt.get("service_before") + _require( + isinstance(service_before, dict) + and set(service_before) + == { + "load_state", + "active_state", + "sub_state", + "main_pid", + "n_restarts", + } + and isinstance(service_before.get("main_pid"), int) + and not isinstance(service_before.get("main_pid"), bool) + and service_before["main_pid"] >= 0 + and isinstance(service_before.get("n_restarts"), int) + and not isinstance(service_before.get("n_restarts"), bool) + and service_before["n_restarts"] >= 0 + and service_before["active_state"] == previous_service["active_state"] + and service_before["sub_state"] == previous_service["sub_state"] + and service_before["n_restarts"] == prior_n_restarts, + "installed receipt service posture differed from rollback state", + ) + _require( + ( + prior_posture == "absent" + and service_before["load_state"] == "not-found" + and service_before["main_pid"] == 0 + ) + or ( + prior_posture == "inactive" + and service_before["load_state"] == "loaded" + and service_before["main_pid"] == 0 + ) + or ( + prior_posture == "active" + and service_before["load_state"] == "loaded" + and service_before["main_pid"] > 1 + ), + "installed receipt load or process posture differed from rollback state", + ) + _require( + receipt.get("transition") + == _install_transition( + str(outcome), + prior_posture=prior_posture, + prior_n_restarts=prior_n_restarts, + service_action="restart", + managed_files_replaced=True, + ), + "installed receipt transition was invalid", + ) + verification = receipt.get("verification") + _require( + isinstance(verification, dict) + and set(verification) + == { + "active_state", + "sub_state", + "main_pid", + "n_restarts", + "invocation_id", + "supervising_process", + "container_id", + "container_pid", + "health", + "validated_process_environment_fields", + "dropins", + "environment_files", + } + and verification.get("active_state") == "active" + and verification.get("sub_state") == "running" + and isinstance(verification.get("main_pid"), int) + and not isinstance(verification.get("main_pid"), bool) + and verification["main_pid"] > 1 + and isinstance(verification.get("n_restarts"), int) + and not isinstance(verification.get("n_restarts"), bool) + and verification["n_restarts"] >= 0 + and isinstance(verification.get("invocation_id"), str) + and HEX_32.fullmatch(verification["invocation_id"]) is not None + and isinstance(verification.get("container_id"), str) + and HEX_64.fullmatch(verification["container_id"]) is not None + and isinstance(verification.get("container_pid"), int) + and not isinstance(verification.get("container_pid"), bool) + and verification["container_pid"] > 1 + and verification.get("health") == "healthy" + and verification.get("dropins") == "absent" + and verification.get("environment_files") == "absent", + "installed receipt verification was invalid", + ) + supervising_process = verification["supervising_process"] + environment_fields = verification["validated_process_environment_fields"] + expected_start = shlex.split(_rendered_service_fields(release)["ExecStart"][0]) + expected_executable = expected_start[0] + expected_cmdline = b"\0".join(item.encode("utf-8") for item in expected_start) + b"\0" + _require( + isinstance(supervising_process, dict) + and set(supervising_process) == {"executable", "cmdline_sha256", "cgroup"} + and isinstance(supervising_process.get("executable"), str) + and supervising_process["executable"] in {expected_executable, str(Path(expected_executable).resolve())} + and isinstance(supervising_process.get("cmdline_sha256"), str) + and supervising_process["cmdline_sha256"] == _sha256_bytes(expected_cmdline) + and supervising_process.get("cgroup") == f"/system.slice/{SERVICE}" + and isinstance(environment_fields, list) + and all(isinstance(field, str) and field for field in environment_fields) + and environment_fields == sorted(set(environment_fields)) + and "PATH" in environment_fields, + "installed receipt verification detail was invalid", + ) + _require( + receipt.get("authority") + == { + "image_pulled": False, + "cloud_or_database_authority_changed": False, + "transport_authority_changed": False, + }, + "installed receipt authority was invalid", + ) + installed_files = receipt.get("installed_files") + _require( + isinstance(installed_files, dict) + and set(installed_files) + == { + "unit_sha256", + "helper_sha256", + "release_descriptor_sha256", + "rollback_state_sha256", + }, + "installed receipt file fields were not exact", + ) + + def _validate_install_receipt( receipt: Mapping[str, Any], *, release: Mapping[str, Any], paths: InstallPaths, + rollback_state: Mapping[str, Any], rollback_state_raw: bytes, ) -> None: installed_files = receipt.get("installed_files") rollback = receipt.get("rollback") + schema = receipt.get("schema") + legacy_result_valid = schema == LEGACY_INSTALL_RECEIPT_SCHEMA and receipt.get("result") == "installed" + if schema == INSTALL_RECEIPT_SCHEMA: + _validate_current_install_receipt_contract( + receipt, + release=release, + rollback_state=rollback_state, + ) + current_outcome_valid = schema == INSTALL_RECEIPT_SCHEMA _require( - receipt.get("schema") == INSTALL_RECEIPT_SCHEMA + (legacy_result_valid or current_outcome_valid) and receipt.get("status") == "pass" and receipt.get("release_sha256") == release["release_sha256"] and isinstance(installed_files, dict) @@ -1644,7 +1951,6 @@ def _validate_managed_prior_state(paths: InstallPaths, helper_source: bytes) -> _require(paths.unit.read_bytes() == package.render_systemd_unit(release).encode(), "installed unit drifted") _require(paths.helper.is_file() and not paths.helper.is_symlink(), "installed helper was unsafe") rollback_state, rollback_state_raw = _read_rollback_state(paths, release["release_sha256"]) - del rollback_state receipt, _receipt_raw = _read_strict_json_file( paths.receipt, label="install receipt", @@ -1655,6 +1961,7 @@ def _validate_managed_prior_state(paths: InstallPaths, helper_source: bytes) -> receipt, release=release, paths=paths, + rollback_state=rollback_state, rollback_state_raw=rollback_state_raw, ) _require(bool(helper_source), "service-control helper source was empty") @@ -1665,7 +1972,7 @@ def _systemctl(action: str, runner: Runner) -> None: if action == "daemon-reload": argv = [SYSTEMCTL, "daemon-reload"] else: - _require(action in {"restart", "stop", "reset-failed"}, "systemd action was not allowed") + _require(action in {"start", "restart", "stop", "reset-failed"}, "systemd action was not allowed") argv = [SYSTEMCTL, action, SERVICE] _run_checked(argv, runner, label=f"systemd {action}") @@ -1817,11 +2124,26 @@ def _install_release_transaction( _ensure_directory(paths.state_dir, 0o700, created) _ensure_directory(paths.docker_config, 0o700, created) _require(not any(paths.docker_config.iterdir()), "service Docker config directory was not empty") - _validate_local_image(release, runner, paths) + try: + _validate_local_image(release, runner, paths) + except VerificationError: + raise + except InstallError as exc: + raise VerificationError( + "image_identity", + "image_contract_mismatch", + str(exc), + ) from exc except Exception as exc: cleanup_complete = _cleanup_created_directories(created) if not cleanup_complete: raise InstallError("preflight failed and cleanup was incomplete") from exc + _attach_recovery( + exc, + rollback_executed=False, + restored_posture=prior_posture.kind, + prior_posture_preserved=True, + ) raise snapshots = { @@ -1841,19 +2163,133 @@ def _install_release_transaction( and snapshots[paths.helper].content == helper_content ) if exact_already_installed: - verified = verify_running_release(release, runner=runner, paths=paths, sleeper=sleeper) - existing_receipt = package.load_json(paths.receipt, "install receipt") - return {**existing_receipt, "result": "already_installed", "verification": verified} + exact_snapshots = _current_managed_snapshots(paths) + exact_release, existing_receipt, _exact_rollback = _validate_managed_snapshots( + exact_snapshots, + paths=paths, + ) + _require(exact_release == release, "exact managed release changed before convergence") + receipt_raw = exact_snapshots["receipt"].content + if prior_posture.kind == "active": + try: + verified = verify_running_release(release, runner=runner, paths=paths, sleeper=sleeper) + _verification_require( + verified["main_pid"] == before["MainPID"] + and verified["n_restarts"] == before["NRestarts"] + and verified["invocation_id"] == before["InvocationID"], + "startup_identity", + "service_identity_changed", + "active exact release changed during verification", + ) + _assert_snapshot_identities(exact_snapshots, paths=paths) + except Exception as exc: + _attach_recovery( + exc, + rollback_executed=False, + restored_posture="active", + prior_posture_preserved=True, + ) + raise + transition = _install_transition( + "verified_noop", + prior_posture="active", + prior_n_restarts=before["NRestarts"], + service_action="none", + managed_files_replaced=False, + ) + return _install_execution_result( + outcome="verified_noop", + release=release, + receipt=existing_receipt, + receipt_raw=receipt_raw, + transition=transition, + verification=verified, + ) + + _require(prior_posture.kind == "inactive", "exact managed release posture was invalid") + start_attempted = False + try: + _require_candidate_absent(runner=runner, paths=paths, sleeper=sleeper) + start_attempted = True + try: + _systemctl("start", runner) + except InstallError as exc: + raise VerificationError( + "service_activation", + "service_not_active", + "controlled service start failed", + ) from exc + verified = verify_running_release(release, runner=runner, paths=paths, sleeper=sleeper) + _verification_require( + verified["n_restarts"] == before["NRestarts"], + "startup_identity", + "service_identity_changed", + "controlled service start changed the restart counter", + ) + _assert_snapshot_identities(exact_snapshots, paths=paths) + transition = _install_transition( + "controlled_start", + prior_posture="inactive", + prior_n_restarts=before["NRestarts"], + service_action="start", + managed_files_replaced=False, + ) + return _install_execution_result( + outcome="controlled_start", + release=release, + receipt=existing_receipt, + receipt_raw=receipt_raw, + transition=transition, + verification=verified, + ) + except Exception as exc: + try: + if start_attempted: + _systemctl("stop", runner) + _require_candidate_absent(runner=runner, paths=paths, sleeper=sleeper) + restored_posture = _stable_service_posture( + release, + runner=runner, + paths=paths, + sleeper=sleeper, + ) + _require( + restored_posture.kind == "inactive" + and restored_posture.state["NRestarts"] == before["NRestarts"], + "controlled start rollback did not restore the prior inactive posture", + ) + except InstallError as recovery_exc: + raise InstallError("controlled service start failed and cleanup was incomplete") from recovery_exc + _attach_recovery( + exc, + rollback_executed=start_attempted, + restored_posture="inactive", + prior_posture_preserved=True, + ) + if isinstance(exc, InstallError): + raise + raise InstallError("controlled service start failed and prior posture was restored") from exc if prior_active: _require(prior_release is not None, "active managed service had no prior release") - prior_verification = verify_running_release(prior_release, runner=runner, paths=paths, sleeper=sleeper) - _require( - prior_verification["main_pid"] == before["MainPID"] - and prior_verification["n_restarts"] == before["NRestarts"] - and prior_verification["invocation_id"] == before["InvocationID"], - "active prior service changed before the trusted stop", - ) + try: + prior_verification = verify_running_release(prior_release, runner=runner, paths=paths, sleeper=sleeper) + _verification_require( + prior_verification["main_pid"] == before["MainPID"] + and prior_verification["n_restarts"] == before["NRestarts"] + and prior_verification["invocation_id"] == before["InvocationID"], + "startup_identity", + "service_identity_changed", + "active prior service changed before the trusted stop", + ) + except Exception as exc: + _attach_recovery( + exc, + rollback_executed=False, + restored_posture="active", + prior_posture_preserved=True, + ) + raise candidate_may_be_active = False try: @@ -1893,10 +2329,19 @@ def _install_release_transaction( ) attempted_paths.add(paths.rollback_state) _publish_managed_file(paths.rollback_state, rollback_state_bytes, 0o600, installed_identities) + outcome = "installed" if prior_release is None else "replaced" + transition = _install_transition( + outcome, + prior_posture=prior_posture.kind, + prior_n_restarts=before["NRestarts"], + service_action="restart", + managed_files_replaced=True, + ) receipt = { "schema": INSTALL_RECEIPT_SCHEMA, "status": "pass", - "result": "installed", + "outcome": outcome, + "transition": transition, "installer_source": dict(source_binding), "release_sha256": release["release_sha256"], "image": { @@ -1936,7 +2381,14 @@ def _install_release_transaction( receipt_bytes = _canonical_json(receipt) attempted_paths.add(paths.receipt) _publish_managed_file(paths.receipt, receipt_bytes, 0o600, installed_identities) - return receipt + return _install_execution_result( + outcome=outcome, + release=release, + receipt=receipt, + receipt_raw=receipt_bytes, + transition=transition, + verification=verified, + ) except Exception as exc: rollback_errors: list[str] = [] if candidate_may_be_active: @@ -1992,6 +2444,12 @@ def _install_release_transaction( rollback_errors.append("directory_cleanup") if rollback_errors: raise InstallError("installation failed and automatic rollback was incomplete") from exc + _attach_recovery( + exc, + rollback_executed=True, + restored_posture=prior_posture.kind, + prior_posture_preserved=True, + ) if isinstance(exc, InstallError): raise raise InstallError("installation failed and was rolled back") from exc @@ -2032,8 +2490,17 @@ def _validate_managed_snapshots( receipt = _json_from_snapshot(snapshots["receipt"], label="snapshot install receipt") installed_files = receipt.get("installed_files") rollback = receipt.get("rollback") + schema = receipt.get("schema") + legacy_result_valid = schema == LEGACY_INSTALL_RECEIPT_SCHEMA and receipt.get("result") == "installed" + if schema == INSTALL_RECEIPT_SCHEMA: + _validate_current_install_receipt_contract( + receipt, + release=release, + rollback_state=rollback_state, + ) + current_outcome_valid = schema == INSTALL_RECEIPT_SCHEMA _require( - receipt.get("schema") == INSTALL_RECEIPT_SCHEMA + (legacy_result_valid or current_outcome_valid) and receipt.get("status") == "pass" and receipt.get("release_sha256") == release_sha256 and isinstance(receipt.get("installer_source"), dict) diff --git a/ops/run_gcp_leoclean_nosend_iap.py b/ops/run_gcp_leoclean_nosend_iap.py index f616677..8dc8b0c 100644 --- a/ops/run_gcp_leoclean_nosend_iap.py +++ b/ops/run_gcp_leoclean_nosend_iap.py @@ -31,9 +31,9 @@ from ops import install_gcp_leoclean_nosend_release as installer # noqa: E402 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" +REMOTE_RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v2" +REMOTE_FAILURE_SCHEMA = "livingip.leocleanNoSendIapFailure.v2" +REMOTE_OPERATION_FAILURE_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v2" RESULT_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapResultReceipt.v1" FAILURE_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapFailureReceipt.v1" LOCAL_FAILURE_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapLocalFailureReceipt.v1" @@ -944,6 +944,113 @@ def _validate_database_summary(value: Any, request_id: str, expected_name: str) ) +def _validate_installer_transition(value: Any) -> None: + _require( + isinstance(value, dict) + and set(value) + == { + "outcome", + "prior_posture", + "prior_n_restarts", + "final_posture", + "service_action", + "managed_files_replaced", + } + and isinstance(value.get("prior_n_restarts"), int) + and not isinstance(value.get("prior_n_restarts"), bool) + and value["prior_n_restarts"] >= 0 + and value.get("final_posture") == "active" + and ( + ( + value.get("outcome"), + value.get("prior_posture"), + value.get("service_action"), + value.get("managed_files_replaced"), + ) + in { + ("verified_noop", "active", "none", False), + ("controlled_start", "inactive", "start", False), + ("installed", "absent", "restart", True), + } + or ( + value.get("outcome") == "replaced" + and value.get("prior_posture") in {"active", "inactive"} + and value.get("service_action") == "restart" + and value.get("managed_files_replaced") is True + ) + ), + "installer transition was invalid", + ) + + +def _validate_rollback_summary(value: Any) -> str: + _require( + isinstance(value, dict) + and set(value) + == { + "status", + "mode", + "executed", + "restored_posture", + "prior_posture_preserved", + }, + "remote rollback summary was invalid", + ) + installer_automatic = ( + value.get("status") == "pass" + and value.get("mode") == "installer_automatic" + and value.get("executed") is True + and value.get("restored_posture") in {"active", "inactive", "absent"} + and value.get("prior_posture_preserved") is True + ) + explicit = ( + value.get("status") == "pass" + and value.get("mode") == "explicit" + and value.get("executed") is True + and value.get("restored_posture") in {"active", "inactive", "absent"} + and value.get("prior_posture_preserved") is True + ) + posture_recovery = ( + value.get("status") == "pass" + and value.get("mode") == "posture_recovery" + and isinstance(value.get("executed"), bool) + and value.get("restored_posture") in {"active", "inactive"} + and value.get("prior_posture_preserved") is True + ) + not_run = value == { + "status": "not_run", + "mode": "not_applicable", + "executed": False, + "restored_posture": "unchanged", + "prior_posture_preserved": True, + } + explicit_failed = ( + value.get("status") == "fail" + and value.get("mode") == "explicit" + and isinstance(value.get("executed"), bool) + and value.get("restored_posture") == "unknown" + and value.get("prior_posture_preserved") is False + ) + posture_recovery_failed = ( + value.get("status") == "fail" + and value.get("mode") == "posture_recovery" + and isinstance(value.get("executed"), bool) + and value.get("restored_posture") == "unknown" + and value.get("prior_posture_preserved") is False + ) + kinds = { + "installer_automatic": installer_automatic, + "explicit": explicit, + "posture_recovery": posture_recovery, + "not_run": not_run, + "explicit_failed": explicit_failed, + "posture_recovery_failed": posture_recovery_failed, + } + matches = [kind for kind, matched in kinds.items() if matched] + _require(len(matches) == 1, "remote rollback summary was invalid") + return matches[0] + + def validate_remote_result(plan: Mapping[str, Any], result: Mapping[str, Any]) -> None: _require( set(result) @@ -999,10 +1106,12 @@ def validate_remote_result(plan: Mapping[str, Any], result: Mapping[str, Any]) - database = result["database"] if operation == "install": _require( - isinstance(service, dict) and set(service) == {"name", "before_restart", "after_restart"}, + isinstance(service, dict) + and set(service) == {"name", "installer_transition", "before_restart", "after_restart"}, "install service result was invalid", ) _require(service["name"] == SERVICE, "install service target differed") + _validate_installer_transition(service["installer_transition"]) _validate_verification(service["before_restart"], "pre-restart verification") _validate_verification(service["after_restart"], "post-restart verification") before = service["before_restart"] @@ -1079,15 +1188,21 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any]) schema = result.get("schema") if schema == REMOTE_FAILURE_SCHEMA: _require(set(result) == common_fields, "remote failure fields were not exact") + rollback_kind = _validate_rollback_summary(rollback) 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"}, + and ( + ( + plan["operation"] == "install" + and rollback_kind + in { + "installer_automatic", + "explicit", + "posture_recovery", + "not_run", + } + ) + or (plan["operation"] == "verify" and rollback_kind == "not_run") ) ) else: @@ -1096,19 +1211,11 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any]) "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", - } + rollback_kind = _validate_rollback_summary(rollback) + rollback_passed = rollback_kind in {"explicit", "posture_recovery"} + rollback_failed = rollback_kind in {"explicit_failed", "posture_recovery_failed"} fail_closed = result.get("fail_closed") + _require(isinstance(fail_closed, dict), "remote fail-closed summary was invalid") if rollback_passed: fail_closed_contract = fail_closed == { "status": "not_required", @@ -1118,6 +1225,7 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any]) "main_pid": None, "n_restarts": None, "candidate_container": "not_observed", + "service_action_executed": False, } else: fail_closed_contract = ( @@ -1130,6 +1238,7 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any]) "main_pid": None, "n_restarts": None, "candidate_container": "unknown", + "service_action_executed": fail_closed.get("service_action_executed"), } or ( isinstance(fail_closed, dict) @@ -1142,6 +1251,7 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any]) "main_pid", "n_restarts", "candidate_container", + "service_action_executed", } and fail_closed.get("status") == "pass" and fail_closed.get("service_posture") == "inactive" @@ -1152,8 +1262,14 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any]) and not isinstance(fail_closed.get("n_restarts"), bool) and fail_closed["n_restarts"] >= 0 and fail_closed.get("candidate_container") == "absent" + and isinstance(fail_closed.get("service_action_executed"), bool) ) ) + if isinstance(fail_closed, dict): + _require( + isinstance(fail_closed.get("service_action_executed"), bool), + "remote fail-closed action summary was invalid", + ) failure_contract = ( plan["operation"] == "install" and ( diff --git a/tests/test_gcp_leoclean_nosend_iap_runner.py b/tests/test_gcp_leoclean_nosend_iap_runner.py index 897b72a..46bd4d4 100644 --- a/tests/test_gcp_leoclean_nosend_iap_runner.py +++ b/tests/test_gcp_leoclean_nosend_iap_runner.py @@ -26,6 +26,24 @@ def completed( 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) @@ -148,7 +166,9 @@ def verifier_failure_result( "rollback": { "status": "pass", "mode": "installer_automatic", - "restored_posture": "prior", + "executed": True, + "restored_posture": "absent", + "prior_posture_preserved": True, }, "authority": { "sending_enabled": False, @@ -169,7 +189,9 @@ def operational_failure_result(plan: dict[str, Any]) -> dict[str, Any]: result["rollback"] = { "status": "fail", "mode": "explicit", + "executed": True, "restored_posture": "unknown", + "prior_posture_preserved": False, } result["fail_closed"] = { "status": "pass", @@ -179,6 +201,7 @@ def operational_failure_result(plan: dict[str, Any]) -> dict[str, Any]: "main_pid": 0, "n_restarts": 0, "candidate_container": "absent", + "service_action_executed": True, } return result @@ -221,6 +244,14 @@ def successful_remote_result(plan: dict[str, Any]) -> dict[str, Any]: 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, } @@ -857,10 +888,65 @@ def test_operational_failure_rejects_extra_rollback_fields(tmp_path: Path) -> No failure = operational_failure_result(plan) failure["rollback"]["credential"] = "not-allowed" - with pytest.raises(local.IapRequestError, match="did not match"): + 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, @@ -1030,10 +1116,17 @@ def test_outer_lock_and_reused_request_id_fail_closed(tmp_path: Path) -> None: def test_rollback_uses_only_explicit_receipt_bound_interface(tmp_path: Path) -> None: - observed: dict[str, Any] = {} + observed: list[dict[str, Any]] = [] def rollback(release_sha256: str, **kwargs: Any) -> dict[str, Any]: - observed.update({"release_sha256": release_sha256, **kwargs}) + 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) @@ -1046,9 +1139,9 @@ def test_rollback_uses_only_explicit_receipt_bound_interface(tmp_path: Path) -> ) assert result["status"] == "pass" - assert observed["release_sha256"] == "a" * 64 - assert observed["execute_rollback"] is True - assert "install_release" not in observed + 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: @@ -1062,6 +1155,26 @@ def test_rollback_fails_when_explicit_interface_is_absent(tmp_path: Path) -> Non ) +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, @@ -1097,10 +1210,14 @@ def test_install_verifier_failure_returns_bounded_automatic_rollback_result( return object() def fail_install(*_args: Any, **_kwargs: Any) -> dict[str, Any]: - raise local.installer.VerificationError( + 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, @@ -1131,7 +1248,9 @@ def test_install_verifier_failure_returns_bounded_automatic_rollback_result( assert result["rollback"] == { "status": "pass", "mode": "installer_automatic", - "restored_posture": "prior", + "executed": True, + "restored_posture": "absent", + "prior_posture_preserved": True, } assert result["authority"]["sending_enabled"] is False @@ -1189,26 +1308,47 @@ def test_post_install_proof_failure_executes_explicit_rollback( 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: {"status": "pass"}, - verify_running_release=lambda *_args, **_kwargs: next(verifications), - rollback_release=lambda release_sha256, **_kwargs: ( - rollbacks.append(release_sha256) - or { - "schema": "livingip.leocleanNoSendRollbackResult.v1", - "status": "pass", - "mode": "executed", - "rolled_back_release_sha256": release_sha256, - "restored_release_sha256": None, - "restored_service_posture": "absent", - "verification": 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", @@ -1256,7 +1396,9 @@ def test_post_install_proof_failure_executes_explicit_rollback( assert result["rollback"] == { "status": "pass", "mode": "explicit", + "executed": True, "restored_posture": "absent", + "prior_posture_preserved": True, } assert result["fail_closed"] == { "status": "not_required", @@ -1266,9 +1408,413 @@ def test_post_install_proof_failure_executes_explicit_rollback( "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, @@ -1329,19 +1875,43 @@ def test_incomplete_rollback_stops_service_and_reports_inert_posture( ), ] ) + + 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: {"status": "pass"}, + 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=lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("rollback failed")), + 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), @@ -1377,7 +1947,9 @@ def test_incomplete_rollback_stops_service_and_reports_inert_posture( assert result["rollback"] == { "status": "fail", "mode": "explicit", + "executed": True, "restored_posture": "unknown", + "prior_posture_preserved": False, } assert result["fail_closed"] == { "status": "pass", @@ -1387,6 +1959,7 @@ def test_incomplete_rollback_stops_service_and_reports_inert_posture( "main_pid": 0, "n_restarts": 0, "candidate_container": "absent", + "service_action_executed": True, } assert result["authority"]["sending_enabled"] is False @@ -1418,6 +1991,7 @@ def test_fail_closed_stop_refuses_untrusted_unit_or_dropin() -> None: "main_pid": None, "n_restarts": None, "candidate_container": "unknown", + "service_action_executed": False, } diff --git a/tests/test_gcp_leoclean_nosend_installer.py b/tests/test_gcp_leoclean_nosend_installer.py index 893b45f..1e83e58 100644 --- a/tests/test_gcp_leoclean_nosend_installer.py +++ b/tests/test_gcp_leoclean_nosend_installer.py @@ -110,6 +110,7 @@ class FakeHost: self.current_release: dict[str, object] | None = None self.loaded_release: dict[str, object] | None = None self.generation = 0 + self.start_calls = 0 self.restart_calls = 0 self.stop_calls = 0 self.fail_dropin_release: str | None = None @@ -157,7 +158,7 @@ class FakeHost: (directory / "cmdline").write_bytes(b"\0".join(item.encode() for item in command) + b"\0") (directory / "cgroup").write_text(f"0::/system.slice/{package.SERVICE}\n", encoding="utf-8") - def _start(self) -> None: + def _start(self, action: str = "restart") -> None: release = self.loaded_release assert release is not None image = release["image"] @@ -166,7 +167,10 @@ class FakeHost: runtime = inspection["runtime_config"] assert isinstance(runtime, dict) self.generation += 1 - self.restart_calls += 1 + if action == "start": + self.start_calls += 1 + else: + self.restart_calls += 1 self.active = True self.main_pid = 1000 + self.generation self.invocation = f"{self.generation:032x}" @@ -274,6 +278,9 @@ class FakeHost: if action == "restart": self._start() return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"") + if action == "start": + self._start("start") + return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"") if action == "stop": self._stop() return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"") @@ -397,7 +404,7 @@ def test_disposable_root_install_is_digest_bound_idempotent_and_exact(tmp_path: host = FakeHost(paths) host.add_release(release, image) - receipt = installer.install_release( + result = installer.install_release( bundle, paths=paths, source_binding=source_binding(), @@ -405,10 +412,26 @@ def test_disposable_root_install_is_digest_bound_idempotent_and_exact(tmp_path: sleeper=lambda _seconds: None, execute_restart=True, ) + receipt = json.loads(paths.receipt.read_text(encoding="utf-8")) - assert receipt["status"] == "pass" - assert receipt["schema"] == "livingip.leocleanNoSendInstallReceipt.v2" - assert receipt["release_sha256"] == release["release_sha256"] + assert result["status"] == "pass" + assert result["schema"] == installer.INSTALL_EXECUTION_RESULT_SCHEMA + assert result["outcome"] == "installed" + assert result["transition"] == { + "outcome": "installed", + "prior_posture": "absent", + "prior_n_restarts": 0, + "final_posture": "active", + "service_action": "restart", + "managed_files_replaced": True, + } + assert result["release_sha256"] == release["release_sha256"] + assert result["persisted_receipt"] == { + "schema": installer.INSTALL_RECEIPT_SCHEMA, + "sha256": package.sha256_file(paths.receipt), + } + assert receipt["schema"] == installer.INSTALL_RECEIPT_SCHEMA + assert receipt["outcome"] == "installed" assert host.restart_calls == 1 assert paths.unit.read_text(encoding="utf-8") == package.render_systemd_unit(release) assert paths.unit.stat().st_mode & 0o777 == 0o644 @@ -423,6 +446,12 @@ def test_disposable_root_install_is_digest_bound_idempotent_and_exact(tmp_path: assert not list(paths.docker_config.iterdir()) assert not list(paths.unit.parent.glob(f".{paths.unit.name}.*.tmp")) + managed_before = { + role: path.read_bytes() + for role, (path, _mode) in installer._managed_files(paths).items() + } + identity_before = (host.main_pid, host.n_restarts, host.invocation) + calls_before = len(host.calls) repeated = installer.install_release( bundle, paths=paths, @@ -432,8 +461,31 @@ def test_disposable_root_install_is_digest_bound_idempotent_and_exact(tmp_path: execute_restart=True, ) - assert repeated["result"] == "already_installed" + assert repeated["outcome"] == "verified_noop" + assert repeated["schema"] == installer.INSTALL_EXECUTION_RESULT_SCHEMA + assert repeated["persisted_receipt"] == { + "schema": installer.INSTALL_RECEIPT_SCHEMA, + "sha256": package.sha256_file(paths.receipt), + } + assert repeated["transition"] == { + "outcome": "verified_noop", + "prior_posture": "active", + "prior_n_restarts": 0, + "final_posture": "active", + "service_action": "none", + "managed_files_replaced": False, + } assert host.restart_calls == 1 + assert host.start_calls == 0 + assert (host.main_pid, host.n_restarts, host.invocation) == identity_before + assert { + role: path.read_bytes() + for role, (path, _mode) in installer._managed_files(paths).items() + } == managed_before + assert not any( + call[0] == installer.SYSTEMCTL and call[1] in {"start", "restart", "stop", "daemon-reload"} + for call in host.calls[calls_before:] + ) def test_first_install_accepts_production_shaped_systemd_not_found_defaults(tmp_path: Path) -> None: @@ -465,10 +517,391 @@ def test_first_install_accepts_production_shaped_systemd_not_found_defaults(tmp_ execute_restart=True, ) - assert repeated["result"] == "already_installed" + assert repeated["outcome"] == "verified_noop" assert host.restart_calls == 1 +def test_exact_inactive_release_converges_with_one_controlled_start_and_preserves_history( + tmp_path: Path, +) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + host._stop() + managed_before = { + role: path.read_bytes() + for role, (path, _mode) in installer._managed_files(paths).items() + } + calls_before = len(host.calls) + restart_calls = host.restart_calls + stop_calls = host.stop_calls + + result = installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert result["outcome"] == "controlled_start" + assert result["schema"] == installer.INSTALL_EXECUTION_RESULT_SCHEMA + assert result["transition"] == { + "outcome": "controlled_start", + "prior_posture": "inactive", + "prior_n_restarts": 0, + "final_posture": "active", + "service_action": "start", + "managed_files_replaced": False, + } + assert host.active + assert host.start_calls == 1 + assert host.restart_calls == restart_calls + assert host.stop_calls == stop_calls + assert { + role: path.read_bytes() + for role, (path, _mode) in installer._managed_files(paths).items() + } == managed_before + assert [ + call[1] + for call in host.calls[calls_before:] + if call[0] == installer.SYSTEMCTL and call[1] in {"start", "restart", "stop", "daemon-reload"} + ] == ["start"] + + +@pytest.mark.parametrize( + "role", + ["unit", "helper", "release", "receipt", "rollback_state"], +) +def test_each_missing_managed_file_is_rejected_as_partial_before_mutation( + tmp_path: Path, + role: str, +) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + managed_path = installer._managed_files(paths)[role][0] + managed_path.unlink() + calls_before = len(host.calls) + + with pytest.raises(installer.InstallError, match="managed install state was partial"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert not any( + call[0] == installer.SYSTEMCTL and call[1] in {"start", "restart", "stop", "daemon-reload"} + for call in host.calls[calls_before:] + ) + + +@pytest.mark.parametrize( + "mutation", + [ + "outer_extra", + "rollback_extra", + "previous_release_integer", + "previous_active_integer", + "automatic_integer", + "service_restart_drift", + "rollback_posture_drift", + "load_state_drift", + "main_pid_drift", + "verification_forged", + "authority_forged", + "invocation_integer", + "container_integer", + "cmdline_integer", + "environment_nonstring", + ], +) +def test_current_install_receipt_is_exact_and_bound_to_rollback_state( + tmp_path: Path, + mutation: str, +) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + receipt = json.loads(paths.receipt.read_text(encoding="utf-8")) + if mutation == "outer_extra": + receipt["unexpected"] = "field" + elif mutation == "rollback_extra": + receipt["rollback"]["unexpected"] = "field" + elif mutation == "previous_release_integer": + receipt["rollback"]["previous_release_present"] = 0 + elif mutation == "previous_active_integer": + receipt["rollback"]["previous_service_active"] = 0 + elif mutation == "automatic_integer": + receipt["rollback"]["automatic_on_failure"] = 1 + elif mutation == "service_restart_drift": + receipt["service_before"]["n_restarts"] += 1 + elif mutation == "rollback_posture_drift": + receipt["rollback"]["previous_service_posture"] = "active" + elif mutation == "load_state_drift": + receipt["service_before"]["load_state"] = "loaded" + elif mutation == "main_pid_drift": + receipt["service_before"]["main_pid"] = 12345 + elif mutation == "verification_forged": + receipt["verification"] = {"status": "forged"} + elif mutation == "authority_forged": + receipt["authority"] = {"cloud_or_database_authority_changed": True} + elif mutation == "invocation_integer": + receipt["verification"]["invocation_id"] = int("1" * 32) + elif mutation == "container_integer": + receipt["verification"]["container_id"] = int("2" * 64) + elif mutation == "cmdline_integer": + receipt["verification"]["supervising_process"]["cmdline_sha256"] = int("3" * 64) + else: + receipt["verification"]["validated_process_environment_fields"] = [{}] + paths.receipt.write_text(json.dumps(receipt, sort_keys=True) + "\n", encoding="utf-8") + paths.receipt.chmod(0o600) + stop_calls = host.stop_calls + + with pytest.raises(installer.InstallError, match="installed receipt"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + with pytest.raises(installer.InstallError, match="installed receipt"): + installer._validate_managed_snapshots( + installer._current_managed_snapshots(paths), + paths=paths, + ) + + assert host.stop_calls == stop_calls + + +def test_legacy_v2_install_receipt_remains_readable_without_rewrite(tmp_path: Path) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + legacy = json.loads(paths.receipt.read_text(encoding="utf-8")) + legacy["schema"] = installer.LEGACY_INSTALL_RECEIPT_SCHEMA + legacy["result"] = "installed" + legacy.pop("outcome") + legacy.pop("transition") + paths.receipt.write_text(json.dumps(legacy, indent=2, sort_keys=True) + "\n", encoding="utf-8") + paths.receipt.chmod(0o600) + receipt_before = paths.receipt.read_bytes() + + result = installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert result["outcome"] == "verified_noop" + assert paths.receipt.read_bytes() == receipt_before + + +@pytest.mark.parametrize("failure_mode", ["start_command", "post_start_verification"]) +def test_exact_inactive_start_failure_restores_inactive_without_rewriting_history( + tmp_path: Path, + failure_mode: str, +) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + host._stop() + managed_before = { + role: path.read_bytes() + for role, (path, _mode) in installer._managed_files(paths).items() + } + if failure_mode == "post_start_verification": + host.fail_dropin_release = str(release["release_sha256"]) + + def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + completed = host(arguments) + if failure_mode == "start_command" and arguments[:2] == [installer.SYSTEMCTL, "start"]: + return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"injected") + return completed + + with pytest.raises(installer.VerificationError) as caught: + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=runner, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert caught.value.rollback_executed is True + assert caught.value.restored_posture == "inactive" + assert caught.value.prior_posture_preserved is True + assert not host.active + assert host.container is None + assert host.start_calls == 1 + assert { + role: path.read_bytes() + for role, (path, _mode) in installer._managed_files(paths).items() + } == managed_before + + +@pytest.mark.parametrize("cleanup_failure", ["stop_command", "restart_counter_drift"]) +def test_exact_inactive_cleanup_failure_never_claims_prior_posture_preserved( + tmp_path: Path, + cleanup_failure: str, +) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + host._stop() + host.fail_dropin_release = str(release["release_sha256"]) + managed_before = { + role: path.read_bytes() + for role, (path, _mode) in installer._managed_files(paths).items() + } + + def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + completed = host(arguments) + if arguments[:2] == [installer.SYSTEMCTL, "stop"]: + if cleanup_failure == "stop_command": + return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"injected") + host.n_restarts += 1 + return completed + + with pytest.raises(installer.InstallError, match="cleanup was incomplete") as caught: + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=runner, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert getattr(caught.value, "prior_posture_preserved", None) is not True + assert not host.active + assert { + role: path.read_bytes() + for role, (path, _mode) in installer._managed_files(paths).items() + } == managed_before + + +@pytest.mark.parametrize("prior_posture", ["active", "inactive"]) +def test_exact_release_rejects_receipt_replacement_race( + tmp_path: Path, + prior_posture: str, +) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + if prior_posture == "inactive": + host._stop() + replaced = False + service_show_calls = 0 + + def replace_receipt() -> None: + nonlocal replaced + if replaced: + return + replaced = True + paths.receipt.unlink() + paths.receipt.write_text('{"schema":"forged"}\n', encoding="utf-8") + paths.receipt.chmod(0o600) + + def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + nonlocal service_show_calls + completed = host(arguments) + if prior_posture == "active" and arguments[:2] == [installer.SYSTEMCTL, "show"]: + service_show_calls += 1 + if service_show_calls == 3: + replace_receipt() + if prior_posture == "inactive" and arguments[:2] == [installer.SYSTEMCTL, "start"]: + replace_receipt() + return completed + + with pytest.raises(installer.InstallError, match="file identity drifted"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=runner, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert replaced + assert host.active is (prior_posture == "active") + + def test_first_install_waits_for_complete_cid_and_healthy_container(tmp_path: Path) -> None: bundle, release, image = release_bundle(tmp_path, "2") paths = install_paths(tmp_path) @@ -1135,7 +1568,7 @@ def test_upgrade_stops_verified_prior_release_before_unit_replacement(tmp_path: execute_restart=True, ) start = len(host.calls) - receipt = installer.install_release( + result = installer.install_release( second_bundle, paths=paths, source_binding=source_binding(), @@ -1143,14 +1576,101 @@ def test_upgrade_stops_verified_prior_release_before_unit_replacement(tmp_path: sleeper=lambda _seconds: None, execute_restart=True, ) + receipt = json.loads(paths.receipt.read_text(encoding="utf-8")) actions = [call[1] for call in host.calls[start:] if call[0] == installer.SYSTEMCTL] assert actions.index("stop") < actions.index("daemon-reload") < actions.index("restart") - assert receipt["release_sha256"] == second_release["release_sha256"] + assert result["release_sha256"] == second_release["release_sha256"] + assert result["outcome"] == "replaced" + assert result["transition"] == { + "outcome": "replaced", + "prior_posture": "active", + "prior_n_restarts": 0, + "final_posture": "active", + "service_action": "restart", + "managed_files_replaced": True, + } + assert receipt["rollback"]["previous_service_posture"] == "active" assert host.current_release == second_release assert host.stop_calls == 1 +def test_inactive_prior_release_is_replaced_without_claiming_active_posture(tmp_path: Path) -> None: + first_bundle, first_release, first_image = release_bundle(tmp_path, "2") + second_bundle, second_release, second_image = release_bundle(tmp_path, "4") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(first_release, first_image) + host.add_release(second_release, second_image) + installer.install_release( + first_bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + host._stop() + stop_calls = host.stop_calls + + result = installer.install_release( + second_bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + receipt = json.loads(paths.receipt.read_text(encoding="utf-8")) + + assert result["outcome"] == "replaced" + assert result["transition"]["prior_posture"] == "inactive" + assert result["transition"]["prior_n_restarts"] == 0 + assert receipt["rollback"]["previous_release_present"] is True + assert receipt["rollback"]["previous_service_active"] is False + assert receipt["rollback"]["previous_service_posture"] == "inactive" + assert host.stop_calls == stop_calls + assert host.active + + +def test_active_prior_verification_failure_reports_unchanged_recovery_posture( + tmp_path: Path, +) -> None: + first_bundle, first_release, first_image = release_bundle(tmp_path, "2") + second_bundle, second_release, second_image = release_bundle(tmp_path, "4") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(first_release, first_image) + host.add_release(second_release, second_image) + installer.install_release( + first_bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + process_root = paths.proc_root / str(host.main_pid) + (process_root / "environ").write_bytes(b"PATH=/unexpected\0") + stop_calls = host.stop_calls + + with pytest.raises(installer.VerificationError) as caught: + installer.install_release( + second_bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert caught.value.rollback_executed is False + assert caught.value.restored_posture == "active" + assert caught.value.prior_posture_preserved is True + assert host.stop_calls == stop_calls + assert host.active + + def test_stale_loaded_unit_is_rejected_before_trusting_old_exec_stop(tmp_path: Path) -> None: first_bundle, first_release, first_image = release_bundle(tmp_path, "2") second_bundle, second_release, second_image = release_bundle(tmp_path, "4") @@ -1327,6 +1847,45 @@ def test_transitional_prior_service_posture_is_rejected_before_mutation(tmp_path assert host.restart_calls == 1 +def test_failed_prior_service_posture_is_rejected_before_mutation(tmp_path: Path) -> None: + first_bundle, first_release, first_image = release_bundle(tmp_path, "2") + second_bundle, second_release, second_image = release_bundle(tmp_path, "4") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(first_release, first_image) + host.add_release(second_release, second_image) + installer.install_release( + first_bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + host.property_overrides.update( + { + "ActiveState": "failed", + "SubState": "failed", + "MainPID": "0", + "InvocationID": "", + "ControlGroup": "", + } + ) + + with pytest.raises(installer.InstallError, match="transitional or failed"): + installer.install_release( + second_bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert host.stop_calls == 0 + assert host.restart_calls == 1 + + def test_changing_prior_service_posture_fails_two_readback_gate(tmp_path: Path) -> None: first_bundle, first_release, first_image = release_bundle(tmp_path, "2") second_bundle, second_release, second_image = release_bundle(tmp_path, "4") @@ -1366,7 +1925,7 @@ def test_wrong_local_image_is_rejected_without_service_restart(tmp_path: Path) - host = FakeHost(paths) host.add_release(release, image) - with pytest.raises(installer.InstallError, match=r"finalized (release|inspection)"): + with pytest.raises(installer.VerificationError, match=r"finalized (release|inspection)") as caught: installer.install_release( bundle, paths=paths, @@ -1376,6 +1935,9 @@ def test_wrong_local_image_is_rejected_without_service_restart(tmp_path: Path) - execute_restart=True, ) + assert caught.value.rollback_executed is False + assert caught.value.restored_posture == "absent" + assert caught.value.prior_posture_preserved is True assert host.restart_calls == 0 assert not paths.unit.exists() assert not paths.state_dir.exists()