from __future__ import annotations import json from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from .config import ConfigStore from .graylog_mcp import GraylogMcpClient from .metrics import prometheus_metrics from .feedback import FeedbackStore HTML = """ SignalScope Monitor

SignalScope Monitor

FortiGate AI/ML Analyzer

Live Status

Waiting for monitor data.

Events and Anomalies

Baseline and Stream Health

AI Assessment

LLM assessment disabled.

Anomalies

Recommendations

Field Baseline Deviations

Related Activity Across Sources

Block Candidates

Threat Intelligence

Policy Findings

Diagnostics

Runtime Configuration

""" def serve_dashboard(host: str, port: int, status_file: str, *, image_dir: str | None = None, config_file: str = "state/fgai-config.json") -> None: status_path = Path(status_file) image_root = Path(image_dir) if image_dir else None config_store = ConfigStore(config_file) class Handler(BaseHTTPRequestHandler): def do_GET(self) -> None: if self.path == "/": self._send(200, "text/html; charset=utf-8", HTML.encode("utf-8")) return if self.path == "/api/config": self._send(200, "application/json", json.dumps(config_store.public()).encode("utf-8")) return if self.path == "/api/feedback": self._send(200, "application/json", json.dumps(FeedbackStore().entries()).encode("utf-8")) return if self.path == "/api/graylog/streams": config = config_store.read() try: client = GraylogMcpClient(str(config.get("graylog_mcp_url", "")), str(config.get("graylog_mcp_token", ""))) client.probe() result = client.call_tool("list_streams", {}) content = result.get("result", {}).get("content", []) text = next((item.get("text", "") for item in content if isinstance(item, dict)), "") if isinstance(result.get("result"), dict) and result["result"].get("isError"): fallback = client.call_tool("list_resource", {"resource_type": "stream"}) resources = fallback.get("result", {}).get("structuredContent", {}).get("resources", []) streams = [ {"id": str(item.get("uri", "")).rsplit(":", 1)[-1], "title": item.get("title", item.get("name", "")), "description": item.get("description", "")} for item in resources if isinstance(item, dict) and item.get("uri") ] else: if not text: raise RuntimeError("Graylog list_streams returned no text content") streams = json.loads(text) body = {"streams": streams, "selected": config.get("graylog_streams", [])} self._send(200, "application/json", json.dumps(body).encode("utf-8")) except Exception as exc: self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8")) return if self.path.startswith("/api/graylog/fields?"): stream_id = self.path.split("stream_id=", 1)[-1].split("&", 1)[0] config = config_store.read() try: client = GraylogMcpClient(str(config.get("graylog_mcp_url", "")), str(config.get("graylog_mcp_token", ""))) client.probe() result = client.call_tool("list_fields", {"streams": [stream_id]}) content = result.get("result", {}).get("content", []) text = next((item.get("text", "") for item in content if isinstance(item, dict)), "") payload = json.loads(text) fields = payload.get("fields", payload) if isinstance(payload, dict) else payload self._send(200, "application/json", json.dumps({"fields": fields}).encode("utf-8")) except Exception as exc: self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8")) return if self.path == "/api/status": if status_path.exists(): body = status_path.read_bytes() else: body = json.dumps({"summary": {}, "anomalies": [], "block_candidates": []}).encode("utf-8") self._send(200, "application/json", body) return if self.path == "/metrics": try: status = json.loads(status_path.read_text(encoding="utf-8")) if status_path.exists() else {} except json.JSONDecodeError: status = {} self._send(200, "text/plain; version=0.0.4; charset=utf-8", prometheus_metrics(status).encode("utf-8")) return if self.path.startswith("/images/") and image_root: image_path = image_root / Path(self.path).name if image_path.exists() and image_path.is_file(): content_type = "image/png" if image_path.suffix.lower() == ".png" else "application/octet-stream" self._send(200, content_type, image_path.read_bytes()) return self._send(404, "text/plain; charset=utf-8", b"not found") def do_POST(self) -> None: if self.path == "/api/feedback" and self._is_loopback_client(): try: payload = json.loads(self.rfile.read(min(int(self.headers.get("Content-Length", "0")), 16_384)).decode("utf-8")) self._send(200, "application/json", json.dumps(FeedbackStore().add(payload)).encode("utf-8")) except (ValueError, json.JSONDecodeError) as exc: self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8")) return if self.path != "/api/config" or not self._is_loopback_client(): self._send(403, "application/json", b'{"error":"configuration is local-only"}') return try: length = int(self.headers.get("Content-Length", "0")) payload = json.loads(self.rfile.read(min(length, 32_768)).decode("utf-8")) if not isinstance(payload, dict): raise ValueError("configuration must be an object") body = json.dumps(config_store.update(payload)).encode("utf-8") except (ValueError, json.JSONDecodeError) as exc: self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8")) return self._send(200, "application/json", body) def _is_loopback_client(self) -> bool: return self.client_address[0] in {"127.0.0.1", "::1"} def log_message(self, format: str, *args: object) -> None: return def _send(self, status: int, content_type: str, body: bytes) -> None: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) server = ThreadingHTTPServer((host, port), Handler) print(f"Dashboard listening on http://{host}:{port}") print(f"Reading status from {status_path}") server.serve_forever()