From 74edd09eb842fa18ad151e496985bf4f2268de9b Mon Sep 17 00:00:00 2001 From: Hermes CT107 Date: Fri, 12 Jun 2026 17:59:34 +0000 Subject: [PATCH] Initial import from Proxmox host --- .env.example | 15 ++ .gitignore | 7 + README.md | 150 ++++++++++++++++ pyproject.toml | 25 +++ src/opnsense_mcp/__init__.py | 3 + src/opnsense_mcp/__main__.py | 4 + src/opnsense_mcp/audit.py | 23 +++ src/opnsense_mcp/client.py | 177 +++++++++++++++++++ src/opnsense_mcp/config.py | 25 +++ src/opnsense_mcp/guards.py | 77 +++++++++ src/opnsense_mcp/server.py | 83 +++++++++ src/opnsense_mcp/tools/__init__.py | 25 +++ src/opnsense_mcp/tools/aliases.py | 148 ++++++++++++++++ src/opnsense_mcp/tools/dhcp.py | 58 +++++++ src/opnsense_mcp/tools/diagnostics.py | 147 ++++++++++++++++ src/opnsense_mcp/tools/firewall.py | 238 ++++++++++++++++++++++++++ src/opnsense_mcp/tools/interfaces.py | 61 +++++++ src/opnsense_mcp/tools/nat.py | 173 +++++++++++++++++++ src/opnsense_mcp/tools/routes.py | 87 ++++++++++ src/opnsense_mcp/tools/system.py | 69 ++++++++ 20 files changed, 1595 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 README.md create mode 100644 pyproject.toml create mode 100644 src/opnsense_mcp/__init__.py create mode 100644 src/opnsense_mcp/__main__.py create mode 100644 src/opnsense_mcp/audit.py create mode 100644 src/opnsense_mcp/client.py create mode 100644 src/opnsense_mcp/config.py create mode 100644 src/opnsense_mcp/guards.py create mode 100644 src/opnsense_mcp/server.py create mode 100644 src/opnsense_mcp/tools/__init__.py create mode 100644 src/opnsense_mcp/tools/aliases.py create mode 100644 src/opnsense_mcp/tools/dhcp.py create mode 100644 src/opnsense_mcp/tools/diagnostics.py create mode 100644 src/opnsense_mcp/tools/firewall.py create mode 100644 src/opnsense_mcp/tools/interfaces.py create mode 100644 src/opnsense_mcp/tools/nat.py create mode 100644 src/opnsense_mcp/tools/routes.py create mode 100644 src/opnsense_mcp/tools/system.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..7c4e751 --- /dev/null +++ b/.env.example @@ -0,0 +1,15 @@ +# Copy to .env or set in ~/.hermes/config.yaml under mcp_servers.opnsense.env +# Create keys in OPNsense: System → Access → Users → API keys + +OPNSENSE_URL=https://10.77.30.1:40443 +OPNSENSE_API_KEY=REPLACE_WITH_YOUR_API_KEY +OPNSENSE_API_SECRET=REPLACE_WITH_YOUR_API_SECRET +OPNSENSE_VERIFY_SSL=false +OPNSENSE_TIMEOUT=60 + +# Safety — enable writes only when ready +OPNSENSE_ALLOW_WRITES=false +OPNSENSE_DRY_RUN=false +OPNSENSE_REQUIRE_DESCRIPTION_PREFIX=mcp: +OPNSENSE_BLOCK_WAN_INBOUND_ANY=true +OPNSENSE_AUDIT_LOG_PATH=/root/.hermes/logs/opnsense-mcp-audit.jsonl diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ed9fb92 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +.env +*.pyc +__pycache__/ +*.egg-info/ +dist/ +build/ +.venv/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..78ba1ab --- /dev/null +++ b/README.md @@ -0,0 +1,150 @@ +# OPNsense MCP + +Model Context Protocol server for [OPNsense](https://opnsense.org/) firewall management. Designed for use with [Hermes Agent](https://github.com/NousResearch/hermes-agent) and any MCP-compatible client. + +## Features + +- **60+ tools** covering system info, interfaces, firewall rules, aliases, NAT, DHCP, diagnostics, routes, and gateways +- **Safety guardrails**: write operations disabled by default, dry-run mode, WAN rule blocking, required description prefix +- **Savepoint workflow**: create savepoint → apply → cancel rollback (matches OPNsense official API pattern) +- **Audit logging**: all mutations logged to JSONL + +## Quick Start + +### 1. Create an OPNsense API key + +In OPNsense: **System → Access → Users** → edit user → add API key. Download the key/secret pair. + +Grant only the privileges needed. For read-only use, limit to read privileges. For writes, include firewall, aliases, routes, etc. + +### 2. Install + +```bash +pip install -e /path/to/opnsense-mcp +``` + +Or with the Hermes venv: + +```bash +/usr/local/lib/hermes-agent/venv/bin/pip install -e /root/opnsense-mcp +``` + +### 3. Configure environment + +```bash +cp .env.example .env +# Edit .env with your API key, secret, and firewall URL +``` + +| Variable | Default | Description | +|----------|---------|-------------| +| `OPNSENSE_URL` | `https://10.77.30.1:40443` | Firewall base URL | +| `OPNSENSE_API_KEY` | — | API key (required) | +| `OPNSENSE_API_SECRET` | — | API secret (required) | +| `OPNSENSE_VERIFY_SSL` | `false` | Verify TLS certificate | +| `OPNSENSE_ALLOW_WRITES` | `false` | Enable mutating operations | +| `OPNSENSE_DRY_RUN` | `false` | Log writes without executing | +| `OPNSENSE_REQUIRE_DESCRIPTION_PREFIX` | `mcp:` | Prefix for new firewall rules | +| `OPNSENSE_BLOCK_WAN_INBOUND_ANY` | `true` | Block WAN allow-all rules | + +### 4. Register with Hermes + +```bash +hermes mcp add opnsense \ + --command /usr/local/lib/hermes-agent/venv/bin/opnsense-mcp \ + --env OPNSENSE_URL=https://10.77.30.1:40443 \ + --env OPNSENSE_API_KEY=your_key \ + --env OPNSENSE_API_SECRET=your_secret \ + --env OPNSENSE_VERIFY_SSL=false \ + --env OPNSENSE_ALLOW_WRITES=false +``` + +Or add to `~/.hermes/config.yaml`: + +```yaml +mcp_servers: + opnsense: + command: "/usr/local/lib/hermes-agent/venv/bin/opnsense-mcp" + env: + OPNSENSE_URL: "https://10.77.30.1:40443" + OPNSENSE_API_KEY: "your_key" + OPNSENSE_API_SECRET: "your_secret" + OPNSENSE_VERIFY_SSL: "false" + OPNSENSE_ALLOW_WRITES: "false" + timeout: 120 +``` + +Restart the Hermes gateway after adding. + +### 5. Test + +```bash +hermes mcp test opnsense +``` + +Or ask Hermes: *"Use opnsense_test_connection to verify firewall access"* + +## Tool Categories + +### Read-only (always available with valid API key) + +| Tool | Description | +|------|-------------| +| `opnsense_test_connection` | Verify API auth and connectivity | +| `opnsense_get_system_info` | Version, uptime, health | +| `opnsense_list_interfaces` | Interface status and IPs | +| `opnsense_search_firewall_rules` | Search filter rules | +| `opnsense_search_aliases` | Search aliases | +| `opnsense_search_dhcp_leases` | DHCP lease table | +| `opnsense_get_arp_table` | ARP / IP→MAC | +| `opnsense_get_gateway_status` | WAN gateway health | +| `opnsense_get_firewall_log` | Recent blocked/ passed traffic | +| `opnsense_get_pf_states` | Active connection states | +| `opnsense_search_port_forwards` | Port forward rules | + +### Write operations (require `OPNSENSE_ALLOW_WRITES=true`) + +| Tool | Description | +|------|-------------| +| `opnsense_add_firewall_rule` | Add filter rule (prefix required) | +| `opnsense_toggle_firewall_rule` | Enable/disable rule | +| `opnsense_create_firewall_savepoint` | Snapshot before changes | +| `opnsense_apply_firewall` | Apply pending changes | +| `opnsense_cancel_firewall_rollback` | Confirm changes (cancel 60s rollback) | +| `opnsense_safe_apply_firewall_rule_change` | Savepoint + toggle + apply workflow | +| `opnsense_alias_add_entry` | Add IP to alias | +| `opnsense_add_d_nat_rule` | Add port forward | + +## Safe Change Workflow + +```text +1. opnsense_create_firewall_savepoint → returns revision +2. opnsense_toggle_firewall_rule(uuid) → make change +3. opnsense_apply_firewall(revision) → apply with 60s auto-rollback +4. Verify connectivity +5. opnsense_cancel_firewall_rollback(revision) → keep change +``` + +Or use `opnsense_safe_apply_firewall_rule_change` which combines steps 1–3. + +## Standalone Usage + +Run directly as an MCP stdio server: + +```bash +OPNSENSE_API_KEY=... OPNSENSE_API_SECRET=... opnsense-mcp +``` + +Compatible with Cursor, Claude Desktop, and other MCP clients. + +## Security Notes + +- Never commit API keys. Use environment variables or Hermes secrets. +- Start with read-only (`ALLOW_WRITES=false`) until you trust the agent workflows. +- Use a dedicated OPNsense user with minimal privileges. +- Review audit log at `OPNSENSE_AUDIT_LOG_PATH`. +- Enable Hermes `approvals: mode: manual` for production use. + +## License + +MIT diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..137ddc4 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,25 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "opnsense-mcp" +version = "1.0.0" +description = "Model Context Protocol server for OPNsense firewall management" +readme = "README.md" +requires-python = ">=3.10" +license = "MIT" +authors = [{ name = "OPNsense MCP" }] +keywords = ["mcp", "opnsense", "firewall", "hermes"] +dependencies = [ + "mcp>=1.0.0", + "httpx>=0.27.0", + "pydantic>=2.0.0", + "pydantic-settings>=2.0.0", +] + +[project.scripts] +opnsense-mcp = "opnsense_mcp.server:main" + +[tool.hatch.build.targets.wheel] +packages = ["src/opnsense_mcp"] diff --git a/src/opnsense_mcp/__init__.py b/src/opnsense_mcp/__init__.py new file mode 100644 index 0000000..c9a437c --- /dev/null +++ b/src/opnsense_mcp/__init__.py @@ -0,0 +1,3 @@ +"""OPNsense MCP server — Model Context Protocol integration for OPNsense firewalls.""" + +__version__ = "1.0.0" diff --git a/src/opnsense_mcp/__main__.py b/src/opnsense_mcp/__main__.py new file mode 100644 index 0000000..6a06c97 --- /dev/null +++ b/src/opnsense_mcp/__main__.py @@ -0,0 +1,4 @@ +from opnsense_mcp.server import main + +if __name__ == "__main__": + main() diff --git a/src/opnsense_mcp/audit.py b/src/opnsense_mcp/audit.py new file mode 100644 index 0000000..b9efeb5 --- /dev/null +++ b/src/opnsense_mcp/audit.py @@ -0,0 +1,23 @@ +"""Append-only audit log for mutating operations.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from opnsense_mcp.config import settings + + +def audit_log(action: str, details: dict[str, Any], *, dry_run: bool = False) -> None: + entry = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "action": action, + "dry_run": dry_run, + **details, + } + path = Path(settings.audit_log_path) + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(entry, default=str) + "\n") diff --git a/src/opnsense_mcp/client.py b/src/opnsense_mcp/client.py new file mode 100644 index 0000000..db57ec2 --- /dev/null +++ b/src/opnsense_mcp/client.py @@ -0,0 +1,177 @@ +"""HTTP client for the OPNsense REST API.""" + +from __future__ import annotations + +import json +from typing import Any +from urllib.parse import urlencode + +import httpx + +from opnsense_mcp.config import settings +from opnsense_mcp.guards import GuardError, require_credentials + + +class OpnsenseAPIError(Exception): + """Raised when the OPNsense API returns an error.""" + + def __init__(self, message: str, status_code: int | None = None, body: Any = None): + super().__init__(message) + self.status_code = status_code + self.body = body + + +class OpnsenseClient: + """Thin wrapper around OPNsense /api/{module}/{controller}/{command} endpoints.""" + + def __init__( + self, + url: str | None = None, + api_key: str | None = None, + api_secret: str | None = None, + *, + verify_ssl: bool | None = None, + timeout: float | None = None, + ): + self.base_url = (url or settings.url).rstrip("/") + self.api_key = api_key if api_key is not None else settings.api_key + self.api_secret = api_secret if api_secret is not None else settings.api_secret + self.verify_ssl = verify_ssl if verify_ssl is not None else settings.verify_ssl + self.timeout = timeout if timeout is not None else settings.timeout + + def _build_url( + self, + module: str, + controller: str, + command: str, + *path_parts: str | int, + params: dict[str, Any] | None = None, + ) -> str: + segments = [self.base_url, "api", module, controller, command] + for part in path_parts: + if part is not None and str(part) != "": + segments.append(str(part)) + url = "/".join(segments) + if params: + clean = {k: v for k, v in params.items() if v is not None} + if clean: + url = f"{url}?{urlencode(clean)}" + return url + + def _request( + self, + method: str, + module: str, + controller: str, + command: str, + *path_parts: str | int, + params: dict[str, Any] | None = None, + data: dict[str, Any] | None = None, + ) -> Any: + require_credentials() + url = self._build_url(module, controller, command, *path_parts, params=params) + with httpx.Client(verify=self.verify_ssl, timeout=self.timeout) as client: + response = client.request( + method, + url, + auth=(self.api_key, self.api_secret), + json=data, + ) + if response.status_code >= 400: + try: + body = response.json() + except json.JSONDecodeError: + body = response.text + raise OpnsenseAPIError( + f"OPNsense API error {response.status_code} for {method} {url}", + status_code=response.status_code, + body=body, + ) + if not response.content: + return {} + try: + return response.json() + except json.JSONDecodeError: + return {"raw": response.text} + + def get( + self, + module: str, + controller: str, + command: str, + *path_parts: str | int, + params: dict[str, Any] | None = None, + ) -> Any: + return self._request("GET", module, controller, command, *path_parts, params=params) + + def post( + self, + module: str, + controller: str, + command: str, + *path_parts: str | int, + params: dict[str, Any] | None = None, + data: dict[str, Any] | None = None, + ) -> Any: + return self._request( + "POST", module, controller, command, *path_parts, params=params, data=data + ) + + # ── Search helpers ──────────────────────────────────────────────────── + + @staticmethod + def search_body( + search_phrase: str = "", + current: int = 1, + row_count: int = 100, + **extra: Any, + ) -> dict[str, Any]: + body: dict[str, Any] = { + "current": current, + "rowCount": row_count, + "sort": {}, + "searchPhrase": search_phrase, + } + body.update(extra) + return body + + def search_grid( + self, + module: str, + controller: str, + command: str, + search_phrase: str = "", + row_count: int = 100, + **extra: Any, + ) -> Any: + """POST to a search* endpoint returning {rows, total, ...}.""" + return self.post( + module, + controller, + command, + data=self.search_body(search_phrase, row_count=row_count, **extra), + ) + + def search_get( + self, + module: str, + controller: str, + command: str, + search_phrase: str = "", + row_count: int = 100, + **extra: Any, + ) -> Any: + """GET variant used by some firewall search endpoints.""" + params = {"current": 1, "rowCount": row_count, "searchPhrase": search_phrase} + params.update(extra) + return self.get(module, controller, command, params=params) + + +_client: OpnsenseClient | None = None + + +def get_client() -> OpnsenseClient: + global _client + if _client is None: + _client = OpnsenseClient() + return _client diff --git a/src/opnsense_mcp/config.py b/src/opnsense_mcp/config.py new file mode 100644 index 0000000..e094815 --- /dev/null +++ b/src/opnsense_mcp/config.py @@ -0,0 +1,25 @@ +"""Configuration loaded from environment variables.""" + +from __future__ import annotations + +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_prefix="OPNSENSE_", extra="ignore") + + url: str = "https://10.77.30.1:40443" + api_key: str = "" + api_secret: str = "" + verify_ssl: bool = False + timeout: float = 60.0 + + # Safety controls + allow_writes: bool = False + dry_run: bool = False + require_description_prefix: str = "mcp:" + block_wan_inbound_any: bool = True + audit_log_path: str = "/root/.hermes/logs/opnsense-mcp-audit.jsonl" + + +settings = Settings() diff --git a/src/opnsense_mcp/guards.py b/src/opnsense_mcp/guards.py new file mode 100644 index 0000000..8156e9b --- /dev/null +++ b/src/opnsense_mcp/guards.py @@ -0,0 +1,77 @@ +"""Safety guardrails for mutating firewall operations.""" + +from __future__ import annotations + +from typing import Any + +from opnsense_mcp.audit import audit_log +from opnsense_mcp.config import settings + + +class GuardError(Exception): + """Raised when a safety guard blocks an operation.""" + + +def require_credentials() -> None: + if not settings.api_key or not settings.api_secret: + raise GuardError( + "OPNSENSE_API_KEY and OPNSENSE_API_SECRET must be set. " + "Create an API key in OPNsense: System → Access → Users." + ) + + +def require_writes(action: str, payload: dict[str, Any] | None = None) -> None: + require_credentials() + if settings.dry_run: + audit_log(action, {"payload": payload or {}, "blocked": "dry_run"}, dry_run=True) + return + if not settings.allow_writes: + raise GuardError( + f"Write operation '{action}' blocked. Set OPNSENSE_ALLOW_WRITES=true to enable." + ) + + +def is_dry_run_response(action: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: + audit_log(action, {"payload": payload or {}}, dry_run=True) + return { + "dry_run": True, + "action": action, + "payload": payload, + "message": "Dry run — no changes applied. Set OPNSENSE_DRY_RUN=false to execute.", + } + + +def validate_firewall_rule(rule: dict[str, Any], *, is_new: bool = False) -> None: + """Reject dangerous or mis-tagged firewall rules.""" + description = str(rule.get("description", "")) + if is_new and settings.require_description_prefix: + prefix = settings.require_description_prefix + if prefix and not description.lower().startswith(prefix.lower()): + raise GuardError( + f"New rules must have description starting with '{prefix}'. " + f"Got: '{description}'" + ) + + if not settings.block_wan_inbound_any: + return + + interface = str(rule.get("interface", rule.get("interface_name", ""))).lower() + action = str(rule.get("action", "pass")).lower() + source = str(rule.get("source_net", rule.get("source", "any"))).lower() + dest = str(rule.get("destination_net", rule.get("destination", "any"))).lower() + + wan_markers = ("wan", "opt", "pppoe", "dhcp") + is_wan = any(m in interface for m in wan_markers) or interface in ("", "any") + + if action in ("pass", "allow") and is_wan: + if source in ("any", "", "0.0.0.0/0", "::/0") and dest not in ("any", "", "this firewall"): + raise GuardError( + "Blocked: inbound WAN allow rule with source 'any' is not permitted. " + "Use a specific source CIDR or disable OPNSENSE_BLOCK_WAN_INBOUND_ANY." + ) + + +def validate_alias_mutation(name: str) -> None: + protected = {"bogons", "bogonsv6", "virusprot", "sshlockout", "__wan_network"} + if name.lower() in protected: + raise GuardError(f"Alias '{name}' is protected and cannot be modified via MCP.") diff --git a/src/opnsense_mcp/server.py b/src/opnsense_mcp/server.py new file mode 100644 index 0000000..59e6fea --- /dev/null +++ b/src/opnsense_mcp/server.py @@ -0,0 +1,83 @@ +"""OPNsense MCP server entrypoint.""" + +from __future__ import annotations + +import json + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp import __version__ +from opnsense_mcp.config import settings +from opnsense_mcp.tools import register_all + +mcp = FastMCP( + "opnsense", + instructions=( + "OPNsense firewall management MCP. Use read tools freely for diagnostics. " + "Write tools require OPNSENSE_ALLOW_WRITES=true. New firewall rules must use " + f"description prefix '{settings.require_description_prefix}'. " + "For risky changes, use savepoint + apply + cancel_rollback workflow." + ), +) + + +@mcp.tool() +def opnsense_get_config() -> str: + """Get current MCP server configuration (no secrets exposed).""" + return json.dumps( + { + "url": settings.url, + "verify_ssl": settings.verify_ssl, + "allow_writes": settings.allow_writes, + "dry_run": settings.dry_run, + "require_description_prefix": settings.require_description_prefix, + "block_wan_inbound_any": settings.block_wan_inbound_any, + "api_key_configured": bool(settings.api_key), + "version": __version__, + }, + indent=2, + ) + + +@mcp.tool() +def opnsense_test_connection() -> str: + """Test connectivity and authentication to the OPNsense API.""" + from opnsense_mcp.client import OpnsenseAPIError, get_client + + client = get_client() + try: + firmware = client.get("core", "firmware", "status") + gateways = client.get("routes", "gateway", "status") + return json.dumps( + { + "status": "ok", + "url": settings.url, + "firmware": firmware, + "gateways": gateways, + }, + indent=2, + default=str, + ) + except OpnsenseAPIError as exc: + return json.dumps( + { + "status": "error", + "url": settings.url, + "message": str(exc), + "status_code": exc.status_code, + "body": exc.body, + }, + indent=2, + default=str, + ) + + +register_all(mcp) + + +def main() -> None: + mcp.run() + + +if __name__ == "__main__": + main() diff --git a/src/opnsense_mcp/tools/__init__.py b/src/opnsense_mcp/tools/__init__.py new file mode 100644 index 0000000..91e4e44 --- /dev/null +++ b/src/opnsense_mcp/tools/__init__.py @@ -0,0 +1,25 @@ +"""Tool package registration.""" + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp.tools import ( + aliases, + dhcp, + diagnostics, + firewall, + interfaces, + nat, + routes, + system, +) + + +def register_all(mcp: FastMCP) -> None: + system.register(mcp) + interfaces.register(mcp) + firewall.register(mcp) + aliases.register(mcp) + nat.register(mcp) + dhcp.register(mcp) + diagnostics.register(mcp) + routes.register(mcp) diff --git a/src/opnsense_mcp/tools/aliases.py b/src/opnsense_mcp/tools/aliases.py new file mode 100644 index 0000000..088a311 --- /dev/null +++ b/src/opnsense_mcp/tools/aliases.py @@ -0,0 +1,148 @@ +"""Firewall alias tools.""" + +from __future__ import annotations + +import json +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp.audit import audit_log +from opnsense_mcp.client import get_client +from opnsense_mcp.config import settings +from opnsense_mcp.guards import is_dry_run_response, require_writes, validate_alias_mutation + + +def register(mcp: FastMCP) -> None: + @mcp.tool() + def opnsense_search_aliases(search_phrase: str = "", row_count: int = 100) -> str: + """Search firewall aliases (host, network, port, URL, etc.).""" + client = get_client() + result = client.search_grid("firewall", "alias", "searchItem", search_phrase, row_count) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_alias(uuid: str = "", name: str = "") -> str: + """Get alias details by UUID or by name.""" + client = get_client() + if name: + result = client.get("firewall", "alias", "getAliasUUID", name) + return json.dumps(result, indent=2, default=str) + result = client.get("firewall", "alias", "getItem", uuid if uuid else None) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_list_network_aliases() -> str: + """List all network-type aliases.""" + client = get_client() + result = client.get("firewall", "alias", "listNetworkAliases") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_add_alias(alias: dict[str, Any]) -> str: + """Add a firewall alias. Fields: name, type (host/network/port/url/etc), content, description, enabled.""" + name = str(alias.get("name", "")) + validate_alias_mutation(name) + payload = {"alias": alias} + if settings.dry_run: + return json.dumps(is_dry_run_response("add_alias", payload), indent=2) + require_writes("add_alias", payload) + client = get_client() + result = client.post("firewall", "alias", "addItem", data=payload) + audit_log("add_alias", {"alias": alias, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_update_alias(uuid: str, alias: dict[str, Any]) -> str: + """Update an existing alias by UUID.""" + validate_alias_mutation(str(alias.get("name", ""))) + payload = {"alias": alias} + if settings.dry_run: + return json.dumps(is_dry_run_response("update_alias", {"uuid": uuid, **payload}), indent=2) + require_writes("update_alias", {"uuid": uuid, **payload}) + client = get_client() + result = client.post("firewall", "alias", "setItem", uuid, data=payload) + audit_log("update_alias", {"uuid": uuid, "alias": alias, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_delete_alias(uuid: str) -> str: + """Delete an alias by UUID.""" + payload = {"uuid": uuid} + if settings.dry_run: + return json.dumps(is_dry_run_response("delete_alias", payload), indent=2) + require_writes("delete_alias", payload) + client = get_client() + result = client.post("firewall", "alias", "delItem", uuid) + audit_log("delete_alias", {"uuid": uuid, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_toggle_alias(uuid: str, enabled: bool = True) -> str: + """Enable or disable an alias.""" + payload = {"uuid": uuid, "enabled": enabled} + if settings.dry_run: + return json.dumps(is_dry_run_response("toggle_alias", payload), indent=2) + require_writes("toggle_alias", payload) + client = get_client() + flag = "1" if enabled else "0" + result = client.post("firewall", "alias", "toggleItem", uuid, flag) + audit_log("toggle_alias", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_alias_add_entry(alias_name: str, entry: str) -> str: + """Add a host/IP/network entry to an existing alias (alias_util/add).""" + validate_alias_mutation(alias_name) + payload = {"alias": alias_name, "entry": entry} + if settings.dry_run: + return json.dumps(is_dry_run_response("alias_add_entry", payload), indent=2) + require_writes("alias_add_entry", payload) + client = get_client() + result = client.post("firewall", "alias_util", "add", alias_name, data={"address": entry}) + audit_log("alias_add_entry", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_alias_remove_entry(alias_name: str, entry: str) -> str: + """Remove an entry from an alias (alias_util/delete).""" + validate_alias_mutation(alias_name) + payload = {"alias": alias_name, "entry": entry} + if settings.dry_run: + return json.dumps(is_dry_run_response("alias_remove_entry", payload), indent=2) + require_writes("alias_remove_entry", payload) + client = get_client() + result = client.post("firewall", "alias_util", "delete", alias_name, data={"address": entry}) + audit_log("alias_remove_entry", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_alias_list_entries(alias_name: str) -> str: + """List all entries/contents of a specific alias.""" + client = get_client() + result = client.get("firewall", "alias_util", "list", alias_name) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_alias_flush(alias_name: str) -> str: + """Remove all entries from an alias (alias_util/flush).""" + validate_alias_mutation(alias_name) + payload = {"alias": alias_name} + if settings.dry_run: + return json.dumps(is_dry_run_response("alias_flush", payload), indent=2) + require_writes("alias_flush", payload) + client = get_client() + result = client.post("firewall", "alias_util", "flush", alias_name) + audit_log("alias_flush", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_reconfigure_aliases() -> str: + """Apply/reconfigure alias changes.""" + if settings.dry_run: + return json.dumps(is_dry_run_response("reconfigure_aliases", {}), indent=2) + require_writes("reconfigure_aliases", {}) + client = get_client() + result = client.post("firewall", "alias", "reconfigure") + audit_log("reconfigure_aliases", {"result": result}) + return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/dhcp.py b/src/opnsense_mcp/tools/dhcp.py new file mode 100644 index 0000000..4fbcf9a --- /dev/null +++ b/src/opnsense_mcp/tools/dhcp.py @@ -0,0 +1,58 @@ +"""DHCP lease and service tools.""" + +from __future__ import annotations + +import json + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp.audit import audit_log +from opnsense_mcp.client import get_client +from opnsense_mcp.config import settings +from opnsense_mcp.guards import is_dry_run_response, require_writes + + +def register(mcp: FastMCP) -> None: + @mcp.tool() + def opnsense_search_dhcp_leases(search_phrase: str = "", row_count: int = 200) -> str: + """Search active DHCPv4 leases (IP, MAC, hostname, expiry).""" + client = get_client() + try: + result = client.search_get( + "dhcpv4", "leases", "searchLease", search_phrase, row_count + ) + except Exception: + result = client.search_grid( + "dhcpv4", "leases", "searchLease", search_phrase, row_count + ) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_delete_dhcp_lease(ip: str) -> str: + """Delete/release a DHCP lease by IP address.""" + payload = {"ip": ip} + if settings.dry_run: + return json.dumps(is_dry_run_response("delete_dhcp_lease", payload), indent=2) + require_writes("delete_dhcp_lease", payload) + client = get_client() + result = client.post("dhcpv4", "leases", "delLease", ip) + audit_log("delete_dhcp_lease", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_dhcp_service_status() -> str: + """Get DHCPv4 service status.""" + client = get_client() + result = client.get("dhcpv4", "service", "status") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_dhcp_service_restart() -> str: + """Restart the DHCPv4 service.""" + if settings.dry_run: + return json.dumps(is_dry_run_response("dhcp_restart", {}), indent=2) + require_writes("dhcp_restart", {}) + client = get_client() + result = client.post("dhcpv4", "service", "restart") + audit_log("dhcp_restart", {"result": result}) + return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/diagnostics.py b/src/opnsense_mcp/tools/diagnostics.py new file mode 100644 index 0000000..6af156e --- /dev/null +++ b/src/opnsense_mcp/tools/diagnostics.py @@ -0,0 +1,147 @@ +"""Diagnostics: logs, states, ARP, ping, traffic.""" + +from __future__ import annotations + +import json +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp.audit import audit_log +from opnsense_mcp.client import get_client +from opnsense_mcp.config import settings +from opnsense_mcp.guards import is_dry_run_response, require_writes + + +def register(mcp: FastMCP) -> None: + @mcp.tool() + def opnsense_get_arp_table(search_phrase: str = "") -> str: + """Get the ARP table (IP to MAC mappings). Optionally filter with search_phrase.""" + client = get_client() + if search_phrase: + result = client.get( + "diagnostics", "interface", "searchArp", params={"searchPhrase": search_phrase} + ) + else: + result = client.get("diagnostics", "interface", "getArp") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_routes() -> str: + """Get the routing table.""" + client = get_client() + result = client.get("diagnostics", "interface", "getRoutes") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_firewall_log(limit: int = 100) -> str: + """Get recent firewall log entries.""" + client = get_client() + result = client.get("diagnostics", "firewall", "log", params={"limit": limit}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_firewall_log_filters() -> str: + """Get available firewall log filter options.""" + client = get_client() + result = client.get("diagnostics", "firewall", "logFilters") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_pf_states(limit: int = 100) -> str: + """Get active PF firewall state table entries.""" + client = get_client() + result = client.get("diagnostics", "firewall", "pfStates", params={"limit": limit}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_query_pf_states(filter_text: str = "", limit: int = 100) -> str: + """Query/filter the PF state table.""" + client = get_client() + result = client.post( + "diagnostics", + "firewall", + "queryStates", + data={"filter": filter_text, "limit": limit}, + ) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_firewall_stats() -> str: + """Get PF firewall statistics.""" + client = get_client() + result = client.get("diagnostics", "firewall", "stats") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_pf_statistics(section: str = "") -> str: + """Get detailed PF statistics, optionally filtered by section.""" + client = get_client() + params = {"section": section} if section else None + result = client.get("diagnostics", "firewall", "pfStatistics", params=params) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_flush_firewall_states() -> str: + """Flush all firewall state table entries. Disrupts active connections.""" + if settings.dry_run: + return json.dumps(is_dry_run_response("flush_states", {}), indent=2) + require_writes("flush_states", {}) + client = get_client() + result = client.post("diagnostics", "firewall", "flushStates") + audit_log("flush_states", {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_kill_firewall_states(source: str = "", destination: str = "") -> str: + """Kill firewall states matching source and/or destination IP.""" + payload: dict[str, Any] = {} + if source: + payload["source"] = source + if destination: + payload["destination"] = destination + if settings.dry_run: + return json.dumps(is_dry_run_response("kill_states", payload), indent=2) + require_writes("kill_states", payload) + client = get_client() + result = client.post("diagnostics", "firewall", "killStates", data=payload or None) + audit_log("kill_states", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_flush_arp_table() -> str: + """Flush the ARP cache.""" + if settings.dry_run: + return json.dumps(is_dry_run_response("flush_arp", {}), indent=2) + require_writes("flush_arp", {}) + client = get_client() + result = client.post("diagnostics", "interface", "flushArp") + audit_log("flush_arp", {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_dns_reverse_lookup(hostname: str = "", ip: str = "") -> str: + """Perform reverse DNS lookup from the firewall.""" + client = get_client() + params: dict[str, str] = {} + if hostname: + params["hostname"] = hostname + if ip: + params["ip"] = ip + result = client.get("diagnostics", "dns", "reverseLookup", params=params or None) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_traffic_top(interfaces: str = "") -> str: + """Get top talkers by traffic. interfaces: comma-separated interface names.""" + client = get_client() + params = {"interfaces": interfaces} if interfaces else None + result = client.get("diagnostics", "traffic", "top", params=params) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_activity() -> str: + """Get current system activity/process snapshot.""" + client = get_client() + result = client.get("diagnostics", "activity", "getActivity") + return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/firewall.py b/src/opnsense_mcp/tools/firewall.py new file mode 100644 index 0000000..bda914b --- /dev/null +++ b/src/opnsense_mcp/tools/firewall.py @@ -0,0 +1,238 @@ +"""Firewall filter rule tools.""" + +from __future__ import annotations + +import json +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp.audit import audit_log +from opnsense_mcp.client import get_client +from opnsense_mcp.config import settings +from opnsense_mcp.guards import ( + GuardError, + is_dry_run_response, + require_writes, + validate_firewall_rule, +) + + +def _rule_payload(rule: dict[str, Any]) -> dict[str, Any]: + return {"rule": rule} + + +def register(mcp: FastMCP) -> None: + @mcp.tool() + def opnsense_search_firewall_rules( + search_phrase: str = "", + interface: str = "", + row_count: int = 100, + ) -> str: + """Search firewall filter rules. Optionally filter by interface (e.g. lan, wan).""" + client = get_client() + extra: dict[str, Any] = {} + if interface: + extra["interface"] = interface + try: + result = client.search_get( + "firewall", "filter", "searchRule", search_phrase, row_count, **extra + ) + except Exception: + result = client.search_grid( + "firewall", "filter", "searchRule", search_phrase, row_count, **extra + ) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_firewall_rule(uuid: str) -> str: + """Get a single firewall rule by UUID.""" + client = get_client() + result = client.get("firewall", "filter", "getRule", uuid) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_firewall_interface_list() -> str: + """List interfaces available for firewall rule assignment.""" + client = get_client() + result = client.get("firewall", "filter", "getInterfaceList") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_firewall_rule_stats() -> str: + """Get per-rule hit counters and statistics.""" + client = get_client() + result = client.get("firewall", "filter_util", "ruleStats") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_add_firewall_rule(rule: dict[str, Any]) -> str: + """Add a new firewall filter rule. Rule dict uses OPNsense field names (description, interface, source_net, destination_net, protocol, action, etc.). Descriptions must start with the configured prefix (default 'mcp:').""" + validate_firewall_rule(rule, is_new=True) + payload = _rule_payload(rule) + if settings.dry_run: + return json.dumps(is_dry_run_response("add_firewall_rule", payload), indent=2) + require_writes("add_firewall_rule", payload) + client = get_client() + result = client.post("firewall", "filter", "addRule", data=payload) + audit_log("add_firewall_rule", {"rule": rule, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_update_firewall_rule(uuid: str, rule: dict[str, Any]) -> str: + """Update an existing firewall rule by UUID.""" + validate_firewall_rule(rule, is_new=False) + payload = _rule_payload(rule) + if settings.dry_run: + return json.dumps( + is_dry_run_response("update_firewall_rule", {"uuid": uuid, **payload}), indent=2 + ) + require_writes("update_firewall_rule", {"uuid": uuid, **payload}) + client = get_client() + result = client.post("firewall", "filter", "setRule", uuid, data=payload) + audit_log("update_firewall_rule", {"uuid": uuid, "rule": rule, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_delete_firewall_rule(uuid: str) -> str: + """Delete a firewall rule by UUID.""" + payload = {"uuid": uuid} + if settings.dry_run: + return json.dumps(is_dry_run_response("delete_firewall_rule", payload), indent=2) + require_writes("delete_firewall_rule", payload) + client = get_client() + result = client.post("firewall", "filter", "delRule", uuid) + audit_log("delete_firewall_rule", {"uuid": uuid, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_toggle_firewall_rule(uuid: str, enabled: bool = True) -> str: + """Enable or disable a firewall rule. enabled=true to enable, false to disable.""" + payload = {"uuid": uuid, "enabled": enabled} + if settings.dry_run: + return json.dumps(is_dry_run_response("toggle_firewall_rule", payload), indent=2) + require_writes("toggle_firewall_rule", payload) + client = get_client() + flag = "1" if enabled else "0" + result = client.post("firewall", "filter", "toggleRule", uuid, flag) + audit_log("toggle_firewall_rule", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_create_firewall_savepoint() -> str: + """Create a firewall configuration savepoint for rollback. Returns revision ID — call cancel_rollback within 60s to keep changes, or let it auto-revert.""" + if settings.dry_run: + return json.dumps(is_dry_run_response("create_savepoint", {}), indent=2) + require_writes("create_savepoint", {}) + client = get_client() + result = client.post("firewall", "filter", "savepoint") + audit_log("create_savepoint", {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_apply_firewall(rollback_revision: str = "") -> str: + """Apply pending firewall changes. Pass rollback_revision from savepoint to enable 60s auto-rollback.""" + payload = {"rollback_revision": rollback_revision} + if settings.dry_run: + return json.dumps(is_dry_run_response("apply_firewall", payload), indent=2) + require_writes("apply_firewall", payload) + client = get_client() + if rollback_revision: + result = client.post("firewall", "filter", "apply", rollback_revision) + else: + result = client.post("firewall", "filter", "apply") + audit_log("apply_firewall", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_cancel_firewall_rollback(rollback_revision: str) -> str: + """Confirm firewall changes by cancelling the auto-rollback timer for a savepoint revision.""" + payload = {"rollback_revision": rollback_revision} + if settings.dry_run: + return json.dumps(is_dry_run_response("cancel_rollback", payload), indent=2) + require_writes("cancel_rollback", payload) + client = get_client() + result = client.post("firewall", "filter", "cancelRollback", rollback_revision) + audit_log("cancel_rollback", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_revert_firewall(revision: str) -> str: + """Revert firewall configuration to a specific revision.""" + payload = {"revision": revision} + if settings.dry_run: + return json.dumps(is_dry_run_response("revert_firewall", payload), indent=2) + require_writes("revert_firewall", payload) + client = get_client() + result = client.post("firewall", "filter", "revert", revision) + audit_log("revert_firewall", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_safe_apply_firewall_rule_change( + uuid: str, + enabled: bool, + confirm: bool = False, + rollback_revision: str = "", + ) -> str: + """Safely toggle a rule with savepoint + apply + 60s rollback window. First call with confirm=false to apply; then call again with confirm=true and the returned rollback_revision to keep changes.""" + if confirm: + if not rollback_revision: + raise GuardError( + "rollback_revision is required when confirm=true. " + "Pass the revision returned from the initial safe_apply call." + ) + if settings.dry_run: + return json.dumps( + is_dry_run_response( + "safe_apply_confirm", {"uuid": uuid, "rollback_revision": rollback_revision} + ), + indent=2, + ) + require_writes("safe_apply_confirm", {"uuid": uuid, "rollback_revision": rollback_revision}) + client = get_client() + result = client.post("firewall", "filter", "cancelRollback", rollback_revision) + audit_log( + "safe_apply_confirm", + {"uuid": uuid, "rollback_revision": rollback_revision, "result": result}, + ) + return json.dumps( + {"status": "confirmed", "rollback_revision": rollback_revision, "result": result}, + indent=2, + default=str, + ) + + if settings.dry_run: + return json.dumps( + is_dry_run_response( + "safe_apply", + {"uuid": uuid, "enabled": enabled, "step": "would savepoint+toggle+apply"}, + ), + indent=2, + ) + require_writes("safe_apply", {"uuid": uuid, "enabled": enabled}) + client = get_client() + sp = client.post("firewall", "filter", "savepoint") + revision = sp.get("revision", "") + flag = "1" if enabled else "0" + client.post("firewall", "filter", "toggleRule", uuid, flag) + apply_result = ( + client.post("firewall", "filter", "apply", revision) + if revision + else client.post("firewall", "filter", "apply") + ) + audit_log("safe_apply", {"uuid": uuid, "enabled": enabled, "revision": revision}) + return json.dumps( + { + "status": "applied_with_rollback", + "rollback_revision": revision, + "rollback_seconds": 60, + "message": ( + "Verify connectivity, then call opnsense_safe_apply_firewall_rule_change " + "with confirm=true and rollback_revision set to keep changes." + ), + "apply_result": apply_result, + }, + indent=2, + default=str, + ) diff --git a/src/opnsense_mcp/tools/interfaces.py b/src/opnsense_mcp/tools/interfaces.py new file mode 100644 index 0000000..3feea9b --- /dev/null +++ b/src/opnsense_mcp/tools/interfaces.py @@ -0,0 +1,61 @@ +"""Network interface tools.""" + +from __future__ import annotations + +import json + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp.client import get_client + + +def register(mcp: FastMCP) -> None: + @mcp.tool() + def opnsense_list_interfaces(details: bool = True) -> str: + """List all network interfaces with status, IP addresses, and traffic stats.""" + client = get_client() + result = client.get( + "interfaces", "overview", "interfacesInfo", params={"details": str(details).lower()} + ) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_interface(identifier: str) -> str: + """Get detailed info for a single interface by name (e.g. lan, wan, opt1).""" + client = get_client() + result = client.get("interfaces", "overview", "getInterface", params={"if": identifier}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_reload_interface(identifier: str) -> str: + """Reload/restart a network interface.""" + from opnsense_mcp.audit import audit_log + from opnsense_mcp.config import settings + from opnsense_mcp.guards import is_dry_run_response, require_writes + + payload = {"identifier": identifier} + if settings.dry_run: + return json.dumps(is_dry_run_response("reload_interface", payload), indent=2) + require_writes("reload_interface", payload) + client = get_client() + result = client.post( + "interfaces", "overview", "reloadInterface", params={"identifier": identifier} + ) + audit_log("reload_interface", {"identifier": identifier, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_list_vlans(search_phrase: str = "", row_count: int = 50) -> str: + """List VLAN interface configurations.""" + client = get_client() + result = client.search_grid( + "interfaces", "vlan_settings", "searchItem", search_phrase, row_count + ) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_vip_status() -> str: + """Get virtual IP (CARP/VIP) status.""" + client = get_client() + result = client.get("diagnostics", "interface", "getVipStatus") + return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/nat.py b/src/opnsense_mcp/tools/nat.py new file mode 100644 index 0000000..e4f9eae --- /dev/null +++ b/src/opnsense_mcp/tools/nat.py @@ -0,0 +1,173 @@ +"""NAT rule tools: source NAT, destination NAT, 1:1, NPT.""" + +from __future__ import annotations + +import json +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp.audit import audit_log +from opnsense_mcp.client import get_client +from opnsense_mcp.config import settings +from opnsense_mcp.guards import is_dry_run_response, require_writes + + +def _search_nat(nat_type: str, search_phrase: str, row_count: int) -> str: + client = get_client() + try: + result = client.search_get("firewall", nat_type, "searchRule", search_phrase, row_count) + except Exception: + result = client.search_grid("firewall", nat_type, "searchRule", search_phrase, row_count) + return json.dumps(result, indent=2, default=str) + + +def _get_nat(nat_type: str, uuid: str) -> str: + client = get_client() + result = client.get("firewall", nat_type, "getRule", uuid) + return json.dumps(result, indent=2, default=str) + + +def _add_nat(nat_type: str, action: str, rule: dict[str, Any]) -> str: + payload = {"rule": rule} + if settings.dry_run: + return json.dumps(is_dry_run_response(action, payload), indent=2) + require_writes(action, payload) + client = get_client() + result = client.post("firewall", nat_type, "addRule", data=payload) + audit_log(action, {"rule": rule, "result": result}) + return json.dumps(result, indent=2, default=str) + + +def _update_nat(nat_type: str, action: str, uuid: str, rule: dict[str, Any]) -> str: + payload = {"rule": rule} + if settings.dry_run: + return json.dumps(is_dry_run_response(action, {"uuid": uuid, **payload}), indent=2) + require_writes(action, {"uuid": uuid, **payload}) + client = get_client() + result = client.post("firewall", nat_type, "setRule", uuid, data=payload) + audit_log(action, {"uuid": uuid, "rule": rule, "result": result}) + return json.dumps(result, indent=2, default=str) + + +def _delete_nat(nat_type: str, action: str, uuid: str) -> str: + if settings.dry_run: + return json.dumps(is_dry_run_response(action, {"uuid": uuid}), indent=2) + require_writes(action, {"uuid": uuid}) + client = get_client() + result = client.post("firewall", nat_type, "delRule", uuid) + audit_log(action, {"uuid": uuid, "result": result}) + return json.dumps(result, indent=2, default=str) + + +def _toggle_nat(nat_type: str, action: str, uuid: str, enabled: bool) -> str: + payload = {"uuid": uuid, "enabled": enabled} + if settings.dry_run: + return json.dumps(is_dry_run_response(action, payload), indent=2) + require_writes(action, payload) + client = get_client() + flag = "1" if enabled else "0" + result = client.post("firewall", nat_type, "toggleRule", uuid, flag) + audit_log(action, payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + +def register(mcp: FastMCP) -> None: + # ── Source NAT ────────────────────────────────────────────────────── + @mcp.tool() + def opnsense_search_source_nat_rules(search_phrase: str = "", row_count: int = 100) -> str: + """Search outbound / source NAT rules.""" + return _search_nat("source_nat", search_phrase, row_count) + + @mcp.tool() + def opnsense_get_source_nat_rule(uuid: str) -> str: + """Get an outbound NAT rule by UUID.""" + return _get_nat("source_nat", uuid) + + @mcp.tool() + def opnsense_add_source_nat_rule(rule: dict[str, Any]) -> str: + """Add an outbound / source NAT rule.""" + return _add_nat("source_nat", "add_source_nat_rule", rule) + + @mcp.tool() + def opnsense_update_source_nat_rule(uuid: str, rule: dict[str, Any]) -> str: + """Update an outbound NAT rule by UUID.""" + return _update_nat("source_nat", "update_source_nat_rule", uuid, rule) + + @mcp.tool() + def opnsense_delete_source_nat_rule(uuid: str) -> str: + """Delete an outbound NAT rule by UUID.""" + return _delete_nat("source_nat", "delete_source_nat_rule", uuid) + + @mcp.tool() + def opnsense_toggle_source_nat_rule(uuid: str, enabled: bool = True) -> str: + """Enable or disable an outbound NAT rule.""" + return _toggle_nat("source_nat", "toggle_source_nat_rule", uuid, enabled) + + # ── Destination NAT / Port Forwards ─────────────────────────────────── + @mcp.tool() + def opnsense_search_d_nat_rules(search_phrase: str = "", row_count: int = 100) -> str: + """Search destination NAT / port forward rules.""" + return _search_nat("d_nat", search_phrase, row_count) + + @mcp.tool() + def opnsense_search_port_forwards(search_phrase: str = "", row_count: int = 100) -> str: + """Search port forward rules (alias for destination NAT search).""" + return _search_nat("d_nat", search_phrase, row_count) + + @mcp.tool() + def opnsense_get_d_nat_rule(uuid: str) -> str: + """Get a destination NAT / port forward rule by UUID.""" + return _get_nat("d_nat", uuid) + + @mcp.tool() + def opnsense_add_d_nat_rule(rule: dict[str, Any]) -> str: + """Add a destination NAT / port forward rule.""" + return _add_nat("d_nat", "add_d_nat_rule", rule) + + @mcp.tool() + def opnsense_update_d_nat_rule(uuid: str, rule: dict[str, Any]) -> str: + """Update a destination NAT rule by UUID.""" + return _update_nat("d_nat", "update_d_nat_rule", uuid, rule) + + @mcp.tool() + def opnsense_delete_d_nat_rule(uuid: str) -> str: + """Delete a destination NAT rule by UUID.""" + return _delete_nat("d_nat", "delete_d_nat_rule", uuid) + + @mcp.tool() + def opnsense_toggle_d_nat_rule(uuid: str, enabled: bool = True) -> str: + """Enable or disable a destination NAT rule.""" + return _toggle_nat("d_nat", "toggle_d_nat_rule", uuid, enabled) + + # ── 1:1 NAT ───────────────────────────────────────────────────────── + @mcp.tool() + def opnsense_search_one_to_one_rules(search_phrase: str = "", row_count: int = 100) -> str: + """Search 1:1 NAT rules.""" + return _search_nat("one_to_one", search_phrase, row_count) + + @mcp.tool() + def opnsense_get_one_to_one_rule(uuid: str) -> str: + """Get a 1:1 NAT rule by UUID.""" + return _get_nat("one_to_one", uuid) + + @mcp.tool() + def opnsense_add_one_to_one_rule(rule: dict[str, Any]) -> str: + """Add a 1:1 NAT rule.""" + return _add_nat("one_to_one", "add_one_to_one_rule", rule) + + @mcp.tool() + def opnsense_delete_one_to_one_rule(uuid: str) -> str: + """Delete a 1:1 NAT rule by UUID.""" + return _delete_nat("one_to_one", "delete_one_to_one_rule", uuid) + + # ── NPT (IPv6) ────────────────────────────────────────────────────── + @mcp.tool() + def opnsense_search_npt_rules(search_phrase: str = "", row_count: int = 100) -> str: + """Search NPT (IPv6 prefix translation) rules.""" + return _search_nat("npt", search_phrase, row_count) + + @mcp.tool() + def opnsense_get_npt_rule(uuid: str) -> str: + """Get an NPT rule by UUID.""" + return _get_nat("npt", uuid) diff --git a/src/opnsense_mcp/tools/routes.py b/src/opnsense_mcp/tools/routes.py new file mode 100644 index 0000000..7689891 --- /dev/null +++ b/src/opnsense_mcp/tools/routes.py @@ -0,0 +1,87 @@ +"""Routing and gateway tools.""" + +from __future__ import annotations + +import json +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp.audit import audit_log +from opnsense_mcp.client import get_client +from opnsense_mcp.config import settings +from opnsense_mcp.guards import is_dry_run_response, require_writes + + +def register(mcp: FastMCP) -> None: + @mcp.tool() + def opnsense_get_gateway_status() -> str: + """Get WAN/gateway status including latency, packet loss, and online/offline state.""" + client = get_client() + result = client.get("routes", "gateway", "status") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_search_routes(search_phrase: str = "", row_count: int = 100) -> str: + """Search static route entries.""" + client = get_client() + result = client.search_grid("routes", "routes", "searchRoute", search_phrase, row_count) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_route(uuid: str = "") -> str: + """Get a static route by UUID, or route configuration template if uuid empty.""" + client = get_client() + if uuid: + result = client.get("routes", "routes", "getRoute", uuid) + else: + result = client.get("routes", "routes", "get") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_add_route(route: dict[str, Any]) -> str: + """Add a static route. Fields depend on OPNsense route model (network, gateway, description, etc.).""" + payload = {"route": route} + if settings.dry_run: + return json.dumps(is_dry_run_response("add_route", payload), indent=2) + require_writes("add_route", payload) + client = get_client() + result = client.post("routes", "routes", "addRoute", data=payload) + audit_log("add_route", {"route": route, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_delete_route(uuid: str) -> str: + """Delete a static route by UUID.""" + payload = {"uuid": uuid} + if settings.dry_run: + return json.dumps(is_dry_run_response("delete_route", payload), indent=2) + require_writes("delete_route", payload) + client = get_client() + result = client.post("routes", "routes", "delRoute", uuid) + audit_log("delete_route", {"uuid": uuid, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_toggle_route(uuid: str, enabled: bool = True) -> str: + """Enable or disable a static route.""" + payload = {"uuid": uuid, "enabled": enabled} + if settings.dry_run: + return json.dumps(is_dry_run_response("toggle_route", payload), indent=2) + require_writes("toggle_route", payload) + client = get_client() + flag = "1" if enabled else "0" + result = client.post("routes", "routes", "toggleRoute", uuid, flag) + audit_log("toggle_route", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_reconfigure_routes() -> str: + """Apply static route configuration changes.""" + if settings.dry_run: + return json.dumps(is_dry_run_response("reconfigure_routes", {}), indent=2) + require_writes("reconfigure_routes", {}) + client = get_client() + result = client.post("routes", "routes", "reconfigure") + audit_log("reconfigure_routes", {"result": result}) + return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/system.py b/src/opnsense_mcp/tools/system.py new file mode 100644 index 0000000..a9a54d4 --- /dev/null +++ b/src/opnsense_mcp/tools/system.py @@ -0,0 +1,69 @@ +"""System, firmware, and service tools.""" + +from __future__ import annotations + +import json +from typing import Any + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp.client import get_client + + +def register(mcp: FastMCP) -> None: + @mcp.tool() + def opnsense_get_system_info() -> str: + """Get OPNsense system information: version, hostname, uptime, CPU, memory, disk.""" + client = get_client() + info = client.get("diagnostics", "system", "systemInformation") + health = client.get("diagnostics", "systemhealth", "getSystemHealth") + return json.dumps({"system": info, "health": health}, indent=2, default=str) + + @mcp.tool() + def opnsense_get_firmware_status() -> str: + """Get firmware version and update status.""" + client = get_client() + result = client.get("core", "firmware", "status") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_list_services(search_phrase: str = "", row_count: int = 50) -> str: + """List system services and their running/stopped status.""" + client = get_client() + result = client.search_grid("core", "service", "search", search_phrase, row_count) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_service_action(service: str, action: str) -> str: + """Control a system service. action: start, stop, restart, reconfigure.""" + from opnsense_mcp.audit import audit_log + from opnsense_mcp.config import settings + from opnsense_mcp.guards import GuardError, is_dry_run_response, require_writes + + allowed = {"start", "stop", "restart", "reconfigure"} + if action not in allowed: + raise GuardError(f"action must be one of: {', '.join(sorted(allowed))}") + + payload = {"service": service, "action": action} + if settings.dry_run: + return json.dumps(is_dry_run_response("service_action", payload), indent=2) + require_writes("service_action", payload) + client = get_client() + result = client.post("core", "service", action, data={"service": service}) + audit_log("service_action", {"service": service, "action": action, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_system_time() -> str: + """Get current system time on the firewall.""" + client = get_client() + result = client.get("diagnostics", "system", "systemTime") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_system_resources() -> str: + """Get CPU load, memory usage, and mbuf statistics.""" + client = get_client() + resources = client.get("diagnostics", "system", "systemResources") + memory = client.get("diagnostics", "system", "memory") + return json.dumps({"resources": resources, "memory": memory}, indent=2, default=str)