#!/usr/bin/env python3 """Tests for configs/felhom-os-apply (`11` §5.4.1). A fake runner plays the host and the guest: nothing is executed for real except the local `dpkg --compare-versions` (pure, no network). Every refusal R1–R13 has a test; the red-proof (each test fails when its rule is removed) is `audits/os-guest-lane-2026-10-04/partB/redproof.txt`. Run: python3 configs/test_felhom_os_apply.py (also run by internal/osupdate's Go test) """ import importlib.machinery import importlib.util import json import os import pathlib import re import stat as statmod import subprocess import unittest HERE = pathlib.Path(__file__).resolve().parent _loader = importlib.machinery.SourceFileLoader("osapply", os.environ.get("OSAPPLY_UNDER_TEST", str(HERE / "felhom-os-apply"))) # red-proof seam _spec = importlib.util.spec_from_loader("osapply", _loader) osapply = importlib.util.module_from_spec(_spec) _loader.exec_module(osapply) PLAN = "/var/lib/felhom-agent/os/plan-t1.json" CONF_OK = ("arch: amd64\nmp0: local-lvm:vm-9201-disk-1,mp=/var/lib/felhom,backup=1,size=70G\n" "mp8: /mnt/felhom-drives,mp=/mnt/felhom-drives\nrootfs: local-lvm:vm-9201-disk-0,size=32G\n") DEB = "Debian:13.7/stable" SEC = "Debian-Security:13/stable-security" def dpkg_cmp(a, op, b): return subprocess.run(["dpkg", "--compare-versions", a, op, b]).returncode == 0 class St: def __init__(self, mode=statmod.S_IFREG | 0o600, uid=999, size=100): self.st_mode, self.st_uid, self.st_size = mode, uid, size class Fake: """The host + one guest. `installed` / `live` (name -> versions in the live archive) / `snapshot` (versions the snapshot archive adds) / `extra_sim` (lines the simulation adds) / `dpkg_audit` / `free`.""" def __init__(self): self.plan = {"release_id": "os-t1", "layer": "guest", "lane": "fast", "vmid": 9201, "mode": "apply", "snapshot": "20261004T080000Z", "packages": [{"name": "libc6", "version": "2.41-12+deb13u4", "origin": "Debian"}, {"name": "openssl", "version": "3.5.7-1~deb13u3", "origin": "Debian-Security"}]} self.files = {"/etc/pve/lxc/9201.conf": CONF_OK} self.stats = {PLAN: St()} self.installed = {"libc6": "2.41-12+deb13u3", "openssl": "3.5.6-1~deb13u1", "bash": "5.2.37-2+b9"} self.live = {"libc6": {"2.41-12+deb13u4"}, "openssl": {"3.5.7-1~deb13u3"}} self.snapshot = {} self.snap_active = False self.extra_sim = [] self.dpkg_audit = "" self.free = 10 * 1024 ** 3 self.install_rc = 0 self.calls = [] self.logs = [] self.written = {} self.status = "status: running" self.lock_held = False self.services = {} self.files[osapply.INSTALL_STATE] = json.dumps({"mode": "appliance"}) self.stats[osapply.INSTALL_STATE] = St(mode=statmod.S_IFREG | 0o644, uid=0) # Docker / trust / facts state (v0.142.0) self.live_restore = "true" self.ids = ["aaa111", "bbb222"] self.engine = "29.7.2" self.daemon_json = '{"log-driver": "json-file"}' self.reload_enables = True self.sig_rc = 0 self.nonces = {} self.clock = 1791115200.0 # 2026-10-04T12:00:00Z self.saved_reports = [] # R-868: (plan path, report) the wrapper kept on disk self.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": False}) self.stats[osapply.TRUST_FILE] = St(mode=statmod.S_IFREG | 0o644, uid=0) self.files[osapply.TRUST_SIGNERS] = 'felhom-op-1 namespaces="felhom-op-v1" ssh-ed25519 AAAA\n' self.stats[osapply.TRUST_SIGNERS] = St(mode=statmod.S_IFREG | 0o644, uid=0) def now(self): return self.clock def sleep(self, s): self.sleeps = getattr(self, "sleeps", []) + [(len(self.calls), s)] # (calls made before it, seconds) def verify_sig(self, signers, key_id, ns, blob, sig): self.verified = (signers, key_id, ns, blob, sig) return self.sig_rc def read_nonces(self): return dict(self.nonces) def write_nonces(self, d): self.nonces = dict(d) # Runner interface def read_file(self, p): if p == PLAN: return json.dumps(self.plan) if p not in self.files: raise OSError("no such file") return self.files[p] def stat(self, p): if p not in self.stats: raise OSError("no such file") return self.stats[p] def agent_uid(self): return 999 def log(self, line): self.logs.append(line) def save_report(self, plan_path, report): self.saved_reports.append((plan_path, json.loads(json.dumps(report)))) return plan_path.replace("/plan-", "/report-") def host(self, argv, timeout=600, stdin=None): self.calls.append(("host", argv)) if argv[0] == "/usr/sbin/pct" and argv[1] == "status": return 0, self.status + "\n", "" if argv[0] == "dpkg" and argv[1] == "--compare-versions": return (0 if dpkg_cmp(argv[2], argv[3], argv[4]) else 1), "", "" if argv[0] == "systemctl" and argv[1] == "is-active": return 0, "\n".join(self.services.get(s, "active") for s in argv[2:]) + "\n", "" return self.emulate(argv) def write_file(self, layer, vmid, path, body): self.written[path] = body self.write_layer = layer if path == osapply.SNAPSHOT_LIST: self.snap_active = True def avail(self, n): v = set(self.live.get(n, set())) if self.snap_active: v |= self.snapshot.get(n, set()) return v def guest(self, vmid, argv, timeout=1800): self.calls.append(("guest", vmid, argv)) return self.emulate(argv) def emulate(self, argv): a = [x for x in argv if not re.match(r"^[A-Z_]+=", x) and x != "env"] cmd = a[0] if cmd == "dpkg-query": return 0, "".join(f"{n}\t{v}\tii \n" for n, v in self.installed.items()), "" if cmd == "dpkg" and a[1] == "--compare-versions": return (0 if dpkg_cmp(a[2], a[3], a[4]) else 1), "", "" if cmd == "dpkg" and a[1] == "--audit": return 0, self.dpkg_audit, "" if cmd == "dpkg" and a[1] == "--configure": self.configured_calls = getattr(self, "configured_calls", 0) + 1 self.dpkg_journal = [] # `dpkg --configure -a` replays and empties the update journal return 0, "", "" if cmd == "fuser": return (0, " 123", "") if self.lock_held else (1, "", "") if cmd == "apt-cache" and a[1] == "madison": self.madison_calls = getattr(self, "madison_calls", 0) + 1 return 0, "".join(f" {n} | {v} | http://deb.debian.org trixie/main amd64 Packages\n" for n in a[2:] for v in self.avail(n)), "" if cmd == "apt-cache" and a[1] == "policy": out = "" for n in a[2:]: out += f"{n}:\n Installed: {self.installed.get(n)}\n Version table:\n *** {self.installed.get(n)} 500\n 500 http://deb.debian.org/debian trixie/main amd64 Packages\n" return 0, out, "" if cmd == "apt-get": if "update" in a: return 0, "", "" if "clean" in a: return 0, "", "" if "-f" in a: self.dpkg_audit = "" return 0, "Setting up x (1) ...\n" if getattr(self, "repaired", False) else "", "" if "--print-uris" in a or "-s" in a: return self.sim(a) # --print-uris prints and installs nothing, with or without -s (9202, 2026-10-05) if "install" in a: if getattr(self, "dpkg_journal", []): # real apt (demo-hp 2026-10-05): a non-empty update journal refuses every install return 100, "", "E: dpkg was interrupted, you must manually run 'sudo dpkg --configure -a' to correct the problem.\n" if self.install_rc: return self.install_rc, "", "E: boom" for x in a: if "=" in x and not x.startswith("-") and "::" not in x: n, v = x.split("=", 1) self.installed[n] = v return 0, getattr(self, "install_out", "Setting up libc6 ...\n"), "" if cmd == "df": return 0, f"Avail\n{self.free}\n", "" if cmd == "docker" and a[1] == "info": return 0, self.live_restore + "\n", "" if cmd == "docker" and a[1] == "version": return 0, self.engine + "\n", "" if cmd == "docker" and a[1:3] == ["ps", "-q"]: return 0, "".join(i + "\n" for i in self.ids), "" # R-528: the memory-kill check's engine (oom_image None = unreadable; oom_state / oom_event the engine's answer) if cmd == "date" and a[1:] == ["+%s"]: # each read is 3 s later than the last, so the order of the reads is visible in the values self.oom_epochs = getattr(self, "oom_epochs", []) + [int(self.clock) + 3 * len(getattr(self, "oom_epochs", []))] return 0, f"{self.oom_epochs[-1]}\n", "" if cmd == "docker" and a[1:4] == ["inspect", "-f", "{{.Config.Image}}"]: img = getattr(self, "oom_image", "gitea.dooplex.hu/admin/felhom-controller:0.300.0") return (0, img + "\n", "") if img is not None else (1, "", "Error: No such object: felhom-controller") if cmd == "docker" and a[1] == "run": self.oom_runs = getattr(self, "oom_runs", []) + [a] return 137, "", "" if cmd == "docker" and a[1:4] == ["inspect", "-f", "{{.State.OOMKilled}} {{.State.ExitCode}}"]: if getattr(self, "oom_inspect_raises", False): raise subprocess.TimeoutExpired(a, 10) return 0, getattr(self, "oom_state", "true 137") + "\n", "" if cmd == "docker" and a[1] == "events": self.oom_events_argv = a return 0, ("oom\n" if getattr(self, "oom_event", True) else ""), "" if cmd == "docker" and a[1:3] == ["rm", "-f"]: self.oom_removed = getattr(self, "oom_removed", []) + a[3:] return 0, a[3] + "\n", "" if cmd == "docker" and a[1] == "inspect": mounts = {"aaa111": "/felhom-controller|/var/run/docker.sock;/app/data;", "bbb222": "/app|/data;"} return 0, mounts.get(a[-1], "/other|;") + "\n", "" if cmd == "docker" and a[1] == "restart": self.restarted_containers = a[2:] return 0, "", "" if cmd == "docker" and a[1] == "exec": return (1, "", "Cannot connect to the Docker daemon") if getattr(self, "controller_blind", False) else (0, self.engine + "\n", "") if cmd == "docker": ids = self.ids + ["x"] * 2 return 0, f"felhom-controller\trunning\tUp 1 hour (healthy)\t{ids[0]}\napp\trunning\tUp 1 hour (healthy)\t{ids[1]}\n", "" if cmd == "cat" and a[1] == osapply.DAEMON_JSON: return (0, self.daemon_json, "") if self.daemon_json is not None else (1, "", "No such file") if cmd == "cat" and a[1] == "/etc/debian_version": return 0, "13.7\n", "" if cmd == "pveversion": return 0, f"pve-manager/{self.installed.get('pve-manager', '9.2.2')}/abcdef (running kernel: 7.0.14-20-pve)\n", "" if cmd == "uname": return 0, "7.0.14-20-pve\n", "" if cmd == "apt-mark": return 0, getattr(self, "held", ""), "" if cmd == "systemctl" and a[1] == "reload": self.reloads = getattr(self, "reloads", 0) + 1 if self.reload_enables and '"live-restore": true' in self.written.get(osapply.DAEMON_JSON, ""): self.live_restore = "true" return 0, "", "" if cmd == "systemctl" and a[1] == "restart": self.restarted = True return 0, "", "" if cmd == "getent": return 0, "1.2.3.4 deb.debian.org\n", "" if cmd == "sh" and a[2] == osapply.DPKG_STATE_SCRIPT: return 0, self.dpkg_audit + osapply.JOURNAL_MARK + "\n" + "".join(j + "\n" for j in getattr(self, "dpkg_journal", [])), "" if cmd == "sh": if "vmlinuz" in a[2]: return 0, "/boot/vmlinuz-7.0.2-6-pve\n/boot/vmlinuz-7.0.14-20-pve\n", "" if "engine=" in a[2]: return 0, f"debian=13.7\nengine={self.engine}\ncontainerd=2.3.3-1~debian.13~trixie\nlive={self.live_restore}\n", "" if "os-release" in a[2]: return 0, "trixie\n", "" if "(deleted)" in a[2]: return 0, getattr(self, "restart_out", ""), "" return 0, "", "" if cmd == "rm": self.snap_active = False return 0, "", "" return 1, "", f"unexpected guest call {a}" def sim(self, a): if "--print-uris" in a: if "-s" in a: # real apt (measured 9202 2026-10-05): with -s it prints the SIMULATION, no URI list return 0, "Inst libc6 [2.41-12+deb13u4] (2.41-12+deb13u4 Debian:13.7/stable [amd64])\n", "" # real apt's line shape, verbatim from 9202 2026-10-05 (audits/night-fixes-2026-10-05/partC/) return 0, ("Need to get 4347 kB of archives.\n" "'http://deb.debian.org/debian/pool/main/b/bash/bash_5.2.37-2%2bb10_amd64.deb' bash_5.2.37-2+b10_amd64.deb 1500792 MD5Sum:27b11721fea83d73b96e0f7023863771\n" "'http://deb.debian.org/debian/pool/main/g/glibc/libc6_2.41-12%2bdeb13u4_amd64.deb' libc6_2.41-12+deb13u4_amd64.deb 2846580 MD5Sum:5559581916477ef1f57ea9f82cecf22e\n" + getattr(self, "extra_uris", "")), "" if "dist-upgrade" in a: if getattr(self, "pending_sim", None) is not None and not getattr(self, "_pending_used", False): self._pending_used = True return 0, "\n".join(self.pending_sim) + "\n", "" return 0, "Inst bash [5.2.37-2+b9] (5.2.37-2+b10 Debian:13.7/stable [amd64])\n", "" out = "" for x in a: if "=" in x and not x.startswith("-") and "::" not in x: n, v = x.split("=", 1) if v not in self.avail(n): return 100, "", f"E: Version '{v}' for '{n}' was not found" origin = getattr(self, "origins", {}).get(n) or ("Docker CE:trixie" if n in osapply.DOCKER_NAMES else SEC if n == "openssl" else DEB) out += f"Inst {n} [{self.installed[n]}] ({v} {origin} [amd64])\n" out += "".join(l + "\n" for l in self.extra_sim) return 0, out, "" def run(f): import io import contextlib buf = io.StringIO() with contextlib.redirect_stdout(buf): rc = osapply.main(["felhom-os-apply", "--plan", PLAN], runner=f) line = [l for l in buf.getvalue().splitlines() if l.startswith("OSAPPLY-REPORT ")][-1] return rc, json.loads(line[len("OSAPPLY-REPORT "):]) class Happy(unittest.TestCase): def test_apply_installs_exactly_the_plan(self): f = Fake() rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(f.installed["libc6"], "2.41-12+deb13u4") self.assertEqual(f.installed["openssl"], "3.5.7-1~deb13u3") self.assertEqual(f.installed["bash"], "5.2.37-2+b9", "a package outside the plan was changed") self.assertEqual(rep["plan"]["upgrade"], 2) self.assertIn("installed", rep) self.assertEqual(rep["pending"][0]["name"], "bash") self.assertTrue(any(l.startswith("os-apply: REPAIR ") for l in f.logs), "the repair line must always print") self.assertTrue(any(l.startswith("os-apply: DONE rc=0") for l in f.logs)) inst = [c for c in f.calls if c[0] == "guest" and "install" in c[2] and "-s" not in c[2] and "-f" not in c[2] and "--print-uris" not in c[2]] self.assertTrue(inst and "Dpkg::Options::=--force-confold" in inst[0][2], "must keep existing config files") def test_already_current_is_a_no_op(self): f = Fake() f.installed.update(libc6="2.41-12+deb13u4", openssl="3.5.7-1~deb13u3") rc, rep = run(f) self.assertEqual(rc, 0) self.assertEqual(rep["plan"]["upgrade"], 0) def test_inventory_installs_nothing(self): f = Fake() f.plan["mode"] = "inventory" rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(f.installed["libc6"], "2.41-12+deb13u3") self.assertIn("installed", rep) def test_health_mode(self): f = Fake() f.plan["mode"] = "health" rc, rep = run(f) self.assertEqual(rc, 0) self.assertEqual(rep["health"]["controller"], "healthy") class Repair(unittest.TestCase): def test_repair_runs_first_and_is_reported(self): f = Fake() f.dpkg_audit = "The following packages have been unpacked but not yet configured.\n perl Larry Wall's\n" f.repaired = True rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(rep["repair"]["half_configured_before"], 1) self.assertEqual(rep["repair"]["fixed"], 1) order = [i for i, c in enumerate(f.calls) if c[0] == "guest" and c[2][-1:] != ["update"]] first_cfg = next(i for i, c in enumerate(f.calls) if c[0] == "guest" and "--configure" in c[2]) first_upd = next(i for i, c in enumerate(f.calls) if c[0] == "guest" and "update" in c[2]) self.assertLess(first_cfg, first_upd, "the repair must run before anything else touches apt") self.assertTrue(order) class Snapshot(unittest.TestCase): def test_a_replaced_version_comes_from_the_snapshot(self): f = Fake() f.live["openssl"] = {"3.5.7-1~deb13u4"} # Debian moved on f.snapshot["openssl"] = {"3.5.7-1~deb13u3"} rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(rep["plan"]["from_snapshot"], 1) self.assertEqual(f.installed["openssl"], "3.5.7-1~deb13u3", "must install the APPROVED version, not the newer one") body = f.written[osapply.SNAPSHOT_LIST] self.assertIn("snapshot.debian.org/archive/debian/20261004T080000Z trixie main", body) self.assertIn("debian-security/20261004T080000Z trixie-security main", body) self.assertFalse(f.snap_active, "the temporary snapshot sources must be removed after the run") def test_snapshot_does_not_have_it_either(self): f = Fake() f.live["openssl"] = set() rc, rep = run(f) self.assertEqual((rc, rep["refused"]["code"]), (2, "R7")) self.assertFalse(f.snap_active) class Refusals(unittest.TestCase): def refused(self, f, code): rc, rep = run(f) self.assertEqual(rc, 2, rep) self.assertEqual(rep["refused"]["code"], code, rep) self.assertTrue(any(l.startswith(f"os-apply: REFUSED: {code} ") for l in f.logs), f.logs) inst = [c for c in f.calls if c[0] == "guest" and "install" in c[2] and "-s" not in c[2] and "-f" not in c[2] and "--print-uris" not in c[2]] self.assertEqual(inst, [], "a refusal must install nothing") return rep def test_R1_usage(self): import io import contextlib f = Fake() with contextlib.redirect_stdout(io.StringIO()): self.assertEqual(osapply.main(["felhom-os-apply", "--plan", PLAN, "--extra"], runner=f), 2) self.assertEqual(osapply.main(["felhom-os-apply", "--plan"], runner=f), 2) def test_R1_path_outside_the_plan_dir(self): import io import contextlib f = Fake() with contextlib.redirect_stdout(io.StringIO()): rc = osapply.main(["felhom-os-apply", "--plan", "/tmp/plan-x.json"], runner=f) self.assertEqual(rc, 2) self.assertTrue(any("R1" in l for l in f.logs)) def test_R1_symlink(self): f = Fake() f.stats[PLAN] = St(mode=statmod.S_IFLNK | 0o777) self.refused(f, "R1") def test_R1_not_owned_by_the_agent(self): f = Fake() f.stats[PLAN] = St(uid=0) self.refused(f, "R1") def test_R2_non_debian_origin_in_the_plan(self): f = Fake() f.plan["packages"][0]["origin"] = "Proxmox" self.refused(f, "R2") def test_R2_non_debian_origin_in_the_simulation(self): f = Fake() f.installed["libc6"] = "2.41-12+deb13u3" orig = f.sim def sim(a): rc, out, err = orig(a) return rc, out.replace("Debian:13.7/stable", "Proxmox Debian Repository:stable"), err f.sim = sim self.refused(f, "R2") def test_R3_slow_lane(self): f = Fake() f.plan["lane"] = "slow" self.refused(f, "R3") def test_R4_removal(self): f = Fake() f.extra_sim = ["Remv bash [5.2.37-2+b9]"] self.refused(f, "R4") def test_R5_downgrade(self): f = Fake() f.installed["libc6"] = "2.41-12+deb13u4" f.plan["packages"] = [{"name": "openssl", "version": "3.5.7-1~deb13u3", "origin": "Debian-Security"}] f.extra_sim = ["Inst openssl [3.5.6-1~deb13u1] (3.5.5-1 Debian:13.7/stable [amd64])"] orig = f.sim def sim(a): # the simulation answers with a LOWER version than installed rc, out, err = orig(a) return rc, "\n".join(l for l in out.splitlines() if not l.startswith("Inst openssl [3.5.6-1~deb13u1] (3.5.7")) + "\n", err f.sim = sim f.plan["packages"][0]["version"] = "3.5.7-1~deb13u3" rep = run(f)[1] # The plan asks 3.5.7; the simulation goes to 3.5.5: that is BOTH a wrong version (R6) and a downgrade. self.assertIn(rep["refused"]["code"], ("R5", "R6")) def test_R5_downgrade_exact(self): f = Fake() f.plan["packages"] = [{"name": "openssl", "version": "3.5.7-1~deb13u3", "origin": "Debian-Security"}] f.installed["openssl"] = "3.5.6-1~deb13u1" orig = f.sim def sim(a): rc, out, err = orig(a) return rc, out.replace("[3.5.6-1~deb13u1]", "[3.5.8-1]"), err f.sim = sim self.refused(f, "R5") def test_R6_new_package(self): f = Fake() f.extra_sim = ["Inst newthing (1.0 Debian:13.7/stable [amd64])"] self.refused(f, "R6") def test_R6_unlisted_package(self): f = Fake() f.extra_sim = ["Inst bash [5.2.37-2+b9] (5.2.37-2+b10 Debian:13.7/stable [amd64])"] self.refused(f, "R6") def test_R6_allow_new_is_slow_lane(self): f = Fake() f.plan["allow_new"] = ["proxmox-kernel-x"] self.refused(f, "R6") def test_R7_not_downloadable_and_no_snapshot(self): f = Fake() f.live["openssl"] = set() f.plan["snapshot"] = "" self.refused(f, "R7") def test_R8_free_space(self): f = Fake() f.free = 100 * 1024 * 1024 self.refused(f, "R8") # R-865: the download is the real one. 2 GB of URIs, 5 GB free: 5 GB < 3 x 2 GB -> R8, with the size in the line. # COMPANION RED-PROOF: put "-s" back into download_bytes -> 0 B -> no refusal -> this test fails. def test_R8_counts_the_real_download(self): f = Fake() f.free = 5 * 1024 ** 3 f.extra_uris = "'http://deb.debian.org/debian/pool/main/b/big/big_1_amd64.deb' big_1_amd64.deb 2000000000 MD5Sum:x\n" rep = self.refused(f, "R8") self.assertIn("download 2004347372 B", str(rep)) def test_download_bytes_never_simulates(self): f = Fake() rc, rep = run(f) self.assertEqual(rc, 0, rep) calls = [c[2] for c in f.calls if c[0] == "guest" and "--print-uris" in c[2]] self.assertTrue(calls, "download_bytes was never called") for c in calls: self.assertNotIn("-s", c, f"--print-uris with -s prints no URIs: {c}") def test_R9_guest_locked_by_a_backup(self): f = Fake() f.files["/etc/pve/lxc/9201.conf"] = CONF_OK + "lock: backup\n" self.refused(f, "R9") def test_R9_apt_lock_held(self): f = Fake() f.lock_held = True self.refused(f, "R9") def test_R10_not_the_boxs_own_guest(self): f = Fake() f.files["/etc/pve/lxc/9201.conf"] = CONF_OK.replace("mp8: /mnt/felhom-drives,", "mp8: /mnt/hdd_1/scratch,") self.refused(f, "R10") def test_R10_reserved_vmid(self): f = Fake() f.plan["vmid"] = 990003 self.refused(f, "R10") def test_R10_bind_only_in_a_snapshot_section(self): f = Fake() f.files["/etc/pve/lxc/9201.conf"] = "rootfs: x\n[snap1]\nmp8: /mnt/felhom-drives,mp=/mnt/felhom-drives\n" self.refused(f, "R10") def test_R10_not_running(self): f = Fake() f.status = "status: stopped" self.refused(f, "R10") def test_R11_duplicate(self): f = Fake() f.plan["packages"].append(dict(f.plan["packages"][0])) self.refused(f, "R11") def test_R11_bad_version_string(self): f = Fake() f.plan["packages"][0]["version"] = "1.0; rm -rf /" self.refused(f, "R11") def test_R11_bad_name(self): f = Fake() f.plan["packages"][0]["name"] = "--purge" self.refused(f, "R11") def test_R12_unknown_layer(self): f = Fake() f.plan["layer"] = "vm" self.refused(f, "R12") def test_R12_host_on_a_byo_box(self): f = Fake() f.plan["layer"] = "host" f.files[osapply.INSTALL_STATE] = json.dumps({"mode": "byo"}) self.refused(f, "R12") def test_R12_host_without_an_install_record(self): f = Fake() f.plan["layer"] = "host" del f.stats[osapply.INSTALL_STATE] self.refused(f, "R12") def test_R12_host_record_not_root_owned(self): # the agent can write agent.json's deployment_mode; only a ROOT-owned record proves anything f = Fake() f.plan["layer"] = "host" f.stats[osapply.INSTALL_STATE] = St(mode=statmod.S_IFREG | 0o644, uid=999) self.refused(f, "R12") def test_R14_kernel_package_in_a_host_plan(self): f = Fake() f.plan["layer"] = "host" f.plan["packages"].append({"name": "linux-image-amd64", "version": "6.12.1-1", "origin": "Debian"}) self.refused(f, "R14") def test_R14_kernel_package_pulled_by_the_simulation(self): f = Fake() f.plan["layer"] = "host" f.extra_sim = ["Inst grub-common [2.12-9] (2.12-10 Debian:13.7/stable [amd64])"] f.installed["grub-common"] = "2.12-9" f.plan["packages"].append({"name": "grub-common", "version": "2.12-10", "origin": "Debian"}) self.refused(f, "R14") def test_R13_repair_does_not_fix_it(self): f = Fake() f.dpkg_audit = "The following packages are broken\n perl\n" orig = f.guest def guest(vmid, argv, timeout=1800): rc, out, err = orig(vmid, argv, timeout) if "-f" in argv: f.dpkg_audit = "The following packages are broken\n perl\n" return rc, out, err f.guest = guest self.refused(f, "R13") class Conffiles(unittest.TestCase): # dpkg's two shapes, measured live 2026-10-04: an unchanged file is UPDATED; a locally changed one is KEPT. def test_updated_vs_kept(self): f = Fake() f.install_out = ("Installing new version of config file /etc/debian_version ...\n" "Configuration file '/etc/ssh/sshd_config'\n ==> Modified (by you or by a script) since installation.\n" " ==> Keeping old config file as default.\n") rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertIn("os-apply: CONFFILE updated /etc/debian_version (it was not changed locally)", f.logs) self.assertTrue(any(l.startswith("os-apply: CONFFILE kept /etc/ssh/sshd_config") for l in f.logs), f.logs) self.assertEqual(rep["conffiles_kept"], ["/etc/ssh/sshd_config"]) self.assertFalse(any("kept /etc/debian_version" in l for l in f.logs), "an updated file must not be reported as kept") class HostLayer(unittest.TestCase): def test_host_runs_on_the_host_not_in_the_guest(self): f = Fake() f.plan["layer"] = "host" rc, rep = run(f) self.assertEqual(rc, 0, rep) inst = [c for c in f.calls if c[0] == "host" and "install" in c[1] and "-s" not in c[1] and "-f" not in c[1]] self.assertTrue(inst, "the host install must run on the host") self.assertFalse([c for c in f.calls if c[0] == "guest" and "install" in c[2]], "nothing installed in the guest") self.assertEqual(sorted(rep["health_after"]["host_services"]), sorted(osapply.HOST_SERVICES)) self.assertTrue(rep["health_after"]["guest_running"]) def test_pending_fast_skips_proxmox_docker_and_kernel(self): f = Fake() f.plan["layer"] = "host" f.plan["select"] = "pending-fast" f.plan["packages"] = [] f.installed.update({"pve-manager": "9.2.2", "linux-image-amd64": "6.12.1", "docker-ce": "29.7"}) f.pending_sim = [ "Inst libc6 [2.41-12+deb13u3] (2.41-12+deb13u4 Debian:13.7/stable [amd64])", "Inst openssl [3.5.6-1~deb13u1] (3.5.7-1~deb13u3 Debian:13.7/stable, Debian-Security:13/stable-security [amd64])", "Inst pve-manager [9.2.2] (9.2.21 Proxmox Debian Repository:stable [amd64])", "Inst linux-image-amd64 [6.12.1] (6.12.9 Debian:13.7/stable [amd64])", "Inst docker-ce [29.7] (29.8 Docker CE:trixie [amd64])", "Inst brand-new (1.0 Debian:13.7/stable [amd64])", ] rc, rep = run(f) self.assertEqual(rc, 0, rep) got = sorted(u["name"] for u in rep["upgraded"]) self.assertEqual(got, ["libc6", "openssl"], "pending-fast must take only installed, Debian-origin, non-kernel packages") def test_reboot_needed_when_pid1_or_lxc_start(self): f = Fake() f.plan["layer"] = "host" f.restart_out = "1 systemd\n2101 lxc-start\n530 sshd\n" rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertTrue(rep["reboot_needed"]) self.assertIn("lxc-start", rep["restart_needed"]) def test_reboot_needed_for_lxc_start_alone(self): # lxc-start runs the guest; only a guest restart (or a host reboot) replaces it f = Fake() f.plan["layer"] = "host" f.restart_out = "2101 lxc-start\n530 sshd\n" rc, rep = run(f) self.assertTrue(rep["reboot_needed"], rep) def test_every_layer_scans_every_pass(self): # R-849 (v0.142.0): host AND guest are scanned on every pass, so a reboot / restart clears the flag. for layer in ("host", "guest"): f = Fake() f.plan["layer"] = layer f.plan["mode"] = "inventory" f.restart_out = "2101 lxc-start\n" if layer == "host" else "1 systemd\n" rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertTrue(rep["reboot_scanned"] and rep["reboot_needed"], (layer, rep)) f = Fake() f.plan["layer"] = layer f.plan["mode"] = "inventory" f.restart_out = "" rc, rep = run(f) self.assertTrue(rep["reboot_scanned"] and rep["reboot_needed"] is False, (layer, rep)) def test_no_reboot_for_ordinary_daemons(self): f = Fake() f.plan["layer"] = "host" f.restart_out = "530 sshd\n611 cron\n" rc, rep = run(f) self.assertFalse(rep["reboot_needed"], rep) class Speed(unittest.TestCase): # R-845: no `pct exec` per package — version checks on the host, madison once, the restart scan only after an install. def test_no_per_package_guest_calls(self): f = Fake() for i in range(40): f.installed[f"pkg{i}"] = "1.0-1" f.live[f"pkg{i}"] = {"1.0-2"} f.plan["packages"].append({"name": f"pkg{i}", "version": "1.0-2", "origin": "Debian"}) rc, rep = run(f) self.assertEqual(rc, 0, rep) guest_cmp = [c for c in f.calls if c[0] == "guest" and "--compare-versions" in c[2]] self.assertEqual(guest_cmp, [], "version comparisons must run on the host") self.assertEqual(f.madison_calls, 1, "madison must run once for all packages") guest_calls = len([c for c in f.calls if c[0] == "guest"]) self.assertLess(guest_calls, 30, f"{guest_calls} guest calls for 42 packages — something is per-package again") class Failure(unittest.TestCase): def test_install_failure_is_rc3_with_dpkg_state(self): f = Fake() f.install_rc = 100 rc, rep = run(f) self.assertEqual(rc, 3) self.assertEqual(rep["failed"]["rc"], 100) self.assertTrue(any(l.startswith("os-apply: FAILED rc=100 step=install") for l in f.logs)) class RestartSkipPattern(unittest.TestCase): """The cgroup filter runs as `grep -q PATTERN /proc//cgroup`; check it with grep itself against the cgroup lines measured on demo-felhom 2026-10-04.""" def grep(self, pattern, line): return subprocess.run(["grep", "-q", pattern], input=line + "\n", text=True).returncode == 0 def test_restart_skip_patterns_against_real_cgroups(self): host = osapply.RESTART_SKIP_CGROUP["host"] self.assertTrue(self.grep(host, "0::/lxc/9201/ns/system.slice/docker.service"), "a guest process must be skipped") self.assertFalse(self.grep(host, "0::/lxc.monitor/9201"), "lxc-start must NOT be skipped (it runs the guest)") self.assertFalse(self.grep(host, "0::/system.slice/pve-cluster.service"), "a host daemon must NOT be skipped") guest = osapply.RESTART_SKIP_CGROUP["guest"] self.assertTrue(self.grep(guest, "0::/system.slice/docker-0123abcd.scope")) self.assertFalse(self.grep(guest, "0::/system.slice/cron.service")) DOCKER_SET = [{"name": "docker-ce", "version": "5:29.8.2-1~debian.13~trixie", "origin": "Docker CE"}, {"name": "containerd.io", "version": "2.3.6-1~debian.13~trixie", "origin": "Docker CE"}] def docker_fake(signed=None, undo=False, ring0=False): f = Fake() f.installed.update({"docker-ce": "5:29.7.2-1~debian.13~trixie", "containerd.io": "2.3.3-1~debian.13~trixie"}) f.live["docker-ce"] = {"5:29.8.2-1~debian.13~trixie", "5:29.7.2-1~debian.13~trixie"} f.live["containerd.io"] = {"2.3.6-1~debian.13~trixie", "2.3.3-1~debian.13~trixie"} f.plan = {"release_id": "os-docker-t1", "layer": "docker", "lane": "slow", "vmid": 9201, "mode": "apply", "packages": [dict(p) for p in DOCKER_SET]} if undo: f.plan["undo"] = True if ring0: f.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": True}) if signed is not None: f.plan["signed"] = signed return f def signed_job(packages=DOCKER_SET, host="demo-hp-bb76ea", op="os_docker_step", undo=False, nonce="n1", issued="2026-10-04T11:50:00Z", expires="2026-10-04T12:30:00Z"): import base64 params = {"packages": packages, "undo": undo} blob = json.dumps({"expires_at": expires, "issued_at": issued, "key_id": "felhom-op-1", "nonce": nonce, "op": op, "params": params, "target": {"guest_id": "", "host_id": host}}, sort_keys=True).encode() return {"blob_b64": base64.b64encode(blob).decode(), "sig": "-----BEGIN SSH SIGNATURE-----\nx\n-----END SSH SIGNATURE-----\n"} class DockerLane(unittest.TestCase): """`11` §5.8, agent v0.142.0. Each test names the refusal it pins; the red-proof file mutates each one.""" def refused(self, f, code): rc, rep = run(f) self.assertEqual(rc, 2, rep) self.assertEqual(rep["refused"]["code"], code, rep) self.assertEqual(f.installed.get("docker-ce", "5:29.7.2-1~debian.13~trixie"), "5:29.7.2-1~debian.13~trixie") return rep def test_docker_package_in_a_fast_plan_is_refused(self): f = Fake() f.plan["packages"].append({"name": "docker-ce", "version": "5:29.8.2-1~debian.13~trixie", "origin": "Debian"}) self.refused(f, "R2") def test_docker_layer_in_the_fast_lane_is_refused(self): f = docker_fake(ring0=True) f.plan["lane"] = "fast" self.refused(f, "R3") def test_no_authority_is_refused(self): self.refused(docker_fake(), "R3") def test_ring0_mark_allows_pending_docker(self): f = docker_fake(ring0=True) f.plan["select"], f.plan["packages"] = "pending-docker", [] f.pending_sim = ["Inst docker-ce [5:29.7.2-1~debian.13~trixie] (5:29.8.2-1~debian.13~trixie Docker CE:trixie [amd64])", "Inst containerd.io [2.3.3-1~debian.13~trixie] (2.3.6-1~debian.13~trixie Docker CE:trixie [amd64])", "Inst bash [5.2.37-2+b9] (5.2.37-2+b10 Debian:13.7/stable [amd64])"] rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(rep["authority"], "ring0") self.assertEqual(f.installed["docker-ce"], "5:29.8.2-1~debian.13~trixie") self.assertEqual(f.installed["bash"], "5.2.37-2+b9", "a Debian package must not ride a Docker step") self.assertFalse(getattr(f, "restarted", False), "never a docker restart") def test_signed_job_applies_exactly_its_packages(self): f = docker_fake(signed=signed_job()) rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(rep["authority"], "signed") self.assertEqual(f.installed["containerd.io"], "2.3.6-1~debian.13~trixie") self.assertEqual(f.verified[0], osapply.TRUST_SIGNERS, "the ROOT-OWNED signers file, not the agent's config") self.assertIn("n1", f.nonces) def test_bad_signature_is_refused(self): f = docker_fake(signed=signed_job()) f.sig_rc = 255 self.refused(f, "R3") self.assertEqual(f.nonces, {}, "a bad signature must not burn a nonce") def test_signed_job_for_another_host_is_refused(self): self.refused(docker_fake(signed=signed_job(host="demo-felhom-8363b5")), "R3") def test_signed_job_other_op_is_refused(self): self.refused(docker_fake(signed=signed_job(op="agent_update")), "R3") def test_expired_signed_job_is_refused(self): self.refused(docker_fake(signed=signed_job(expires="2026-10-04T11:55:00Z")), "R3") def test_a_window_longer_than_24h_is_refused(self): # R-1280. RED-PROOF: drop the MAX_WINDOW_S check → the job (current, signed, 24 h + 1 s) runs. self.refused(docker_fake(signed=signed_job(issued="2026-10-03T11:59:59Z", expires="2026-10-04T12:00:00Z")), "R3") def test_a_window_of_24h_is_accepted(self): rc, rep = run(docker_fake(signed=signed_job(issued="2026-10-03T12:30:00Z", expires="2026-10-04T12:30:00Z"))) self.assertNotEqual((rep.get("refused") or {}).get("code"), "R3", rep) def test_replayed_signed_job_is_refused(self): f = docker_fake(signed=signed_job()) f.nonces = {"n1": f.clock + 600} self.refused(f, "R3") def test_plan_must_equal_the_signed_packages(self): f = docker_fake(signed=signed_job(packages=DOCKER_SET[:1])) self.refused(f, "R3") def test_agent_writable_trust_file_is_refused(self): f = docker_fake(ring0=True) f.stats[osapply.TRUST_FILE] = St(mode=statmod.S_IFREG | 0o644, uid=999) self.refused(f, "R3") def test_live_restore_off_is_refused(self): f = docker_fake(signed=signed_job()) f.live_restore = "false" self.refused(f, "R15") def test_undo_needs_a_signed_job(self): self.refused(docker_fake(undo=True, ring0=True), "R3") def test_signed_undo_downgrades(self): old = [{"name": "docker-ce", "version": "5:29.7.2-1~debian.13~trixie", "origin": "Docker CE"}] f = docker_fake(signed=signed_job(packages=old, undo=True), undo=True) f.plan["packages"] = [dict(p) for p in old] f.installed["docker-ce"] = "5:29.8.2-1~debian.13~trixie" rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(f.installed["docker-ce"], "5:29.7.2-1~debian.13~trixie") self.assertTrue(rep["undo"]) def test_unsigned_downgrade_is_refused(self): f = docker_fake(ring0=True) f.plan["packages"] = [{"name": "docker-ce", "version": "5:29.6.0-1~debian.13~trixie", "origin": "Docker CE"}] f.live["docker-ce"].add("5:29.6.0-1~debian.13~trixie") rc, rep = run(f) self.assertEqual(rep["plan"]["upgrade"], 0, "an older version on an unsigned step is 'already', never installed") def test_step_restarts_only_the_socket_users(self): # R-858: after an engine step, ONLY the container that mounts the docker socket is restarted (here the controller) f = docker_fake(signed=signed_job()) rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(f.restarted_containers, ["felhom-controller"]) self.assertEqual(rep["socket_restarted"], ["felhom-controller"]) def test_no_install_restarts_nothing(self): f = docker_fake(signed=signed_job()) f.installed.update({"docker-ce": "5:29.8.2-1~debian.13~trixie", "containerd.io": "2.3.6-1~debian.13~trixie"}) rc, rep = run(f) self.assertFalse(hasattr(f, "restarted_containers"), "nothing installed -> no container restart") def test_health_says_when_the_controller_cannot_reach_docker(self): f = docker_fake(signed=signed_job()) f.controller_blind = True rc, rep = run(f) self.assertFalse(rep["health_after"]["controller_docker_ok"]) def test_health_carries_container_ids(self): f = docker_fake(signed=signed_job()) rc, rep = run(f) self.assertEqual(rep["health_after"]["containers"]["app"]["id"], "bbb222") self.assertEqual(rep["docker_engine"], "29.7.2") class LiveRestore(unittest.TestCase): def lr(self): f = Fake() f.plan = {"release_id": "lr", "layer": "guest", "lane": "fast", "vmid": 9201, "mode": "live-restore-on", "packages": []} f.live_restore = "false" return f def test_turns_it_on_with_a_reload_never_a_restart(self): f = self.lr() rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(json.loads(f.written[osapply.DAEMON_JSON]), {"log-driver": "json-file", "live-restore": True}) self.assertEqual(f.reloads, 1) self.assertFalse(getattr(f, "restarted", False)) self.assertEqual(rep["live_restore"]["result"], "on") self.assertTrue(rep["live_restore"]["same_ids"]) def test_already_on_writes_nothing(self): f = self.lr() f.live_restore = "true" rc, rep = run(f) self.assertEqual(rc, 0) self.assertNotIn(osapply.DAEMON_JSON, f.written) def test_invalid_daemon_json_is_left_alone(self): f = self.lr() f.daemon_json = "{not json" rc, rep = run(f) self.assertEqual(rep["refused"]["code"], "R16") self.assertNotIn(osapply.DAEMON_JSON, f.written) def test_reload_that_does_not_enable_puts_the_file_back(self): f = self.lr() f.reload_enables = False rc, rep = run(f) self.assertEqual(rc, 3, rep) self.assertEqual(f.written[osapply.DAEMON_JSON], '{"log-driver": "json-file"}') self.assertFalse(getattr(f, "restarted", False)) class Facts(unittest.TestCase): def facts(self, f=None): f = f or Fake() f.plan = {"release_id": "facts", "layer": "host", "lane": "fast", "vmid": 9201, "mode": "facts", "packages": []} rc, rep = run(f) self.assertEqual(rc, 0, rep) return f, rep["facts"] def test_reads_host_and_guest(self): f = Fake() f.files["/proc/sys/kernel/tainted"] = "4225\n" # 4096 + 128 (D: oops) + 1 f.files["/proc/sys/kernel/panic"] = "10\n" f.held = "tzdata\n" f.files["/etc/default/grub"] = "GRUB_DEFAULT=saved\n" f.files["/boot/grub/grubenv"] = "# GRUB Environment Block\nsaved_entry=gnulinux-advanced-x>gnulinux-7.0.14-20-pve-advanced-x\n" _, fa = self.facts(f) h, g = fa["host"], fa["guest"] self.assertEqual((h["debian"], h["kernel_running"], h["kernel_next_boot"]), ("13.7", "7.0.14-20-pve", "7.0.14-20-pve")) self.assertEqual(h["held"], ["tzdata"]) self.assertTrue(h["oops_this_boot"]) self.assertEqual(h["kernel_panic"], 10) self.assertEqual((g["debian"], g["docker_engine"], g["live_restore"]), ("13.7", "29.7.2", "on")) self.assertEqual(g["containerd"], "2.3.3-1~debian.13~trixie") def test_next_entry_wins_and_default_zero_is_the_newest(self): f = Fake() f.files["/etc/default/grub"] = "GRUB_DEFAULT=saved\n" f.files["/boot/grub/grubenv"] = "saved_entry=gnulinux-advanced-x>gnulinux-7.0.2-6-pve-advanced-x\nnext_entry=gnulinux-advanced-x>gnulinux-7.0.14-20-pve-advanced-x\n" _, fa = self.facts(f) self.assertEqual(fa["host"]["kernel_next_boot"], "7.0.14-20-pve") self.assertIn("next_entry", fa["host"]["kernel_next_boot_source"]) f = Fake() f.files["/etc/default/grub"] = "GRUB_DEFAULT=0\n" _, fa = self.facts(f) self.assertEqual(fa["host"]["kernel_next_boot"], "7.0.14-20-pve", "dpkg order, not string order (7.0.2 < 7.0.14)") def test_stopped_guest_is_unknown_never_guessed(self): f = Fake() f.status = "status: stopped" _, fa = self.facts(f) self.assertEqual(fa["guest"]["docker_engine"], "unknown") self.assertIn("R10", fa["guest"]["unknown_reason"]) self.assertEqual(fa["host"]["tainted"], None, "unreadable -> None, not 0") class RealSignatureCheck(unittest.TestCase): """The REAL Runner.verify_sig with the real ssh-keygen and a throwaway key, in the installer's allowed_signers form (` namespaces="felhom-op-v1" `). Nothing leaves the temp dir.""" def setUp(self): import shutil if not shutil.which("ssh-keygen"): self.skipTest("ssh-keygen not available") import tempfile self.d = tempfile.mkdtemp() self.key = os.path.join(self.d, "k") subprocess.run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", "felhom-op-1", "-f", self.key], check=True) pub = open(self.key + ".pub").read().strip() self.signers = os.path.join(self.d, "signers") open(self.signers, "w").write(f'felhom-op-1 namespaces="felhom-op-v1" {pub}\n') def sign(self, blob, ns="felhom-op-v1"): bp = os.path.join(self.d, "blob") open(bp, "wb").write(blob) if os.path.exists(bp + ".sig"): os.remove(bp + ".sig") # ssh-keygen -Y sign asks before overwriting (it would wait on stdin) subprocess.run(["ssh-keygen", "-q", "-Y", "sign", "-f", self.key, "-n", ns, bp], check=True, stdin=subprocess.DEVNULL, timeout=30) return open(bp + ".sig").read() def test_good_signature_verifies(self): blob = b'{"op":"os_docker_step"}' self.assertEqual(osapply.Runner().verify_sig(self.signers, "felhom-op-1", "felhom-op-v1", blob, self.sign(blob)), 0) def test_changed_blob_wrong_namespace_or_wrong_principal_fail(self): blob = b'{"op":"os_docker_step"}' sig = self.sign(blob) r = osapply.Runner() self.assertNotEqual(r.verify_sig(self.signers, "felhom-op-1", "felhom-op-v1", blob + b" ", sig), 0) self.assertNotEqual(r.verify_sig(self.signers, "someone-else", "felhom-op-v1", blob, sig), 0) self.assertNotEqual(r.verify_sig(self.signers, "felhom-op-1", "felhom-op-v1", blob, self.sign(blob, ns="other-ns")), 0) class UnsentReport(unittest.TestCase): """R-868 (v0.144.0): an apply pass keeps its report on disk until the agent has sent it. COMPANION RED-PROOF: drop the r.save_report call in main() -> test_apply_keeps_a_copy fails.""" def test_apply_keeps_a_copy_with_the_agents_ids(self): f = Fake() f.plan.update(run_id="20261005T0257-ab12", trigger="debug", ring=0) rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(len(f.saved_reports), 1, "an apply pass must keep its report on disk") path, saved = f.saved_reports[0] self.assertEqual(path, PLAN) self.assertEqual(saved, rep, "the copy is the report the agent would have read") self.assertEqual((saved["run_id"], saved["trigger"], saved["ring"]), ("20261005T0257-ab12", "debug", 0)) def test_a_refusal_is_kept_too(self): f = Fake() f.free = 1 rc, rep = run(f) self.assertEqual(rc, 2) self.assertEqual(f.saved_reports[0][1]["refused"]["code"], "R8") def test_other_modes_keep_nothing(self): f = Fake() f.plan["mode"] = "health" run(f) self.assertEqual(f.saved_reports, []) def test_odd_ids_are_dropped_not_trusted(self): f = Fake() f.plan.update(run_id="../../etc/x", trigger="Night; rm", ring=True) rc, rep = run(f) self.assertEqual(rc, 0, rep) for k in ("run_id", "trigger", "ring"): self.assertNotIn(k, rep) class SaveReportOnDisk(unittest.TestCase): """The real Runner.save_report on a temp dir: root writes into the AGENT's directory, so a symlink must never be followed — neither for the directory nor for the file name.""" def setUp(self): import tempfile self.tmp = tempfile.mkdtemp() self.dir = os.path.join(self.tmp, "os") os.mkdir(self.dir) self.prev = osapply.PLAN_DIR osapply.PLAN_DIR = self.dir self.r = osapply.Runner() self.r.agent_uid = lambda: os.getuid() def tearDown(self): import shutil osapply.PLAN_DIR = self.prev shutil.rmtree(self.tmp) def test_writes_0600_next_to_the_plan(self): p = self.r.save_report(os.path.join(self.dir, "plan-r1-guest-apply.json"), {"mode": "apply"}) self.assertEqual(p, os.path.join(self.dir, "report-r1-guest-apply.json")) st = os.stat(p) self.assertEqual(statmod.S_IMODE(st.st_mode), 0o600) with open(p) as fh: self.assertEqual(json.load(fh), {"mode": "apply"}) def test_a_symlink_at_the_name_is_replaced_not_followed(self): victim = os.path.join(self.tmp, "victim") with open(victim, "w") as fh: fh.write("untouched") os.symlink(victim, os.path.join(self.dir, "report-r2-guest-apply.json")) self.r.save_report(os.path.join(self.dir, "plan-r2-guest-apply.json"), {"mode": "apply"}) with open(victim) as fh: self.assertEqual(fh.read(), "untouched") self.assertFalse(os.path.islink(os.path.join(self.dir, "report-r2-guest-apply.json"))) def test_a_symlinked_directory_is_refused(self): real = os.path.join(self.tmp, "elsewhere") os.mkdir(real) link = os.path.join(self.tmp, "linked") os.symlink(real, link) osapply.PLAN_DIR = link with self.assertRaises(OSError): self.r.save_report(os.path.join(link, "plan-r3-guest-apply.json"), {"mode": "apply"}) self.assertEqual(os.listdir(real), []) class AgentDiesMidPass(unittest.TestCase): """R-868, MEASURED LIVE 2026-10-05 05:45 UTC on demo-hp (agent v0.144.0): the agent was kill -9-ed while apt-get ran; apt finished all 13 packages, but the wrapper's next log line went to a stderr pipe nobody reads any more -> BrokenPipeError -> the wrapper died before save_report: no DONE in the journal, no kept copy, no report. COMPANION RED-PROOF: let Runner.log write to stderr unguarded -> this test fails (BrokenPipeError).""" def test_a_dead_reader_does_not_stop_the_report_copy(self): import io, sys, contextlib class DeadPipe(io.TextIOBase): dead = False def write(self, s): if DeadPipe.dead: raise BrokenPipeError(32, "Broken pipe") return len(s) def flush(self): if DeadPipe.dead: raise BrokenPipeError(32, "Broken pipe") class DyingFake(Fake): def emulate(self, argv): a = [x for x in argv if not re.match(r"^[A-Z_]+=", x) and x != "env"] if a and a[0] == "apt-get" and "install" in a and "-s" not in a and "--print-uris" not in a: DeadPipe.dead = True # the agent is killed while apt-get runs return super().emulate(argv) f = DyingFake() journal = [] f.log = lambda line: osapply.Runner.log(f, line) # the REAL log(): stderr, then the journal prev_run = osapply.subprocess.run osapply.subprocess.run = lambda argv, *a, **kw: (journal.append(argv[-1]) if argv[0] == "logger" else prev_run(argv, *a, **kw)) # never the real `logger` prev_err, prev_out = sys.stderr, sys.stdout sys.stderr, sys.stdout = DeadPipe(), DeadPipe() try: rc = osapply.main(["felhom-os-apply", "--plan", PLAN], runner=f) finally: sys.stderr, sys.stdout = prev_err, prev_out osapply.subprocess.run = prev_run DeadPipe.dead = False self.assertEqual(rc, 0) self.assertEqual(len(f.saved_reports), 1, "the kept copy is the only way this report reaches the hub") self.assertEqual(len(f.saved_reports[0][1]["upgraded"]), 2) self.assertTrue(any(l.startswith("os-apply: DONE") for l in journal), "the journal must still get the DONE line") class CrashLeftTheJournal(unittest.TestCase): """R-876 — THE MEASURED SHAPE (demo-hp 2026-10-05, a crash while dpkg unpacked): `dpkg --audit` CLEAN, but /var/lib/dpkg/updates holds 3 files; apt refuses every install ("dpkg was interrupted"). v0.144.1 logged `REPAIR configured=0 fixed=0`, then `FAILED rc=100`, every pass, until a person ran `dpkg --configure -a`. COMPANION RED-PROOFS: drop `or journal` from repair()'s condition -> test 1 fails; drop the interrupted-retry -> test 3 fails; make repair() always run -> test 2 fails (R-845's speed).""" def test_the_next_pass_repairs_by_itself_and_finishes(self): f = Fake() f.dpkg_audit = "" f.dpkg_journal = ["0000", "0001", "0002"] rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(rep["repair"]["journal_before"], 3) self.assertTrue(rep["repair"]["clean_after"]) self.assertEqual(len(rep["upgraded"]), 2) cfg = next(i for i, c in enumerate(f.calls) if c[0] == "guest" and "--configure" in c[2]) inst = next(i for i, c in enumerate(f.calls) if c[0] == "guest" and "install" in c[2] and "-s" not in c[2] and "-f" not in c[2] and "--print-uris" not in c[2]) self.assertLess(cfg, inst, "the repair must run before the install") self.assertFalse(any("INTERRUPTED" in l for l in f.logs), "the journal check must catch it BEFORE apt refuses") def test_a_clean_pass_still_costs_one_state_call(self): f = Fake() rc, rep = run(f) self.assertEqual(rc, 0, rep) state = [c for c in f.calls if c[0] == "guest" and c[2][:2] == ["sh", "-c"] and c[2][2] == osapply.DPKG_STATE_SCRIPT] self.assertEqual(len(state), 1, "R-845: a clean pass reads dpkg's state ONCE and runs no repair") self.assertEqual(getattr(f, "configured_calls", 0), 0) def test_apt_interrupted_is_repaired_and_retried_once(self): """The belt: the journal probe saw nothing (a race, an odd layout), apt still says interrupted.""" f = Fake() orig = f.emulate state = {"first": True} def emulate(argv): a = [x for x in argv if not re.match(r"^[A-Z_]+=", x) and x != "env"] if a[0] == "apt-get" and "install" in a and "-s" not in a and "-f" not in a and "--print-uris" not in a and state["first"]: state["first"] = False return 100, "", "E: dpkg was interrupted, you must manually run 'sudo dpkg --configure -a' to correct the problem.\n" return orig(argv) f.emulate = emulate rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertTrue(any("INTERRUPTED" in l for l in f.logs), f.logs) self.assertTrue(any(l.startswith("os-apply: REPAIR ") and l.endswith("forced") for l in f.logs), f.logs) class OOMCheck(unittest.TestCase): """R-528 (`09` decision 157): after a Docker engine step the wrapper proves the engine reports a memory kill (OOMKilled=true AND the `oom` event). Reported only; the hub decides. Red-proofs: audits/readback-2026-10-07/F/.""" def apply(self, **kw): f = docker_fake(signed=signed_job()) for k, v in kw.items(): setattr(f, k, v) rc, rep = run(f) self.assertEqual(rc, 0, rep) return f, rep def test_pass_when_oomkilled_and_the_event(self): f, rep = self.apply() oc = rep["oom_check"] self.assertEqual(oc["result"], "pass", oc) self.assertEqual((oc["oom_killed"], oc["oom_event"], oc["exit_code"]), (True, True, 137)) self.assertEqual(oc["image"], "gitea.dooplex.hu/admin/felhom-controller:0.300.0") self.assertEqual(sorted(oc), ["detail", "exit_code", "image", "oom_event", "oom_killed", "result"]) run_argv = f.oom_runs[0] name = run_argv[run_argv.index("--name") + 1] self.assertTrue(re.match(r"^felhom-oomcheck-[0-9]+-[0-9a-f]{8}$", name), name) for flag, val in (("--pull", "never"), ("--network", "none"), ("--memory", "64m"), ("--memory-swap", "64m"), ("--label", "felhom.oomcheck=1"), ("--entrypoint", "sh")): self.assertEqual(run_argv[run_argv.index(flag) + 1], val, flag) self.assertNotIn("-v", run_argv) self.assertNotIn("-p", run_argv) self.assertEqual(run_argv[-3:], ["gitea.dooplex.hu/admin/felhom-controller:0.300.0", "-c", osapply.OOMCHECK_SCRIPT]) self.assertIn(f"container={name}", f.oom_events_argv) self.assertIn("event=oom", f.oom_events_argv) self.assertEqual(f.oom_removed, [name], "the check container must be removed") self.assertTrue(rep["health_after"], "health is read before the check") # bounded: every call has a timeout and the clock is read twice — the worst case stays within 90 s self.assertLessEqual(sum(osapply.OOMCHECK_TIMEOUTS.values()) + osapply.OOMCHECK_TIMEOUTS["clock"] + osapply.OOMCHECK_SETTLE, 90) def test_events_window_ends_after_the_settle_wait(self): # Measured (F1/F2): an --until taken right after the run missed the oom event. The window must end after a # wait of >= 2 s that comes AFTER the run, at the guest epoch read after that wait, + 1. f, rep = self.apply() run_i = next(i for i, c in enumerate(f.calls) if c[-1][:2] == ["docker", "run"]) date_i = [i for i, c in enumerate(f.calls) if c[-1] == ["date", "+%s"]] waits = [(i, s) for i, s in getattr(f, "sleeps", []) if i > run_i] self.assertTrue(waits and waits[0][1] >= 2, f"no settle wait after the run: {getattr(f, 'sleeps', None)}") self.assertTrue(date_i[-1] >= waits[0][0], "the end epoch must be read after the wait") ev = f.oom_events_argv since, until = int(ev[ev.index("--since") + 1]), int(ev[ev.index("--until") + 1]) self.assertEqual(until, f.oom_epochs[-1] + 1, "until = the guest epoch read after the wait, + 1") self.assertGreater(until, f.oom_epochs[0] + 1) self.assertLess(since, f.oom_epochs[0] + 1) def test_oomkilled_false_is_fail(self): f, rep = self.apply(oom_state="false 0") oc = rep["oom_check"] self.assertEqual(oc["result"], "fail", oc) self.assertIn("OOMKilled=true", oc["detail"]) self.assertTrue(oc["oom_event"]) self.assertEqual(len(f.oom_removed), 1) def test_no_event_is_fail(self): f, rep = self.apply(oom_event=False) oc = rep["oom_check"] self.assertEqual(oc["result"], "fail", oc) self.assertIn("the oom event", oc["detail"]) self.assertTrue(oc["oom_killed"]) def test_image_unreadable_is_error(self): f, rep = self.apply(oom_image=None) oc = rep["oom_check"] self.assertEqual(oc["result"], "error", oc) self.assertIsNone(oc["image"]) self.assertIn("image could not be read", oc["detail"]) self.assertFalse(hasattr(f, "oom_runs"), "no container is started without an image") def test_container_removed_even_when_inspect_raises(self): f, rep = self.apply(oom_inspect_raises=True) oc = rep["oom_check"] self.assertEqual(oc["result"], "error", oc) self.assertEqual(len(f.oom_removed), 1, "the container must be removed even when inspect raised") self.assertTrue(f.oom_removed[0].startswith(osapply.OOMCHECK_PREFIX)) self.assertNotIn("failed", rep, "the check never turns the step into a failure") def test_never_on_guest_or_host_or_in_health_mode(self): g = Fake() rc, rep = run(g) self.assertEqual(rc, 0, rep) h = Fake() h.plan["layer"] = "host" h.plan["packages"] = [{"name": "bash", "version": "5.2.37-2+b10", "origin": "Debian"}] h.installed["bash"] = "5.2.37-2+b9" h.live["bash"] = {"5.2.37-2+b10"} rc_h, rep_h = run(h) self.assertEqual(rc_h, 0, rep_h) d = docker_fake(signed=signed_job()) d.plan["mode"] = "health" rc_d, rep_d = run(d) self.assertEqual(rc_d, 0, rep_d) for f, r in ((g, rep), (h, rep_h), (d, rep_d)): self.assertNotIn("oom_check", r) self.assertFalse(hasattr(f, "oom_runs"), r.get("layer")) self.assertFalse(any(c[-1][:2] == ["docker", "run"] for c in f.calls)) def test_mode_oom_check_runs_only_the_check(self): f = docker_fake() # no authority: the check changes nothing, so it needs none f.plan["mode"], f.plan["packages"] = "oom-check", [] rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(rep["oom_check"]["result"], "pass", rep) self.assertFalse(any("apt-get" in c[-1] or "dpkg-query" in c[-1] for c in f.calls), f.calls) self.assertEqual(len(f.oom_removed), 1) def test_mode_oom_check_keeps_the_refusals(self): f = docker_fake() f.plan["mode"], f.plan["packages"] = "oom-check", [] f.files["/etc/pve/lxc/9201.conf"] = "arch: amd64\n" rc, rep = run(f) self.assertEqual((rc, rep["refused"]["code"]), (2, "R10"), rep) self.assertFalse(hasattr(f, "oom_runs")) g = Fake() g.plan["mode"] = "oom-check" rc, rep = run(g) self.assertEqual((rc, rep["refused"]["code"]), (2, "R11"), rep) PVE = "Proxmox Debian Repository:stable" PVE_SET = [{"name": "pve-manager", "version": "9.2.21", "origin": "Proxmox Debian Repository"}, {"name": "libpve-common-perl", "version": "9.1.9", "origin": "Proxmox Debian Repository"}] def pve_fake(signed=None, ring0=False): f = Fake() f.installed.update({"pve-manager": "9.2.2", "libpve-common-perl": "9.1.1", "proxmox-kernel-helper": "9.0.4"}) f.live["pve-manager"] = {"9.2.21", "9.2.2"} f.live["libpve-common-perl"] = {"9.1.9", "9.1.1"} f.origins = {"pve-manager": PVE, "libpve-common-perl": PVE, "proxmox-kernel-helper": PVE, "shim-signed": PVE, "proxmox-firewall-data": PVE} f.plan = {"release_id": "os-pve-t1", "layer": "pve", "lane": "slow", "vmid": 9201, "mode": "apply", "packages": [dict(p) for p in PVE_SET]} if ring0: f.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": True}) if signed is not None: f.plan["signed"] = signed return f class PVELane(unittest.TestCase): """R-812 option A (`09` §3 decision 163): the host's Proxmox USERSPACE packages, slow lane, no kernel / boot / firmware / microcode (R14), no removal, a new package only from PVE_NEW_ALLOW. Red-proof: audits/day-2026-10-07/B/.""" def refused(self, f, code): rc, rep = run(f) self.assertEqual(rc, 2, rep) self.assertEqual(rep["refused"]["code"], code, rep) self.assertEqual(f.installed["pve-manager"], "9.2.2", "nothing may be installed on a refusal") return rep # THE RED TEST (design-R-812 §5): before the pve layer existed this plan was refused R12. def test_pve_manager_plan_is_installed_on_the_host(self): f = pve_fake(signed=signed_job(packages=PVE_SET, op="os_pve_step")) rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(f.installed["pve-manager"], "9.2.21") self.assertEqual(rep["authority"], "signed") self.assertEqual(rep["pve_manager"], "9.2.21", "pveversion after the step is reported") inst = [c for c in f.calls if c[0] == "host" and "install" in c[1] and "-s" not in c[1] and "-f" not in c[1] and "--print-uris" not in c[1]] self.assertTrue(inst, "the pve layer installs on the HOST") self.assertFalse([c for c in f.calls if c[0] == "guest" and "install" in c[2]], "nothing installed in the guest") self.assertEqual(sorted(rep["health_after"]["host_services"]), sorted(osapply.HOST_SERVICES)) def test_kernel_in_a_pve_plan_is_refused(self): f = pve_fake(ring0=True) f.plan["packages"].append({"name": "proxmox-kernel-7.0", "version": "7.0.14-20", "origin": "Proxmox Debian Repository"}) self.refused(f, "R14") def test_shim_in_a_pve_plan_is_refused(self): f = pve_fake(ring0=True) f.plan["packages"].append({"name": "shim-signed", "version": "1.47+pmx1", "origin": "Proxmox Debian Repository"}) self.refused(f, "R14") def test_kernel_pulled_in_by_the_simulation_is_refused(self): f = pve_fake(ring0=True) f.installed["proxmox-kernel-helper"] = "9.0.4" f.extra_sim = ["Inst proxmox-kernel-helper [9.0.4] (9.0.6 Proxmox Debian Repository:stable [all])"] self.refused(f, "R6") # not in the plan — refused before the name check; R14 below when it IS in the plan g = pve_fake(ring0=True) g.live["proxmox-kernel-helper"] = {"9.0.6"} g.plan["packages"] = [dict(PVE_SET[0])] g.extra_sim = ["Inst proxmox-kernel-helper [9.0.4] (9.0.6 Proxmox Debian Repository:stable [all])"] # a kernel-helper the plan did not name is R6; the R14 name check covers a listed one (test above) self.refused(g, "R6") def test_debian_package_in_a_pve_plan_is_refused(self): f = pve_fake(ring0=True) f.plan["packages"].append({"name": "libc6", "version": "2.41-12+deb13u4", "origin": "Debian"}) self.refused(f, "R2") def test_debian_origin_in_the_pve_simulation_is_refused(self): f = pve_fake(ring0=True) f.origins["libpve-common-perl"] = DEB self.refused(f, "R2") def test_docker_package_in_a_pve_plan_is_refused(self): f = pve_fake(ring0=True) f.plan["packages"].append({"name": "docker-ce", "version": "5:29.8.2-1~debian.13~trixie", "origin": "Proxmox Debian Repository"}) self.refused(f, "R2") def test_pve_in_the_fast_lane_is_refused(self): f = pve_fake(ring0=True) f.plan["lane"] = "fast" self.refused(f, "R3") def test_no_authority_is_refused(self): self.refused(pve_fake(), "R3") def test_docker_signed_op_does_not_authorize_a_pve_step(self): self.refused(pve_fake(signed=signed_job(packages=PVE_SET, op="os_docker_step")), "R3") def test_pve_on_a_byo_box_is_refused(self): f = pve_fake(ring0=True) f.files[osapply.INSTALL_STATE] = json.dumps({"mode": "byo"}) self.refused(f, "R12") def test_unlisted_new_package_is_refused(self): f = pve_fake(ring0=True) f.extra_sim = ["Inst proxmox-new-thing (1.0 Proxmox Debian Repository:stable [all])"] self.refused(f, "R6") def test_allow_listed_new_package_is_accepted(self): f = pve_fake(ring0=True) f.extra_sim = ["Inst proxmox-firewall-data (0.1 Proxmox Debian Repository:stable [all])"] rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(f.installed["pve-manager"], "9.2.21") def test_allow_new_is_never_an_open_door_in_the_fast_lane(self): f = Fake() f.plan["layer"] = "host" f.extra_sim = ["Inst proxmox-firewall-data (0.1 Proxmox Debian Repository:stable [all])"] rc, rep = run(f) self.assertEqual((rc, rep["refused"]["code"]), (2, "R6"), rep) def test_undo_is_refused(self): f = pve_fake(signed=signed_job(packages=PVE_SET, op="os_pve_step")) f.plan["undo"] = True self.refused(f, "R5") def test_ring0_pending_pve_takes_only_installed_proxmox_userspace(self): f = pve_fake(ring0=True) f.plan["select"], f.plan["packages"] = "pending-pve", [] f.installed.update({"linux-image-amd64": "6.12.1", "tailscale": "1.102.2"}) f.pending_sim = [ "Inst pve-manager [9.2.2] (9.2.21 Proxmox Debian Repository:stable [amd64])", "Inst libpve-common-perl [9.1.1] (9.1.9 Proxmox Debian Repository:stable [all])", "Inst proxmox-kernel-helper [9.0.4] (9.0.6 Proxmox Debian Repository:stable [all])", "Inst proxmox-kernel-7.0.14-20-pve-signed (7.0.14-20 Proxmox Debian Repository:stable [amd64])", "Inst proxmox-firewall-data (0.1 Proxmox Debian Repository:stable [all])", "Inst libc6 [2.41-12+deb13u3] (2.41-12+deb13u4 Debian:13.7/stable [amd64])", "Inst tailscale [1.102.2] (1.102.5 Tailscale:pkgs.tailscale.com [amd64])", ] rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(rep["authority"], "ring0") self.assertEqual(sorted(u["name"] for u in rep["upgraded"]), ["libpve-common-perl", "pve-manager"], "pending-pve: installed, Proxmox-origin, never kernel/boot/firmware, never Debian or other origins") self.assertEqual(f.installed["libc6"], "2.41-12+deb13u3") # 2026-10-09, demo-hp (Secure Boot on): proxmox-secure-boot-support's upgrade pulls shim-signed-common; in the pve # plan that refused the whole step (R6) and skipped the night's kernel step. It is a boot-chain package: left out. # RED-PROOF: drop proxmox-secure-boot-support from HOST_SLOW_RE -> it is in the plan -> this test fails. def test_ring0_pending_pve_leaves_the_secure_boot_meta_out(self): f = pve_fake(ring0=True) f.plan["select"], f.plan["packages"] = "pending-pve", [] f.installed.update({"proxmox-secure-boot-support": "9.0.1"}) f.pending_sim = [ "Inst pve-manager [9.2.2] (9.2.21 Proxmox Debian Repository:stable [amd64])", "Inst proxmox-secure-boot-support [9.0.1] (9.0.2 Proxmox Debian Repository:stable [all])", "Inst shim-signed-common [1.48+pmx1+16.1-1+pmx1] (1.51+pmx1+16.1-2+pmx1 Proxmox Debian Repository:stable [all])", ] rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(sorted(u["name"] for u in rep["upgraded"]), ["pve-manager"]) self.assertEqual(f.installed["proxmox-secure-boot-support"], "9.0.1") # 2026-11-05..07, demo-felhom (EFI): proxmox-grub 2.12-9+pmx2 -> pmx4 depends on grub-efi-amd64 of the same version; # HOST_SLOW_RE's `grub` is anchored at the start, so proxmox-grub fell into the pve plan, the simulation pulled # grub-efi-amd64 and the whole pve step was refused R6 every night. It is a boot-chain package: left out. # RED-PROOF: drop proxmox-grub from HOST_SLOW_RE -> it is in the plan -> this test fails. def test_ring0_pending_pve_leaves_proxmox_grub_out(self): f = pve_fake(ring0=True) f.plan["select"], f.plan["packages"] = "pending-pve", [] f.installed.update({"proxmox-grub": "2.12-9+pmx2"}) f.pending_sim = [ "Inst pve-manager [9.2.2] (9.2.21 Proxmox Debian Repository:stable [amd64])", "Inst proxmox-grub [2.12-9+pmx2] (2.12-9+pmx4 Proxmox Debian Repository:stable [amd64])", "Inst grub-efi-amd64 [2.12-9+pmx2] (2.12-9+pmx4 Proxmox Debian Repository:stable [amd64])", ] rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(sorted(u["name"] for u in rep["upgraded"]), ["pve-manager"]) self.assertEqual(f.installed["proxmox-grub"], "2.12-9+pmx2") def test_select_pending_pve_needs_the_pve_layer(self): f = Fake() f.plan["layer"], f.plan["select"], f.plan["packages"] = "host", "pending-pve", [] rc, rep = run(f) self.assertEqual((rc, rep["refused"]["code"]), (2, "R11"), rep) # ---------- the kernel lane (R-836, `09` §3 decision 172, `11` §5.11) ---------- U = "1af1fcc6-639c-416b-a7e5-c4470d41a502" OLD, NEW = "7.0.2-6-pve", "7.0.14-22-pve" KERNEL_SET = [{"name": "proxmox-kernel-7.0", "version": "7.0.14-22", "origin": "Proxmox Debian Repository"}, {"name": "proxmox-kernel-7.0.14-22-pve-signed", "version": "7.0.14-22", "origin": "Proxmox Debian Repository"}] class KFake(Fake): """A Proxmox host with GRUB on UEFI: /boot on the root LV, a vfat ESP, the bundle's two generators. update-grub is emulated like the real 10_linux: WITHOUT the felhom default file the NEWEST kernel becomes the default (measured, R-836 — that is the defect the lane exists for); with it, the kernel it names.""" def __init__(self): super().__init__() self.running = OLD self.boot = {OLD} self.efi, self.esp_fs, self.boot_mount, self.mods, self.snippets, self.pin = True, "vfat", "", True, True, False self.env = None # None = no env block on the ESP; else {"felhom_next": ...} self.tree = {} # files put_file wrote self.reboots = [] self.update_grubs = 0 self.grub_runs_in_postinst = True self.installed.update({"proxmox-kernel-7.0": "7.0.2-6", "proxmox-default-kernel": "2.1.0", "pve-firmware": "3.18-3", "proxmox-kernel-7.0.2-6-pve-signed": "7.0.2-6", "pve-manager": "9.2.21"}) self.live.update({"proxmox-kernel-7.0": {"7.0.14-22", "7.0.2-6"}, "proxmox-kernel-7.0.14-22-pve-signed": {"7.0.14-22"}, "proxmox-kernel-7.0.14-23-pve-signed": {"7.0.14-23"}, "pve-firmware": {"3.18-7", "3.18-3"}}) self.origins = {} self.kernel_pending = ["Inst pve-firmware [3.18-3] (3.18-7 Proxmox Debian Repository:stable [all])", "Inst proxmox-kernel-7.0.14-22-pve-signed (7.0.14-22 Proxmox Debian Repository:stable [amd64])", "Inst proxmox-kernel-7.0 [7.0.2-6] (7.0.14-22 Proxmox Debian Repository:stable [amd64])", "Inst pve-manager [9.2.21] (9.2.22 Proxmox Debian Repository:stable [amd64])", "Inst libc6 [2.41-12+deb13u3] (2.41-12+deb13u4 Debian:13.7/stable [amd64])"] self.plan = {"release_id": "kernel-t1", "layer": "kernel", "lane": "slow", "vmid": 9201, "mode": "apply", "select": "pending-kernel", "packages": []} self.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": True}) self.files[osapply.CRASH_GUARD_STATE] = json.dumps({"armed": True, "tripped": False, "unclean_boots_in_window": 0}) self.cfg = self.render() # -- GRUB -- def newest(self): best = None for k in self.boot: if best is None or dpkg_cmp(k[:-4], "gt", best[:-4]): best = k return best def render(self): d = self.tree.get(osapply.KERNEL_DEFAULT_CFG) if d: dflt = re.search(r'GRUB_DEFAULT="([^"]+)"', d).group(1) else: dflt = f"gnulinux-advanced-{U}>gnulinux-{self.newest()}-advanced-{U}" cfg = ('if [ "${next_entry}" ] ; then\n set default="${next_entry}"\nelse\n' f' set default="{dflt}"\nfi\n' f"submenu 'Advanced options' $menuentry_id_option 'gnulinux-advanced-{U}' {{\n") for k in sorted(self.boot): cfg += f" menuentry 'Proxmox VE, with Linux {k}' $menuentry_id_option 'gnulinux-{k}-advanced-{U}' {{ }}\n" cfg += "}\n" if self.snippets: cfg += "### BEGIN /etc/grub.d/01_felhom_oneshot ###\nload_env felhom_next\n" cfg += "".join(f"menuentry 'Felhom one-shot: {k}' --id felhom-oneshot-{k} {{ }}\n" for k in sorted(self.boot)) return cfg def lexists(self, p): if p == "/sys/firmware/efi": return self.efi if p.startswith("/usr/lib/grub/x86_64-efi/"): return self.mods if p in (osapply.ONESHOT_SNIPPET, osapply.ONESHOT_ENTRIES): return self.snippets if p == osapply.KERNEL_PIN_FILE: return self.pin m = re.match(r"^/boot/(vmlinuz|initrd\.img)-(.+)$", p) if m: return m.group(2) in self.boot return p in self.tree def put_file(self, path, data, mode): self.tree[path] = data.decode() def read_file(self, p): if p == osapply.GRUB_CFG: return self.cfg if p in self.tree: return self.tree[p] return super().read_file(p) def host(self, argv, timeout=600, stdin=None): if argv[0] == "uname": self.calls.append(("host", argv)) return 0, self.running + "\n", "" if argv[0] == "findmnt": self.calls.append(("host", argv)) if argv[-1] == osapply.ESP_MOUNT: return (0, self.esp_fs + "\n", "") if self.esp_fs else (1, "", "") return (0, self.boot_mount + "\n", "") if self.boot_mount else (1, "", "") if argv[0] == "grub-editenv": self.calls.append(("host", argv)) if argv[2] == "list": return (1, "", "no such file") if self.env is None else \ (0, "".join(f"{k}={v}\n" for k, v in self.env.items()), "") if argv[2] == "create": self.env = {} return 0, "", "" if argv[2] == "set": k, v = argv[3].split("=", 1) self.env[k] = v return 0, "", "" if argv[2] == "unset": self.env.pop(argv[3], None) return 0, "", "" if argv[0] == "update-grub": self.calls.append(("host", argv)) self.update_grubs += 1 self.cfg = self.render() return 0, "", "" if argv[0] == "mkdir": self.calls.append(("host", argv)) return 0, "", "" if argv[:2] == ["systemctl", "reboot"]: self.calls.append(("host", argv)) self.reboots.append(self.running) return 0, "", "" return super().host(argv, timeout, stdin) def emulate(self, argv): a = [x for x in argv if not re.match(r"^[A-Z_]+=", x) and x != "env"] if a[0] == "apt-get" and "install" in a and "-s" not in a and "--print-uris" not in a and "-f" not in a: if self.install_rc: return self.install_rc, "", "E: boom" for x in a: if "=" in x and not x.startswith("-") and "::" not in x: n, v = x.split("=", 1) self.installed[n] = v m = re.match(r"^proxmox-kernel-[0-9]+\.[0-9]+$", n) if m: self.installed[f"proxmox-kernel-{v}-pve-signed"] = v if m or osapply.KERNEL_IMAGE_RE.match(n): self.boot.add(v + "-pve") if self.grub_runs_in_postinst: self.cfg = self.render() # the kernel's postinst runs update-grub (zz-update-grub) return 0, "Setting up proxmox-kernel ...\n", "" return super().emulate(argv) def sim(self, a): if "--print-uris" in a: return 0, "", "" if "dist-upgrade" in a: return 0, "\n".join(self.kernel_pending) + "\n", "" out, named = "", set() for x in a: if "=" in x and not x.startswith("-") and "::" not in x: named.add(x.split("=", 1)[0]) for x in a: if "=" in x and not x.startswith("-") and "::" not in x: n, v = x.split("=", 1) if v not in self.avail(n): return 100, "", f"E: Version '{v}' for '{n}' was not found" o = self.origins.get(n, PVE) out += f"Inst {n} [{self.installed[n]}] ({v} {o} [amd64])\n" if n in self.installed else f"Inst {n} ({v} {o} [amd64])\n" if re.match(r"^proxmox-kernel-[0-9]+\.[0-9]+$", n): img = f"proxmox-kernel-{v}-pve-signed" if img not in self.installed and img not in named: out += f"Inst {img} ({v} {PVE} [amd64])\n" out += "".join(l + "\n" for l in self.extra_sim) return 0, out, "" def kfake(**kw): f = KFake() for k, v in kw.items(): setattr(f, k, v) return f def kmode(f, mode, **extra): f.plan = {"release_id": "kernel-t1", "layer": "kernel", "lane": "slow", "vmid": 9201, "mode": mode, "packages": []} f.plan.update(extra) return run(f) def staged(f=None): f = f or kfake() rc, rep = run(f) assert rc == 0, rep return f class KernelLane(unittest.TestCase): """R-836 / `09` §3 decision 172: stage a kernel once through the ESP flag; the default moves only after a healthy one-shot boot. Red-proof: audits/kernel-lane-2026-10-07/A/redproof.txt.""" def refused(self, f, code, mode=None, **extra): rc, rep = kmode(f, mode, **extra) if mode else run(f) self.assertEqual(rc, 2, rep) self.assertEqual(rep["refused"]["code"], code, rep) return rep # --- the four red tests the brief names (Part A 3) --- def test_installing_a_kernel_does_not_change_the_grub_default(self): f = kfake() rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertIn(NEW, f.boot, "the new kernel was installed") self.assertEqual(osapply.Kernel.default_kver(f.cfg), OLD, "the default must stay the kernel the box runs (R-836: an install made the new one default)") self.assertIn(f"gnulinux-{OLD}-advanced-{U}", f.tree[osapply.KERNEL_DEFAULT_CFG]) self.assertEqual(f.env, {"felhom_next": NEW}, "the ONLY way to the new kernel is the one-shot flag") self.assertEqual(f.reboots, [], "staging never reboots") st = json.loads(f.tree[osapply.KERNEL_STATE]) self.assertEqual((st["phase"], st["from"], st["to"]), ("staged", OLD, NEW)) self.assertEqual(rep["kernel"]["default"], OLD) def test_a_box_without_the_esp_flag_is_refused(self): for attr, val, frag in (("esp_fs", "ext4", "vfat"), ("efi", False, "UEFI"), ("snippets", False, "bundle"), ("mods", False, "module"), ("boot_mount", "/boot", "separate"), ("pin", True, "pinned")): f = kfake(**{attr: val}) rep = self.refused(f, "R20") self.assertIn(frag, rep["refused"]["reason"], attr) self.assertNotIn(NEW, f.boot, f"{attr}: nothing may be installed on a refusal") self.assertIsNone(f.env, f"{attr}: no flag on a refusal") def test_a_reboot_without_the_flag_is_refused(self): f = staged() f.env = {"felhom_next": ""} self.refused(f, "R22", mode="kernel-reboot") self.assertEqual(f.reboots, []) def test_a_kernel_outside_the_approved_set_is_refused(self): # a signed set that names only the meta-package: the image the sources pull in was never signed for f = kfake() f.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": False}) f.plan.update(select="listed", packages=[dict(KERNEL_SET[0])], signed=signed_job(packages=[KERNEL_SET[0]], op="os_kernel_step")) self.refused(f, "R23") self.assertNotIn(NEW, f.boot) # a signed set names 7.0.14-22; the sources would ALSO bring 7.0.14-23 f2 = kfake() f2.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": False}) f2.plan.update(select="listed", packages=[dict(p) for p in KERNEL_SET], signed=signed_job(packages=KERNEL_SET, op="os_kernel_step")) f2.extra_sim = ["Inst proxmox-kernel-7.0.14-23-pve-signed (7.0.14-23 Proxmox Debian Repository:stable [amd64])"] self.refused(f2, "R23") # a name that is not a kernel package at all g = kfake() g.plan.update(select="listed", packages=[{"name": "pve-manager", "version": "9.2.22", "origin": "Proxmox Debian Repository"}]) self.refused(g, "R23") # the household was told about another kernel h = kfake() h.plan["expect_kver"] = "7.0.14-23-pve" self.refused(h, "R23") self.assertNotIn(NEW, h.boot) def test_the_snippet_clears_the_flag_on_use(self): """01_felhom_oneshot: the flag is copied, CLEARED and SAVED before any `set default`, all inside the one branch that runs only when the flag is set; a default is set only for an installed kernel's own entry.""" text = (HERE / "felhom-grub-oneshot.sh").read_text() body = text[text.index("cat <gnulinux-9.9.9-1-pve-advanced-{U}"\n' f.cfg = f.render() self.refused(f, "R20", mode="kernel-revert") self.assertEqual(len(f.reboots), 1) def test_cancel_clears_the_flag(self): f = staged() rc, rep = kmode(f, "kernel-cancel") self.assertEqual(rc, 0, rep) self.assertFalse(f.env.get("felhom_next"), "the flag is gone") self.assertEqual(rep["kernel"]["phase"], "cancelled") def test_status_is_read_only_and_names_setup_problems(self): f = kfake(esp_fs="ext4") rc, rep = kmode(f, "kernel-status") self.assertEqual(rc, 0, rep) self.assertTrue(rep["kernel"]["setup_problems"]) self.assertEqual(f.tree, {}) self.assertFalse([c for c in f.calls if c[1][0] in ("update-grub", "systemctl", "apt-get")]) def test_facts_carry_the_kernel_lane(self): f = staged() f.plan = {"release_id": "facts", "layer": "host", "lane": "fast", "vmid": 9201, "mode": "facts", "packages": []} rc, rep = run(f) self.assertEqual(rc, 0, rep) h = rep["facts"]["host"] self.assertEqual(h["kernel_lane"]["phase"], "staged") self.assertEqual(h["kernel_next_boot"], NEW) self.assertIn("one-shot", h["kernel_next_boot_source"]) # ---------- the clean-up of old kernels (R-1368, `09` §3 decision 286, `11` §5.11.1) ---------- K = ["7.0.2-6-pve", "7.0.14-20-pve", "7.0.14-22-pve", "7.0.14-26-pve", "7.0.14-28-pve"] # five installed kernels C_FROM, C_RUN = "7.0.14-26-pve", "7.0.14-28-pve" GB = 1024 ** 3 def img(k): return f"proxmox-kernel-{k}-signed" class CFake(KFake): """A ring-0 appliance after a good kernel step 26 -> 28: five kernels installed, the box runs its newest, the default (grub.cfg and the pinned file) is the running kernel, the flag is empty. `apt-get purge` is emulated like the real one: the image leaves dpkg and /boot, and the kernel's postrm runs update-grub.""" def __init__(self): super().__init__() self.running = C_RUN self.boot = set(K) self.modules = set(K) # /lib/modules/ for n in [x for x in self.installed if osapply.KERNEL_IMAGE_RE.match(x)]: del self.installed[n] for k in K: self.installed[img(k)] = k[:-4] self.installed["proxmox-kernel-7.0"] = C_RUN[:-4] self.env = {"felhom_next": ""} self.tree[osapply.KERNEL_DEFAULT_CFG] = (f'GRUB_DEFAULT="gnulinux-advanced-{U}>gnulinux-{C_RUN}-advanced-{U}"\n') self.state = {"phase": "good", "step_id": "ring0-x", "from": C_FROM, "to": C_RUN, "default_before": C_FROM, "vmid": 9201, "staged_at": "2026-10-01T02:50:00Z", "rebooted_at": "2026-10-01T02:51:00Z", "result_at": "2026-10-01T03:10:00Z", "self_revert_used": False} self.write_state() self.cfg = self.render() self.purges, self.purge_rc, self.purge_moves_default, self.purge_extra_sim = [], 0, False, [] self.plan = {"release_id": "kernel", "layer": "kernel", "lane": "slow", "vmid": 9201, "mode": "kernel-cleanup", "packages": []} def write_state(self): if self.state is None: self.tree.pop(osapply.KERNEL_STATE, None) else: self.tree[osapply.KERNEL_STATE] = json.dumps(self.state) def lexists(self, p): m = re.match(r"^/lib/modules/(.+)$", p) if m: return m.group(1) in self.modules return super().lexists(p) def changing(self): """Every command that changes the box (the purge, a repair, update-grub, a flag write, a file write).""" out = [] for c in self.calls: a = [x for x in c[-1] if not re.match(r"^[A-Z_]+=", x) and x != "env"] if (a[0] == "apt-get" and "-s" not in a and "--print-uris" not in a) or a[0] in ("update-grub", "dpkg-reconfigure") \ or (a[0] == "dpkg" and a[1] == "--configure") or (a[0] == "grub-editenv" and a[2] != "list") \ or (a[0] == "systemctl" and a[1] == "reboot"): out.append(a) return out def emulate(self, argv): a = [x for x in argv if not re.match(r"^[A-Z_]+=", x) and x != "env"] if a[0] == "apt-get" and "purge" in a and "-s" not in a: names = [x for x in a[a.index("purge") + 1:] if not x.startswith("-")] self.purges.append(names) if self.purge_rc: self.dpkg_journal = ["0001"] # a purge cut in the middle leaves dpkg's update journal return self.purge_rc, "", "E: Sub-process /usr/bin/dpkg returned an error code (1)" for n in names: self.installed.pop(n, None) m = osapply.KERNEL_IMAGE_RE.match(n) if m: self.boot.discard(m.group(1)) self.free += GB if self.purge_moves_default: self.tree.pop(osapply.KERNEL_DEFAULT_CFG, None) self.cfg = self.render() # the kernel's postrm runs update-grub (zz-update-grub) return 0, "Removing proxmox-kernel ...\n", "" return super().emulate(argv) def sim(self, a): if "purge" in a: names = [x for x in a[a.index("purge") + 1:] if not x.startswith("-")] if getattr(self, "purge_sim_rc", 0): return self.purge_sim_rc, "", "E: Unable to locate package" out = "".join(f"Purg {n} [{self.installed[n]}]\n" for n in names if n in self.installed) return 0, out + "".join(l + "\n" for l in self.purge_extra_sim), "" return super().sim(a) def cfake(**kw): f = CFake() for k, v in kw.items(): setattr(f, k, v) return f def cstate(f, **kw): f.state.update(kw) f.write_state() return f class KernelCleanup(unittest.TestCase): """§5 of the release-2a brief: one test per acceptance scenario, each on the stored effect (the exact apt line or the absence of any changing command, the report, the exit code). Red-proofs (the four mutations the brief names): felhom.eu audits/release-2a-2026-11-14/A/redproof.txt.""" REMOVE = [img(k) for k in ("7.0.14-20-pve", "7.0.14-22-pve", "7.0.2-6-pve")] # sorted by name def nothing_changed(self, f): self.assertEqual(f.purges, [], "no purge") self.assertEqual(f.changing(), [], "no command that changes the box") self.assertEqual(sorted(f.boot), sorted(K), "every kernel is still there") def refused(self, f, code): before = (f.tree.get(osapply.KERNEL_STATE), f.tree.get(osapply.KERNEL_DEFAULT_CFG), f.cfg, dict(f.env or {})) rc, rep = run(f) self.assertEqual(rc, 2, rep) self.assertEqual(rep["refused"]["code"], code, rep) self.nothing_changed(f) self.assertEqual((f.tree.get(osapply.KERNEL_STATE), f.tree.get(osapply.KERNEL_DEFAULT_CFG), f.cfg, dict(f.env or {})), before) return rep def not_run(self, f, code): rc, rep = run(f) self.assertEqual(rc, 0, rep) c = rep["kernel_cleanup"] self.assertEqual(c["outcome"], "not_run", rep) self.assertIn(code, [x["code"] for x in c["not_run"]], rep) self.assertIsNone(rep.get("refused")) self.nothing_changed(f) return rep # --- scenario 1: the normal night --- def test_s1_the_normal_night_purges_exactly_the_old_kernels_in_one_command(self): f = cfake() state0, pin0 = f.tree[osapply.KERNEL_STATE], f.tree[osapply.KERNEL_DEFAULT_CFG] rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(f.purges, [self.REMOVE], "ONE apt-get purge of the exact names") purge = [c[-1] for c in f.calls if "purge" in c[-1] and "-s" not in c[-1]] self.assertEqual(len(purge), 1) self.assertEqual(purge[0][purge[0].index("apt-get"):], ["apt-get", "-y", "-q", "purge"] + self.REMOVE) self.assertNotIn("autoremove", " ".join(" ".join(c[-1]) for c in f.calls), "never apt's own clean-up") c = rep["kernel_cleanup"] self.assertEqual(c["outcome"], "removed") self.assertEqual(c["removed"], self.REMOVE) self.assertEqual(sorted(c["kept"]), sorted([C_FROM, C_RUN])) self.assertEqual(c["bytes_gained"], 3 * GB) self.assertEqual(sorted(f.boot), sorted([C_FROM, C_RUN])) # the default in grub.cfg, the pinned default file and the step's state file are unchanged self.assertEqual(osapply.Kernel.default_kver(f.cfg), C_RUN) self.assertEqual(f.tree[osapply.KERNEL_DEFAULT_CFG], pin0) self.assertEqual(f.tree[osapply.KERNEL_STATE], state0) self.assertEqual(f.env, {"felhom_next": ""}) self.assertEqual(f.reboots, [], "the clean-up never reboots") self.assertEqual(f.update_grubs, 0, "the wrapper itself runs no update-grub (the kernel's postrm did)") # leftovers are reported, never deleted self.assertEqual(c["leftovers"], [f"/lib/modules/{k}" for k in ("7.0.14-20-pve", "7.0.14-22-pve", "7.0.2-6-pve")]) self.assertEqual(f.modules, set(K), "a left-over folder is reported, not deleted") # --- scenario 2: the second night --- def test_s2_the_second_night_removes_nothing(self): f = cfake() self.assertEqual(run(f)[0], 0) f.calls.clear() f.purges.clear() rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(rep["kernel_cleanup"]["outcome"], "nothing") self.assertEqual(rep["kernel_cleanup"]["removal"], []) self.assertEqual(f.purges, []) self.assertEqual(f.changing(), [], "no apt call that changes anything") self.assertFalse([c for c in f.calls if "purge" in c[-1]], "not even a simulation: there is nothing to simulate") # --- scenario 3: a step is not over (P1) --- def test_s3_a_step_that_is_not_over_refuses(self): for phase in ("staged", "oneshot", "judging", "reverting", "some-new-phase"): f = cstate(cfake(), phase=phase) rep = self.refused(f, "P1") self.assertIn(phase, rep["refused"]["reason"]) # --- scenario 4: the flag (P2); the pinned default against grub.cfg (P3) --- def test_s4_the_flag_names_a_kernel_refuses(self): f = cfake(env={"felhom_next": "7.0.14-26-pve"}) self.refused(f, "P2") def test_s4_pinned_default_and_grub_cfg_disagree_refuses(self): f = cfake() f.tree[osapply.KERNEL_DEFAULT_CFG] = f'GRUB_DEFAULT="gnulinux-advanced-{U}>gnulinux-{C_FROM}-advanced-{U}"\n' self.refused(f, "P3") # grub.cfg still says 28, the file says 26 g = cfake() del g.tree[osapply.KERNEL_DEFAULT_CFG] self.refused(g, "P3") # no pinned file at all h = cfake() # both name a kernel that is not installed h.tree[osapply.KERNEL_DEFAULT_CFG] = f'GRUB_DEFAULT="gnulinux-advanced-{U}>gnulinux-7.0.14-99-pve-advanced-{U}"\n' h.cfg = h.cfg.replace(f"gnulinux-{C_RUN}-advanced-{U}\"", f"gnulinux-7.0.14-99-pve-advanced-{U}\"", 1) self.refused(h, "P3") # --- scenario 5: the box fell back / does not run its newest kernel --- def test_s5_a_box_that_fell_back_keeps_every_kernel(self): for phase in ("fell_back", "self_reverted", "revert_failed", "cancelled"): self.not_run(cstate(cfake(running=C_FROM), phase=phase), "step-not-good") def test_s5_a_box_that_does_not_run_its_newest_kernel_keeps_every_kernel(self): f = cfake(running=C_FROM) # the step says good, but the box runs the older kernel self.not_run(f, "not-on-newest") # --- scenario 6: the wrong case (K1) — one case per kept-kernel kind --- def k1(self, kind, prepare): """The keep computation is BROKEN on purpose (it forgets `kind`); K1 reads the box by itself and must refuse before any apt call.""" f = cfake() victim = prepare(f) real = osapply.Kernel.cleanup_keep def broken(self_, images, sources): keep = real(self_, images, [(k, why) for k, why in sources if k != victim]) keep.pop(victim, None) # … also when it is the newest installed return keep osapply.Kernel.cleanup_keep = broken try: rc, rep = run(f) finally: osapply.Kernel.cleanup_keep = real self.assertEqual(rc, 2, f"{kind}: {rep}") self.assertEqual(rep["refused"]["code"], "K1", f"{kind}: {rep}") self.assertIn(img(victim), rep["refused"]["reason"]) self.assertFalse([c for c in f.calls if "purge" in c[-1]], f"{kind}: refused BEFORE any apt call, also the simulation") self.nothing_changed(f) def test_s6_k1_the_running_kernel(self): self.k1("running", lambda f: C_RUN) def test_s6_k1_the_steps_from(self): self.k1("from", lambda f: C_FROM) def test_s6_k1_the_steps_to(self): def prep(f): # a step whose `to` is not the running kernel any more cstate(f, to="7.0.14-22-pve") return "7.0.14-22-pve" self.k1("to", prep) def test_s6_k1_default_before(self): def prep(f): cstate(f, default_before="7.0.14-20-pve") return "7.0.14-20-pve" self.k1("default_before", prep) def test_s6_k1_the_grub_cfg_default_and_the_pinned_file(self): # the default (both places) names 26; the box runs 28 after a hand boot. The keep computation forgets 26. def prep(f): f.tree[osapply.KERNEL_DEFAULT_CFG] = f'GRUB_DEFAULT="gnulinux-advanced-{U}>gnulinux-7.0.14-22-pve-advanced-{U}"\n' f.cfg = f.render() return "7.0.14-22-pve" self.k1("default", prep) def test_s6_k1_the_flag(self): # a flag is a refusal by itself (P2, first); with the keep computation broken, K1 is in the SAME refusal f = cfake(env={"felhom_next": "7.0.14-20-pve"}) real = osapply.Kernel.cleanup_keep osapply.Kernel.cleanup_keep = lambda s_, images, sources: real(s_, images, [(k, w) for k, w in sources if k != "7.0.14-20-pve"]) try: rc, rep = run(f) finally: osapply.Kernel.cleanup_keep = real self.assertEqual((rc, rep["refused"]["code"]), (2, "P2"), rep) self.assertIn("K1 " + img("7.0.14-20-pve") + " is a keep-kernel (the one-shot flag)", rep["refused"]["reason"]) self.assertFalse([c for c in f.calls if "purge" in c[-1]]) self.nothing_changed(f) def test_s6_k1_an_operators_boot_pin(self): def prep(f): f.tree[osapply.KERNEL_PIN_FILE] = "7.0.14-20-pve\n" return "7.0.14-20-pve" self.k1("boot pin", prep) def test_s6_every_kept_kind_is_in_the_keep_set_and_never_in_the_removal_set(self): f = cfake() cstate(f, default_before="7.0.14-20-pve") f.tree[osapply.KERNEL_PIN_FILE] = "7.0.14-22-pve\n" f.plan["dry_run"] = True rc, rep = run(f) c = rep["kernel_cleanup"] self.assertEqual(sorted(c["kept"]), sorted(["7.0.14-20-pve", "7.0.14-22-pve", C_FROM, C_RUN])) self.assertEqual(c["removal"], [img("7.0.2-6-pve")]) def test_s6_the_running_kernel_is_not_an_installed_package(self): f = cfake() del f.installed[img(C_RUN)] # the box runs a kernel dpkg does not know (an unsigned hand install, say) f.installed["proxmox-kernel-7.0.14-28-pve-other"] = "7.0.14-28" self.refused(f, "P3") # P3 is first (the default names a kernel that is not an installed image) … rc, rep = run(f) self.assertIn("K1", rep["refused"]["reason"], "… and K1 is in the same refusal") def test_no_step_record_keeps_the_next_lower_kernel(self): f = cstate(cfake(), **{"from": None, "to": None, "default_before": None}) f.plan["dry_run"] = True rc, rep = run(f) self.assertIn("7.0.14-26-pve", rep["kernel_cleanup"]["kept"]) self.assertIn("next lower", " ".join(rep["kernel_cleanup"]["kept"]["7.0.14-26-pve"])) # --- scenario 7: apt would do more (S1, S2) --- def test_s7_the_simulation_would_remove_something_else(self): f = cfake(purge_extra_sim=["Purg proxmox-kernel-7.0 [7.0.14-28]", "Remv proxmox-ve [9.2.0]"]) rep = self.refused(f, "S2") self.assertIn("proxmox-ve", rep["refused"]["reason"]) def test_s7_the_simulation_would_install_something(self): f = cfake(purge_extra_sim=["Inst proxmox-kernel-7.0.14-28-pve (7.0.14-28 Proxmox Debian Repository:stable [amd64])"]) rep = self.refused(f, "S1") self.assertIn("proxmox-kernel-7.0.14-28-pve", rep["refused"]["reason"]) def test_s7_the_simulation_fails(self): self.refused(cfake(purge_sim_rc=100), "S1") # --- scenario 8: not our box --- def test_s8_a_byo_host_is_refused_by_the_appliance_rule(self): f = cfake() f.files[osapply.INSTALL_STATE] = json.dumps({"mode": "byo"}) self.refused(f, "R12") def test_s8_a_box_without_the_ring0_mark_does_not_run(self): f = cfake() f.files[osapply.TRUST_FILE] = json.dumps({"host_id": "tester-1-d70be4", "ring0_slow_lane": False}) rep = self.not_run(f, "no-ring0-mark") self.assertIn("ring 0 only in this release", rep["kernel_cleanup"]["not_run"][0]["reason"]) g = cfake() # an agent-writable trust file proves nothing g.stats[osapply.TRUST_FILE] = St(mode=statmod.S_IFREG | 0o666, uid=0) self.not_run(g, "no-ring0-mark") h = cfake() # the agent's plan cannot claim the ring h.files[osapply.TRUST_FILE] = json.dumps({"host_id": "x", "ring0_slow_lane": False}) h.plan["ring"] = 0 self.not_run(h, "no-ring0-mark") # --- scenario 9: a plan that names packages --- def test_s9_cleanup_plan_names_refused(self): for extra in ({"packages": [{"name": img("7.0.14-28-pve"), "version": "7.0.14-28", "origin": "Proxmox Debian Repository"}]}, {"select": "pending-kernel"}, {"allow_new": ["x"]}, {"signed": {"blob_b64": "eA==", "sig": "x"}}, {"expect_kver": "7.0.14-20-pve"}, {"dry_run": "yes"}): f = cfake() f.plan.update(extra) self.refused(f, "C1") # --- scenario 10: no kernel step ever ran --- def test_s10_no_state_file_does_not_run(self): f = cfake() f.state = None f.write_state() rep = self.not_run(f, "no-step") self.assertIn("no kernel step yet", rep["kernel_cleanup"]["not_run"][0]["reason"]) # --- scenario 11: a power cut in the purge --- def test_s11_a_cut_purge_is_repaired_first_and_the_rule_is_computed_again(self): f = cfake(purge_rc=100) rc, rep = run(f) self.assertEqual(rc, 3, rep) self.assertEqual(rep["failed"]["step"], "purge") self.assertEqual(rep["kernel_cleanup"]["outcome"], "failed") # the cut removed ONE of the three before it stopped f.installed.pop(self.REMOVE[0]) f.boot.discard("7.0.14-20-pve") f.purge_rc, f.purges = 0, [] f.calls.clear() rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(rep["repair"]["journal_before"], 1, "repair first: dpkg's journal was replayed") i_repair = [i for i, c in enumerate(f.calls) if c[-1][-3:-1] == ["--configure", "-a"] or "--configure" in c[-1]][0] i_purge = [i for i, c in enumerate(f.calls) if "purge" in c[-1]][0] self.assertLess(i_repair, i_purge, "the repair runs BEFORE the rule is computed and before any purge") self.assertEqual(f.purges, [self.REMOVE[1:]], "the rule was computed again from the box: the two that are left") self.assertEqual(rep["kernel_cleanup"]["outcome"], "removed") # --- the 20-hour window, the dry run, after-the-purge proof --- def test_inside_a_steps_20_hour_window_it_does_not_run(self): f = cfake() cstate(f, result_at=osapply._iso(f.clock - 3600)) # the step ended good one hour ago self.not_run(f, "window") def test_the_dry_run_reports_and_changes_nothing(self): f = cfake() f.plan["dry_run"] = True rc, rep = run(f) self.assertEqual(rc, 0, rep) c = rep["kernel_cleanup"] self.assertEqual((c["outcome"], c["would"], c["removal"]), ("dry_run", "remove", self.REMOVE)) self.assertEqual(sorted(c["kept"]), sorted([C_FROM, C_RUN])) self.nothing_changed(f) self.assertNotIn("repair", rep, "a dry run does not even repair") # a dry run of a refusal says which one, with exit 0 g = cstate(cfake(), phase="judging") g.plan["dry_run"] = True rc, rep = run(g) self.assertEqual((rc, rep["kernel_cleanup"]["would"]), (0, "refuse"), rep) self.assertEqual([x["code"] for x in rep["kernel_cleanup"]["refusals"]], ["P1"]) self.nothing_changed(g) # … and of a box that would not run h = cfake() h.files[osapply.TRUST_FILE] = json.dumps({"host_id": "x", "ring0_slow_lane": False}) h.plan["dry_run"] = True rc, rep = run(h) self.assertEqual((rc, rep["kernel_cleanup"]["would"], rep["kernel_cleanup"]["removal"]), (0, "not_run", self.REMOVE), rep) self.nothing_changed(h) def test_a_purge_that_moves_the_default_is_reported_failed_loudly(self): f = cfake(purge_moves_default=True) # the pinned file vanished in the purge: 10_linux makes the newest the default rc, rep = run(f) self.assertEqual(rc, 3, rep) self.assertEqual(rep["failed"]["step"], "verify") self.assertIn("pinned default file changed", rep["failed"]["reason"]) self.assertEqual(f.update_grubs, 0, "never repaired by hand in code") self.assertTrue(any("FAILED" in l and "verify" in l for l in f.logs)) def test_the_apt_lock_is_respected(self): self.refused(cfake(lock_held=True), "R9") def test_both_image_packages_of_one_old_kernel_go_together(self): f = cfake() f.installed["proxmox-kernel-7.0.14-20-pve"] = "7.0.14-20" # the unsigned image beside the signed one rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertIn("proxmox-kernel-7.0.14-20-pve", f.purges[0]) self.assertIn(img("7.0.14-20-pve"), f.purges[0]) def test_cleanup_needs_no_new_sudo_line(self): """The wrapper is reached by `--plan` only: the new mode rides the one sudo line the agent already has.""" sudo = (HERE / "felhom-agent.sudoers").read_text() live = [l.strip().rstrip("\\,").strip() for l in sudo.splitlines() if l.strip() and not l.strip().startswith("#")] self.assertEqual([l for l in live if "felhom-os-apply" in l and "Cmnd_Alias" not in l and "NOPASSWD" not in l], [osapply.SELF_CHECK_SUDO_LINE], "exactly the one --plan line reaches the wrapper") self.assertFalse([l for l in live if "kernel-cleanup" in l or ("apt" in l and ("purge" in l or "remove" in l))], "no sudo line removes a package") # ---------- the security-profile refresh (R-1388, R-1301, `11` §5.12) ---------- class PFake(Fake): """A host with two AppArmor cache folders. `stale`: profile files whose cached copy in the running kernel's folder is an old compile; the refresh command rewrites the cache from `fresh`. lxc-containers compiles differently every time (measured on demo-hp). `failed`: unit -> ExecStart program.""" def __init__(self, stale=("usr.sbin.chronyd",), failed=None, kernel="7.0.14-28-pve", record=None): super().__init__() self.kernel = kernel self.fresh = {"usr.sbin.chronyd": b"chrony-4.8", "usr.bin.man": b"man", "lxc-containers": b"lxc-0", "nvidia_modprobe": b"nv"} self.names = {"usr.sbin.chronyd": "/usr/sbin/chronyd", "usr.bin.man": "/usr/bin/man\nman_filter\nman_groff", "lxc-containers": "lxc-container-default\nlxc-container-default-cgns", "nvidia_modprobe": "nvidia_modprobe\nnvidia_modprobe//kmod"} self.cache = {"ae1a4050.0": {k: b"old-" + k.encode() for k in self.fresh}, "f6da0254.0": dict(self.fresh)} for k in stale: self.cache["f6da0254.0"][k] = b"OLD-COPY-" + k.encode() self.failed = dict({"chrony.service": "/usr/sbin/chronyd"} if failed is None else failed) self.restarts, self.parser_runs, self.parser_rc, self.tree = [], 0, 0, {} if record is not None: self.tree[osapply.PROFILE_STATE] = json.dumps(record) self.plan = {"release_id": "profile-refresh", "layer": "host", "mode": "profile-refresh"} def isdir(self, p): return p in (osapply.PROFILE_DIR, osapply.PROFILE_CACHE) or p.startswith(osapply.PROFILE_CACHE + "/") and p.count("/") == 4 def list_dir(self, p): if p == osapply.PROFILE_CACHE: return sorted(self.cache) d = p.rsplit("/", 1)[1] return sorted(list(self.cache.get(d, {})) + [".features"]) def read_bytes(self, p): d, f = p.split("/")[-2:] return self.cache[d][f] def put_file(self, path, data, mode): self.tree[path] = data.decode() def read_file(self, p): if p in self.tree: return self.tree[p] return super().read_file(p) def host(self, argv, timeout=600, stdin=None): if argv[0] == "uname": self.calls.append(("host", argv)) return 0, self.kernel + "\n", "" if argv == osapply.PROFILE_REFRESH_CMD: self.calls.append(("host", argv)) self.parser_runs += 1 if self.parser_rc: return self.parser_rc, "", "AppArmor parser error for /etc/apparmor.d/x" self.fresh["lxc-containers"] = b"lxc-%d" % self.parser_runs # never the same binary twice self.cache["f6da0254.0"] = dict(self.fresh) return 0, "", "" if argv[:2] == ["apparmor_parser", "-N"]: self.calls.append(("host", argv)) return 0, self.names[argv[2].rsplit("/", 1)[1]] + "\n", "" if argv[:2] == ["systemctl", "list-units"]: self.calls.append(("host", argv)) return 0, "".join(f"{u} loaded failed failed x\n" for u in self.failed), "" if argv[:2] == ["systemctl", "show"]: self.calls.append(("host", argv)) return 0, f"{{ path={self.failed[argv[-1]]} ; argv[]={self.failed[argv[-1]]} -F 1 ; ignore_errors=no }}\n", "" if argv[:2] == ["systemctl", "restart"]: self.calls.append(("host", argv)) self.restarts.append(argv[2]) return 0, "", "" return super().host(argv, timeout, stdin) class ProfileRefresh(unittest.TestCase): """Red-proofs: felhom.eu audits/release-2a-2026-11-14/B/redproof.txt.""" def test_a_stale_profile_is_refreshed_and_only_its_failed_unit_restarts(self): f = PFake(failed={"chrony.service": "/usr/sbin/chronyd", "postfix.service": "/usr/sbin/postfix", "other.service": "/usr/bin/man-not"}) rc, rep = run(f) self.assertEqual(rc, 0, rep) p = rep["profile_refresh"] self.assertEqual(p["outcome"], "refreshed") self.assertEqual(f.parser_runs, 1) self.assertIn(("host", ["apparmor_parser", "--replace", "--skip-read-cache", "--write-cache", "--", "/etc/apparmor.d"]), f.calls) self.assertEqual(f.cache["f6da0254.0"]["usr.sbin.chronyd"], b"chrony-4.8", "the cache holds the fresh compile") self.assertEqual(p["rewritten"], ["f6da0254.0/lxc-containers", "f6da0254.0/usr.sbin.chronyd"]) self.assertEqual(f.restarts, ["chrony.service"], "ONLY the failed unit whose own program's profile was stale") self.assertEqual([x["unit"] for x in p["restarted"]], ["chrony.service"]) rec = json.loads(f.tree[osapply.PROFILE_STATE]) self.assertEqual(rec["kernel"], "7.0.14-28-pve") def test_a_fresh_cache_changes_nothing_and_restarts_nothing(self): f = PFake(stale=()) # chrony is failed for another reason; its profile is NOT stale rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(rep["profile_refresh"]["rewritten"], ["f6da0254.0/lxc-containers"], "only the file that never compiles the same") self.assertEqual(f.restarts, [], "a failed unit whose profile was not stale is NOT restarted") self.assertEqual(f.cache["f6da0254.0"]["usr.sbin.chronyd"], b"chrony-4.8") def test_a_stale_profile_of_a_running_program_restarts_nothing(self): f = PFake(failed={}) # the unit is active: the kernel swaps the profile of a running task itself rc, rep = run(f) self.assertEqual((rc, f.restarts, rep["profile_refresh"]["outcome"]), (0, [], "refreshed"), rep) def test_a_failed_unit_of_another_program_is_not_restarted(self): f = PFake(stale=("usr.bin.man",), failed={"chrony.service": "/usr/sbin/chronyd"}) rc, rep = run(f) self.assertEqual((rc, f.restarts), (0, []), rep) g = PFake(stale=("nvidia_modprobe",), failed={"nvidia.service": "nvidia_modprobe"}) # a profile name that is not a path rc, rep = run(g) self.assertEqual((rc, g.restarts), (0, []), rep) def test_once_per_kernel(self): f = PFake() self.assertEqual(run(f)[0], 0) f.calls.clear() f.cache["f6da0254.0"]["usr.sbin.chronyd"] = b"OLD-COPY" # even with a stale file: this kernel was done rc, rep = run(f) self.assertEqual((rc, rep["profile_refresh"]["outcome"], f.parser_runs), (0, "not_needed", 1), rep) self.assertFalse([c for c in f.calls if c[1][0] in ("apparmor_parser", "systemctl")], "no parser, no systemctl") f.kernel = "7.0.14-29-pve" # the box starts on another kernel: again rc, rep = run(f) self.assertEqual((rc, rep["profile_refresh"]["outcome"], f.parser_runs), (0, "refreshed", 2), rep) self.assertEqual(json.loads(f.tree[osapply.PROFILE_STATE])["kernel"], "7.0.14-29-pve") def test_a_failed_parser_writes_no_record_so_the_next_start_tries_again(self): f = PFake() f.parser_rc = 1 rc, rep = run(f) self.assertEqual((rc, rep["failed"]["step"], rep["profile_refresh"]["outcome"]), (3, "profile-refresh", "failed"), rep) self.assertNotIn(osapply.PROFILE_STATE, f.tree) self.assertEqual(f.restarts, []) f.parser_rc = 0 self.assertEqual(run(f)[1]["profile_refresh"]["outcome"], "refreshed") def test_the_plan_names_nothing(self): for extra in ({"profiles": ["usr.sbin.chronyd"]}, {"units": ["ssh.service"]}, {"packages": [{"name": "chrony", "version": "4.8", "origin": "Debian"}]}, {"force": True}, {"dry_run": 1}, {"layer": "guest"}): f = PFake() f.plan.update(extra) rc, rep = run(f) self.assertEqual(rc, 2, rep) self.assertIn(rep["refused"]["code"], ("F1", "R11"), rep) self.assertEqual((f.parser_runs, f.restarts), (0, [])) def test_a_byo_host_is_refused(self): f = PFake() f.files[osapply.INSTALL_STATE] = json.dumps({"mode": "byo"}) rc, rep = run(f) self.assertEqual((rc, rep["refused"]["code"], f.parser_runs), (2, "R12", 0), rep) def test_the_dry_run_changes_nothing(self): f = PFake() f.plan["dry_run"] = True rc, rep = run(f) self.assertEqual((rc, rep["profile_refresh"]["outcome"], rep["profile_refresh"]["would"]), (0, "dry_run", "refresh"), rep) self.assertEqual((f.parser_runs, f.restarts), (0, [])) self.assertNotIn(osapply.PROFILE_STATE, f.tree) def test_no_more_than_three_restarts(self): f = PFake(failed={f"chrony{i}.service": "/usr/sbin/chronyd" for i in range(5)}) rc, rep = run(f) self.assertEqual((rc, len(f.restarts), len(rep["profile_refresh"]["not_restarted"])), (0, 3, 2), rep) def test_profile_refresh_changes_no_package(self): f = PFake() run(f) self.assertFalse([c for c in f.calls if c[1][0] in ("apt-get", "dpkg", "apt")], "no package command at all") # ---------- trust in Proxmox by a copy of its root CA (R-1380, D80, felhom.eu `03` §12a) ---------- CA_PEM = b"-----BEGIN CERTIFICATE-----\nQUJDREVGR0hJSktMTU5PUFFSU1RVVldYWVo=\n-----END CERTIFICATE-----\n" CA_PEM_2 = b"-----BEGIN CERTIFICATE-----\nWllYV1ZVVFNSUVBPTk1MS0pJSEdGRURDQkE=\n-----END CERTIFICATE-----\n" class TFake(Fake): """A Proxmox host: the root CA file, the copy (or none), and openssl's answer about the bytes it is given.""" def __init__(self, source=CA_PEM, copy=None, ca=True, ends="Aug 17 15:44:42 2036 GMT"): super().__init__() self.source, self.tree, self.ca, self.ends, self.openssl_rc = source, {}, ca, ends, 0 if copy is not None: self.tree[osapply.PVE_CA_COPY] = copy self.openssl_stdin, self.puts = [], [] self.serves_new = False # does pveproxy serve a certificate that verifies with the NEW root alone? self.dead_ca = set() # certificate texts openssl says are not a live CA (ended) self.handshakes = [] self.plan = {"release_id": "pve-ca", "layer": "host", "mode": "pve-ca-copy"} def read_bytes(self, p): if p == osapply.PVE_CA_SOURCE: if self.source is None: raise FileNotFoundError(p) return self.source if p in self.tree: return self.tree[p] raise FileNotFoundError(p) def put_file(self, path, data, mode): self.puts.append((path, data, mode)) self.tree[path] = data def host(self, argv, timeout=600, stdin=None): if argv[0] == "openssl" and argv[1] == "s_client": self.handshakes.append(argv) return (0, "Verify return code: 0 (ok)\n", "") if self.serves_new else (1, "", "Verify return code: 19 (self-signed certificate in certificate chain)\n") if argv[0] == "openssl" and "-checkend" in argv: return (1, "X509v3 Basic Constraints: critical\n CA:TRUE\nCertificate will expire\n", "") if stdin.strip() in self.dead_ca else \ (0, "X509v3 Basic Constraints: critical\n CA:TRUE\nCertificate will not expire\n", "") if argv[0] == "openssl": self.calls.append(("host", argv)) self.openssl_stdin.append(stdin) if self.openssl_rc: return self.openssl_rc, "", "Could not read certificate from " return 0, ("X509v3 Basic Constraints: critical\n CA:%s\nnotAfter=%s\n" % ("TRUE" if self.ca else "FALSE", self.ends)), "" return super().host(argv, timeout, stdin) class PVECACopy(unittest.TestCase): """Red-proofs: felhom.eu audits/night-build-2026-11-14/A/redproof-wrapper.txt.""" def refused(self, f, code): rc, rep = run(f) self.assertEqual((rc, rep["refused"]["code"]), (2, code), rep) self.assertEqual(f.puts, [], "nothing is written on a refusal") return rep def test_a_new_box_gets_the_copy(self): f = TFake() rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(f.puts, [(osapply.PVE_CA_COPY, CA_PEM, 0o644)], "ONE file, the fixed place, the source's bytes, 0644") self.assertEqual(f.openssl_stdin, [CA_PEM.decode()], "the bytes that are checked are the bytes that are written") c = rep["pve_ca_copy"] self.assertEqual((c["outcome"], c["path"], c["replaced"]), ("written", "/etc/felhom-agent/pve-root-ca.pem", False)) self.assertEqual(c["not_after"], "2036-08-17T15:44:42Z") self.assertGreater(c["days_left"], 3000) def test_the_same_copy_is_not_written_again(self): f = TFake(copy=CA_PEM) rc, rep = run(f) self.assertEqual((rc, rep["pve_ca_copy"]["outcome"], f.puts), (0, "same", []), rep) def test_a_renewed_ca_refreshes_the_copy(self): f = TFake(source=CA_PEM_2, copy=CA_PEM) rc, rep = run(f) self.assertEqual((rc, rep["pve_ca_copy"]["outcome"], rep["pve_ca_copy"]["replaced"]), (0, "written", True), rep) # the new certificate FIRST, and the copy's previous one kept behind it: Proxmox serves its old certificate until # pveproxy is reloaded, and an agent that restarts in that window must still have a root that verifies it bundle = f.tree[osapply.PVE_CA_COPY] self.assertEqual(bundle, CA_PEM_2 + osapply.PEM_CERT_RE.findall(CA_PEM)[0] + b"\n") self.assertTrue(rep["pve_ca_copy"]["kept_previous"]) self.assertEqual(f.openssl_stdin, [CA_PEM_2.decode()], "the SOURCE is checked in full; the kept one by its own check") def test_the_previous_root_stays_only_for_the_changeover(self): old = osapply.PEM_CERT_RE.findall(CA_PEM)[0] bundle = CA_PEM_2 + old + b"\n" # Proxmox still serves the OLD certificate: nothing is written, the old root stays f = TFake(source=CA_PEM_2, copy=bundle) rc, rep = run(f) self.assertEqual((rc, rep["pve_ca_copy"]["outcome"], f.puts), (0, "same", []), rep) self.assertTrue(rep["pve_ca_copy"]["kept_previous"]) self.assertEqual(len(f.handshakes), 1) self.assertIn(osapply.PVE_CA_SOURCE, f.handshakes[0], "the handshake is verified with the NEW root alone (the source file)") self.assertIn("-no-CApath", f.handshakes[0]) self.assertIn("-verify_return_error", f.handshakes[0]) # pveproxy was reloaded and serves a certificate of the NEW root: the old root is dropped f = TFake(source=CA_PEM_2, copy=bundle) f.serves_new = True rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertEqual(f.tree[osapply.PVE_CA_COPY], CA_PEM_2, "after the changeover the copy is the new root alone") self.assertEqual((rep["pve_ca_copy"]["outcome"], rep["pve_ca_copy"]["dropped_previous"], rep["pve_ca_copy"]["kept_previous"]), ("written", True, False)) # the old root has ENDED while Proxmox still serves the old certificate: it verifies nothing, so it is dropped f = TFake(source=CA_PEM_2, copy=bundle) f.dead_ca = {old.decode()} rc, rep = run(f) self.assertEqual(f.tree[osapply.PVE_CA_COPY], CA_PEM_2) # an ended previous root is not kept at the change itself either f = TFake(source=CA_PEM_2, copy=CA_PEM) f.dead_ca = {old.decode()} rc, rep = run(f) self.assertEqual(f.tree[osapply.PVE_CA_COPY], CA_PEM_2) self.assertFalse(rep["pve_ca_copy"]["kept_previous"]) def test_a_copy_with_anything_unchecked_in_it_is_rewritten(self): old = osapply.PEM_CERT_RE.findall(CA_PEM)[0] for name, copy in (("text after the source", CA_PEM_2 + b"anything at all\n"), ("a third certificate", CA_PEM_2 + old + b"\n" + old + b"\n"), ("a key in the file", CA_PEM_2 + b"-----BEGIN PRIVATE KEY-----\nAAAA\n-----END PRIVATE KEY-----\n"), ("the two in the wrong order", CA_PEM + osapply.PEM_CERT_RE.findall(CA_PEM_2)[0] + b"\n")): f = TFake(source=CA_PEM_2, copy=copy) f.serves_new = True rc, rep = run(f) self.assertEqual(rc, 0, (name, rep)) got = f.tree[osapply.PVE_CA_COPY] self.assertTrue(got.startswith(CA_PEM_2), name) self.assertNotIn(b"anything", got, name) self.assertNotIn(b"PRIVATE KEY", got, name) self.assertLessEqual(len(osapply.PEM_CERT_RE.findall(got)), 2, name) # a first copy keeps nothing behind it, and makes no handshake f4 = TFake() rc, rep = run(f4) self.assertFalse(rep["pve_ca_copy"]["kept_previous"]) self.assertEqual((f4.tree[osapply.PVE_CA_COPY], f4.handshakes), (CA_PEM, [])) def test_a_source_that_is_not_a_ca_certificate_is_refused(self): self.refused(TFake(ca=False), "T3") # a leaf certificate self.refused(TFake(source=b"hello\n"), "T3") # not a certificate at all self.refused(TFake(source=CA_PEM + CA_PEM_2), "T3") # two certificates (a chain, a bundle) self.refused(TFake(source=CA_PEM + b"-----BEGIN PRIVATE KEY-----\nAAAA\n-----END PRIVATE KEY-----\n"), "T3") self.refused(TFake(source=b""), "T3") self.refused(TFake(source=CA_PEM + b"x" * (osapply.PVE_CA_MAX_BYTES + 1)), "T3") f = TFake() f.openssl_rc = 1 # openssl cannot parse it self.refused(f, "T3") self.refused(TFake(ends="Aug 17 15:44:42 2020 GMT"), "T3") # an ended CA self.refused(TFake(ends="garbage"), "T3") def test_no_source_is_refused(self): self.refused(TFake(source=None), "T2") def test_with_the_real_openssl_and_the_real_runner(self): """NOT a fake: a CA and a leaf made by openssl in a temp folder, judged through Runner.host as on a box (the fake cannot show a wrong argument or a bytes/text mix-up). Skipped where openssl is missing.""" import shutil, tempfile if not shutil.which("openssl"): self.skipTest("no openssl here") d = tempfile.mkdtemp() try: def sh(*a): subprocess.run(a, check=True, capture_output=True, cwd=d) sh("openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", "ca.key", "-out", "ca.pem", "-days", "30", "-subj", "/CN=test CA", "-addext", "basicConstraints=critical,CA:TRUE") sh("openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", "leaf.key", "-out", "leaf.pem", "-days", "30", "-subj", "/CN=leaf", "-addext", "basicConstraints=CA:FALSE") real = osapply.Runner() for name, want in (("ca.pem", "written"), ("leaf.pem", "T3")): data = open(os.path.join(d, name), "rb").read() f = TFake(source=data) f.host = lambda argv, timeout=600, stdin=None: real.host(argv, timeout, stdin) if argv[0] == "openssl" else Fake.host(f, argv, timeout, stdin) f.clock = __import__("time").time() rc, rep = run(f) got = rep["pve_ca_copy"]["outcome"] if rc == 0 else rep["refused"]["code"] self.assertEqual(got, want, f"{name}: {rep}") if want == "written": self.assertIn(rep["pve_ca_copy"]["days_left"], (29, 30)) # the two checks the changeover uses, with the REAL openssl and a REAL TLS server ended = True try: # an ENDED CA needs openssl 3.4+ (-not_after); older: that one check is not made sh("openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", "old.key", "-out", "old.pem", "-not_before", "20190101000000Z", "-not_after", "20200101000000Z", "-subj", "/CN=ended CA", "-addext", "basicConstraints=critical,CA:TRUE") except subprocess.CalledProcessError: ended = False sh("openssl", "req", "-newkey", "rsa:2048", "-nodes", "-keyout", "srv.key", "-out", "srv.csr", "-subj", "/CN=pve") sh("openssl", "x509", "-req", "-in", "srv.csr", "-CA", "ca.pem", "-CAkey", "ca.key", "-CAcreateserial", "-out", "srv.pem", "-days", "5") app = osapply.Apply.__new__(osapply.Apply) app.r = real block = lambda n: osapply.PEM_CERT_RE.findall(open(os.path.join(d, n), "rb").read())[0] self.assertTrue(app.is_live_ca(block("ca.pem")), "a CA that has not ended") self.assertFalse(app.is_live_ca(block("leaf.pem")), "a leaf is not a CA") if ended: self.assertFalse(app.is_live_ca(block("old.pem")), "an ended CA is not a live one") import socket s0 = socket.socket(); s0.bind(("127.0.0.1", 0)); port = s0.getsockname()[1]; s0.close() srv = subprocess.Popen(["openssl", "s_server", "-accept", f"127.0.0.1:{port}", "-cert", "srv.pem", "-key", "srv.key", "-quiet"], cwd=d, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) try: saved, osapply.PVE_LOCAL_API = osapply.PVE_LOCAL_API, f"127.0.0.1:{port}" ok = False for _ in range(40): # the server needs a moment to listen if app.pve_serves_with(os.path.join(d, "ca.pem")): ok = True break __import__("time").sleep(0.1) self.assertTrue(ok, "the server's certificate is signed by ca.pem: the handshake must verify with it alone") self.assertFalse(app.pve_serves_with(os.path.join(d, "leaf.pem")), "another certificate as the CA file must NOT verify") osapply.PVE_LOCAL_API = "127.0.0.1:1" self.assertFalse(app.pve_serves_with(os.path.join(d, "ca.pem")), "no answer is a no") finally: osapply.PVE_LOCAL_API = saved srv.kill(); srv.wait() finally: shutil.rmtree(d, ignore_errors=True) def test_a_plan_that_names_a_path_or_a_file_is_refused(self): for extra in ({"source": "/etc/shadow"}, {"path": "/etc/sudoers.d/x"}, {"dest": "/root/.ssh/authorized_keys"}, {"content_b64": "QQ=="}, {"packages": []}, {"file": "x"}, {"vmid": 9201}, {"dry_run": True}): f = TFake() f.plan.update(extra) rep = self.refused(f, "T1") self.assertEqual(f.openssl_stdin, [], "refused before the source is even read") g = TFake() g.plan["layer"] = "kernel" self.refused(g, "R11") def test_a_refused_source_keeps_an_older_copy(self): f = TFake(source=b"not a certificate", copy=CA_PEM) self.refused(f, "T3") self.assertEqual(f.tree[osapply.PVE_CA_COPY], CA_PEM, "written whole or not at all") def test_it_runs_on_a_byo_host_too_and_changes_nothing_else(self): f = TFake() f.files[osapply.INSTALL_STATE] = json.dumps({"mode": "byo"}) rc, rep = run(f) self.assertEqual((rc, rep["pve_ca_copy"]["outcome"]), (0, "written"), rep) self.assertEqual([c[1][0] for c in f.calls], ["openssl"], "one openssl call; no package, no service, no other file") if __name__ == "__main__": unittest.main()