OS updates, guest fast lane (11 §8 step 2): felhom-os-apply wrapper (R1-R13 refusals, repair first, snapshot.debian.org fallback), FELHOM_OSAPPLY sudoers, the OS leg after the primary backup, hub os_update block + os-report, --selftest=os-update
gates / gates (push) Successful in 18s
gates / gates (push) Successful in 18s
No automatic undo: a customer guest cannot be snapshotted (R-837, measured). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
This commit is contained in:
@@ -0,0 +1,452 @@
|
||||
#!/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
|
||||
|
||||
# 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 host(self, argv, timeout=600):
|
||||
self.calls.append(("host", argv))
|
||||
if argv[1] == "status":
|
||||
return 0, self.status + "\n", ""
|
||||
return 1, "", "unexpected host call"
|
||||
|
||||
def guest_write(self, vmid, path, body):
|
||||
self.written[path] = body
|
||||
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))
|
||||
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":
|
||||
return 0, "", ""
|
||||
if cmd == "fuser":
|
||||
return (0, " 123", "") if self.lock_held else (1, "", "")
|
||||
if cmd == "apt-cache" and a[1] == "madison":
|
||||
return 0, "".join(f" {a[2]} | {v} | http://deb.debian.org trixie/main amd64 Packages\n" for v in self.avail(a[2])), ""
|
||||
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 "-s" in a:
|
||||
return self.sim(a)
|
||||
if "install" 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
|
||||
return 0, "Setting up libc6 ...\n", ""
|
||||
if cmd == "df":
|
||||
return 0, f"Avail\n{self.free}\n", ""
|
||||
if cmd == "docker":
|
||||
return 0, "felhom-controller\trunning\tUp 1 hour (healthy)\napp\trunning\tUp 1 hour (healthy)\n", ""
|
||||
if cmd == "getent":
|
||||
return 0, "1.2.3.4 deb.debian.org\n", ""
|
||||
if cmd == "sh":
|
||||
if "os-release" in a[2]:
|
||||
return 0, "trixie\n", ""
|
||||
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:
|
||||
return 0, "'http://x/libc6.deb' libc6.deb 4000000 SHA256:x\n", ""
|
||||
if "dist-upgrade" in a:
|
||||
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 = 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]]
|
||||
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)
|
||||
self.assertIn("restart_needed", 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]]
|
||||
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")
|
||||
|
||||
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_host_layer(self):
|
||||
f = Fake()
|
||||
f.plan["layer"] = "host"
|
||||
self.refused(f, "R12")
|
||||
|
||||
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 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))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user