From a0fabc307260ffcc4119019331b7ff66ddb5f2a8 Mon Sep 17 00:00:00 2001 From: fwazb Date: Tue, 21 Jul 2026 10:58:36 -0700 Subject: [PATCH] Add fail-closed leoclean staging installer --- docs/gcp-leoclean-nosend-service-package.md | 96 +- docs/workstreams/gcp-leoclean/goals.md | 6 +- ops/gcp_leoclean_nosend_package.py | 50 +- ops/install_gcp_leoclean_nosend_release.py | 1363 +++++++++++++++++++ ops/leoclean_nosend_service_control.py | 319 +++++ tests/test_gcp_leoclean_nosend_installer.py | 1208 ++++++++++++++++ tests/test_gcp_leoclean_nosend_package.py | 12 +- 7 files changed, 3018 insertions(+), 36 deletions(-) create mode 100644 ops/install_gcp_leoclean_nosend_release.py create mode 100644 ops/leoclean_nosend_service_control.py create mode 100644 tests/test_gcp_leoclean_nosend_installer.py diff --git a/docs/gcp-leoclean-nosend-service-package.md b/docs/gcp-leoclean-nosend-service-package.md index bec9237..42ffa1f 100644 --- a/docs/gcp-leoclean-nosend-service-package.md +++ b/docs/gcp-leoclean-nosend-service-package.md @@ -246,6 +246,65 @@ extra descriptor fields, wrong-architecture images, missing repository-digest bindings, mismatched image labels, duplicate JSON keys, and non-finite JSON values are rejected. +## Install, restart, verify, and roll back + +The installer consumes one finalized release-v3 bundle. Its default invocation +is a dry run: it validates the exact two-file bundle, re-renders the unit from +the descriptor, binds the installer/package/control-helper source closure to +the clean Git revision and this checkout's fixed contract root, captures those +stable bytes once, and prints the proposed target without changing the host: + +```bash +python3 ops/install_gcp_leoclean_nosend_release.py \ + --bundle /tmp/leoclean-nosend-release +``` + +After separate host-restart authorization, the same reviewed revision can be +executed as root: + +```bash +sudo python3 ops/install_gcp_leoclean_nosend_release.py \ + --bundle /tmp/leoclean-nosend-release \ + --execute-restart +``` + +The executable path never pulls an image. Before touching the service it +requires the exact digest reference and normalized config digest already to be +present on the local `unix:///var/run/docker.sock` daemon through an empty, +service-private Docker config directory. The complete executable transaction, +including rollback, holds a nonblocking root-owned mode-`0600` lock at +`/run/leoclean-gcp-nosend-installer.lock`. It installs only the deterministic +unit, a standalone release-bound container-control helper, the exact release +descriptor, and a mode-`0600` install receipt. Atomic publications retain the +published inode identity across durability/readback failure; rollback refuses +to overwrite a pathname that no longer names that inode. The helper controls a +container only through a mode-`0600` CID file in the service's mode-`0700` +runtime directory, then rechecks the full ID, name, immutable image reference, +config digest, release hash, role label, and running state. After the trusted ID +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 +`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 +that currently loaded release-bound unit and only afterward replaces and +reloads the unit. Candidate verification also rejects filesystem or effective +systemd drop-ins, environment files, forbidden inherited process environment, +an unexpected fragment, image/config/runtime drift, a non-healthy container, +or any PID, container-ID, invocation-ID, or restart-counter change across the +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. + +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 +boundaries. + ## Offline evidence and live gate Offline tests prove strict receipt/descriptor/inspection schemas, exact @@ -281,30 +340,43 @@ is absent, installation is separate infrastructure work; the deploy must stop instead of falling back to the production sync script or a mutable host virtual environment. +The installer tests use a disposable filesystem root and exact fake +systemd/Docker readbacks. They prove dry-run non-mutation, local digest/config +binding, first install, no-restart idempotence, old-unit-first upgrade ordering, +effective-environment and later-drop-in rejection, PID/restart-race rejection, +automatic rollback to a running prior release, first-install cleanup, managed +file posture, contract-root source binding, stale-loaded-unit rejection, +supervising-process binding, lock contention, post-rename pathname replacement, +exact inactive-posture rollback, and full-ID/config-digest-only container +control including name-rebind refusal. They do not constitute a live staging +restart or a Cloud SQL identity proof. + Publishing the branch and stacked PR is only a review handoff. After both the digest-inspection package and this receipt producer receive exact-head human review and merge, separate live gates remain: staging repository/tag-policy preflight; authorized build/push receipt production; authorized independent -digest finalization; installer/restart/rollback; and Cloud SQL/secret/service -deployment proof. CI carries no Artifact Registry credentials and performs no -push. IAM/secret changes, Cloud SQL access, proposal staging, production -promotion, Telegram traffic, and VPS actions remain separate authority. +digest finalization; an authorized installer/restart execution; and Cloud +SQL/secret/service deployment proof. CI carries no Artifact Registry +credentials and performs no push. IAM/secret changes, Cloud SQL access, +proposal staging, production promotion, Telegram traffic, and VPS actions +remain separate authority. ## Rollback and sunset -This slice's automatic rollback removes an unpublished bundle only. Receipt -pathnames are never removed automatically; the private retained link and any -ambiguous publication residue require explicit operator inspection. Tool-owned -build and candidate tags, the verified +Release finalization's automatic rollback removes an unpublished bundle only. +Receipt pathnames are never removed automatically; the private retained link +and any ambiguous publication residue require explicit operator inspection. +Tool-owned build and candidate tags, the verified repository-digest reference, and content-addressable layers may remain because a Docker-config lock cannot establish exclusive ownership of daemon-global cache state. A remotely pushed candidate is intentionally not deleted during automatic rollback and cannot be consumed without a valid receipt plus independent finalization. Any later cache cleanup requires separately proven -daemon isolation. This slice installs no unit and changes no running service. -Restoring a prior descriptor, unit, and digest -belongs to the later installer slice, which must never restart or edit the -production Leo, Telegram, or VPS services. +daemon isolation. The installer does not delete the bundle, local image, or any +Artifact Registry object. Its bounded rollback restores the managed prior +descriptor, helper, unit, receipt, and service state; it must never restart or +edit the production Leo, Telegram, or VPS services. A live staging execution +and rollback proof remain separately authorized. The fresh tmpfs profile intentionally has no session continuity. That is appropriate for isolated gateway/database and fresh-turn parity proof, but it diff --git a/docs/workstreams/gcp-leoclean/goals.md b/docs/workstreams/gcp-leoclean/goals.md index 5d5876d..f3624ee 100644 --- a/docs/workstreams/gcp-leoclean/goals.md +++ b/docs/workstreams/gcp-leoclean/goals.md @@ -112,8 +112,10 @@ cut over safely and retire the VPS after soak. staging repository/tag policy. 4. Under explicit live authorization, build and push one candidate, publish its receipt, and independently finalize the exact digest. -5. Add the separate fail-closed installer/restart/verification/rollback slice - that consumes the exact inspection-bound release bundle. +5. Review and merge the offline fail-closed + installer/restart/verification/rollback slice that consumes the exact + inspection-bound release bundle; authorize its first staging execution + separately. 6. Provision the scoped Cloud SQL role/secret and deploy the separate staging service. 7. Prove effective identity, reads, function-only proposal staging, denied diff --git a/ops/gcp_leoclean_nosend_package.py b/ops/gcp_leoclean_nosend_package.py index 75ebac9..7133adb 100644 --- a/ops/gcp_leoclean_nosend_package.py +++ b/ops/gcp_leoclean_nosend_package.py @@ -52,6 +52,13 @@ PLATFORM = "linux/amd64" IMAGE_REPOSITORY = "europe-west6-docker.pkg.dev/teleo-501523/teleo/leoclean-nosend-staging" CONTAINER_NAME = "livingip-leoclean-nosend" LOCAL_BUILD_REPOSITORY = "livingip-leoclean-nosend-build" +DOCKER_BINARY = "/usr/bin/docker" +DOCKER_HOST = "unix:///var/run/docker.sock" +SERVICE_DOCKER_CONFIG = "/var/lib/livingip/leoclean-nosend/docker-config" +SERVICE_CONTROL_HELPER = "/usr/local/libexec/livingip/leoclean-nosend-service-control" +SERVICE_RUNTIME_DIRECTORY = "leoclean-gcp-nosend" +SERVICE_CIDFILE = f"/run/{SERVICE_RUNTIME_DIRECTORY}/container.cid" +SERVICE_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" HERMES_COMMIT = "b2f477a30b3c05d0f383c543af98496ae8a96070" PYTHON_VERSION = "3.11.9" @@ -1477,9 +1484,7 @@ def validate_prepared_image_context(context: Path) -> tuple[dict[str, Any], dict } _require(actual_identity_paths == expected_identity_paths, "prepared identity source entry set drifted") actual_script_paths = { - path.relative_to(context).as_posix() - for path in (context / "scripts").rglob("*") - if path.is_file() + path.relative_to(context).as_posix() for path in (context / "scripts").rglob("*") if path.is_file() } _require( actual_script_paths == set(IDENTITY_CONTRACT_SOURCES), @@ -1496,9 +1501,7 @@ def validate_prepared_image_context(context: Path) -> tuple[dict[str, Any], dict _require( isinstance(artifact_files, list) and all( - isinstance(entry, dict) - and isinstance(entry.get("path"), str) - and isinstance(entry.get("mode"), str) + isinstance(entry, dict) and isinstance(entry.get("path"), str) and isinstance(entry.get("mode"), str) for entry in artifact_files ), "prepared artifact file manifest is invalid", @@ -1518,9 +1521,7 @@ def validate_prepared_image_context(context: Path) -> tuple[dict[str, Any], dict _require(bool(re.fullmatch(r"0o[0-7]{3,4}", raw_mode)), "prepared artifact file mode is invalid") expected_file_modes[f"artifact/{entry['path']}"] = int(raw_mode.removeprefix("0o"), 8) expected_file_modes.update({f"identity/{name}": 0o600 for name in IDENTITY_FILES}) - expected_file_modes.update( - {f"identity-sources/{relative}": 0o644 for relative in expected_identity_paths} - ) + expected_file_modes.update({f"identity-sources/{relative}": 0o644 for relative in expected_identity_paths}) expected_file_modes.update({relative: 0o644 for relative in IDENTITY_CONTRACT_SOURCES}) actual_files = {path.relative_to(context).as_posix() for path in context.rglob("*") if path.is_file()} _require(actual_files == set(expected_file_modes), "prepared image file entry set drifted") @@ -1531,16 +1532,13 @@ def validate_prepared_image_context(context: Path) -> tuple[dict[str, Any], dict ) expected_directories = {"artifact", "identity", "identity-sources", "scripts"} expected_directories.update( - f"artifact/{relative}" - for relative in _parent_directory_paths({entry["path"] for entry in artifact_files}) + f"artifact/{relative}" for relative in _parent_directory_paths({entry["path"] for entry in artifact_files}) ) expected_directories.update( _parent_directory_paths({f"identity-sources/{relative}" for relative in expected_identity_paths}) ) expected_directories.update(_parent_directory_paths(set(IDENTITY_CONTRACT_SOURCES))) - actual_directories = { - path.relative_to(context).as_posix() for path in context.rglob("*") if path.is_dir() - } + actual_directories = {path.relative_to(context).as_posix() for path in context.rglob("*") if path.is_dir()} _require(actual_directories == expected_directories, "prepared image directory entry set drifted") _require(stat.S_IMODE(context.stat().st_mode) == 0o700, "prepared image context mode drifted") for relative in actual_directories: @@ -2486,15 +2484,20 @@ def validate_release_descriptor(value: dict[str, Any]) -> None: def render_systemd_unit(release: dict[str, Any]) -> str: validate_release_descriptor(release) image = release["image"]["reference"] + config_digest = release["image"]["config_digest"] release_sha = release["release_sha256"] profile_tmpfs = f"{PROFILE_ROOT}:rw,noexec,nosuid,nodev,size=64m,uid=0,gid=0,mode=0700" temp_tmpfs = f"/tmp:rw,noexec,nosuid,nodev,size=64m,uid={RUNTIME_UID},gid={RUNTIME_GID},mode=0700" postgres_data_tmpfs = "/var/lib/postgresql/data:ro,noexec,nosuid,nodev,size=64k,uid=0,gid=0,mode=0000" exec_start = " ".join( [ - "/usr/bin/docker run", + DOCKER_BINARY, + f"--host={DOCKER_HOST}", + f"--config={SERVICE_DOCKER_CONFIG}", + "run", "--rm", f"--name={CONTAINER_NAME}", + f"--cidfile={SERVICE_CIDFILE}", "--pull=never", "--read-only", "--cap-drop=ALL", @@ -2510,6 +2513,7 @@ def render_systemd_unit(release: dict[str, Any]) -> str: f"--tmpfs={temp_tmpfs}", f"--tmpfs={postgres_data_tmpfs}", f"--label=livingip.release.sha256={release_sha}", + "--label=livingip.container.role=leoclean-gcp-nosend", image, "service", ] @@ -2523,9 +2527,15 @@ Wants=network-online.target [Service] Type=simple +Environment=PATH={SERVICE_PATH} +UnsetEnvironment=DOCKER_HOST DOCKER_CONFIG DOCKER_CONTEXT DOCKER_TLS_VERIFY DOCKER_CERT_PATH DOCKER_API_VERSION DOCKER_CUSTOM_HEADERS BUILDKIT_HOST BUILDX_BUILDER +UnsetEnvironment=HTTP_PROXY HTTPS_PROXY ALL_PROXY NO_PROXY http_proxy https_proxy all_proxy no_proxy +RuntimeDirectory={SERVICE_RUNTIME_DIRECTORY} +RuntimeDirectoryMode=0700 +RuntimeDirectoryPreserve=no +ExecStartPre={SERVICE_CONTROL_HELPER} prestart {release_sha} {image} {config_digest} ExecStart={exec_start} -ExecStop=/usr/bin/docker stop --time=30 {CONTAINER_NAME} -ExecStopPost=-/usr/bin/docker rm --force {CONTAINER_NAME} +ExecStop={SERVICE_CONTROL_HELPER} stop {release_sha} {image} {config_digest} Restart=on-failure RestartSec=5s TimeoutStartSec=120s @@ -2552,7 +2562,11 @@ SystemCallArchitectures=native WantedBy=multi-user.target """ _require("--publish" not in unit and " -p " not in unit, "service unit publishes a port") - _require("docker.sock" not in unit, "service unit mounts the Docker socket") + _require( + "--volume=/var/run/docker.sock" not in unit and "--mount=type=bind,source=/var/run/docker.sock" not in unit, + "service unit mounts the Docker socket", + ) + _require("docker rm" not in unit and "--force" not in unit, "service unit contains destructive cleanup") _require( "EnvironmentFile" not in unit and "LoadCredential" not in unit, "service unit adds an unreviewed credential source", diff --git a/ops/install_gcp_leoclean_nosend_release.py b/ops/install_gcp_leoclean_nosend_release.py new file mode 100644 index 0000000..0faab6e --- /dev/null +++ b/ops/install_gcp_leoclean_nosend_release.py @@ -0,0 +1,1363 @@ +#!/usr/bin/python3 -I +"""Install, restart, verify, and roll back one leoclean no-send release.""" + +from __future__ import annotations + +import argparse +import contextlib +import fcntl +import hashlib +import json +import os +import re +import shlex +import stat +import subprocess +import sys +import tempfile +import time +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +CONTRACT_ROOT = Path(__file__).resolve().parents[1] +if str(CONTRACT_ROOT) not in sys.path: + sys.path.insert(0, str(CONTRACT_ROOT)) + +from ops import gcp_leoclean_nosend_package as package # noqa: E402 + +INSTALL_RECEIPT_SCHEMA = "livingip.leocleanNoSendInstallReceipt.v1" +SERVICE = package.SERVICE +SYSTEMCTL = "/usr/bin/systemctl" +MAX_COMMAND_OUTPUT_BYTES = 1024 * 1024 +MAX_RELEASE_BYTES = 8 * 1024 * 1024 +MAX_SOURCE_BYTES = 8 * 1024 * 1024 +HEX_64 = re.compile(r"^[0-9a-f]{64}$") +HEX_40 = re.compile(r"^[0-9a-f]{40}$") +HEX_32 = re.compile(r"^[0-9a-f]{32}$") +SOURCE_PATHS = ( + "ops/gcp_leoclean_nosend_package.py", + "ops/install_gcp_leoclean_nosend_release.py", + "ops/leoclean_nosend_service_control.py", +) +SERVICE_EXEC_PROPERTIES = ("ExecStartPre", "ExecStart", "ExecStop") +SERVICE_SECURITY_PROPERTIES = ( + "RuntimeDirectory", + "RuntimeDirectoryMode", + "RuntimeDirectoryPreserve", + "UMask", + "NoNewPrivileges", + "CapabilityBoundingSet", + "AmbientCapabilities", + "PrivateDevices", + "PrivateTmp", + "ProtectSystem", + "ProtectHome", + "ProtectKernelTunables", + "ProtectKernelModules", + "ProtectKernelLogs", + "ProtectControlGroups", + "LockPersonality", + "RestrictSUIDSGID", + "RestrictRealtime", + "RestrictAddressFamilies", + "SystemCallArchitectures", +) +ABSENT_SERVICE_IDENTITY_PROPERTIES = ( + "FragmentPath", + "DropInPaths", + "EnvironmentFiles", + "Environment", + "ControlGroup", + *SERVICE_EXEC_PROPERTIES, + "RuntimeDirectory", +) +SERVICE_PROPERTIES = ( + "LoadState", + "ActiveState", + "SubState", + "MainPID", + "NRestarts", + "InvocationID", + "FragmentPath", + "DropInPaths", + "EnvironmentFiles", + "Environment", + "ControlGroup", + *SERVICE_EXEC_PROPERTIES, + *SERVICE_SECURITY_PROPERTIES, +) +FORBIDDEN_PROCESS_PREFIXES = ( + "DOCKER_", + "BUILDKIT_", + "BUILDX_", + "PG", + "CLOUDSDK_", + "GOOGLE_", + "TELEO_", + "HTTP_PROXY", + "HTTPS_PROXY", + "ALL_PROXY", + "NO_PROXY", + "LD_", + "PYTHON", +) + +Runner = Callable[[list[str]], subprocess.CompletedProcess[bytes]] +Sleeper = Callable[[float], None] + + +class InstallError(RuntimeError): + """A bounded installer failure safe to report without command output.""" + + +class AtomicWriteError(InstallError): + """An atomic publication failure retaining any published inode identity.""" + + def __init__(self, message: str, published_identity: tuple[int, int] | None = None) -> None: + super().__init__(message) + self.published_identity = published_identity + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise InstallError(message) + + +@dataclass(frozen=True) +class InstallPaths: + root: Path + unit: Path + helper: Path + state_dir: Path + docker_config: Path + release: Path + receipt: Path + lock: Path + proc_root: Path + dropin_directories: tuple[Path, ...] + + @classmethod + def under(cls, root: Path, *, proc_root: Path | None = None) -> InstallPaths: + resolved = root.resolve() + + def rooted(absolute: str) -> Path: + return resolved / absolute.removeprefix("/") + + state = rooted("/var/lib/livingip/leoclean-nosend") + return cls( + root=resolved, + unit=rooted(f"/etc/systemd/system/{SERVICE}"), + helper=rooted(package.SERVICE_CONTROL_HELPER), + state_dir=state, + docker_config=state / "docker-config", + release=state / "release.json", + receipt=state / "install-receipt.json", + lock=rooted("/run/leoclean-gcp-nosend-installer.lock"), + proc_root=proc_root or rooted("/proc"), + dropin_directories=tuple( + rooted(path) + for path in ( + f"/etc/systemd/system/{SERVICE}.d", + f"/run/systemd/system/{SERVICE}.d", + f"/usr/lib/systemd/system/{SERVICE}.d", + f"/lib/systemd/system/{SERVICE}.d", + ) + ), + ) + + +@dataclass(frozen=True) +class FileSnapshot: + existed: bool + content: bytes = b"" + mode: int = 0 + + +@dataclass(frozen=True) +class PreparedInstall: + release: dict[str, Any] + unit: bytes + helper: bytes + source_binding: dict[str, Any] + + +@dataclass(frozen=True) +class ServicePosture: + kind: str + state: dict[str, Any] + + +def _sha256_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _canonical_json(value: Mapping[str, Any]) -> bytes: + try: + return (json.dumps(value, allow_nan=False, indent=2, sort_keys=True) + "\n").encode("utf-8") + except (TypeError, ValueError) as exc: + raise InstallError("installer receipt could not be encoded") from exc + + +def _default_runner(argv: list[str]) -> subprocess.CompletedProcess[bytes]: + return subprocess.run( + argv, + check=False, + stdin=subprocess.DEVNULL, + capture_output=True, + timeout=120, + env={"PATH": package.SERVICE_PATH}, + ) + + +def _run(argv: list[str], runner: Runner, *, label: str) -> subprocess.CompletedProcess[bytes]: + try: + completed = runner(argv) + except (OSError, subprocess.SubprocessError) as exc: + raise InstallError(f"{label} failed") from exc + _require( + len(completed.stdout) <= MAX_COMMAND_OUTPUT_BYTES and len(completed.stderr) <= MAX_COMMAND_OUTPUT_BYTES, + f"{label} output exceeded the safety limit", + ) + return completed + + +def _run_checked(argv: list[str], runner: Runner, *, label: str) -> subprocess.CompletedProcess[bytes]: + completed = _run(argv, runner, label=label) + _require(completed.returncode == 0, f"{label} failed") + return completed + + +def _strict_json_bytes(raw: bytes, label: str) -> Any: + _require(len(raw) <= MAX_COMMAND_OUTPUT_BYTES, f"{label} output exceeded the safety limit") + + def reject_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + value: dict[str, Any] = {} + for key, item in pairs: + if key in value: + raise ValueError("duplicate key") + value[key] = item + return value + + try: + return json.loads( + raw.decode("utf-8", "strict"), + object_pairs_hook=reject_duplicates, + parse_constant=lambda _value: (_ for _ in ()).throw(ValueError("constant")), + ) + except (UnicodeError, ValueError, json.JSONDecodeError) as exc: + raise InstallError(f"{label} output was invalid") from exc + + +def _read_stable_file( + path: Path, + *, + label: str, + max_bytes: int, + expected_mode: int | None = None, +) -> bytes: + flags = os.O_RDONLY + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + descriptor = os.open(path, flags) + except OSError as exc: + raise InstallError(f"{label} was unsafe") from exc + try: + before = os.fstat(descriptor) + _require( + stat.S_ISREG(before.st_mode) + and before.st_nlink == 1 + and before.st_size <= max_bytes + and (expected_mode is None or stat.S_IMODE(before.st_mode) == expected_mode), + f"{label} was unsafe", + ) + chunks: list[bytes] = [] + remaining = max_bytes + 1 + while remaining: + chunk = os.read(descriptor, min(65536, remaining)) + if not chunk: + break + chunks.append(chunk) + remaining -= len(chunk) + content = b"".join(chunks) + after = os.fstat(descriptor) + named = path.stat(follow_symlinks=False) + _require( + len(content) == after.st_size + and ( + before.st_dev, + before.st_ino, + before.st_mode, + before.st_nlink, + before.st_size, + before.st_mtime_ns, + before.st_ctime_ns, + ) + == ( + after.st_dev, + after.st_ino, + after.st_mode, + after.st_nlink, + after.st_size, + after.st_mtime_ns, + after.st_ctime_ns, + ) + and (after.st_dev, after.st_ino, after.st_mode, after.st_nlink) + == (named.st_dev, named.st_ino, named.st_mode, named.st_nlink), + f"{label} changed while it was read", + ) + return content + except OSError as exc: + raise InstallError(f"{label} could not be read") from exc + finally: + os.close(descriptor) + + +def _read_stable_source(path: Path) -> bytes: + return _read_stable_file(path, label="installer source file", max_bytes=MAX_SOURCE_BYTES) + + +def reviewed_source_binding(repo_root: Path = CONTRACT_ROOT) -> dict[str, Any]: + root = repo_root.resolve() + _require(root == CONTRACT_ROOT, "installer source repository was not the contract root") + revision = package._git_commit(root) + expected = {relative: package.sha256_file(root / relative) for relative in SOURCE_PATHS} + package._verify_git_source_closure(root, revision, expected) + stable = { + "revision": revision, + "files": [{"path": relative, "sha256": expected[relative]} for relative in sorted(expected)], + } + return {**stable, "sha256": package.canonical_sha256(stable)} + + +def _validate_source_binding(value: Mapping[str, Any]) -> None: + _require(set(value) == {"revision", "files", "sha256"}, "installer source binding was invalid") + revision = value.get("revision") + files = value.get("files") + _require( + isinstance(revision, str) and HEX_40.fullmatch(revision) is not None, "installer source binding was invalid" + ) + _require(isinstance(files, list) and len(files) == len(SOURCE_PATHS), "installer source binding was invalid") + observed_paths: list[str] = [] + for item in files: + _require(isinstance(item, dict) and set(item) == {"path", "sha256"}, "installer source binding was invalid") + path = item.get("path") + digest = item.get("sha256") + _require( + isinstance(path, str) and isinstance(digest, str) and HEX_64.fullmatch(digest) is not None, + "installer source binding was invalid", + ) + observed_paths.append(path) + _require(observed_paths == sorted(SOURCE_PATHS), "installer source binding was invalid") + stable = {"revision": revision, "files": files} + _require(value.get("sha256") == package.canonical_sha256(stable), "installer source binding was invalid") + + +def _bound_source_files(source_binding: Mapping[str, Any]) -> dict[str, bytes]: + _validate_source_binding(source_binding) + _require( + Path(__file__).resolve() == CONTRACT_ROOT / "ops/install_gcp_leoclean_nosend_release.py" + and Path(package.__file__).resolve() == CONTRACT_ROOT / "ops/gcp_leoclean_nosend_package.py", + "executing installer source was outside the contract root", + ) + expected = {str(item["path"]): str(item["sha256"]) for item in source_binding["files"]} + sources: dict[str, bytes] = {} + for relative in sorted(SOURCE_PATHS): + path = (CONTRACT_ROOT / relative).resolve() + _require(path == CONTRACT_ROOT / relative, "installer source path escaped the contract root") + content = _read_stable_source(path) + _require(_sha256_bytes(content) == expected[relative], "executing installer source differed from its binding") + sources[relative] = content + return sources + + +def validate_release_bundle(bundle: Path) -> tuple[dict[str, Any], bytes]: + _require(not bundle.is_symlink(), "release bundle was missing or unsafe") + root = bundle.resolve() + _require(root.is_dir() and not root.is_symlink(), "release bundle was missing or unsafe") + _require(stat.S_IMODE(root.stat().st_mode) & 0o022 == 0, "release bundle was writable by group or other") + entries = sorted(path.name for path in root.iterdir()) + _require(entries == [SERVICE, "release.json"], "release bundle entry set was not exact") + release_path = root / "release.json" + unit_path = root / SERVICE + release_bytes = _read_stable_file( + release_path, + label="release descriptor", + max_bytes=MAX_RELEASE_BYTES, + expected_mode=0o644, + ) + unit = _read_stable_file( + unit_path, + label="release unit", + max_bytes=MAX_RELEASE_BYTES, + expected_mode=0o644, + ) + try: + encoded_release = release_bytes.decode("utf-8", "strict") + release = package._strict_json_loads(encoded_release, "release descriptor") + _require(isinstance(release, dict), "release descriptor was not a JSON object") + package.validate_release_descriptor(release) + except (UnicodeError, package.PackageError) as exc: + raise InstallError("release bundle validation failed") from exc + _require(unit == package.render_systemd_unit(release).encode("utf-8"), "release unit differed from its descriptor") + return release, unit + + +def _prepare_install(bundle: Path, source_binding: Mapping[str, Any]) -> PreparedInstall: + sources = _bound_source_files(source_binding) + release, unit = validate_release_bundle(bundle) + helper = sources["ops/leoclean_nosend_service_control.py"] + _require(helper.startswith(b"#!/usr/bin/python3 -I\n"), "service-control helper was invalid") + return PreparedInstall(release, unit, helper, dict(source_binding)) + + +def _build_install_plan(prepared: PreparedInstall, paths: InstallPaths) -> dict[str, Any]: + release = prepared.release + return { + "schema": "livingip.leocleanNoSendInstallPlan.v1", + "status": "pass", + "mode": "dry_run", + "installer_source": prepared.source_binding, + "release": { + "schema": release["schema"], + "release_sha256": release["release_sha256"], + "image_reference": release["image"]["reference"], + "image_config_digest": release["image"]["config_digest"], + "image_input_sha256": release["image"]["input_sha256"], + "unit_sha256": _sha256_bytes(prepared.unit), + "helper_sha256": _sha256_bytes(prepared.helper), + }, + "target": { + "service": SERVICE, + "unit": str(paths.unit), + "helper": str(paths.helper), + "state_directory": str(paths.state_dir), + "docker_host": package.DOCKER_HOST, + "docker_config": str(paths.docker_config), + }, + "authority": { + "image_pull": False, + "database_or_secret_provisioning": False, + "gcp_control_plane": False, + "production_or_transport": False, + "restart_requires_execute_flag": True, + }, + } + + +def build_install_plan( + bundle: Path, + *, + paths: InstallPaths, + source_binding: Mapping[str, Any], +) -> dict[str, Any]: + return _build_install_plan(_prepare_install(bundle, source_binding), paths) + + +def _snapshot(path: Path) -> FileSnapshot: + try: + observed = path.stat(follow_symlinks=False) + except FileNotFoundError: + return FileSnapshot(False) + except OSError as exc: + raise InstallError("installed file could not be identified") from exc + _require(stat.S_ISREG(observed.st_mode), "installed path was not a regular file") + try: + content = path.read_bytes() + except OSError as exc: + raise InstallError("installed file could not be read") from exc + return FileSnapshot(True, content, stat.S_IMODE(observed.st_mode)) + + +def _ensure_directory(path: Path, mode: int, created: list[Path]) -> None: + missing: list[Path] = [] + cursor = path + while not cursor.exists(): + _require(not cursor.is_symlink(), "install directory path was unsafe") + missing.append(cursor) + cursor = cursor.parent + observed = cursor.stat(follow_symlinks=False) + _require(stat.S_ISDIR(observed.st_mode), "install directory parent was unsafe") + for directory in reversed(missing): + directory.mkdir(mode=mode if directory == path else 0o755) + created.append(directory) + observed = path.stat(follow_symlinks=False) + _require( + stat.S_ISDIR(observed.st_mode) and observed.st_uid == os.geteuid() and observed.st_mode & 0o022 == 0, + "install directory was unsafe", + ) + if path in created: + path.chmod(mode) + + +def _fsync_directory(path: Path) -> None: + flags = os.O_RDONLY + if hasattr(os, "O_DIRECTORY"): + flags |= os.O_DIRECTORY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + descriptor = os.open(path, flags) + try: + _require(stat.S_ISDIR(os.fstat(descriptor).st_mode), "fsync directory was invalid") + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def _atomic_write(path: Path, content: bytes, mode: int) -> tuple[int, int]: + _require(path.parent.is_dir() and not path.parent.is_symlink(), "install destination parent was unsafe") + if path.exists() or path.is_symlink(): + _require(path.is_file() and not path.is_symlink(), "install destination was unsafe") + descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) + temporary = Path(temporary_name) + published_identity: tuple[int, int] | None = None + try: + os.fchmod(descriptor, mode) + view = memoryview(content) + while view: + written = os.write(descriptor, view) + _require(written > 0, "install file write made no progress") + view = view[written:] + os.fsync(descriptor) + observed = os.fstat(descriptor) + _require(stat.S_ISREG(observed.st_mode) and observed.st_size == len(content), "staged install file was invalid") + staged_identity = (observed.st_dev, observed.st_ino) + os.replace(temporary, path) + published_identity = staged_identity + _fsync_directory(path.parent) + installed = path.stat(follow_symlinks=False) + os.lseek(descriptor, 0, os.SEEK_SET) + installed_content = b"" + while len(installed_content) < len(content) + 1: + chunk = os.read(descriptor, min(65536, len(content) + 1 - len(installed_content))) + if not chunk: + break + installed_content += chunk + after = os.fstat(descriptor) + named_after_readback = path.stat(follow_symlinks=False) + _require( + stat.S_ISREG(installed.st_mode) + and (installed.st_dev, installed.st_ino) == staged_identity + and (after.st_dev, after.st_ino) == staged_identity + and (named_after_readback.st_dev, named_after_readback.st_ino) == staged_identity + and stat.S_IMODE(installed.st_mode) == mode + and stat.S_IMODE(named_after_readback.st_mode) == mode + and installed_content == content, + "installed file verification failed", + ) + return staged_identity + except (OSError, InstallError) as exc: + raise AtomicWriteError("atomic install file publication failed", published_identity) from exc + finally: + if descriptor >= 0: + os.close(descriptor) + if temporary.exists() or temporary.is_symlink(): + temporary.unlink() + + +def _publish_managed_file( + path: Path, + content: bytes, + mode: int, + installed_identities: dict[Path, tuple[int, int]], +) -> None: + try: + installed_identities[path] = _atomic_write(path, content, mode) + except AtomicWriteError as exc: + if exc.published_identity is not None: + installed_identities[path] = exc.published_identity + raise + + +def _restore_snapshot(path: Path, snapshot: FileSnapshot, installed_identity: tuple[int, int] | None) -> None: + if installed_identity is not None: + observed = path.stat(follow_symlinks=False) + _require( + stat.S_ISREG(observed.st_mode) and (observed.st_dev, observed.st_ino) == installed_identity, + "rollback refused a replaced installed file", + ) + if snapshot.existed: + _atomic_write(path, snapshot.content, snapshot.mode) + return + if path.exists() or path.is_symlink(): + path.unlink() + _fsync_directory(path.parent) + + +def _dropins_absent(paths: InstallPaths) -> None: + for directory in paths.dropin_directories: + if not directory.exists() and not directory.is_symlink(): + continue + _require(directory.is_dir() and not directory.is_symlink(), "service drop-in path was unsafe") + _require(not any(directory.iterdir()), "service has an unreviewed drop-in") + + +def _parse_properties(raw: bytes) -> dict[str, str]: + try: + lines = raw.decode("utf-8", "strict").splitlines() + except UnicodeError as exc: + raise InstallError("systemd property output was invalid") from exc + value: dict[str, str] = {} + for line in lines: + key, separator, item = line.partition("=") + _require( + separator == "=" and key in SERVICE_PROPERTIES and key not in value, "systemd property output was invalid" + ) + value[key] = item + _require(set(value) == set(SERVICE_PROPERTIES), "systemd property output was incomplete") + return value + + +def _service_state(runner: Runner) -> dict[str, Any]: + argv = [SYSTEMCTL, "show", SERVICE, "--no-pager"] + argv.extend(f"--property={name}" for name in SERVICE_PROPERTIES) + completed = _run_checked(argv, runner, label="systemd service readback") + raw = _parse_properties(completed.stdout) + try: + main_pid = int(raw["MainPID"] or "0") + n_restarts = int(raw["NRestarts"] or "0") + except ValueError as exc: + raise InstallError("systemd numeric property was invalid") from exc + _require(main_pid >= 0 and n_restarts >= 0, "systemd numeric property was invalid") + return {**raw, "MainPID": main_pid, "NRestarts": n_restarts} + + +def _rendered_service_fields(release: Mapping[str, Any]) -> dict[str, list[str]]: + fields: dict[str, list[str]] = {} + in_service = False + for line in package.render_systemd_unit(dict(release)).splitlines(): + if line == "[Service]": + in_service = True + continue + if line.startswith("["): + in_service = False + continue + if not in_service or not line: + continue + key, separator, value = line.partition("=") + _require(separator == "=" and key, "rendered service contract was invalid") + fields.setdefault(key, []).append(value) + return fields + + +def _parse_effective_exec(value: str) -> list[str]: + matched = re.fullmatch( + r"\{ path=([^ ;]+) ; argv\[\]=([^;]+?) ; ignore_errors=(yes|no) ;.*\}", + value, + ) + _require(matched is not None and matched.group(3) == "no", "effective service command was invalid") + try: + argv = shlex.split(matched.group(2)) + except ValueError as exc: + raise InstallError("effective service command was invalid") from exc + _require(argv and argv[0] == matched.group(1), "effective service command was invalid") + return argv + + +def _validate_unit_configuration( + state: Mapping[str, Any], + paths: InstallPaths, + release: Mapping[str, Any], +) -> None: + _require(state.get("LoadState") == "loaded", "service unit was not loaded") + try: + fragment = Path(str(state.get("FragmentPath", ""))).resolve() + except OSError as exc: + raise InstallError("service fragment path was invalid") from exc + _require(fragment == paths.unit.resolve(), "service loaded an unexpected fragment") + expected_unit = package.render_systemd_unit(dict(release)).encode("utf-8") + installed_unit = _read_stable_file( + paths.unit, + label="installed service unit", + max_bytes=MAX_RELEASE_BYTES, + expected_mode=0o644, + ) + _require(installed_unit == expected_unit, "installed service unit differed from the release") + _require(not str(state.get("DropInPaths", "")).strip(), "service has an effective later drop-in") + _require(not str(state.get("EnvironmentFiles", "")).strip(), "service has an effective environment file") + try: + configured_environment = shlex.split(str(state.get("Environment", ""))) + except ValueError as exc: + raise InstallError("service environment property was invalid") from exc + _require( + configured_environment == [f"PATH={package.SERVICE_PATH}"], + "service configured environment was not exact", + ) + fields = _rendered_service_fields(release) + for property_name in SERVICE_EXEC_PROPERTIES: + expected_values = fields.get(property_name) + _require(expected_values is not None and len(expected_values) == 1, "rendered service command was invalid") + expected_argv = shlex.split(expected_values[0]) + _require( + _parse_effective_exec(str(state.get(property_name, ""))) == expected_argv, + "effective service command differed from the release", + ) + for property_name in SERVICE_SECURITY_PROPERTIES: + expected_values = fields.get(property_name) + _require(expected_values is not None and len(expected_values) == 1, "rendered service security was invalid") + observed_value = str(state.get(property_name, "")) + expected_value = expected_values[0] + if property_name in {"RestrictAddressFamilies", "SystemCallArchitectures"}: + _require( + sorted(shlex.split(observed_value)) == sorted(shlex.split(expected_value)), + "effective service security differed from the release", + ) + else: + _require(observed_value == expected_value, "effective service security differed from the release") + + +def _stable_service_posture( + release: Mapping[str, Any] | None, + *, + runner: Runner, + paths: InstallPaths, + sleeper: Sleeper, +) -> ServicePosture: + first = _service_state(runner) + sleeper(0.05) + second = _service_state(runner) + _require(first == second, "service posture changed between readbacks") + if release is None: + _require( + first["LoadState"] == "not-found" + and first["ActiveState"] == "inactive" + and first["SubState"] == "dead" + and first["MainPID"] == 0 + and first["NRestarts"] == 0 + and first["InvocationID"] == "" + and all(first[property_name] == "" for property_name in ABSENT_SERVICE_IDENTITY_PROPERTIES), + "unmanaged service posture was not exactly absent", + ) + return ServicePosture("absent", first) + + _dropins_absent(paths) + _validate_unit_configuration(first, paths, release) + if ( + first["ActiveState"] == "active" + and first["SubState"] == "running" + and first["MainPID"] > 1 + and HEX_32.fullmatch(str(first["InvocationID"])) is not None + and first["ControlGroup"] == f"/system.slice/{SERVICE}" + ): + return ServicePosture("active", first) + _require( + first["ActiveState"] == "inactive" + and first["SubState"] == "dead" + and first["MainPID"] == 0 + and first["InvocationID"] == "" + and first["ControlGroup"] == "", + "managed service posture was transitional or failed", + ) + return ServicePosture("inactive", first) + + +def _parse_process_environment(raw: bytes) -> dict[str, str]: + _require(raw and raw.endswith(b"\0") and b"\0\0" not in raw, "service process environment was malformed") + environment: dict[str, str] = {} + try: + for record in raw[:-1].split(b"\0"): + key, separator, value = record.partition(b"=") + _require(bool(key) and separator == b"=", "service process environment was malformed") + decoded_key = key.decode("utf-8", "strict") + decoded_value = value.decode("utf-8", "strict") + _require(decoded_key not in environment, "service process environment was malformed") + environment[decoded_key] = decoded_value + except UnicodeError as exc: + raise InstallError("service process environment was malformed") from exc + return environment + + +def _validate_process_environment(pid: int, paths: InstallPaths) -> tuple[str, ...]: + _require(pid > 1, "service main PID was invalid") + try: + environment = _parse_process_environment((paths.proc_root / str(pid) / "environ").read_bytes()) + except OSError as exc: + raise InstallError("service process environment was unreadable") from exc + _require(environment.get("PATH") == package.SERVICE_PATH, "service process PATH was not exact") + forbidden = tuple( + sorted( + key for key in environment if any(key.upper().startswith(prefix) for prefix in FORBIDDEN_PROCESS_PREFIXES) + ) + ) + _require(not forbidden, "service process inherited a forbidden environment field") + return tuple(sorted(environment)) + + +def _validate_supervising_process( + pid: int, + paths: InstallPaths, + release: Mapping[str, Any], + control_group: str, +) -> dict[str, Any]: + _require(pid > 1, "service main PID was invalid") + process_root = paths.proc_root / str(pid) + fields = _rendered_service_fields(release) + expected_start = shlex.split(fields["ExecStart"][0]) + expected_executable = expected_start[0] + try: + executable = os.readlink(process_root / "exe") + command_line = (process_root / "cmdline").read_bytes() + cgroup = (process_root / "cgroup").read_text(encoding="utf-8") + except (OSError, UnicodeError) as exc: + raise InstallError("service supervising process identity was unreadable") from exc + accepted_executables = {expected_executable, str(Path(expected_executable).resolve())} + _require(executable in accepted_executables, "service supervising executable differed from the release") + _require( + command_line == b"\0".join(item.encode("utf-8") for item in expected_start) + b"\0", + "service supervising command line differed from the release", + ) + _require(control_group == f"/system.slice/{SERVICE}", "service control group was not exact") + cgroup_paths: list[str] = [] + for line in cgroup.splitlines(): + hierarchy, separator, remainder = line.partition(":") + controllers, second_separator, path = remainder.partition(":") + _require( + bool(hierarchy) and separator == ":" and second_separator == ":" and path, + "service cgroup identity was malformed", + ) + del controllers + cgroup_paths.append(path) + _require(cgroup_paths and set(cgroup_paths) == {control_group}, "service cgroup differed from systemd") + return { + "executable": executable, + "cmdline_sha256": _sha256_bytes(command_line), + "cgroup": control_group, + } + + +def _docker_prefix(paths: InstallPaths) -> list[str]: + return [ + package.DOCKER_BINARY, + f"--host={package.DOCKER_HOST}", + f"--config={paths.docker_config}", + ] + + +def _docker_inspect(arguments: list[str], runner: Runner, paths: InstallPaths, label: str) -> dict[str, Any] | None: + completed = _run([*_docker_prefix(paths), *arguments], runner, label=label) + if completed.returncode != 0: + return None + value = _strict_json_bytes(completed.stdout, label) + _require(isinstance(value, list) and len(value) == 1 and isinstance(value[0], dict), f"{label} was not exact") + return value[0] + + +def _validate_local_image(release: dict[str, Any], runner: Runner, paths: InstallPaths) -> None: + reference = release["image"]["reference"] + inspected = _docker_inspect(["image", "inspect", reference], runner, paths, "local image inspection") + _require(inspected is not None, "digest-bound image was not present locally") + try: + normalized = package.build_image_inspection(release["image_input"], reference, inspected) + except package.PackageError as exc: + raise InstallError("local image differed from the finalized release") from exc + _require(normalized == release["inspection"], "local image differed from the finalized inspection") + + +def _validate_running_container(value: dict[str, Any], release: dict[str, Any]) -> dict[str, Any]: + config = value.get("Config") + state = value.get("State") + health = state.get("Health") if isinstance(state, dict) else None + host = value.get("HostConfig") + labels = config.get("Labels") if isinstance(config, dict) else None + runtime = release["inspection"]["runtime_config"] + container_id = value.get("Id") + _require(isinstance(container_id, str) and HEX_64.fullmatch(container_id) is not None, "container ID was invalid") + _require( + value.get("Name") == f"/{package.CONTAINER_NAME}" + and value.get("Image") == release["image"]["config_digest"] + and isinstance(config, dict) + and config.get("Image") == release["image"]["reference"] + and config.get("Env") == runtime["environment"] + and isinstance(labels, dict) + and labels.get("livingip.release.sha256") == release["release_sha256"] + and labels.get("livingip.container.role") == "leoclean-gcp-nosend" + and isinstance(state, dict) + and state.get("Running") is True + and isinstance(state.get("Pid"), int) + and state["Pid"] > 1 + and isinstance(health, dict) + and health.get("Status") == "healthy" + and isinstance(host, dict) + and host.get("ReadonlyRootfs") is True + and host.get("NetworkMode") == "bridge" + and host.get("PidsLimit") == 512 + and sorted(host.get("CapDrop", [])) == ["ALL"] + and sorted(host.get("CapAdd", [])) == ["CHOWN", "SETGID", "SETPCAP", "SETUID"] + and host.get("SecurityOpt") == ["no-new-privileges:true"], + "running container differed from the release contract", + ) + expected_path = runtime["entrypoint"][0] + expected_args = [*runtime["entrypoint"][1:], *runtime["cmd"]] + _require( + value.get("Path") == expected_path and value.get("Args") == expected_args, + "running container command differed from the release contract", + ) + return {"container_id": container_id, "container_pid": state["Pid"], "health": "healthy"} + + +def verify_running_release( + release: dict[str, Any], + *, + runner: Runner, + paths: InstallPaths, + sleeper: Sleeper = time.sleep, + attempts: int = 30, +) -> dict[str, Any]: + first: dict[str, Any] | None = None + for _attempt in range(attempts): + observed = _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"]), + ) + first_container = _docker_inspect( + ["container", "inspect", package.CONTAINER_NAME], + runner, + paths, + "running container inspection", + ) + _require(first_container is not None, "running container was absent") + container = _validate_running_container(first_container, release) + _validate_local_image(release, runner, paths) + second_container = _docker_inspect( + ["container", "inspect", container["container_id"]], + runner, + paths, + "stable container inspection", + ) + _require(second_container is not None, "running container exited during verification") + stable_container = _validate_running_container(second_container, release) + second = _service_state(runner) + _validate_unit_configuration(second, paths, release) + stable_supervising_process = _validate_supervising_process( + second["MainPID"], + paths, + release, + str(second["ControlGroup"]), + ) + _require( + first["MainPID"] == second["MainPID"] + and first["NRestarts"] == second["NRestarts"] + and first["InvocationID"] == second["InvocationID"] + and first["ControlGroup"] == second["ControlGroup"] + and first["ActiveState"] == second["ActiveState"] == "active" + and first["SubState"] == second["SubState"] == "running" + and container == stable_container + and supervising_process == stable_supervising_process, + "service or container changed during verification", + ) + return { + "active_state": "active", + "sub_state": "running", + "main_pid": first["MainPID"], + "n_restarts": first["NRestarts"], + "invocation_id": first["InvocationID"], + "supervising_process": supervising_process, + **container, + "validated_process_environment_fields": list(validated_environment_fields), + "dropins": "absent", + "environment_files": "absent", + } + + +def _read_managed_release(paths: InstallPaths) -> dict[str, Any] | None: + if not paths.release.exists() and not paths.release.is_symlink(): + return None + try: + value = package.load_json(paths.release, "installed release descriptor") + package.validate_release_descriptor(value) + except (OSError, package.PackageError) as exc: + raise InstallError("installed release descriptor was invalid") from exc + return value + + +def _validate_managed_prior_state(paths: InstallPaths, helper_source: bytes) -> dict[str, Any] | None: + unit_exists = paths.unit.exists() or paths.unit.is_symlink() + helper_exists = paths.helper.exists() or paths.helper.is_symlink() + release_exists = paths.release.exists() or paths.release.is_symlink() + receipt_exists = paths.receipt.exists() or paths.receipt.is_symlink() + if not any((unit_exists, helper_exists, release_exists, receipt_exists)): + return None + _require(all((unit_exists, helper_exists, release_exists, receipt_exists)), "managed install state was partial") + for path, mode in ( + (paths.unit, 0o644), + (paths.helper, 0o755), + (paths.release, 0o600), + (paths.receipt, 0o600), + ): + observed = path.stat(follow_symlinks=False) + _require( + stat.S_ISREG(observed.st_mode) + and observed.st_uid == os.geteuid() + and stat.S_IMODE(observed.st_mode) == mode + and observed.st_nlink == 1, + "managed install file posture drifted", + ) + for directory in (paths.state_dir, paths.docker_config): + observed = directory.stat(follow_symlinks=False) + _require( + stat.S_ISDIR(observed.st_mode) + and observed.st_uid == os.geteuid() + and stat.S_IMODE(observed.st_mode) == 0o700, + "managed install directory posture drifted", + ) + _require(not any(paths.docker_config.iterdir()), "service Docker config directory was not empty") + release = _read_managed_release(paths) + _require(release is not None, "managed release state was absent") + _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") + try: + receipt = package.load_json(paths.receipt, "install receipt") + except (OSError, package.PackageError) as exc: + raise InstallError("installed receipt was invalid") from exc + _require( + receipt.get("schema") == INSTALL_RECEIPT_SCHEMA + and receipt.get("status") == "pass" + and receipt.get("release_sha256") == release["release_sha256"], + "installed receipt was invalid", + ) + _require( + isinstance(receipt.get("installed_files"), dict) + and receipt["installed_files"].get("unit_sha256") == _sha256_bytes(paths.unit.read_bytes()) + and receipt["installed_files"].get("helper_sha256") == _sha256_bytes(paths.helper.read_bytes()), + "installed file receipt drifted", + ) + _require(bool(helper_source), "service-control helper source was empty") + return release + + +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") + argv = [SYSTEMCTL, action, SERVICE] + _run_checked(argv, runner, label=f"systemd {action}") + + +def _cleanup_created_directories(created: list[Path]) -> bool: + complete = True + for directory in reversed(created): + try: + directory.rmdir() + except FileNotFoundError: + pass + except OSError: + complete = False + return complete + + +@contextlib.contextmanager +def _install_lock(paths: InstallPaths): + if paths.root == Path("/"): + _require(os.geteuid() == 0, "live installation lock requires root") + created: list[Path] = [] + _ensure_directory(paths.lock.parent, 0o755, created) + flags = os.O_RDWR | os.O_CREAT + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + descriptor = os.open(paths.lock, flags, 0o600) + except OSError as exc: + raise InstallError("installer lock was unsafe") from exc + try: + observed = os.fstat(descriptor) + named = paths.lock.stat(follow_symlinks=False) + _require( + stat.S_ISREG(observed.st_mode) + and (observed.st_dev, observed.st_ino) == (named.st_dev, named.st_ino) + and observed.st_uid == os.geteuid() + and stat.S_IMODE(observed.st_mode) == 0o600 + and observed.st_nlink == 1, + "installer lock posture was unsafe", + ) + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + raise InstallError("another installer transaction holds the lock") from exc + os.fsync(descriptor) + _fsync_directory(paths.lock.parent) + yield + finally: + try: + fcntl.flock(descriptor, fcntl.LOCK_UN) + finally: + os.close(descriptor) + + +def _install_release_transaction( + bundle: Path, + *, + paths: InstallPaths, + source_binding: Mapping[str, Any], + runner: Runner = _default_runner, + sleeper: Sleeper = time.sleep, + execute_restart: bool = False, +) -> dict[str, Any]: + prepared = _prepare_install(bundle, source_binding) + plan = _build_install_plan(prepared, paths) + if not execute_restart: + return plan + + release = prepared.release + unit_content = prepared.unit + helper_content = prepared.helper + _dropins_absent(paths) + prior_release = _validate_managed_prior_state(paths, helper_content) + prior_posture = _stable_service_posture( + prior_release, + runner=runner, + paths=paths, + sleeper=sleeper, + ) + before = prior_posture.state + prior_active = prior_posture.kind == "active" + + created: list[Path] = [] + try: + _ensure_directory(paths.unit.parent, 0o755, created) + _ensure_directory(paths.helper.parent, 0o755, created) + _ensure_directory(paths.state_dir, 0o700, created) + _ensure_directory(paths.docker_config, 0o700, created) + _require(not any(paths.docker_config.iterdir()), "service Docker config directory was not empty") + _validate_local_image(release, runner, paths) + except Exception as exc: + cleanup_complete = _cleanup_created_directories(created) + if not cleanup_complete: + raise InstallError("preflight failed and cleanup was incomplete") from exc + raise + + snapshots = { + paths.unit: _snapshot(paths.unit), + paths.helper: _snapshot(paths.helper), + paths.release: _snapshot(paths.release), + paths.receipt: _snapshot(paths.receipt), + } + installed_identities: dict[Path, tuple[int, int]] = {} + attempted_paths: set[Path] = set() + + exact_already_installed = ( + prior_release is not None + and prior_release == release + and snapshots[paths.unit].content == unit_content + and snapshots[paths.helper].content == helper_content + ) + if exact_already_installed: + verified = verify_running_release(release, runner=runner, paths=paths, sleeper=sleeper) + existing_receipt = package.load_json(paths.receipt, "install receipt") + return {**existing_receipt, "result": "already_installed", "verification": verified} + + if prior_active: + _require(prior_release is not None, "active managed service had no prior release") + prior_verification = verify_running_release(prior_release, runner=runner, paths=paths, sleeper=sleeper) + _require( + prior_verification["main_pid"] == before["MainPID"] + and prior_verification["n_restarts"] == before["NRestarts"] + and prior_verification["invocation_id"] == before["InvocationID"], + "active prior service changed before the trusted stop", + ) + + candidate_may_be_active = False + try: + if prior_active: + # Stop with the currently loaded unit so its release-bound helper + # validates the old container before the unit is replaced. + _systemctl("stop", runner) + attempted_paths.add(paths.helper) + _publish_managed_file(paths.helper, helper_content, 0o755, installed_identities) + attempted_paths.add(paths.unit) + _publish_managed_file(paths.unit, unit_content, 0o644, installed_identities) + _systemctl("daemon-reload", runner) + configured_posture = _stable_service_posture( + release, + runner=runner, + paths=paths, + sleeper=sleeper, + ) + _require(configured_posture.kind == "inactive", "candidate service was active before its explicit restart") + candidate_may_be_active = True + _systemctl("restart", runner) + verified = verify_running_release(release, runner=runner, paths=paths, sleeper=sleeper) + release_bytes = _canonical_json(release) + attempted_paths.add(paths.release) + _publish_managed_file(paths.release, release_bytes, 0o600, installed_identities) + receipt = { + "schema": INSTALL_RECEIPT_SCHEMA, + "status": "pass", + "result": "installed", + "installer_source": dict(source_binding), + "release_sha256": release["release_sha256"], + "image": { + "reference": release["image"]["reference"], + "digest": release["image"]["digest"], + "config_digest": release["image"]["config_digest"], + "input_sha256": release["image"]["input_sha256"], + }, + "installed_files": { + "unit_sha256": _sha256_bytes(unit_content), + "helper_sha256": _sha256_bytes(helper_content), + "release_descriptor_sha256": _sha256_bytes(release_bytes), + }, + "service_before": { + "load_state": before["LoadState"], + "active_state": before["ActiveState"], + "sub_state": before["SubState"], + "main_pid": before["MainPID"], + "n_restarts": before["NRestarts"], + }, + "verification": verified, + "rollback": { + "previous_release_present": prior_release is not None, + "previous_service_active": prior_active, + "previous_service_posture": prior_posture.kind, + "automatic_on_failure": True, + }, + "authority": { + "image_pulled": False, + "cloud_or_database_authority_changed": False, + "transport_authority_changed": False, + }, + } + receipt_bytes = _canonical_json(receipt) + attempted_paths.add(paths.receipt) + _publish_managed_file(paths.receipt, receipt_bytes, 0o600, installed_identities) + return receipt + except Exception as exc: + rollback_errors: list[str] = [] + if candidate_may_be_active: + try: + _systemctl("stop", runner) + except InstallError: + rollback_errors.append("candidate_stop") + for path in (paths.receipt, paths.release, paths.unit, paths.helper): + if path not in attempted_paths: + continue + try: + _restore_snapshot(path, snapshots[path], installed_identities.get(path)) + except (OSError, InstallError): + rollback_errors.append("file_restore") + try: + _systemctl("daemon-reload", runner) + except InstallError: + rollback_errors.append("daemon_reload") + if prior_active: + try: + _systemctl("restart", runner) + _require(prior_release is not None, "prior release was missing during rollback") + restored = verify_running_release(prior_release, runner=runner, paths=paths, sleeper=sleeper) + _require( + restored["n_restarts"] == before["NRestarts"], + "rollback did not restore the prior active posture", + ) + except InstallError: + rollback_errors.append("prior_restart") + else: + try: + if prior_posture.kind == "absent": + _systemctl("reset-failed", runner) + restored_posture = _stable_service_posture( + prior_release, + runner=runner, + paths=paths, + sleeper=sleeper, + ) + _require( + restored_posture.kind == prior_posture.kind + and restored_posture.state["NRestarts"] == before["NRestarts"], + "rollback did not restore the prior service posture", + ) + except InstallError: + rollback_errors.append("inactive_cleanup") + if not _cleanup_created_directories(created): + rollback_errors.append("directory_cleanup") + if rollback_errors: + raise InstallError("installation failed and automatic rollback was incomplete") from exc + if isinstance(exc, InstallError): + raise + raise InstallError("installation failed and was rolled back") from exc + + +def install_release( + bundle: Path, + *, + paths: InstallPaths, + source_binding: Mapping[str, Any], + runner: Runner = _default_runner, + sleeper: Sleeper = time.sleep, + execute_restart: bool = False, +) -> dict[str, Any]: + if not execute_restart: + return _install_release_transaction( + bundle, + paths=paths, + source_binding=source_binding, + runner=runner, + sleeper=sleeper, + execute_restart=False, + ) + with _install_lock(paths): + return _install_release_transaction( + bundle, + paths=paths, + source_binding=source_binding, + runner=runner, + sleeper=sleeper, + execute_restart=True, + ) + + +class _Parser(argparse.ArgumentParser): + def error(self, _message: str) -> None: + raise InstallError("invalid arguments") + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = _Parser(description=__doc__, add_help=False) + parser.add_argument("--bundle", required=True, type=Path) + parser.add_argument("--execute-restart", action="store_true") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + try: + args = parse_args(argv) + root = Path("/") + if args.execute_restart: + _require(os.geteuid() == 0, "live installation requires root") + result = install_release( + args.bundle.absolute(), + paths=InstallPaths.under(root), + source_binding=reviewed_source_binding(), + execute_restart=args.execute_restart, + ) + sys.stdout.write(json.dumps(result, allow_nan=False, separators=(",", ":"), sort_keys=True) + "\n") + return 0 + except (InstallError, package.PackageError) as exc: + sys.stderr.write( + json.dumps({"error": str(exc), "status": "fail"}, separators=(",", ":"), sort_keys=True) + "\n" + ) + return 65 + except Exception: + sys.stderr.write('{"error":"installer failed","status":"fail"}\n') + return 65 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ops/leoclean_nosend_service_control.py b/ops/leoclean_nosend_service_control.py new file mode 100644 index 0000000..f14b9a2 --- /dev/null +++ b/ops/leoclean_nosend_service_control.py @@ -0,0 +1,319 @@ +#!/usr/bin/python3 -I +"""Control one digest-bound leoclean no-send container by verified full ID.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import stat +import subprocess +import sys +import time +from collections.abc import Callable +from pathlib import Path +from typing import Any + +DOCKER_BINARY = "/usr/bin/docker" +DOCKER_HOST = "unix:///var/run/docker.sock" +DOCKER_CONFIG = "/var/lib/livingip/leoclean-nosend/docker-config" +CONTAINER_NAME = "livingip-leoclean-nosend" +CIDFILE = Path("/run/leoclean-gcp-nosend/container.cid") +IMAGE_REPOSITORY = "europe-west6-docker.pkg.dev/teleo-501523/teleo/leoclean-nosend-staging" +SAFE_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" +MAX_DOCKER_OUTPUT_BYTES = 1024 * 1024 +HEX_64 = re.compile(r"^[0-9a-f]{64}$") +IMAGE_REFERENCE = re.compile(rf"^{re.escape(IMAGE_REPOSITORY)}@sha256:[0-9a-f]{{64}}$") +IMAGE_CONFIG_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") + +Runner = Callable[[list[str]], subprocess.CompletedProcess[bytes]] + + +class ControlError(RuntimeError): + """A bounded service-control failure.""" + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise ControlError(message) + + +def _strict_json(raw: bytes) -> Any: + _require(len(raw) <= MAX_DOCKER_OUTPUT_BYTES, "docker output exceeded the safety limit") + + def reject_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + value: dict[str, Any] = {} + for key, item in pairs: + if key in value: + raise ValueError("duplicate key") + value[key] = item + return value + + try: + return json.loads( + raw.decode("utf-8", "strict"), + object_pairs_hook=reject_duplicates, + parse_constant=lambda _value: (_ for _ in ()).throw(ValueError("constant")), + ) + except (UnicodeError, ValueError, json.JSONDecodeError) as exc: + raise ControlError("docker inspection was invalid") from exc + + +def _default_runner(argv: list[str]) -> subprocess.CompletedProcess[bytes]: + environment = {"PATH": SAFE_PATH} + return subprocess.run( + argv, + check=False, + stdin=subprocess.DEVNULL, + capture_output=True, + timeout=45, + env=environment, + ) + + +def _docker(arguments: list[str], runner: Runner) -> subprocess.CompletedProcess[bytes]: + argv = [ + DOCKER_BINARY, + f"--host={DOCKER_HOST}", + f"--config={DOCKER_CONFIG}", + *arguments, + ] + try: + completed = runner(argv) + except (OSError, subprocess.SubprocessError) as exc: + raise ControlError("docker command failed") from exc + _require( + len(completed.stdout) <= MAX_DOCKER_OUTPUT_BYTES and len(completed.stderr) <= MAX_DOCKER_OUTPUT_BYTES, + "docker output exceeded the safety limit", + ) + return completed + + +def _inspect(reference: str, runner: Runner) -> dict[str, Any] | None: + completed = _docker(["container", "inspect", reference], runner) + if completed.returncode != 0: + try: + error = completed.stderr.decode("utf-8", "strict").strip() + except UnicodeError as exc: + raise ControlError("container absence could not be proven") from exc + absence = re.compile( + rf"^(?:Error response from daemon: |Error: )?No such (?:container|object): {re.escape(reference)}$" + ) + _require(absence.fullmatch(error) is not None, "container absence could not be proven") + return None + value = _strict_json(completed.stdout) + _require( + isinstance(value, list) and len(value) == 1 and isinstance(value[0], dict), + "container inspection was not exact", + ) + return value[0] + + +def _validate_container( + value: dict[str, Any], + *, + container_id: str, + release_sha256: str, + image_reference: str, + image_config_digest: str, +) -> bool: + config = value.get("Config") + labels = config.get("Labels") if isinstance(config, dict) else None + state = value.get("State") + _require( + value.get("Id") == container_id + and value.get("Name") == f"/{CONTAINER_NAME}" + and value.get("Image") == image_config_digest + and isinstance(config, dict) + and config.get("Image") == image_reference + and isinstance(labels, dict) + and labels.get("livingip.release.sha256") == release_sha256 + and labels.get("livingip.container.role") == "leoclean-gcp-nosend" + and isinstance(state, dict) + and isinstance(state.get("Running"), bool), + "container identity did not match the installed release", + ) + return bool(state["Running"]) + + +def _cidfile_parent_descriptor(cidfile: Path) -> int: + flags = os.O_RDONLY + if hasattr(os, "O_DIRECTORY"): + flags |= os.O_DIRECTORY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + descriptor = os.open(cidfile.parent, flags) + observed = os.fstat(descriptor) + named = os.stat(cidfile.parent, follow_symlinks=False) + except OSError as exc: + raise ControlError("container ID directory was unsafe") from exc + if not ( + stat.S_ISDIR(observed.st_mode) + and (observed.st_dev, observed.st_ino) == (named.st_dev, named.st_ino) + and observed.st_uid == os.geteuid() + and observed.st_mode & 0o077 == 0 + ): + os.close(descriptor) + raise ControlError("container ID directory was unsafe") + return descriptor + + +def _read_cidfile(cidfile: Path) -> tuple[str, tuple[int, int]] | None: + parent_descriptor = _cidfile_parent_descriptor(cidfile) + descriptor: int | None = None + try: + flags = os.O_RDONLY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + descriptor = os.open(cidfile.name, flags, dir_fd=parent_descriptor) + except FileNotFoundError: + return None + observed = os.fstat(descriptor) + named = os.stat(cidfile.name, dir_fd=parent_descriptor, follow_symlinks=False) + _require( + stat.S_ISREG(observed.st_mode) + and (observed.st_dev, observed.st_ino) == (named.st_dev, named.st_ino) + and observed.st_uid == os.geteuid() + and stat.S_IMODE(observed.st_mode) == 0o600 + and observed.st_nlink == 1 + and observed.st_size <= 65, + "container ID file was unsafe", + ) + raw = os.read(descriptor, 66) + try: + container_id = raw.decode("ascii", "strict").strip() + except UnicodeError as exc: + raise ControlError("container ID file was invalid") from exc + _require(HEX_64.fullmatch(container_id) is not None, "container ID file was invalid") + return container_id, (observed.st_dev, observed.st_ino) + except OSError as exc: + raise ControlError("container ID file could not be read") from exc + finally: + if descriptor is not None: + os.close(descriptor) + os.close(parent_descriptor) + + +def _unlink_cidfile(cidfile: Path, identity: tuple[int, int]) -> None: + parent_descriptor = _cidfile_parent_descriptor(cidfile) + try: + observed = os.stat(cidfile.name, dir_fd=parent_descriptor, follow_symlinks=False) + _require( + stat.S_ISREG(observed.st_mode) and (observed.st_dev, observed.st_ino) == identity, + "container ID file changed during cleanup", + ) + os.unlink(cidfile.name, dir_fd=parent_descriptor) + os.fsync(parent_descriptor) + except OSError as exc: + raise ControlError("container ID file cleanup failed") from exc + finally: + os.close(parent_descriptor) + + +def _wait_absent(container_id: str, runner: Runner, sleeper: Callable[[float], None]) -> None: + for _attempt in range(60): + if _inspect(container_id, runner) is None: + return + sleeper(0.25) + raise ControlError("container did not stop within the bounded wait") + + +def _require_name_unbound(runner: Runner) -> None: + _require(_inspect(CONTAINER_NAME, runner) is None, "container name was rebound after the trusted container exited") + + +def _stop_exact(container_id: str, runner: Runner, sleeper: Callable[[float], None]) -> None: + completed = _docker(["container", "stop", "--time=30", container_id], runner) + _require(completed.returncode == 0, "container stop failed") + _wait_absent(container_id, runner, sleeper) + + +def control_container( + action: str, + release_sha256: str, + image_reference: str, + image_config_digest: str, + *, + runner: Runner = _default_runner, + sleeper: Callable[[float], None] = time.sleep, + cidfile: Path = CIDFILE, +) -> dict[str, object]: + _require(action in {"prestart", "stop"}, "service-control action was invalid") + _require(HEX_64.fullmatch(release_sha256) is not None, "release hash was invalid") + _require(IMAGE_REFERENCE.fullmatch(image_reference) is not None, "image reference was invalid") + _require(IMAGE_CONFIG_DIGEST.fullmatch(image_config_digest) is not None, "image config digest was invalid") + + cid = _read_cidfile(cidfile) + by_name = _inspect(CONTAINER_NAME, runner) + if cid is None: + _require(by_name is None, "container name is occupied without a trusted ID file") + return {"action": action, "status": "pass", "container": "absent"} + + container_id, cid_identity = cid + by_id = _inspect(container_id, runner) + if by_id is None: + _require(by_name is None, "container name was rebound after the trusted container exited") + _require_name_unbound(runner) + _unlink_cidfile(cidfile, cid_identity) + return {"action": action, "status": "pass", "container": "absent"} + + running = _validate_container( + by_id, + container_id=container_id, + release_sha256=release_sha256, + image_reference=image_reference, + image_config_digest=image_config_digest, + ) + _require( + by_name is not None and by_name.get("Id") == container_id, + "container name and trusted ID did not agree", + ) + if running: + _stop_exact(container_id, runner, sleeper) + else: + _wait_absent(container_id, runner, sleeper) + _require_name_unbound(runner) + _unlink_cidfile(cidfile, cid_identity) + return {"action": action, "status": "pass", "container": "stopped"} + + +class _Parser(argparse.ArgumentParser): + def error(self, _message: str) -> None: + raise ControlError("invalid arguments") + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = _Parser(description=__doc__, add_help=False) + parser.add_argument("action", choices=("prestart", "stop")) + parser.add_argument("release_sha256") + parser.add_argument("image_reference") + parser.add_argument("image_config_digest") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + try: + args = parse_args(argv) + result = control_container( + args.action, + args.release_sha256, + args.image_reference, + args.image_config_digest, + ) + sys.stdout.write(json.dumps(result, separators=(",", ":"), sort_keys=True) + "\n") + return 0 + except ControlError as exc: + sys.stderr.write( + json.dumps({"error": str(exc), "status": "fail"}, separators=(",", ":"), sort_keys=True) + "\n" + ) + return 65 + except Exception: + sys.stderr.write('{"error":"service control failed","status":"fail"}\n') + return 65 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_gcp_leoclean_nosend_installer.py b/tests/test_gcp_leoclean_nosend_installer.py new file mode 100644 index 0000000..28fa409 --- /dev/null +++ b/tests/test_gcp_leoclean_nosend_installer.py @@ -0,0 +1,1208 @@ +from __future__ import annotations + +import copy +import json +import os +import shlex +import subprocess +from pathlib import Path +from unittest import mock + +import pytest + +from ops import gcp_leoclean_nosend_package as package +from ops import install_gcp_leoclean_nosend_release as installer +from ops import leoclean_nosend_service_control as control +from tests.test_gcp_leoclean_nosend_package import ( + build_push_receipt_fixture, + docker_inspection_fixture, + image_input_fixture, + write_json, +) + +# Representative production-shaped values from systemd's synthesized +# properties for a system service with LoadState=not-found. These are manager +# defaults, not evidence that a unit fragment, command, environment, runtime +# directory, or process exists. +PRODUCTION_SHAPED_NOT_FOUND_SECURITY = { + "RuntimeDirectory": "", + "RuntimeDirectoryMode": "0755", + "RuntimeDirectoryPreserve": "no", + "UMask": "0022", + "NoNewPrivileges": "no", + "CapabilityBoundingSet": ( + "cap_chown cap_dac_override cap_dac_read_search cap_fowner cap_fsetid cap_kill " + "cap_setgid cap_setuid cap_setpcap cap_linux_immutable cap_net_bind_service " + "cap_net_broadcast cap_net_admin cap_net_raw cap_ipc_lock cap_ipc_owner " + "cap_sys_module cap_sys_rawio cap_sys_chroot cap_sys_ptrace cap_sys_pacct " + "cap_sys_admin cap_sys_boot cap_sys_nice cap_sys_resource cap_sys_time " + "cap_sys_tty_config cap_mknod cap_lease cap_audit_write cap_audit_control " + "cap_setfcap cap_mac_override cap_mac_admin cap_syslog cap_wake_alarm " + "cap_block_suspend cap_audit_read cap_perfmon cap_bpf cap_checkpoint_restore" + ), + "AmbientCapabilities": "", + "PrivateDevices": "no", + "PrivateTmp": "no", + "ProtectSystem": "no", + "ProtectHome": "no", + "ProtectKernelTunables": "no", + "ProtectKernelModules": "no", + "ProtectKernelLogs": "no", + "ProtectControlGroups": "no", + "LockPersonality": "no", + "RestrictSUIDSGID": "no", + "RestrictRealtime": "no", + "RestrictAddressFamilies": "", + "SystemCallArchitectures": "", +} + + +def source_binding() -> dict[str, object]: + stable: dict[str, object] = { + "revision": "a" * 40, + "files": [ + {"path": path, "sha256": package.sha256_file(installer.CONTRACT_ROOT / path)} + for path in sorted(installer.SOURCE_PATHS) + ], + } + return {**stable, "sha256": package.canonical_sha256(stable)} + + +def release_bundle(tmp_path: Path, digest_character: str) -> tuple[Path, dict[str, object], dict[str, object]]: + fixture_root = tmp_path / f"input-{digest_character}" + fixture_root.mkdir(mode=0o755) + # Production image creation rejects any dirty reviewed source closure. + # This synthetic fixture tests an uncommitted installer candidate without + # weakening the production check itself. + with mock.patch.object(package, "_verify_git_source_closure"): + image_input = image_input_fixture(fixture_root) + reference = f"{package.IMAGE_REPOSITORY}@sha256:{digest_character * 64}" + config_digest = "sha256:" + chr(ord(digest_character) + 1) * 64 + image_inspection = docker_inspection_fixture( + image_input, + reference, + config_digest=config_digest, + ) + inspection = package.build_image_inspection(image_input, reference, image_inspection) + release = package.build_release_descriptor( + image_input, + inspection, + build_push_receipt_fixture(image_input, reference, config_digest=config_digest), + ) + bundle = tmp_path / f"bundle-{digest_character}" + bundle.mkdir(mode=0o755) + write_json(bundle / "release.json", release, mode=0o644) + (bundle / package.SERVICE).write_text(package.render_systemd_unit(release), encoding="utf-8") + (bundle / package.SERVICE).chmod(0o644) + return bundle, release, image_inspection + + +class FakeHost: + def __init__(self, paths: installer.InstallPaths) -> None: + self.paths = paths + self.images: dict[str, dict[str, object]] = {} + self.releases: dict[bytes, dict[str, object]] = {} + self.active = False + self.main_pid = 0 + self.n_restarts = 0 + self.invocation = "" + self.container: dict[str, object] | None = None + self.current_release: dict[str, object] | None = None + self.loaded_release: dict[str, object] | None = None + self.generation = 0 + self.restart_calls = 0 + self.stop_calls = 0 + self.fail_dropin_release: str | None = None + self.race_release: str | None = None + self.race_kind: str | None = None + self.race_applied = False + self.property_overrides: dict[str, str] = {} + self.not_found_security_defaults: dict[str, str] = {} + self.unstable_readbacks = False + self.show_count = 0 + self.calls: list[list[str]] = [] + + def add_release(self, release: dict[str, object], image_inspection: dict[str, object]) -> None: + image = release["image"] + assert isinstance(image, dict) + self.images[str(image["reference"])] = copy.deepcopy(image_inspection) + self.releases[package.render_systemd_unit(release).encode()] = release + + def _release_from_unit(self) -> dict[str, object]: + value = self.releases.get(self.paths.unit.read_bytes()) + assert value is not None + return value + + def _effective_fields(self) -> dict[str, list[str]]: + assert self.loaded_release is not None + return installer._rendered_service_fields(self.loaded_release) + + @staticmethod + def _effective_exec(value: str) -> str: + argv = shlex.split(value) + return ( + f"{{ path={argv[0]} ; argv[]={value} ; ignore_errors=no ; " + "start_time=[n/a] ; stop_time=[n/a] ; pid=0 ; code=(null) ; status=0/0 }" + ) + + def _write_process_environment(self) -> None: + directory = self.paths.proc_root / str(self.main_pid) + directory.mkdir(parents=True, exist_ok=True) + (directory / "environ").write_bytes(f"PATH={package.SERVICE_PATH}\0INVOCATION_ID={self.invocation}\0".encode()) + fields = self._effective_fields() + command = shlex.split(fields["ExecStart"][0]) + (directory / "exe").symlink_to(command[0]) + (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: + release = self.loaded_release + assert release is not None + image = release["image"] + inspection = release["inspection"] + assert isinstance(image, dict) and isinstance(inspection, dict) + runtime = inspection["runtime_config"] + assert isinstance(runtime, dict) + self.generation += 1 + self.restart_calls += 1 + self.active = True + self.main_pid = 1000 + self.generation + self.invocation = f"{self.generation:032x}" + container_id = f"{self.generation:064x}" + labels = dict(image["labels"]) + labels.update( + { + "livingip.release.sha256": release["release_sha256"], + "livingip.container.role": "leoclean-gcp-nosend", + } + ) + self.container = { + "Id": container_id, + "Name": f"/{package.CONTAINER_NAME}", + "Image": image["config_digest"], + "Path": runtime["entrypoint"][0], + "Args": [*runtime["entrypoint"][1:], *runtime["cmd"]], + "Config": { + "Image": image["reference"], + "Env": runtime["environment"], + "Labels": labels, + }, + "State": {"Running": True, "Pid": 2000 + self.generation, "Health": {"Status": "healthy"}}, + "HostConfig": { + "ReadonlyRootfs": True, + "NetworkMode": "bridge", + "PidsLimit": 512, + "CapDrop": ["ALL"], + "CapAdd": ["CHOWN", "SETGID", "SETPCAP", "SETUID"], + "SecurityOpt": ["no-new-privileges:true"], + }, + } + self.current_release = release + self._write_process_environment() + + def _stop(self) -> None: + self.stop_calls += 1 + self.active = False + self.main_pid = 0 + self.invocation = "" + self.container = None + self.current_release = None + + def _show(self, arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + self.show_count += 1 + loaded = self.loaded_release is not None + release_sha = self.current_release.get("release_sha256") if self.current_release else None + dropin = bool(self.active and release_sha == self.fail_dropin_release) + value: dict[str, str] = { + "LoadState": "loaded" if loaded else "not-found", + "ActiveState": "active" if self.active else "inactive", + "SubState": "running" if self.active else "dead", + "MainPID": str(self.main_pid), + "NRestarts": str(self.n_restarts), + "InvocationID": self.invocation, + "FragmentPath": str(self.paths.unit) if loaded else "", + "DropInPaths": f"{self.paths.unit}.d/override.conf" if dropin else "", + "EnvironmentFiles": "", + "Environment": f"PATH={package.SERVICE_PATH}" if loaded else "", + "ControlGroup": f"/system.slice/{package.SERVICE}" if self.active else "", + } + fields = self._effective_fields() if loaded else {} + for property_name in installer.SERVICE_EXEC_PROPERTIES: + value[property_name] = self._effective_exec(fields[property_name][0]) if loaded else "" + for property_name in installer.SERVICE_SECURITY_PROPERTIES: + value[property_name] = ( + fields[property_name][0] if loaded else self.not_found_security_defaults.get(property_name, "") + ) + value.update(self.property_overrides) + if self.unstable_readbacks: + value["NRestarts"] = str(self.n_restarts + self.show_count % 2) + payload = "".join(f"{name}={value[name]}\n" for name in installer.SERVICE_PROPERTIES).encode() + return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"") + + def __call__(self, arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + self.calls.append(arguments) + if arguments[0] == installer.SYSTEMCTL: + action = arguments[1] + if action == "show": + return self._show(arguments) + if action == "daemon-reload": + self.loaded_release = self._release_from_unit() if self.paths.unit.exists() else None + return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"") + if action == "reset-failed": + self.n_restarts = 0 + return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"") + if action == "restart": + self._start() + return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"") + if action == "stop": + self._stop() + return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"") + raise AssertionError(arguments) + + assert arguments[:3] == [ + package.DOCKER_BINARY, + f"--host={package.DOCKER_HOST}", + f"--config={self.paths.docker_config}", + ] + operation = arguments[3:] + if operation[:2] == ["image", "inspect"]: + image = self.images.get(operation[2]) + if image is None: + return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"not found") + return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([image]).encode(), stderr=b"") + if operation[:2] == ["container", "inspect"]: + if self.container is None or operation[2] not in {package.CONTAINER_NAME, self.container["Id"]}: + return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"not found") + observed = copy.deepcopy(self.container) + release_sha = self.current_release.get("release_sha256") if self.current_release else None + if operation[2] != package.CONTAINER_NAME and release_sha == self.race_release and not self.race_applied: + self.race_applied = True + if self.race_kind == "container_pid": + observed["State"]["Pid"] += 1 + elif self.race_kind == "service_restart": + self.main_pid += 1 + self.n_restarts += 1 + self.invocation = "f" * 32 + self._write_process_environment() + return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([observed]).encode(), stderr=b"") + raise AssertionError(arguments) + + +def install_paths(tmp_path: Path) -> installer.InstallPaths: + root = tmp_path / "root" + root.mkdir(mode=0o700) + return installer.InstallPaths.under(root, proc_root=root / "proc") + + +def test_dry_run_validates_bundle_without_creating_install_paths(tmp_path: Path) -> None: + bundle, release, _image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + + result = installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + execute_restart=False, + ) + + assert result["mode"] == "dry_run" + assert result["release"]["release_sha256"] == release["release_sha256"] + assert not paths.unit.exists() + assert not paths.state_dir.exists() + + +def test_dry_run_rejects_unbound_installer_source_without_creating_paths(tmp_path: Path) -> None: + bundle, _release, _image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + binding = source_binding() + binding["sha256"] = "f" * 64 + + with pytest.raises(installer.InstallError, match="source binding"): + installer.install_release( + bundle, + paths=paths, + source_binding=binding, + execute_restart=False, + ) + + assert not paths.unit.exists() + assert not paths.state_dir.exists() + + +def test_alternate_repo_root_cannot_bind_the_executing_installer(tmp_path: Path) -> None: + with pytest.raises(installer.InstallError, match="not the contract root"): + installer.reviewed_source_binding(tmp_path) + + with pytest.raises(installer.InstallError, match="invalid arguments"): + installer.parse_args(["--bundle", str(tmp_path), "--repo-root", str(tmp_path)]) + + +def test_rehashed_alternate_source_bytes_cannot_bypass_actual_source_binding(tmp_path: Path) -> None: + bundle, _release, _image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + binding = source_binding() + files = copy.deepcopy(binding["files"]) + assert isinstance(files, list) and isinstance(files[0], dict) + files[0]["sha256"] = "f" * 64 + stable = {"revision": binding["revision"], "files": files} + forged = {**stable, "sha256": package.canonical_sha256(stable)} + + with pytest.raises(installer.InstallError, match="executing installer source differed"): + installer.install_release( + bundle, + paths=paths, + source_binding=forged, + execute_restart=False, + ) + + +def test_disposable_root_install_is_digest_bound_idempotent_and_exact(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) + + receipt = installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert receipt["status"] == "pass" + assert receipt["release_sha256"] == release["release_sha256"] + 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 + assert paths.helper.stat().st_mode & 0o777 == 0o755 + assert paths.release.stat().st_mode & 0o777 == 0o600 + assert paths.receipt.stat().st_mode & 0o777 == 0o600 + assert not list(paths.docker_config.iterdir()) + assert not list(paths.unit.parent.glob(f".{paths.unit.name}.*.tmp")) + + repeated = installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert repeated["result"] == "already_installed" + assert host.restart_calls == 1 + + +def test_first_install_accepts_production_shaped_systemd_not_found_defaults(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) + host.not_found_security_defaults = dict(PRODUCTION_SHAPED_NOT_FOUND_SECURITY) + + receipt = installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert receipt["status"] == "pass" + assert receipt["release_sha256"] == release["release_sha256"] + assert host.restart_calls == 1 + + +@pytest.mark.parametrize("property_name", installer.ABSENT_SERVICE_IDENTITY_PROPERTIES) +def test_first_install_rejects_identity_bearing_not_found_property( + tmp_path: Path, + property_name: str, +) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + host.not_found_security_defaults = dict(PRODUCTION_SHAPED_NOT_FOUND_SECURITY) + host.property_overrides[property_name] = "unexpected-authority" + + with pytest.raises(installer.InstallError, match="exactly absent"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert host.restart_calls == 0 + assert not paths.unit.exists() + assert not paths.state_dir.exists() + + +def test_execute_transaction_uses_exclusive_owner_only_lock(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) + + with installer._install_lock(paths): + observed = paths.lock.stat(follow_symlinks=False) + assert observed.st_uid == os.geteuid() + assert observed.st_mode & 0o777 == 0o600 + with pytest.raises(installer.InstallError, match="holds the lock"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert not host.calls + assert not paths.unit.exists() + + +def test_upgrade_stops_verified_prior_release_before_unit_replacement(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, + ) + start = len(host.calls) + receipt = installer.install_release( + second_bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + 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 + + +def test_stale_loaded_unit_is_rejected_before_trusting_old_exec_stop(tmp_path: Path) -> None: + first_bundle, first_release, first_image = release_bundle(tmp_path, "2") + second_bundle, second_release, second_image = release_bundle(tmp_path, "4") + 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.loaded_release = second_release + + with pytest.raises(installer.InstallError, match="effective service command differed"): + installer.install_release( + second_bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert host.active + assert host.current_release == first_release + assert host.stop_calls == 0 + assert host.restart_calls == 1 + + +def test_effective_security_drift_is_rejected_before_old_service_stop(tmp_path: Path) -> None: + first_bundle, first_release, first_image = release_bundle(tmp_path, "2") + second_bundle, second_release, second_image = release_bundle(tmp_path, "4") + 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["NoNewPrivileges"] = "no" + + with pytest.raises(installer.InstallError, match="effective service security differed"): + 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 + + +@pytest.mark.parametrize( + ("field", "error"), + [ + ("exe", "supervising executable differed"), + ("cmdline", "supervising command line differed"), + ("cgroup", "cgroup differed from systemd"), + ], +) +def test_supervising_process_drift_is_rejected_before_old_service_stop( + tmp_path: Path, + field: str, + error: str, +) -> 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) + if field == "exe": + executable = process_root / "exe" + executable.unlink() + executable.symlink_to("/bin/false") + elif field == "cmdline": + (process_root / "cmdline").write_bytes(b"/usr/bin/false\0") + else: + (process_root / "cgroup").write_text("0::/system.slice/other.service\n", encoding="utf-8") + + with pytest.raises(installer.InstallError, match=error): + 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_prior_managed_file_mode_drift_fails_before_service_mutation(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, + ) + paths.receipt.chmod(0o644) + + with pytest.raises(installer.InstallError, match="file posture drifted"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert host.active + assert host.stop_calls == 0 + assert host.restart_calls == 1 + + +def test_transitional_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": "deactivating", "SubState": "stop-sigterm"}) + + 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") + 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.show_count = 0 + host.unstable_readbacks = True + + with pytest.raises(installer.InstallError, match="changed between readbacks"): + 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_wrong_local_image_is_rejected_without_service_restart(tmp_path: Path) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + image["Id"] = "sha256:" + "9" * 64 + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + + with pytest.raises(installer.InstallError, match=r"finalized (release|inspection)"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert host.restart_calls == 0 + assert not paths.unit.exists() + assert not paths.state_dir.exists() + + +def test_preflight_reports_incomplete_directory_cleanup(tmp_path: Path) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + image["Id"] = "sha256:" + "9" * 64 + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + original_call = host.__call__ + + def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + result = original_call(arguments) + if arguments[3:5] == ["image", "inspect"]: + (paths.state_dir / "unexpected").write_text("preserve", encoding="utf-8") + return result + + with pytest.raises(installer.InstallError, match="cleanup was incomplete"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=runner, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert host.restart_calls == 0 + assert (paths.state_dir / "unexpected").read_text(encoding="utf-8") == "preserve" + + +def test_post_rename_race_refuses_to_mutate_replaced_pathname( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + original_fsync = installer._fsync_directory + raced = False + + def replace_after_helper_rename(directory: Path) -> None: + nonlocal raced + if directory == paths.helper.parent and paths.helper.exists() and not raced: + raced = True + paths.helper.unlink() + paths.helper.write_bytes(b"independent replacement\n") + paths.helper.chmod(0o755) + raise OSError("injected post-rename durability failure") + original_fsync(directory) + + monkeypatch.setattr(installer, "_fsync_directory", replace_after_helper_rename) + with pytest.raises(installer.InstallError, match="automatic rollback was incomplete"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert raced + assert paths.helper.read_bytes() == b"independent replacement\n" + assert host.restart_calls == 0 + + +@pytest.mark.parametrize("race_kind", ["container_pid", "service_restart"]) +def test_pid_or_restart_race_fails_and_cleans_first_install(tmp_path: Path, race_kind: str) -> None: + bundle, release, image = release_bundle(tmp_path, "2") + paths = install_paths(tmp_path) + host = FakeHost(paths) + host.add_release(release, image) + host.race_release = str(release["release_sha256"]) + host.race_kind = race_kind + + with pytest.raises(installer.InstallError, match="changed during verification"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert not host.active + assert host.container is None + assert not paths.unit.exists() + assert not paths.helper.exists() + assert not paths.state_dir.exists() + + +def test_restart_command_failure_after_side_effect_stops_candidate_and_cleans_install(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) + failed_restart = False + + def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + nonlocal failed_restart + result = host(arguments) + if arguments[:2] == [installer.SYSTEMCTL, "restart"] and not failed_restart: + failed_restart = True + return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"restart failed") + return result + + with pytest.raises(installer.InstallError, match="systemd restart failed"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=runner, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert failed_restart + assert not host.active + assert host.stop_calls == 1 + assert not paths.unit.exists() + assert not paths.state_dir.exists() + + +def test_later_dropin_failure_restores_previous_release_and_running_service(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, + ) + first_unit = paths.unit.read_bytes() + first_state = paths.release.read_bytes() + first_receipt = paths.receipt.read_bytes() + host.fail_dropin_release = str(second_release["release_sha256"]) + + with pytest.raises(installer.InstallError, match="later drop-in"): + installer.install_release( + second_bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert paths.unit.read_bytes() == first_unit + assert paths.release.read_bytes() == first_state + assert paths.receipt.read_bytes() == first_receipt + assert host.active + assert host.current_release == first_release + assert host.stop_calls == 2 + assert host.restart_calls == 3 + assert not list(paths.state_dir.glob(".*.tmp")) + + +def test_failed_upgrade_restores_exact_inactive_prior_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 + restart_calls = host.restart_calls + prior_n_restarts = host.n_restarts + first_unit = paths.unit.read_bytes() + host.fail_dropin_release = str(second_release["release_sha256"]) + + with pytest.raises(installer.InstallError, match="effective later drop-in"): + installer.install_release( + second_bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert not host.active + assert host.loaded_release == first_release + assert host.n_restarts == prior_n_restarts + assert host.stop_calls == stop_calls + 1 + assert host.restart_calls == restart_calls + 1 + assert paths.unit.read_bytes() == first_unit + + +def test_effective_process_environment_override_is_rejected_and_rolled_back(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) + original_start = host._start + + def start_with_override() -> None: + original_start() + environment = paths.proc_root / str(host.main_pid) / "environ" + environment.write_bytes(f"PATH={package.SERVICE_PATH}\0DOCKER_HOST=tcp://unsafe.invalid:2375\0".encode()) + + host._start = start_with_override # type: ignore[method-assign] + + with pytest.raises(installer.InstallError, match="forbidden environment"): + installer.install_release( + bundle, + paths=paths, + source_binding=source_binding(), + runner=host, + sleeper=lambda _seconds: None, + execute_restart=True, + ) + + assert not host.active + assert not paths.state_dir.exists() + + +def test_control_helper_stops_only_trusted_full_container_id(tmp_path: Path) -> None: + cid_parent = tmp_path / "runtime" + cid_parent.mkdir(mode=0o700) + cidfile = cid_parent / "container.cid" + container_id = "c" * 64 + cidfile.write_text(container_id, encoding="ascii") + cidfile.chmod(0o600) + release_sha = "d" * 64 + image_reference = f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}" + container: dict[str, object] | None = { + "Id": container_id, + "Name": f"/{package.CONTAINER_NAME}", + "Image": "sha256:" + "f" * 64, + "Config": { + "Image": image_reference, + "Labels": { + "livingip.release.sha256": release_sha, + "livingip.container.role": "leoclean-gcp-nosend", + }, + }, + "State": {"Running": True}, + } + calls: list[list[str]] = [] + + def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + nonlocal container + calls.append(arguments) + operation = arguments[3:] + if operation[:2] == ["container", "inspect"]: + if container is None or operation[2] not in {package.CONTAINER_NAME, container_id}: + error = f"Error: No such container: {operation[2]}\n".encode() + return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=error) + return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([container]).encode(), stderr=b"") + if operation[:2] == ["container", "stop"]: + assert operation == ["container", "stop", "--time=30", container_id] + container = None + return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"") + raise AssertionError(arguments) + + result = control.control_container( + "stop", + release_sha, + image_reference, + "sha256:" + "f" * 64, + runner=runner, + sleeper=lambda _seconds: None, + cidfile=cidfile, + ) + + assert result == {"action": "stop", "status": "pass", "container": "stopped"} + assert not cidfile.exists() + stop_calls = [call[3:] for call in calls if call[3:5] == ["container", "stop"]] + assert stop_calls == [["container", "stop", "--time=30", container_id]] + + +def test_control_helper_refuses_config_digest_mismatch_without_stopping(tmp_path: Path) -> None: + cid_parent = tmp_path / "runtime" + cid_parent.mkdir(mode=0o700) + cidfile = cid_parent / "container.cid" + container_id = "c" * 64 + cidfile.write_text(container_id, encoding="ascii") + cidfile.chmod(0o600) + release_sha = "d" * 64 + image_reference = f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}" + container = { + "Id": container_id, + "Name": f"/{package.CONTAINER_NAME}", + "Image": "sha256:" + "1" * 64, + "Config": { + "Image": image_reference, + "Labels": { + "livingip.release.sha256": release_sha, + "livingip.container.role": "leoclean-gcp-nosend", + }, + }, + "State": {"Running": True}, + } + calls: list[list[str]] = [] + + def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + calls.append(arguments) + return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([container]).encode(), stderr=b"") + + with pytest.raises(control.ControlError, match="identity did not match"): + control.control_container( + "stop", + release_sha, + image_reference, + "sha256:" + "2" * 64, + runner=runner, + sleeper=lambda _seconds: None, + cidfile=cidfile, + ) + + assert cidfile.exists() + assert all(call[3:5] != ["container", "stop"] for call in calls) + + +def test_control_helper_retains_cid_when_name_rebinds_after_trusted_id_disappears(tmp_path: Path) -> None: + cid_parent = tmp_path / "runtime" + cid_parent.mkdir(mode=0o700) + cidfile = cid_parent / "container.cid" + container_id = "c" * 64 + rebound_id = "b" * 64 + cidfile.write_text(container_id, encoding="ascii") + cidfile.chmod(0o600) + release_sha = "d" * 64 + image_reference = f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}" + config_digest = "sha256:" + "f" * 64 + trusted = { + "Id": container_id, + "Name": f"/{package.CONTAINER_NAME}", + "Image": config_digest, + "Config": { + "Image": image_reference, + "Labels": { + "livingip.release.sha256": release_sha, + "livingip.container.role": "leoclean-gcp-nosend", + }, + }, + "State": {"Running": True}, + } + stopped = False + rebound = False + calls: list[list[str]] = [] + + def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + nonlocal stopped, rebound + calls.append(arguments) + operation = arguments[3:] + if operation[:2] == ["container", "stop"]: + stopped = True + return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"") + if operation[:2] == ["container", "inspect"]: + reference = operation[2] + if reference == container_id and stopped: + rebound = True + error = f"Error: No such container: {reference}\n".encode() + return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=error) + if reference == package.CONTAINER_NAME and rebound: + replacement = {"Id": rebound_id, "Name": f"/{package.CONTAINER_NAME}"} + return subprocess.CompletedProcess( + arguments, + 0, + stdout=json.dumps([replacement]).encode(), + stderr=b"", + ) + return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([trusted]).encode(), stderr=b"") + raise AssertionError(arguments) + + with pytest.raises(control.ControlError, match="name was rebound"): + control.control_container( + "stop", + release_sha, + image_reference, + config_digest, + runner=runner, + sleeper=lambda _seconds: None, + cidfile=cidfile, + ) + + assert stopped and rebound + assert cidfile.read_text(encoding="ascii") == container_id + assert [call[3:5] for call in calls].count(["container", "stop"]) == 1 + + +def test_control_helper_refuses_untrusted_name_collision_without_mutation(tmp_path: Path) -> None: + cid_parent = tmp_path / "runtime" + cid_parent.mkdir(mode=0o700) + calls: list[list[str]] = [] + + def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + calls.append(arguments) + operation = arguments[3:] + if operation == ["container", "inspect", package.CONTAINER_NAME]: + return subprocess.CompletedProcess( + arguments, + 0, + stdout=json.dumps([{"Id": "f" * 64, "Name": f"/{package.CONTAINER_NAME}"}]).encode(), + stderr=b"", + ) + raise AssertionError(arguments) + + with pytest.raises(control.ControlError, match="without a trusted ID"): + control.control_container( + "prestart", + "d" * 64, + f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}", + "sha256:" + "f" * 64, + runner=runner, + sleeper=lambda _seconds: None, + cidfile=cid_parent / "container.cid", + ) + + assert all(call[3:5] != ["container", "stop"] for call in calls) + + +def test_control_helper_does_not_treat_daemon_failure_as_container_absence(tmp_path: Path) -> None: + cid_parent = tmp_path / "runtime" + cid_parent.mkdir(mode=0o700) + cidfile = cid_parent / "container.cid" + + def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]: + return subprocess.CompletedProcess( + arguments, + 1, + stdout=b"", + stderr=b"Cannot connect to the Docker daemon\n", + ) + + with pytest.raises(control.ControlError, match="absence could not be proven"): + control.control_container( + "prestart", + "d" * 64, + f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}", + "sha256:" + "f" * 64, + runner=runner, + sleeper=lambda _seconds: None, + cidfile=cidfile, + ) + + assert not cidfile.exists() + + +def test_release_bundle_symlink_is_rejected(tmp_path: Path) -> None: + bundle, _release, _image = release_bundle(tmp_path, "2") + alias = tmp_path / "bundle-alias" + alias.symlink_to(bundle, target_is_directory=True) + + with pytest.raises(installer.InstallError, match="missing or unsafe"): + installer.validate_release_bundle(alias) + + +def test_service_unit_binds_local_docker_authority_and_private_cidfile(tmp_path: Path) -> None: + _bundle, release, _image = release_bundle(tmp_path, "2") + unit = package.render_systemd_unit(release) + + assert f"--host={control.DOCKER_HOST}" in unit + assert f"--config={control.DOCKER_CONFIG}" in unit + assert f"--cidfile={control.CIDFILE}" in unit + assert str(release["image"]["config_digest"]) in unit + assert "UMask=0077" in unit + assert "docker rm" not in unit + assert "--force" not in unit diff --git a/tests/test_gcp_leoclean_nosend_package.py b/tests/test_gcp_leoclean_nosend_package.py index 60883a8..fd044b9 100644 --- a/tests/test_gcp_leoclean_nosend_package.py +++ b/tests/test_gcp_leoclean_nosend_package.py @@ -1822,9 +1822,7 @@ def test_registry_inspection_absent_pull_dependent_retains_new_reference( calls: list[list[str]] = [] environments: list[dict[str, str]] = [] state = {"present": False, "dependent": False, "inspection_calls": 0} - payload = json.dumps( - [docker_inspection_fixture(image_input, reference, config_digest=config_digest)] - ).encode() + payload = json.dumps([docker_inspection_fixture(image_input, reference, config_digest=config_digest)]).encode() monkeypatch.setenv("DOCKER_HOST", "tcp://unreviewed.example:2375") monkeypatch.setenv("DOCKER_CONTEXT", "unreviewed") @@ -2703,7 +2701,13 @@ def test_rendered_unit_is_isolated_digest_pinned_and_has_no_credential_or_transp assert "NoNewPrivileges=yes" in unit assert "EnvironmentFile" not in unit assert "LoadCredential" not in unit - assert "docker.sock" not in unit + assert f"--host={package.DOCKER_HOST}" in unit + assert f"--config={package.SERVICE_DOCKER_CONFIG}" in unit + assert f"--cidfile={package.SERVICE_CIDFILE}" in unit + assert package.SERVICE_CONTROL_HELPER in unit + assert "docker rm" not in unit + assert "--force" not in unit + assert "--volume=/var/run/docker.sock" not in unit assert "--publish" not in unit assert "teleo-prod-1" not in unit assert "77.42.65.182" not in unit