96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
# 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)")
|