diff --git a/python_scripts/systemd_service_auditor.py b/python_scripts/systemd_service_auditor.py index e12b62a..a4c6ada 100644 --- a/python_scripts/systemd_service_auditor.py +++ b/python_scripts/systemd_service_auditor.py @@ -1,50 +1,387 @@ #!/usr/bin/env python3 +"""Audit running systemd services for unit files in unexpected locations. + +A unit file that lives outside the standard systemd directories (or inside a +world-writable directory such as ``/tmp``) is a classic persistence trick: an +attacker who can drop a service file there gets their payload restarted on +every boot. This script enumerates the services that are running *right now* +and reports the ones whose ``FragmentPath`` looks wrong. + +How it works +------------ +1. ``systemctl list-units --type=service --state=running`` gives the service + names. ``--plain`` is passed so the output has no bullet/colour markers and + can be parsed reliably. +2. For each name, ``systemctl show --property=FragmentPath`` gives the + unit file that systemd actually loaded. +3. The path is resolved (symlinks followed) and compared *by path component* + against the standard directories. Component comparison matters: a naive + ``str.startswith("/lib/systemd/system")`` would happily accept + ``/lib/systemd/system-evil/backdoor.service``. + +Trust boundary +-------------- +Everything printed by ``systemctl`` is treated as untrusted input. Unit names +are validated against a conservative character class before they are handed +back to a subprocess, every ``systemctl`` call is bounded by a timeout, and a +failure to query one unit is recorded and reported instead of being silently +swallowed. + +Exit codes +---------- +0 No non-standard or high-risk unit files were found. +1 At least one non-standard or high-risk unit file was found. +2 The audit could not run (systemctl missing, timed out, or failed). + +Usage +----- + python3 systemd_service_auditor.py + python3 systemd_service_auditor.py --json + python3 systemd_service_auditor.py --standard-path /opt/systemd/system + python3 systemd_service_auditor.py --no-resolve-symlinks --timeout 5 + +Requires systemd on the host; the script itself makes no state changes. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re import subprocess +import sys +from dataclasses import asdict, dataclass, field + +SYSTEMCTL = "systemctl" + +#: Directories a distribution ships unit files in. Override with +#: ``--standard-path`` when auditing a system that keeps units elsewhere. +DEFAULT_STANDARD_DIRS: tuple[str, ...] = ( + "/lib/systemd/system", + "/usr/lib/systemd/system", + "/etc/systemd/system", +) + +#: Directories that are writable by non-root users or are wiped on reboot. +HIGH_RISK_DIRS: tuple[str, ...] = ( + "/tmp", + "/var/tmp", + "/dev/shm", + "/home", +) -def audit_systemd_services(): +#: Default per-call timeout for a ``systemctl`` invocation, in seconds. +DEFAULT_TIMEOUT = 15.0 + +#: Unit names systemd accepts: alphanumerics plus ``: _ . @ -`` and the +#: ``\x2d``-style escapes some units use. Anything else is not passed on to a +#: subprocess. +SERVICE_NAME_RE = re.compile(r"^[A-Za-z0-9_.@:\\-]+$") + +SEVERITY_HIGH = "high" +SEVERITY_WARNING = "warning" +SEVERITY_INFO = "info" + + +class AuditError(RuntimeError): + """Raised when the audit cannot produce a trustworthy result. + + Used for conditions that make the whole run meaningless (systemctl absent, + timing out, or exiting non-zero) rather than for a single bad unit, which + is recorded in :attr:`AuditReport.errors` and does not abort the scan. """ - Evaluates active systemd services, tracking metadata traces - to find non-standard startup configs. + + +@dataclass(frozen=True) +class Finding: + """One suspicious (or merely noteworthy) service unit file.""" + + service: str + fragment_path: str + reason: str + severity: str + + +@dataclass +class AuditReport: + """Result of a single audit run.""" + + findings: list[Finding] = field(default_factory=list) + errors: list[str] = field(default_factory=list) + services_checked: int = 0 + + @property + def has_problems(self) -> bool: + """True when a finding warrants a non-zero exit code.""" + return any(f.severity in (SEVERITY_HIGH, SEVERITY_WARNING) for f in self.findings) + + +# --------------------------------------------------------------------------- +# systemctl helpers +# --------------------------------------------------------------------------- + +def _run_systemctl(args: list[str], timeout: float = DEFAULT_TIMEOUT) -> str: + """Run ``systemctl `` and return its stdout. + + ``args`` is always a list, so no shell is involved and unit names cannot be + interpreted as shell syntax. Raises :class:`AuditError` if systemctl is + missing, exceeds ``timeout``, or exits non-zero. """ - print("=== Systemd Service Auditor ===") - print("[*] Listing active services and checking for non-standard paths...") - try: - # Get active services - result = subprocess.run(['systemctl', 'list-units', '--type=service', '--state=running', '--no-legend'], - capture_output=True, text=True) - lines = result.stdout.splitlines() - except Exception as e: - print(f"[!] Error running systemctl: {e}") - return - - non_standard = 0 - for line in lines: - service_name = line.split()[0] - - # Check service fragment path + completed = subprocess.run( + [SYSTEMCTL, *args], + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + except FileNotFoundError as exc: + raise AuditError("systemctl not found; is systemd installed?") from exc + except subprocess.TimeoutExpired as exc: + raise AuditError( + f"systemctl {args[0] if args else ''} timed out after {timeout}s" + ) from exc + except OSError as exc: # e.g. permission denied on the binary + raise AuditError(f"could not execute systemctl: {exc}") from exc + + if completed.returncode != 0: + detail = (completed.stderr or completed.stdout or "").strip() + raise AuditError( + f"systemctl {' '.join(args)} exited {completed.returncode}" + + (f": {detail}" if detail else "") + ) + return completed.stdout + + +def list_running_services(timeout: float = DEFAULT_TIMEOUT) -> list[str]: + """Return the names of currently running service units. + + Blank lines, the ``●`` bullet some systemd versions emit, and names that do + not match :data:`SERVICE_NAME_RE` are dropped rather than forwarded to a + second subprocess. + """ + output = _run_systemctl( + ["list-units", "--type=service", "--state=running", "--no-legend", "--plain"], + timeout=timeout, + ) + + services: list[str] = [] + for raw_line in output.splitlines(): + line = raw_line.strip().lstrip("\u25cf\u2022* ").strip() + if not line: + continue + name = line.split()[0] + if SERVICE_NAME_RE.match(name): + services.append(name) + return services + + +def get_fragment_path(service: str, timeout: float = DEFAULT_TIMEOUT) -> str | None: + """Return the ``FragmentPath`` systemd reports for ``service``. + + Returns ``None`` when the property is present but empty, which is normal + for transient units that have no unit file on disk. + """ + if not SERVICE_NAME_RE.match(service): + raise ValueError(f"refusing to query malformed unit name: {service!r}") + + output = _run_systemctl( + ["show", service, "--property=FragmentPath"], timeout=timeout + ) + for line in output.splitlines(): + key, sep, value = line.partition("=") + if sep and key.strip() == "FragmentPath": + return value.strip() or None + return None + + +# --------------------------------------------------------------------------- +# Classification +# --------------------------------------------------------------------------- + +def _is_within(path: str, directory: str) -> bool: + """True when ``path`` is ``directory`` itself or sits beneath it. + + Comparison is done on normalised path components, so ``/tmp-evil`` is not + considered to be inside ``/tmp`` and ``/lib/systemd/system-evil`` is not + considered to be a standard systemd directory. + """ + path = os.path.normpath(path) + directory = os.path.normpath(directory) + return path == directory or path.startswith(directory + os.sep) + + +def classify_path( + service: str, + fragment_path: str | None, + standard_dirs: tuple[str, ...] = DEFAULT_STANDARD_DIRS, + high_risk_dirs: tuple[str, ...] = HIGH_RISK_DIRS, + resolve_symlinks: bool = True, +) -> Finding | None: + """Classify one unit file location, returning a :class:`Finding` or ``None``. + + Symlinks are resolved first: ``/etc/systemd/system/foo.service`` is very + often a symlink to ``/lib/systemd/system/foo.service`` and is perfectly + normal, so flagging it would be a false positive. + """ + if not fragment_path: + return Finding(service, "", "missing-fragment-path", SEVERITY_INFO) + + path = os.path.realpath(fragment_path) if resolve_symlinks else fragment_path + path = os.path.normpath(path) + + if any(_is_within(path, d) for d in high_risk_dirs): + return Finding(service, fragment_path, "high-risk-path", SEVERITY_HIGH) + + if not any(_is_within(path, d) for d in standard_dirs): + return Finding(service, fragment_path, "non-standard-path", SEVERITY_WARNING) + + return None + + +def audit_services( + standard_dirs: tuple[str, ...] = DEFAULT_STANDARD_DIRS, + high_risk_dirs: tuple[str, ...] = HIGH_RISK_DIRS, + timeout: float = DEFAULT_TIMEOUT, + resolve_symlinks: bool = True, +) -> AuditReport: + """Audit every running service and return the findings. + + A failure while querying a single unit is appended to + :attr:`AuditReport.errors` and the scan continues, so one unreadable unit + cannot mask the rest of the report. Raises :class:`AuditError` if the unit + list itself cannot be retrieved. + """ + report = AuditReport() + services = list_running_services(timeout=timeout) + report.services_checked = len(services) + + for service in services: try: - show_result = subprocess.run(['systemctl', 'show', service_name, '--property=FragmentPath'], - capture_output=True, text=True) - path = show_result.stdout.strip().split('=')[-1] - - # Standard paths are usually /lib/systemd/system/ or /etc/systemd/system/ - if not any(path.startswith(p) for p in ['/lib/systemd/system', '/usr/lib/systemd/system', '/etc/systemd/system']): - if path: # Ensure path isn't empty - print(f"[!] NON-STANDARD PATH: {service_name} -> {path}") - non_standard += 1 - - # Check for services running from /tmp or /home - if any(p in path for p in ['/tmp', '/home', '/var/tmp']): - print(f" [!] HIGH RISK: Service {service_name} is running from {path}") - - except Exception: + fragment_path = get_fragment_path(service, timeout=timeout) + except (AuditError, ValueError) as exc: + report.errors.append(f"{service}: {exc}") continue + finding = classify_path( + service, + fragment_path, + standard_dirs=standard_dirs, + high_risk_dirs=high_risk_dirs, + resolve_symlinks=resolve_symlinks, + ) + if finding is not None: + report.findings.append(finding) + + return report + + +# --------------------------------------------------------------------------- +# Reporting +# --------------------------------------------------------------------------- + +def print_report(report: AuditReport) -> None: + """Print a human-readable report in the tool's traditional format.""" + for finding in report.findings: + if finding.reason == "high-risk-path": + print(f"[!] NON-STANDARD PATH: {finding.service} -> {finding.fragment_path}") + print(f" [!] HIGH RISK: Service {finding.service} is running from {finding.fragment_path}") + elif finding.reason == "non-standard-path": + print(f"[!] NON-STANDARD PATH: {finding.service} -> {finding.fragment_path}") + else: # missing-fragment-path - informational only + print(f"[*] NO UNIT FILE: {finding.service} has no FragmentPath (transient unit?)") + + for error in report.errors: + print(f"[!] Could not inspect {error}") + + non_standard = sum(1 for f in report.findings if f.severity != SEVERITY_INFO) if non_standard == 0: print("[+] All active systemd services are running from standard locations.") else: print(f"[*] Audit complete. {non_standard} non-standard services found.") + +def report_to_dict(report: AuditReport) -> dict: + """Serialise a report for ``--json`` consumers.""" + return { + "services_checked": report.services_checked, + "findings": [asdict(f) for f in report.findings], + "errors": report.errors, + "has_problems": report.has_problems, + } + + +def audit_systemd_services() -> AuditReport: + """Run the default audit and print it. + + Kept as the historical entry point so ``from systemd_service_auditor + import audit_systemd_services`` keeps working; prefer :func:`audit_services` + when you want the data without the printing. + """ + print("=== Systemd Service Auditor ===") + print("[*] Listing active services and checking for non-standard paths...") + + try: + report = audit_services() + except AuditError as exc: + print(f"[!] Error running systemctl: {exc}") + return AuditReport(errors=[str(exc)]) + + print_report(report) + return report + + +def main(argv: list[str] | None = None) -> int: + """Command-line entry point. Returns the process exit code.""" + parser = argparse.ArgumentParser( + description="Flag running systemd services whose unit files live outside standard locations.", + ) + parser.add_argument( + "--standard-path", + dest="standard_paths", + action="append", + metavar="DIR", + help="treat DIR as a standard unit directory (repeatable)", + ) + parser.add_argument( + "--timeout", + type=float, + default=DEFAULT_TIMEOUT, + metavar="SECONDS", + help=f"per-systemctl-call timeout (default: {DEFAULT_TIMEOUT})", + ) + parser.add_argument( + "--json", + action="store_true", + help="emit machine-readable JSON instead of text", + ) + parser.add_argument( + "--no-resolve-symlinks", + action="store_true", + help="classify the raw FragmentPath without resolving symlinks", + ) + args = parser.parse_args(argv) + + standard_dirs = tuple(args.standard_paths) if args.standard_paths else DEFAULT_STANDARD_DIRS + + try: + report = audit_services( + standard_dirs=standard_dirs, + timeout=args.timeout, + resolve_symlinks=not args.no_resolve_symlinks, + ) + except AuditError as exc: + print(f"[!] Error running systemctl: {exc}", file=sys.stderr) + return 2 + + if args.json: + print(json.dumps(report_to_dict(report), indent=2)) + else: + print_report(report) + + return 1 if report.has_problems else 0 + + if __name__ == "__main__": - audit_systemd_services() + sys.exit(main()) diff --git a/tests/test_systemd_service_auditor.py b/tests/test_systemd_service_auditor.py new file mode 100644 index 0000000..aa89fd0 --- /dev/null +++ b/tests/test_systemd_service_auditor.py @@ -0,0 +1,360 @@ +"""Tests for python_scripts/systemd_service_auditor.py. + +Focus areas: + * parsing of ``systemctl`` output, including blank lines, the ``●`` bullet + marker and unit names that must not reach a subprocess + * the trust boundary around unit names (no shell, validated names) + * path classification - component boundaries and symlink resolution, the + two places where a naive ``str.startswith``/``in`` check goes wrong + * failure handling - missing systemctl, timeouts, non-zero exits, and a + single unreadable unit not aborting the whole audit + * the exit-code contract used by CI +""" + +import json + +import pytest + +import systemd_service_auditor as auditor + +# --- Fakes ------------------------------------------------------------------ + + +class FakeSystemctl: + """Stands in for ``auditor._run_systemctl``. + + ``responses`` maps the first systemctl argument to either a string + (returned as stdout) or an exception instance to raise. + """ + + def __init__(self, responses): + self.responses = responses + self.calls = [] + + def __call__(self, args, timeout=auditor.DEFAULT_TIMEOUT): + self.calls.append(list(args)) + key = args[0] + if key not in self.responses: + raise AssertionError(f"unexpected systemctl call: {args}") + value = self.responses[key] + if isinstance(value, Exception): + raise value + if callable(value): + return value(args) + return value + + +@pytest.fixture +def fake(monkeypatch): + def install(responses): + stub = FakeSystemctl(responses) + monkeypatch.setattr(auditor, "_run_systemctl", stub) + return stub + + return install + + +# --- list_running_services -------------------------------------------------- + +def test_list_running_services_parses_names(fake): + fake({"list-units": "cron.service loaded active running Regular background\n" + "ssh.service loaded active running OpenBSD Secure Shell\n"}) + assert auditor.list_running_services() == ["cron.service", "ssh.service"] + + +def test_list_running_services_skips_blank_and_bullet_lines(fake): + fake({"list-units": "\n\u25cf broken.service loaded failed failed Broken\n" + " \nsshd.service loaded active running ssh\n"}) + assert auditor.list_running_services() == ["broken.service", "sshd.service"] + + +def test_list_running_services_drops_malformed_names(fake): + fake({"list-units": "evil;rm -rf /.service loaded active running nope\n" + "good.service loaded active running fine\n"}) + assert auditor.list_running_services() == ["good.service"] + + +def test_list_running_services_requests_plain_output(fake): + stub = fake({"list-units": ""}) + auditor.list_running_services() + assert "--plain" in stub.calls[0] + assert "--no-legend" in stub.calls[0] + + +def test_list_running_services_propagates_audit_error(fake): + fake({"list-units": auditor.AuditError("systemctl list-units exited 1")}) + with pytest.raises(auditor.AuditError): + auditor.list_running_services() + + +def test_list_running_services_raises_on_nonzero_exit(fake): + fake({"list-units": auditor.AuditError("systemctl list-units exited 1: no bus")}) + with pytest.raises(auditor.AuditError): + auditor.list_running_services() + + +# --- _run_systemctl --------------------------------------------------------- + +def test_run_systemctl_reports_nonzero_exit(monkeypatch): + class Completed: + returncode = 1 + stdout = "" + stderr = "Failed to connect to bus\n" + + monkeypatch.setattr(auditor.subprocess, "run", lambda *a, **kw: Completed()) + with pytest.raises(auditor.AuditError, match="Failed to connect to bus"): + auditor._run_systemctl(["list-units"]) + + +def test_run_systemctl_translates_missing_binary(monkeypatch): + def raise_missing(*args, **kwargs): + raise FileNotFoundError("systemctl") + + monkeypatch.setattr(auditor.subprocess, "run", raise_missing) + with pytest.raises(auditor.AuditError, match="systemctl not found"): + auditor._run_systemctl(["list-units"]) + + +def test_run_systemctl_translates_timeout(monkeypatch): + def raise_timeout(*args, **kwargs): + raise auditor.subprocess.TimeoutExpired("systemctl", 5) + + monkeypatch.setattr(auditor.subprocess, "run", raise_timeout) + with pytest.raises(auditor.AuditError, match="timed out after 5"): + auditor._run_systemctl(["list-units"], timeout=5) + + +def test_run_systemctl_translates_os_error(monkeypatch): + def raise_oserror(*args, **kwargs): + raise PermissionError("denied") + + monkeypatch.setattr(auditor.subprocess, "run", raise_oserror) + with pytest.raises(auditor.AuditError, match="could not execute systemctl"): + auditor._run_systemctl(["list-units"]) + + +def test_run_systemctl_passes_argv_without_shell(monkeypatch): + captured = {} + + class Completed: + returncode = 0 + stdout = "FragmentPath=/etc/systemd/system/x.service\n" + stderr = "" + + def fake_run(argv, **kwargs): + captured["argv"] = argv + captured["kwargs"] = kwargs + return Completed() + + monkeypatch.setattr(auditor.subprocess, "run", fake_run) + auditor._run_systemctl(["show", "x.service", "--property=FragmentPath"]) + assert captured["argv"] == ["systemctl", "show", "x.service", "--property=FragmentPath"] + assert "shell" not in captured["kwargs"] + assert captured["kwargs"]["timeout"] == auditor.DEFAULT_TIMEOUT + + +# --- get_fragment_path ------------------------------------------------------ + +def test_get_fragment_path_parses_property(fake): + fake({"show": "FragmentPath=/etc/systemd/system/cron.service\n"}) + assert auditor.get_fragment_path("cron.service") == "/etc/systemd/system/cron.service" + + +def test_get_fragment_path_empty_value_is_none(fake): + fake({"show": "FragmentPath=\n"}) + assert auditor.get_fragment_path("cron.service") is None + + +def test_get_fragment_path_missing_property_is_none(fake): + fake({"show": ""}) + assert auditor.get_fragment_path("cron.service") is None + + +def test_get_fragment_path_rejects_malformed_name(fake): + stub = fake({}) + with pytest.raises(ValueError, match="malformed unit name"): + auditor.get_fragment_path("evil service") + assert stub.calls == [] + + +# --- classify_path ---------------------------------------------------------- + +def test_classify_standard_path_returns_none(): + assert auditor.classify_path("cron.service", "/lib/systemd/system/cron.service") is None + + +def test_classify_standard_path_under_usr_lib(): + assert auditor.classify_path("a.service", "/usr/lib/systemd/system/a.service") is None + + +def test_classify_prefix_boundary_is_not_standard(): + """Regression: '/lib/systemd/system-evil' must not pass a startswith() check.""" + finding = auditor.classify_path("x.service", "/lib/systemd/system-evil/x.service") + assert finding is not None + assert finding.reason == "non-standard-path" + assert finding.severity == auditor.SEVERITY_WARNING + + +def test_classify_substring_is_not_high_risk(): + """Regression: '/opt/homegrown' must not match a bare 'in /home' test.""" + finding = auditor.classify_path("x.service", "/opt/homegrown/x.service") + assert finding is not None + assert finding.reason == "non-standard-path" + + +def test_classify_sibling_of_home_is_not_high_risk(): + finding = auditor.classify_path("x.service", "/home2/x.service") + assert finding.reason == "non-standard-path" + + +@pytest.mark.parametrize("path", [ + "/tmp/x.service", + "/var/tmp/x.service", + "/dev/shm/x.service", + "/home/krish/x.service", +]) +def test_classify_high_risk_directories(path): + finding = auditor.classify_path("x.service", path) + assert finding is not None + assert finding.reason == "high-risk-path" + assert finding.severity == auditor.SEVERITY_HIGH + + +def test_classify_missing_path_is_informational(): + finding = auditor.classify_path("run-abc.scope", None) + assert finding is not None + assert finding.reason == "missing-fragment-path" + assert finding.severity == auditor.SEVERITY_INFO + + +def test_classify_resolves_symlink_to_standard_dir(tmp_path): + target = tmp_path / "standard" + target.mkdir() + real_unit = target / "cron.service" + real_unit.write_text("[Unit]\n", encoding="utf-8") + link = tmp_path / "cron.service" + link.symlink_to(real_unit) + + # Without resolution the symlink location looks non-standard ... + unresolved = auditor.classify_path( + "cron.service", str(link), standard_dirs=(str(target),), resolve_symlinks=False + ) + assert unresolved is not None + assert unresolved.reason == "non-standard-path" + + # ... and with resolution it is correctly treated as standard. + assert auditor.classify_path( + "cron.service", str(link), standard_dirs=(str(target),), resolve_symlinks=True + ) is None + + +def test_classify_custom_standard_dir(): + assert auditor.classify_path( + "x.service", "/opt/systemd/system/x.service", + standard_dirs=("/opt/systemd/system",), + ) is None + + +# --- audit_services --------------------------------------------------------- + +def _show_map(mapping): + def responder(args): + return f"FragmentPath={mapping.get(args[1], '')}\n" + + return responder + + +def test_audit_services_end_to_end(fake): + fake({ + "list-units": "cron.service loaded active running cron\n" + "backdoor.service loaded active running evil\n" + "tmpd.service loaded active running evil\n", + "show": _show_map({ + "cron.service": "/lib/systemd/system/cron.service", + "backdoor.service": "/opt/payload/backdoor.service", + "tmpd.service": "/tmp/tmpd.service", + }), + }) + report = auditor.audit_services() + assert report.services_checked == 3 + assert {f.service for f in report.findings} == {"backdoor.service", "tmpd.service"} + assert report.has_problems is True + + +def test_audit_services_clean_run(fake): + fake({ + "list-units": "cron.service loaded active running cron\n", + "show": _show_map({"cron.service": "/lib/systemd/system/cron.service"}), + }) + report = auditor.audit_services() + assert report.findings == [] + assert report.errors == [] + assert report.has_problems is False + + +def test_audit_services_records_error_and_continues(fake): + def responder(args): + if args[1] == "broken.service": + raise auditor.AuditError("systemctl show broken.service exited 1") + return "FragmentPath=/tmp/x.service\n" + + fake({ + "list-units": "broken.service loaded active running x\n" + "bad.service loaded active running y\n", + "show": responder, + }) + report = auditor.audit_services() + assert report.services_checked == 2 + assert len(report.errors) == 1 and "broken.service" in report.errors[0] + assert [f.service for f in report.findings] == ["bad.service"] + + +def test_audit_services_propagates_list_failure(fake): + fake({"list-units": auditor.AuditError("systemctl not found")}) + with pytest.raises(auditor.AuditError): + auditor.audit_services() + + +# --- CLI -------------------------------------------------------------------- + +def test_main_clean_exit_code(fake, capsys): + fake({ + "list-units": "cron.service loaded active running cron\n", + "show": _show_map({"cron.service": "/lib/systemd/system/cron.service"}), + }) + assert auditor.main([]) == 0 + assert "standard locations" in capsys.readouterr().out + + +def test_main_findings_exit_code(fake): + fake({ + "list-units": "bad.service loaded active running x\n", + "show": _show_map({"bad.service": "/tmp/bad.service"}), + }) + assert auditor.main([]) == 1 + + +def test_main_unavailable_systemctl_exit_code(fake, capsys): + fake({"list-units": auditor.AuditError("systemctl not found")}) + assert auditor.main([]) == 2 + assert "systemctl not found" in capsys.readouterr().err + + +def test_main_json_output_is_parseable(fake, capsys): + fake({ + "list-units": "bad.service loaded active running x\n", + "show": _show_map({"bad.service": "/opt/x/bad.service"}), + }) + assert auditor.main(["--json"]) == 1 + payload = json.loads(capsys.readouterr().out) + assert payload["services_checked"] == 1 + assert payload["findings"][0]["service"] == "bad.service" + assert payload["has_problems"] is True + + +def test_main_honours_custom_standard_path(fake): + fake({ + "list-units": "x.service loaded active running x\n", + "show": _show_map({"x.service": "/opt/systemd/system/x.service"}), + }) + assert auditor.main(["--standard-path", "/opt/systemd/system"]) == 0