"""Exercise installer guards without installing software or touching services.""" import hashlib import importlib.util import json import os from pathlib import Path import re import stat import subprocess import sys import tempfile import unittest from unittest.mock import patch SOURCE = (Path(__file__).resolve().parents[1] / "install.sh").read_text() BLOCKS = re.findall(r"<<'PY'\n(.*?)\nPY", SOURCE, re.DOTALL) PREFLIGHT = next(block for block in BLOCKS if "# INSTALLATION_PREFLIGHT:" in block) OWNERSHIP = next(block for block in BLOCKS if "# INSTALLATION_OWNERSHIP:" in block) RELEASE = next(block for block in BLOCKS if "# INSTALLATION_RELEASE:" in block) CLEANUP = re.search(r"installation_cleanup\(\) \{.*?\n\}", SOURCE, re.DOTALL).group() DOCKER = SOURCE[SOURCE.index("# INSTALLATION_DOCKER:"):SOURCE.index("install -d -m 0700 /etc/electric-cherry")] SPEC = importlib.util.spec_from_file_location("package_release", Path(__file__).resolve().parents[1] / "package_release.py") PACKAGER = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(PACKAGER) class InstallerTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.base = Path(self.temporary.name).resolve() self.root = self.base / "managed" self.config = self.base / "configuration" / "agent.json" self.config.parent.mkdir() self.controller = "https://controller.example" self.saved = {"root": str(self.root), "controller": self.controller, "hostId": "host-one", "credential": "private"} self.overrides = {} def tearDown(self): self.temporary.cleanup() def preflight(self, root=None): original_lstat = Path.lstat def root_owned(path): # Simulate Linux root-owned parents while keeping the actual inode # types. No filesystem mutations or elevated privileges are needed. info = original_lstat(path) fields = list(info) fields[4] = 0 if stat.S_ISDIR(info.st_mode): fields[0] &= ~0o022 override = self.overrides.get(str(path), {}) fields[0] = override.get("mode", fields[0]) fields[4] = override.get("uid", fields[4]) return os.stat_result(fields) argv = ["-", self.controller, str(root or self.root), str(self.config)] with patch.object(sys, "argv", argv), patch.object(Path, "lstat", root_owned): exec(compile(PREFLIGHT, "installer-preflight", "exec"), {}) def legacy_installation(self): for directory in ("agent", "instances", "libraries"): (self.root / directory).mkdir(parents=True, exist_ok=True) self.config.write_text(json.dumps(self.saved)) def ownership(self): with patch.object(sys, "argv", ["-", str(self.root), str(self.config)]): exec(compile(OWNERSHIP, "installer-ownership", "exec"), {}) def test_new_directory_and_empty_directory_are_allowed_without_adopting_files(self): self.preflight() self.assertFalse(self.root.exists()) self.root.mkdir() self.preflight() (self.root / "unrelated.txt").write_text("keep") with self.assertRaisesRegex(SystemExit, "not owned"): self.preflight() self.assertEqual((self.root / "unrelated.txt").read_text(), "keep") def test_known_legacy_layout_can_upgrade_and_marker_binds_exact_enrollment(self): self.legacy_installation() original_config = self.config.read_bytes() (self.root / "libraries" / "game.ff").write_bytes(b"game data") self.preflight() self.ownership() marker = self.root / ".electric-cherry.json" self.assertEqual(json.loads(marker.read_text()), { "version": 1, "root": str(self.root), "hostId": "host-one", "controller": self.controller}) self.assertEqual(stat.S_IMODE(marker.stat().st_mode), 0o600) self.assertEqual(self.config.read_bytes(), original_config) self.assertEqual((self.root / "libraries" / "game.ff").read_bytes(), b"game data") self.preflight() def test_legacy_upgrade_never_adopts_unrelated_top_level_files(self): self.legacy_installation() (self.root / "personal-backup.tar").write_bytes(b"keep") with self.assertRaisesRegex(SystemExit, "not owned"): self.preflight() self.assertFalse((self.root / ".electric-cherry.json").exists()) def test_reenrollment_cannot_silently_change_controller_or_root(self): self.legacy_installation() for field, value in (("controller", "https://another.example"), ("root", str(self.base / "other"))): with self.subTest(field=field): self.config.write_text(json.dumps({**self.saved, field: value})) with self.assertRaisesRegex(SystemExit, "different controller or data directory"): self.preflight() def test_existing_marker_must_match_controller_host_and_root(self): self.legacy_installation() self.ownership() marker = self.root / ".electric-cherry.json" valid = json.loads(marker.read_text()) for field, value in (("hostId", "host-two"), ("root", str(self.base)), ("controller", "https://another.example")): with self.subTest(field=field): marker.write_text(json.dumps({**valid, field: value})) with self.assertRaisesRegex(SystemExit, "ownership does not match"): self.preflight() def test_lexical_aliases_and_system_roots_are_rejected_before_mutation(self): for root in ("/var/lib/..", "/var/lib/../lib", "/var//lib/managed", "/var/lib/./managed", "/var/lib/managed/", "/var/lib", "/usr/local/libexec", "/", "/games", str(self.config.parent), str(self.config.parent / "data")): with self.subTest(root=root), self.assertRaisesRegex(SystemExit, "canonical, dedicated"): self.preflight(root) self.assertFalse(self.root.exists()) def test_symlink_roots_ancestors_metadata_and_managed_subdirectories_are_rejected(self): elsewhere = self.base / "elsewhere" elsewhere.mkdir() self.root.symlink_to(elsewhere, target_is_directory=True) with self.assertRaisesRegex(SystemExit, "not symlinks"): self.preflight() with self.assertRaisesRegex(SystemExit, "not symlinks"): self.preflight(str(self.root / "nested")) self.root.unlink() self.legacy_installation() (self.root / "libraries").rmdir() (self.root / "libraries").symlink_to(elsewhere, target_is_directory=True) with self.assertRaisesRegex(SystemExit, "not symlinks"): self.preflight() (self.root / "libraries").unlink() (self.root / "libraries").mkdir() self.config.unlink() real_config = self.base / "real-config.json" real_config.write_text(json.dumps(self.saved)) self.config.symlink_to(real_config) with self.assertRaisesRegex(SystemExit, "regular, root-owned"): self.preflight() def test_untrusted_parents_and_metadata_cannot_authorize_cleanup(self): self.legacy_installation() self.overrides[str(self.base)] = {"uid": 1000} with self.assertRaisesRegex(SystemExit, "root-owned"): self.preflight() self.overrides = {str(self.base): {"mode": stat.S_IFDIR | 0o777}} with self.assertRaisesRegex(SystemExit, "root-owned"): self.preflight() self.overrides = {str(self.config): {"uid": 1000}} with self.assertRaisesRegex(SystemExit, "regular, root-owned"): self.preflight() def test_pending_deletion_blocks_upgrade_even_with_dangling_request_symlink(self): self.legacy_installation() request = self.config.with_name("decommission.json") request.write_text("{}") with self.assertRaisesRegex(SystemExit, "deletion is in progress"): self.preflight() request.unlink() request.symlink_to(self.base / "missing") with self.assertRaisesRegex(SystemExit, "deletion is in progress"): self.preflight() def test_pending_update_recovery_is_preserved_and_blocks_manual_replacement(self): self.legacy_installation() update = self.config.parent / "update" update.mkdir() journal = update / "journal.json" saved = '{"phase":"swapping","targetVersion":"1.2.0"}' journal.write_text(saved) with self.assertRaisesRegex(SystemExit, "automatic update needs recovery"): self.preflight() self.assertEqual(journal.read_text(), saved) journal.unlink() journal.symlink_to(self.base / "missing") with self.assertRaisesRegex(SystemExit, "automatic update needs recovery"): self.preflight() def test_update_directory_and_lock_cannot_follow_symlinks(self): self.legacy_installation() elsewhere = self.base / "unrelated" elsewhere.mkdir() update = self.config.parent / "update" update.symlink_to(elsewhere, target_is_directory=True) with self.assertRaisesRegex(SystemExit, "not symlinks"): self.preflight() update.unlink() lock = self.config.with_name("update.lock") lock.symlink_to(elsewhere / "sentinel") with self.assertRaisesRegex(SystemExit, "regular, root-owned"): self.preflight() self.assertFalse((elsewhere / "sentinel").exists()) lock.unlink() activity = self.root / "agent" / "activity.lock" activity.symlink_to(elsewhere / "activity-sentinel") with self.assertRaisesRegex(SystemExit, "regular, root-owned"): self.preflight() self.assertFalse((elsewhere / "activity-sentinel").exists()) def test_data_root_cannot_contain_the_updater_or_service_installation(self): for path in ("/etc/systemd", "/etc/systemd/system", "/usr/local/libexec/electric-cherry-update.py", "/usr/local/libexec/electric-cherry-agent.py.update", "/usr/local/libexec/electric-cherry-uninstall.py.update"): with self.subTest(path=path), self.assertRaisesRegex(SystemExit, "outside the agent executable and service"): self.preflight(path) def test_mounted_parent_is_allowed_but_data_root_and_nested_mounts_are_rejected(self): mountinfo = Path('/proc/1/mountinfo') original_exists, original_read = Path.exists, Path.read_text for mount, allowed in ((self.base, True), (self.root, False), (self.root / 'libraries', False)): def exists(path): return True if path == mountinfo else original_exists(path) def read(path, *args, **kwargs): return f'29 24 8:1 / {mount} rw - ext4 /dev/sda1 rw' if path == mountinfo else original_read(path, *args, **kwargs) with self.subTest(mount=mount), patch.object(Path, 'exists', exists), patch.object(Path, 'read_text', read): if allowed: self.preflight() else: with self.assertRaisesRegex(SystemExit, "dedicated subdirectory of the mounted disk"): self.preflight() class DockerInstallerTests(unittest.TestCase): def run_docker_setup(self, version="13", cli=False, engine=False, available=True, data=False, config=False, apt_status=0, supply_cli=True): with tempfile.TemporaryDirectory() as temporary: base = Path(temporary) docker_data = base / "docker-data" docker_config = base / "daemon.json" if data: docker_data.mkdir() (docker_data / "sentinel").write_text("existing containers") if config: docker_config.write_text("existing configuration") block = DOCKER.replace("/var/lib/docker", str(docker_data)).replace("/etc/docker/daemon.json", str(docker_config)) # Simulate Debian's package split and Docker state. All commands that # could install packages or touch services are shell stubs. script = r''' set -euo pipefail command() { [ "$*" = '-v docker' ] && [ "$EC_TEST_CLI" = true ]; } dpkg-query() { [ "$*" = '-W -f=${Status} docker.io' ] || exit 90 [ "$EC_TEST_ENGINE" = true ] || return 1 printf 'install ok installed' } apt-get() { printf 'apt-get %s\n' "$*" [ "$EC_TEST_APT_STATUS" = 0 ] || return "$EC_TEST_APT_STATUS" if [ "$EC_TEST_SUPPLY_CLI" = true ]; then case " $* " in *' docker-cli '*) EC_TEST_CLI=true ;; *' docker.io '*) if [ "$VERSION_ID" = 12 ]; then EC_TEST_CLI=true; fi ;; esac fi } systemctl() { printf 'systemctl %s\n' "$*"; } docker() { [ "$EC_TEST_CLI" = true ] || return 127 [ "$*" = info ] || exit 91 if [ "$EC_TEST_AVAILABLE" != true ]; then echo 'Cannot connect to the Docker daemon' >&2 return 1 fi } ''' environment = {**os.environ, "VERSION_ID": version, "EC_TEST_CLI": str(cli).lower(), "EC_TEST_ENGINE": str(engine).lower(), "EC_TEST_AVAILABLE": str(available).lower(), "EC_TEST_APT_STATUS": str(apt_status), "EC_TEST_SUPPLY_CLI": str(supply_cli).lower()} result = subprocess.run(["bash", "-c", script + block], env=environment, capture_output=True, text=True) if data: self.assertEqual((docker_data / "sentinel").read_text(), "existing containers") if config: self.assertEqual(docker_config.read_text(), "existing configuration") return result def test_fresh_debian_12_and_13_install_both_engine_and_client(self): for version, packages in (("12", "docker.io"), ("13", "docker.io docker-cli")): with self.subTest(version=version): result = self.run_docker_setup(version=version) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.splitlines(), [ f"apt-get install -y --no-install-recommends {packages}", "systemctl enable --now docker"]) def test_rerun_repairs_debian_13_missing_client_without_touching_engine(self): result = self.run_docker_setup(engine=True, data=True, config=True) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.splitlines(), [ "apt-get install -y --no-install-recommends --no-remove docker-cli"]) def test_existing_docker_including_ce_is_reused_without_package_or_service_changes(self): for engine in (False, True): with self.subTest(debian_engine=engine): result = self.run_docker_setup(cli=True, engine=engine, data=True, config=True) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, "") def test_unknown_existing_data_or_configuration_still_blocks_engine_installation(self): for data, config in ((True, False), (False, True)): with self.subTest(data=data, config=config): result = self.run_docker_setup(data=data, config=config) self.assertNotEqual(result.returncode, 0) self.assertIn("Existing Docker data/configuration", result.stderr) self.assertEqual(result.stdout, "") def test_unavailable_existing_engine_exposes_docker_error_without_restarting_it(self): result = self.run_docker_setup(cli=True, available=False) self.assertNotEqual(result.returncode, 0) self.assertIn("Cannot connect to the Docker daemon", result.stderr) self.assertIn("journalctl -u docker", result.stderr) self.assertEqual(result.stdout, "") def test_package_failure_stops_before_starting_docker(self): result = self.run_docker_setup(apt_status=42) self.assertEqual(result.returncode, 42) self.assertNotIn("systemctl", result.stdout) def test_missing_client_is_distinguished_from_an_unavailable_daemon(self): result = self.run_docker_setup(supply_cli=False) self.assertNotEqual(result.returncode, 0) self.assertIn("Docker CLI is missing", result.stderr) self.assertNotIn("Docker is installed but unavailable", result.stderr) class ReleaseTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.source = Path(self.temporary.name) / "source" self.target = Path(self.temporary.name) / "release" self.source.mkdir() (self.source / "ec_agent.py").write_text('VERSION = "3.12.7"\nraise RuntimeError("must never execute during packaging")\n') (self.source / "ec_agent_update.py").write_text('raise RuntimeError("must never execute during verification")\n') (self.source / "install.sh").write_text("#!/bin/bash\n") PACKAGER.package(self.source, self.target) def tearDown(self): self.temporary.cleanup() def verify(self): argv = ["-", str(self.target / "manifest.json"), str(self.target / "ec-agent.py"), str(self.target / "ec-agent-update.py")] with patch.object(sys, "argv", argv): exec(compile(RELEASE, "installer-release", "exec"), {}) def manifest(self, **changes): path = self.target / "manifest.json" value = json.loads(path.read_text()) value.update(changes) path.write_text(json.dumps(value)) return value def test_package_derives_version_and_hashes_both_sources_without_executing_them(self): manifest = self.manifest() self.assertEqual(manifest["version"], "3.12.7") self.assertEqual(manifest["updaterProtocolVersion"], 1) self.assertEqual(manifest["sha256"], hashlib.sha256((self.source / "ec_agent.py").read_bytes()).hexdigest()) self.assertEqual(manifest["updaterSha256"], hashlib.sha256((self.source / "ec_agent_update.py").read_bytes()).hexdigest()) self.assertEqual((self.target / "ec-agent-update.py").read_bytes(), (self.source / "ec_agent_update.py").read_bytes()) self.verify() def test_changed_agent_and_updater_downloads_are_rejected(self): for filename, error in (("ec-agent.py", "Agent release checksum"), ("ec-agent-update.py", "Updater release checksum")): with self.subTest(filename=filename): path = self.target / filename original = path.read_bytes() path.write_bytes(original + b"# changed\n") with self.assertRaisesRegex(SystemExit, error): self.verify() path.write_bytes(original) def test_installer_rejects_missing_or_unsupported_updater_protocol(self): for protocol in (None, 0, 2, "1"): self.manifest(updaterProtocolVersion=protocol) with self.subTest(protocol=protocol), self.assertRaisesRegex(SystemExit, "Unsupported agent or updater"): self.verify() def test_installer_rejects_manifest_version_mismatch_and_invalid_code(self): self.manifest(version="3.12.8") with self.assertRaisesRegex(SystemExit, "version does not match"): self.verify() malformed = b"def malformed(:\n" (self.target / "ec-agent-update.py").write_bytes(malformed) self.manifest(version="3.12.7", updaterSha256=hashlib.sha256(malformed).hexdigest()) with self.assertRaises(SyntaxError): self.verify() def test_packaging_requires_one_literal_semantic_agent_version(self): for source in ('VERSION = "bad"', 'VERSION = "01.2.3"', 'VERSION = str(3)', 'VERSION = "1.2.3"\nVERSION = "1.2.4"'): (self.source / "ec_agent.py").write_text(source) with self.subTest(source=source), self.assertRaisesRegex(ValueError, "literal semantic VERSION"): PACKAGER.package(self.source, self.target) def test_installer_serializes_changes_and_enables_update_timer_only_after_agent_starts(self): locked = SOURCE.index("flock --exclusive 9") self.assertLess(SOURCE.index("systemctl stop electric-cherry-update.timer"), locked) self.assertLess(locked, SOURCE.index("installation_preflight", locked)) self.assertLess(SOURCE.index("installation_preflight", locked), SOURCE.index('install -m 0755 "$temporary_dir/ec-agent.py"')) self.assertLess(locked, SOURCE.index("flock --exclusive 8")) self.assertLess(SOURCE.index("flock --exclusive 8"), SOURCE.index("systemctl stop electric-cherry-agent.service")) self.assertLess(SOURCE.index("flock --unlock 8", SOURCE.index("systemctl daemon-reload")), SOURCE.index("systemctl restart electric-cherry-agent")) self.assertLess(SOURCE.index("systemctl is-active --quiet electric-cherry-agent"), SOURCE.index("systemctl enable --now electric-cherry-update.timer")) self.assertIn("ExecStart=/usr/bin/python3 /usr/local/libexec/electric-cherry-update.py", SOURCE) self.assertIn("OnUnitInactiveSec=5min", SOURCE) self.assertIn("RandomizedDelaySec=30s", SOURCE) self.assertIn("AccuracySec=1s", SOURCE) self.assertNotIn("Persistent=true", SOURCE) def test_failed_install_restores_only_previously_active_timer_after_releasing_locks(self): for previous, finished, code in ((True, False, 37), (False, False, 37), (True, True, 0)): with self.subTest(previous=previous, finished=finished): scratch = Path(self.temporary.name) / "scratch" scratch.mkdir(exist_ok=True) log = Path(self.temporary.name) / "cleanup.log" log.unlink(missing_ok=True) script = CLEANUP + r''' flock() { printf 'unlock:%s\n' "$2" >> "$EC_TEST_LOG"; } systemctl() { if { : >&8; } 2>/dev/null || { : >&9; } 2>/dev/null; then printf 'lock-still-open\n' >> "$EC_TEST_LOG" return 1 fi printf '%s\n' "$*" >> "$EC_TEST_LOG" } temporary_dir="$EC_TEST_TEMP" updater_timer_was_active="$EC_TEST_PREVIOUS" installer_finished="$EC_TEST_FINISHED" update_lock_held=true activity_lock_held=true exec 8>"$EC_TEST_LOCK" exec 9>"$EC_TEST_LOCK" trap installation_cleanup EXIT exit "$EC_TEST_EXIT" ''' environment = {**os.environ, "EC_TEST_LOG": str(log), "EC_TEST_TEMP": str(scratch), "EC_TEST_PREVIOUS": str(previous).lower(), "EC_TEST_FINISHED": str(finished).lower(), "EC_TEST_LOCK": str(Path(self.temporary.name) / "test.lock"), "EC_TEST_EXIT": str(code)} result = subprocess.run(["bash", "-c", script], env=environment, capture_output=True) self.assertEqual(result.returncode, code, result.stderr) expected = ["unlock:8", "unlock:9"] if previous and not finished: expected.append("start electric-cherry-update.timer") self.assertEqual(log.read_text().splitlines(), expected) self.assertFalse(scratch.exists()) if __name__ == "__main__": unittest.main()