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

Preserve no-send IAP failure receipts

Browse source
This commit is contained in:
fwazb 2026-07-23 14:29:47 -07:00
parent 0312209c2d
commit 445ffdc9fc
4 changed files with 1136 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -344,6 +344,33 @@ contact GCP or Cloud SQL, change Telegram traffic, touch the VPS, approve a
proposal, or promote production. Those remain later, separately authorized
boundaries.
## IAP result and failure receipts
The fixed-target IAP runner treats the remote `result.json`, the remote process
exit status/stdout, the IAP transport, and the private local result receipt as
one bounded result path. A successful operation is accepted only when the
structured result matches the exact request, revision, artifact, image, config
digest, service, database proof, and no-send authority contract.
Post-install failures are returned with allowlisted `failure_stage` and
`failure_reason` values. If explicit rollback passes, its exact restored
posture is recorded. If rollback cannot be proven, the remote helper stops the
staging service only after exact-binding the currently loaded unit, `ExecStop`,
supervising process, and container to the reviewed release. It then proves
`inactive/dead`, `MainPID=0`, and absence of the named candidate container
before reporting a fail-closed posture. Drift blocks the stop; a failure to
prove that posture is reported as unknown and is never converted into success.
If IAP returns no output, malformed output, a contract mismatch, or a transport
interruption, the local runner writes a private self-hashed failure receipt
with the bounded stage/reason, transfer-cleanup result, and `remote_state:
unknown`. Command stdout, stderr, tokens, passwords, and secret values are not
copied into a receipt or exception. That local-only receipt records requested
authority, not unproven remote outcomes. Operators must not infer a successful
or inert remote service from a local transport-failure receipt. Remote
operation failures and local transport failures use distinct outer receipt
schemas; the successful result-receipt v1 contract is unchanged.
## Offline evidence and live gate
Offline tests prove strict receipt/descriptor/inspection schemas, exact

363
ops/gcp_leoclean_nosend_iap_remote.py
View file

@ -16,6 +16,7 @@ import subprocess
import sys
import tarfile
import tempfile
import time
import zipfile
from collections.abc import Callable, Mapping
from pathlib import Path
@ -24,6 +25,7 @@ from typing import Any
SCHEMA = "livingip.leocleanNoSendIapRequest.v1"
RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v1"
FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapFailure.v1"
OPERATION_FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v1"
PROJECT = "teleo-501523"
ZONE = "europe-west6-a"
INSTANCE = "teleo-staging-1"
@ -48,6 +50,17 @@ MAX_MEMBER_BYTES = 2 * 1024 * 1024
MAX_MANIFEST_BYTES = 128 * 1024
MAX_COMMAND_OUTPUT_BYTES = 1024 * 1024
MAX_TOKEN_BYTES = 64 * 1024
POST_INSTALL_FAILURES = frozenset(
{
("install_receipt", "install_receipt_invalid"),
("service_before_restart", "service_verification_failed"),
("database_before_restart", "database_verification_failed"),
("service_restart", "service_restart_failed"),
("service_after_restart", "service_verification_failed"),
("restart_identity", "restart_identity_mismatch"),
("database_after_restart", "database_verification_failed"),
}
)
EXPECTED_MEMBERS = {
"request-manifest.json",
"release-artifact.zip",
@ -852,6 +865,236 @@ def _is_verification_failure(installer_module: Any, failure: Exception) -> bool:
return isinstance(failure_type, type) and isinstance(failure, failure_type)
def _post_install_failure_coordinates(
installer_module: Any,
failure: Exception,
*,
fallback_stage: str,
fallback_reason: str,
) -> tuple[str, str]:
if _is_verification_failure(installer_module, failure):
failure_stage = getattr(failure, "failure_stage", None)
failure_reason = getattr(failure, "failure_reason", None)
_require(
isinstance(failure_stage, str)
and isinstance(failure_reason, str)
and (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES,
"verification failure was not allowlisted",
)
return failure_stage, failure_reason
_require(
(fallback_stage, fallback_reason) in POST_INSTALL_FAILURES,
"post-install failure was not allowlisted",
)
return fallback_stage, fallback_reason
def _rollback_proven_service(posture: str) -> dict[str, Any]:
_require(posture in {"active", "inactive", "absent"}, "rollback posture was invalid")
return {
"status": "not_required",
"service_posture": posture,
"active_state": "not_observed",
"sub_state": "not_observed",
"main_pid": None,
"n_restarts": None,
"candidate_container": "not_observed",
}
def _fail_closed_inert_service(
*,
installer_module: Any,
release: Mapping[str, Any],
paths: Any,
runner: Runner,
) -> dict[str, Any]:
def contract_runner(argv: list[str]) -> subprocess.CompletedProcess[bytes]:
return _run(
argv,
runner=runner,
label="fail-closed service command",
require_success=False,
)
try:
systemctl = getattr(installer_module, "_systemctl", None)
stable_service_posture = getattr(installer_module, "_stable_service_posture", None)
verify_running_release = getattr(installer_module, "verify_running_release", None)
require_candidate_absent = getattr(installer_module, "_require_candidate_absent", None)
_require(
callable(systemctl)
and callable(stable_service_posture)
and callable(verify_running_release)
and callable(require_candidate_absent),
"reviewed installer lacked fail-closed service controls",
)
before = stable_service_posture(
release,
runner=contract_runner,
paths=paths,
sleeper=time.sleep,
)
_require(before.kind in {"active", "inactive"}, "managed service posture was not trusted")
if before.kind == "active":
verification = verify_running_release(
release,
runner=contract_runner,
paths=paths,
sleeper=time.sleep,
)
_require(
verification.get("main_pid") == before.state.get("MainPID")
and verification.get("invocation_id") == before.state.get("InvocationID")
and verification.get("n_restarts") == before.state.get("NRestarts"),
"running service identity changed before fail-closed stop",
)
systemctl("stop", contract_runner)
after = stable_service_posture(
release,
runner=contract_runner,
paths=paths,
sleeper=time.sleep,
)
_require(
after.kind == "inactive"
and after.state.get("ActiveState") == "inactive"
and after.state.get("SubState") == "dead"
and after.state.get("MainPID") == 0
and isinstance(after.state.get("NRestarts"), int)
and not isinstance(after.state.get("NRestarts"), bool)
and after.state["NRestarts"] >= 0,
"service did not stop into a stable inert posture",
)
require_candidate_absent(
runner=contract_runner,
paths=paths,
sleeper=time.sleep,
)
return {
"status": "pass",
"service_posture": "inactive",
"active_state": "inactive",
"sub_state": "dead",
"main_pid": 0,
"n_restarts": after.state["NRestarts"],
"candidate_container": "absent",
}
except Exception:
return {
"status": "fail",
"service_posture": "unknown",
"active_state": "unknown",
"sub_state": "unknown",
"main_pid": None,
"n_restarts": None,
"candidate_container": "unknown",
}
def _operational_failure_result(
*,
manifest: Mapping[str, Any],
release: Mapping[str, Any],
failure_stage: str,
failure_reason: str,
rollback: Mapping[str, str],
fail_closed: Mapping[str, Any],
archive_sha256: str,
helper_sha256: str,
installer_module: Any,
) -> dict[str, Any]:
_require(
manifest.get("operation") == "install"
and set(rollback) == {"status", "mode", "restored_posture"}
and (
(failure_stage, failure_reason) in POST_INSTALL_FAILURES
or (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES
),
"post-install failure was not allowlisted",
)
rollback_passed = (
rollback.get("status") == "pass"
and rollback.get("mode") == "explicit"
and rollback.get("restored_posture") in {"active", "inactive", "absent"}
)
rollback_failed = rollback == {
"status": "fail",
"mode": "explicit",
"restored_posture": "unknown",
}
verification_pair = (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES
_require(
(rollback_passed and not verification_pair) or rollback_failed,
"post-install rollback summary was invalid",
)
if rollback_passed:
_require(
fail_closed == _rollback_proven_service(str(rollback["restored_posture"])),
"post-install restored posture summary was invalid",
)
else:
_require(
fail_closed
in (
{
"status": "fail",
"service_posture": "unknown",
"active_state": "unknown",
"sub_state": "unknown",
"main_pid": None,
"n_restarts": None,
"candidate_container": "unknown",
},
{
"status": "pass",
"service_posture": "inactive",
"active_state": "inactive",
"sub_state": "dead",
"main_pid": 0,
"n_restarts": fail_closed.get("n_restarts"),
"candidate_container": "absent",
},
)
and (
fail_closed.get("status") != "pass"
or (
isinstance(fail_closed.get("n_restarts"), int)
and not isinstance(fail_closed.get("n_restarts"), bool)
and fail_closed["n_restarts"] >= 0
)
),
"post-install fail-closed summary was invalid",
)
return {
"schema": OPERATION_FAILURE_RESULT_SCHEMA,
"status": "fail",
"operation": manifest["operation"],
"request_id": manifest["request_id"],
"runner_revision": manifest["runner_revision"],
"release_revision": manifest["release_revision"],
"release_sha256": release["release_sha256"],
"image_digest": release["image"]["digest"],
"image_config_digest": release["image"]["config_digest"],
"release_artifact_sha256": manifest["release_artifact"]["archive_sha256"],
"request_binding": {
"archive_sha256": archive_sha256,
"helper_sha256": helper_sha256,
"manifest_sha256": manifest["manifest_sha256"],
},
"failure_stage": failure_stage,
"failure_reason": failure_reason,
"rollback": dict(rollback),
"fail_closed": dict(fail_closed),
"authority": {
"sending_enabled": False,
"production_traffic_changed": False,
"vps_touched": False,
"canonical_proposal_approved": False,
},
}
def execute_request(
manifest: Mapping[str, Any],
payload_root: Path,
@ -919,7 +1162,6 @@ def execute_request(
installer_module=installer_module,
)
) from exc
_require(install_receipt.get("status") == "pass", "installation did not pass")
elif operation == "rollback":
rollback = _execute_explicit_rollback(
installer_module=installer_module,
@ -945,12 +1187,19 @@ def execute_request(
)
if operation == "install":
failure_stage = "install_receipt"
failure_reason = "install_receipt_invalid"
try:
_require(install_receipt.get("status") == "pass", "installation did not pass")
failure_stage = "service_before_restart"
failure_reason = "service_verification_failed"
before = installer_module.verify_running_release(
release,
runner=contract_runner,
paths=paths,
)
failure_stage = "database_before_restart"
failure_reason = "database_verification_failed"
before_database = _verify_database(
container_id=before["container_id"],
run_id=f"{manifest['request_id']}-pre",
@ -959,16 +1208,22 @@ def execute_request(
package_module=package_module,
runner=runner,
)
failure_stage = "service_restart"
failure_reason = "service_restart_failed"
_run(
[installer_module.SYSTEMCTL, "restart", SERVICE],
runner=runner,
label="fixed service restart",
)
failure_stage = "service_after_restart"
failure_reason = "service_verification_failed"
after = installer_module.verify_running_release(
release,
runner=contract_runner,
paths=paths,
)
failure_stage = "restart_identity"
failure_reason = "restart_identity_mismatch"
_require(
before["main_pid"] != after["main_pid"]
and before["invocation_id"] != after["invocation_id"]
@ -976,6 +1231,8 @@ def execute_request(
and after["n_restarts"] >= before["n_restarts"],
"post-install restart did not create a new exact service/container identity",
)
failure_stage = "database_after_restart"
failure_reason = "database_verification_failed"
after_database = _verify_database(
container_id=after["container_id"],
run_id=f"{manifest['request_id']}-post",
@ -985,6 +1242,12 @@ def execute_request(
runner=runner,
)
except Exception as exc:
bounded_stage, bounded_reason = _post_install_failure_coordinates(
installer_module,
exc,
fallback_stage=failure_stage,
fallback_reason=failure_reason,
)
try:
recovery = _execute_explicit_rollback(
installer_module=installer_module,
@ -1000,8 +1263,30 @@ def execute_request(
archive_sha256=archive_sha256,
helper_sha256=helper_sha256,
)
except Exception as rollback_exc:
raise RemoteError("install_failed_rollback_incomplete") from rollback_exc
except Exception:
fail_closed = _fail_closed_inert_service(
installer_module=installer_module,
release=release,
paths=paths,
runner=runner,
)
raise RemoteOperationFailure(
_operational_failure_result(
manifest=manifest,
release=release,
failure_stage=bounded_stage,
failure_reason=bounded_reason,
rollback={
"status": "fail",
"mode": "explicit",
"restored_posture": "unknown",
},
fail_closed=fail_closed,
archive_sha256=archive_sha256,
helper_sha256=helper_sha256,
installer_module=installer_module,
)
) from exc
if _is_verification_failure(installer_module, exc):
raise RemoteOperationFailure(
_verification_failure_result(
@ -1018,7 +1303,24 @@ def execute_request(
installer_module=installer_module,
)
) from exc
raise RemoteError("install_failed_rolled_back") from exc
restored_posture = str(rollback_result["service"]["restored_posture"])
raise RemoteOperationFailure(
_operational_failure_result(
manifest=manifest,
release=release,
failure_stage=bounded_stage,
failure_reason=bounded_reason,
rollback={
"status": "pass",
"mode": "explicit",
"restored_posture": restored_posture,
},
fail_closed=_rollback_proven_service(restored_posture),
archive_sha256=archive_sha256,
helper_sha256=helper_sha256,
installer_module=installer_module,
)
) from exc
service = {
"name": SERVICE,
"before_restart": before,
@ -1089,6 +1391,35 @@ def execute_request(
}
def _execute_and_persist_result(
manifest: Mapping[str, Any],
payload_root: Path,
request_dir: Path,
archive_sha256: str,
helper_sha256: str,
*,
runner: Runner,
) -> dict[str, Any]:
try:
result = execute_request(
manifest,
payload_root,
request_dir,
archive_sha256,
helper_sha256,
runner=runner,
)
except RemoteOperationFailure as exc:
result = exc.result
result_path = request_dir / "result.json"
_copy_into_request(
result_path,
(json.dumps(result, allow_nan=False, indent=2, sort_keys=True) + "\n").encode("utf-8"),
0o600,
)
return result
def receive_request(
operation: str,
request_id: str,
@ -1154,24 +1485,14 @@ def receive_request(
release_revision=release_revision,
)
verify_service_account(runner=runner)
try:
result = execute_request(
manifest,
payload_root,
request_dir,
archive_sha256,
helper_sha256,
runner=runner,
)
except RemoteOperationFailure as exc:
result = exc.result
result_path = request_dir / "result.json"
_copy_into_request(
result_path,
(json.dumps(result, allow_nan=False, indent=2, sort_keys=True) + "\n").encode("utf-8"),
0o600,
return _execute_and_persist_result(
manifest,
payload_root,
request_dir,
archive_sha256,
helper_sha256,
runner=runner,
)
return result
finally:
os.close(lock_descriptor)
finally:

308
ops/run_gcp_leoclean_nosend_iap.py
View file

@ -33,7 +33,10 @@ SCHEMA = "livingip.leocleanNoSendIapRequest.v1"
PLAN_SCHEMA = "livingip.leocleanNoSendIapPlan.v1"
REMOTE_RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v1"
REMOTE_FAILURE_SCHEMA = "livingip.leocleanNoSendIapFailure.v1"
REMOTE_OPERATION_FAILURE_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v1"
RESULT_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapResultReceipt.v1"
FAILURE_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapFailureReceipt.v1"
LOCAL_FAILURE_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapLocalFailureReceipt.v1"
EXECUTION_SCHEMA = "livingip.leocleanNoSendIapExecution.v1"
PROJECT = package.PROJECT
ZONE = package.ZONE
@ -78,6 +81,26 @@ FORBIDDEN_OUTPUT_MARKERS = (
"77.42.65.182",
"teleo-prod-1",
)
POST_INSTALL_FAILURES = frozenset(
{
("install_receipt", "install_receipt_invalid"),
("service_before_restart", "service_verification_failed"),
("database_before_restart", "database_verification_failed"),
("service_restart", "service_restart_failed"),
("service_after_restart", "service_verification_failed"),
("restart_identity", "restart_identity_mismatch"),
("database_after_restart", "database_verification_failed"),
}
)
LOCAL_FAILURES = frozenset(
{
("iap_transport", "upload_failed"),
("iap_transport", "remote_command_interrupted"),
("remote_receipt", "absent"),
("remote_receipt", "malformed"),
("remote_receipt", "contract_mismatch"),
}
)
BOOTSTRAP_CODE = textwrap.dedent(
r"""
import fcntl, hashlib, io, json, os, pwd, re, stat, sys, tarfile
@ -203,7 +226,7 @@ def _strict_json(raw: bytes, label: str) -> dict[str, Any]:
object_pairs_hook=reject_duplicates,
parse_constant=lambda _value: (_ for _ in ()).throw(ValueError("constant")),
)
except (UnicodeError, ValueError, json.JSONDecodeError) as exc:
except (UnicodeError, ValueError, json.JSONDecodeError, RecursionError) as exc:
raise IapRequestError(f"{label} was invalid JSON") from exc
_require(isinstance(value, dict), f"{label} was not a JSON object")
return value
@ -717,6 +740,17 @@ def _action_receipts(commands: tuple[list[str], list[str], list[str]]) -> list[d
]
def _expected_authority(operation: str) -> dict[str, bool]:
_require(operation in OPERATIONS, "operation was invalid")
return {
"gcp_mutation": operation in {"install", "rollback"},
"service_restart": operation in {"install", "rollback"},
"database_or_secret_provisioning": False,
"production_or_transport": False,
"vps": False,
}
def prepare_request(
operation: str,
request_id: str,
@ -833,13 +867,7 @@ def prepare_request(
"ssh_key": str(ssh_key),
},
"actions": actions,
"authority": {
"gcp_mutation": operation in {"install", "rollback"},
"service_restart": operation in {"install", "rollback"},
"database_or_secret_provisioning": False,
"production_or_transport": False,
"vps": False,
},
"authority": _expected_authority(operation),
}
@ -1028,33 +1056,114 @@ def validate_remote_result(plan: Mapping[str, Any], result: Mapping[str, Any]) -
def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any]) -> None:
_require(
set(result)
== {
"schema",
"status",
"operation",
"request_id",
"runner_revision",
"release_revision",
"release_sha256",
"image_digest",
"image_config_digest",
"release_artifact_sha256",
"request_binding",
"failure_stage",
"failure_reason",
"rollback",
"authority",
},
"remote failure fields were not exact",
)
request = plan["request"]
release = plan["release"]
rollback = result.get("rollback")
common_fields = {
"schema",
"status",
"operation",
"request_id",
"runner_revision",
"release_revision",
"release_sha256",
"image_digest",
"image_config_digest",
"release_artifact_sha256",
"request_binding",
"failure_stage",
"failure_reason",
"rollback",
"authority",
}
schema = result.get("schema")
if schema == REMOTE_FAILURE_SCHEMA:
_require(set(result) == common_fields, "remote failure fields were not exact")
failure_contract = (
(result.get("failure_stage"), result.get("failure_reason")) in installer.VERIFICATION_FAILURES
and rollback
in (
{"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"},
{"status": "pass", "mode": "explicit", "restored_posture": "active"},
{"status": "pass", "mode": "explicit", "restored_posture": "inactive"},
{"status": "pass", "mode": "explicit", "restored_posture": "absent"},
{"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"},
)
)
else:
_require(
schema == REMOTE_OPERATION_FAILURE_SCHEMA and set(result) == common_fields | {"fail_closed"},
"remote failure fields were not exact",
)
failure_pair = (result.get("failure_stage"), result.get("failure_reason"))
rollback_passed = (
isinstance(rollback, dict)
and set(rollback) == {"status", "mode", "restored_posture"}
and rollback.get("status") == "pass"
and rollback.get("mode") == "explicit"
and rollback.get("restored_posture") in {"active", "inactive", "absent"}
)
rollback_failed = rollback == {
"status": "fail",
"mode": "explicit",
"restored_posture": "unknown",
}
fail_closed = result.get("fail_closed")
if rollback_passed:
fail_closed_contract = fail_closed == {
"status": "not_required",
"service_posture": rollback["restored_posture"],
"active_state": "not_observed",
"sub_state": "not_observed",
"main_pid": None,
"n_restarts": None,
"candidate_container": "not_observed",
}
else:
fail_closed_contract = (
fail_closed
== {
"status": "fail",
"service_posture": "unknown",
"active_state": "unknown",
"sub_state": "unknown",
"main_pid": None,
"n_restarts": None,
"candidate_container": "unknown",
}
or (
isinstance(fail_closed, dict)
and set(fail_closed)
== {
"status",
"service_posture",
"active_state",
"sub_state",
"main_pid",
"n_restarts",
"candidate_container",
}
and fail_closed.get("status") == "pass"
and fail_closed.get("service_posture") == "inactive"
and fail_closed.get("active_state") == "inactive"
and fail_closed.get("sub_state") == "dead"
and fail_closed.get("main_pid") == 0
and isinstance(fail_closed.get("n_restarts"), int)
and not isinstance(fail_closed.get("n_restarts"), bool)
and fail_closed["n_restarts"] >= 0
and fail_closed.get("candidate_container") == "absent"
)
)
failure_contract = (
plan["operation"] == "install"
and (
(failure_pair in POST_INSTALL_FAILURES and (rollback_passed or rollback_failed))
or (failure_pair in installer.VERIFICATION_FAILURES and rollback_failed)
)
and fail_closed_contract
)
_require(
result.get("schema") == REMOTE_FAILURE_SCHEMA
and result.get("status") == "fail"
result.get("status") == "fail"
and result.get("operation") == plan["operation"]
and result.get("request_id") == plan["request_id"]
and result.get("runner_revision") == plan["runner_revision"]
@ -1069,15 +1178,7 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any])
"helper_sha256": request["helper_sha256"],
"manifest_sha256": request["manifest_sha256"],
}
and (result.get("failure_stage"), result.get("failure_reason")) in installer.VERIFICATION_FAILURES
and rollback
in (
{"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"},
{"status": "pass", "mode": "explicit", "restored_posture": "active"},
{"status": "pass", "mode": "explicit", "restored_posture": "inactive"},
{"status": "pass", "mode": "explicit", "restored_posture": "absent"},
{"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"},
)
and failure_contract
and result.get("authority")
== {
"sending_enabled": False,
@ -1089,33 +1190,8 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any])
)
def _write_result_receipt(
plan: Mapping[str, Any],
result: Mapping[str, Any],
*,
status: str = "pass",
) -> tuple[Path, dict[str, Any]]:
_require(status in {"pass", "fail"} and result.get("status") == status, "local result status was invalid")
archive = Path(str(plan["request"]["archive"]))
output = archive.parent / f"livingip-leoclean-nosend-result-{plan['request_id']}.json"
def _publish_local_receipt(output: Path, stable: Mapping[str, Any]) -> dict[str, Any]:
_require(not output.exists() and not output.is_symlink(), "local result receipt already existed")
stable = {
"schema": RESULT_RECEIPT_SCHEMA,
"status": status,
"operation": plan["operation"],
"request_id": plan["request_id"],
"runner_revision": plan["runner_revision"],
"release_revision": plan["release_revision"],
"request": {
"archive_sha256": plan["request"]["archive_sha256"],
"helper_sha256": plan["request"]["helper_sha256"],
"manifest_sha256": plan["request"]["manifest_sha256"],
},
"release": dict(plan["release"]),
"release_artifact": dict(plan["release_artifact"]),
"expected": dict(plan["expected"]),
"remote_result": dict(result),
}
receipt = {**stable, "receipt_sha256": package.canonical_sha256(stable)}
content = _canonical_json(receipt)
descriptor, temporary_name = tempfile.mkstemp(prefix=".iap-result-", dir=output.parent)
@ -1142,6 +1218,74 @@ def _write_result_receipt(
temporary.unlink()
except FileNotFoundError:
pass
return receipt
def _write_result_receipt(
plan: Mapping[str, Any],
result: Mapping[str, Any],
*,
status: str = "pass",
) -> tuple[Path, dict[str, Any]]:
_require(status in {"pass", "fail"} and result.get("status") == status, "local result status was invalid")
archive = Path(str(plan["request"]["archive"]))
output = archive.parent / f"livingip-leoclean-nosend-result-{plan['request_id']}.json"
stable = {
"schema": RESULT_RECEIPT_SCHEMA if status == "pass" else FAILURE_RECEIPT_SCHEMA,
"status": status,
"operation": plan["operation"],
"request_id": plan["request_id"],
"runner_revision": plan["runner_revision"],
"release_revision": plan["release_revision"],
"request": {
"archive_sha256": plan["request"]["archive_sha256"],
"helper_sha256": plan["request"]["helper_sha256"],
"manifest_sha256": plan["request"]["manifest_sha256"],
},
"release": dict(plan["release"]),
"release_artifact": dict(plan["release_artifact"]),
"expected": dict(plan["expected"]),
"remote_result": dict(result),
}
receipt = _publish_local_receipt(output, stable)
return output, receipt
def _write_local_failure_receipt(
plan: Mapping[str, Any],
*,
failure_stage: str,
failure_reason: str,
cleanup_status: str,
) -> tuple[Path, dict[str, Any]]:
_require(
(failure_stage, failure_reason) in LOCAL_FAILURES and cleanup_status in {"pass", "fail"},
"local failure classification was invalid",
)
archive = Path(str(plan["request"]["archive"]))
output = archive.parent / f"livingip-leoclean-nosend-result-{plan['request_id']}.json"
stable = {
"schema": LOCAL_FAILURE_RECEIPT_SCHEMA,
"status": "fail",
"operation": plan["operation"],
"request_id": plan["request_id"],
"runner_revision": plan["runner_revision"],
"release_revision": plan["release_revision"],
"request": {
"archive_sha256": plan["request"]["archive_sha256"],
"helper_sha256": plan["request"]["helper_sha256"],
"manifest_sha256": plan["request"]["manifest_sha256"],
},
"release": dict(plan["release"]),
"release_artifact": dict(plan["release_artifact"]),
"expected": dict(plan["expected"]),
"failure_stage": failure_stage,
"failure_reason": failure_reason,
"cleanup": {"status": cleanup_status, "mode": "failed_transfer"},
"remote_state": "unknown",
"requested_authority": dict(plan["authority"]),
}
receipt = _publish_local_receipt(output, stable)
return output, receipt
@ -1177,6 +1321,7 @@ def execute_plan(
)
operation = plan.get("operation")
_require(operation in {"install", "verify", "rollback"}, "operation was not executable")
_require(plan.get("authority") == _expected_authority(operation), "request authority was not exact")
request_id = plan.get("request_id")
runner_revision = plan.get("runner_revision")
release_revision = plan.get("release_revision")
@ -1261,9 +1406,12 @@ def execute_plan(
scp, ssh, cleanup = expected_commands
validate_remote_main(runner_revision, runner=git_runner)
environment = dict(gcloud_env or gcloud_environment())
failure_stage = "iap_transport"
failure_reason = "upload_failed"
try:
try:
_run_bounded(list(scp), runner=runner, label="IAP upload", environment=environment)
failure_reason = "remote_command_interrupted"
completed = _run_bounded(
list(ssh),
runner=runner,
@ -1271,8 +1419,17 @@ def execute_plan(
environment=environment,
accepted_returncodes=(0, 65),
)
failure_stage = "remote_receipt"
if not completed.stdout:
failure_reason = "absent"
raise IapRequestError("remote result was absent")
failure_reason = "malformed"
result = _strict_json(completed.stdout, "remote result")
encoded = json.dumps(result, allow_nan=False, sort_keys=True).casefold()
try:
encoded = json.dumps(result, allow_nan=False, sort_keys=True).casefold()
except (TypeError, ValueError, RecursionError) as exc:
raise IapRequestError("remote result could not be normalized") from exc
failure_reason = "contract_mismatch"
_require(
not any(marker in encoded for marker in FORBIDDEN_OUTPUT_MARKERS),
"remote result was not sanitized",
@ -1282,6 +1439,7 @@ def execute_plan(
else:
validate_remote_result(plan, result)
except IapRequestError as exc:
cleanup_status = "pass"
try:
_run_bounded(
list(cleanup),
@ -1289,9 +1447,19 @@ def execute_plan(
label="IAP failed-transfer cleanup",
environment=environment,
)
except IapRequestError as cleanup_exc:
raise IapRequestError("IAP operation failed and remote upload cleanup was unproven") from cleanup_exc
raise exc
except IapRequestError:
cleanup_status = "fail"
receipt_path, _receipt = _write_local_failure_receipt(
plan,
failure_stage=failure_stage,
failure_reason=failure_reason,
cleanup_status=cleanup_status,
)
raise IapRequestError(
"IAP operation failed closed "
f"at {failure_stage}/{failure_reason}; "
f"sanitized receipt: {receipt_path}"
) from exc
finally:
_cleanup_local_ssh_key(ssh_key)
if completed.returncode == 65:

530
tests/test_gcp_leoclean_nosend_iap_runner.py
View file

@ -159,6 +159,106 @@ def verifier_failure_result(
}
def operational_failure_result(plan: dict[str, Any]) -> dict[str, Any]:
result = verifier_failure_result(
plan,
failure_stage="database_before_restart",
failure_reason="database_verification_failed",
)
result["schema"] = local.REMOTE_OPERATION_FAILURE_SCHEMA
result["rollback"] = {
"status": "fail",
"mode": "explicit",
"restored_posture": "unknown",
}
result["fail_closed"] = {
"status": "pass",
"service_posture": "inactive",
"active_state": "inactive",
"sub_state": "dead",
"main_pid": 0,
"n_restarts": 0,
"candidate_container": "absent",
}
return result
def successful_remote_result(plan: dict[str, Any]) -> dict[str, Any]:
def verification(main_pid: int, invocation: str, container: str) -> dict[str, Any]:
return {
"active_state": "active",
"sub_state": "running",
"main_pid": main_pid,
"n_restarts": 0,
"invocation_id": invocation * 32,
"supervising_process": {"status": "pass"},
"container_id": container * 64,
"container_pid": main_pid + 1,
"health": "healthy",
"validated_process_environment_fields": ["PATH"],
"dropins": "absent",
"environment_files": "absent",
}
def database(name: str, receipt_hash: str) -> dict[str, Any]:
return {
"schema": "livingip.leocleanNoSendContainerDatabaseVerification.v1",
"status": "pass",
"receipt_sha256": receipt_hash * 64,
"receipt_path": f"/var/lib/livingip/leoclean-nosend-iap/requests/{plan['request_id']}/{name}",
"proof": {
"proposal_staging": "function_only",
"proposal_transaction": "rolled_back",
"durable_proposal_created": False,
"canonical_direct_writes": "denied",
"stage_direct_writes": "denied",
"role_escalation": "denied",
},
}
before = verification(101, "2", "3")
if plan["operation"] == "install":
after = verification(103, "5", "6")
service_result = {
"name": local.SERVICE,
"before_restart": before,
"after_restart": after,
}
database_result = {
"status": "pass",
"before_restart": database("database-before-restart.json", "4"),
"after_restart": database("database-after-restart.json", "7"),
}
else:
service_result = {"name": local.SERVICE, "verification": before}
database_result = database("database-verify.json", "4")
return {
"schema": local.REMOTE_RESULT_SCHEMA,
"status": "pass",
"operation": plan["operation"],
"request_id": plan["request_id"],
"runner_revision": plan["runner_revision"],
"release_revision": plan["release_revision"],
"release_sha256": plan["release"]["release_sha256"],
"image_digest": plan["release"]["image_digest"],
"image_config_digest": plan["release"]["image_config_digest"],
"release_artifact_sha256": plan["release_artifact"]["archive_sha256"],
"request_binding": {
"archive_sha256": plan["request"]["archive_sha256"],
"helper_sha256": plan["request"]["helper_sha256"],
"manifest_sha256": plan["request"]["manifest_sha256"],
},
"service": service_result,
"database": database_result,
"authority": {
"sending_enabled": False,
"production_traffic_changed": False,
"vps_touched": False,
"canonical_proposal_approved": False,
},
}
def test_gcloud_commands_are_fixed_iap_only_and_expire_keys(tmp_path: Path) -> None:
scp, ssh, cleanup = fixed_commands(tmp_path)
@ -183,6 +283,10 @@ def test_embedded_root_bootstrap_compiles() -> None:
compile(local.BOOTSTRAP_CODE, "<leoclean-iap-bootstrap>", "exec")
def test_remote_and_local_post_install_failure_contracts_match() -> None:
assert remote.POST_INSTALL_FAILURES == local.POST_INSTALL_FAILURES
def test_live_origin_main_readback_is_exact() -> None:
revision = "a" * 40
responses = iter(
@ -499,6 +603,125 @@ def test_local_result_receipt_is_private_and_self_hashed(tmp_path: Path) -> None
assert json.loads(path.read_text(encoding="utf-8")) == receipt
def test_execute_success_returns_private_receipt(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
plan = executable_plan(tmp_path, "install")
remote_result = successful_remote_result(plan)
monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
calls = 0
def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
nonlocal calls
calls += 1
if calls == 1:
return completed(argv)
return completed(argv, stdout=json.dumps(remote_result).encode())
execution = local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
receipt_path = Path(execution["receipt_path"])
receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
assert calls == 2
assert execution["status"] == "pass"
assert stat.S_IMODE(receipt_path.stat().st_mode) == 0o600
assert receipt["remote_result"] == remote_result
assert receipt["receipt_sha256"] == execution["receipt_sha256"]
@pytest.mark.parametrize("remote_returncode", [0, 65], ids=["success", "failure"])
def test_execute_remote_without_output_records_absent_receipt(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
remote_returncode: int,
) -> None:
plan = executable_plan(tmp_path, "verify")
monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
calls = 0
def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
nonlocal calls
calls += 1
return completed(argv, returncode=remote_returncode if calls == 2 else 0)
with pytest.raises(local.IapRequestError, match="remote_receipt/absent"):
local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
assert calls == 3
assert receipt["schema"] == local.LOCAL_FAILURE_RECEIPT_SCHEMA
assert (receipt["failure_stage"], receipt["failure_reason"]) == ("remote_receipt", "absent")
assert receipt["cleanup"] == {"status": "pass", "mode": "failed_transfer"}
assert receipt["remote_state"] == "unknown"
def test_execute_transport_interruption_records_bounded_failure(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
plan = executable_plan(tmp_path, "verify")
monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
calls = 0
def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
nonlocal calls
calls += 1
if calls == 2:
raise subprocess.TimeoutExpired(argv, 300, output=b"access_token=hidden")
return completed(argv)
with pytest.raises(local.IapRequestError, match="iap_transport/remote_command_interrupted") as caught:
local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
receipt_text = receipt_path.read_text(encoding="utf-8")
receipt = json.loads(receipt_text)
assert calls == 3
assert (receipt["failure_stage"], receipt["failure_reason"]) == (
"iap_transport",
"remote_command_interrupted",
)
assert "hidden" not in receipt_text
assert "access_token" not in receipt_text
assert "hidden" not in str(caught.value)
@pytest.mark.parametrize(
"payload",
[
b"{not-json",
(b"[" * 1100) + b"0" + (b"]" * 1100),
],
ids=["invalid-json", "recursive-json"],
)
def test_execute_malformed_remote_receipt_fails_closed(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
payload: bytes,
) -> None:
plan = executable_plan(tmp_path, "verify")
monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
calls = 0
def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
nonlocal calls
calls += 1
if calls == 2:
return completed(argv, stdout=payload)
return completed(argv)
with pytest.raises(local.IapRequestError, match="remote_receipt/malformed"):
local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
assert calls == 3
assert (receipt["failure_stage"], receipt["failure_reason"]) == ("remote_receipt", "malformed")
assert receipt["remote_state"] == "unknown"
def test_execute_records_sanitized_verifier_failure(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
@ -532,6 +755,94 @@ def test_execute_records_sanitized_verifier_failure(
assert "remote_operation_failed" not in str(caught.value)
def test_execute_records_operational_failure_and_inert_state(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
plan = executable_plan(tmp_path, "install")
failure = operational_failure_result(plan)
monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
calls = 0
def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
nonlocal calls
calls += 1
if calls == 1:
return completed(argv)
return completed(argv, returncode=65, stdout=json.dumps(failure).encode())
with pytest.raises(
local.IapRequestError,
match="database_before_restart/database_verification_failed",
):
local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
assert calls == 2
assert receipt["remote_result"] == failure
assert receipt["remote_result"]["fail_closed"]["service_posture"] == "inactive"
def test_operational_failure_persists_and_propagates_to_local_receipt(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
) -> None:
plan = executable_plan(tmp_path, "install")
failure = operational_failure_result(plan)
request_dir = private_directory(tmp_path / "remote-request")
payload_root = private_directory(tmp_path / "remote-payload")
def fail_execute(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
raise remote.RemoteOperationFailure(failure)
monkeypatch.setattr(remote, "execute_request", fail_execute)
persisted = remote._execute_and_persist_result(
plan,
payload_root,
request_dir,
plan["request"]["archive_sha256"],
plan["request"]["helper_sha256"],
runner=lambda argv, **_kwargs: completed(argv),
)
remote_receipt_path = request_dir / "result.json"
assert persisted == failure
assert stat.S_IMODE(remote_receipt_path.stat().st_mode) == 0o600
assert json.loads(remote_receipt_path.read_text(encoding="utf-8")) == failure
monkeypatch.setattr(remote, "receive_request", lambda *_args: persisted)
previous_umask = os.umask(0o077)
try:
returncode = remote.main(["install", "request", "runner", "release", "archive", "helper"])
finally:
os.umask(previous_umask)
captured = capsys.readouterr()
assert returncode == 65
assert captured.err == ""
assert json.loads(captured.out) == failure
monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
calls = 0
def local_runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
nonlocal calls
calls += 1
if calls == 1:
return completed(argv)
return completed(argv, returncode=65, stdout=captured.out.encode())
with pytest.raises(local.IapRequestError, match="database_before_restart/database_verification_failed"):
local.execute_plan(plan, runner=local_runner, gcloud_env={"PATH": "/usr/bin"})
local_receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
local_receipt = json.loads(local_receipt_path.read_text(encoding="utf-8"))
stable = {key: value for key, value in local_receipt.items() if key != "receipt_sha256"}
assert local_receipt["schema"] == local.FAILURE_RECEIPT_SCHEMA
assert local_receipt["remote_result"] == failure
assert local_receipt["receipt_sha256"] == local.package.canonical_sha256(stable)
def test_remote_failure_rejects_unallowlisted_reason(tmp_path: Path) -> None:
plan = executable_plan(tmp_path, "install")
failure = verifier_failure_result(plan)
@ -541,6 +852,15 @@ def test_remote_failure_rejects_unallowlisted_reason(tmp_path: Path) -> None:
local.validate_remote_failure(plan, failure)
def test_operational_failure_rejects_extra_rollback_fields(tmp_path: Path) -> None:
plan = executable_plan(tmp_path, "install")
failure = operational_failure_result(plan)
failure["rollback"]["credential"] = "not-allowed"
with pytest.raises(local.IapRequestError, match="did not match"):
local.validate_remote_failure(plan, failure)
def test_remote_main_emits_structured_failure_on_stdout(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
@ -570,8 +890,16 @@ def test_execute_failure_does_not_forward_command_output(monkeypatch: pytest.Mon
with pytest.raises(local.IapRequestError) as caught:
local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
receipt_text = receipt_path.read_text(encoding="utf-8")
receipt = json.loads(receipt_text)
assert "extremely-secret" not in str(caught.value)
assert "password" not in str(caught.value).casefold()
assert "extremely-secret" not in receipt_text
assert "password" not in receipt_text.casefold()
assert receipt["cleanup"] == {"status": "fail", "mode": "failed_transfer"}
assert "authority" not in receipt
assert receipt["requested_authority"] == plan["authority"]
def test_execute_rejects_unsanitized_remote_result(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
@ -595,7 +923,16 @@ def test_execute_rejects_unsanitized_remote_result(monkeypatch: pytest.MonkeyPat
with pytest.raises(local.IapRequestError) as caught:
local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
receipt_text = receipt_path.read_text(encoding="utf-8")
receipt = json.loads(receipt_text)
assert "not-allowed" not in str(caught.value)
assert "not-allowed" not in receipt_text
assert (receipt["failure_stage"], receipt["failure_reason"]) == (
"remote_receipt",
"contract_mismatch",
)
assert receipt["cleanup"] == {"status": "pass", "mode": "failed_transfer"}
def test_execute_rederives_commands_and_rejects_plan_tampering(tmp_path: Path) -> None:
@ -606,6 +943,22 @@ def test_execute_rederives_commands_and_rejects_plan_tampering(tmp_path: Path) -
local.execute_plan(plan, runner=lambda argv, **_kwargs: completed(argv))
def test_execute_rejects_authority_tampering_before_dispatch(tmp_path: Path) -> None:
plan = executable_plan(tmp_path, "verify")
plan["authority"]["credential"] = "not-allowed"
calls: list[list[str]] = []
with pytest.raises(local.IapRequestError, match="request authority was not exact") as caught:
local.execute_plan(
plan,
runner=lambda argv, **_kwargs: (calls.append(argv) or completed(argv)),
)
assert calls == []
assert "not-allowed" not in str(caught.value)
assert not (tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json").exists()
@pytest.mark.parametrize("fail_pull", [False, True])
def test_ephemeral_registry_credentials_are_always_removed(tmp_path: Path, fail_pull: bool) -> None:
calls: list[tuple[list[str], bytes | None]] = []
@ -838,6 +1191,7 @@ def test_post_install_proof_failure_executes_explicit_rollback(
installer_module = SimpleNamespace(
SYSTEMCTL="/usr/bin/systemctl",
VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES,
InstallPaths=Paths,
validate_release_bundle=lambda _bundle: (release, b"unit"),
_validate_source_binding=lambda _binding: None,
@ -881,7 +1235,7 @@ def test_post_install_proof_failure_executes_explicit_rollback(
return completed(argv, returncode=1)
return completed(argv)
with pytest.raises(remote.RemoteError, match="install_failed_rolled_back"):
with pytest.raises(remote.RemoteOperationFailure) as caught:
remote.execute_request(
manifest,
tmp_path,
@ -891,6 +1245,180 @@ def test_post_install_proof_failure_executes_explicit_rollback(
runner=runner,
)
assert rollbacks == [release["release_sha256"]]
expected_failure = {
"first_database": ("database_before_restart", "database_verification_failed"),
"restart": ("service_restart", "service_restart_failed"),
"second_database": ("database_after_restart", "database_verification_failed"),
}[failure_point]
result = caught.value.result
assert result["schema"] == remote.OPERATION_FAILURE_RESULT_SCHEMA
assert (result["failure_stage"], result["failure_reason"]) == expected_failure
assert result["rollback"] == {
"status": "pass",
"mode": "explicit",
"restored_posture": "absent",
}
assert result["fail_closed"] == {
"status": "not_required",
"service_posture": "absent",
"active_state": "not_observed",
"sub_state": "not_observed",
"main_pid": None,
"n_restarts": None,
"candidate_container": "not_observed",
}
def test_incomplete_rollback_stops_service_and_reports_inert_posture(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
release = {
"release_sha256": "a" * 64,
"image": {
"reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'b' * 64}",
"digest": f"sha256:{'b' * 64}",
"config_digest": f"sha256:{'c' * 64}",
},
"image_input": {"runtime": {"teleo_git_head": "d" * 40}},
}
manifest = {
"operation": "install",
"request_id": "iap-abcdefghijkl",
"runner_revision": "e" * 40,
"release_revision": "d" * 40,
"manifest_sha256": "f" * 64,
"release": {
"release_sha256": release["release_sha256"],
"image_reference": release["image"]["reference"],
"image_digest": release["image"]["digest"],
"image_config_digest": release["image"]["config_digest"],
},
"release_artifact": {"archive_sha256": "1" * 64},
"installer_source": {"revision": "e" * 40},
}
stops: list[str] = []
absence_checks: list[bool] = []
class Paths:
@staticmethod
def under(_root: Path) -> object:
return object()
postures = iter(
[
SimpleNamespace(
kind="active",
state={
"ActiveState": "active",
"SubState": "running",
"MainPID": 10,
"NRestarts": 0,
"InvocationID": "2" * 32,
},
),
SimpleNamespace(
kind="inactive",
state={
"ActiveState": "inactive",
"SubState": "dead",
"MainPID": 0,
"NRestarts": 0,
"InvocationID": "",
},
),
]
)
installer_module = SimpleNamespace(
SYSTEMCTL="/usr/bin/systemctl",
InstallPaths=Paths,
validate_release_bundle=lambda _bundle: (release, b"unit"),
_validate_source_binding=lambda _binding: None,
install_release=lambda *_args, **_kwargs: {"status": "pass"},
verify_running_release=lambda *_args, **_kwargs: {
"main_pid": 10,
"invocation_id": "2" * 32,
"container_id": "3" * 64,
"n_restarts": 0,
},
rollback_release=lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("rollback failed")),
_systemctl=lambda action, _runner: stops.append(action),
_stable_service_posture=lambda *_args, **_kwargs: next(postures),
_require_candidate_absent=lambda **_kwargs: absence_checks.append(True),
VERIFICATION_FAILURES=frozenset(),
)
package_module = SimpleNamespace(
render_systemd_unit=lambda _release: "unit",
canonical_sha256=local.package.canonical_sha256,
)
monkeypatch.setattr(remote, "validate_original_release_artifact", lambda *_args, **_kwargs: None)
monkeypatch.setattr(remote, "_load_contract_modules", lambda _root: (package_module, installer_module))
monkeypatch.setattr(remote, "pull_immutable_image", lambda *_args, **_kwargs: None)
monkeypatch.setattr(
remote,
"_verify_database",
lambda **_kwargs: (_ for _ in ()).throw(remote.RemoteError("database failed")),
)
monkeypatch.setattr(remote.time, "sleep", lambda _seconds: None)
with pytest.raises(remote.RemoteOperationFailure) as caught:
remote.execute_request(
manifest,
tmp_path,
tmp_path,
"6" * 64,
"7" * 64,
runner=lambda argv, **_kwargs: completed(argv),
)
result = caught.value.result
assert stops == ["stop"]
assert absence_checks == [True]
assert result["rollback"] == {
"status": "fail",
"mode": "explicit",
"restored_posture": "unknown",
}
assert result["fail_closed"] == {
"status": "pass",
"service_posture": "inactive",
"active_state": "inactive",
"sub_state": "dead",
"main_pid": 0,
"n_restarts": 0,
"candidate_container": "absent",
}
assert result["authority"]["sending_enabled"] is False
def test_fail_closed_stop_refuses_untrusted_unit_or_dropin() -> None:
stop_calls: list[str] = []
installer_module = SimpleNamespace(
_systemctl=lambda action, _runner: stop_calls.append(action),
_stable_service_posture=lambda *_args, **_kwargs: (_ for _ in ()).throw(
local.installer.InstallError("unit or drop-in drift")
),
verify_running_release=lambda *_args, **_kwargs: {"status": "unexpected"},
_require_candidate_absent=lambda **_kwargs: None,
)
result = remote._fail_closed_inert_service(
installer_module=installer_module,
release={"release_sha256": "a" * 64},
paths=object(),
runner=lambda argv, **_kwargs: completed(argv),
)
assert stop_calls == []
assert result == {
"status": "fail",
"service_posture": "unknown",
"active_state": "unknown",
"sub_state": "unknown",
"main_pid": None,
"n_restarts": None,
"candidate_container": "unknown",
}
@pytest.mark.parametrize(