Old kernels removed by the box (R-1368, wrapper mode kernel-cleanup); security profiles refreshed after a kernel change (R-1388, mode profile-refresh); Go 1.27.4 (R-1391)
gates / gates (push) Successful in 40s
gates / gates (push) Successful in 40s
This commit is contained in:
admin committed
1 parent
91a2924ad0
commit
cc63dd5204
11 files changed
+1542
-10
No files matched your search
+368
-3
@@ -155,9 +155,28 @@ KERNEL_IMAGE_RE = re.compile(r"^proxmox-kernel-([0-9]+\.[0-9]+\.[0-9]+-[0-9]+-pv
|
||||
# helper and the firmware the kernel loads. Everything else in HOST_SLOW_RE (grub, shim, microcode, efibootmgr) stays out.
|
||||
KERNEL_UPGRADE_RE = re.compile(r"^(proxmox-default-kernel|proxmox-kernel-[0-9]+\.[0-9]+|proxmox-kernel-helper|pve-firmware)$")
|
||||
KERNEL_MIN_GAP = 20 * 3600 # never two kernel steps in one night
|
||||
KERNEL_MODES = ("kernel-status", "kernel-reboot", "kernel-boot", "kernel-good", "kernel-revert", "kernel-cancel")
|
||||
KERNEL_MODES = ("kernel-status", "kernel-reboot", "kernel-boot", "kernel-good", "kernel-revert", "kernel-cancel",
|
||||
"kernel-cleanup")
|
||||
# phases: staged → oneshot → judging → good | reverting → self_reverted | revert_failed; oneshot → fell_back; staged → cancelled
|
||||
KERNEL_ACTIVE = ("staged", "oneshot", "judging", "reverting")
|
||||
# The clean-up of old kernels (R-1368, `09` §3 decision 286, `11` §5.11.1). A step that ENDED in one of these phases
|
||||
# and is not "good" keeps every kernel (the box fell back or reverted); any other phase is "not ended" (P1).
|
||||
# The security-profile refresh (R-1388, R-1301, `11` §5.12; measured on demo-hp 2026-11-14,
|
||||
# felhom.eu audits/profile-refresh-spike-2026-11-14/). AppArmor keeps one cache folder per kernel feature set; a
|
||||
# profile that changed while ANOTHER kernel ran leaves an old compiled copy in this kernel's folder, with a time that
|
||||
# looks newer than its source — and the kernel loads it at the next start on this kernel.
|
||||
PROFILE_DIR = "/etc/apparmor.d"
|
||||
PROFILE_CACHE = "/var/cache/apparmor"
|
||||
PROFILE_STATE = "/var/lib/felhom-os-apply/profile-refresh.json" # 0644 root: the kernel the last refresh ran on
|
||||
PROFILE_REFRESH_CMD = ["apparmor_parser", "--replace", "--skip-read-cache", "--write-cache", "--", PROFILE_DIR]
|
||||
PROFILE_PATH_NAME_RE = re.compile(r"^/[A-Za-z0-9._/+-]+$") # a profile whose name is one plain program path
|
||||
UNIT_NAME_RE = re.compile(r"^[A-Za-z0-9@:._-]+\.service$")
|
||||
PROFILE_RESTART_MAX = 3
|
||||
KERNEL_ENDED_NOT_GOOD = ("self_reverted", "revert_failed", "fell_back", "cancelled")
|
||||
KERNEL_LEFTOVER_PATHS = ("/lib/modules/{v}", "/boot/vmlinuz-{v}", "/boot/initrd.img-{v}", "/boot/System.map-{v}",
|
||||
"/boot/config-{v}")
|
||||
PINNED_KVER_RE = re.compile(r"gnulinux-([0-9]+\.[0-9]+\.[0-9]+-[0-9]+-pve)-advanced-")
|
||||
ANY_KVER_RE = re.compile(r"[0-9]+\.[0-9]+\.[0-9]+-[0-9]+-pve")
|
||||
|
||||
# ---------- the config bundle (R-840, agent v0.143.0, `11` §5.4.2) ----------
|
||||
# A signed `agent_config_update` job carries {agent_version, bundle_sha256}; the bundle is ONE JSON file built from this
|
||||
@@ -495,7 +514,7 @@ class Apply:
|
||||
def check_plan(self, plan):
|
||||
mode = plan.get("mode", "apply")
|
||||
if mode not in ("apply", "inventory", "health", "facts", "live-restore-on", "bundle", "bundle-policy", "agent_update",
|
||||
"oom-check") \
|
||||
"oom-check", "profile-refresh") \
|
||||
+ KERNEL_MODES:
|
||||
raise Refused("R11", f"unknown mode {mode!r}")
|
||||
if mode == "agent_update":
|
||||
@@ -510,6 +529,16 @@ class Apply:
|
||||
if plan.get("layer") != "host":
|
||||
raise Refused("R11", "bundle-policy is a host-layer mode")
|
||||
return mode, "host", 0, "bundle-policy"
|
||||
if mode == "profile-refresh":
|
||||
if plan.get("layer") != "host":
|
||||
raise Refused("R11", "profile-refresh is a host-layer mode")
|
||||
# it takes NOTHING from the plan but dry_run: no profile name, no unit name, no package (R-1388)
|
||||
for k in ("packages", "allow_new", "signed", "select", "undo", "snapshot", "profiles", "units", "force"):
|
||||
if plan.get(k):
|
||||
raise Refused("F1", f"a profile-refresh plan carries {k!r} — it takes nothing from a plan but dry_run")
|
||||
if not isinstance(plan.get("dry_run", False), bool):
|
||||
raise Refused("F1", "dry_run must be true or false")
|
||||
return mode, "host", 0, "profile-refresh"
|
||||
layer = plan.get("layer")
|
||||
if layer not in ("guest", "host", "docker", "pve", "kernel"):
|
||||
raise Refused("R12", f"layer {layer!r} is not guest, host, docker, pve or kernel")
|
||||
@@ -540,6 +569,17 @@ class Apply:
|
||||
rid = plan.get("release_id", "")
|
||||
if not isinstance(rid, str) or not re.match(r"^[A-Za-z0-9._:-]{1,80}$", rid):
|
||||
raise Refused("R11", f"release_id {rid!r} is not a plain id")
|
||||
if mode == "kernel-cleanup":
|
||||
# `11` §5.11.1, rule 4 of the release: the clean-up NEVER takes a package name (or any choice of what to
|
||||
# remove) from a plan — the wrapper computes the set from the box. Pinned by test_cleanup_plan_names_refused.
|
||||
for k in ("packages", "allow_new", "signed", "expect_kver", "undo", "snapshot"):
|
||||
if plan.get(k):
|
||||
raise Refused("C1", f"a kernel-cleanup plan carries {k!r} — the clean-up takes nothing from a plan "
|
||||
f"but dry_run; the wrapper computes the set from the box")
|
||||
if plan.get("select", "listed") != "listed":
|
||||
raise Refused("C1", "a kernel-cleanup plan carries a select — the wrapper computes the set from the box")
|
||||
if not isinstance(plan.get("dry_run", False), bool):
|
||||
raise Refused("C1", "dry_run must be true or false")
|
||||
if plan.get("allow_new"):
|
||||
raise Refused("R6", "allow_new is a slow-lane field; the fast lane never adds a package")
|
||||
select = plan.get("select", "listed")
|
||||
@@ -1084,6 +1124,9 @@ class Apply:
|
||||
return Bundle(self).from_plan(plan)
|
||||
if self.mode == "bundle-policy":
|
||||
return Bundle(self).policy_pass()
|
||||
if self.mode == "profile-refresh":
|
||||
self.check_appliance()
|
||||
return self.profile_refresh(plan)
|
||||
if self.mode == "agent_update":
|
||||
return self.agent_update(plan)
|
||||
if self.layer in ("host", "pve", "kernel"):
|
||||
@@ -1149,6 +1192,107 @@ class Apply:
|
||||
self.report["oom_check"] = self.oom_check()
|
||||
return 0
|
||||
|
||||
# ---------- the security-profile refresh (R-1388, R-1301, `11` §5.12) ----------
|
||||
def profile_cache(self):
|
||||
"""{"<folder>/<file>": sha256} of every cached profile, every kernel's folder."""
|
||||
out = {}
|
||||
for d in self.r.list_dir(PROFILE_CACHE):
|
||||
dp = f"{PROFILE_CACHE}/{d}"
|
||||
if not self.r.isdir(dp):
|
||||
continue
|
||||
for f in self.r.list_dir(dp):
|
||||
if f.startswith("."):
|
||||
continue
|
||||
try:
|
||||
out[f"{d}/{f}"] = sha256_hex(self.r.read_bytes(f"{dp}/{f}"))
|
||||
except OSError:
|
||||
out[f"{d}/{f}"] = "unreadable"
|
||||
return out
|
||||
|
||||
def profile_refresh(self, plan):
|
||||
"""Load every AppArmor profile from a FRESH compile (the cache ignored and rewritten), once per kernel the box
|
||||
starts on; then restart ONLY a failed service whose own program's profile had a stale cached copy.
|
||||
|
||||
Changes no package and takes no name from the plan. Measured on demo-hp with both guests running: 0.5 s, no
|
||||
guest or container touched, the same profiles loaded in the same modes.
|
||||
|
||||
WHY NOT a "loaded hash changed" check: for a file that defines several profiles the kernel's hash differs
|
||||
between a cache load and a fresh load of the SAME policy (17 of 119 on demo-hp), and one file (lxc-containers)
|
||||
compiles to a different binary every time. The only reliable sign of a stale copy is the CACHED FILE changing
|
||||
when it is rewritten — that is what the restart rule reads."""
|
||||
r, dry = self.r, plan.get("dry_run") is True
|
||||
res = {"dry_run": dry, "outcome": None, "rewritten": [], "restarted": [], "not_restarted": []}
|
||||
self.report["profile_refresh"] = res
|
||||
rc, out, _ = r.host(["uname", "-r"], 30)
|
||||
kernel = out.strip() if rc == 0 else ""
|
||||
try:
|
||||
rec = json.loads(r.read_file(PROFILE_STATE))
|
||||
rec = rec if isinstance(rec, dict) else {}
|
||||
except (OSError, ValueError):
|
||||
rec = {}
|
||||
res["kernel"], res["last_refresh_kernel"], res["last_refresh_at"] = kernel or "unknown", rec.get("kernel"), rec.get("at")
|
||||
if not kernel:
|
||||
raise Refused("F2", "the running kernel could not be read")
|
||||
if not r.isdir(PROFILE_DIR) or not r.isdir(PROFILE_CACHE):
|
||||
res["outcome"] = "not_needed"
|
||||
res["reason"] = "this box has no AppArmor profile folder or no profile cache"
|
||||
return 0
|
||||
needed = rec.get("kernel") != kernel
|
||||
if not needed:
|
||||
res["outcome"], res["reason"] = "not_needed", f"the profiles were refreshed on this kernel ({kernel}) at {rec.get('at')}"
|
||||
return 0
|
||||
if dry:
|
||||
res["outcome"], res["would"] = "dry_run", "refresh"
|
||||
return 0
|
||||
before = self.profile_cache()
|
||||
t0 = time.time()
|
||||
rc, out, err = r.host(list(PROFILE_REFRESH_CMD), 600)
|
||||
res["seconds"] = round(time.time() - t0, 1)
|
||||
if rc != 0:
|
||||
res["outcome"] = "failed"
|
||||
self.report["failed"] = {"rc": rc, "step": "profile-refresh", "tail": (out + err).strip().splitlines()[-3:]}
|
||||
r.log(f"os-apply: FAILED rc={rc} step=profile-refresh (no record written: the next start tries again)")
|
||||
return 3
|
||||
after = self.profile_cache()
|
||||
res["rewritten"] = sorted(k for k in after if before.get(k) != after[k])
|
||||
# --- the narrow restart rule: a FAILED service, whose ExecStart program IS the name of a profile whose cached
|
||||
# copy was stale. Never "whatever failed"; never a unit named by the plan. ---
|
||||
programs = {}
|
||||
for key in res["rewritten"]:
|
||||
fname = key.split("/", 1)[1]
|
||||
rc_n, names, _ = r.host(["apparmor_parser", "-N", f"{PROFILE_DIR}/{fname}"], 60)
|
||||
if rc_n != 0:
|
||||
continue
|
||||
for n in names.split():
|
||||
if PROFILE_PATH_NAME_RE.match(n):
|
||||
programs[n] = fname
|
||||
if programs:
|
||||
rc_u, units, _ = r.host(["systemctl", "list-units", "--type=service", "--state=failed", "--no-legend", "--plain"], 60)
|
||||
for line in (units.splitlines() if rc_u == 0 else []):
|
||||
unit = (line.split() or [""])[0]
|
||||
if not UNIT_NAME_RE.match(unit):
|
||||
continue
|
||||
_, es, _ = r.host(["systemctl", "show", "-p", "ExecStart", "--value", unit], 30)
|
||||
m = re.search(r"\bpath=(\S+)", es)
|
||||
prog = m.group(1) if m else ""
|
||||
if prog not in programs:
|
||||
continue
|
||||
if len(res["restarted"]) >= PROFILE_RESTART_MAX:
|
||||
res["not_restarted"].append({"unit": unit, "program": prog, "reason": f"more than {PROFILE_RESTART_MAX} units match — look at the box"})
|
||||
continue
|
||||
rc_r, o2, e2 = r.host(["systemctl", "restart", unit], 120)
|
||||
_, act, _ = r.host(["systemctl", "is-active", unit], 30)
|
||||
res["restarted"].append({"unit": unit, "program": prog, "profile_file": programs[prog], "rc": rc_r,
|
||||
"active": act.strip() == "active"})
|
||||
r.log(f"os-apply: PROFILE REFRESH restarted {unit} (failed; its program {prog} had a stale cached profile) "
|
||||
f"rc={rc_r} now={act.strip() or 'unknown'}")
|
||||
r.put_file(PROFILE_STATE, (json.dumps({"kernel": kernel, "at": _iso(r.now()), "rewritten": res["rewritten"],
|
||||
"restarted": [x["unit"] for x in res["restarted"]]}, indent=2, sort_keys=True) + "\n").encode(), 0o644)
|
||||
res["outcome"] = "refreshed"
|
||||
r.log(f"os-apply: PROFILE REFRESH DONE kernel={kernel} (last {rec.get('kernel') or 'never'}) seconds={res['seconds']} "
|
||||
f"rewritten={len(res['rewritten'])} ({', '.join(res['rewritten'][:8])}) restarted={[x['unit'] for x in res['restarted']]}")
|
||||
return 0
|
||||
|
||||
# ---------- R-528: the memory-kill check ----------
|
||||
def oom_check(self):
|
||||
"""Run one throwaway container over its memory cap in the guest and read what the engine says about it.
|
||||
@@ -1493,6 +1637,11 @@ class Kernel:
|
||||
kernel-good the one-shot boot was healthy: the new kernel becomes the GRUB default.
|
||||
kernel-revert the one-shot boot was NOT healthy: reboot ONCE into the old kernel (still the default).
|
||||
kernel-cancel drop a staged step: clear the flag (the package stays installed, the default never moved).
|
||||
kernel-cleanup remove old kernel images by the rule of `11` §5.11.1 (R-1368, decision 286). The plan names nothing
|
||||
(C1); `dry_run: true` computes and reports, and changes nothing. Its own refusals: P1–P3, K1, N1,
|
||||
S1, S2 (the design's table) and C1 (the plan names something). "Not run" (exit 0, nothing
|
||||
changed): no ring-0 mark, no kernel step yet, the newest step did not end good, the box does not
|
||||
run its newest kernel, inside a step's 20 h window.
|
||||
kernel-status read only."""
|
||||
|
||||
def __init__(self, apply, plan):
|
||||
@@ -1646,7 +1795,8 @@ class Kernel:
|
||||
self.report["kernel"] = v
|
||||
return 0
|
||||
fn = {"apply": self.stage, "kernel-reboot": self.reboot_staged, "kernel-boot": self.after_boot,
|
||||
"kernel-good": self.good, "kernel-revert": self.revert, "kernel-cancel": self.cancel}[mode]
|
||||
"kernel-good": self.good, "kernel-revert": self.revert, "kernel-cancel": self.cancel,
|
||||
"kernel-cleanup": self.cleanup}[mode]
|
||||
rc = fn()
|
||||
self.report["kernel"] = self.view()
|
||||
return rc
|
||||
@@ -1882,6 +2032,221 @@ class Kernel:
|
||||
self.log(f"os-apply: KERNEL CANCELLED {st.get('to')} (installed, never the default; flag cleared)")
|
||||
return 0
|
||||
|
||||
# ---------- the clean-up of old kernels (R-1368, `09` §3 decision 286, `11` §5.11.1) ----------
|
||||
def kernel_images(self):
|
||||
"""kver -> [installed image package names] (dpkg status ii only)."""
|
||||
out = {}
|
||||
for name in sorted(self.a.installed()):
|
||||
m = KERNEL_IMAGE_RE.match(name)
|
||||
if m:
|
||||
out.setdefault(m.group(1), []).append(name)
|
||||
return out
|
||||
|
||||
def kver_newest(self, kvers):
|
||||
best = None
|
||||
for k in kvers:
|
||||
if best is None or self.a.dpkg_cmp(k[:-4], "gt", best[:-4]):
|
||||
best = k
|
||||
return best
|
||||
|
||||
def read_or_none(self, path):
|
||||
try:
|
||||
return self.r.read_file(path)
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
def cleanup_sources(self):
|
||||
"""Every place that names a kernel the lane still needs, read from the box ONCE: (kver, why) pairs. This is
|
||||
the keep set's only input, and K1's only input — K1 reads it again by itself (cleanup_protected)."""
|
||||
st = self.state()
|
||||
cfg = self.grub_cfg()
|
||||
dflt = self.default_kver(cfg)
|
||||
m = PINNED_KVER_RE.search(self.read_or_none(KERNEL_DEFAULT_CFG) or "")
|
||||
pinned = m.group(1) if m else None
|
||||
flag = self.flag() or ""
|
||||
m = ANY_KVER_RE.search(flag)
|
||||
m2 = ANY_KVER_RE.search(self.read_or_none(KERNEL_PIN_FILE) or "")
|
||||
src = [(self.running(), "running"),
|
||||
(st.get("from"), "the kernel step's `from` (the last kernel that booted well; a self-revert boots it)"),
|
||||
(st.get("to"), "the kernel step's `to`"),
|
||||
(st.get("default_before"), "the kernel step's `default_before`"),
|
||||
(dflt if KVER_RE.match(dflt) else None, "the default in grub.cfg"),
|
||||
(pinned, "the pinned default file"),
|
||||
(m.group(0) if m else None, "the one-shot flag"),
|
||||
(m2.group(0) if m2 else None, "an operator's boot pin")]
|
||||
return st, dflt, pinned, flag, [(k, why) for k, why in src if isinstance(k, str) and KVER_RE.match(k)]
|
||||
|
||||
def cleanup_keep(self, images, sources):
|
||||
"""The keep set: every kernel a source names, and the newest installed. (cleanup() adds the next lower
|
||||
installed kernel when no step record names the last good one.)"""
|
||||
keep = {}
|
||||
for k, why in sources:
|
||||
keep.setdefault(k, []).append(why)
|
||||
newest = self.kver_newest(images)
|
||||
if newest:
|
||||
keep.setdefault(newest, []).append("newest installed")
|
||||
return keep
|
||||
|
||||
def cleanup_protected(self):
|
||||
"""K1's own reading (`11` §5.11.1: "K1 is the rule's own check and must stay its own"): the kernels that must
|
||||
never reach the purge line, read from the box AGAIN — not taken from the keep set that built the removal set."""
|
||||
_, _, _, _, sources = self.cleanup_sources()
|
||||
return {k: why for k, why in sources}
|
||||
|
||||
def cleanup_simulate(self, names):
|
||||
"""apt's simulation of exactly `names`: (rc, removes, installs, last line)."""
|
||||
rc, out, err = self.a.x(APT_ENV + ["apt-get", "-s", "-q", "purge"] + names)
|
||||
removes = sorted({l.split()[1] for l in out.splitlines() if l.startswith(("Remv ", "Purg ")) and len(l.split()) > 1})
|
||||
installs = sorted({l.split()[1] for l in out.splitlines() if l.startswith(("Inst ", "Conf ")) and len(l.split()) > 1})
|
||||
return rc, removes, installs, ((err or out).strip().splitlines() or ["?"])[-1][:200]
|
||||
|
||||
def cleanup(self):
|
||||
a = self.a
|
||||
dry = self.plan.get("dry_run") is True
|
||||
res = {"dry_run": dry, "outcome": None, "not_run": [], "refusals": [], "removed": [], "leftovers": []}
|
||||
self.report["kernel_cleanup"] = res
|
||||
not_run, refuse = res["not_run"], res["refusals"]
|
||||
# --- may this box run the clean-up at all? (ring 0 only in this release; the mark is ROOT-owned) ---
|
||||
try:
|
||||
ring0 = a.load_trust().get("ring0_slow_lane") is True
|
||||
except Refused as e:
|
||||
ring0 = False
|
||||
not_run.append({"code": "no-ring0-mark", "reason": f"no usable trust record: {e.reason}"})
|
||||
if not ring0 and not not_run:
|
||||
not_run.append({"code": "no-ring0-mark", "reason": "this box has no root-owned ring-0 mark — the clean-up runs on "
|
||||
"ring 0 only in this release"})
|
||||
if a.apt_lock_held():
|
||||
raise Refused("R9", "another apt/dpkg holds the lock on the host")
|
||||
if not dry and not not_run:
|
||||
a.repair() # repair first (`11` §5.4.1): a purge a power cut interrupted is made whole, THEN the rule is computed
|
||||
# --- read the box ---
|
||||
images = self.kernel_images()
|
||||
running = self.running()
|
||||
newest = self.kver_newest(images)
|
||||
raw_state = self.read_or_none(KERNEL_STATE)
|
||||
st, dflt, pinned, flag, sources = self.cleanup_sources()
|
||||
phase = st.get("phase", "none") if raw_state is not None else "none"
|
||||
res.update(running=running or "unknown", newest=newest, phase=phase, default=dflt, pinned_default=pinned,
|
||||
installed={k: images[k] for k in sorted(images, key=lambda v: [int(x) for x in re.findall(r"\d+", v)])})
|
||||
# --- "when it may run" (decision 286): not run, nothing changed ---
|
||||
if raw_state is None or not st:
|
||||
not_run.append({"code": "no-step", "reason": "no kernel step yet (no state file) — nothing proves which kernel booted well"})
|
||||
elif phase in KERNEL_ENDED_NOT_GOOD:
|
||||
not_run.append({"code": "step-not-good", "reason": f"the newest kernel step ended {phase!r}, not good — every kernel stays"})
|
||||
elif phase != "good":
|
||||
refuse.append({"code": "P1", "reason": f"a kernel step is in phase {phase!r} — not ended (or not a phase this wrapper knows)"})
|
||||
if running and newest and running in images and running != newest:
|
||||
not_run.append({"code": "not-on-newest", "reason": f"the box runs {running}, not its newest installed kernel {newest} — every kernel stays"})
|
||||
times = [x for x in (_parse_iso(st.get(k)) for k in ("staged_at", "rebooted_at", "result_at")) if x is not None]
|
||||
if times and self.r.now() - max(times) < KERNEL_MIN_GAP:
|
||||
not_run.append({"code": "window", "reason": "a kernel step staged, booted or ended within the last 20 hours — never in its night"})
|
||||
# --- the refusal table ---
|
||||
if flag:
|
||||
refuse.append({"code": "P2", "reason": f"the one-shot flag names {flag!r}"})
|
||||
if pinned is None or pinned != dflt or pinned not in images:
|
||||
refuse.append({"code": "P3", "reason": f"the pinned default {pinned!r} and grub.cfg's default {dflt!r} do not name the same installed kernel"})
|
||||
if not running or running not in images:
|
||||
refuse.append({"code": "K1", "reason": f"the running kernel {running or 'unknown'!r} is not an installed kernel image package"})
|
||||
# --- the keep set and the removal set (exact package names) ---
|
||||
keep = self.cleanup_keep(images, sources)
|
||||
if running and not (st.get("to") == running and st.get("from") in images):
|
||||
lower = [k for k in images if a.dpkg_cmp(k[:-4], "lt", running[:-4])] if running in images else []
|
||||
lg = self.kver_newest(lower)
|
||||
if lg:
|
||||
keep.setdefault(lg, []).append("the next lower installed kernel (no step record names the last good one)")
|
||||
removal = sorted(n for k in images if k not in keep for n in images[k])
|
||||
res["kept"] = {k: keep[k] for k in sorted(keep) if k in images}
|
||||
res["removal"] = removal
|
||||
# K1 — its own reading of the box, against the removal set
|
||||
by_name = {n: k for k, names in images.items() for n in names}
|
||||
protected = self.cleanup_protected()
|
||||
for n in removal:
|
||||
if n not in by_name:
|
||||
refuse.append({"code": "N1", "reason": f"{n} is not an installed kernel image package"})
|
||||
elif by_name[n] in protected or by_name[n] == newest:
|
||||
refuse.append({"code": "K1", "reason": f"{n} is a keep-kernel ({protected.get(by_name[n], 'newest installed')})"})
|
||||
# S1 / S2 — apt's simulation of exactly that set
|
||||
if removal and not [x for x in refuse if x["code"] in ("K1", "N1")]:
|
||||
rc, removes, installs, tail = self.cleanup_simulate(removal)
|
||||
res["simulation"] = {"rc": rc, "removes": removes, "installs": installs}
|
||||
if rc != 0:
|
||||
refuse.append({"code": "S1", "reason": "the simulation failed: " + tail})
|
||||
if installs:
|
||||
refuse.append({"code": "S1", "reason": f"the simulation would install or configure {', '.join(installs[:6])}"})
|
||||
extra = [p for p in removes if p not in removal]
|
||||
if extra:
|
||||
refuse.append({"code": "S2", "reason": f"the simulation would also remove {', '.join(extra[:8])} — not in the removal set"})
|
||||
missing = [p for p in removal if p not in removes]
|
||||
if missing and rc == 0:
|
||||
refuse.append({"code": "S1", "reason": f"the simulation does not remove {', '.join(missing[:6])}"})
|
||||
self.log(f"os-apply: KERNEL CLEANUP {'DRY RUN ' if dry else ''}running={running or 'unknown'} newest={newest} phase={phase} "
|
||||
f"installed={len(images)} keep={len(res['kept'])} removal={len(removal)} not_run={[x['code'] for x in not_run]} "
|
||||
f"refusals={[x['code'] for x in refuse]}")
|
||||
# --- the verdict ---
|
||||
if dry:
|
||||
res["outcome"] = "dry_run"
|
||||
res["would"] = "not_run" if not_run else "refuse" if refuse else "remove" if removal else "nothing"
|
||||
return 0
|
||||
if not_run:
|
||||
res["outcome"] = "not_run"
|
||||
return 0
|
||||
if refuse:
|
||||
raise Refused(refuse[0]["code"], "; ".join(f"{x['code']} {x['reason']}" for x in refuse)[:600])
|
||||
if not removal:
|
||||
res["outcome"] = "nothing"
|
||||
self.log("os-apply: KERNEL CLEANUP nothing to remove")
|
||||
return 0
|
||||
# --- the purge: ONE apt-get purge of the exact names; then prove the boot setup did not move ---
|
||||
gens_before = {p: (self.r.lexists(p), self.read_or_none(p)) for p in (ONESHOT_SNIPPET, ONESHOT_ENTRIES)}
|
||||
pin_before = self.read_or_none(KERNEL_DEFAULT_CFG)
|
||||
free_before = a.free_bytes()
|
||||
t0 = time.time()
|
||||
rc, out, err = a.x(APT_ENV + ["apt-get", "-y", "-q", "purge"] + removal)
|
||||
res["seconds"] = round(time.time() - t0, 1)
|
||||
if rc != 0:
|
||||
_, aud, _ = a.x(["dpkg", "--audit"])
|
||||
res["outcome"] = "failed"
|
||||
self.report["failed"] = {"rc": rc, "step": "purge", "dpkg_audit": (aud.strip().splitlines() or ["clean"])[0],
|
||||
"tail": (out + err).strip().splitlines()[-3:]}
|
||||
self.log(f"os-apply: FAILED rc={rc} step=purge (the next run repairs dpkg first, then computes the rule again)")
|
||||
return 3
|
||||
after = self.kernel_images()
|
||||
after_names = {n for names in after.values() for n in names}
|
||||
res["removed"] = [n for n in removal if n not in after_names]
|
||||
free_after = a.free_bytes()
|
||||
res["bytes_gained"] = free_after - free_before if free_before >= 0 and free_after >= 0 else None
|
||||
problems = []
|
||||
if len(res["removed"]) != len(removal):
|
||||
problems.append(f"still installed after the purge: {', '.join(n for n in removal if n in after_names)}")
|
||||
gone = [k for k in res["kept"] if k not in after]
|
||||
if gone:
|
||||
problems.append(f"a KEPT kernel is no longer installed: {', '.join(gone)}")
|
||||
got = self.default_kver(self.grub_cfg())
|
||||
if got != dflt:
|
||||
problems.append(f"the default in grub.cfg moved from {dflt} to {got}")
|
||||
if self.read_or_none(KERNEL_DEFAULT_CFG) != pin_before:
|
||||
problems.append("the pinned default file changed")
|
||||
if self.read_or_none(KERNEL_STATE) != raw_state:
|
||||
problems.append("the kernel step's state file changed")
|
||||
for p, (was, body) in gens_before.items():
|
||||
if self.r.lexists(p) != was or self.read_or_none(p) != body:
|
||||
problems.append(f"{p} changed")
|
||||
for k in sorted(set(by_name[n] for n in res["removed"])):
|
||||
if k in after:
|
||||
continue # another image package of the same kernel is still installed
|
||||
res["leftovers"] += [p.format(v=k) for p in KERNEL_LEFTOVER_PATHS if self.r.lexists(p.format(v=k))]
|
||||
if problems:
|
||||
# reported LOUDLY, never repaired by hand here (`11` §5.11.1): the operator looks at the box
|
||||
res["outcome"] = "failed"
|
||||
self.report["failed"] = {"rc": 3, "step": "verify", "reason": "; ".join(problems)[:600]}
|
||||
self.log("os-apply: FAILED rc=3 step=verify after the kernel purge — " + "; ".join(problems))
|
||||
return 3
|
||||
res["outcome"] = "removed"
|
||||
self.log(f"os-apply: KERNEL CLEANUP DONE removed={len(res['removed'])} ({', '.join(res['removed'])}) "
|
||||
f"kept={', '.join(res['kept'])} gained={res['bytes_gained']} B seconds={res['seconds']} "
|
||||
f"leftovers={len(res['leftovers'])}; default {dflt} (proved from grub.cfg)")
|
||||
return 0
|
||||
|
||||
|
||||
class Bundle:
|
||||
"""The config bundle (R-840, `11` §5.4.2): every root-owned file the installer's step 5 writes, installed as ONE
|
||||
|
||||
@@ -2031,5 +2031,607 @@ class KernelLane(unittest.TestCase):
|
||||
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/<kver>
|
||||
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")
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user