351 lines
18 KiB
Python
351 lines
18 KiB
Python
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import stat
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location("ec_uninstall_agent", Path(__file__).resolve().parents[1] / "ec_agent.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
INSTANCE = "a53d99da-a1eb-4f01-9b10-a6268b556542"
|
|
|
|
|
|
class CleanupDocker:
|
|
def __init__(self):
|
|
self.calls = []
|
|
self.items = {}
|
|
self.images = {}
|
|
|
|
def inspect(self, identity):
|
|
return self.items.get(identity)
|
|
|
|
def call(self, *args, **kwargs):
|
|
self.calls.append(args)
|
|
if args[:2] == ("container", "ls"):
|
|
return "\n".join(self.items)
|
|
if args[:2] == ("container", "rm"):
|
|
self.items.pop(args[-1], None)
|
|
return ""
|
|
if args[:2] == ("image", "inspect"):
|
|
item = self.images.get(args[-1])
|
|
return json.dumps([item]) if item else None
|
|
if args[:2] == ("image", "rm"):
|
|
self.images.pop(args[-1], None)
|
|
return ""
|
|
raise AssertionError("Unexpected Docker request: " + str(args))
|
|
|
|
|
|
class UninstallTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.base = Path(self.temp.name).resolve()
|
|
self.root = self.base / "host-data"
|
|
self.config_dir = self.base / "etc" / "electric-cherry"
|
|
self.bin_dir = self.base / "libexec"
|
|
self.unit_dir = self.base / "systemd"
|
|
for directory in (self.config_dir, self.bin_dir, self.unit_dir):
|
|
directory.mkdir(parents=True)
|
|
replacements = {
|
|
"AGENT_CONFIG": self.config_dir / "agent.json",
|
|
"AGENT_EXECUTABLE": self.bin_dir / "electric-cherry-agent.py",
|
|
"AGENT_UNIT": self.unit_dir / "electric-cherry-agent.service",
|
|
"UNINSTALL_EXECUTABLE": self.bin_dir / "electric-cherry-uninstall.py",
|
|
"UNINSTALL_UNIT": self.unit_dir / "electric-cherry-uninstall.service",
|
|
"UNINSTALL_REQUEST": self.config_dir / "decommission.json",
|
|
"UPDATE_EXECUTABLE": self.bin_dir / "electric-cherry-update.py",
|
|
"UPDATE_UNIT": self.unit_dir / "electric-cherry-update.service",
|
|
"UPDATE_TIMER": self.unit_dir / "electric-cherry-update.timer",
|
|
}
|
|
self.patches = [patch.object(module, key, value) for key, value in replacements.items()]
|
|
# Unit tests run without root privileges. Keep actual path, symlink,
|
|
# file-type and containment checks while substituting owner checks.
|
|
def test_protected_path(value, missing=False):
|
|
path = Path(value)
|
|
if any(current.is_symlink() for current in (path, *path.parents)):
|
|
raise module.AgentError("Symlink ownership is invalid")
|
|
if not missing and not path.exists():
|
|
raise module.AgentError("Host uninstall files are unavailable")
|
|
return path
|
|
self.patches.extend([patch.object(module, "protected_path", side_effect=test_protected_path), patch.object(module, "systemctl"), patch.object(module, "sync_parent")])
|
|
for entry in self.patches:
|
|
entry.start()
|
|
self.config = {"hostId": "host-one", "controller": "https://controller.example", "credential": "secret", "root": str(self.root), "configPath": str(module.AGENT_CONFIG)}
|
|
self.agent = module.Agent(self.config)
|
|
self.docker = CleanupDocker()
|
|
for path in (module.AGENT_EXECUTABLE, module.AGENT_UNIT, module.UNINSTALL_EXECUTABLE, module.UNINSTALL_UNIT, module.UPDATE_EXECUTABLE, module.UPDATE_UNIT, module.UPDATE_TIMER):
|
|
path.write_text("installed")
|
|
for path in (module.AGENT_EXECUTABLE, module.UNINSTALL_EXECUTABLE):
|
|
path.with_suffix(".py.update").write_text("interrupted update")
|
|
module.atomic_json(module.AGENT_CONFIG, self.config)
|
|
module.atomic_json(self.root / module.ROOT_MARKER, {"version": 1, "root": str(self.root), "hostId": "host-one", "controller": "https://controller.example"})
|
|
self.game = self.root / "libraries" / "t6" / "library" / "content" / "zone" / "all" / "dlc.ff"
|
|
self.game.parent.mkdir(parents=True)
|
|
self.game.write_bytes(b"game DLC")
|
|
(self.base / "unrelated.txt").write_text("keep")
|
|
self.operation = {"id": "delete-one", "kind": "host.decommission", "payload": {"hostId": "host-one", "operationId": "delete-one", "receiptToken": "receipt-only-secret", "receiptUrl": "/api/agent/decommission/delete-one"}}
|
|
|
|
def tearDown(self):
|
|
self.agent.journal.close()
|
|
for entry in reversed(self.patches):
|
|
entry.stop()
|
|
self.temp.cleanup()
|
|
|
|
def prepare(self):
|
|
result = self.agent.execute(self.operation)
|
|
self.assertEqual(result["state"], "running", result)
|
|
return module.Uninstaller(module.UNINSTALL_REQUEST, self.docker)
|
|
|
|
def container(self, identity, owned=True, host="host-one", root=None, image="image-one"):
|
|
labels = {module.LABEL + ".managed": "true", module.LABEL + ".instance": INSTANCE, module.LABEL + ".generation": "1", module.LABEL + ".host": host} if owned else {}
|
|
return {"Id": identity, "Image": image, "Config": {"Labels": labels}, "State": {"Running": False, "Status": "exited"}, "Mounts": [{"Type": "bind", "Source": str((root or self.root) / "instances" / INSTANCE / "1" / "app"), "Destination": "/home/plutainer/app"}]}
|
|
|
|
def test_final_receipt_follows_data_and_self_removal(self):
|
|
worker = self.prepare()
|
|
stages = []
|
|
def receipt(stage, error=None):
|
|
stages.append(stage)
|
|
self.assertFalse(self.root.exists())
|
|
self.assertFalse(module.AGENT_CONFIG.exists())
|
|
self.assertFalse(module.AGENT_EXECUTABLE.exists())
|
|
self.assertFalse(module.AGENT_UNIT.exists())
|
|
self.assertFalse(module.UPDATE_EXECUTABLE.exists())
|
|
self.assertFalse(module.UPDATE_UNIT.exists())
|
|
self.assertFalse(module.UPDATE_TIMER.exists())
|
|
self.assertFalse(module.AGENT_EXECUTABLE.with_suffix(".py.update").exists())
|
|
self.assertFalse(module.UNINSTALL_EXECUTABLE.with_suffix(".py.update").exists())
|
|
if stage == "cleaned":
|
|
self.assertTrue(module.UNINSTALL_REQUEST.exists())
|
|
self.assertTrue(module.UNINSTALL_EXECUTABLE.exists())
|
|
else:
|
|
self.assertFalse(self.config_dir.exists())
|
|
self.assertFalse(module.UNINSTALL_EXECUTABLE.exists())
|
|
self.assertFalse(module.UNINSTALL_UNIT.exists())
|
|
worker.receipt = receipt
|
|
worker.run()
|
|
self.assertEqual(stages, ["cleaned", "completed"])
|
|
self.assertEqual((self.base / "unrelated.txt").read_text(), "keep")
|
|
module.systemctl.assert_any_call("disable", "--now", "electric-cherry-agent.service")
|
|
module.systemctl.assert_any_call("disable", "--now", "electric-cherry-update.timer")
|
|
module.systemctl.assert_any_call("stop", "electric-cherry-update.service")
|
|
module.systemctl.assert_any_call("disable", "electric-cherry-uninstall.service")
|
|
|
|
def test_handoff_is_durable_and_never_reports_operation_completed(self):
|
|
worker = self.prepare()
|
|
self.assertEqual(worker.job["operationId"], "delete-one")
|
|
self.assertNotIn("credential", worker.job)
|
|
self.assertTrue(self.agent.stopping.is_set())
|
|
module.systemctl.assert_any_call("enable", "--now", "--no-block", "electric-cherry-uninstall.service")
|
|
self.assertEqual(self.agent.journal.get("delete-one")["state"], "running")
|
|
|
|
def test_failed_checkpoint_cannot_leave_a_runnable_uninstall_request(self):
|
|
with patch.object(self.agent.journal, "checkpoint", side_effect=OSError("journal unavailable")):
|
|
result = self.agent.execute(self.operation)
|
|
self.assertEqual(result["state"], "failed")
|
|
self.assertFalse(module.UNINSTALL_REQUEST.exists())
|
|
module.systemctl.assert_not_called()
|
|
|
|
def test_pending_cleanup_resumes_after_controller_unavailable(self):
|
|
worker = self.prepare()
|
|
worker.receipt = lambda *_: (_ for _ in ()).throw(module.AgentError("offline"))
|
|
with self.assertRaisesRegex(module.AgentError, "offline"):
|
|
worker.run()
|
|
self.assertFalse(self.root.exists())
|
|
self.assertTrue(module.UNINSTALL_REQUEST.exists())
|
|
self.assertTrue(module.UNINSTALL_EXECUTABLE.exists())
|
|
self.assertTrue(json.loads(module.UNINSTALL_REQUEST.read_text())["dataRemoved"])
|
|
before = len(self.docker.calls)
|
|
resumed = module.Uninstaller(module.UNINSTALL_REQUEST, self.docker)
|
|
stages = []
|
|
resumed.receipt = lambda stage, error=None: stages.append(stage)
|
|
resumed.run()
|
|
self.assertEqual(stages, ["cleaned", "completed"])
|
|
self.assertEqual(len(self.docker.calls), before)
|
|
|
|
def test_lost_final_receipt_retries_after_all_files_are_gone(self):
|
|
worker = self.prepare()
|
|
stages = []
|
|
def receipt(stage, error=None):
|
|
stages.append(stage)
|
|
if stage == "completed":
|
|
self.assertFalse(self.config_dir.exists())
|
|
if stages.count("completed") == 1:
|
|
raise module.AgentError("lost reply")
|
|
worker.receipt = receipt
|
|
with patch.object(module.time, "sleep") as wait:
|
|
worker.run()
|
|
self.assertEqual(stages, ["cleaned", "completed", "completed"])
|
|
wait.assert_called_once_with(15)
|
|
|
|
def test_partial_self_removal_retries_without_its_unit_or_executable(self):
|
|
worker = self.prepare()
|
|
stages = []
|
|
worker.receipt = lambda stage, error=None: stages.append(stage)
|
|
original = shutil.rmtree
|
|
attempts = 0
|
|
def interrupted(path, *args, **kwargs):
|
|
nonlocal attempts
|
|
if Path(path) == self.config_dir:
|
|
attempts += 1
|
|
if attempts == 1:
|
|
self.assertFalse(module.UNINSTALL_UNIT.exists())
|
|
self.assertFalse(module.UNINSTALL_EXECUTABLE.exists())
|
|
raise OSError("temporarily busy")
|
|
return original(path, *args, **kwargs)
|
|
with patch.object(module.shutil, "rmtree", side_effect=interrupted), patch.object(module.time, "sleep") as wait:
|
|
worker.run()
|
|
self.assertEqual(stages, ["cleaned", "failed", "completed"])
|
|
self.assertFalse(self.config_dir.exists())
|
|
wait.assert_called_once_with(15)
|
|
|
|
def test_partial_tree_removal_recovers_without_deleted_marker(self):
|
|
worker = self.prepare()
|
|
original = shutil.rmtree
|
|
def interrupted(path, *args, **kwargs):
|
|
if Path(path) == self.root:
|
|
(self.root / module.ROOT_MARKER).unlink()
|
|
self.game.unlink()
|
|
raise OSError("power loss")
|
|
return original(path, *args, **kwargs)
|
|
worker.receipt = lambda *_: None
|
|
with patch.object(module.shutil, "rmtree", side_effect=interrupted):
|
|
with self.assertRaisesRegex(OSError, "power loss"):
|
|
worker.run()
|
|
self.assertTrue(json.loads(module.UNINSTALL_REQUEST.read_text())["removalStarted"])
|
|
self.assertFalse((self.root / module.ROOT_MARKER).exists())
|
|
resumed = module.Uninstaller(module.UNINSTALL_REQUEST, self.docker)
|
|
resumed.receipt = lambda *_: None
|
|
resumed.run()
|
|
self.assertFalse(self.root.exists())
|
|
|
|
def test_replaced_data_root_cannot_be_deleted_on_resume(self):
|
|
self.prepare()
|
|
self.root.rename(self.base / "original-data")
|
|
self.root.mkdir()
|
|
(self.root / "precious.txt").write_text("keep")
|
|
with self.assertRaisesRegex(module.AgentError, "changed"):
|
|
module.Uninstaller(module.UNINSTALL_REQUEST, self.docker)
|
|
self.assertTrue((self.root / "precious.txt").exists())
|
|
|
|
def test_only_owned_containers_and_unused_recorded_images_are_removed(self):
|
|
aliases = {"example/runtime:one": "image-one", "example/runtime:shared": "image-shared", "example/runtime:changed": "image-old"}
|
|
module.atomic_json(self.root / "agent" / "runtime-images.json", aliases)
|
|
self.docker.images = {"example/runtime:one": {"Id": "image-one"}, "example/runtime:shared": {"Id": "image-shared"}, "example/runtime:changed": {"Id": "image-new"}, "example/unrelated:one": {"Id": "image-unrelated"}}
|
|
self.docker.items = {"ours": self.container("ours"), "other": self.container("other", host="host-two", root=self.base / "different-install", image="image-shared")}
|
|
worker = self.prepare()
|
|
worker.receipt = lambda *_: None
|
|
worker.run()
|
|
self.assertNotIn("ours", self.docker.items)
|
|
self.assertIn("other", self.docker.items)
|
|
self.assertNotIn("example/runtime:one", self.docker.images)
|
|
self.assertIn("example/runtime:shared", self.docker.images)
|
|
self.assertIn("example/runtime:changed", self.docker.images)
|
|
self.assertIn("example/unrelated:one", self.docker.images)
|
|
self.assertFalse(any("prune" in call or "volume" in call for call in self.docker.calls))
|
|
|
|
def test_unrelated_container_using_root_blocks_any_removal(self):
|
|
self.docker.items = {"foreign": self.container("foreign", owned=False)}
|
|
worker = self.prepare()
|
|
stages = []
|
|
worker.receipt = lambda stage, error=None: stages.append((stage, error))
|
|
with self.assertRaisesRegex(module.AgentError, "unrelated container"):
|
|
worker.run()
|
|
self.assertEqual(stages[0][0], "failed")
|
|
self.assertTrue(self.game.exists())
|
|
self.assertTrue(module.AGENT_EXECUTABLE.exists())
|
|
self.assertIn("foreign", self.docker.items)
|
|
|
|
def test_running_or_unknown_managed_container_blocks_removal(self):
|
|
worker = self.prepare()
|
|
worker.receipt = lambda *_: None
|
|
for state in ({"Running": True, "Status": "running"}, {"Running": False, "Status": "restarting"}, {}):
|
|
with self.subTest(state=state):
|
|
self.docker.items = {"ours": {**self.container("ours"), "State": state}}
|
|
with self.assertRaisesRegex(module.AgentError, "still running or its state is unknown"):
|
|
worker.run()
|
|
self.assertTrue(self.game.exists())
|
|
self.assertIn("ours", self.docker.items)
|
|
self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls))
|
|
|
|
def test_legacy_container_requires_exact_managed_bind_mount(self):
|
|
worker = self.prepare()
|
|
legacy = self.container("legacy")
|
|
legacy["Config"]["Labels"].pop(module.LABEL + ".host")
|
|
self.assertTrue(worker.owned_container(legacy))
|
|
legacy["Mounts"][0]["Source"] = str(self.base / "elsewhere")
|
|
self.assertFalse(worker.owned_container(legacy))
|
|
|
|
def test_missing_marker_and_wrong_host_never_schedule_helper(self):
|
|
(self.root / module.ROOT_MARKER).unlink()
|
|
self.assertFalse(self.agent.can_uninstall())
|
|
result = self.agent.execute(self.operation)
|
|
self.assertEqual(result["state"], "failed")
|
|
module.systemctl.assert_not_called()
|
|
self.assertTrue(self.game.exists())
|
|
|
|
def test_receipt_path_and_identity_cannot_be_redirected(self):
|
|
for key, value in (("receiptUrl", "https://attacker.example"), ("hostId", "other"), ("operationId", "other")):
|
|
with self.subTest(key=key):
|
|
operation = {**self.operation, "id": "check-" + key, "payload": {**self.operation["payload"], "operationId": "check-" + key, "receiptUrl": "/api/agent/decommission/check-" + key, key: value}}
|
|
self.assertEqual(self.agent.execute(operation)["state"], "failed")
|
|
module.systemctl.assert_not_called()
|
|
|
|
def test_symlink_inside_owned_tree_is_unlinked_without_following(self):
|
|
outside = self.base / "unrelated.txt"
|
|
(self.root / "libraries" / "symlink").symlink_to(outside)
|
|
worker = self.prepare()
|
|
worker.receipt = lambda *_: None
|
|
worker.run()
|
|
self.assertEqual(outside.read_text(), "keep")
|
|
|
|
def test_unsafe_roots_and_installation_overlap_are_rejected(self):
|
|
for value in ("/", "/var/lib", "/srv", "/var/lib/../lib", "/var//lib/anything", str(self.config_dir), str(self.bin_dir)):
|
|
with self.subTest(value=value):
|
|
with self.assertRaises(module.AgentError):
|
|
module.uninstall_root(value)
|
|
|
|
def test_root_mount_or_nested_mount_cannot_be_recursively_deleted(self):
|
|
original = Path.read_text
|
|
for mount in (self.root, self.root / "libraries"):
|
|
def read_text(path, *args, **kwargs):
|
|
if str(path) == "/proc/1/mountinfo":
|
|
return "21 1 8:1 / " + str(mount) + " rw - ext4 /dev/sda1 rw\n"
|
|
return original(path, *args, **kwargs)
|
|
with patch.object(Path, "read_text", read_text):
|
|
with self.assertRaisesRegex(module.AgentError, "Unmount"):
|
|
module.uninstall_root(str(self.root))
|
|
|
|
def test_marker_directory_is_rejected_before_reading(self):
|
|
marker = self.root / module.ROOT_MARKER
|
|
marker.unlink()
|
|
marker.mkdir()
|
|
with self.assertRaisesRegex(module.AgentError, "regular file"):
|
|
module.uninstall_ownership(self.config)
|
|
|
|
def test_boot_reconciles_handoff_instead_of_marking_complete(self):
|
|
self.prepare()
|
|
self.agent.journal.close()
|
|
self.agent = module.Agent(self.config)
|
|
self.assertEqual(self.agent.journal.get("delete-one")["state"], "unknown")
|
|
result = self.agent.execute(self.operation)
|
|
self.assertEqual(result["state"], "running")
|
|
self.assertEqual(json.loads(module.UNINSTALL_REQUEST.read_text())["operationId"], "delete-one")
|
|
|
|
|
|
class OwnershipPathTests(unittest.TestCase):
|
|
def test_root_ownership_group_writes_and_symlinks_are_rejected(self):
|
|
for mode, uid in ((stat.S_IFDIR | 0o777, 0), (stat.S_IFDIR | 0o755, 1000), (stat.S_IFLNK | 0o777, 0)):
|
|
with self.subTest(mode=mode, uid=uid):
|
|
value = os.stat_result((mode, 1, 1, 1, uid, 0, 0, 0, 0, 0))
|
|
with patch.object(Path, "lstat", return_value=value):
|
|
with self.assertRaisesRegex(module.AgentError, "only by root"):
|
|
module.protected_path("/var/lib/electric-cherry")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|