Add tray action layer over CLI and systemctl

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

enodia_sentinel/tray/actions.py

@@ -0,0 +1,96 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+"""Action layer for the tray applet: drives the CLI and systemctl.
+
+No GUI imports. Every subprocess call is timeout-bounded and check=False so a
+failed command degrades to an ActionResult rather than raising.
+"""
+from __future__ import annotations
+
+import json
+import subprocess
+import webbrowser
+from dataclasses import dataclass
+from typing import Callable, Sequence
+
+from ..config import Config
+
+Runner = Callable[[Sequence[str], int], "subprocess.CompletedProcess"]
+
+DAEMON_UNIT = "enodia-sentinel.service"
+WEB_UNIT = "enodia-sentinel-web.service"
+
+
+def _default_runner(cmd: Sequence[str], timeout: int) -> "subprocess.CompletedProcess":
+    try:
+        return subprocess.run(list(cmd), capture_output=True, text=True,
+                              timeout=timeout, check=False)
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        return subprocess.CompletedProcess(list(cmd), 127, "", str(exc))
+
+
+@dataclass(frozen=True)
+class ActionResult:
+    ok: bool
+    message: str
+
+
+def dashboard_url(cfg: Config) -> str:
+    return f"https://127.0.0.1:{cfg.web_port}/"
+
+
+def fetch_status(runner: Runner = _default_runner) -> "dict | None":
+    proc = runner(["enodia-sentinel", "status", "--json"], 10)
+    try:
+        data = json.loads(proc.stdout)
+        return data if isinstance(data, dict) else None
+    except (ValueError, TypeError):
+        return None
+
+
+def _systemctl(action: str, unit: str, runner: Runner) -> ActionResult:
+    proc = runner(["systemctl", action, unit], 30)
+    if proc.returncode == 0:
+        return ActionResult(True, f"{unit}: {action} ok")
+    lines = (proc.stderr or proc.stdout or "").strip().splitlines()
+    detail = lines[-1] if lines else f"exit {proc.returncode}"
+    return ActionResult(False, f"{unit}: {action} failed — {detail}")
+
+
+def start_daemon(runner: Runner = _default_runner) -> ActionResult:
+    return _systemctl("start", DAEMON_UNIT, runner)
+
+
+def stop_daemon(runner: Runner = _default_runner) -> ActionResult:
+    return _systemctl("stop", DAEMON_UNIT, runner)
+
+
+def restart_daemon(runner: Runner = _default_runner) -> ActionResult:
+    return _systemctl("restart", DAEMON_UNIT, runner)
+
+
+def is_active(unit: str, runner: Runner = _default_runner) -> bool:
+    proc = runner(["systemctl", "is-active", unit], 10)
+    return proc.stdout.strip() == "active"
+
+
+def open_dashboard(cfg: Config, runner: Runner = _default_runner,
+                   opener: Callable[[str], object] = webbrowser.open) -> ActionResult:
+    if not is_active(WEB_UNIT, runner):
+        started = _systemctl("start", WEB_UNIT, runner)
+        if not started.ok:
+            return started
+    url = dashboard_url(cfg)
+    opener(url)
+    return ActionResult(True, f"Opened {url}")
+
+
+def run_check(runner: Runner = _default_runner) -> ActionResult:
+    proc = runner(["enodia-sentinel", "check", "--json"], 120)
+    try:
+        alerts = json.loads(proc.stdout)
+    except (ValueError, TypeError):
+        return ActionResult(False, "Quick check failed to produce JSON")
+    count = len(alerts) if isinstance(alerts, list) else 0
+    if count == 0:
+        return ActionResult(True, "Quick check: no findings")
+    return ActionResult(True, f"Quick check: {count} finding(s)")

tests/test_tray_actions.py

@@ -0,0 +1,85 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+"""Tray action layer: drives CLI + systemctl through an injectable runner."""
+import json
+import subprocess
+import unittest
+
+from enodia_sentinel.config import Config
+from enodia_sentinel.tray import actions
+
+
+def _proc(rc=0, out="", err=""):
+    return subprocess.CompletedProcess(args=[], returncode=rc,
+                                       stdout=out, stderr=err)
+
+
+class _Recorder:
+    """Runner that returns scripted results and records the commands seen."""
+    def __init__(self, results):
+        self._results = list(results)
+        self.calls = []
+
+    def __call__(self, cmd, timeout):
+        self.calls.append(list(cmd))
+        return self._results.pop(0)
+
+
+class TestActions(unittest.TestCase):
+    def test_dashboard_url_is_loopback_with_port(self):
+        cfg = Config()
+        cfg.web_port = 9999
+        self.assertEqual(actions.dashboard_url(cfg), "https://127.0.0.1:9999/")
+
+    def test_fetch_status_parses_json_even_on_nonzero_exit(self):
+        runner = _Recorder([_proc(rc=1, out=json.dumps({"running": True}))])
+        self.assertEqual(actions.fetch_status(runner), {"running": True})
+
+    def test_fetch_status_returns_none_on_garbage(self):
+        runner = _Recorder([_proc(out="not json")])
+        self.assertIsNone(actions.fetch_status(runner))
+
+    def test_start_daemon_ok(self):
+        runner = _Recorder([_proc(rc=0)])
+        res = actions.start_daemon(runner)
+        self.assertTrue(res.ok)
+        self.assertEqual(runner.calls[0],
+                         ["systemctl", "start", "enodia-sentinel.service"])
+
+    def test_stop_daemon_failure_carries_stderr_tail(self):
+        runner = _Recorder([_proc(rc=1, err="line1\nAccess denied")])
+        res = actions.stop_daemon(runner)
+        self.assertFalse(res.ok)
+        self.assertIn("Access denied", res.message)
+
+    def test_open_dashboard_starts_web_if_inactive(self):
+        # is-active -> inactive, then start -> ok
+        runner = _Recorder([_proc(out="inactive"), _proc(rc=0)])
+        opened = []
+        res = actions.open_dashboard(Config(), runner, opener=opened.append)
+        self.assertTrue(res.ok)
+        self.assertEqual(runner.calls[0],
+                         ["systemctl", "is-active", "enodia-sentinel-web.service"])
+        self.assertEqual(runner.calls[1],
+                         ["systemctl", "start", "enodia-sentinel-web.service"])
+        self.assertEqual(opened, ["https://127.0.0.1:8787/"])
+
+    def test_open_dashboard_skips_start_when_active(self):
+        runner = _Recorder([_proc(out="active")])
+        opened = []
+        res = actions.open_dashboard(Config(), runner, opener=opened.append)
+        self.assertTrue(res.ok)
+        self.assertEqual(len(runner.calls), 1)
+        self.assertEqual(opened, ["https://127.0.0.1:8787/"])
+
+    def test_run_check_counts_findings(self):
+        runner = _Recorder([_proc(out=json.dumps([{"signature": "rsh"},
+                                                   {"signature": "suid"}]))])
+        res = actions.run_check(runner)
+        self.assertTrue(res.ok)
+        self.assertIn("2", res.message)
+
+    def test_run_check_no_findings(self):
+        runner = _Recorder([_proc(out="[]")])
+        res = actions.run_check(runner)
+        self.assertTrue(res.ok)
+        self.assertIn("no findings", res.message.lower())