#!/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) 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_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") if __name__ == "__main__": unittest.main(verbosity=2)