From b62e05113d81dfd467751f01798d7ef54b3c5d11 Mon Sep 17 00:00:00 2001 From: Hermes CT107 Date: Sat, 4 Jul 2026 20:16:00 +0000 Subject: [PATCH] OPNsense MCP fixes: O1,O2,O4,O6 - narrow except, health check, register host, VLAN tools --- src/opnsense_mcp/tools/__init__.py | 2 + src/opnsense_mcp/tools/dhcp.py | 6 +- src/opnsense_mcp/tools/firewall.py | 4 +- src/opnsense_mcp/tools/hosts.py | 125 +++++++++++++++++++++++++++ src/opnsense_mcp/tools/interfaces.py | 42 +++++++++ src/opnsense_mcp/tools/nat.py | 4 +- src/opnsense_mcp/tools/system.py | 36 ++++++++ 7 files changed, 212 insertions(+), 7 deletions(-) create mode 100644 src/opnsense_mcp/tools/hosts.py diff --git a/src/opnsense_mcp/tools/__init__.py b/src/opnsense_mcp/tools/__init__.py index 278d8a8..cc5b1f0 100644 --- a/src/opnsense_mcp/tools/__init__.py +++ b/src/opnsense_mcp/tools/__init__.py @@ -10,6 +10,7 @@ from opnsense_mcp.tools import ( dns, firewall, firmware, + hosts, interfaces, logs, nat, @@ -21,6 +22,7 @@ from opnsense_mcp.tools import ( def register_all(mcp: FastMCP) -> None: system.register(mcp) + hosts.register(mcp) interfaces.register(mcp) firewall.register(mcp) aliases.register(mcp) diff --git a/src/opnsense_mcp/tools/dhcp.py b/src/opnsense_mcp/tools/dhcp.py index 662f6b0..7be4cf3 100644 --- a/src/opnsense_mcp/tools/dhcp.py +++ b/src/opnsense_mcp/tools/dhcp.py @@ -7,7 +7,7 @@ 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.client import OpnsenseAPIError, get_client from opnsense_mcp.config import settings from opnsense_mcp.guards import is_dry_run_response, require_writes @@ -21,7 +21,7 @@ def register(mcp: FastMCP) -> None: result = client.search_get( "dhcpv4", "leases", "searchLease", search_phrase, row_count ) - except Exception: + except OpnsenseAPIError: result = client.search_grid( "dhcpv4", "leases", "searchLease", search_phrase, row_count ) @@ -65,7 +65,7 @@ def register(mcp: FastMCP) -> None: result = client.search_grid( "kea", "dhcpv4", "searchReservation", search_phrase, row_count ) - except Exception: + except OpnsenseAPIError: result = client.search_grid( "dhcpv4", "settings", "searchReservation", search_phrase, row_count ) diff --git a/src/opnsense_mcp/tools/firewall.py b/src/opnsense_mcp/tools/firewall.py index bda914b..22b858e 100644 --- a/src/opnsense_mcp/tools/firewall.py +++ b/src/opnsense_mcp/tools/firewall.py @@ -8,7 +8,7 @@ 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.client import OpnsenseAPIError, get_client from opnsense_mcp.config import settings from opnsense_mcp.guards import ( GuardError, @@ -38,7 +38,7 @@ def register(mcp: FastMCP) -> None: result = client.search_get( "firewall", "filter", "searchRule", search_phrase, row_count, **extra ) - except Exception: + except OpnsenseAPIError: result = client.search_grid( "firewall", "filter", "searchRule", search_phrase, row_count, **extra ) diff --git a/src/opnsense_mcp/tools/hosts.py b/src/opnsense_mcp/tools/hosts.py new file mode 100644 index 0000000..454ebe9 --- /dev/null +++ b/src/opnsense_mcp/tools/hosts.py @@ -0,0 +1,125 @@ +"""Host registration composite tool: DHCP reservation + DNS override.""" + +from __future__ import annotations + +import json + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp.audit import audit_log +from opnsense_mcp.client import OpnsenseAPIError, 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_register_host( + mac: str, + ip: str, + hostname: str, + domain: str = "", + description: str = "", + ) -> str: + """Register a host by creating a DHCP static mapping and DNS host override, then reconfigure both services. + + Returns a combined result with dhcp_result, dns_result, dhcp_reconfigure, dns_reconfigure. + """ + payload = { + "mac": mac, + "ip": ip, + "hostname": hostname, + "domain": domain, + "description": description, + } + + if settings.dry_run: + return json.dumps( + is_dry_run_response("register_host", payload), indent=2 + ) + + require_writes("register_host", payload) + client = get_client() + + # ── 1. Add DHCP static mapping (try Kea, fall back to ISC) ── + dhcp_payload = { + "reservation": { + "hw_address": mac, + "ip_address": ip, + "hostname": hostname, + "description": description, + } + } + dhcp_result = {} + try: + dhcp_result = client.post( + "kea", "dhcpv4", "addReservation", data=dhcp_payload + ) + except OpnsenseAPIError: + try: + dhcp_result = client.post( + "dhcpv4", "settings", "addReservation", data=dhcp_payload + ) + except OpnsenseAPIError as exc: + dhcp_result = { + "error": str(exc), + "status_code": exc.status_code, + } + + # ── 2. Add DNS host override ── + dns_host = { + "enabled": "1", + "hostname": hostname, + "domain": domain, + "rr": "A", + "server": ip, + "description": description, + } + dns_payload = {"host": dns_host} + dns_result = {} + try: + dns_result = client.post( + "unbound", "settings", "addHostOverride", data=dns_payload + ) + except OpnsenseAPIError as exc: + dns_result = { + "error": str(exc), + "status_code": exc.status_code, + } + + # ── 3. Reconfigure DHCP service ── + dhcp_reconf = {} + try: + dhcp_reconf = client.post("dhcpv4", "service", "reconfigure") + except OpnsenseAPIError as exc: + dhcp_reconf = { + "error": str(exc), + "status_code": exc.status_code, + } + + # ── 4. Reconfigure Unbound DNS ── + dns_reconf = {} + try: + dns_reconf = client.post("unbound", "service", "reconfigure") + except OpnsenseAPIError as exc: + dns_reconf = { + "error": str(exc), + "status_code": exc.status_code, + } + + audit_log("register_host", payload | { + "dhcp_result": dhcp_result, + "dns_result": dns_result, + }) + + return json.dumps( + { + "host": {"mac": mac, "ip": ip, "hostname": hostname, "domain": domain}, + "dhcp_result": dhcp_result, + "dns_result": dns_result, + "dhcp_reconfigure": dhcp_reconf, + "dns_reconfigure": dns_reconf, + }, + indent=2, + default=str, + ) diff --git a/src/opnsense_mcp/tools/interfaces.py b/src/opnsense_mcp/tools/interfaces.py index 3feea9b..69272cd 100644 --- a/src/opnsense_mcp/tools/interfaces.py +++ b/src/opnsense_mcp/tools/interfaces.py @@ -59,3 +59,45 @@ def register(mcp: FastMCP) -> None: client = get_client() result = client.get("diagnostics", "interface", "getVipStatus") return json.dumps(result, indent=2, default=str) + @mcp.tool() + def opnsense_add_vlan( + interface: str = "opt1", + vlan_tag: int = 0, + description: str = "", + ) -> str: + """Add a new VLAN interface on a parent interface (e.g. opt1). Returns the new UUID.""" + from opnsense_mcp.audit import audit_log + from opnsense_mcp.config import settings + from opnsense_mcp.guards import is_dry_run_response, require_writes + + vlan = { + "vlanif": interface, + "tag": str(vlan_tag), + "descr": description, + } + payload = {"vlan": vlan} + if settings.dry_run: + return json.dumps(is_dry_run_response("add_vlan", payload), indent=2) + require_writes("add_vlan", payload) + client = get_client() + result = client.post( + "interfaces", "vlan_settings", "addItem", data=payload + ) + audit_log("add_vlan", {"vlan": vlan, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_delete_vlan(uuid: str) -> str: + """Delete a VLAN interface configuration by UUID.""" + 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 = {"uuid": uuid} + if settings.dry_run: + return json.dumps(is_dry_run_response("delete_vlan", payload), indent=2) + require_writes("delete_vlan", payload) + client = get_client() + result = client.post("interfaces", "vlan_settings", "delItem", uuid) + audit_log("delete_vlan", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/nat.py b/src/opnsense_mcp/tools/nat.py index e4f9eae..b23caa0 100644 --- a/src/opnsense_mcp/tools/nat.py +++ b/src/opnsense_mcp/tools/nat.py @@ -8,7 +8,7 @@ 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.client import OpnsenseAPIError, get_client from opnsense_mcp.config import settings from opnsense_mcp.guards import is_dry_run_response, require_writes @@ -17,7 +17,7 @@ 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: + except OpnsenseAPIError: result = client.search_grid("firewall", nat_type, "searchRule", search_phrase, row_count) return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/system.py b/src/opnsense_mcp/tools/system.py index a9a54d4..0334b3a 100644 --- a/src/opnsense_mcp/tools/system.py +++ b/src/opnsense_mcp/tools/system.py @@ -67,3 +67,39 @@ def register(mcp: FastMCP) -> None: resources = client.get("diagnostics", "system", "systemResources") memory = client.get("diagnostics", "system", "memory") return json.dumps({"resources": resources, "memory": memory}, indent=2, default=str) + @mcp.tool() + def opnsense_health_check() -> str: + """Comprehensive health snapshot: gateways, interfaces, services, system resources, firmware, WireGuard status.""" + client = get_client() + try: + gateways = client.get("routes", "gateway", "status") + except Exception: + gateways = {"error": "gateway check failed"} + try: + interfaces = client.get("interfaces", "overview", "interfacesInfo") + except Exception: + interfaces = {"error": "interface check failed"} + try: + services = client.search_grid("core", "service", "search", "", 50) + except Exception: + services = {"error": "service check failed"} + try: + resources = client.get("diagnostics", "system", "systemResources") + except Exception: + resources = {"error": "resource check failed"} + try: + firmware = client.get("core", "firmware", "status") + except Exception: + firmware = {"error": "firmware check failed"} + try: + wg_status = client.get("wireguard", "service", "show") + except Exception: + wg_status = {"error": "WireGuard check failed or not installed"} + return json.dumps({ + "gateways": gateways, + "interfaces": interfaces, + "services": services, + "resources": resources, + "firmware": firmware, + "wireguard": wg_status, + }, indent=2, default=str)