teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Converge leoclean no-send installation state

Browse source
This commit is contained in:
fwazb 2026-07-23 16:25:17 -07:00
parent 6696f3adc0
commit e5a5cc0ca2
6 changed files with 2268 additions and 139 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
docs/gcp-leoclean-nosend-service-package.md
View file

@ -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

480
ops/gcp_leoclean_nosend_iap_remote.py
View file

@ -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,
),
)
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,6 +1582,37 @@ def execute_request(
fallback_stage=failure_stage,
fallback_reason=failure_reason,
)
if installer_transition is None:
fail_closed = _fail_closed_inert_service(
installer_module=installer_module,
release=release,
paths=paths,
runner=runner,
)
raise RemoteOperationFailure(
_operational_failure_result(
manifest=manifest,
release=release,
failure_stage="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,
@ -1256,6 +1621,10 @@ def execute_request(
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,
@ -1263,6 +1632,26 @@ def execute_request(
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,

487
ops/install_gcp_leoclean_nosend_release.py
View file

@ -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")
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:
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)
existing_receipt = package.load_json(paths.receipt, "install receipt")
return {**existing_receipt, "result": "already_installed", "verification": verified}
_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")
try:
prior_verification = verify_running_release(prior_release, runner=runner, paths=paths, sleeper=sleeper)
_require(
_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)

162
ops/run_gcp_leoclean_nosend_iap.py
View file

@ -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 (

616
tests/test_gcp_leoclean_nosend_iap_runner.py
View file

@ -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,7 +888,62 @@ 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)
@ -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,17 +1308,16 @@ def test_post_install_proof_failure_executes_explicit_rollback(
def under(_root: Path) -> object:
return object()
installer_module = SimpleNamespace(
SYSTEMCTL="/usr/bin/systemctl",
VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES,
InstallPaths=Paths,
validate_release_bundle=lambda _bundle: (release, b"unit"),
_validate_source_binding=lambda _binding: None,
install_release=lambda *_args, **_kwargs: {"status": "pass"},
verify_running_release=lambda *_args, **_kwargs: next(verifications),
rollback_release=lambda release_sha256, **_kwargs: (
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)
or {
return {
"schema": "livingip.leocleanNoSendRollbackResult.v1",
"status": "pass",
"mode": "executed",
@ -1208,7 +1326,29 @@ def test_post_install_proof_failure_executes_explicit_rollback(
"restored_service_posture": "absent",
"verification": None,
}
installer_module = SimpleNamespace(
SYSTEMCTL="/usr/bin/systemctl",
VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES,
INSTALL_EXECUTION_RESULT_SCHEMA=local.installer.INSTALL_EXECUTION_RESULT_SCHEMA,
LEGACY_INSTALL_RECEIPT_SCHEMA=local.installer.LEGACY_INSTALL_RECEIPT_SCHEMA,
INSTALL_RECEIPT_SCHEMA=local.installer.INSTALL_RECEIPT_SCHEMA,
InstallPaths=Paths,
validate_release_bundle=lambda _bundle: (release, b"unit"),
_validate_source_binding=lambda _binding: None,
install_release=lambda *_args, **_kwargs: install_execution_result(
release,
{
"outcome": "installed",
"prior_posture": "absent",
"prior_n_restarts": 0,
"final_posture": "active",
"service_action": "restart",
"managed_files_replaced": True,
},
),
verify_running_release=lambda *_args, **_kwargs: next(verifications),
rollback_release=rollback_release,
)
package_module = SimpleNamespace(
render_systemd_unit=lambda _release: "unit",
@ -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,
}

590
tests/test_gcp_leoclean_nosend_installer.py
View file

@ -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,6 +167,9 @@ class FakeHost:
runtime = inspection["runtime_config"]
assert isinstance(runtime, dict)
self.generation += 1
if action == "start":
self.start_calls += 1
else:
self.restart_calls += 1
self.active = True
self.main_pid = 1000 + self.generation
@ -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,94 @@ 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(),
runner=host,
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 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(),
@ -1144,11 +1664,11 @@ def test_upgrade_stops_verified_prior_release_before_unit_replacement(tmp_path:
execute_restart=True,
)
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 host.current_release == second_release
assert host.stop_calls == 1
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:
@ -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()