admin/felhom-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8401a3091749daa9f1db6c521a47c4331b335c5f
felhom-agent/configs/test_felhom_os_apply.py
T

1126 lines
50 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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):
pass
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":
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 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), ""
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 == "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":
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 = "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/<pid>/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_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
(`<key_id> namespaces="felhom-op-v1" <type> <b64> <comment>`). 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)
if __name__ == "__main__":
unittest.main()
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")
Reference in New Issue View Git Blame Copy Permalink