# tests/plugins/host_agent/test_maint_helper.py """#5766: the root helper the agent can start, and the agent's side of it. The helper runs as root and reads files an unprivileged agent wrote, so what it refuses matters as much as what it does: a link, a big file, an action not on its list. The agent leaves requests and reports results; Steward serves the helper and the units from one copy. """ import importlib.util import json import os from pathlib import Path import pytest from plugins.host_agent import agent as a FILES = Path(__file__).resolve().parents[3] / "plugins" / "host_agent" / "files" @pytest.fixture def maint(): spec = importlib.util.spec_from_file_location("steward_maint", FILES / "maint.py") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod def _write(path, req): path.write_text(json.dumps(req)) return str(path) def test_the_helper_reads_a_request_naming_one_of_its_actions(maint, tmp_path): for action in ("update", "reboot", "drain", "undrain"): assert maint.read_request(_write(tmp_path / "r.json", {"id": "job-1", "action": action})) node = "abcdefghijklmnopqrstuvwxy" assert maint.read_request(_write(tmp_path / "n.json", {"id": "job-2", "action": "drain", "node": node}))["node"] == node @pytest.mark.parametrize("req", [ {"id": "job-1", "action": "shell"}, # not on the list {"id": "job-1", "action": "update; rm -rf /"}, {"id": "../../etc/x", "action": "update"}, # an id that walks paths {"id": "", "action": "update"}, {"id": "job-1", "action": "drain", "node": "--force"}, # a flag, not a node id ["update"], ]) def test_the_helper_refuses_anything_else(maint, tmp_path, req): assert maint.read_request(_write(tmp_path / "r.json", req)) is None def test_the_helper_never_follows_a_link_the_agent_planted(maint, tmp_path): target = tmp_path / "secret.json" _write(target, {"id": "job-1", "action": "update"}) link = tmp_path / "r.json" link.symlink_to(target) assert maint.read_request(str(link)) is None def test_the_helper_refuses_a_big_file_and_a_non_file(maint, tmp_path): big = tmp_path / "big.json" big.write_text(json.dumps({"id": "job-1", "action": "update", "pad": "x" * 5000})) assert maint.read_request(str(big)) is None (tmp_path / "dir.json").mkdir() assert maint.read_request(str(tmp_path / "dir.json")) is None assert maint.read_request(str(tmp_path / "missing.json")) is None def test_the_helper_counts_what_an_upgrade_changed(maint): assert maint.upgraded_count("12 upgraded, 2 newly installed, 0 to remove and 0 not upgraded.") == 14 dnf = "Transaction Summary\n====\nInstall 1 Package\nUpgrade 23 Packages\n" assert maint.upgraded_count(dnf) == 24 assert maint.upgraded_count("nothing to do") is None def test_a_request_is_deleted_whatever_it_held_and_a_repeat_is_not_run_twice(maint, tmp_path, monkeypatch): requests, results = tmp_path / "req", tmp_path / "res" requests.mkdir() monkeypatch.setattr(maint, "REQUESTS", str(requests)) monkeypatch.setattr(maint, "RESULTS", str(results)) ran = [] monkeypatch.setattr(maint, "do_node", lambda avail, node: ran.append(avail) or {"ok": True}) _write(requests / "bad.json", {"id": "x", "action": "shell"}) _write(requests / "job-1.json", {"id": "job-1", "action": "drain"}) maint.main() assert list(requests.iterdir()) == [] assert json.loads((results / "job-1.json").read_text())["state"] == "done" _write(requests / "job-1.json", {"id": "job-1", "action": "drain"}) maint.main() assert ran == ["drain"] def test_a_reboot_says_so_before_it_goes(maint, tmp_path, monkeypatch): monkeypatch.setattr(maint, "RESULTS", str(tmp_path)) called = [] monkeypatch.setattr(maint, "run", lambda cmd, deadline=0, env=None: called.append(cmd) or (True, "")) monkeypatch.setattr(maint.os, "sync", lambda: None) out = maint.handle({"id": "job-9", "action": "reboot"}) assert out["state"] == "rebooting" assert json.loads((tmp_path / "job-9.json").read_text())["state"] == "rebooting" assert called == [["systemctl", "reboot"]] # ── the agent's side ───────────────────────────────────────────────────────── def test_the_agent_leaves_a_whole_request_the_helper_accepts(maint, tmp_path): a.request_maintenance("job-3", "drain", node="abcdefghijklmnopqrstuvwxy", requests_dir=str(tmp_path)) assert sorted(os.listdir(tmp_path)) == ["job-3.json"] # no temp file left behind assert maint.read_request(str(tmp_path / "job-3.json"))["action"] == "drain" def test_the_agent_refuses_to_ask_for_anything_else(tmp_path): with pytest.raises(ValueError): a.request_maintenance("job-3", "shell", requests_dir=str(tmp_path)) with pytest.raises(ValueError): a.request_maintenance("../x", "update", requests_dir=str(tmp_path)) assert os.listdir(tmp_path) == [] def test_each_result_state_is_reported_once(tmp_path): res = tmp_path / "job-4.json" res.write_text(json.dumps({"id": "job-4", "action": "update", "state": "running"})) seen: set = set() events, keys = a.maintenance_results(seen, results_dir=str(tmp_path)) assert [e["state"] for e in events] == ["running"] and events[0]["kind"] == "maintenance" assert a.maintenance_results(seen, results_dir=str(tmp_path))[0] # not posted yet: said again seen.update(keys) assert a.maintenance_results(seen, results_dir=str(tmp_path)) == ([], []) res.write_text(json.dumps({"id": "job-4", "action": "update", "state": "done", "ok": True, "packages": 12, "reboot_required": True})) events, _ = a.maintenance_results(seen, results_dir=str(tmp_path)) assert events == [{"kind": "maintenance", "id": "job-4", "action": "update", "state": "done", "ok": True, "packages": 12, "reboot_required": True}] def test_an_old_or_broken_result_is_not_reported(tmp_path): old = tmp_path / "job-5.json" old.write_text(json.dumps({"id": "job-5", "action": "update", "state": "done"})) week_ago = os.stat(old).st_mtime - 7 * 86400 os.utime(old, (week_ago, week_ago)) (tmp_path / "job-6.json").write_text("{not json") (tmp_path / "job-7.json").write_text(json.dumps({"id": "../x", "state": "done"})) assert a.maintenance_results(set(), results_dir=str(tmp_path)) == ([], []) assert a.maintenance_results(set(), results_dir=str(tmp_path / "missing")) == ([], []) def test_the_agent_says_whether_the_helper_is_installed(tmp_path): unit, requests = tmp_path / "steward-maint.path", tmp_path / "requests" assert a.maint_helper_installed(str(unit), str(requests)) is False unit.write_text("[Path]\n") requests.mkdir() assert a.maint_helper_installed(str(unit), str(requests)) is True # ── what Steward serves ────────────────────────────────────────────────────── def test_the_helper_and_the_agent_agree_on_paths_and_actions(maint): assert a.MAINT_REQUESTS == maint.REQUESTS and a.MAINT_RESULTS == maint.RESULTS assert a.MAINT_ACTIONS == maint.ACTIONS path_unit = (FILES / "steward-maint.path").read_text() assert f"PathExistsGlob={maint.REQUESTS}/*.json" in path_unit async def test_steward_serves_only_the_listed_files(app): from plugins.host_agent.routes import AGENT_FILES, agent_file assert sorted(AGENT_FILES) == sorted(p.name for p in FILES.iterdir() if p.is_file()) async with app.test_request_context("/plugins/host_agent/files/maint.py"): ok = await agent_file("maint.py") assert ok.status_code == 200 assert "ACTIONS = (" in await ok.get_data(as_text=True) for name in ("../agent.py", "agent.py", "routes.py", "../../steward/config.py"): _, status = await agent_file(name) assert status == 404, name # ── pacman, zypper and yum hosts (#5804) ───────────────────────────────────── def test_the_helper_counts_what_zypper_and_pacman_changed(maint): zypper = ("The following 15 packages are going to be upgraded:\n bash vim ...\n" "15 packages to upgrade, 2 new.\nOverall download size: 48.2 MiB.\n") assert maint.upgraded_count(zypper) == 17 pacman = ("resolving dependencies...\nlooking for conflicting packages...\n\n" "Packages (12) linux-6.11.2.arch1-1 openssl-3.4.0-1 ...\n\n" "Total Download Size: 152.31 MiB\n") assert maint.upgraded_count(pacman) == 12 yum = "Transaction Summary\n====\nUpgrade 4 Packages\n" assert maint.upgraded_count(yum) == 4 def _manager(monkeypatch, maint, binaries): monkeypatch.setattr(maint.shutil, "which", lambda name: f"/usr/bin/{name}" if name in binaries else None) @pytest.mark.parametrize("binaries, manager", [ (("apt-get",), "apt"), (("dnf", "yum"), "dnf"), (("yum",), "yum"), (("zypper",), "zypper"), (("pacman",), "pacman"), ((), None)]) def test_the_helper_picks_the_hosts_package_manager(maint, monkeypatch, binaries, manager): _manager(monkeypatch, maint, binaries) assert maint.host_manager() == manager def test_each_manager_upgrades_without_asking(maint, monkeypatch): commands = {m: [step[1] for step in maint.update_steps(m)] for m in ("apt", "dnf", "yum", "pacman")} assert commands["pacman"] == [["pacman", "-Syu", "--noconfirm"]] assert commands["yum"] == [["yum", "-y", "update"]] assert commands["dnf"] == [["dnf", "-y", "upgrade", "--refresh"]] assert commands["apt"][-1][-1] == "dist-upgrade" monkeypatch.setattr(maint, "_os_release_id", lambda path="/etc/os-release": "opensuse-leap") leap = [step[1] for step in maint.update_steps("zypper")] assert leap == [["zypper", "--non-interactive", "refresh"], ["zypper", "--non-interactive", "update", "--auto-agree-with-licenses"]] monkeypatch.setattr(maint, "_os_release_id", lambda path="/etc/os-release": "opensuse-tumbleweed") assert maint.update_steps("zypper")[-1][1][2] == "dup" # a rolling release upgrades with dup def test_a_zypper_update_that_asks_for_a_reboot_still_worked(maint, monkeypatch): """zypper exits 102 when the update needs a reboot: done, not failed.""" _manager(monkeypatch, maint, ("zypper",)) monkeypatch.setattr(maint, "_os_release_id", lambda path="/etc/os-release": "sles") ran = [] def fake_run(cmd, deadline=0, env=None, ok_codes=(0,)): ran.append(cmd[2]) code = 102 if cmd[2] == "update" else 0 return code in ok_codes, "3 packages to upgrade.\n" monkeypatch.setattr(maint, "run", fake_run) monkeypatch.setattr(maint, "reboot_required", lambda: True) got = maint.do_update() assert got["ok"] and got["packages"] == 3 and got["reboot_required"] is True assert ran == ["refresh", "update"] def test_a_failed_step_says_which(maint, monkeypatch): _manager(monkeypatch, maint, ("pacman",)) monkeypatch.setattr(maint, "run", lambda cmd, **kw: (False, "error: failed to synchronize all databases")) got = maint.do_update() assert not got["ok"] and got["detail"].startswith("pacman -Syu failed: error: failed to synchronize") def test_a_host_with_no_package_manager_it_knows_says_so(maint, monkeypatch): _manager(monkeypatch, maint, ()) assert "apt, dnf, yum, zypper, pacman" in maint.do_update()["detail"] def test_a_pacman_host_needs_a_reboot_once_the_running_kernel_is_replaced(maint, monkeypatch, tmp_path): _manager(monkeypatch, maint, ("pacman",)) monkeypatch.setattr(maint, "REBOOT_MARKER", str(tmp_path / "reboot-required")) # none monkeypatch.setattr(maint, "KERNEL_MODULES", str(tmp_path)) (tmp_path / os.uname().release).mkdir() assert maint.reboot_required() is False (tmp_path / os.uname().release).rmdir() assert maint.reboot_required() is True # ── check: what an update would change (#6098) ────────────────────────────── APT_SIM = """Reading package lists... Calculating upgrade... The following packages will be upgraded: containerd.io docker-ce docker-ce-cli openssl Inst docker-ce [5:27.0.3-1~debian.12~bookworm] (5:27.1.1-1~debian.12~bookworm Docker CE:bookworm [amd64]) Inst docker-ce-cli [5:27.0.3-1] (5:27.1.1-1 Docker CE:bookworm [amd64]) Inst openssl [3.0.13-1] (3.0.14-1 Debian-Security:12/stable-security [amd64]) Conf docker-ce (5:27.1.1-1~debian.12~bookworm Docker CE:bookworm [amd64]) """ def test_a_check_reads_what_each_manager_would_upgrade(maint): assert maint.pending_names("apt", APT_SIM) == ["docker-ce", "docker-ce-cli", "openssl"] dnf = ("\nmoby-engine.x86_64 26.1.3-1.fc40 updates\n" "kernel-core.x86_64 6.10.3-200.fc40 updates\n" "Obsoleting Packages\n") assert maint.pending_names("dnf", dnf) == ["moby-engine", "kernel-core"] table = ("S | Repository | Name | Current Version | Available Version | Arch\n" "--+------------+--------+-----------------+-------------------+-------\n" "v | repo-oss | docker | 24.0.7 | 25.0.3 | x86_64\n" "v | repo-oss | vim | 9.0 | 9.1 | x86_64\n") assert maint.pending_names("zypper", table) == ["docker", "vim"] dup = ("The following 2 packages are going to be upgraded:\n" " containerd runc\n\nThe following package is going to be installed:\n foo\n" "2 packages to upgrade, 1 new.\n") assert maint.pending_names("zypper", dup) == ["containerd", "runc", "foo"] assert maint.pending_names("pacman", "docker 1:27.0-1 -> 1:27.1-1\nlinux 6.10-1 -> 6.10.3-1\n") == [ "docker", "linux"] @pytest.mark.parametrize("name, restarts_docker", [ ("docker-ce", True), ("docker.io", True), ("containerd.io", True), ("moby-engine", True), ("docker", True), ("containerd", True), ("docker-ce-cli", False), ("docker-compose-plugin", False), ("docker-buildx-plugin", False), ("runc", False), ("python3-docker", False)]) def test_only_the_packages_that_restart_the_daemon_count_as_docker(maint, name, restarts_docker): assert bool(maint.DOCKER_PACKAGES.match(name)) is restarts_docker def test_the_check_names_the_containers_outside_the_swarm(maint, monkeypatch): monkeypatch.setattr(maint.shutil, "which", lambda b: "/usr/bin/docker") inspected = ("/ak-outpost|unless-stopped|\n" "/traefik_router.abc.def|no|t0k3n\n" # a swarm task: the drain moves it "/scratch||\n") def fake_run(cmd, deadline=0, env=None, ok_codes=(0,), keep=0): return True, ("id1 id2 id3\n" if cmd[1] == "ps" else inspected) monkeypatch.setattr(maint, "run", fake_run) assert maint.standalone_containers() == [{"name": "ak-outpost", "restart": "unless-stopped"}, {"name": "scratch", "restart": "no"}] def test_a_check_says_whether_docker_is_upgraded_and_changes_nothing(maint, monkeypatch): _manager(monkeypatch, maint, ("apt-get",)) ran = [] def fake_run(cmd, deadline=0, env=None, ok_codes=(0,), keep=0): ran.append(cmd) return True, APT_SIM if "-s" in cmd else "Hit:1 http://deb.debian.org bookworm InRelease\n" monkeypatch.setattr(maint, "run", fake_run) monkeypatch.setattr(maint, "standalone_containers", lambda: [{"name": "ak-outpost", "restart": "always"}]) got = maint.do_check() assert got == {"ok": True, "pending": 3, "docker_update": True, "docker_packages": ["docker-ce"], "containers": [{"name": "ak-outpost", "restart": "always"}]} assert ran == [["apt-get", "update"], ["apt-get", "-s", "dist-upgrade"]] # a simulation only assert "check" in maint.ACTIONS def test_a_check_that_cant_list_says_which_step(maint, monkeypatch): _manager(monkeypatch, maint, ("dnf",)) monkeypatch.setattr(maint, "run", lambda cmd, **kw: (False, "Curl error (6): Couldn't resolve host")) got = maint.do_check() assert not got["ok"] and got["detail"].startswith("dnf check-update failed: Curl error")