luna/enodia-sentinal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add dashboard theme settings and contrast tests

Browse source
This commit is contained in:
Luna 2026-06-16 02:29:45 -07:00
parent 7f4d5b42fd
commit bfef23fc4a
7 changed files with 424 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

60
enodia_sentinel/web.py
View file

@ -316,63 +316,71 @@ class _Handler(BaseHTTPRequestHandler):
supplied = parse_qs(urlparse(self.path).query).get("token", [""])[0]
return hmac.compare_digest(supplied, token)
def _json(self, obj, status=200):
body = json.dumps(obj).encode("utf-8")
def _send_body(self, body: bytes, ctype: str, status=200, send_body=True):
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
if send_body:
self.wfile.write(body)
def _json(self, obj, status=200, send_body=True):
body = json.dumps(obj).encode("utf-8")
self._send_body(body, "application/json", status, send_body)
def _json_or_not_found(self, obj, send_body=True):
self._json(obj if obj else {"error": "not found"},
200 if obj else 404, send_body=send_body)
def do_GET(self):
self._handle_get(send_body=True)
def do_HEAD(self):
self._handle_get(send_body=False)
def _handle_get(self, send_body=True):
path = urlparse(self.path).path
if not self._authorized():
self._json({"error": "unauthorized"}, 401)
self._json({"error": "unauthorized"}, 401, send_body=send_body)
return
cfg: Config = self.server.cfg # type: ignore[attr-defined]
if path in ("/", "/index.html", "/dashboard.html"):
self._serve_static("dashboard.html", "text/html; charset=utf-8")
self._serve_static("dashboard.html", "text/html; charset=utf-8",
send_body=send_body)
elif path == "/api/status":
self._json(daemon_status(cfg))
self._json(daemon_status(cfg), send_body=send_body)
elif path == "/api/alerts":
self._json(list_alerts(cfg))
self._json(list_alerts(cfg), send_body=send_body)
elif path.startswith("/api/alerts/"):
alert = get_alert(cfg, path.rsplit("/", 1)[-1])
self._json(alert if alert else {"error": "not found"},
200 if alert else 404)
self._json_or_not_found(alert, send_body=send_body)
elif path == "/api/events":
self._json({"events": tail_events(cfg)})
self._json({"events": tail_events(cfg)}, send_body=send_body)
elif path == "/api/posture":
self._json(posture_report(cfg))
self._json(posture_report(cfg), send_body=send_body)
elif path == "/api/rules":
self._json(rules_catalog(cfg))
self._json(rules_catalog(cfg), send_body=send_body)
elif path == "/api/incidents":
self._json(list_incidents(cfg))
self._json(list_incidents(cfg), send_body=send_body)
elif path.startswith("/api/incidents/"):
iid = unquote(path.rsplit("/", 1)[-1])
inc = get_incident(cfg, iid)
self._json(inc if inc else {"error": "not found"},
200 if inc else 404)
self._json_or_not_found(inc, send_body=send_body)
elif path.startswith("/api/respond/plan/"):
iid = unquote(path.rsplit("/", 1)[-1])
plan = response_plan(cfg, iid)
self._json(plan if plan else {"error": "not found"},
200 if plan else 404)
self._json_or_not_found(plan, send_body=send_body)
else:
self._json({"error": "not found"}, 404)
self._json({"error": "not found"}, 404, send_body=send_body)
def _serve_static(self, name, ctype):
def _serve_static(self, name, ctype, send_body=True):
try:
body = (_STATIC / name).read_bytes()
except OSError:
self._json({"error": "missing asset"}, 500)
self._json({"error": "missing asset"}, 500, send_body=send_body)
return
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
self._send_body(body, ctype, send_body=send_body)
def build_server(cfg: Config) -> tuple[ThreadingHTTPServer, str, str]: