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
HTML = """
fgAI Monitor
fgAI Monitor
Live Status
Waiting for monitor data.
AI Assessment
LLM assessment disabled.
Anomalies
Recommendations
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/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)), "")
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 == "/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.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/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()
