diff --git a/src/fgai/dashboard.py b/src/fgai/dashboard.py index 0bbb4a4..01b21f1 100644 --- a/src/fgai/dashboard.py +++ b/src/fgai/dashboard.py @@ -85,6 +85,7 @@ async function refresh() { const a = data.anomaly_summary || {}; const baseline = data.baseline || {}; const threat = (data.capabilities || {}).threat_intel || {}; + const mcp = (data.capabilities || {}).graylog_mcp || {}; const configuration = data.configuration || {}; document.getElementById('stamp').textContent = data.generated_at ? `Updated ${new Date(data.generated_at * 1000).toLocaleString()}` : 'Waiting for monitor data'; document.getElementById('metrics').innerHTML = [ @@ -98,7 +99,7 @@ async function refresh() { capability('Baseline', baseline.enabled ? 'on' : 'warn', baseline.enabled ? `${baseline.sources_ready || 0} sources ready` : 'disabled'), capability('Ollama', llm.enabled && llm.status !== 'error' ? 'on' : 'warn', llm.enabled ? (llm.status || 'starting') : 'disabled'), capability('Threat Intel', threat.enabled && threat.configured ? 'on' : 'warn', threat.enabled ? `${threat.provider || 'unknown'}${threat.configured ? '' : ', key missing'}` : 'disabled'), - capability('Log source', configuration.log_source === 'graylog_mcp' ? 'warn' : 'on', configuration.log_source === 'graylog_mcp' ? 'Graylog MCP configured, connector pending' : 'local syslog') + capability('Graylog MCP', mcp.status === 'connected' ? 'on' : 'warn', configuration.log_source === 'graylog_mcp' ? (mcp.status || 'checking') : 'not selected') ].join(''); document.getElementById('liveStatus').innerHTML = [ `Log file: ${esc(data.log_path || '')}`, diff --git a/src/fgai/graylog_mcp.py b/src/fgai/graylog_mcp.py new file mode 100644 index 0000000..d5b37ec --- /dev/null +++ b/src/fgai/graylog_mcp.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import base64 +import json +from urllib import error, request + + +class GraylogMcpClient: + """Small Streamable HTTP MCP client used for Graylog connection checks.""" + + def __init__(self, url: str, token: str, *, timeout: int = 15) -> None: + self.url = url.rstrip("/") + self.token = token + self.timeout = timeout + self.session_id: str | None = None + + def _call(self, method: str, params: dict[str, object] | None = None, *, notification: bool = False) -> dict[str, object]: + payload: dict[str, object] = {"jsonrpc": "2.0", "method": method} + if not notification: + payload["id"] = 1 + if params is not None: + payload["params"] = params + credentials = base64.b64encode(f"{self.token}:token".encode("utf-8")).decode("ascii") + headers = { + "Authorization": f"Basic {credentials}", + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + } + if self.session_id: + headers["Mcp-Session-Id"] = self.session_id + req = request.Request(self.url, data=json.dumps(payload).encode("utf-8"), method="POST", headers=headers) + try: + with request.urlopen(req, timeout=self.timeout) as response: + self.session_id = response.headers.get("Mcp-Session-Id", self.session_id) + body = response.read().decode("utf-8") + except error.HTTPError as exc: + raise RuntimeError(f"http_{exc.code}") from exc + except Exception as exc: + raise RuntimeError(f"connection_error: {exc}") from exc + if notification: + return {} + if body.startswith("data:"): + body = next((line[5:].strip() for line in body.splitlines() if line.startswith("data:")), "") + try: + response_data = json.loads(body) + except json.JSONDecodeError as exc: + raise RuntimeError("invalid_mcp_response") from exc + if "error" in response_data: + raise RuntimeError(f"mcp_error: {response_data['error']}") + return response_data + + def probe(self) -> dict[str, object]: + initialized = self._call( + "initialize", + {"protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": {"name": "fgAI", "version": "0.1"}}, + ) + self._call("notifications/initialized", notification=True) + tools = self._call("tools/list") + tool_list = tools.get("result", {}).get("tools", []) if isinstance(tools.get("result"), dict) else [] + names = [str(item.get("name")) for item in tool_list if isinstance(item, dict) and item.get("name")] + version = initialized.get("result", {}).get("serverInfo", {}).get("version", "") if isinstance(initialized.get("result"), dict) else "" + return {"status": "connected", "tools": names, "server_version": version} diff --git a/src/fgai/monitor.py b/src/fgai/monitor.py index dfea679..66a7be5 100644 --- a/src/fgai/monitor.py +++ b/src/fgai/monitor.py @@ -7,6 +7,7 @@ from pathlib import Path from .anomaly import anomaly_summary, detect_source_anomalies from .baseline import BaselineStore from .config import ConfigStore +from .graylog_mcp import GraylogMcpClient from .llm import ollama_dashboard_assessment from .logs import local_in_failures, read_events, summarize_events, top_field_values from .mitigation import parse_allowlist, suggest_block_candidates @@ -34,6 +35,17 @@ def build_status( config_exists = bool(config_store and config_store.path.exists()) runtime_values = config_store.read() if config_exists and config_store else {} runtime_config = config_store.public() if config_store else {} + mcp_status: dict[str, object] = {"status": "not_configured"} + if runtime_values.get("log_source") == "graylog_mcp": + url = str(runtime_values.get("graylog_mcp_url", "")) + token = str(runtime_values.get("graylog_mcp_token", "")) + if not url or not token: + mcp_status = {"status": "missing_configuration"} + else: + try: + mcp_status = GraylogMcpClient(url, token).probe() + except RuntimeError as exc: + mcp_status = {"status": "error", "error": str(exc)} intel_ips = sorted( { ip @@ -68,7 +80,7 @@ def build_status( "summary": summarize_events(events), "anomaly_summary": anomaly_summary(anomalies), "baseline": {"enabled": bool(baseline), "sources_ready": len(profiles), "new_events_recorded": baseline_events}, - "capabilities": {"threat_intel": threat_intel_status}, + "capabilities": {"threat_intel": threat_intel_status, "graylog_mcp": mcp_status}, "configuration": runtime_config, "diagnostics": { "top_source_ips": top_field_values(events, "srcip", limit=10), diff --git a/tests/test_graylog_mcp.py b/tests/test_graylog_mcp.py new file mode 100644 index 0000000..caeb949 --- /dev/null +++ b/tests/test_graylog_mcp.py @@ -0,0 +1,37 @@ +import json +import unittest +from unittest.mock import patch + +from fgai.graylog_mcp import GraylogMcpClient + + +class _Response: + def __init__(self, body: dict[str, object], session: str = "session-1") -> None: + self.body = body + self.headers = {"Mcp-Session-Id": session} + + def read(self) -> bytes: + return json.dumps(self.body).encode("utf-8") + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + +class GraylogMcpTests(unittest.TestCase): + def test_probe_initializes_and_lists_tools(self): + responses = iter([ + _Response({"result": {"serverInfo": {"version": "7.1.6"}}}), + _Response({}), + _Response({"result": {"tools": [{"name": "search_messages"}, {"name": "aggregate_messages"}]}}), + ]) + with patch("fgai.graylog_mcp.request.urlopen", side_effect=responses): + status = GraylogMcpClient("http://graylog/api/mcp", "raw-token").probe() + self.assertEqual(status["status"], "connected") + self.assertIn("search_messages", status["tools"]) + + +if __name__ == "__main__": + unittest.main()