luna/enodia-sentinal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
enodia-sentinal/tests/test_web.py
Luna 11d91a40b4 Add dashboard integrity/watchdog console 
Surface integrity and watchdog state in the read-only dashboard via a new
/api/integrity endpoint and integrity_report(): watchdog/heartbeat
verdict, FIM baseline and package-DB anchor freshness, pacman keyring and
SigLevel posture, and Sentinel's own self-integrity footprint. The
endpoint summarizes existing anchors and heartbeats only — it runs no live
FIM or package verification from the request path, keeping the dashboard
read-only.

Add the enodia.integrity.v1 schema (schemas.py + docs/SCHEMAS.md) with a
contract test, a new "Integrity" dashboard tab and summary metric, and
web tests for the report shape, schema contract, endpoint, and console
wiring. Docs updated; completes the v1.0 "dashboard to local console"
roadmap item.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 17:39:53 -07:00

373 lines
14 KiB
Python
Raw Blame History

# 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)
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()
Reference in a new issue View git blame Copy permalink