"""Exercise installer guards without installing software or touching services.""" import json import os from pathlib import Path import re import stat 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) 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_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() if __name__ == "__main__": unittest.main()