2103 lines
82 KiB
Python
2103 lines
82 KiB
Python
#!/usr/bin/python3 -I
|
|
"""Install, restart, verify, and roll back one leoclean no-send release."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import binascii
|
|
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.v2"
|
|
ROLLBACK_STATE_SCHEMA = "livingip.leocleanNoSendRollbackState.v1"
|
|
ROLLBACK_RESULT_SCHEMA = "livingip.leocleanNoSendRollbackResult.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
|
|
MAX_ROLLBACK_STATE_BYTES = 32 * 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",
|
|
)
|
|
MANAGED_FILE_ROLES = frozenset({"unit", "helper", "release", "receipt", "rollback_state"})
|
|
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
|
|
rollback_state: 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",
|
|
rollback_state=state / "rollback-state.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
|
|
identity: tuple[int, ...] | None = None
|
|
|
|
|
|
@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) and observed.st_nlink == 1,
|
|
"installed path was not a regular file",
|
|
)
|
|
content = _read_stable_file(
|
|
path,
|
|
label="installed file",
|
|
max_bytes=MAX_ROLLBACK_STATE_BYTES,
|
|
expected_mode=stat.S_IMODE(observed.st_mode),
|
|
)
|
|
named = path.stat(follow_symlinks=False)
|
|
_require(
|
|
(named.st_dev, named.st_ino, named.st_mode, named.st_nlink)
|
|
== (observed.st_dev, observed.st_ino, observed.st_mode, observed.st_nlink),
|
|
"installed file changed while it was snapshotted",
|
|
)
|
|
return FileSnapshot(
|
|
True,
|
|
content,
|
|
stat.S_IMODE(observed.st_mode),
|
|
(
|
|
observed.st_dev,
|
|
observed.st_ino,
|
|
observed.st_mode,
|
|
observed.st_nlink,
|
|
observed.st_size,
|
|
observed.st_mtime_ns,
|
|
observed.st_ctime_ns,
|
|
),
|
|
)
|
|
|
|
|
|
def _snapshot_to_json(snapshot: FileSnapshot) -> dict[str, Any]:
|
|
if not snapshot.existed:
|
|
return {"existed": False}
|
|
_require(snapshot.identity is not None, "rollback snapshot identity was missing")
|
|
return {
|
|
"existed": True,
|
|
"mode": snapshot.mode,
|
|
"sha256": _sha256_bytes(snapshot.content),
|
|
"content_base64": base64.b64encode(snapshot.content).decode("ascii"),
|
|
}
|
|
|
|
|
|
def _snapshot_from_json(value: Any, *, label: str) -> FileSnapshot:
|
|
_require(isinstance(value, dict), f"{label} was invalid")
|
|
if value == {"existed": False}:
|
|
return FileSnapshot(False)
|
|
_require(
|
|
set(value) == {"existed", "mode", "sha256", "content_base64"}
|
|
and value.get("existed") is True
|
|
and isinstance(value.get("mode"), int)
|
|
and not isinstance(value.get("mode"), bool)
|
|
and 0 <= value["mode"] <= 0o777
|
|
and isinstance(value.get("sha256"), str)
|
|
and HEX_64.fullmatch(value["sha256"]) is not None
|
|
and isinstance(value.get("content_base64"), str),
|
|
f"{label} was invalid",
|
|
)
|
|
try:
|
|
content = base64.b64decode(value["content_base64"], validate=True)
|
|
except (ValueError, binascii.Error) as exc:
|
|
raise InstallError(f"{label} was invalid") from exc
|
|
_require(
|
|
len(content) <= MAX_ROLLBACK_STATE_BYTES and _sha256_bytes(content) == value["sha256"],
|
|
f"{label} was invalid",
|
|
)
|
|
return FileSnapshot(True, content, value["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, ...] | None,
|
|
) -> tuple[int, int] | None:
|
|
if installed_identity is not None:
|
|
observed = path.stat(follow_symlinks=False)
|
|
observed_identity = (
|
|
observed.st_dev,
|
|
observed.st_ino,
|
|
observed.st_mode,
|
|
observed.st_nlink,
|
|
observed.st_size,
|
|
observed.st_mtime_ns,
|
|
observed.st_ctime_ns,
|
|
)
|
|
_require(
|
|
stat.S_ISREG(observed.st_mode)
|
|
and (observed.st_dev, observed.st_ino) == installed_identity[:2]
|
|
and (len(installed_identity) == 2 or observed_identity == installed_identity),
|
|
"rollback refused a replaced installed file",
|
|
)
|
|
if snapshot.existed:
|
|
return _atomic_write(path, snapshot.content, snapshot.mode)
|
|
if path.exists() or path.is_symlink():
|
|
path.unlink()
|
|
_fsync_directory(path.parent)
|
|
return None
|
|
|
|
|
|
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:
|
|
raw = _read_stable_file(
|
|
paths.release,
|
|
label="installed release descriptor",
|
|
max_bytes=MAX_RELEASE_BYTES,
|
|
expected_mode=0o600,
|
|
)
|
|
value = _strict_json_bytes(raw, "installed release descriptor")
|
|
_require(isinstance(value, dict), "installed release descriptor was invalid")
|
|
package.validate_release_descriptor(value)
|
|
except (OSError, InstallError, package.PackageError) as exc:
|
|
raise InstallError("installed release descriptor was invalid") from exc
|
|
return value
|
|
|
|
|
|
def _managed_files(paths: InstallPaths) -> dict[str, tuple[Path, int]]:
|
|
return {
|
|
"unit": (paths.unit, 0o644),
|
|
"helper": (paths.helper, 0o755),
|
|
"release": (paths.release, 0o600),
|
|
"receipt": (paths.receipt, 0o600),
|
|
"rollback_state": (paths.rollback_state, 0o600),
|
|
}
|
|
|
|
|
|
def _read_strict_json_file(path: Path, *, label: str, max_bytes: int, mode: int) -> tuple[dict[str, Any], bytes]:
|
|
raw = _read_stable_file(path, label=label, max_bytes=max_bytes, expected_mode=mode)
|
|
value = _strict_json_bytes(raw, label)
|
|
_require(isinstance(value, dict), f"{label} was invalid")
|
|
return value, raw
|
|
|
|
|
|
def _validate_rollback_state(value: Mapping[str, Any], *, current_release_sha256: str) -> None:
|
|
_require(
|
|
set(value)
|
|
== {
|
|
"schema",
|
|
"current_release_sha256",
|
|
"previous_files",
|
|
"previous_directories",
|
|
"previous_service",
|
|
"sha256",
|
|
}
|
|
and value.get("schema") == ROLLBACK_STATE_SCHEMA
|
|
and value.get("current_release_sha256") == current_release_sha256,
|
|
"rollback state was invalid",
|
|
)
|
|
previous_files = value.get("previous_files")
|
|
_require(
|
|
isinstance(previous_files, dict) and set(previous_files) == MANAGED_FILE_ROLES,
|
|
"rollback state was invalid",
|
|
)
|
|
decoded = {
|
|
role: _snapshot_from_json(previous_files[role], label=f"rollback {role} snapshot")
|
|
for role in sorted(previous_files)
|
|
}
|
|
prior_presence = {role: snapshot.existed for role, snapshot in decoded.items()}
|
|
_require(
|
|
not any(prior_presence.values()) or all(prior_presence.values()),
|
|
"rollback prior managed file set was partial",
|
|
)
|
|
previous_directories = value.get("previous_directories")
|
|
_require(
|
|
isinstance(previous_directories, dict)
|
|
and set(previous_directories) == {"state_dir_existed", "docker_config_existed"}
|
|
and all(isinstance(item, bool) for item in previous_directories.values()),
|
|
"rollback state was invalid",
|
|
)
|
|
previous_service = value.get("previous_service")
|
|
_require(
|
|
isinstance(previous_service, dict)
|
|
and set(previous_service) == {"posture", "active_state", "sub_state", "n_restarts"}
|
|
and previous_service.get("posture") in {"absent", "inactive", "active"}
|
|
and previous_service.get("active_state") in {"inactive", "active"}
|
|
and previous_service.get("sub_state") in {"dead", "running"}
|
|
and isinstance(previous_service.get("n_restarts"), int)
|
|
and not isinstance(previous_service.get("n_restarts"), bool)
|
|
and previous_service["n_restarts"] >= 0,
|
|
"rollback state was invalid",
|
|
)
|
|
expected_state = {
|
|
"absent": ("inactive", "dead"),
|
|
"inactive": ("inactive", "dead"),
|
|
"active": ("active", "running"),
|
|
}[previous_service["posture"]]
|
|
_require(
|
|
(previous_service["active_state"], previous_service["sub_state"]) == expected_state
|
|
and ((previous_service["posture"] == "absent") == (not any(prior_presence.values()))),
|
|
"rollback state was invalid",
|
|
)
|
|
stable = {key: item for key, item in value.items() if key != "sha256"}
|
|
_require(value.get("sha256") == package.canonical_sha256(stable), "rollback state self-hash drifted")
|
|
|
|
|
|
def _read_rollback_state(paths: InstallPaths, release_sha256: str) -> tuple[dict[str, Any], bytes]:
|
|
value, raw = _read_strict_json_file(
|
|
paths.rollback_state,
|
|
label="rollback state",
|
|
max_bytes=MAX_ROLLBACK_STATE_BYTES,
|
|
mode=0o600,
|
|
)
|
|
_validate_rollback_state(value, current_release_sha256=release_sha256)
|
|
return value, raw
|
|
|
|
|
|
def _validate_install_receipt(
|
|
receipt: Mapping[str, Any],
|
|
*,
|
|
release: Mapping[str, Any],
|
|
paths: InstallPaths,
|
|
rollback_state_raw: bytes,
|
|
) -> None:
|
|
installed_files = receipt.get("installed_files")
|
|
rollback = receipt.get("rollback")
|
|
_require(
|
|
receipt.get("schema") == INSTALL_RECEIPT_SCHEMA
|
|
and receipt.get("status") == "pass"
|
|
and receipt.get("release_sha256") == release["release_sha256"]
|
|
and isinstance(installed_files, dict)
|
|
and set(installed_files)
|
|
== {"unit_sha256", "helper_sha256", "release_descriptor_sha256", "rollback_state_sha256"}
|
|
and installed_files.get("unit_sha256")
|
|
== _sha256_bytes(
|
|
_read_stable_file(
|
|
paths.unit, label="installed service unit", max_bytes=MAX_RELEASE_BYTES, expected_mode=0o644
|
|
)
|
|
)
|
|
and installed_files.get("helper_sha256")
|
|
== _sha256_bytes(
|
|
_read_stable_file(
|
|
paths.helper, label="installed service helper", max_bytes=MAX_SOURCE_BYTES, expected_mode=0o755
|
|
)
|
|
)
|
|
and installed_files.get("release_descriptor_sha256")
|
|
== _sha256_bytes(
|
|
_read_stable_file(
|
|
paths.release,
|
|
label="installed release descriptor",
|
|
max_bytes=MAX_RELEASE_BYTES,
|
|
expected_mode=0o600,
|
|
)
|
|
)
|
|
and installed_files.get("rollback_state_sha256") == _sha256_bytes(rollback_state_raw)
|
|
and isinstance(rollback, dict)
|
|
and rollback.get("state_sha256") == _sha256_bytes(rollback_state_raw)
|
|
and rollback.get("state_path") == str(paths.rollback_state),
|
|
"installed receipt was invalid",
|
|
)
|
|
_require(isinstance(receipt.get("installer_source"), dict), "installed receipt source binding was invalid")
|
|
_validate_source_binding(receipt["installer_source"])
|
|
_require(
|
|
receipt.get("image")
|
|
== {
|
|
"reference": release["image"]["reference"],
|
|
"digest": release["image"]["digest"],
|
|
"config_digest": release["image"]["config_digest"],
|
|
"input_sha256": release["image"]["input_sha256"],
|
|
},
|
|
"installed receipt image binding was invalid",
|
|
)
|
|
|
|
|
|
def _validate_managed_prior_state(paths: InstallPaths, helper_source: bytes) -> dict[str, Any] | None:
|
|
managed = _managed_files(paths)
|
|
presence = {role: path.exists() or path.is_symlink() for role, (path, _mode) in managed.items()}
|
|
if not any(presence.values()):
|
|
return None
|
|
_require(all(presence.values()), "managed install state was partial")
|
|
for path, mode in managed.values():
|
|
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")
|
|
rollback_state, rollback_state_raw = _read_rollback_state(paths, release["release_sha256"])
|
|
del rollback_state
|
|
receipt, _receipt_raw = _read_strict_json_file(
|
|
paths.receipt,
|
|
label="install receipt",
|
|
max_bytes=MAX_RELEASE_BYTES,
|
|
mode=0o600,
|
|
)
|
|
_validate_install_receipt(
|
|
receipt,
|
|
release=release,
|
|
paths=paths,
|
|
rollback_state_raw=rollback_state_raw,
|
|
)
|
|
_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
|
|
|
|
|
|
def _directory_existed_safely(path: Path, mode: int) -> bool:
|
|
if not path.exists() and not path.is_symlink():
|
|
return False
|
|
observed = path.stat(follow_symlinks=False)
|
|
_require(
|
|
stat.S_ISDIR(observed.st_mode) and observed.st_uid == os.geteuid() and stat.S_IMODE(observed.st_mode) == mode,
|
|
"prior install directory posture was unsafe",
|
|
)
|
|
return True
|
|
|
|
|
|
def _prior_directory_state(paths: InstallPaths) -> dict[str, bool]:
|
|
state_existed = _directory_existed_safely(paths.state_dir, 0o700)
|
|
docker_existed = _directory_existed_safely(paths.docker_config, 0o700)
|
|
if docker_existed:
|
|
_require(state_existed and not any(paths.docker_config.iterdir()), "prior Docker config posture was unsafe")
|
|
if state_existed:
|
|
allowed = {paths.docker_config.name} if docker_existed else set()
|
|
observed = {path.name for path in paths.state_dir.iterdir()}
|
|
managed_names = {paths.release.name, paths.receipt.name, paths.rollback_state.name}
|
|
_require(
|
|
observed <= allowed | managed_names,
|
|
"prior install state directory contained an unmanaged entry",
|
|
)
|
|
return {
|
|
"state_dir_existed": state_existed,
|
|
"docker_config_existed": docker_existed,
|
|
}
|
|
|
|
|
|
def _build_rollback_state(
|
|
release_sha256: str,
|
|
snapshots: Mapping[Path, FileSnapshot],
|
|
paths: InstallPaths,
|
|
previous_directories: Mapping[str, bool],
|
|
prior_posture: ServicePosture,
|
|
) -> dict[str, Any]:
|
|
path_to_role = {path: role for role, (path, _mode) in _managed_files(paths).items()}
|
|
_require(set(snapshots) == set(path_to_role), "rollback snapshot set was not exact")
|
|
stable = {
|
|
"schema": ROLLBACK_STATE_SCHEMA,
|
|
"current_release_sha256": release_sha256,
|
|
"previous_files": {
|
|
path_to_role[path]: _snapshot_to_json(snapshots[path])
|
|
for path in sorted(snapshots, key=lambda item: path_to_role[item])
|
|
},
|
|
"previous_directories": dict(previous_directories),
|
|
"previous_service": {
|
|
"posture": prior_posture.kind,
|
|
"active_state": prior_posture.state["ActiveState"],
|
|
"sub_state": prior_posture.state["SubState"],
|
|
"n_restarts": prior_posture.state["NRestarts"],
|
|
},
|
|
}
|
|
value = {**stable, "sha256": package.canonical_sha256(stable)}
|
|
_validate_rollback_state(value, current_release_sha256=release_sha256)
|
|
return value
|
|
|
|
|
|
@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"
|
|
previous_directories = _prior_directory_state(paths)
|
|
|
|
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),
|
|
paths.rollback_state: _snapshot(paths.rollback_state),
|
|
}
|
|
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)
|
|
rollback_state = _build_rollback_state(
|
|
release["release_sha256"],
|
|
snapshots,
|
|
paths,
|
|
previous_directories,
|
|
prior_posture,
|
|
)
|
|
rollback_state_bytes = _canonical_json(rollback_state)
|
|
_require(
|
|
len(rollback_state_bytes) <= MAX_ROLLBACK_STATE_BYTES,
|
|
"rollback state exceeded the safety limit",
|
|
)
|
|
attempted_paths.add(paths.rollback_state)
|
|
_publish_managed_file(paths.rollback_state, rollback_state_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),
|
|
"rollback_state_sha256": _sha256_bytes(rollback_state_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,
|
|
"state_path": str(paths.rollback_state),
|
|
"state_sha256": _sha256_bytes(rollback_state_bytes),
|
|
},
|
|
"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.rollback_state, 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 _json_from_snapshot(snapshot: FileSnapshot, *, label: str) -> dict[str, Any]:
|
|
_require(snapshot.existed, f"{label} was absent")
|
|
value = _strict_json_bytes(snapshot.content, label)
|
|
_require(isinstance(value, dict), f"{label} was invalid")
|
|
return value
|
|
|
|
|
|
def _validate_managed_snapshots(
|
|
snapshots: Mapping[str, FileSnapshot],
|
|
*,
|
|
paths: InstallPaths,
|
|
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
|
|
expected_modes = {role: mode for role, (_path, mode) in _managed_files(paths).items()}
|
|
_require(set(snapshots) == set(expected_modes), "managed snapshot set was not exact")
|
|
_require(all(snapshot.existed for snapshot in snapshots.values()), "managed snapshot set was partial")
|
|
_require(
|
|
all(snapshots[role].mode == expected_modes[role] for role in expected_modes),
|
|
"managed snapshot mode drifted",
|
|
)
|
|
release = _json_from_snapshot(snapshots["release"], label="snapshot release descriptor")
|
|
try:
|
|
package.validate_release_descriptor(release)
|
|
except package.PackageError as exc:
|
|
raise InstallError("snapshot release descriptor was invalid") from exc
|
|
release_sha256 = release["release_sha256"]
|
|
_require(
|
|
snapshots["unit"].content == package.render_systemd_unit(release).encode("utf-8")
|
|
and snapshots["helper"].content.startswith(b"#!/usr/bin/python3 -I\n"),
|
|
"snapshot installed files differed from the release",
|
|
)
|
|
rollback_state = _json_from_snapshot(snapshots["rollback_state"], label="snapshot rollback state")
|
|
_validate_rollback_state(rollback_state, current_release_sha256=release_sha256)
|
|
receipt = _json_from_snapshot(snapshots["receipt"], label="snapshot install receipt")
|
|
installed_files = receipt.get("installed_files")
|
|
rollback = receipt.get("rollback")
|
|
_require(
|
|
receipt.get("schema") == INSTALL_RECEIPT_SCHEMA
|
|
and receipt.get("status") == "pass"
|
|
and receipt.get("release_sha256") == release_sha256
|
|
and isinstance(receipt.get("installer_source"), dict)
|
|
and isinstance(installed_files, dict)
|
|
and installed_files.get("unit_sha256") == _sha256_bytes(snapshots["unit"].content)
|
|
and installed_files.get("helper_sha256") == _sha256_bytes(snapshots["helper"].content)
|
|
and installed_files.get("release_descriptor_sha256") == _sha256_bytes(snapshots["release"].content)
|
|
and installed_files.get("rollback_state_sha256") == _sha256_bytes(snapshots["rollback_state"].content)
|
|
and isinstance(rollback, dict)
|
|
and rollback.get("state_path") == str(paths.rollback_state)
|
|
and rollback.get("state_sha256") == _sha256_bytes(snapshots["rollback_state"].content),
|
|
"snapshot install receipt was invalid",
|
|
)
|
|
_validate_source_binding(receipt["installer_source"])
|
|
return release, receipt, rollback_state
|
|
|
|
|
|
def _current_managed_snapshots(paths: InstallPaths) -> dict[str, FileSnapshot]:
|
|
return {role: _snapshot(path) for role, (path, _mode) in _managed_files(paths).items()}
|
|
|
|
|
|
def _previous_snapshots(rollback_state: Mapping[str, Any]) -> dict[str, FileSnapshot]:
|
|
previous_files = rollback_state["previous_files"]
|
|
_require(isinstance(previous_files, dict), "rollback prior file set was invalid")
|
|
return {
|
|
role: _snapshot_from_json(previous_files[role], label=f"rollback {role} snapshot")
|
|
for role in sorted(previous_files)
|
|
}
|
|
|
|
|
|
def _prepare_explicit_rollback(
|
|
expected_release_sha256: str,
|
|
*,
|
|
paths: InstallPaths,
|
|
source_binding: Mapping[str, Any],
|
|
runner: Runner,
|
|
sleeper: Sleeper,
|
|
) -> dict[str, Any]:
|
|
_require(HEX_64.fullmatch(expected_release_sha256) is not None, "expected release hash was invalid")
|
|
_bound_source_files(source_binding)
|
|
managed_release = _validate_managed_prior_state(paths, b"explicit rollback")
|
|
_require(managed_release is not None, "explicit rollback found no managed release")
|
|
current_snapshots = _current_managed_snapshots(paths)
|
|
current_release, current_receipt, rollback_state = _validate_managed_snapshots(
|
|
current_snapshots,
|
|
paths=paths,
|
|
)
|
|
_require(
|
|
current_release == managed_release and current_release["release_sha256"] == expected_release_sha256,
|
|
"installed release did not match the expected release hash",
|
|
)
|
|
previous_snapshots = _previous_snapshots(rollback_state)
|
|
previous_presence = [snapshot.existed for snapshot in previous_snapshots.values()]
|
|
previous_release: dict[str, Any] | None
|
|
if any(previous_presence):
|
|
_require(all(previous_presence), "rollback prior managed file set was partial")
|
|
previous_release, _previous_receipt, _previous_rollback = _validate_managed_snapshots(
|
|
previous_snapshots,
|
|
paths=paths,
|
|
)
|
|
else:
|
|
previous_release = None
|
|
current_posture = _stable_service_posture(
|
|
current_release,
|
|
runner=runner,
|
|
paths=paths,
|
|
sleeper=sleeper,
|
|
)
|
|
current_verification: dict[str, Any] | None = None
|
|
if current_posture.kind == "active":
|
|
current_verification = verify_running_release(
|
|
current_release,
|
|
runner=runner,
|
|
paths=paths,
|
|
sleeper=sleeper,
|
|
)
|
|
_require(
|
|
current_verification["main_pid"] == current_posture.state["MainPID"]
|
|
and current_verification["n_restarts"] == current_posture.state["NRestarts"]
|
|
and current_verification["invocation_id"] == current_posture.state["InvocationID"],
|
|
"installed service changed during rollback preparation",
|
|
)
|
|
previous_service = rollback_state["previous_service"]
|
|
_require(
|
|
(previous_release is None) == (previous_service["posture"] == "absent"),
|
|
"rollback prior release differed from its recorded posture",
|
|
)
|
|
return {
|
|
"current_release": current_release,
|
|
"current_receipt": current_receipt,
|
|
"current_snapshots": current_snapshots,
|
|
"current_posture": current_posture,
|
|
"current_verification": current_verification,
|
|
"previous_snapshots": previous_snapshots,
|
|
"previous_release": previous_release,
|
|
"previous_directories": rollback_state["previous_directories"],
|
|
"previous_service": previous_service,
|
|
"rollback_state_sha256": _sha256_bytes(current_snapshots["rollback_state"].content),
|
|
}
|
|
|
|
|
|
def _build_rollback_plan(
|
|
expected_release_sha256: str,
|
|
prepared: Mapping[str, Any],
|
|
*,
|
|
paths: InstallPaths,
|
|
source_binding: Mapping[str, Any],
|
|
) -> dict[str, Any]:
|
|
previous_release = prepared["previous_release"]
|
|
return {
|
|
"schema": ROLLBACK_RESULT_SCHEMA,
|
|
"status": "pass",
|
|
"mode": "dry_run",
|
|
"installer_source": dict(source_binding),
|
|
"expected_release_sha256": expected_release_sha256,
|
|
"installed_release_sha256": prepared["current_release"]["release_sha256"],
|
|
"installed_receipt_release_sha256": prepared["current_receipt"]["release_sha256"],
|
|
"rollback_state_sha256": prepared["rollback_state_sha256"],
|
|
"current_service_posture": prepared["current_posture"].kind,
|
|
"target_service_posture": prepared["previous_service"]["posture"],
|
|
"target_release_sha256": previous_release["release_sha256"] if previous_release is not None else None,
|
|
"target": {
|
|
"service": SERVICE,
|
|
"unit": str(paths.unit),
|
|
"helper": str(paths.helper),
|
|
"state_directory": str(paths.state_dir),
|
|
},
|
|
"authority": {
|
|
"execute_requires_root_and_execute_rollback": True,
|
|
"image_pulled": False,
|
|
"cloud_or_database_authority_changed": False,
|
|
"transport_authority_changed": False,
|
|
},
|
|
}
|
|
|
|
|
|
def _remove_rollback_owned_directories(paths: InstallPaths, previous_directories: Mapping[str, bool]) -> None:
|
|
_require(not any(paths.docker_config.iterdir()), "rollback Docker config directory was not empty")
|
|
if not previous_directories["docker_config_existed"]:
|
|
paths.docker_config.rmdir()
|
|
_fsync_directory(paths.state_dir)
|
|
if not previous_directories["state_dir_existed"]:
|
|
_require(not any(paths.state_dir.iterdir()), "rollback state directory was not empty")
|
|
paths.state_dir.rmdir()
|
|
_fsync_directory(paths.state_dir.parent)
|
|
|
|
|
|
def _assert_snapshot_identities(
|
|
snapshots: Mapping[str, FileSnapshot],
|
|
*,
|
|
paths: InstallPaths,
|
|
) -> None:
|
|
for role, (path, expected_mode) in _managed_files(paths).items():
|
|
snapshot = snapshots[role]
|
|
_require(snapshot.existed and snapshot.identity is not None, "installed rollback file identity was missing")
|
|
try:
|
|
observed = path.stat(follow_symlinks=False)
|
|
except OSError as exc:
|
|
raise InstallError("installed rollback file identity drifted") from exc
|
|
observed_identity = (
|
|
observed.st_dev,
|
|
observed.st_ino,
|
|
observed.st_mode,
|
|
observed.st_nlink,
|
|
observed.st_size,
|
|
observed.st_mtime_ns,
|
|
observed.st_ctime_ns,
|
|
)
|
|
_require(
|
|
stat.S_ISREG(observed.st_mode)
|
|
and observed.st_nlink == 1
|
|
and stat.S_IMODE(observed.st_mode) == expected_mode
|
|
and observed_identity == snapshot.identity,
|
|
"installed rollback file identity drifted",
|
|
)
|
|
|
|
|
|
def _execute_explicit_rollback(
|
|
expected_release_sha256: str,
|
|
prepared: Mapping[str, Any],
|
|
*,
|
|
paths: InstallPaths,
|
|
runner: Runner,
|
|
sleeper: Sleeper,
|
|
) -> dict[str, Any]:
|
|
current_release = prepared["current_release"]
|
|
current_snapshots = prepared["current_snapshots"]
|
|
current_posture: ServicePosture = prepared["current_posture"]
|
|
previous_snapshots = prepared["previous_snapshots"]
|
|
previous_release = prepared["previous_release"]
|
|
previous_service = prepared["previous_service"]
|
|
role_paths = {role: path for role, (path, _mode) in _managed_files(paths).items()}
|
|
restored_identities: dict[Path, tuple[int, int] | None] = {}
|
|
mutated_paths: set[Path] = set()
|
|
files_mutated = False
|
|
service_stopped = False
|
|
target_may_be_active = False
|
|
try:
|
|
_assert_snapshot_identities(current_snapshots, paths=paths)
|
|
if current_posture.kind == "active":
|
|
fresh_verification = verify_running_release(
|
|
current_release,
|
|
runner=runner,
|
|
paths=paths,
|
|
sleeper=sleeper,
|
|
)
|
|
_require(
|
|
fresh_verification == prepared["current_verification"],
|
|
"service or container drifted before rollback stop",
|
|
)
|
|
before_stop = _service_state(runner)
|
|
_require(
|
|
before_stop["MainPID"] == current_posture.state["MainPID"]
|
|
and before_stop["NRestarts"] == current_posture.state["NRestarts"]
|
|
and before_stop["InvocationID"] == current_posture.state["InvocationID"]
|
|
and before_stop["ActiveState"] == "active"
|
|
and before_stop["SubState"] == "running",
|
|
"service PID or invocation drifted before rollback stop",
|
|
)
|
|
_systemctl("stop", runner)
|
|
service_stopped = True
|
|
stopped = _stable_service_posture(
|
|
current_release,
|
|
runner=runner,
|
|
paths=paths,
|
|
sleeper=sleeper,
|
|
)
|
|
_require(stopped.kind == "inactive", "installed service did not stop into a trusted posture")
|
|
for role in ("receipt", "rollback_state", "release", "unit", "helper"):
|
|
path = role_paths[role]
|
|
files_mutated = True
|
|
mutated_paths.add(path)
|
|
restored_identities[path] = _restore_snapshot(
|
|
path,
|
|
previous_snapshots[role],
|
|
current_snapshots[role].identity,
|
|
)
|
|
_systemctl("daemon-reload", runner)
|
|
if previous_service["posture"] == "active":
|
|
_require(previous_release is not None, "rollback active target release was missing")
|
|
target_may_be_active = True
|
|
_systemctl("restart", runner)
|
|
final_verification = verify_running_release(
|
|
previous_release,
|
|
runner=runner,
|
|
paths=paths,
|
|
sleeper=sleeper,
|
|
)
|
|
_require(
|
|
final_verification["n_restarts"] == previous_service["n_restarts"],
|
|
"rollback did not restore the recorded active posture",
|
|
)
|
|
final_posture = "active"
|
|
elif previous_service["posture"] == "inactive":
|
|
_require(previous_release is not None, "rollback inactive target release was missing")
|
|
final = _stable_service_posture(
|
|
previous_release,
|
|
runner=runner,
|
|
paths=paths,
|
|
sleeper=sleeper,
|
|
)
|
|
_require(
|
|
final.kind == "inactive" and final.state["NRestarts"] == previous_service["n_restarts"],
|
|
"rollback did not restore the recorded inactive posture",
|
|
)
|
|
final_verification = None
|
|
final_posture = "inactive"
|
|
else:
|
|
_require(previous_release is None, "rollback absent target unexpectedly had a release")
|
|
_systemctl("reset-failed", runner)
|
|
final = _stable_service_posture(
|
|
None,
|
|
runner=runner,
|
|
paths=paths,
|
|
sleeper=sleeper,
|
|
)
|
|
_require(final.kind == "absent", "rollback did not restore the absent posture")
|
|
_remove_rollback_owned_directories(paths, prepared["previous_directories"])
|
|
final_verification = None
|
|
final_posture = "absent"
|
|
return {
|
|
"schema": ROLLBACK_RESULT_SCHEMA,
|
|
"status": "pass",
|
|
"mode": "executed",
|
|
"rolled_back_release_sha256": expected_release_sha256,
|
|
"restored_release_sha256": previous_release["release_sha256"] if previous_release is not None else None,
|
|
"restored_service_posture": final_posture,
|
|
"verification": final_verification,
|
|
"authority": {
|
|
"image_pulled": False,
|
|
"cloud_or_database_authority_changed": False,
|
|
"transport_authority_changed": False,
|
|
},
|
|
}
|
|
except Exception as exc:
|
|
recovery_errors: list[str] = []
|
|
if target_may_be_active:
|
|
try:
|
|
_systemctl("stop", runner)
|
|
except InstallError:
|
|
recovery_errors.append("target_stop")
|
|
if files_mutated:
|
|
for role in ("receipt", "rollback_state", "release", "unit", "helper"):
|
|
path = role_paths[role]
|
|
if path not in mutated_paths:
|
|
continue
|
|
try:
|
|
if restored_identities.get(path) is None:
|
|
_require(
|
|
not path.exists() and not path.is_symlink(),
|
|
"rollback recovery refused a replaced pathname",
|
|
)
|
|
_restore_snapshot(path, current_snapshots[role], restored_identities.get(path))
|
|
except (OSError, InstallError):
|
|
recovery_errors.append("file_restore")
|
|
try:
|
|
_systemctl("daemon-reload", runner)
|
|
except InstallError:
|
|
recovery_errors.append("daemon_reload")
|
|
if service_stopped and current_posture.kind == "active":
|
|
try:
|
|
_systemctl("restart", runner)
|
|
verify_running_release(current_release, runner=runner, paths=paths, sleeper=sleeper)
|
|
except InstallError:
|
|
recovery_errors.append("current_restart")
|
|
if recovery_errors:
|
|
raise InstallError("explicit rollback failed and recovery was incomplete") from exc
|
|
if isinstance(exc, InstallError):
|
|
raise
|
|
raise InstallError("explicit rollback failed without changing the installed release") from exc
|
|
|
|
|
|
def rollback_release(
|
|
expected_release_sha256: str,
|
|
*,
|
|
paths: InstallPaths,
|
|
source_binding: Mapping[str, Any],
|
|
runner: Runner = _default_runner,
|
|
sleeper: Sleeper = time.sleep,
|
|
execute_rollback: bool = False,
|
|
) -> dict[str, Any]:
|
|
if not execute_rollback:
|
|
prepared = _prepare_explicit_rollback(
|
|
expected_release_sha256,
|
|
paths=paths,
|
|
source_binding=source_binding,
|
|
runner=runner,
|
|
sleeper=sleeper,
|
|
)
|
|
return _build_rollback_plan(
|
|
expected_release_sha256,
|
|
prepared,
|
|
paths=paths,
|
|
source_binding=source_binding,
|
|
)
|
|
with _install_lock(paths):
|
|
prepared = _prepare_explicit_rollback(
|
|
expected_release_sha256,
|
|
paths=paths,
|
|
source_binding=source_binding,
|
|
runner=runner,
|
|
sleeper=sleeper,
|
|
)
|
|
return _execute_explicit_rollback(
|
|
expected_release_sha256,
|
|
prepared,
|
|
paths=paths,
|
|
runner=runner,
|
|
sleeper=sleeper,
|
|
)
|
|
|
|
|
|
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", type=Path)
|
|
parser.add_argument("--execute-restart", action="store_true")
|
|
parser.add_argument("--rollback-release")
|
|
parser.add_argument("--execute-rollback", action="store_true")
|
|
args = parser.parse_args(argv)
|
|
if args.rollback_release is None:
|
|
_require(args.bundle is not None and not args.execute_rollback, "invalid arguments")
|
|
else:
|
|
_require(
|
|
args.bundle is None and not args.execute_restart and HEX_64.fullmatch(args.rollback_release) is not None,
|
|
"invalid arguments",
|
|
)
|
|
_require(not (args.execute_restart and args.execute_rollback), "invalid arguments")
|
|
return args
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
try:
|
|
args = parse_args(argv)
|
|
root = Path("/")
|
|
if args.execute_restart or args.execute_rollback:
|
|
_require(os.geteuid() == 0, "live installation or rollback requires root")
|
|
paths = InstallPaths.under(root)
|
|
binding = reviewed_source_binding()
|
|
if args.rollback_release is not None:
|
|
result = rollback_release(
|
|
args.rollback_release,
|
|
paths=paths,
|
|
source_binding=binding,
|
|
execute_rollback=args.execute_rollback,
|
|
)
|
|
else:
|
|
_require(args.bundle is not None, "invalid arguments")
|
|
result = install_release(
|
|
args.bundle.absolute(),
|
|
paths=paths,
|
|
source_binding=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())
|