1208 lines
45 KiB
Python
1208 lines
45 KiB
Python
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
|