nearxos
6185b9b85a
Self-hosted, Docker-based agentic troubleshooting platform: FastAPI backend + LangGraph agent, Next.js UI, tiered LLM routing (local Ollama -> Gemini -> DeepSeek -> OpenRouter), MCP server manager, encrypted device credentials, RBAC, audit log, project-memory + Obsidian integrations, and editable troubleshooting decision rules tuned for the GeneseasX vessel stack. Co-authored-by: Cursor <cursoragent@cursor.com>
109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
"""FortiGate MCP server (FortiOS REST API).
|
|
|
|
Read-only diagnostics by default. Token (API key) auth. Self-signed certs are
|
|
accepted (typical for standalone FortiGates). Writes require FORTIGATE_ALLOW_WRITES=true.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
|
|
import httpx
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
mcp = FastMCP("fortigate")
|
|
|
|
ALLOW_WRITES = os.environ.get("FORTIGATE_ALLOW_WRITES", "false").lower() == "true"
|
|
|
|
_active: dict | None = None # {"host","api_key","port"}
|
|
|
|
|
|
def _resolve(host, api_key, port) -> dict:
|
|
if host:
|
|
return {"host": host, "api_key": api_key, "port": port or 443}
|
|
if _active:
|
|
return _active
|
|
raise RuntimeError("No FortiGate target. Call fortigate_connect first.")
|
|
|
|
|
|
def _base(target: dict) -> str:
|
|
return f"https://{target['host']}:{target.get('port', 443)}"
|
|
|
|
|
|
def _get(target: dict, path: str, params: dict | None = None) -> dict:
|
|
url = f"{_base(target)}{path}"
|
|
headers = {"Authorization": f"Bearer {target.get('api_key')}"}
|
|
with httpx.Client(verify=False, timeout=20) as client: # noqa: S501 (self-signed device)
|
|
r = client.get(url, headers=headers, params=params)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
@mcp.tool()
|
|
def fortigate_connect(host: str, api_key: str, port: int = 443, username: str | None = None) -> str:
|
|
"""Set the active FortiGate target (token/API-key auth) and verify connectivity."""
|
|
global _active
|
|
target = {"host": host, "api_key": api_key, "port": port}
|
|
try:
|
|
_get(target, "/api/v2/monitor/system/status")
|
|
except Exception as exc: # noqa: BLE001
|
|
return f"connect failed: {exc}"
|
|
_active = target
|
|
return f"connected to FortiGate {host}"
|
|
|
|
|
|
@mcp.tool()
|
|
def fortigate_test_connection(host: str | None = None, api_key: str | None = None, port: int = 443) -> str:
|
|
"""Test connectivity to a FortiGate."""
|
|
target = _resolve(host, api_key, port)
|
|
try:
|
|
_get(target, "/api/v2/monitor/system/status")
|
|
return "ok"
|
|
except Exception as exc: # noqa: BLE001
|
|
return f"failed: {exc}"
|
|
|
|
|
|
@mcp.tool()
|
|
def fortigate_get_system_status(host: str | None = None, api_key: str | None = None, port: int = 443) -> str:
|
|
"""System status: model, version, serial, uptime."""
|
|
target = _resolve(host, api_key, port)
|
|
return json.dumps(_get(target, "/api/v2/monitor/system/status"), indent=2)[:6000]
|
|
|
|
|
|
@mcp.tool()
|
|
def fortigate_list_interfaces(host: str | None = None, api_key: str | None = None, port: int = 443) -> str:
|
|
"""List interfaces and their link/IP state."""
|
|
target = _resolve(host, api_key, port)
|
|
data = _get(target, "/api/v2/monitor/system/interface")
|
|
return json.dumps(data, indent=2)[:8000]
|
|
|
|
|
|
@mcp.tool()
|
|
def fortigate_list_policies(host: str | None = None, api_key: str | None = None, port: int = 443) -> str:
|
|
"""List firewall policies (cmdb)."""
|
|
target = _resolve(host, api_key, port)
|
|
data = _get(target, "/api/v2/cmdb/firewall/policy")
|
|
return json.dumps(data, indent=2)[:8000]
|
|
|
|
|
|
@mcp.tool()
|
|
def fortigate_get_routing(host: str | None = None, api_key: str | None = None, port: int = 443) -> str:
|
|
"""Active routing table."""
|
|
target = _resolve(host, api_key, port)
|
|
return json.dumps(_get(target, "/api/v2/monitor/router/ipv4"), indent=2)[:8000]
|
|
|
|
|
|
@mcp.tool()
|
|
def fortigate_get_logs(lines: int = 50, host: str | None = None, api_key: str | None = None, port: int = 443) -> str:
|
|
"""Recent event log (best-effort; depends on logging config)."""
|
|
target = _resolve(host, api_key, port)
|
|
try:
|
|
data = _get(target, "/api/v2/log/memory/event", params={"rows": lines})
|
|
return json.dumps(data, indent=2)[:8000]
|
|
except Exception as exc: # noqa: BLE001
|
|
return f"log fetch failed (logging may be disk/forticloud): {exc}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run()
|