376 lines
14 KiB
Python
376 lines
14 KiB
Python
# 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<body>.*?)\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()
|