From a1a1aea16fd5d22d7f48ad61ea33abb24229b7f8 Mon Sep 17 00:00:00 2001 From: fwazb Date: Thu, 23 Jul 2026 12:06:35 -0700 Subject: [PATCH] Report no-send verifier startup failures --- ops/gcp_leoclean_nosend_iap_remote.py | 174 +++++++++++-- ops/install_gcp_leoclean_nosend_release.py | 251 +++++++++++++++---- ops/run_gcp_leoclean_nosend_iap.py | 105 +++++++- tests/test_gcp_leoclean_nosend_iap_runner.py | 172 ++++++++++++- tests/test_gcp_leoclean_nosend_installer.py | 72 ++++++ 5 files changed, 693 insertions(+), 81 deletions(-) diff --git a/ops/gcp_leoclean_nosend_iap_remote.py b/ops/gcp_leoclean_nosend_iap_remote.py index eab9789..0bfde9f 100644 --- a/ops/gcp_leoclean_nosend_iap_remote.py +++ b/ops/gcp_leoclean_nosend_iap_remote.py @@ -23,6 +23,7 @@ from typing import Any SCHEMA = "livingip.leocleanNoSendIapRequest.v1" RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v1" +FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapFailure.v1" PROJECT = "teleo-501523" ZONE = "europe-west6-a" INSTANCE = "teleo-staging-1" @@ -77,6 +78,14 @@ class RemoteError(RuntimeError): """A remote request failed closed without exposing command output.""" +class RemoteOperationFailure(RemoteError): + """A sanitized, request-bound failure that can be returned to the caller.""" + + def __init__(self, result: Mapping[str, Any]) -> None: + super().__init__("remote_operation_failed") + self.result = dict(result) + + def _require(condition: bool, message: str) -> None: if not condition: raise RemoteError(message) @@ -781,6 +790,68 @@ def _rollback_result( } +def _verification_failure_result( + *, + manifest: Mapping[str, Any], + release: Mapping[str, Any], + failure: Exception, + rollback: Mapping[str, str], + archive_sha256: str, + helper_sha256: str, + installer_module: Any, +) -> dict[str, Any]: + 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", + ) + _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"}, + ), + "verification rollback summary was invalid", + ) + return { + "schema": 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), + "authority": { + "sending_enabled": False, + "production_traffic_changed": False, + "vps_touched": False, + "canonical_proposal_approved": False, + }, + } + + +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) + + def execute_request( manifest: Mapping[str, Any], payload_root: Path, @@ -821,18 +892,33 @@ def execute_request( paths = installer_module.InstallPaths.under(Path("/")) if operation == "install": pull_immutable_image(release["image"]["reference"], request_dir, runner=runner) - install_receipt = installer_module.install_release( - bundle, - paths=paths, - source_binding=source_binding, - runner=lambda argv: _run( - argv, - runner=runner, - label="installer command", - require_success=False, - ), - execute_restart=True, - ) + try: + install_receipt = installer_module.install_release( + bundle, + paths=paths, + source_binding=source_binding, + runner=lambda argv: _run( + argv, + runner=runner, + label="installer command", + require_success=False, + ), + execute_restart=True, + ) + except Exception as exc: + if not _is_verification_failure(installer_module, exc): + raise + raise RemoteOperationFailure( + _verification_failure_result( + manifest=manifest, + release=release, + failure=exc, + rollback={"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"}, + archive_sha256=archive_sha256, + helper_sha256=helper_sha256, + installer_module=installer_module, + ) + ) from exc _require(install_receipt.get("status") == "pass", "installation did not pass") elif operation == "rollback": rollback = _execute_explicit_rollback( @@ -907,7 +993,7 @@ def execute_request( source_binding=source_binding, runner=runner, ) - _rollback_result( + rollback_result = _rollback_result( manifest=manifest, release=release, rollback=recovery, @@ -916,6 +1002,22 @@ def execute_request( ) except Exception as rollback_exc: raise RemoteError("install_failed_rollback_incomplete") from rollback_exc + if _is_verification_failure(installer_module, exc): + raise RemoteOperationFailure( + _verification_failure_result( + manifest=manifest, + release=release, + failure=exc, + rollback={ + "status": "pass", + "mode": "explicit", + "restored_posture": rollback_result["service"]["restored_posture"], + }, + archive_sha256=archive_sha256, + helper_sha256=helper_sha256, + installer_module=installer_module, + ) + ) from exc raise RemoteError("install_failed_rolled_back") from exc service = { "name": SERVICE, @@ -928,11 +1030,26 @@ def execute_request( "after_restart": after_database, } else: - verification = installer_module.verify_running_release( - release, - runner=contract_runner, - paths=paths, - ) + try: + verification = installer_module.verify_running_release( + release, + runner=contract_runner, + paths=paths, + ) + except Exception as exc: + if not _is_verification_failure(installer_module, exc): + raise + raise RemoteOperationFailure( + _verification_failure_result( + manifest=manifest, + release=release, + failure=exc, + rollback={"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"}, + archive_sha256=archive_sha256, + helper_sha256=helper_sha256, + installer_module=installer_module, + ) + ) from exc database = _verify_database( container_id=verification["container_id"], run_id=f"{manifest['request_id']}-verify", @@ -1037,14 +1154,17 @@ def receive_request( release_revision=release_revision, ) verify_service_account(runner=runner) - result = execute_request( - manifest, - payload_root, - request_dir, - archive_sha256, - helper_sha256, - 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, @@ -1066,7 +1186,7 @@ def main(argv: list[str] | None = None) -> int: _require(len(arguments) == 6, "remote arguments were not exact") result = receive_request(*arguments) sys.stdout.write(json.dumps(result, allow_nan=False, separators=(",", ":"), sort_keys=True) + "\n") - return 0 + return 0 if result.get("status") == "pass" else 65 except RemoteError as exc: error = str(exc) if error not in {"install_failed_rolled_back", "install_failed_rollback_incomplete"}: diff --git a/ops/install_gcp_leoclean_nosend_release.py b/ops/install_gcp_leoclean_nosend_release.py index 8395f54..8f7740e 100644 --- a/ops/install_gcp_leoclean_nosend_release.py +++ b/ops/install_gcp_leoclean_nosend_release.py @@ -110,6 +110,26 @@ FORBIDDEN_PROCESS_PREFIXES = ( "LD_", "PYTHON", ) +VERIFICATION_FAILURES = frozenset( + { + ("service_activation", "service_readback_invalid"), + ("service_activation", "service_not_active"), + ("invocation_identity", "invocation_id_invalid"), + ("unit_dropins", "dropin_contract_mismatch"), + ("unit_configuration", "unit_contract_mismatch"), + ("startup_identity", "service_identity_changed"), + ("startup_identity", "container_id_invalid"), + ("process_environment", "process_environment_mismatch"), + ("supervising_process", "supervising_process_mismatch"), + ("container_startup", "container_contract_mismatch"), + ("container_startup", "container_not_healthy"), + ("image_identity", "image_contract_mismatch"), + ("stable_container", "container_stability_mismatch"), + ("stable_unit_configuration", "unit_contract_drift"), + ("stable_supervising_process", "supervising_process_drift"), + ("stable_service_identity", "service_identity_drift"), + } +) Runner = Callable[[list[str]], subprocess.CompletedProcess[bytes]] Sleeper = Callable[[float], None] @@ -119,6 +139,22 @@ class InstallError(RuntimeError): """A bounded installer failure safe to report without command output.""" +class VerificationError(InstallError): + """A bounded verifier failure carrying only allowlisted diagnostic codes.""" + + def __init__( + self, + failure_stage: str, + failure_reason: str, + message: str = "release verification failed", + ) -> 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 + + class AtomicWriteError(InstallError): """An atomic publication failure retaining any published inode identity.""" @@ -132,6 +168,29 @@ def _require(condition: bool, message: str) -> None: raise InstallError(message) +def _verification_require( + condition: bool, + failure_stage: str, + failure_reason: str, + message: str = "release verification failed", +) -> None: + if not condition: + raise VerificationError(failure_stage, failure_reason, message) + + +def _verification_check( + failure_stage: str, + failure_reason: str, + check: Callable[[], Any], +) -> Any: + try: + return check() + except VerificationError: + raise + except InstallError as exc: + raise VerificationError(failure_stage, failure_reason, str(exc)) from exc + + @dataclass(frozen=True) class InstallPaths: root: Path @@ -1060,80 +1119,180 @@ def verify_running_release( ) -> dict[str, Any]: first: dict[str, Any] | None = None for _attempt in range(attempts): - observed = _service_state(runner) + observed = _verification_check( + "service_activation", + "service_readback_invalid", + lambda: _service_state(runner), + ) if observed["ActiveState"] == "active" and observed["SubState"] == "running" and observed["MainPID"] > 1: first = observed break sleeper(0.25) - _require(first is not None, "service did not become active within the bounded wait") - _require(HEX_32.fullmatch(str(first["InvocationID"])) is not None, "service invocation ID was invalid") - _dropins_absent(paths) - _validate_unit_configuration(first, paths, release) - validated_environment_fields = _validate_process_environment(first["MainPID"], paths) - supervising_process = _validate_supervising_process( - first["MainPID"], - paths, - release, - str(first["ControlGroup"]), + _verification_require( + first is not None, + "service_activation", + "service_not_active", + "service did not become active within the bounded wait", ) + _verification_require( + HEX_32.fullmatch(str(first["InvocationID"])) is not None, + "invocation_identity", + "invocation_id_invalid", + "service invocation ID was invalid", + ) + _verification_check( + "unit_dropins", + "dropin_contract_mismatch", + lambda: _dropins_absent(paths), + ) + _verification_check( + "unit_configuration", + "unit_contract_mismatch", + lambda: _validate_unit_configuration(first, paths, release), + ) + validated_environment_fields: tuple[str, ...] | None = None + supervising_process: dict[str, Any] | None = None container: dict[str, Any] | None = None for _attempt in range(attempts): - current = _service_state(runner) - _require( + current = _verification_check( + "startup_identity", + "service_identity_changed", + lambda: _service_state(runner), + ) + _verification_require( current["MainPID"] == first["MainPID"] and current["NRestarts"] == first["NRestarts"] and current["InvocationID"] == first["InvocationID"] and current["ControlGroup"] == first["ControlGroup"] and current["ActiveState"] == "active" and current["SubState"] == "running", + "startup_identity", + "service_identity_changed", "service changed during startup verification", ) - container_id = _read_startup_container_id(paths) + container_id = _verification_check( + "startup_identity", + "container_id_invalid", + lambda: _read_startup_container_id(paths), + ) if container_id is None: sleeper(0.25) continue - first_container = _docker_inspect( - ["container", "inspect", package.CONTAINER_NAME], - runner, - paths, - "running container inspection", + if validated_environment_fields is None: + validated_environment_fields = _verification_check( + "process_environment", + "process_environment_mismatch", + lambda current=current: _validate_process_environment(current["MainPID"], paths), + ) + supervising_process = _verification_check( + "supervising_process", + "supervising_process_mismatch", + lambda current=current: _validate_supervising_process( + current["MainPID"], + paths, + release, + str(current["ControlGroup"]), + ), + ) + first_container = _verification_check( + "container_startup", + "container_contract_mismatch", + lambda: _docker_inspect( + ["container", "inspect", package.CONTAINER_NAME], + runner, + paths, + "running container inspection", + ), ) if first_container is None: sleeper(0.25) continue - container = _validate_starting_container( - first_container, - release, - expected_container_id=container_id, + container = _verification_check( + "container_startup", + "container_contract_mismatch", + lambda first_container=first_container, container_id=container_id: _validate_starting_container( + first_container, + release, + expected_container_id=container_id, + ), ) if container is not None: break sleeper(0.25) - _require(container is not None, "container did not become healthy within the bounded wait") - _validate_local_image(release, runner, paths) - second_container = _docker_inspect( - ["container", "inspect", container["container_id"]], - runner, - paths, - "stable container inspection", + _verification_require( + container is not None, + "container_startup", + "container_not_healthy", + "container did not become healthy within the bounded wait", ) - _require(second_container is not None, "running container exited during verification") - stable_container = _validate_starting_container( - second_container, - release, - expected_container_id=container["container_id"], + _verification_require( + validated_environment_fields is not None and supervising_process is not None, + "startup_identity", + "container_id_invalid", + "container identity was not established before verification", ) - _require(stable_container is not None, "running container changed during verification") - second = _service_state(runner) - _validate_unit_configuration(second, paths, release) - stable_supervising_process = _validate_supervising_process( - second["MainPID"], - paths, - release, - str(second["ControlGroup"]), + _verification_check( + "image_identity", + "image_contract_mismatch", + lambda: _validate_local_image(release, runner, paths), ) - stable_container_id = _read_startup_container_id(paths) - _require( + second_container = _verification_check( + "stable_container", + "container_stability_mismatch", + lambda: _docker_inspect( + ["container", "inspect", container["container_id"]], + runner, + paths, + "stable container inspection", + ), + ) + _verification_require( + second_container is not None, + "stable_container", + "container_stability_mismatch", + "running container exited during verification", + ) + stable_container = _verification_check( + "stable_container", + "container_stability_mismatch", + lambda: _validate_starting_container( + second_container, + release, + expected_container_id=container["container_id"], + ), + ) + _verification_require( + stable_container is not None, + "stable_container", + "container_stability_mismatch", + "running container changed during verification", + ) + second = _verification_check( + "stable_service_identity", + "service_identity_drift", + lambda: _service_state(runner), + ) + _verification_check( + "stable_unit_configuration", + "unit_contract_drift", + lambda: _validate_unit_configuration(second, paths, release), + ) + stable_supervising_process = _verification_check( + "stable_supervising_process", + "supervising_process_drift", + lambda: _validate_supervising_process( + second["MainPID"], + paths, + release, + str(second["ControlGroup"]), + ), + ) + stable_container_id = _verification_check( + "stable_service_identity", + "service_identity_drift", + lambda: _read_startup_container_id(paths), + ) + _verification_require( first["MainPID"] == second["MainPID"] and first["NRestarts"] == second["NRestarts"] and first["InvocationID"] == second["InvocationID"] @@ -1143,6 +1302,8 @@ def verify_running_release( and container == stable_container and stable_container_id == container["container_id"] and supervising_process == stable_supervising_process, + "stable_service_identity", + "service_identity_drift", "service or container changed during verification", ) return { diff --git a/ops/run_gcp_leoclean_nosend_iap.py b/ops/run_gcp_leoclean_nosend_iap.py index 617acbc..3461c5a 100644 --- a/ops/run_gcp_leoclean_nosend_iap.py +++ b/ops/run_gcp_leoclean_nosend_iap.py @@ -32,6 +32,7 @@ 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" RESULT_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapResultReceipt.v1" EXECUTION_SCHEMA = "livingip.leocleanNoSendIapExecution.v1" PROJECT = package.PROJECT @@ -215,6 +216,7 @@ def _run_bounded( label: str, input_bytes: bytes | None = None, environment: Mapping[str, str] | None = None, + accepted_returncodes: tuple[int, ...] = (0,), ) -> subprocess.CompletedProcess[bytes]: try: completed = runner( @@ -232,7 +234,7 @@ def _run_bounded( len(completed.stdout) <= MAX_COMMAND_OUTPUT_BYTES and len(completed.stderr) <= MAX_COMMAND_OUTPUT_BYTES, f"{label} output exceeded the safety limit", ) - _require(completed.returncode == 0, f"{label} failed") + _require(completed.returncode in accepted_returncodes, f"{label} failed") return completed @@ -1025,13 +1027,81 @@ def validate_remote_result(plan: Mapping[str, Any], result: Mapping[str, Any]) - ) -def _write_result_receipt(plan: Mapping[str, Any], result: Mapping[str, Any]) -> tuple[Path, dict[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") + _require( + result.get("schema") == REMOTE_FAILURE_SCHEMA + and 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"] + and result.get("release_revision") == plan["release_revision"] + and result.get("release_sha256") == release["release_sha256"] + and result.get("image_digest") == release["image_digest"] + and result.get("image_config_digest") == release["image_config_digest"] + and result.get("release_artifact_sha256") == plan["release_artifact"]["archive_sha256"] + and result.get("request_binding") + == { + "archive_sha256": request["archive_sha256"], + "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 result.get("authority") + == { + "sending_enabled": False, + "production_traffic_changed": False, + "vps_touched": False, + "canonical_proposal_approved": False, + }, + "remote failure did not match the request", + ) + + +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" _require(not output.exists() and not output.is_symlink(), "local result receipt already existed") stable = { "schema": RESULT_RECEIPT_SCHEMA, - "status": "pass", + "status": status, "operation": plan["operation"], "request_id": plan["request_id"], "runner_revision": plan["runner_revision"], @@ -1194,7 +1264,23 @@ def execute_plan( try: try: _run_bounded(list(scp), runner=runner, label="IAP upload", environment=environment) - completed = _run_bounded(list(ssh), runner=runner, label="IAP remote operation", environment=environment) + completed = _run_bounded( + list(ssh), + runner=runner, + label="IAP remote operation", + environment=environment, + accepted_returncodes=(0, 65), + ) + result = _strict_json(completed.stdout, "remote result") + encoded = json.dumps(result, allow_nan=False, sort_keys=True).casefold() + _require( + not any(marker in encoded for marker in FORBIDDEN_OUTPUT_MARKERS), + "remote result was not sanitized", + ) + if completed.returncode == 65: + validate_remote_failure(plan, result) + else: + validate_remote_result(plan, result) except IapRequestError as exc: try: _run_bounded( @@ -1208,10 +1294,13 @@ def execute_plan( raise exc finally: _cleanup_local_ssh_key(ssh_key) - result = _strict_json(completed.stdout, "remote result") - validate_remote_result(plan, result) - encoded = json.dumps(result, allow_nan=False, sort_keys=True).casefold() - _require(not any(marker in encoded for marker in FORBIDDEN_OUTPUT_MARKERS), "remote result was not sanitized") + if completed.returncode == 65: + receipt_path, _receipt = _write_result_receipt(plan, result, status="fail") + raise IapRequestError( + "IAP verifier failed closed " + f"at {result['failure_stage']}/{result['failure_reason']}; " + f"sanitized receipt: {receipt_path}" + ) receipt_path, receipt = _write_result_receipt(plan, result) return { "schema": EXECUTION_SCHEMA, diff --git a/tests/test_gcp_leoclean_nosend_iap_runner.py b/tests/test_gcp_leoclean_nosend_iap_runner.py index f568867..ebf6d89 100644 --- a/tests/test_gcp_leoclean_nosend_iap_runner.py +++ b/tests/test_gcp_leoclean_nosend_iap_runner.py @@ -121,6 +121,44 @@ def executable_plan(tmp_path: Path, operation: str) -> dict[str, Any]: } +def verifier_failure_result( + plan: dict[str, Any], + *, + failure_stage: str = "process_environment", + failure_reason: str = "process_environment_mismatch", +) -> dict[str, Any]: + return { + "schema": local.REMOTE_FAILURE_SCHEMA, + "status": "fail", + "operation": plan["operation"], + "request_id": plan["request_id"], + "runner_revision": plan["runner_revision"], + "release_revision": plan["release_revision"], + "release_sha256": plan["release"]["release_sha256"], + "image_digest": plan["release"]["image_digest"], + "image_config_digest": plan["release"]["image_config_digest"], + "release_artifact_sha256": plan["release_artifact"]["archive_sha256"], + "request_binding": { + "archive_sha256": plan["request"]["archive_sha256"], + "helper_sha256": plan["request"]["helper_sha256"], + "manifest_sha256": plan["request"]["manifest_sha256"], + }, + "failure_stage": failure_stage, + "failure_reason": failure_reason, + "rollback": { + "status": "pass", + "mode": "installer_automatic", + "restored_posture": "prior", + }, + "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) @@ -451,7 +489,7 @@ def test_gcloud_environment_selects_absolute_compatible_python( def test_local_result_receipt_is_private_and_self_hashed(tmp_path: Path) -> None: plan = executable_plan(tmp_path, "verify") - result = {"bounded": "remote-result"} + result = {"status": "pass", "bounded": "remote-result"} path, receipt = local._write_result_receipt(plan, result) stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"} @@ -461,6 +499,64 @@ 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_records_sanitized_verifier_failure( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + plan = executable_plan(tmp_path, "install") + failure = verifier_failure_result(plan) + monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None) + calls = 0 + + def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]: + nonlocal calls + calls += 1 + if calls == 1: + return completed(argv) + return completed( + argv, + returncode=65, + stdout=json.dumps(failure).encode(), + stderr=b'{"error":"remote_operation_failed","status":"fail"}', + ) + + with pytest.raises(local.IapRequestError, match="process_environment/process_environment_mismatch") as caught: + local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"}) + + receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json" + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + assert calls == 2 + assert stat.S_IMODE(receipt_path.stat().st_mode) == 0o600 + assert receipt["status"] == "fail" + assert receipt["remote_result"] == failure + assert "remote_operation_failed" not in str(caught.value) + + +def test_remote_failure_rejects_unallowlisted_reason(tmp_path: Path) -> None: + plan = executable_plan(tmp_path, "install") + failure = verifier_failure_result(plan) + failure["failure_reason"] = "password=not-allowlisted" + + with pytest.raises(local.IapRequestError, match="did not match"): + local.validate_remote_failure(plan, failure) + + +def test_remote_main_emits_structured_failure_on_stdout( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + failure = verifier_failure_result(executable_plan(tmp_path, "install")) + monkeypatch.setattr(remote, "receive_request", lambda *_args: failure) + + returncode = remote.main(["install", "request", "runner", "release", "archive", "helper"]) + + captured = capsys.readouterr() + assert returncode == 65 + assert json.loads(captured.out) == failure + assert captured.err == "" + + def test_execute_failure_does_not_forward_command_output(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: plan = executable_plan(tmp_path, "install") monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None) @@ -609,6 +705,80 @@ def test_rollback_fails_when_explicit_interface_is_absent(tmp_path: Path) -> Non ) +def test_install_verifier_failure_returns_bounded_automatic_rollback_result( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + release = { + "release_sha256": "a" * 64, + "image": { + "reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'b' * 64}", + "digest": f"sha256:{'b' * 64}", + "config_digest": f"sha256:{'c' * 64}", + }, + "image_input": {"runtime": {"teleo_git_head": "d" * 40}}, + } + manifest = { + "operation": "install", + "request_id": "iap-abcdefghijkl", + "runner_revision": "e" * 40, + "release_revision": "d" * 40, + "manifest_sha256": "f" * 64, + "release": { + "release_sha256": release["release_sha256"], + "image_reference": release["image"]["reference"], + "image_digest": release["image"]["digest"], + "image_config_digest": release["image"]["config_digest"], + }, + "release_artifact": {"archive_sha256": "1" * 64}, + "installer_source": {"revision": "e" * 40}, + } + + class Paths: + @staticmethod + def under(_root: Path) -> object: + return object() + + def fail_install(*_args: Any, **_kwargs: Any) -> dict[str, Any]: + raise local.installer.VerificationError( + "process_environment", + "process_environment_mismatch", + ) + + installer_module = SimpleNamespace( + VerificationError=local.installer.VerificationError, + VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES, + InstallPaths=Paths, + validate_release_bundle=lambda _bundle: (release, b"unit"), + _validate_source_binding=lambda _binding: None, + install_release=fail_install, + ) + package_module = SimpleNamespace(render_systemd_unit=lambda _release: "unit") + monkeypatch.setattr(remote, "validate_original_release_artifact", lambda *_args, **_kwargs: None) + monkeypatch.setattr(remote, "_load_contract_modules", lambda _root: (package_module, installer_module)) + monkeypatch.setattr(remote, "pull_immutable_image", lambda *_args, **_kwargs: None) + + with pytest.raises(remote.RemoteOperationFailure) as caught: + remote.execute_request( + manifest, + tmp_path, + tmp_path, + "6" * 64, + "7" * 64, + runner=lambda argv, **_kwargs: completed(argv), + ) + + result = caught.value.result + assert result["failure_stage"] == "process_environment" + assert result["failure_reason"] == "process_environment_mismatch" + assert result["rollback"] == { + "status": "pass", + "mode": "installer_automatic", + "restored_posture": "prior", + } + assert result["authority"]["sending_enabled"] is False + + @pytest.mark.parametrize("failure_point", ["first_database", "restart", "second_database"]) def test_post_install_proof_failure_executes_explicit_rollback( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/test_gcp_leoclean_nosend_installer.py b/tests/test_gcp_leoclean_nosend_installer.py index e2bd5dc..55c762f 100644 --- a/tests/test_gcp_leoclean_nosend_installer.py +++ b/tests/test_gcp_leoclean_nosend_installer.py @@ -522,6 +522,78 @@ def test_first_install_waits_for_complete_cid_and_healthy_container(tmp_path: Pa assert host.stop_calls == 0 +def test_first_install_defers_process_identity_until_release_cid_exists(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) + delayed_container_id: str | None = None + process_identity_restored = False + + def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + nonlocal delayed_container_id + result = host(arguments) + if arguments[:2] == [installer.SYSTEMCTL, "restart"]: + assert host.container is not None + delayed_container_id = str(host.container["Id"]) + paths.cidfile.unlink() + process_root = paths.proc_root / str(host.main_pid) + (process_root / "environ").unlink() + (process_root / "exe").unlink() + (process_root / "cmdline").unlink() + (process_root / "cgroup").unlink() + return result + + def sleeper(_seconds: float) -> None: + nonlocal process_identity_restored + if host.active and delayed_container_id is not None and not process_identity_restored: + host._write_process_environment() + paths.cidfile.write_text(delayed_container_id, encoding="ascii") + paths.cidfile.chmod(0o600) + process_identity_restored = True + + receipt = installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=runner, + sleeper=sleeper, + execute_restart=True, + ) + + assert receipt["status"] == "pass" + assert process_identity_restored + assert host.stop_calls == 0 + + +def test_process_environment_failure_has_bounded_stage_and_reason(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, + ) + process_root = paths.proc_root / str(host.main_pid) + (process_root / "environ").write_bytes(b"PATH=/unexpected\0") + + with pytest.raises(installer.VerificationError) as caught: + installer.verify_running_release( + release, + runner=host, + paths=paths, + sleeper=lambda _seconds: None, + ) + + assert caught.value.failure_stage == "process_environment" + assert caught.value.failure_reason == "process_environment_mismatch" + + def test_first_install_health_timeout_rolls_back_to_absent(tmp_path: Path) -> None: bundle, release, image = release_bundle(tmp_path, "2") paths = install_paths(tmp_path)