From 4545b2b5a2adae37accb22759567f09eeb744cf2 Mon Sep 17 00:00:00 2001 From: Luna Date: Tue, 30 Jun 2026 06:05:48 -0700 Subject: [PATCH] Add tray action layer over CLI and systemctl Co-Authored-By: Claude Opus 4.8 --- enodia_sentinel/tray/actions.py | 96 +++++++++++++++++++++++++++++++++ tests/test_tray_actions.py | 85 +++++++++++++++++++++++++++++ 2 files changed, 181 insertions(+) create mode 100644 enodia_sentinel/tray/actions.py create mode 100644 tests/test_tray_actions.py diff --git a/enodia_sentinel/tray/actions.py b/enodia_sentinel/tray/actions.py new file mode 100644 index 0000000..a58fd12 --- /dev/null +++ b/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)") diff --git a/tests/test_tray_actions.py b/tests/test_tray_actions.py new file mode 100644 index 0000000..3907341 --- /dev/null +++ b/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())