Add status command (roadmap v0.8: status --json)
Add `enodia-sentinel status [--json]` — a local health and alert summary: daemon running state, heartbeat freshness, per-severity alert counts, last alert time, and eBPF state. Reuses web.daemon_status (the same data behind /api/status), so the CLI and dashboard never diverge. Exit code reflects health (0 = running and fresh, 1 = down or stale), so it doubles as a cron/monitoring probe. Three CLI tests drive it against a temp log_dir; no daemon or root needed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
9ebc355936
commit
f2d896a2d7
4 changed files with 99 additions and 2 deletions
|
|
@ -182,6 +182,23 @@ Exit code:
|
||||||
|
|
||||||
## Evidence and Operator Commands
|
## Evidence and Operator Commands
|
||||||
|
|
||||||
|
### `status`
|
||||||
|
|
||||||
|
```bash
|
||||||
|
enodia-sentinel status
|
||||||
|
enodia-sentinel status --json
|
||||||
|
```
|
||||||
|
|
||||||
|
Prints a local health and alert summary: daemon running state, heartbeat age and
|
||||||
|
staleness, total alerts with per-severity counts, last alert time, and eBPF
|
||||||
|
monitor state. `--json` emits the same data as the dashboard `/api/status`
|
||||||
|
endpoint, for cron/monitoring.
|
||||||
|
|
||||||
|
Exit code:
|
||||||
|
|
||||||
|
- `0`: daemon is running and the heartbeat is fresh.
|
||||||
|
- `1`: daemon is down or the heartbeat is stale.
|
||||||
|
|
||||||
### `web`
|
### `web`
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
|
|
@ -238,7 +255,6 @@ does not require pip or a virtualenv.
|
||||||
The roadmap reserves these command shapes:
|
The roadmap reserves these command shapes:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
enodia-sentinel status --json
|
|
||||||
enodia-sentinel incident list
|
enodia-sentinel incident list
|
||||||
enodia-sentinel incident show <incident-id>
|
enodia-sentinel incident show <incident-id>
|
||||||
enodia-sentinel incident export <incident-id>
|
enodia-sentinel incident export <incident-id>
|
||||||
|
|
|
||||||
|
|
@ -31,7 +31,7 @@ work that is bigger than single alerts.
|
||||||
world-writable PATH components, loose sensitive-file permissions, and disabled
|
world-writable PATH components, loose sensitive-file permissions, and disabled
|
||||||
package signatures. Still to come: unexpected enabled services and unsafe
|
package signatures. Still to come: unexpected enabled services and unsafe
|
||||||
systemd unit settings.
|
systemd unit settings.
|
||||||
- Add a machine-readable `status --json` command for automation.
|
- ✅ Add a machine-readable `status --json` command for automation.
|
||||||
- Update the red-team harness so every current detection has a named drill and
|
- Update the red-team harness so every current detection has a named drill and
|
||||||
expected `sid`.
|
expected `sid`.
|
||||||
- Document incident response runbooks for reverse shells, persistence, trojaned
|
- Document incident response runbooks for reverse shells, persistence, trojaned
|
||||||
|
|
|
||||||
|
|
@ -75,6 +75,8 @@ def main(argv: list[str] | None = None) -> int:
|
||||||
help="packages to verify (0 = config default; rotates)")
|
help="packages to verify (0 = config default; rotates)")
|
||||||
sub.add_parser("rootcheck",
|
sub.add_parser("rootcheck",
|
||||||
help="anti-rootkit cross-view: hidden procs/modules/ports")
|
help="anti-rootkit cross-view: hidden procs/modules/ports")
|
||||||
|
st = sub.add_parser("status", help="print daemon health/alert summary")
|
||||||
|
st.add_argument("--json", action="store_true", help="emit status as JSON")
|
||||||
po = sub.add_parser("posture",
|
po = sub.add_parser("posture",
|
||||||
help="audit host config hygiene (SSH, sudo, PATH, perms)")
|
help="audit host config hygiene (SSH, sudo, PATH, perms)")
|
||||||
po.add_argument("action", nargs="?", default="check", choices=["check"],
|
po.add_argument("action", nargs="?", default="check", choices=["check"],
|
||||||
|
|
@ -123,6 +125,8 @@ def main(argv: list[str] | None = None) -> int:
|
||||||
return _cmd_pkgdb_verify(cfg, args.sample)
|
return _cmd_pkgdb_verify(cfg, args.sample)
|
||||||
if args.cmd == "rootcheck":
|
if args.cmd == "rootcheck":
|
||||||
return _cmd_rootcheck(cfg)
|
return _cmd_rootcheck(cfg)
|
||||||
|
if args.cmd == "status":
|
||||||
|
return _cmd_status(cfg, args.json)
|
||||||
if args.cmd == "posture":
|
if args.cmd == "posture":
|
||||||
return _cmd_posture(cfg, args.json)
|
return _cmd_posture(cfg, args.json)
|
||||||
if args.cmd == "watchdog":
|
if args.cmd == "watchdog":
|
||||||
|
|
@ -178,6 +182,31 @@ def _cmd_rootcheck(cfg: Config) -> int:
|
||||||
return 1
|
return 1
|
||||||
|
|
||||||
|
|
||||||
|
def _cmd_status(cfg: Config, as_json: bool) -> int:
|
||||||
|
from .web import daemon_status
|
||||||
|
s = daemon_status(cfg)
|
||||||
|
if as_json:
|
||||||
|
import json
|
||||||
|
print(json.dumps(s, indent=2))
|
||||||
|
else:
|
||||||
|
age = s["heartbeat_age"]
|
||||||
|
if age is None:
|
||||||
|
hb = "none (daemon never wrote one)"
|
||||||
|
elif s["heartbeat_stale"]:
|
||||||
|
hb = f"STALE — {int(age)}s ago (> {cfg.heartbeat_max_age}s)"
|
||||||
|
else:
|
||||||
|
hb = f"{int(age)}s ago"
|
||||||
|
counts = ", ".join(f"{k}: {v}" for k, v in sorted(s["counts"].items())) or "none"
|
||||||
|
print(f"enodia-sentinel {s['version']} on {s['host']}")
|
||||||
|
print(f" daemon: {'running' if s['running'] else 'NOT running'}")
|
||||||
|
print(f" heartbeat: {hb}")
|
||||||
|
print(f" alerts: {s['total_alerts']} total ({counts})")
|
||||||
|
print(f" last alert: {s['last_alert'] or 'none'}")
|
||||||
|
print(f" eBPF: {s['ebpf']}")
|
||||||
|
# Health gate for monitoring: unhealthy if the daemon is down or silent.
|
||||||
|
return 0 if (s["running"] and not s["heartbeat_stale"]) else 1
|
||||||
|
|
||||||
|
|
||||||
def _cmd_posture(cfg: Config, as_json: bool) -> int:
|
def _cmd_posture(cfg: Config, as_json: bool) -> int:
|
||||||
from . import posture
|
from . import posture
|
||||||
findings = sorted(posture.run(cfg), key=lambda x: -x.severity)
|
findings = sorted(posture.run(cfg), key=lambda x: -x.severity)
|
||||||
|
|
|
||||||
52
tests/test_status.py
Normal file
52
tests/test_status.py
Normal file
|
|
@ -0,0 +1,52 @@
|
||||||
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||||
|
"""`enodia-sentinel status` tests — drive the CLI against a temp log_dir, so no
|
||||||
|
daemon or root is needed. Exercises the health-gate exit code and --json."""
|
||||||
|
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
|
||||||
|
|
||||||
|
|
||||||
|
class TestStatusCommand(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
self.dir = tempfile.TemporaryDirectory()
|
||||||
|
os.environ["ENODIA_LOG_DIR"] = self.dir.name
|
||||||
|
self.tmp = Path(self.dir.name)
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
os.environ.pop("ENODIA_LOG_DIR", None)
|
||||||
|
self.dir.cleanup()
|
||||||
|
|
||||||
|
def _run(self, *args):
|
||||||
|
buf = io.StringIO()
|
||||||
|
with redirect_stdout(buf):
|
||||||
|
code = main(["status", *args])
|
||||||
|
return code, buf.getvalue()
|
||||||
|
|
||||||
|
def test_no_heartbeat_is_unhealthy(self):
|
||||||
|
code, out = self._run()
|
||||||
|
self.assertEqual(code, 1) # daemon down + no heartbeat
|
||||||
|
self.assertIn("NOT running", out)
|
||||||
|
self.assertIn("none", out)
|
||||||
|
|
||||||
|
def test_stale_heartbeat_is_unhealthy(self):
|
||||||
|
(self.tmp / "heartbeat").write_text("1") # epoch 1 = ancient
|
||||||
|
code, _ = self._run()
|
||||||
|
self.assertEqual(code, 1)
|
||||||
|
|
||||||
|
def test_json_shape(self):
|
||||||
|
code, out = self._run("--json")
|
||||||
|
data = json.loads(out)
|
||||||
|
for key in ("version", "running", "total_alerts", "counts",
|
||||||
|
"heartbeat_age", "heartbeat_stale", "host"):
|
||||||
|
self.assertIn(key, data)
|
||||||
|
self.assertEqual(code, 1) # nothing running in the temp dir
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Loading…
Add table
Add a link
Reference in a new issue