#!/usr/bin/env python3 """Tests for felhom-priv-apply (R-861, `03` §3.1). An in-memory host plays the files; nothing real is written or run. Each refusal rule has a test that feeds it the ATTACK it exists for; each accepted shape is a file the agent really renders (the Go contract tests feed the live renderers too). Red-proof: audits/hub-safety-2026-10-05/partF/. Run: python3 configs/test_felhom_priv_apply.py (also run by internal/privapply's Go test) """ import sys sys.dont_write_bytecode = True import importlib.machinery import importlib.util import os import pathlib import unittest HERE = pathlib.Path(__file__).resolve().parent _loader = importlib.machinery.SourceFileLoader("privapply", os.environ.get("PRIVAPPLY_UNDER_TEST", str(HERE / "felhom-priv-apply"))) _spec = importlib.util.spec_from_loader("privapply", _loader) pa = importlib.util.module_from_spec(_spec) _loader.exec_module(pa) AGENT_UID = 999 class FakeHost: def __init__(self): self.src, self.dest, self.logs, self.writes = {}, {}, [], [] self.owner, self.kind = {}, {} def stage(self, path, text, uid=AGENT_UID, kind="file"): self.src[path] = text.encode() if isinstance(text, str) else text self.owner[path], self.kind[path] = uid, kind def agent_uid(self): return AGENT_UID def read_source(self, path): # mirrors Host.read_source's refusals: absent, not a regular file (a symlink is refused by O_NOFOLLOW), owner, size if path not in self.src: raise pa.Refused("P1", f"cannot open the staged file {path}") if self.kind[path] != "file": raise pa.Refused("P1", f"{path} is not a regular file") if self.owner[path] != AGENT_UID: raise pa.Refused("P1", f"{path} is not owned by felhom-agent") if len(self.src[path]) > pa.MAX_BYTES: raise pa.Refused("P1", f"{path} is larger than {pa.MAX_BYTES} bytes") return self.src[path] def read_dest(self, path): return self.dest.get(path) def install(self, dest, data, mode): self.writes.append((dest, mode)) self.dest[dest] = data def log(self, line): self.logs.append(line) # controller-image (R-861 (a) A1): the ref arrives on stdin and is written INSIDE the guest by root. stdin = b"" guest_writes = None def read_stdin(self, limit): return self.stdin[:limit + 1] def write_guest_image(self, vmid, data): if self.guest_writes is None: self.guest_writes = [] self.guest_writes.append((vmid, data)) LOCAL_UNIT = """# Managed by felhom-agent — do not edit by hand. [Unit] Description=Felhom storage mount 91d2dc2d-2d28-4929-9bdd-3e11fa2f41ae After=local-fs-pre.target [Mount] What=/dev/disk/by-uuid/91d2dc2d-2d28-4929-9bdd-3e11fa2f41ae Where=/mnt/hdd_1 Type=ext4 [Install] WantedBy=multi-user.target """ NET_MOUNT = """# felhom network storage — do not edit by hand. [Unit] Description=Felhom network storage media (nfs) [Mount] What=nas.lan:/volume1/media Where=/mnt/felhom-drives/media Type=nfs4 Options=vers=4.1,soft,timeo=50,retrans=2,noatime,_netdev,retry=0,nosuid,nodev """ NET_AUTOMOUNT = """# felhom network storage — do not edit by hand. [Unit] Description=Felhom network storage automount media (nfs) [Automount] Where=/mnt/felhom-drives/media TimeoutIdleSec=600 [Install] WantedBy=multi-user.target """ NET_NAME = "mnt-felhom\\x2ddrives-media.mount" DNS_BASE = """# felhom split-horizon resolver — host base config (agent-managed; DO NOT EDIT) bind-interfaces listen-address=192.168.0.104 listen-address=127.0.0.1 no-resolv server=1.1.1.1 server=9.9.9.9 """ DNS_GUEST = """# felhom split-horizon DNS — customer demo-hp (agent-managed; DO NOT EDIT) local=/enkisfelhom.hu/ address=/enkisfelhom.hu/192.168.0.138 """ K = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" # base64 of 32 bytes WG = f"""# felhom offsite tunnel — agent-managed (S3); DO NOT EDIT [Interface] PrivateKey = {K} Address = 10.77.0.3/32 MTU = 1280 [Peer] PublicKey = {K} Endpoint = 49.12.1.2:51820 AllowedIPs = 10.77.0.1/32, 10.77.0.250/32 PersistentKeepalive = 25 """ KEYLINE = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIL8z0qCNgA3x2xxAB0Qj5ro8waFjGZ8Ta/sWB63tlLw+ felhom-op-1\n" def run(host, *argv): return pa.main(list(argv), host=host) class Accepts(unittest.TestCase): """What the agent really writes is accepted and installed, root-owned, at the fixed destination.""" def test_local_unit_installed(self): h = FakeHost() h.stage("/var/lib/felhom-agent/units/mnt-hdd_1.mount", LOCAL_UNIT) self.assertEqual(run(h, "unit", "mnt-hdd_1.mount"), 0) self.assertEqual(h.writes, [("/etc/systemd/system/mnt-hdd_1.mount", 0o644)]) def test_local_unit_without_type(self): # the N100's unit has no Type= (autodetect) h = FakeHost() h.stage("/var/lib/felhom-agent/units/mnt-hdd_1.mount", LOCAL_UNIT.replace("Type=ext4\n", "")) self.assertEqual(run(h, "unit", "mnt-hdd_1.mount"), 0) def test_network_pair(self): h = FakeHost() h.stage("/var/lib/felhom-agent/units/" + NET_NAME, NET_MOUNT) h.stage("/var/lib/felhom-agent/units/mnt-felhom\\x2ddrives-media.automount", NET_AUTOMOUNT) self.assertEqual(run(h, "unit", NET_NAME), 0) self.assertEqual(run(h, "unit", "mnt-felhom\\x2ddrives-media.automount"), 0) def test_identical_is_not_rewritten(self): h = FakeHost() h.stage("/var/lib/felhom-agent/units/mnt-hdd_1.mount", LOCAL_UNIT) h.dest["/etc/systemd/system/mnt-hdd_1.mount"] = LOCAL_UNIT.encode() self.assertEqual(run(h, "unit", "mnt-hdd_1.mount"), 0) self.assertEqual(h.writes, []) def test_dnsmasq_both_dropins(self): h = FakeHost() h.stage("/tmp/felhom-resolver-123.conf", DNS_BASE) self.assertEqual(run(h, "dnsmasq", "/tmp/felhom-resolver-123.conf", "felhom-resolver-base.conf"), 0) h.stage("/tmp/felhom-resolver-124.conf", DNS_GUEST) self.assertEqual(run(h, "dnsmasq", "/tmp/felhom-resolver-124.conf", "felhom-demo-hp.conf"), 0) self.assertIn(("/etc/dnsmasq.d/felhom-demo-hp.conf", 0o644), h.writes) def test_wg_installed_0600(self): h = FakeHost() h.stage(pa.WG_SRC, WG) self.assertEqual(run(h, "wg"), 0) self.assertEqual(h.writes, [(pa.WG_DEST, 0o600)]) def test_sshd_template_and_key(self): h = FakeHost() h.stage(pa.SSHD_SRC, pa.render_sshd(8822)) h.stage(pa.KEY_SRC, KEYLINE) self.assertEqual(run(h, "sshd-config"), 0) self.assertEqual(run(h, "sshd-key"), 0) h.stage(pa.KEY_SRC, "") # clearing the operator login is allowed self.assertEqual(run(h, "sshd-key"), 0) def test_escape_matches_systemd(self): self.assertEqual(pa.systemd_escape_path("/mnt/felhom-drives/media"), "mnt-felhom\\x2ddrives-media") self.assertEqual(pa.systemd_escape_path("/mnt/hdd_1"), "mnt-hdd_1") self.assertEqual(pa.systemd_escape_path("/mnt/.x"), "mnt-.x") # a dot is escaped only at the very start class Refuses(unittest.TestCase): """Each rule, with the attack it exists for. Nothing is written on a refusal.""" def refused(self, h, argv, rule): rc = run(h, *argv) self.assertIn(rc, (2, 3), f"{argv} was accepted") self.assertEqual(h.writes, [], f"{argv} wrote something") self.assertTrue(any(f"[{rule}]" in l for l in h.logs), f"{argv}: rule {rule} not logged: {h.logs}") def unit(self, text, name="mnt-hdd_1.mount"): h = FakeHost() h.stage("/var/lib/felhom-agent/units/" + name, text) return h, ["unit", name] def test_U1_name_outside_mnt(self): h = FakeHost() self.refused(h, ["unit", "etc-sudoers.d.mount"], "U1") self.refused(h, ["unit", "../../etc/x.mount"], "U1") self.refused(h, ["unit", "mnt-x.service"], "U1") def test_U2_service_section(self): self.refused(*self.unit(LOCAL_UNIT + "\n[Service]\nExecStart=/bin/sh -c id\n"), "U2") def test_U2_wants_starts_another_unit(self): # review 2026-10-05: Wants=reboot.target would reboot the host for extra in ("Wants=reboot.target", "Requires=felhom-agent-rollback.service", "Before=pve-guests.service"): self.refused(*self.unit(LOCAL_UNIT.replace("After=local-fs-pre.target", "After=local-fs-pre.target\n" + extra)), "U2") self.refused(*self.unit(LOCAL_UNIT.replace("After=local-fs-pre.target", "After=poweroff.target")), "U2") def test_U2_continuation_line(self): t = LOCAL_UNIT.replace("Description=Felhom storage mount 91d2dc2d-2d28-4929-9bdd-3e11fa2f41ae", "Description=Felhom storage mount \\") self.refused(*self.unit(t), "U2") self.refused(*self.unit(LOCAL_UNIT.replace("# Managed by felhom-agent", "# comment \\\n# Managed by felhom-agent")), "U2") def test_U2_unknown_key(self): self.refused(*self.unit(LOCAL_UNIT.replace("Type=ext4", "Type=ext4\nDirectoryMode=0777")), "U2") def test_U3_bind_over_sudoers_dir(self): # the R-861 attack: mount an agent-owned directory over /etc/sudoers.d t = LOCAL_UNIT.replace("Where=/mnt/hdd_1", "Where=/etc/sudoers.d") self.refused(*self.unit(t, "mnt-hdd_1.mount"), "U3") def test_U3_name_must_match_where(self): self.refused(*self.unit(LOCAL_UNIT.replace("Where=/mnt/hdd_1", "Where=/mnt/other")), "U3") def test_U3_traversal_in_where(self): # the name passes U1 and equals the escaped Where — ONLY the Where rule stops a mount at /mnt/../etc = /etc self.assertEqual(pa.systemd_escape_path("/mnt/../etc") + ".mount", "mnt-..-etc.mount") self.refused(*self.unit(LOCAL_UNIT.replace("Where=/mnt/hdd_1", "Where=/mnt/../etc"), "mnt-..-etc.mount"), "U3") def test_U4_what_is_an_agent_directory(self): t = LOCAL_UNIT.replace("What=/dev/disk/by-uuid/91d2dc2d-2d28-4929-9bdd-3e11fa2f41ae", "What=/var/lib/felhom-agent/evil") self.refused(*self.unit(t), "U4") def test_U4_tmpfs(self): self.refused(*self.unit(LOCAL_UNIT.replace("Type=ext4", "Type=tmpfs")), "U4") def test_U5_bind_option(self): self.refused(*self.unit(LOCAL_UNIT.replace("Type=ext4", "Type=ext4\nOptions=bind")), "U5") def test_U5_network_without_nosuid(self): t = NET_MOUNT.replace(",nosuid,nodev", "") self.refused(*self.unit(t, NET_NAME), "U5") def test_U5_suid_option(self): self.refused(*self.unit(LOCAL_UNIT.replace("Type=ext4", "Type=ext4\nOptions=suid,dev")), "U5") def test_U3_network_outside_drives(self): t = NET_MOUNT.replace("/mnt/felhom-drives/media", "/mnt/media") self.refused(*self.unit(t, "mnt-media.mount"), "U3") def test_D1_dhcp_script(self): # runs as root h = FakeHost() h.stage("/tmp/felhom-resolver-1.conf", DNS_BASE + "dhcp-script=/var/lib/felhom-agent/x.sh\n") self.refused(h, ["dnsmasq", "/tmp/felhom-resolver-1.conf", "felhom-x.conf"], "D1") def test_D1_conf_dir_and_log_file(self): for extra in ("conf-dir=/var/lib/felhom-agent\n", "log-facility=/etc/sudoers.d/x\n", "user=root\n"): h = FakeHost() h.stage("/tmp/felhom-resolver-1.conf", DNS_GUEST + extra) self.refused(h, ["dnsmasq", "/tmp/felhom-resolver-1.conf", "felhom-x.conf"], "D1") def test_D2_paths(self): h = FakeHost() self.refused(h, ["dnsmasq", "/etc/shadow", "felhom-x.conf"], "D2") self.refused(h, ["dnsmasq", "/tmp/felhom-resolver-1.conf", "../sudoers.d/x.conf"], "D2") def test_W1_postup(self): # wg-quick runs PostUp as root h = FakeHost() h.stage(pa.WG_SRC, WG.replace("MTU = 1280", "MTU = 1280\nPostUp = /bin/sh -c id")) self.refused(h, ["wg"], "W1") def test_W2_values(self): h = FakeHost() h.stage(pa.WG_SRC, WG.replace("AllowedIPs = 10.77.0.1/32, 10.77.0.250/32", "AllowedIPs = 0.0.0.0/0")) self.refused(h, ["wg"], "W2") def test_S1_sshd_strictmodes(self): # an AuthorizedKeysFile the agent owns + StrictModes no = root login h = FakeHost() h.stage(pa.SSHD_SRC, pa.render_sshd(8822) + "StrictModes no\n") self.refused(h, ["sshd-config"], "S1") h2 = FakeHost() h2.stage(pa.SSHD_SRC, pa.render_sshd(8822).replace("/etc/felhom-sshd/authorized_keys/%u", "/var/lib/felhom-agent/k")) self.refused(h2, ["sshd-config"], "S1") def test_S1_port_22(self): h = FakeHost() h.stage(pa.SSHD_SRC, pa.render_sshd(22)) self.refused(h, ["sshd-config"], "S1") def test_S2_key_options_and_two_keys(self): h = FakeHost() h.stage(pa.KEY_SRC, 'command="/bin/sh" ' + KEYLINE) self.refused(h, ["sshd-key"], "S2") h2 = FakeHost() h2.stage(pa.KEY_SRC, KEYLINE + KEYLINE) self.refused(h2, ["sshd-key"], "S2") def test_P1_symlink_owner_size(self): h = FakeHost() h.stage(pa.WG_SRC, WG, kind="symlink") self.refused(h, ["wg"], "P1") h2 = FakeHost() h2.stage(pa.WG_SRC, WG, uid=0) self.refused(h2, ["wg"], "P1") h3 = FakeHost() h3.stage(pa.WG_SRC, "#" * (pa.MAX_BYTES + 1)) self.refused(h3, ["wg"], "P1") def test_A1_usage(self): h = FakeHost() self.refused(h, ["install", "/etc/shadow"], "A1") self.refused(h, ["wg", "/etc/shadow"], "A1") class ControllerImage(unittest.TestCase): """R-861 (a) A1 (`09` §3 decision 165): the agent can no longer `tee` any image ref into the guest. The root verb reads the ref on stdin, requires our registry + our repository + an x.y.z tag, and writes the guest file itself. RED-PROOF: on the pre-A1 wrapper `controller-image` is not a verb (A1 usage, rc 2) — the accepted case fails.""" def go(self, ref, *argv): h = FakeHost() h.stdin = ref.encode() if isinstance(ref, str) else ref return h, run(h, *(argv or ("controller-image", "9201"))) def test_our_controller_ref_is_written_in_the_guest(self): h, rc = self.go("gitea.dooplex.hu/admin/felhom-controller:0.301.0\n") self.assertEqual(rc, 0, h.logs) self.assertEqual(h.guest_writes, [(9201, b"gitea.dooplex.hu/admin/felhom-controller:0.301.0\n")]) def test_a_foreign_image_is_refused(self): for ref in ("docker.io/library/alpine:latest\n", "alpine\n", "gitea.dooplex.hu/admin/felhom-controller:latest\n", "gitea.dooplex.hu/admin/other:0.1.0\n", "evil.example/admin/felhom-controller:0.301.0\n", "gitea.dooplex.hu/admin/felhom-controller:0.301.0\nalpine\n", "gitea.dooplex.hu/admin/felhom-controller:0.301.0 x\n", "", "\n"): h, rc = self.go(ref) self.assertEqual(rc, 3, f"{ref!r} was accepted") self.assertFalse(h.guest_writes, f"{ref!r} wrote the guest file") self.assertTrue(any("[I1]" in l for l in h.logs), h.logs) def test_oversize_stdin_is_refused(self): h, rc = self.go("gitea.dooplex.hu/admin/felhom-controller:0.301.0" + " " * 300) self.assertEqual(rc, 3) self.assertFalse(h.guest_writes) def test_vmid_must_be_numeric(self): for argv in (("controller-image", "9201;id"), ("controller-image", "-1"), ("controller-image",), ("controller-image", "9201", "9202")): h, rc = self.go("gitea.dooplex.hu/admin/felhom-controller:0.301.0\n", *argv) self.assertIn(rc, (2, 3), argv) self.assertFalse(h.guest_writes, argv) def test_self_check_names_the_verb(self): import io, contextlib buf = io.StringIO() with contextlib.redirect_stdout(buf): pa.main(["--self-check"]) self.assertIn("controller-image", buf.getvalue()) if __name__ == "__main__": unittest.main(verbosity=2)