From 22e4f3a4691d5c4e6b40bb39ffdeefa914b84665 Mon Sep 17 00:00:00 2001 From: Hermes CT107 Date: Fri, 12 Jun 2026 18:54:21 +0000 Subject: [PATCH] Add Unbound DNS, VPN, firmware, logs, and certificate tools (v1.1.0) --- README.md | 15 ++- pyproject.toml | 2 +- src/opnsense_mcp/__init__.py | 2 +- src/opnsense_mcp/tools/__init__.py | 10 ++ src/opnsense_mcp/tools/certs.py | 38 ++++++ src/opnsense_mcp/tools/dhcp.py | 14 ++ src/opnsense_mcp/tools/diagnostics.py | 26 ++++ src/opnsense_mcp/tools/dns.py | 180 ++++++++++++++++++++++++++ src/opnsense_mcp/tools/firmware.py | 71 ++++++++++ src/opnsense_mcp/tools/logs.py | 72 +++++++++++ src/opnsense_mcp/tools/vpn.py | 119 +++++++++++++++++ 11 files changed, 546 insertions(+), 3 deletions(-) create mode 100644 src/opnsense_mcp/tools/certs.py create mode 100644 src/opnsense_mcp/tools/dns.py create mode 100644 src/opnsense_mcp/tools/firmware.py create mode 100644 src/opnsense_mcp/tools/logs.py create mode 100644 src/opnsense_mcp/tools/vpn.py diff --git a/README.md b/README.md index 78ba1ab..2f28bfe 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ Model Context Protocol server for [OPNsense](https://opnsense.org/) firewall man ## Features -- **60+ tools** covering system info, interfaces, firewall rules, aliases, NAT, DHCP, diagnostics, routes, and gateways +- **90+ tools** covering system info, interfaces, firewall rules, aliases, NAT, DHCP, diagnostics, routes, gateways, Unbound DNS, VPN (WireGuard/OpenVPN/IPsec), firmware/updates, system logs, and certificates - **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 @@ -101,6 +101,15 @@ Or ask Hermes: *"Use opnsense_test_connection to verify firewall access"* | `opnsense_get_firewall_log` | Recent blocked/ passed traffic | | `opnsense_get_pf_states` | Active connection states | | `opnsense_search_port_forwards` | Port forward rules | +| `opnsense_search_dns_host_overrides` | Unbound local DNS records | +| `opnsense_wireguard_status` | WireGuard tunnels and peer handshakes | +| `opnsense_openvpn_sessions` | Connected OpenVPN clients | +| `opnsense_ipsec_status` | IPsec service and sessions | +| `opnsense_firmware_check_updates` | Available updates from mirrors | +| `opnsense_firmware_audit` | Package vulnerability audit | +| `opnsense_get_system_log` | Syslog search (system, dhcp, dns, vpn, audit, ...) | +| `opnsense_list_certificates` | Trust store certificates (no private keys) | +| `opnsense_get_interface_traffic` | Per-interface traffic rates | ### Write operations (require `OPNSENSE_ALLOW_WRITES=true`) @@ -114,6 +123,10 @@ Or ask Hermes: *"Use opnsense_test_connection to verify firewall access"* | `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 | +| `opnsense_add_dns_host_override` | Add local DNS record (+ reconfigure) | +| `opnsense_wireguard_toggle_peer` | Enable/disable WireGuard peer | +| `opnsense_unbound_restart` | Restart DNS resolver | +| `opnsense_firmware_update` | Install updates (double confirm) | ## Safe Change Workflow diff --git a/pyproject.toml b/pyproject.toml index 137ddc4..445e94e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "opnsense-mcp" -version = "1.0.0" +version = "1.1.0" description = "Model Context Protocol server for OPNsense firewall management" readme = "README.md" requires-python = ">=3.10" diff --git a/src/opnsense_mcp/__init__.py b/src/opnsense_mcp/__init__.py index c9a437c..27ff12f 100644 --- a/src/opnsense_mcp/__init__.py +++ b/src/opnsense_mcp/__init__.py @@ -1,3 +1,3 @@ """OPNsense MCP server — Model Context Protocol integration for OPNsense firewalls.""" -__version__ = "1.0.0" +__version__ = "1.1.0" diff --git a/src/opnsense_mcp/tools/__init__.py b/src/opnsense_mcp/tools/__init__.py index 91e4e44..278d8a8 100644 --- a/src/opnsense_mcp/tools/__init__.py +++ b/src/opnsense_mcp/tools/__init__.py @@ -4,13 +4,18 @@ from mcp.server.fastmcp import FastMCP from opnsense_mcp.tools import ( aliases, + certs, dhcp, diagnostics, + dns, firewall, + firmware, interfaces, + logs, nat, routes, system, + vpn, ) @@ -23,3 +28,8 @@ def register_all(mcp: FastMCP) -> None: dhcp.register(mcp) diagnostics.register(mcp) routes.register(mcp) + dns.register(mcp) + vpn.register(mcp) + firmware.register(mcp) + logs.register(mcp) + certs.register(mcp) diff --git a/src/opnsense_mcp/tools/certs.py b/src/opnsense_mcp/tools/certs.py new file mode 100644 index 0000000..b078429 --- /dev/null +++ b/src/opnsense_mcp/tools/certs.py @@ -0,0 +1,38 @@ +"""Certificate and CA inspection tools (trust store, read-only).""" + +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_certificates(search_phrase: str = "", row_count: int = 100) -> str: + """List certificates in the trust store (names, purposes, validity dates).""" + client = get_client() + result = client.search_grid("trust", "cert", "search", search_phrase, row_count) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_list_certificate_authorities(search_phrase: str = "", row_count: int = 50) -> str: + """List certificate authorities (CAs) in the trust store.""" + client = get_client() + result = client.search_grid("trust", "ca", "search", search_phrase, row_count) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_certificate(uuid: str) -> str: + """Get certificate details by UUID (subject, issuer, validity — no private key).""" + client = get_client() + result = client.get("trust", "cert", "get", uuid) + # Strip private key material if present + if isinstance(result, dict): + cert = result.get("cert", {}) + if isinstance(cert, dict): + cert.pop("prv", None) + cert.pop("prv_payload", None) + return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/dhcp.py b/src/opnsense_mcp/tools/dhcp.py index 4fbcf9a..662f6b0 100644 --- a/src/opnsense_mcp/tools/dhcp.py +++ b/src/opnsense_mcp/tools/dhcp.py @@ -56,3 +56,17 @@ def register(mcp: FastMCP) -> None: result = client.post("dhcpv4", "service", "restart") audit_log("dhcp_restart", {"result": result}) return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_search_dhcp_reservations(search_phrase: str = "", row_count: int = 200) -> str: + """Search DHCP static reservations. Tries Kea (newer) then ISC legacy endpoint.""" + client = get_client() + try: + result = client.search_grid( + "kea", "dhcpv4", "searchReservation", search_phrase, row_count + ) + except Exception: + result = client.search_grid( + "dhcpv4", "settings", "searchReservation", search_phrase, row_count + ) + return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/diagnostics.py b/src/opnsense_mcp/tools/diagnostics.py index 6af156e..f272cc9 100644 --- a/src/opnsense_mcp/tools/diagnostics.py +++ b/src/opnsense_mcp/tools/diagnostics.py @@ -145,3 +145,29 @@ def register(mcp: FastMCP) -> None: client = get_client() result = client.get("diagnostics", "activity", "getActivity") return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_interface_traffic() -> str: + """Get current per-interface traffic rates (bytes/packets in and out).""" + client = get_client() + result = client.get("diagnostics", "traffic", "interface") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_ndp_table(search_phrase: str = "") -> str: + """Get the IPv6 NDP (neighbor discovery) table.""" + client = get_client() + if search_phrase: + result = client.get( + "diagnostics", "interface", "searchNdp", params={"searchPhrase": search_phrase} + ) + else: + result = client.get("diagnostics", "interface", "getNdp") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_capture_status() -> str: + """Get packet capture job status (diagnostics/packet_capture).""" + client = get_client() + result = client.get("diagnostics", "packet_capture", "searchJobs") + return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/dns.py b/src/opnsense_mcp/tools/dns.py new file mode 100644 index 0000000..b4ade5c --- /dev/null +++ b/src/opnsense_mcp/tools/dns.py @@ -0,0 +1,180 @@ +"""Unbound DNS resolver tools: host/domain overrides, service control, stats.""" + +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: + # ── Service ───────────────────────────────────────────────────────── + @mcp.tool() + def opnsense_unbound_service_status() -> str: + """Get Unbound DNS resolver service status.""" + client = get_client() + result = client.get("unbound", "service", "status") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_unbound_restart() -> str: + """Restart the Unbound DNS resolver service.""" + if settings.dry_run: + return json.dumps(is_dry_run_response("unbound_restart", {}), indent=2) + require_writes("unbound_restart", {}) + client = get_client() + result = client.post("unbound", "service", "restart") + audit_log("unbound_restart", {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_unbound_reconfigure() -> str: + """Apply Unbound configuration changes (required after host/domain override edits).""" + if settings.dry_run: + return json.dumps(is_dry_run_response("unbound_reconfigure", {}), indent=2) + require_writes("unbound_reconfigure", {}) + client = get_client() + result = client.post("unbound", "service", "reconfigure") + audit_log("unbound_reconfigure", {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_unbound_stats() -> str: + """Get Unbound DNS resolver statistics (cache hits, queries, etc.).""" + client = get_client() + result = client.get("unbound", "diagnostics", "stats") + return json.dumps(result, indent=2, default=str) + + # ── Host overrides ────────────────────────────────────────────────── + @mcp.tool() + def opnsense_search_dns_host_overrides(search_phrase: str = "", row_count: int = 100) -> str: + """Search Unbound DNS host overrides (local DNS records).""" + client = get_client() + result = client.search_grid( + "unbound", "settings", "searchHostOverride", search_phrase, row_count + ) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_get_dns_host_override(uuid: str) -> str: + """Get a DNS host override by UUID.""" + client = get_client() + result = client.get("unbound", "settings", "getHostOverride", uuid) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_add_dns_host_override( + hostname: str, + domain: str, + ip: str, + description: str = "", + enabled: bool = True, + ) -> str: + """Add a DNS host override (local A/AAAA record). Call opnsense_unbound_reconfigure afterwards to apply.""" + host = { + "enabled": "1" if enabled else "0", + "hostname": hostname, + "domain": domain, + "rr": "A", + "server": ip, + "description": description, + } + payload = {"host": host} + if settings.dry_run: + return json.dumps(is_dry_run_response("add_dns_host_override", payload), indent=2) + require_writes("add_dns_host_override", payload) + client = get_client() + result = client.post("unbound", "settings", "addHostOverride", data=payload) + audit_log("add_dns_host_override", {"host": host, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_update_dns_host_override(uuid: str, host: dict[str, Any]) -> str: + """Update a DNS host override by UUID. host dict fields: enabled, hostname, domain, rr, server, description.""" + payload = {"host": host} + if settings.dry_run: + return json.dumps( + is_dry_run_response("update_dns_host_override", {"uuid": uuid, **payload}), + indent=2, + ) + require_writes("update_dns_host_override", {"uuid": uuid, **payload}) + client = get_client() + result = client.post("unbound", "settings", "setHostOverride", uuid, data=payload) + audit_log("update_dns_host_override", {"uuid": uuid, "host": host, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_delete_dns_host_override(uuid: str) -> str: + """Delete a DNS host override by UUID.""" + payload = {"uuid": uuid} + if settings.dry_run: + return json.dumps(is_dry_run_response("delete_dns_host_override", payload), indent=2) + require_writes("delete_dns_host_override", payload) + client = get_client() + result = client.post("unbound", "settings", "delHostOverride", uuid) + audit_log("delete_dns_host_override", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_toggle_dns_host_override(uuid: str, enabled: bool = True) -> str: + """Enable or disable a DNS host override.""" + payload = {"uuid": uuid, "enabled": enabled} + if settings.dry_run: + return json.dumps(is_dry_run_response("toggle_dns_host_override", payload), indent=2) + require_writes("toggle_dns_host_override", payload) + client = get_client() + flag = "1" if enabled else "0" + result = client.post("unbound", "settings", "toggleHostOverride", uuid, flag) + audit_log("toggle_dns_host_override", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + # ── Domain overrides ──────────────────────────────────────────────── + @mcp.tool() + def opnsense_search_dns_domain_overrides(search_phrase: str = "", row_count: int = 100) -> str: + """Search Unbound DNS domain overrides (forward zones to other DNS servers).""" + client = get_client() + result = client.search_grid( + "unbound", "settings", "searchDomainOverride", search_phrase, row_count + ) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_add_dns_domain_override( + domain: str, + server: str, + description: str = "", + enabled: bool = True, + ) -> str: + """Add a DNS domain override forwarding a zone to a specific DNS server. Reconfigure to apply.""" + item = { + "enabled": "1" if enabled else "0", + "domain": domain, + "server": server, + "description": description, + } + payload = {"domain": item} + if settings.dry_run: + return json.dumps(is_dry_run_response("add_dns_domain_override", payload), indent=2) + require_writes("add_dns_domain_override", payload) + client = get_client() + result = client.post("unbound", "settings", "addDomainOverride", data=payload) + audit_log("add_dns_domain_override", {"domain": item, "result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_delete_dns_domain_override(uuid: str) -> str: + """Delete a DNS domain override by UUID.""" + payload = {"uuid": uuid} + if settings.dry_run: + return json.dumps(is_dry_run_response("delete_dns_domain_override", payload), indent=2) + require_writes("delete_dns_domain_override", payload) + client = get_client() + result = client.post("unbound", "settings", "delDomainOverride", uuid) + audit_log("delete_dns_domain_override", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/firmware.py b/src/opnsense_mcp/tools/firmware.py new file mode 100644 index 0000000..13abc4e --- /dev/null +++ b/src/opnsense_mcp/tools/firmware.py @@ -0,0 +1,71 @@ +"""Firmware, updates, and plugin management 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_firmware_info() -> str: + """Get firmware product info and full plugin/package list with versions.""" + client = get_client() + result = client.get("core", "firmware", "info") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_firmware_check_updates() -> str: + """Check for available firmware and package updates (queries OPNsense mirrors).""" + client = get_client() + result = client.post("core", "firmware", "check") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_firmware_upgrade_status() -> str: + """Get the status/progress of a running firmware check or upgrade.""" + client = get_client() + result = client.get("core", "firmware", "upgradestatus") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_firmware_audit() -> str: + """Run a security audit of installed packages against known vulnerabilities.""" + client = get_client() + result = client.post("core", "firmware", "audit") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_firmware_changelog(version: str) -> str: + """Get the changelog for a specific OPNsense version (e.g. '24.7').""" + client = get_client() + result = client.post("core", "firmware", "changelog", version) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_firmware_update(confirm: bool = False) -> str: + """Install pending minor updates. Requires confirm=true AND OPNSENSE_ALLOW_WRITES=true. The firewall may restart services or reboot.""" + if not confirm: + return json.dumps( + { + "status": "confirmation_required", + "message": ( + "This installs pending updates and may reboot the firewall. " + "Call again with confirm=true to proceed." + ), + }, + indent=2, + ) + if settings.dry_run: + return json.dumps(is_dry_run_response("firmware_update", {}), indent=2) + require_writes("firmware_update", {}) + client = get_client() + result = client.post("core", "firmware", "update") + audit_log("firmware_update", {"result": result}) + return json.dumps(result, indent=2, default=str) diff --git a/src/opnsense_mcp/tools/logs.py b/src/opnsense_mcp/tools/logs.py new file mode 100644 index 0000000..e4fc67f --- /dev/null +++ b/src/opnsense_mcp/tools/logs.py @@ -0,0 +1,72 @@ +"""System log search tools (syslog via diagnostics/log API).""" + +from __future__ import annotations + +import json + +from mcp.server.fastmcp import FastMCP + +from opnsense_mcp.client import get_client + + +def _search_log(module: str, scope: str, search_phrase: str, count: int) -> str: + client = get_client() + result = client.post( + "diagnostics", + "log", + module, + scope, + data={ + "current": 1, + "rowCount": count, + "searchPhrase": search_phrase, + }, + ) + return json.dumps(result, indent=2, default=str) + + +def register(mcp: FastMCP) -> None: + @mcp.tool() + def opnsense_get_system_log(search_phrase: str = "", count: int = 100) -> str: + """Search the main system log (/var/log/system).""" + return _search_log("core", "system", search_phrase, count) + + @mcp.tool() + def opnsense_get_backend_log(search_phrase: str = "", count: int = 100) -> str: + """Search the configd backend log.""" + return _search_log("core", "configd", search_phrase, count) + + @mcp.tool() + def opnsense_get_boot_log(search_phrase: str = "", count: int = 100) -> str: + """Search the boot/dmesg log.""" + return _search_log("core", "boot", search_phrase, count) + + @mcp.tool() + def opnsense_get_dhcp_log(search_phrase: str = "", count: int = 100) -> str: + """Search the DHCP daemon log.""" + return _search_log("dhcpd", "dhcpd", search_phrase, count) + + @mcp.tool() + def opnsense_get_dns_log(search_phrase: str = "", count: int = 100) -> str: + """Search the Unbound DNS resolver log.""" + return _search_log("unbound", "unbound", search_phrase, count) + + @mcp.tool() + def opnsense_get_openvpn_log(search_phrase: str = "", count: int = 100) -> str: + """Search the OpenVPN log.""" + return _search_log("openvpn", "openvpn", search_phrase, count) + + @mcp.tool() + def opnsense_get_ipsec_log(search_phrase: str = "", count: int = 100) -> str: + """Search the IPsec/charon log.""" + return _search_log("ipsec", "charon", search_phrase, count) + + @mcp.tool() + def opnsense_get_gateway_log(search_phrase: str = "", count: int = 100) -> str: + """Search the gateway monitor (dpinger) log.""" + return _search_log("core", "gateways", search_phrase, count) + + @mcp.tool() + def opnsense_get_audit_log(search_phrase: str = "", count: int = 100) -> str: + """Search the configuration audit log (who changed what).""" + return _search_log("core", "audit", search_phrase, count) diff --git a/src/opnsense_mcp/tools/vpn.py b/src/opnsense_mcp/tools/vpn.py new file mode 100644 index 0000000..758f127 --- /dev/null +++ b/src/opnsense_mcp/tools/vpn.py @@ -0,0 +1,119 @@ +"""VPN tools: WireGuard, OpenVPN, and IPsec status and management.""" + +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: + # ── WireGuard ─────────────────────────────────────────────────────── + @mcp.tool() + def opnsense_wireguard_status() -> str: + """Get WireGuard service status including active tunnels and peer handshakes.""" + client = get_client() + result = client.get("wireguard", "service", "show") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_wireguard_list_servers(search_phrase: str = "", row_count: int = 50) -> str: + """List WireGuard server (instance) configurations.""" + client = get_client() + result = client.search_grid( + "wireguard", "server", "searchServer", search_phrase, row_count + ) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_wireguard_list_peers(search_phrase: str = "", row_count: int = 100) -> str: + """List WireGuard peer (client) configurations.""" + client = get_client() + result = client.search_grid( + "wireguard", "client", "searchClient", search_phrase, row_count + ) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_wireguard_restart() -> str: + """Restart the WireGuard service.""" + if settings.dry_run: + return json.dumps(is_dry_run_response("wireguard_restart", {}), indent=2) + require_writes("wireguard_restart", {}) + client = get_client() + result = client.post("wireguard", "service", "restart") + audit_log("wireguard_restart", {"result": result}) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_wireguard_toggle_peer(uuid: str, enabled: bool = True) -> str: + """Enable or disable a WireGuard peer by UUID. Reconfigure/restart WireGuard to apply.""" + payload = {"uuid": uuid, "enabled": enabled} + if settings.dry_run: + return json.dumps(is_dry_run_response("wireguard_toggle_peer", payload), indent=2) + require_writes("wireguard_toggle_peer", payload) + client = get_client() + flag = "1" if enabled else "0" + result = client.post("wireguard", "client", "toggleClient", uuid, flag) + audit_log("wireguard_toggle_peer", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + # ── OpenVPN ───────────────────────────────────────────────────────── + @mcp.tool() + def opnsense_openvpn_sessions(search_phrase: str = "", row_count: int = 100) -> str: + """List active OpenVPN sessions (connected clients with IPs and bytes).""" + client = get_client() + result = client.search_grid( + "openvpn", "service", "searchSessions", search_phrase, row_count + ) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_openvpn_instances(search_phrase: str = "", row_count: int = 50) -> str: + """List OpenVPN instance configurations (servers and clients).""" + client = get_client() + result = client.search_grid("openvpn", "instances", "search", search_phrase, row_count) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_openvpn_kill_session(session_id: str) -> str: + """Disconnect an OpenVPN session by session id (from searchSessions).""" + payload = {"session_id": session_id} + if settings.dry_run: + return json.dumps(is_dry_run_response("openvpn_kill_session", payload), indent=2) + require_writes("openvpn_kill_session", payload) + client = get_client() + result = client.post("openvpn", "service", "killSession", data=payload) + audit_log("openvpn_kill_session", payload | {"result": result}) + return json.dumps(result, indent=2, default=str) + + # ── IPsec ─────────────────────────────────────────────────────────── + @mcp.tool() + def opnsense_ipsec_status() -> str: + """Get IPsec service status.""" + client = get_client() + result = client.get("ipsec", "service", "status") + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_ipsec_sessions(search_phrase: str = "", row_count: int = 100) -> str: + """List IPsec security associations / active sessions.""" + client = get_client() + result = client.search_grid( + "ipsec", "sessions", "searchPhase1", search_phrase, row_count + ) + return json.dumps(result, indent=2, default=str) + + @mcp.tool() + def opnsense_ipsec_connections(search_phrase: str = "", row_count: int = 50) -> str: + """List configured IPsec connections (swanctl model).""" + client = get_client() + result = client.search_grid( + "ipsec", "connections", "searchConnection", search_phrase, row_count + ) + return json.dumps(result, indent=2, default=str)