diff --git a/README.md b/README.md index 1aa50a6..863554b 100644 --- a/README.md +++ b/README.md @@ -214,6 +214,26 @@ python3 -m enodia_sentinel.cli check # run every detector once, print fin No pip, no virtualenv, no dependencies — it's stdlib-only and installs as a plain package directory plus a launcher wrapper. +### Terminal TUI and completion + +For SSH/tmux sessions, use the stdlib curses dashboard: + +```bash +enodia-sentinel tui +# or, when installed from pyproject metadata: +enodia-sentinel-tui +``` + +Keys: `1` status, `2` alerts, `3` incidents, `4` rules, `r` refresh, `/` filter, +`:` command mode, `Tab` complete command names, `q` quit. + +Shell completion is generated without extra dependencies: + +```bash +enodia-sentinel completion bash > ~/.local/share/bash-completion/completions/enodia-sentinel +enodia-sentinel completion zsh > ~/.zfunc/_enodia-sentinel +``` + ### Desktop tray (optional) A lightweight system-tray applet is available for desktop installs. It is an diff --git a/docs/COMMAND_REFERENCE.md b/docs/COMMAND_REFERENCE.md index 4165bf3..795d294 100644 --- a/docs/COMMAND_REFERENCE.md +++ b/docs/COMMAND_REFERENCE.md @@ -365,6 +365,31 @@ Exit code: - `0`: daemon is running and the heartbeat is fresh. - `1`: daemon is down or the heartbeat is stale. +### `tui` / `enodia-sentinel-tui` + +```bash +enodia-sentinel tui +enodia-sentinel-tui +``` + +Opens the stdlib curses terminal dashboard for SSH/tmux operators. It does not +require the optional tray dependencies and reads the same local state as the CLI: +daemon status, recent alert snapshots, incident index, and event-rule metadata. + +Keys: + +- `1` / `2` / `3` / `4`: status, alerts, incidents, rules. +- `j` / `k` or arrows: scroll. +- `r`: refresh. +- `/`: filter the current view. +- `:`: command mode. +- `Tab`: complete command names in command mode. +- `Ctrl-L`: redraw. +- `q`: quit. + +Command mode supports `status`, `alerts`, `incidents`, `rules`, `help`, +`refresh`, `filter `, `clear`, and `quit`. + ### `web` ```bash @@ -401,6 +426,25 @@ from `status --json`, and runs `check --json` on demand with the result shown as a desktop notification. Requires the `[tray]` extra (`pystray`, `Pillow`); the core agent stays dependency-free without it. +### `completion` + +```bash +enodia-sentinel completion bash +enodia-sentinel completion zsh +``` + +Prints a shell completion script for the CLI. Example installs: + +```bash +mkdir -p ~/.local/share/bash-completion/completions +enodia-sentinel completion bash > ~/.local/share/bash-completion/completions/enodia-sentinel + +mkdir -p ~/.zfunc +enodia-sentinel completion zsh > ~/.zfunc/_enodia-sentinel +``` + +For zsh, ensure `~/.zfunc` is in `fpath` before `compinit`. + ### `triage` ```bash diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index 7bcd310..aa434c4 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -77,6 +77,10 @@ The HTTPS dashboard shows the same posture findings under the Posture tab and summarizes FIM/package anchors plus heartbeat freshness under the Integrity tab for quick remote review. +For SSH/tmux operators, `enodia-sentinel tui` provides a terminal dashboard over +the same local state: status, recent alerts, incidents, and rules. Use `:` for +command mode and `Tab` to complete TUI commands. + On desktop hosts the optional tray applet (`enodia-sentinel-tray`, `[tray]` extra) wraps the same workflow: it drives the same `systemctl` units operators use directly, so daemon start/stop/restart prompts for polkit authorization as diff --git a/docs/PACKAGING.md b/docs/PACKAGING.md index 44f094d..93caef9 100644 --- a/docs/PACKAGING.md +++ b/docs/PACKAGING.md @@ -114,6 +114,17 @@ empty. not installed to a system autostart path by default — desktop users copy it to `~/.config/autostart/` themselves. +## Terminal TUI and Completion + +The terminal dashboard is stdlib-only and available through +`enodia-sentinel tui` in source/wrapper installs. Python packaging metadata also +declares `enodia-sentinel-tui = enodia_sentinel.tui:main` for pip-style script +generation. + +Shell completion has no runtime dependency: operators generate scripts with +`enodia-sentinel completion bash` or `enodia-sentinel completion zsh` and install +them in their normal shell completion path. + ## Package Hardening Checklist - Package version matches `pyproject.toml` and `enodia_sentinel.__version__`. diff --git a/enodia_sentinel/cli.py b/enodia_sentinel/cli.py index a00ad86..d877a5c 100644 --- a/enodia_sentinel/cli.py +++ b/enodia_sentinel/cli.py @@ -208,6 +208,7 @@ def main(argv: list[str] | None = None) -> int: help="sid for show, event JSON path for test, or '-'") rules.add_argument("--json", action="store_true", help="emit JSON") sub.add_parser("web", help="serve the read-only dashboard") + sub.add_parser("tui", help="open the terminal dashboard (curses)") sub.add_parser("triage", help="classify captured alerts as likely-FP vs review") sub.add_parser("fim-baseline", help="build the file-integrity baseline") sub.add_parser("fim-update", help="refresh the FIM baseline (run by the pacman hook)") @@ -249,6 +250,10 @@ def main(argv: list[str] | None = None) -> int: help="heartbeat staleness threshold (seconds)") wd.add_argument("--insecure-tls", action="store_true", help="allow self-signed/untrusted dashboard certificates") + comp = sub.add_parser("completion", + help="print shell completion script for bash or zsh") + comp.add_argument("shell", choices=["bash", "zsh"], + help="shell to generate completion for") args = parser.parse_args(argv) cfg = Config.load(args.config) @@ -270,6 +275,9 @@ def main(argv: list[str] | None = None) -> int: from .web import serve serve(cfg) return 0 + if args.cmd == "tui": + from .tui import run as run_tui + return run_tui(cfg) if args.cmd == "triage": return _cmd_triage(cfg) if args.cmd in ("fim-baseline", "fim-update"): @@ -301,6 +309,10 @@ def main(argv: list[str] | None = None) -> int: if args.cmd == "watchdog": return _cmd_watchdog(cfg, args.url, args.token, args.max_age, not args.insecure_tls) + if args.cmd == "completion": + from .completion import script + print(script(args.shell), end="") + return 0 return _cmd_run(cfg) diff --git a/enodia_sentinel/completion.py b/enodia_sentinel/completion.py new file mode 100644 index 0000000..3e1aeba --- /dev/null +++ b/enodia_sentinel/completion.py @@ -0,0 +1,102 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Shell completion script generation.""" +from __future__ import annotations + +COMMANDS = ( + "run", "check", "baseline", "list-detectors", "rules", "web", "tui", + "triage", "fim-baseline", "fim-update", "fim-check", "pkgdb-check", + "pkgdb-verify", "rootcheck", "status", "incident", "posture", "respond", + "watchdog", "completion", +) + +RULE_ACTIONS = ("list", "show", "test", "docs") +BASELINE_ACTIONS = ("build", "accept", "revoke", "list") +INCIDENT_ACTIONS = ("list", "show", "export") +POSTURE_ACTIONS = ("check",) +RESPOND_ACTIONS = ("plan",) +COMPLETION_SHELLS = ("bash", "zsh") + + +def bash_completion(prog: str = "enodia-sentinel") -> str: + commands = " ".join(COMMANDS) + rule_actions = " ".join(RULE_ACTIONS) + baseline_actions = " ".join(BASELINE_ACTIONS) + incident_actions = " ".join(INCIDENT_ACTIONS) + posture_actions = " ".join(POSTURE_ACTIONS) + respond_actions = " ".join(RESPOND_ACTIONS) + shells = " ".join(COMPLETION_SHELLS) + return f"""# bash completion for {prog} +_{prog.replace('-', '_')}_complete() {{ + local cur prev cmd + COMPREPLY=() + cur="${{COMP_WORDS[COMP_CWORD]}}" + prev="${{COMP_WORDS[COMP_CWORD-1]}}" + cmd="" + for word in "${{COMP_WORDS[@]:1}}"; do + case "$word" in + -* ) ;; + * ) cmd="$word"; break ;; + esac + done + if [[ $COMP_CWORD -le 1 || -z "$cmd" ]]; then + COMPREPLY=( $(compgen -W "--version --config -c {commands}" -- "$cur") ) + return 0 + fi + case "$cmd" in + rules) COMPREPLY=( $(compgen -W "{rule_actions} --json" -- "$cur") ) ;; + baseline) COMPREPLY=( $(compgen -W "{baseline_actions} --reason --expires --force --stale --json" -- "$cur") ) ;; + incident) COMPREPLY=( $(compgen -W "{incident_actions} --json" -- "$cur") ) ;; + posture) COMPREPLY=( $(compgen -W "{posture_actions} --json" -- "$cur") ) ;; + respond) COMPREPLY=( $(compgen -W "{respond_actions} --json" -- "$cur") ) ;; + completion) COMPREPLY=( $(compgen -W "{shells}" -- "$cur") ) ;; + check|fim-check|status) COMPREPLY=( $(compgen -W "--json --packages" -- "$cur") ) ;; + pkgdb-verify) COMPREPLY=( $(compgen -W "--sample" -- "$cur") ) ;; + watchdog) COMPREPLY=( $(compgen -W "--url --token --max-age --insecure-tls" -- "$cur") ) ;; + esac +}} +complete -F _{prog.replace('-', '_')}_complete {prog} +""" + + +def zsh_completion(prog: str = "enodia-sentinel") -> str: + command_specs = " ".join(f"{c}:{c}" for c in COMMANDS) + return f"""#compdef {prog} +# zsh completion for {prog} +local -a commands +commands=({command_specs}) + +_arguments -C \\ + '(-h --help)'{{-h,--help}}'[show help]' \\ + '--version[show version]' \\ + '(-c --config)'{{-c,--config}}'[path to TOML config]:config file:_files' \\ + '1:command:->command' \\ + '*::arg:->args' + +case $state in + command) + _describe 'command' commands + ;; + args) + case $words[2] in + rules) _arguments '1:action:((list show test docs))' '--json[emit JSON]' ;; + baseline) _arguments '1:action:((build accept revoke list))' '2:kind:((fim pkgfile listener suid))' '--reason[reason]' '--expires[TTL]' '--force[force]' '--stale[stale]' '--json[emit JSON]' ;; + incident) _arguments '1:action:((list show export))' '--json[emit JSON]' ;; + posture) _arguments '1:action:((check))' '--json[emit JSON]' ;; + respond) _arguments '1:action:((plan))' '--json[emit JSON]' ;; + completion) _arguments '1:shell:((bash zsh))' ;; + check|status) _arguments '--json[emit JSON]' ;; + fim-check) _arguments '--packages[verify package-owned files]' ;; + pkgdb-verify) _arguments '--sample[package sample size]' ;; + watchdog) _arguments '--url[dashboard URL]' '--token[bearer token]' '--max-age[seconds]' '--insecure-tls[allow self-signed TLS]' ;; + esac + ;; +esac +""" + + +def script(shell: str, prog: str = "enodia-sentinel") -> str: + if shell == "bash": + return bash_completion(prog) + if shell == "zsh": + return zsh_completion(prog) + raise ValueError(f"unsupported shell: {shell}") diff --git a/enodia_sentinel/tui.py b/enodia_sentinel/tui.py new file mode 100644 index 0000000..c9b0124 --- /dev/null +++ b/enodia_sentinel/tui.py @@ -0,0 +1,331 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Stdlib curses terminal UI for SSH/tmux operators.""" +from __future__ import annotations + +import curses +import json +import textwrap +from dataclasses import dataclass, field +from typing import Iterable + +from . import __version__, incident, ruleops +from .config import Config + + +VIEWS = ("status", "alerts", "incidents", "rules", "help") +TUI_COMMANDS = { + "status": "show daemon health and heartbeat", + "alerts": "show recent captured alert snapshots", + "incidents": "show grouped incidents", + "rules": "show event-rule catalog", + "help": "show keys and commands", + "refresh": "reload data from disk", + "filter": "filter current view: filter ", + "clear": "clear filter and message", + "quit": "exit the TUI", +} + + +@dataclass +class TuiState: + view: str = "status" + filter_text: str = "" + message: str = "" + scroll: int = 0 + command: str = "" + command_mode: bool = False + should_quit: bool = False + model: dict = field(default_factory=dict) + + +def complete_command(prefix: str, commands: Iterable[str] = TUI_COMMANDS) -> str: + """Return the unique command completion or the common prefix of matches.""" + if not prefix: + return prefix + matches = sorted(c for c in commands if c.startswith(prefix)) + if not matches: + return prefix + if len(matches) == 1: + return matches[0] + common = matches[0] + for match in matches[1:]: + i = 0 + while i < min(len(common), len(match)) and common[i] == match[i]: + i += 1 + common = common[:i] + return common or prefix + + +def collect_model(cfg: Config) -> dict: + from .web import daemon_status + + return { + "status": daemon_status(cfg), + "alerts": _recent_alerts(cfg), + "incidents": sorted( + incident.load_index(cfg).values(), + key=lambda i: i.get("last_ts", 0.0), + reverse=True, + ), + "rules": ruleops.list_rules(cfg), + } + + +def run_tui_command(state: TuiState, command: str) -> TuiState: + raw = command.strip() + if not raw: + state.command_mode = False + return state + name, _, rest = raw.partition(" ") + if name in VIEWS: + state.view = name + state.scroll = 0 + state.message = f"view: {name}" + elif name == "refresh": + state.message = "refreshed" + elif name == "filter": + state.filter_text = rest.strip() + state.scroll = 0 + state.message = f"filter: {state.filter_text or 'none'}" + elif name == "clear": + state.filter_text = "" + state.message = "" + state.scroll = 0 + elif name == "quit": + state.should_quit = True + else: + state.message = f"unknown command: {name}" + state.command = "" + state.command_mode = False + return state + + +def render_lines(state: TuiState, width: int = 100) -> list[str]: + model = state.model or {} + view = state.view if state.view in VIEWS else "status" + if view == "status": + lines = _render_status(model.get("status") or {}, width) + elif view == "alerts": + lines = _render_alerts(model.get("alerts") or [], width) + elif view == "incidents": + lines = _render_incidents(model.get("incidents") or [], width) + elif view == "rules": + lines = _render_rules(model.get("rules") or [], width) + else: + lines = _render_help(width) + if state.filter_text: + needle = state.filter_text.lower() + lines = [line for line in lines if needle in line.lower()] + if not lines: + lines = [f"No rows match filter: {state.filter_text}"] + return lines + + +def run(cfg: Config) -> int: + curses.wrapper(lambda stdscr: _loop(stdscr, cfg)) + return 0 + + +def main(argv=None) -> int: + import argparse + + parser = argparse.ArgumentParser( + prog="enodia-sentinel-tui", + description="Terminal dashboard for Enodia Sentinel.", + ) + parser.add_argument("-c", "--config", help="path to TOML config") + args = parser.parse_args(argv) + return run(Config.load(args.config)) + + +def _loop(stdscr, cfg: Config) -> None: + curses.curs_set(0) + stdscr.nodelay(False) + state = TuiState(model=collect_model(cfg)) + while not state.should_quit: + _draw(stdscr, state) + key = stdscr.getch() + if state.command_mode: + _handle_command_key(state, key) + if not state.command_mode and not state.should_quit: + if state.message == "refreshed": + state.model = collect_model(cfg) + state.command = "" + continue + if key in (ord("q"), ord("Q")): + state.should_quit = True + elif key in (ord("r"), ord("R")): + state.model = collect_model(cfg) + state.message = "refreshed" + elif key in (ord("1"),): + state.view, state.scroll = "status", 0 + elif key in (ord("2"),): + state.view, state.scroll = "alerts", 0 + elif key in (ord("3"),): + state.view, state.scroll = "incidents", 0 + elif key in (ord("4"),): + state.view, state.scroll = "rules", 0 + elif key in (ord("h"), ord("?")): + state.view, state.scroll = "help", 0 + elif key == ord(":"): + state.command_mode = True + state.command = "" + curses.curs_set(1) + elif key == ord("/"): + state.command_mode = True + state.command = "filter " + curses.curs_set(1) + elif key in (curses.KEY_DOWN, ord("j")): + state.scroll += 1 + elif key in (curses.KEY_UP, ord("k")): + state.scroll = max(0, state.scroll - 1) + elif key == curses.KEY_NPAGE: + state.scroll += 10 + elif key == curses.KEY_PPAGE: + state.scroll = max(0, state.scroll - 10) + elif key == 12: # Ctrl-L + stdscr.clear() + + +def _handle_command_key(state: TuiState, key: int) -> None: + if key in (10, 13, curses.KEY_ENTER): + run_tui_command(state, state.command) + curses.curs_set(0) + elif key in (27,): + state.command_mode = False + state.command = "" + curses.curs_set(0) + elif key in (9,): + head, sep, tail = state.command.partition(" ") + if sep: + return + state.command = complete_command(head) + elif key in (curses.KEY_BACKSPACE, 127, 8): + state.command = state.command[:-1] + elif 32 <= key <= 126: + state.command += chr(key) + + +def _draw(stdscr, state: TuiState) -> None: + height, width = stdscr.getmaxyx() + width = max(width, 20) + stdscr.erase() + title = f" Enodia Sentinel TUI v{__version__} " + tabs = " 1 Status 2 Alerts 3 Incidents 4 Rules h Help r Refresh q Quit" + stdscr.addnstr(0, 0, (title + tabs)[:width - 1], width - 1, curses.A_REVERSE) + meta = f"view={state.view} filter={state.filter_text or '-'} {state.message}" + stdscr.addnstr(1, 0, meta[:width - 1], width - 1) + lines = render_lines(state, width) + body_h = max(0, height - 4) + max_scroll = max(0, len(lines) - body_h) + state.scroll = min(state.scroll, max_scroll) + for row, line in enumerate(lines[state.scroll:state.scroll + body_h], start=2): + stdscr.addnstr(row, 0, line[:width - 1], width - 1) + prompt = ":" + state.command if state.command_mode else "Press ':' for commands; Tab completes command names." + stdscr.addnstr(height - 1, 0, prompt[:width - 1], width - 1, curses.A_REVERSE) + if state.command_mode: + stdscr.move(height - 1, min(width - 1, len(prompt))) + stdscr.refresh() + + +def _render_status(status: dict, width: int) -> list[str]: + if not status: + return ["No status available."] + age = status.get("heartbeat_age") + heartbeat = "none" if age is None else f"{int(age)}s ago" + if status.get("heartbeat_stale"): + heartbeat += " (STALE)" + counts = status.get("counts") or {} + count_text = ", ".join(f"{k}:{v}" for k, v in sorted(counts.items())) or "none" + return [ + f"Host: {status.get('host', '?')}", + f"Version: {status.get('version', '?')}", + f"Daemon: {'running' if status.get('running') else 'stopped'}", + f"Heartbeat: {heartbeat}", + f"Alerts: {status.get('total_alerts', 0)} total ({count_text})", + f"Last alert: {status.get('last_alert') or 'none'}", + f"eBPF: {status.get('ebpf', 'unknown')}", + f"Exec probe: {status.get('ebpf_exec', 'unknown')}", + f"Syscall: {status.get('ebpf_syscall', 'unknown')}", + "", + "Use 2/3/4 to inspect alerts, incidents, and rules. Use ':' for commands.", + ] + + +def _render_alerts(alerts: list[dict], width: int) -> list[str]: + if not alerts: + return ["No alert snapshots found."] + lines = [f"{'TIME':<24} {'SEV':<9} {'SID':<7} SIGNATURE / DETAIL"] + for snap in alerts: + for alert in snap.get("alerts", []): + detail = alert.get("detail", "") + line = ( + f"{snap.get('time', '?')[:24]:<24} {alert.get('severity', '?'):<9} " + f"{alert.get('sid', 0):<7} {alert.get('signature', '?')} " + f"{detail[:max(20, width - 60)]}" + ) + lines.append(line) + return lines + + +def _render_incidents(incidents: list[dict], width: int) -> list[str]: + if not incidents: + return ["No incidents recorded."] + lines = [f"{'INCIDENT':<28} {'LAST SEEN':<24} {'SEV':<9} {'#':>3} SIGNATURES"] + for inc in incidents: + sigs = ", ".join(inc.get("signatures", [])) + lines.append( + f"{inc.get('id', '?'):<28} {inc.get('last_seen', '?')[:24]:<24} " + f"{inc.get('severity', '?'):<9} {inc.get('alert_count', 0):>3} {sigs}" + ) + return lines + + +def _render_rules(rules: list[dict], width: int) -> list[str]: + if not rules: + return ["No rules loaded."] + lines = [f"{'SID':<7} {'EVENT':<8} {'SEV':<9} {'CLASSTYPE':<24} MESSAGE"] + for rule in rules: + lines.append( + f"{rule.get('sid', 0):<7} {rule.get('event', '?'):<8} " + f"{rule.get('severity', '?'):<9} {rule.get('classtype', '?'):<24} " + f"{rule.get('msg', '')[:max(20, width - 52)]}" + ) + return lines + + +def _render_help(width: int) -> list[str]: + lines = [ + "Keys:", + " 1/2/3/4 switch views: status, alerts, incidents, rules", + " j/k or arrows scroll", + " PgUp/PgDn page scroll", + " r refresh", + " / filter current view", + " : command mode", + " Tab complete command names in command mode", + " Ctrl-L redraw", + " q quit", + "", + "Commands:", + ] + for name, help_text in TUI_COMMANDS.items(): + lines.append(f" {name:<10} {help_text}") + return [wrapped for line in lines for wrapped in (textwrap.wrap(line, width) or [""])] + + +def _recent_alerts(cfg: Config, limit: int = 50) -> list[dict]: + rows = [] + for path in sorted(cfg.log_dir.glob("alert-*.json"), reverse=True)[:limit]: + try: + data = json.loads(path.read_text()) + except (OSError, ValueError): + continue + if isinstance(data, dict): + data.setdefault("snapshot", path.name) + rows.append(data) + return rows + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pyproject.toml b/pyproject.toml index 0de2d5b..cfd4020 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,7 @@ dependencies = [] [project.scripts] enodia-sentinel = "enodia_sentinel.cli:main" +enodia-sentinel-tui = "enodia_sentinel.tui:main" enodia-sentinel-tray = "enodia_sentinel.tray.app:main" [project.optional-dependencies] diff --git a/tests/test_docs_versioning.py b/tests/test_docs_versioning.py index 5b6155d..9d7a73f 100644 --- a/tests/test_docs_versioning.py +++ b/tests/test_docs_versioning.py @@ -51,6 +51,8 @@ class TestDocsVersioning(unittest.TestCase): self.assertGreaterEqual(len(commands), 8) for phrase in ("Global options:", "Exit code:", "Expected use:"): self.assertIn(phrase, text) + self.assertIn("### `tui` / `enodia-sentinel-tui`", text) + self.assertIn("### `completion`", text) if __name__ == "__main__": diff --git a/tests/test_tray_imports.py b/tests/test_tray_imports.py index 10dc5de..746ff2b 100644 --- a/tests/test_tray_imports.py +++ b/tests/test_tray_imports.py @@ -26,5 +26,9 @@ class TestTrayBoundaries(unittest.TestCase): data["project"]["scripts"]["enodia-sentinel-tray"], "enodia_sentinel.tray.app:main", ) + self.assertEqual( + data["project"]["scripts"]["enodia-sentinel-tui"], + "enodia_sentinel.tui:main", + ) # Core runtime deps stay empty. self.assertEqual(data["project"]["dependencies"], []) diff --git a/tests/test_tui.py b/tests/test_tui.py new file mode 100644 index 0000000..52eb070 --- /dev/null +++ b/tests/test_tui.py @@ -0,0 +1,81 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +import io +import json +import os +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path + +from enodia_sentinel.cli import main +from enodia_sentinel.config import Config +from enodia_sentinel.tui import ( + TuiState, + collect_model, + complete_command, + render_lines, + run_tui_command, +) + + +class TestTuiCore(unittest.TestCase): + def setUp(self): + self.dir = tempfile.TemporaryDirectory() + self.tmp = Path(self.dir.name) + self.cfg = Config() + self.cfg.log_dir = self.tmp + + def tearDown(self): + self.dir.cleanup() + + def test_command_completion_unique_and_common_prefix(self): + self.assertEqual(complete_command("inc"), "incidents") + self.assertEqual(complete_command("r"), "r") + self.assertEqual(complete_command("zzz"), "zzz") + + def test_run_tui_command_switches_views_and_filters(self): + state = TuiState() + run_tui_command(state, "incidents") + self.assertEqual(state.view, "incidents") + run_tui_command(state, "filter ssh") + self.assertEqual(state.filter_text, "ssh") + run_tui_command(state, "clear") + self.assertEqual(state.filter_text, "") + run_tui_command(state, "quit") + self.assertTrue(state.should_quit) + + def test_collect_model_and_render_status(self): + model = collect_model(self.cfg) + state = TuiState(model=model) + lines = render_lines(state, width=80) + self.assertTrue(any("Daemon:" in line for line in lines)) + self.assertTrue(any("Heartbeat:" in line for line in lines)) + + def test_render_alerts_and_filter(self): + (self.tmp / "alert-1.json").write_text(json.dumps({ + "time": "2026-07-09T01:02:03-07:00", + "alerts": [{ + "severity": "CRITICAL", + "sid": 100010, + "signature": "reverse_shell", + "detail": "bash connected to 8.8.8.8", + }], + })) + state = TuiState(view="alerts", model=collect_model(self.cfg)) + self.assertTrue(any("reverse_shell" in line for line in render_lines(state))) + state.filter_text = "nomatch" + self.assertEqual(render_lines(state), ["No rows match filter: nomatch"]) + + def test_cli_completion_outputs_scripts(self): + for shell in ("bash", "zsh"): + with self.subTest(shell=shell): + buf = io.StringIO() + with redirect_stdout(buf): + code = main(["completion", shell]) + self.assertEqual(code, 0) + self.assertIn("enodia-sentinel", buf.getvalue()) + self.assertIn("tui", buf.getvalue()) + + +if __name__ == "__main__": + unittest.main()