Add read-only web dashboard and phone push notifications

Both zero-dependency (stdlib http.server + urllib), consistent with the
project's no-dependency-tree stance for a security daemon.

Web dashboard (enodia_sentinel/web.py + static/dashboard.html):
- read-only JSON API over the log dir: /api/status, /api/alerts,
  /api/alerts/<id>, /api/events
- bearer-token auth (constant-time; header or ?token=), required on non-
  loopback binds, auto-generated + persisted (0600) when unset
- binds the host's Tailscale IP by default (auto-detected), reachable from the
  tailnet but not the LAN/internet
- self-contained dark SPA: severity cards, live alert list, full snapshot
  viewer; 10s auto-refresh
- path-traversal-safe alert lookup; `enodia-sentinel web` subcommand; daemon
  now writes a pidfile so the dashboard can show live status
- hardened enodia-sentinel-web.service (read-only, no caps)

Phone push (enodia_sentinel/notify/):
- pluggable backends — ntfy, Pushover, generic webhook — each separating a pure
  build() (unit-tested, no network) from send()
- a backend turns on when its config keys are set; pushes gated by
  notify_min_severity; severity → per-service priority/tags
- fired from snapshot.capture on worker threads, errors swallowed
- desktop notify-send retained

Tests: +16 (9 web incl. a real-server 401/200 auth test, 7 notify request-build
cases). 55/55 pass. Live end-to-end verified: daemon → alert → API → page.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Luna 2026-05-31 16:18:00 -07:00
parent 0eb5077551
commit c00fff224c
17 changed files with 960 additions and 5 deletions

85
tests/test_notify.py Normal file
View file

@ -0,0 +1,85 @@
# SPDX-License-Identifier: GPL-3.0-or-later
"""Tests for notification request construction (no network)."""
import json
import unittest
from enodia_sentinel.alert import Alert, Severity
from enodia_sentinel.config import Config
from enodia_sentinel.notify import Notification, backends, enabled_backends, min_severity
def notif(sev=Severity.CRITICAL):
alerts = [Alert(sev, "reverse_shell", "rsh:1", "detail", (1,), 100010,
"c2-reverse-shell")]
return Notification.from_alerts(alerts, host="woofbox",
snapshot_name="alert-x.log",
dashboard_url="https://sentinel.example")
class TestNotification(unittest.TestCase):
def test_summary_fields(self):
n = notif()
self.assertEqual(n.severity, Severity.CRITICAL)
self.assertIn("reverse_shell", n.signatures)
self.assertIn(100010, n.sids)
self.assertIn("woofbox", n.title())
self.assertIn("alert-x.log", n.message())
self.assertIn("sentinel.example", n.message())
class TestEnable(unittest.TestCase):
def test_backends_off_by_default(self):
self.assertEqual(enabled_backends(Config()), [])
def test_ntfy_enabled_when_configured(self):
c = Config()
c.notify_ntfy_url = "https://ntfy.sh"
c.notify_ntfy_topic = "enodia-secret"
self.assertIn(backends.Ntfy, enabled_backends(c))
def test_min_severity(self):
c = Config()
c.notify_min_severity = "CRITICAL"
self.assertEqual(min_severity(c), Severity.CRITICAL)
class TestNtfy(unittest.TestCase):
def test_build(self):
c = Config()
c.notify_ntfy_url = "https://ntfy.sh/"
c.notify_ntfy_topic = "enodia-secret"
c.notify_ntfy_token = "tok_123"
req = backends.Ntfy.build(c, notif(Severity.CRITICAL))
self.assertEqual(req.full_url, "https://ntfy.sh/enodia-secret")
self.assertEqual(req.get_method(), "POST")
self.assertEqual(req.headers["Priority"], "5")
self.assertEqual(req.headers["Authorization"], "Bearer tok_123")
self.assertIn(b"reverse_shell", req.data)
class TestPushover(unittest.TestCase):
def test_build(self):
c = Config()
c.notify_pushover_token = "app"
c.notify_pushover_user = "usr"
req = backends.Pushover.build(c, notif(Severity.HIGH))
self.assertIn("api.pushover.net", req.full_url)
body = req.data.decode()
self.assertIn("token=app", body)
self.assertIn("user=usr", body)
self.assertIn("priority=0", body)
class TestWebhook(unittest.TestCase):
def test_build_json(self):
c = Config()
c.notify_webhook_url = "https://hook.example/x"
req = backends.Webhook.build(c, notif())
payload = json.loads(req.data)
self.assertEqual(payload["host"], "woofbox")
self.assertEqual(payload["severity"], "CRITICAL")
self.assertEqual(req.headers["Content-type"], "application/json")
if __name__ == "__main__":
unittest.main()

126
tests/test_web.py Normal file
View file

@ -0,0 +1,126 @@
# SPDX-License-Identifier: GPL-3.0-or-later
"""Tests for the read-only web dashboard: data layer + auth (real server)."""
import json
import tempfile
import threading
import unittest
import urllib.error
import urllib.request
from pathlib import Path
from enodia_sentinel import web
from enodia_sentinel.config import Config
def _make_cfg(tmp: Path) -> Config:
c = Config()
c.log_dir = tmp
return c
def _write_alert(tmp: Path, name: str, severity: str, sigs):
alerts = [{"signature": s, "sid": 100010, "severity": severity} for s in sigs]
(tmp / f"{name}.json").write_text(json.dumps({
"time": "2026-05-31T00:00:00-07:00", "host": "woofbox",
"severity": severity, "alerts": alerts,
}))
(tmp / f"{name}.log").write_text(f"=== ENODIA SENTINEL ALERT ===\n{severity}\n")
class TestDataLayer(unittest.TestCase):
def setUp(self):
self.dir = tempfile.TemporaryDirectory()
self.tmp = Path(self.dir.name)
self.cfg = _make_cfg(self.tmp)
def tearDown(self):
self.dir.cleanup()
def test_list_and_status(self):
_write_alert(self.tmp, "alert-20260531-000001", "CRITICAL", ["reverse_shell"])
_write_alert(self.tmp, "alert-20260531-000002", "HIGH", ["new_listener"])
alerts = web.list_alerts(self.cfg)
self.assertEqual(len(alerts), 2)
self.assertEqual(alerts[0]["name"], "alert-20260531-000002.log") # newest first
st = web.daemon_status(self.cfg)
self.assertEqual(st["total_alerts"], 2)
self.assertEqual(st["counts"]["CRITICAL"], 1)
self.assertFalse(st["running"]) # no live pidfile
def test_get_alert_and_traversal(self):
_write_alert(self.tmp, "alert-20260531-000001", "CRITICAL", ["x"])
got = web.get_alert(self.cfg, "alert-20260531-000001.log")
self.assertIn("text", got)
self.assertIn("json", got)
# path traversal / bad names rejected
self.assertIsNone(web.get_alert(self.cfg, "../../etc/passwd"))
self.assertIsNone(web.get_alert(self.cfg, "events.log"))
def test_tail_events(self):
(self.tmp / "events.log").write_text("l1\nl2\nl3\n")
self.assertEqual(web.tail_events(self.cfg, 2), ["l2", "l3"])
class TestNetworkHelpers(unittest.TestCase):
def test_is_loopback(self):
self.assertTrue(web.is_loopback("127.0.0.1"))
self.assertFalse(web.is_loopback("100.64.1.2"))
def test_resolve_bind_explicit(self):
c = Config()
c.web_bind = "100.64.1.2"
self.assertEqual(web.resolve_bind(c), "100.64.1.2")
def test_ensure_token_persists(self):
with tempfile.TemporaryDirectory() as d:
c = _make_cfg(Path(d))
t1 = web.ensure_token(c)
t2 = web.ensure_token(c)
self.assertTrue(t1)
self.assertEqual(t1, t2) # stable across calls
class TestAuth(unittest.TestCase):
"""Spin up the real server on loopback and check token enforcement."""
def setUp(self):
self.dir = tempfile.TemporaryDirectory()
self.tmp = Path(self.dir.name)
_write_alert(self.tmp, "alert-20260531-000001", "CRITICAL", ["reverse_shell"])
self.cfg = _make_cfg(self.tmp)
self.cfg.web_bind = "127.0.0.1"
self.cfg.web_port = 0 # ephemeral
self.cfg.web_token = "secret-token"
self.httpd, _bind, _tok = web.build_server(self.cfg)
self.port = self.httpd.server_address[1]
self.t = threading.Thread(target=self.httpd.serve_forever, daemon=True)
self.t.start()
def tearDown(self):
self.httpd.shutdown()
self.dir.cleanup()
def _get(self, path, token=None):
url = f"http://127.0.0.1:{self.port}{path}"
req = urllib.request.Request(url)
if token:
req.add_header("Authorization", f"Bearer {token}")
return urllib.request.urlopen(req, timeout=4)
def test_unauthorized_without_token(self):
with self.assertRaises(urllib.error.HTTPError) as cm:
self._get("/api/status")
self.assertEqual(cm.exception.code, 401)
def test_authorized_with_token(self):
resp = self._get("/api/status", token="secret-token")
data = json.loads(resp.read())
self.assertEqual(data["total_alerts"], 1)
def test_token_via_query_param(self):
resp = self._get("/api/alerts?token=secret-token")
self.assertEqual(resp.status, 200)
if __name__ == "__main__":
unittest.main()