Add Unbound DNS, VPN, firmware, logs, and certificate tools (v1.1.0)

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
 

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"

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"

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)

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)

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)

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)

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)

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)

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)

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)