felhom-agent/configs/test_felhom_mkfs_guarded.py

#!/usr/bin/env python3
"""Tests for configs/felhom-mkfs-guarded.sh — the ONLY mkfs path the agent's sudoers permits.

First tests of this wrapper (agent v0.158.0), written for partitions as places (felhom.eu architecture/16 §6). Every
mount of every node of the disk outside /mnt/felhom-drives refuses — INCLUDING a sibling that is Felhom's own drive
(its raw /mnt/<name> mount): the agent can make that corroborating mount itself, so the root wrapper must not trust
it (security review of 912f7c1). An unlistable disk refuses.

HOW IT RUNS ANYWHERE: the wrapper is copied with five literal host paths replaced (each replacement is asserted, so a
drifted wrapper fails here instead of silently testing nothing): /proc/mounts → a fake mount table, the block-device
test → true (no real /dev nodes here), the pvs candidates → a stub, /sys/block → a fake tree, the mkfs binaries → a
recorder. `lsblk` is a PATH stub answering the three calls the wrapper makes from one JSON disk description.
Every refusal has a positive control (the same disk without the one condition is formatted).
Run: python3 configs/test_felhom_mkfs_guarded.py   (also run by the Go suite, internal/storage)
"""
import json, os, pathlib, subprocess, tempfile, unittest

WRAPPER = pathlib.Path(__file__).resolve().parent / "felhom-mkfs-guarded.sh"

STUB_LSBLK = r'''#!/usr/bin/env python3
import json, os, sys
st = json.load(open(os.environ["MG_STATE"]))
a = sys.argv[1:]; dev = a[-1]
nodes = st["nodes"]   # [{"name": "/dev/sdb", "pk": "", "fstype": ""}, ...]
if a[0] == "-ndo" and a[1] == "PKNAME":
    for n in nodes:
        if n["name"] == dev and n.get("pk"): print(n["pk"])
elif a[0] == "-nrpo" and a[1] == "NAME":
    if not st.get("lsblk_empty"):
        for n in nodes: print(n["name"])
elif a[0] == "-nro" and a[1] == "FSTYPE":
    for n in nodes: print(n.get("fstype", ""))
elif a[0] == "-nro" and a[1] == "MOUNTPOINT":   # the pre-v0.158.0 wrapper's call: one mountpoint per node, like lsblk
    mt = [l.split()[:2] for l in open(os.environ["MG_MOUNTS"]) if l.strip()]
    for n in nodes:
        mps = [m for d, m in mt if d == n["name"]]
        print(mps[0] if mps else "")
'''
STUB_MKFS = '#!/bin/sh\necho "$0 $*" >> "$MG_CALLS"\n'
STUB_PVS = '#!/bin/sh\nprintf "%s\\n" $MG_PVS\n'


class Disk:
    def __init__(self, nodes, mounts, ro="0", pvs="", lsblk_empty=False):
        self.t = tempfile.TemporaryDirectory()
        r = pathlib.Path(self.t.name)
        self.bin = r / "bin"; self.bin.mkdir()
        (self.bin / "lsblk").write_text(STUB_LSBLK); (self.bin / "lsblk").chmod(0o755)
        for n in ("mkfs.ext4", "mkfs.xfs"):
            (self.bin / n).write_text(STUB_MKFS); (self.bin / n).chmod(0o755)
        (self.bin / "pvs").write_text(STUB_PVS); (self.bin / "pvs").chmod(0o755)
        self.state = r / "state.json"
        self.state.write_text(json.dumps({"nodes": nodes, "lsblk_empty": lsblk_empty}))
        self.mounts = r / "mounts"
        self.mounts.write_text("".join(f"{d} {m} ext4 rw 0 0\n" for d, m in mounts))
        self.sys = r / "sys"
        for n in nodes:
            if not n.get("pk"):
                d = self.sys / os.path.basename(n["name"]); d.mkdir(parents=True); (d / "ro").write_text(ro + "\n")
        self.calls = r / "calls"
        self.pvs = pvs
        src = WRAPPER.read_text()
        subs = [
            ("/proc/mounts", str(self.mounts)),
            ('[[ -b "$dev" ]]', "true"),
            ("for c in /usr/sbin/pvs /sbin/pvs; do", f"for c in {self.bin}/pvs; do"),
            ('rof="/sys/block/', f'rof="{self.sys}/'),
            ("exec /usr/sbin/mkfs.ext4", f"exec {self.bin}/mkfs.ext4"),
            ("exec /usr/sbin/mkfs.xfs ", f"exec {self.bin}/mkfs.xfs "),
        ]
        for a, b in subs:
            assert a in src, f"the wrapper no longer contains {a!r} — update this test's substitutions"
            src = src.replace(a, b)
        self.copy = r / "mkfs-guarded"; self.copy.write_text(src)

    def run(self, dev, fstype="ext4"):
        env = {"PATH": f"{self.bin}:/usr/bin:/bin", "MG_STATE": str(self.state), "MG_CALLS": str(self.calls),
               "MG_MOUNTS": str(self.mounts),
               "MG_PVS": self.pvs}
        p = subprocess.run(["bash", str(self.copy), dev, fstype], env=env, capture_output=True, text=True)
        made = self.calls.read_text().split("\n")[0] if self.calls.exists() else ""
        return p.returncode, p.stderr.strip(), made

    def close(self):
        self.t.cleanup()


def disk_sdb(sdb1_fstype="ext4", sdb2_fstype=""):
    return [{"name": "/dev/sdb", "pk": "", "fstype": ""},
            {"name": "/dev/sdb1", "pk": "sdb", "fstype": sdb1_fstype},
            {"name": "/dev/sdb2", "pk": "sdb", "fstype": sdb2_fstype}]


ROOT = ("/dev/sda2", "/")
OWN_SDB1 = [("/dev/sdb1", "/mnt/hdd_1"), ("/dev/sdb1", "/mnt/felhom-drives/hdd_1")]


class Partitions(unittest.TestCase):
    def check(self, mounts, dev="/dev/sdb2", nodes=None, want_ok=True, **kw):
        d = Disk(nodes or disk_sdb(), [ROOT] + mounts, **kw)
        try:
            rc, err, made = d.run(dev)
            if want_ok:
                self.assertEqual(rc, 0, err)
                self.assertTrue(made.endswith(f"mkfs.ext4 -F {dev}"), made)
            else:
                self.assertEqual(rc, 1, (mounts, made))
                self.assertIn("REFUSED", err)
                self.assertEqual(made, "", "a refused format must not run mkfs")
            return err
        finally:
            d.close()

    def test_a_free_part_on_an_unmounted_disk_is_formatted(self):
        self.check([])   # positive control for every refusal below

    def test_the_free_part_next_to_felhoms_own_drive_is_refused(self):
        # The agent could make the „corroborating" managed mount itself: the root wrapper does not trust it.
        self.check(OWN_SDB1, want_ok=False)

    def test_a_sibling_mounted_raw_only_is_refused(self):
        self.check([("/dev/sdb1", "/mnt/hdd_1")], want_ok=False)

    def test_a_sibling_mounted_elsewhere_is_refused_even_with_a_managed_mount(self):
        for elsewhere in ("/srv/data", "/media/usb", "/mnt/a/b", "/home/x"):
            self.check([("/dev/sdb1", elsewhere), ("/dev/sdb1", "/mnt/felhom-drives/hdd_1")], want_ok=False)

    def test_the_target_itself_mounted_raw_is_refused(self):
        self.check([("/dev/sdb2", "/mnt/films"), ("/dev/sdb2", "/mnt/felhom-drives/films")], want_ok=False)

    def test_the_target_mounted_only_under_felhom_drives_is_formatted(self):
        self.check([("/dev/sdb2", "/mnt/felhom-drives/films")])

    def test_an_unlistable_disk_is_refused(self):
        self.check([], want_ok=False, lsblk_empty=True)

    def test_the_catastrophic_cases_still_refuse(self):
        self.check([("/dev/sdb1", "/")], want_ok=False)                     # system disk
        self.check([], nodes=disk_sdb(sdb1_fstype="LVM2_member"), want_ok=False)  # a member signature
        self.check([], pvs="/dev/sdb1", want_ok=False)                      # an LVM PV (pvs)
        self.check([], ro="1", want_ok=False)                               # read-only


if __name__ == "__main__":
    unittest.main()