admin
fdd87178d2
gates / gates (push) Successful in 20s
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
321 lines
13 KiB
Python
321 lines
13 KiB
Python
#!/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_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")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|