Add systemd unit posture checks

Close the remaining v0.8 posture bullet: flag enabled systemd units with
group/world-writable unit files (systemd_unit_writable, sid 100050) and
units whose ExecStart binary runs from a writable/unsafe path such as
/tmp, /dev/shm, /var/tmp, /home, or /run/user (systemd_exec_unsafe_path,
sid 100051).

- posture.systemd_findings/parse_systemctl_show/read_systemd_units, wired
  into posture.run() behind the new posture_systemd config knob.
- read_systemd_units shells out to systemctl with timeouts and fails closed
  on non-systemd hosts; the dashboard /api/posture surfaces findings for free.
- Unit tests for the parser and evaluator; TestRunIntegration patches the
  new reader so run() never touches live systemctl.
- Docs: COMMAND_REFERENCE posture list, ROADMAP v0.8 bullet, RUNBOOKS
  Runbook 2 triggers, config TOML knob.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Luna 2026-06-19 02:27:19 -07:00
parent 11d91a40b4
commit 69798b26e4
7 changed files with 177 additions and 6 deletions

View file

@ -103,6 +103,50 @@ class TestSensitiveFiles(unittest.TestCase):
list(posture.sensitive_file_findings([("/etc/motd", 0o100666)])), [])
class TestSystemd(unittest.TestCase):
SHOW = (
"Id=clean.service\n"
"FragmentPath=/usr/lib/systemd/system/clean.service\n"
"ExecStart={ path=/usr/bin/cleand ; argv[]=/usr/bin/cleand }\n"
"UnitFileState=enabled\n"
"\n"
"Id=evil.service\n"
"FragmentPath=/etc/systemd/system/evil.service\n"
"ExecStart={ path=/tmp/payload ; argv[]=/tmp/payload --daemon }\n"
"UnitFileState=enabled\n"
)
def test_parse_systemctl_show(self):
parsed = posture.parse_systemctl_show(self.SHOW)
self.assertEqual(parsed, [
("clean.service", "/usr/lib/systemd/system/clean.service",
True, ("/usr/bin/cleand",)),
("evil.service", "/etc/systemd/system/evil.service",
True, ("/tmp/payload",)),
])
def test_writable_unit_file_is_high(self):
units = [posture.UnitFact("foo.service", "/etc/systemd/system/foo.service",
0o100664, True, ("/usr/bin/foo",))]
findings = list(posture.systemd_findings(units))
self.assertEqual(len(findings), 1)
self.assertEqual(findings[0].signature, "systemd_unit_writable")
self.assertEqual(findings[0].severity, Severity.HIGH)
def test_exec_from_unsafe_path_is_high(self):
units = [posture.UnitFact("evil.service", "/etc/systemd/system/evil.service",
0o100644, True, ("/tmp/payload",))]
findings = list(posture.systemd_findings(units))
self.assertEqual(len(findings), 1)
self.assertEqual(findings[0].signature, "systemd_exec_unsafe_path")
self.assertEqual(findings[0].severity, Severity.HIGH)
def test_clean_unit_yields_nothing(self):
units = [posture.UnitFact("clean.service", "/usr/lib/systemd/system/clean.service",
0o100644, True, ("/usr/bin/cleand",))]
self.assertEqual(list(posture.systemd_findings(units)), [])
class TestRunIntegration(unittest.TestCase):
"""Drive run() with every reader monkeypatched to known facts."""
@ -110,7 +154,7 @@ class TestRunIntegration(unittest.TestCase):
self.cfg = Config()
self._saved = {n: getattr(posture, n) for n in
("read_sshd_lines", "read_sudoers", "read_path_dirs",
"read_sensitive_files")}
"read_sensitive_files", "read_systemd_units")}
import enodia_sentinel.pkgdb as pkgdb
self._saved_sig = pkgdb.siglevel_alert
pkgdb.siglevel_alert = lambda *a, **k: None
@ -127,6 +171,7 @@ class TestRunIntegration(unittest.TestCase):
posture.read_sudoers = lambda m, d: [("/etc/sudoers", "root ALL=(ALL) ALL\n", 0o440)]
posture.read_path_dirs = lambda dirs: [("/usr/bin", 0o40755, 0)]
posture.read_sensitive_files = lambda *a, **k: [("/etc/shadow", 0o100600)]
posture.read_systemd_units = lambda *a, **k: []
self.assertEqual(list(posture.run(self.cfg)), [])
def test_messy_host_aggregates_findings(self):
@ -135,6 +180,7 @@ class TestRunIntegration(unittest.TestCase):
("/etc/sudoers", "bob ALL=(ALL) NOPASSWD: ALL\n", 0o440)]
posture.read_path_dirs = lambda dirs: [("/usr/local/bin", 0o40777, 0)]
posture.read_sensitive_files = lambda *a, **k: [("/etc/shadow", 0o100644)]
posture.read_systemd_units = lambda *a, **k: []
sigs = {f.signature for f in posture.run(self.cfg)}
self.assertSetEqual(
sigs,