# SPDX-License-Identifier: GPL-3.0-or-later """Tests for the read-only web dashboard: data layer + auth (real server).""" import json import re import ssl import tempfile import threading import unittest import urllib.error import urllib.request from pathlib import Path from enodia_sentinel import web from enodia_sentinel import incident from enodia_sentinel.alert import Alert, Severity from enodia_sentinel.config import Config THEME_IDS = ( "console", "paper", "contrast", "pride", "trans", "dracula", "solarized-dark", "solarized-light", "twilight", ) 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") def _incident_alert(sig: str, pids=()): return Alert(Severity.CRITICAL, sig, f"k:{sig}", f"{sig} detail", tuple(pids), sid=100010, classtype="test") def _theme_blocks(html: str): blocks = {} root = re.search(r":root\{(?P.*?)\n \}", html, re.S) if root: blocks["console"] = _css_vars(root.group("body")) for name, body in re.findall(r':root\[data-theme="([^"]+)"\]\{(.*?)\n \}', html, re.S): blocks[name] = _css_vars(body) return blocks def _css_vars(block: str): return { name: value.strip() for name, value in re.findall(r"--([\w-]+):([^;]+);", block) } def _hex_rgb(value: str): value = value.strip() if not value.startswith("#"): return None if len(value) == 4: value = "#" + "".join(ch * 2 for ch in value[1:]) if not re.fullmatch(r"#[0-9A-Fa-f]{6}", value): return None return tuple(int(value[i:i + 2], 16) / 255.0 for i in (1, 3, 5)) def _linear_channel(value: float) -> float: if value <= 0.04045: return value / 12.92 return ((value + 0.055) / 1.055) ** 2.4 def _relative_luminance(rgb) -> float: r, g, b = (_linear_channel(part) for part in rgb) return 0.2126 * r + 0.7152 * g + 0.0722 * b def _contrast(foreground: str, background: str) -> float: fg = _hex_rgb(foreground) bg = _hex_rgb(background) if fg is None or bg is None: raise AssertionError(f"Expected hex colors, got {foreground!r} / {background!r}") high = max(_relative_luminance(fg), _relative_luminance(bg)) low = min(_relative_luminance(fg), _relative_luminance(bg)) return (high + 0.05) / (low + 0.05) 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"]) def test_status_reports_exec_and_syscall_ebpf(self): (self.tmp / "events.log").write_text( "2026 eBPF exec monitor: enabled\n" "2026 eBPF syscall monitor: disabled (not running as root)\n" ) st = web.daemon_status(self.cfg) self.assertEqual(st["ebpf_exec"], "enabled") self.assertEqual(st["ebpf_syscall"], "disabled (not running as root)") def test_incident_and_response_plan_data(self): _write_alert(self.tmp, "alert-20260531-000001", "CRITICAL", ["reverse_shell"]) iid = incident.record(self.cfg, "alert-20260531-000001.log", [_incident_alert("reverse_shell", [4242])], lineage={4242}, when=1000.0, host="h") incs = web.list_incidents(self.cfg) self.assertEqual(incs[0]["id"], iid) inc = web.get_incident(self.cfg, iid) self.assertEqual(inc["incident"]["id"], iid) self.assertEqual(len(inc["timeline"]), 1) plan = web.response_plan(self.cfg, iid) self.assertEqual(plan["incident_id"], iid) self.assertEqual(plan["mode"], "dry-run") def test_posture_report_shape(self): def runner(_cfg): yield Alert(Severity.HIGH, "ssh_root_login", "k", "root SSH login enabled", sid=100040, classtype="host-posture") yield Alert(Severity.MEDIUM, "sudo_nopasswd", "k2", "passwordless sudo", sid=100043, classtype="host-posture") report = web.posture_report(self.cfg, runner=runner) self.assertEqual(report["count"], 2) self.assertEqual(report["counts"], {"HIGH": 1, "MEDIUM": 1}) self.assertEqual(report["findings"][0]["signature"], "ssh_root_login") def test_rules_catalog_shape(self): catalog = web.rules_catalog(self.cfg) self.assertGreater(catalog["count"], 0) self.assertIn("exec", catalog["by_event"]) self.assertIn("syscall", catalog["by_event"]) self.assertIn("CRITICAL", catalog["by_severity"]) self.assertIn("builtin", catalog["by_origin"]) rule = next(r for r in catalog["rules"] if r["sid"] == 100002) self.assertEqual(rule["event"], "exec") self.assertIn("conditions", rule) def test_integrity_report_shape(self): (self.tmp / "fim-baseline.json").write_text(json.dumps({ "/etc/passwd": {"sha256": "a"}, "/etc/sudoers": {"sha256": "b"}, })) (self.tmp / "pkgdb-anchor.json").write_text(json.dumps({ "fingerprint": "0123456789abcdef0123456789abcdef", "time": 999.0, })) status = { "running": True, "heartbeat_age": 10.0, "heartbeat_stale": False, } report = web.integrity_report(self.cfg, status=status, now=1000.0) self.assertEqual(report["schema"], "enodia.integrity.v1") self.assertEqual(report["checks"]["watchdog"], "ok") self.assertEqual(report["anchors"]["fim_baseline"]["count"], 2) self.assertEqual(report["anchors"]["pkgdb"]["fingerprint_prefix"], "0123456789abcdef") self.assertTrue(report["read_only"]) self.assertIn("sentinel_footprint", report) self.assertIn("hash_chain", report) self.assertEqual(report["checks"]["hash_chain"], "missing") self.assertFalse(report["hash_chain"]["exists"]) def test_dashboard_has_persistent_theme_controls(self): html = (web._STATIC / "dashboard.html").read_text() self.assertIn('id="settingsMenu"', html) self.assertIn('id="settingsBtn"', html) self.assertIn('aria-label="Theme"', html) for theme in THEME_IDS: self.assertIn(f'data-theme="{theme}"', html) self.assertIn("sentinel.theme", html) self.assertIn("#55CDFC", html) self.assertIn("#F7A8B8", html) self.assertGreaterEqual(html.count("#55CDFC"), 8) self.assertGreaterEqual(html.count("#F7A8B8"), 8) trans_block = html.split(':root[data-theme="trans"]{', 1)[1].split("}", 1)[0] self.assertIn("--ink:#000000", trans_block) self.assertIn("--muted:#000000", trans_block) self.assertIn("--soft:#000000", trans_block) self.assertIn("--white:#000000", trans_block) self.assertIn("--panel:#FFFFFF", trans_block) self.assertIn("--panel2:#F7A8B8", trans_block) self.assertIn("--rail:#55CDFC", trans_block) self.assertNotIn("--amber:#FFFFFF", trans_block) self.assertNotIn("--red:#F7A8B8", trans_block) self.assertNotIn("--blue:#55CDFC", trans_block) def test_dashboard_theme_registry_matches_css(self): html = (web._STATIC / "dashboard.html").read_text() blocks = _theme_blocks(html) self.assertEqual(set(blocks), set(THEME_IDS)) options = re.search(r"const THEME_OPTIONS = \[(.*?)\];", html, re.S) self.assertIsNotNone(options) registered = tuple(re.findall(r'\["([^"]+)","[^"]+"\]', options.group(1))) self.assertEqual(registered, THEME_IDS) def test_dashboard_theme_text_contrast(self): html = (web._STATIC / "dashboard.html").read_text() blocks = _theme_blocks(html) token_pairs = ( ("ink", "bg"), ("ink", "panel"), ("muted", "panel"), ("soft", "panel2"), ("white", "panel"), ("active-ink", "accent"), ("red", "panel"), ("amber", "panel"), ("blue", "panel"), ("ok", "panel"), ) for theme, tokens in blocks.items(): with self.subTest(theme=theme): for token in ( "bg", "ink", "muted", "soft", "panel", "panel2", "accent", "amber", "red", "blue", "ok", "white", "active-ink", ): self.assertIn(token, tokens) for foreground, background in token_pairs: ratio = _contrast(tokens[foreground], tokens[background]) self.assertGreaterEqual( ratio, 4.5, f"{theme} {foreground} on {background} contrast is {ratio:.2f}", ) def test_dashboard_has_integrity_console(self): html = (web._STATIC / "dashboard.html").read_text() self.assertIn('data-tab="integrity"', html) self.assertIn('id="integrityTab"', html) self.assertIn('/api/integrity', html) self.assertIn('function renderIntegrity()', html) self.assertIn('Integrity state', html) 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 def test_self_signed_tls_material_is_created(self): with tempfile.TemporaryDirectory() as d: c = _make_cfg(Path(d)) cert, key = web.ensure_tls_cert(c, "127.0.0.1") self.assertTrue(cert.is_file()) self.assertTrue(key.is_file()) 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.assertTrue(Path(self.httpd.tls_cert).is_file()) self.port = self.httpd.server_address[1] self.t = threading.Thread(target=self.httpd.serve_forever, daemon=True) self.t.start() self.ctx = ssl._create_unverified_context() def tearDown(self): self.httpd.shutdown() self.httpd.server_close() self.dir.cleanup() def _get(self, path, token=None): url = f"https://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, context=self.ctx) def _head(self, path, token=None): url = f"https://127.0.0.1:{self.port}{path}" req = urllib.request.Request(url, method="HEAD") if token: req.add_header("Authorization", f"Bearer {token}") return urllib.request.urlopen(req, timeout=4, context=self.ctx) 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_head_dashboard_with_token(self): resp = self._head("/", token="secret-token") self.assertEqual(resp.status, 200) self.assertEqual(resp.headers.get_content_type(), "text/html") self.assertEqual(resp.read(), b"") def test_token_via_query_param(self): resp = self._get("/api/alerts?token=secret-token") self.assertEqual(resp.status, 200) def test_posture_endpoint(self): resp = self._get("/api/posture", token="secret-token") data = json.loads(resp.read()) self.assertIn("findings", data) self.assertIn("count", data) def test_rules_endpoint(self): resp = self._get("/api/rules", token="secret-token") data = json.loads(resp.read()) self.assertIn("rules", data) self.assertGreater(data["count"], 0) def test_integrity_endpoint(self): resp = self._get("/api/integrity", token="secret-token") data = json.loads(resp.read()) self.assertEqual(data["schema"], "enodia.integrity.v1") self.assertIn("watchdog", data) self.assertIn("anchors", data) self.assertTrue(data["read_only"]) if __name__ == "__main__": unittest.main()