# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""NAPALM driver for OPNsense firewalls (REST API).

OPNsense exposes a JSON REST API at ``/api/``.  Authentication uses an
API key/secret pair generated in the OPNsense GUI (System → Access →
Users → edit user → API keys).  Pass them via *optional_args*:

    optional_args={
        "api_key": "<key>",
        "api_secret": "<secret>",
        "verify": False,  # disable TLS verification for self-signed certs
    }

If *api_key* / *api_secret* are omitted the driver falls back to the
positional *username* / *password* arguments.

Config management (load_merge_candidate / commit_config / rollback)
operates on **static routes** via ``/api/routes/routes/``.  Other parts
of the OPNsense configuration (firewall rules, interface settings, …) are
not exposed through a generic config-push endpoint and must be managed
per-module through their respective API controllers.
"""

from __future__ import annotations

import difflib
import json
import logging
import socket
from ipaddress import ip_address, ip_network
from typing import Any

logger = logging.getLogger(__name__)

import requests
from requests.exceptions import RequestException

from napalm_device_types import FingerprintRule, FirewallDriver
from napalm.base.exceptions import ConnectionException, ConnectionClosedException, MergeConfigException

from napalm_opnsense.ping_mixin import OPNsensePingMixin


class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
    """NAPALM driver for OPNsense (read-only, REST API)."""

    VENDOR = "OPNsense"
    DRIVER_NAME = "opnsense"
    SNMP_FINGERPRINT = [
        FingerprintRule("opnsense", weight=8.0, mandatory=True),
    ]
    SSH_FINGERPRINT = [
        FingerprintRule("freebsd", weight=7.0),
    ]
    HTTP_FINGERPRINT = [
        FingerprintRule("opnsense", weight=8.0, mandatory=True),
    ]

    def __init__(
        self,
        hostname: str,
        username: str,
        password: str,
        timeout: int = 60,
        optional_args: dict[str, Any] | None = None,
    ) -> None:
        self.hostname = hostname
        self.username = username
        self.password = password
        self.timeout = timeout
        self.optional_args = optional_args or {}

        # NAPALM standard attributes
        self.force_no_enable = True
        self.use_canonical_interface = False

        # OPNsense REST API settings
        self.base_url = self.optional_args.get("base_url") or f"https://{hostname}"
        # Accept "verify", "verify_ssl", or "ssl_verify" (NetOrk passes "verify_ssl")
        _v = self.optional_args.get("verify", self.optional_args.get("verify_ssl", self.optional_args.get("ssl_verify", True)))
        self.verify = bool(_v)
        self.api_key = self.optional_args.get("api_key") or username
        self.api_secret = self.optional_args.get("api_secret") or password

        self.session: requests.Session | None = None

        # Config-management state
        self._candidate_config: list[dict[str, Any]] | None = None
        # Backup ID of the config snapshot taken just before commit_config().
        # Used by rollback() to restore the exact pre-commit state.
        self._pre_commit_backup_id: str | None = None

    # ------------------------------------------------------------------
    # Connection management
    # ------------------------------------------------------------------

    def open(self) -> None:
        """Open an HTTPS session to the OPNsense API and validate credentials."""
        try:
            s = requests.Session()
            s.verify = self.verify
            s.headers.update({"Accept": "application/json"})
            s.auth = (self.api_key, self.api_secret)
            self.session = s
            # Lightweight connectivity and auth check
            self._get("/api/core/system/status")
        except RequestException as exc:
            self.session = None
            raise ConnectionException(
                f"Cannot connect to OPNsense API at {self.base_url}: {exc}"
            ) from exc

    def close(self) -> None:
        """Close the HTTPS session."""
        if self.session is not None:
            self.session.close()
            self.session = None

    def is_alive(self) -> dict[str, bool]:
        """Return whether the session is usable.

        Performs a lightweight socket-level check without sending a full
        HTTP request to avoid polluting API logs.
        """
        if self.session is None:
            return {"is_alive": False}
        try:
            host = self.hostname
            port = 443
            with socket.create_connection((host, port), timeout=5):
                pass
            return {"is_alive": True}
        except (socket.error, OSError):
            return {"is_alive": False}

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _get(self, path: str) -> dict[str, Any]:
        """Perform a GET request against the OPNsense REST API.

        :raises ConnectionClosedException: if called before :meth:`open`.
        :raises RequestException: on HTTP-level errors.
        """
        if self.session is None:
            raise ConnectionClosedException("Not connected – call open() first.")
        url = self.base_url.rstrip("/") + path
        response = self.session.get(url, timeout=self.timeout)
        response.raise_for_status()
        return response.json()

    def _post(self, path: str, data: dict[str, Any] | None = None) -> dict[str, Any]:
        """Perform a POST request against the OPNsense REST API.

        :param path: API path, e.g. ``/api/routes/routes/addroute``.
        :param data: JSON-serialisable payload (sent as ``application/json``).
        :raises ConnectionClosedException: if called before :meth:`open`.
        :raises RequestException: on HTTP-level errors.
        """
        if self.session is None:
            raise ConnectionClosedException("Not connected – call open() first.")
        url = self.base_url.rstrip("/") + path
        response = self.session.post(url, json=data or {}, timeout=self.timeout)
        response.raise_for_status()
        return response.json()

    # ------------------------------------------------------------------
    # NAPALM getters
    # ------------------------------------------------------------------

    def get_facts(self) -> dict[str, Any]:
        """Return a dictionary of general device facts.

        Calls ``GET /api/core/system/status``.

        Returned keys (NAPALM standard):
        ``vendor``, ``model``, ``hostname``, ``fqdn``, ``os_version``,
        ``serial_number``, ``uptime``, ``interface_list``.
        """
        status = self._get("/api/core/system/status")

        hostname = status.get("hostname") or status.get("name") or ""
        version = status.get("version") or status.get("product_version") or "unknown"

        try:
            interface_list = list(self.get_interfaces().keys())
        except Exception as exc:
            logger.debug("Failed to fetch interface list: %s", exc)
            interface_list = []

        return {
            "vendor": self.VENDOR,
            "model": status.get("model") or self.VENDOR,
            "hostname": hostname,
            "fqdn": hostname,
            "os_version": version,
            "serial_number": status.get("serial") or "",
            "uptime": status.get("uptime", -1),
            "interface_list": interface_list,
        }

    def get_interfaces(self) -> dict[str, dict[str, Any]]:
        """Return interface details keyed by interface name.

        Calls ``GET /api/interfaces/overview/export``.

        Each entry contains NAPALM standard keys:
        ``is_up``, ``is_enabled``, ``description``, ``last_flapped``,
        ``mac_address``, ``speed``, ``mtu``.

        Plus one non-standard key, ``identifier``: OPNsense's *assigned*
        interface name ("lan", "opt1"), as opposed to the physical device the
        dict is keyed by ("em0"). Empty for unassigned interfaces. This is the
        name :meth:`send_wake_on_lan` requires — see the note there.
        """
        data = self._get("/api/interfaces/overview/export")
        interfaces: dict[str, dict[str, Any]] = {}

        # API returns a bare list in newer OPNsense versions;
        # older/wrapped format uses {"interfaces": [...]}
        items: list = data if isinstance(data, list) else data.get("interfaces", [])

        for iface in items:
            name = iface.get("device") or iface.get("name", "")
            if not name:
                continue
            interfaces[name] = {
                "is_up": iface.get("status", "") == "up" if "status" in iface else bool(iface.get("up", False)),
                "is_enabled": bool(iface.get("enabled", True)),
                "description": iface.get("description") or iface.get("descr") or "",
                "last_flapped": -1.0,
                "mac_address": (iface.get("macaddr") or iface.get("mac") or "").lower(),
                "speed": float(iface["speed_mbps"]) if iface.get("speed_mbps") else 0.0,
                "mtu": int(iface["mtu"]) if iface.get("mtu") else 0,
                "identifier": iface.get("identifier") or "",
            }

        return interfaces

    def get_interfaces_ip(self) -> dict[str, dict[str, Any]]:
        """Return IP addresses grouped by interface name.

        Uses ``GET /api/interfaces/overview/export`` (same source as get_interfaces).

        Structure::

            {
                "em0": {
                    "ipv4": {"192.0.2.10": {"prefix_length": 24}},
                    "ipv6": {"2001:db8::1": {"prefix_length": 64}},
                }
            }
        """
        import ipaddress

        data = self._get("/api/interfaces/overview/export")
        items: list = data if isinstance(data, list) else data.get("interfaces", [])
        result: dict[str, dict[str, Any]] = {}

        def _add(ifname: str, cidr: str, family: str) -> None:
            try:
                iface_obj = ipaddress.ip_interface(cidr)
                ip = str(iface_obj.ip)
                prefix = iface_obj.network.prefixlen
                result.setdefault(ifname, {"ipv4": {}, "ipv6": {}})
                result[ifname][family][ip] = {"prefix_length": prefix}
            except ValueError:
                pass

        for iface in items:
            name: str = iface.get("device") or iface.get("name", "")
            if not name:
                continue
            addr4: str = iface.get("addr4", "")
            addr6: str = iface.get("addr6", "")
            if addr4:
                _add(name, addr4, "ipv4")
            if addr6:
                _add(name, addr6, "ipv6")
            if not addr4:
                for entry in iface.get("ipv4") or []:
                    ip_field = entry.get("ipaddr") or entry.get("ip", "")
                    subnet = entry.get("subnetbits") or entry.get("prefix_length")
                    cidr = f"{ip_field}/{subnet}" if subnet and "/" not in ip_field else ip_field
                    if cidr:
                        _add(name, cidr, "ipv4")
            if not addr6:
                for entry in iface.get("ipv6") or []:
                    ip_field = entry.get("ipaddr") or entry.get("ip", "")
                    prefix = entry.get("prefixlen") or entry.get("prefix_length")
                    cidr = f"{ip_field}/{prefix}" if prefix and "/" not in ip_field else ip_field
                    if cidr:
                        _add(name, cidr, "ipv6")

        return result

    def get_networks(self) -> list[dict[str, Any]]:
        """Return the IP networks this firewall is authoritative for.

        Derived from the interface overview (``GET /api/interfaces/overview/export``).
        Loopback and link-local addresses are excluded.

        Each entry::

            {
                "network": "192.168.1.0/24",
                "interface": "em0",
                "gateway": "192.168.1.1",
                "family": "ipv4",
                "prefix_length": 24,
                "vlan_id": 20,
            }

        ``vlan_id`` is ``None`` for interfaces that are not 802.1Q VLAN
        sub-interfaces (e.g. the untagged management/LAN interface).
        """
        import ipaddress

        data = self._get("/api/interfaces/overview/export")
        items: list = data if isinstance(data, list) else data.get("interfaces", [])

        networks: list[dict[str, Any]] = []

        def _add(ifname: str, cidr: str, family: str, vlan_id: int | None) -> None:
            """Parse a CIDR string (e.g. '10.0.0.1/24') and append to networks."""
            try:
                iface_obj = ipaddress.ip_interface(cidr)
                net = iface_obj.network
                if net.is_loopback or net.is_link_local:
                    return
                networks.append({
                    "network": str(net),
                    "interface": ifname,
                    "gateway": str(iface_obj.ip),
                    "family": family,
                    "prefix_length": net.prefixlen,
                    "vlan_id": vlan_id,
                })
            except ValueError:
                pass

        for iface in items:
            name: str = iface.get("device") or iface.get("name", "")
            if not name:
                continue

            vlan_id: int | None = None
            raw_tag = iface.get("vlan_tag")
            if raw_tag:
                try:
                    vlan_id = int(raw_tag)
                except (TypeError, ValueError):
                    vlan_id = None

            # Primary format: flat CIDR strings in addr4/addr6
            addr4: str = iface.get("addr4", "")
            addr6: str = iface.get("addr6", "")
            if addr4:
                _add(name, addr4, "ipv4", vlan_id)
            if addr6:
                _add(name, addr6, "ipv6", vlan_id)

            # Fallback: ipv4/ipv6 arrays where ipaddr may include prefix
            if not addr4:
                for entry in iface.get("ipv4") or []:
                    ip_field = entry.get("ipaddr") or entry.get("ip", "")
                    subnet = entry.get("subnetbits") or entry.get("prefix_length")
                    cidr = f"{ip_field}/{subnet}" if subnet and "/" not in ip_field else ip_field
                    if cidr:
                        _add(name, cidr, "ipv4", vlan_id)
            if not addr6:
                for entry in iface.get("ipv6") or []:
                    ip_field = entry.get("ipaddr") or entry.get("ip", "")
                    prefix = entry.get("prefixlen") or entry.get("prefix_length")
                    cidr = f"{ip_field}/{prefix}" if prefix and "/" not in ip_field else ip_field
                    if cidr:
                        _add(name, cidr, "ipv6", vlan_id)

        return networks

    def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
        """Return the ARP table.

        Calls ``GET /api/diagnostics/interface/get_arp``.

        Each entry contains: ``interface``, ``mac``, ``ip``, ``age``.
        """
        data = self._get("/api/diagnostics/interface/get_arp")
        arp_table: list[dict[str, Any]] = []

        for entry in data if isinstance(data, list) else data.get("arp", []):
            arp_table.append(
                {
                    "interface": entry.get("intf") or entry.get("interface", ""),
                    "mac": (entry.get("mac") or "").lower(),
                    "ip": entry.get("ip") or entry.get("address", ""),
                    "age": float(entry.get("expires", 0)),
                }
            )

        return arp_table

    def get_interfaces_counters(self) -> dict[str, dict[str, Any]]:
        """Return per-interface packet and byte counters.

        Calls ``GET /api/diagnostics/interface/get_interface_statistics``.

        Each entry contains NAPALM standard keys:
        ``tx_errors``, ``rx_errors``, ``tx_discards``, ``rx_discards``,
        ``tx_octets``, ``rx_octets``,
        ``tx_unicast_packets``, ``rx_unicast_packets``,
        ``tx_multicast_packets``, ``rx_multicast_packets``,
        ``tx_broadcast_packets``, ``rx_broadcast_packets``.
        """
        data = self._get("/api/diagnostics/interface/get_interface_statistics")
        counters: dict[str, dict[str, Any]] = {}

        for iface, stats in data.get("statistics", {}).items():
            counters[iface] = {
                "tx_errors": int(stats.get("output-errors", 0)),
                "rx_errors": int(stats.get("input-errors", 0)),
                "tx_discards": int(stats.get("output-drops", 0)),
                "rx_discards": int(stats.get("input-drops", 0)),
                "tx_octets": int(stats.get("output-bytes", 0)),
                "rx_octets": int(stats.get("input-bytes", 0)),
                "tx_unicast_packets": int(stats.get("output-packets", 0)),
                "rx_unicast_packets": int(stats.get("input-packets", 0)),
                "tx_multicast_packets": int(stats.get("output-multicasts", 0)),
                "rx_multicast_packets": int(stats.get("input-multicasts", 0)),
                "tx_broadcast_packets": int(stats.get("output-broadcasts", 0)),
                "rx_broadcast_packets": int(stats.get("input-broadcasts", 0)),
            }

        return counters

    def get_environment(self) -> dict[str, Any]:
        """Return device environment data (CPU, memory, temperature).

        Calls:
        - ``GET /api/diagnostics/system/system_resources`` for CPU and memory.
        - ``GET /api/diagnostics/system/system_temperature`` for temperature sensors.

        ``fan`` and ``power`` fields are not exposed by OPNsense and are
        returned with assumed-healthy placeholder values.
        """
        resources = self._get("/api/diagnostics/system/system_resources")

        cpu_pct = float(resources.get("cpu", {}).get("used", 0))
        mem_total = int(resources.get("memory", {}).get("total", 0) or 0)
        mem_used = int(resources.get("memory", {}).get("used", 0) or 0)

        try:
            temp_data = self._get("/api/diagnostics/system/system_temperature")
        except Exception as exc:
            logger.debug("Failed to fetch system temperature: %s", exc)
            temp_data = {}

        temperature: dict[str, Any] = {}
        for sensor in temp_data.get("data", []):
            name = sensor.get("device") or sensor.get("name", "")
            temp_val = float(sensor.get("temperature", 0))
            temperature[name] = {
                "temperature": temp_val,
                "is_alert": temp_val > 80.0,
                "is_critical": temp_val > 95.0,
            }

        return {
            "fans": {},
            "temperature": temperature,
            "power": {},
            "cpu": {0: {"%usage": cpu_pct}},
            "memory": {
                "available_ram": mem_total - mem_used,
                "used_ram": mem_used,
            },
        }

    def get_route_to(
        self,
        destination: str = "",
        protocol: str = "",
        longer: bool = False,
    ) -> dict[str, list[dict[str, Any]]]:
        """Return routing table entries.

        Calls ``GET /api/diagnostics/interface/get_routes``.

        :param destination: Filter by exact prefix (e.g. ``"192.0.2.0/24"``).
        :param protocol: Filter by protocol name (``"static"``, ``"connected"``).
        :param longer: Ignored (OPNsense does not support longer-prefixes filter).

        Returns a NAPALM-standard route dict keyed by network prefix.
        """
        data = self._get("/api/diagnostics/interface/get_routes")
        routes: dict[str, list[dict[str, Any]]] = {}

        # Optionally enrich with OSPF routes from FRR/Quagga
        ospf_networks: set = set()
        try:
            ospf_data = self._get("/api/quagga/ospf/routes")
            for prefix in (ospf_data if isinstance(ospf_data, list) else ospf_data.get("routes", {}).keys()):
                ospf_networks.add(str(prefix))
        except Exception as exc:
            logger.debug("Failed to fetch OSPF routes: %s", exc)

        route_list = data if isinstance(data, list) else data.get("route", [])
        for route in route_list:
            network = route.get("network") or route.get("destination", "")
            if not network:
                continue

            if destination and network != destination:
                continue

            flags = route.get("flags", "").upper()
            gateway = route.get("gateway") or route.get("nexthop", "")
            iface = route.get("netif") or route.get("interface", "")

            # Clean up BSD link-layer gateway references
            clean_gateway = "" if (not gateway or gateway.startswith("link#") or gateway == "0.0.0.0") else gateway

            # Determine address family from network address
            family = "ipv6" if (":" in network or (gateway and ":" in gateway)) else "ipv4"

            # Determine routing protocol from BSD flags:
            # S = Static, dynamic routes have no S flag
            if network in ospf_networks:
                proto = "ospf"
            elif "S" in flags:
                proto = "static"
            elif not clean_gateway:
                proto = "connected"
            else:
                proto = "kernel"

            if protocol and proto != protocol.lower():
                continue

            entry: dict[str, Any] = {
                "protocol": proto,
                "family": family,
                "current_active": "U" in flags,
                "last_active": False,
                "age": -1,
                "next_hop": clean_gateway,
                "outgoing_interface": iface,
                "selected_next_hop": True,
                "preference": int(route.get("priority", 0)),
                "inactive_reason": "",
                "routing_table": "global",
                "protocol_attributes": {},
            }
            routes.setdefault(network, []).append(entry)

        return routes

    def get_ipv6_neighbors_table(self) -> list[dict[str, Any]]:
        """Return the IPv6 Neighbor Discovery (NDP) table.

        Calls ``GET /api/diagnostics/interface/get_ndp``.

        Each entry contains: ``interface``, ``mac``, ``ip``, ``age``,
        ``state`` (best-effort from NDP flags).
        """
        data = self._get("/api/diagnostics/interface/get_ndp")
        neighbors: list[dict[str, Any]] = []

        rows = data if isinstance(data, list) else data.get("rows", [])
        for entry in rows:
            neighbors.append(
                {
                    "interface": entry.get("intf") or entry.get("interface", ""),
                    "mac": (entry.get("mac") or "").lower(),
                    "ip": entry.get("ip") or entry.get("address", ""),
                    "age": float(entry.get("expires", 0)),
                    "state": entry.get("state", ""),
                }
            )

        return neighbors

    def get_lldp_neighbors(self) -> dict[str, list[dict[str, Any]]]:
        """Return LLDP neighbors grouped by local port.

        Calls ``GET /api/lldpd/service/neighbor``.

        Requires the ``os-lldpd`` plugin to be installed on OPNsense.
        Returns an empty dict if the plugin is not present.
        """
        try:
            data = self._get("/api/lldpd/service/neighbor")
        except Exception as exc:
            logger.debug("LLDP neighbor fetch failed (plugin not installed?): %s", exc)
            return {}

        neighbors: dict[str, list[dict[str, Any]]] = {}
        for row in data.get("rows", []):
            port = row.get("local_port") or row.get("port", "")
            neighbors.setdefault(port, []).append(
                {
                    "hostname": row.get("system_name") or row.get("chassis", ""),
                    "port": row.get("port_id") or row.get("port", ""),
                }
            )
        return neighbors

    def get_lldp_neighbors_detail(
        self, interface: str = ""
    ) -> dict[str, list[dict[str, Any]]]:
        """Return detailed LLDP neighbor information.

        Calls ``GET /api/lldpd/service/neighbor``.

        Requires the ``os-lldpd`` plugin.  Returns an empty dict if the
        plugin is not available.

        :param interface: If set, filter results to this local port.
        """
        try:
            data = self._get("/api/lldpd/service/neighbor")
        except Exception as exc:
            logger.debug("LLDP neighbor detail fetch failed (plugin not installed?): %s", exc)
            return {}

        details: dict[str, list[dict[str, Any]]] = {}
        for row in data.get("rows", []):
            port = row.get("local_port") or row.get("port", "")
            if interface and port != interface:
                continue
            details.setdefault(port, []).append(
                {
                    "parent_interface": "",
                    "remote_port": row.get("port_id") or row.get("port", ""),
                    "remote_port_description": row.get("port_description", ""),
                    "remote_chassis_id": row.get("chassis_id") or row.get("chassis", ""),
                    "remote_system_name": row.get("system_name", ""),
                    "remote_system_description": row.get("system_description", ""),
                    "remote_system_capab": [
                        c.strip().lower()
                        for c in row.get("system_capabilities", "").split(",")
                        if c.strip()
                    ],
                    "remote_system_enable_capab": [
                        c.strip().lower()
                        for c in row.get("enabled_capabilities", "").split(",")
                        if c.strip()
                    ],
                }
            )
        return details

    def get_ntp_servers(self) -> dict[str, dict[str, Any]]:
        """Return configured NTP servers.

        Calls ``GET /api/ntpd/service/status`` which includes the list of
        configured peer addresses in the ``peers`` field.

        Returns a dict keyed by server address with an empty value dict
        (NAPALM standard format).
        """
        try:
            data = self._get("/api/ntpd/service/status")
        except Exception as exc:
            logger.debug("NTP status fetch failed (plugin not installed?): %s", exc)
            return {}

        servers: dict[str, dict[str, Any]] = {}
        for peer in data.get("peers", []):
            addr = peer.get("address") or peer.get("remote", "")
            if addr:
                servers[addr] = {}
        return servers

    def get_vlans(self) -> dict[str, dict[str, Any]]:
        """Return configured VLANs.

        Calls ``GET /api/interfaces/vlan_settings/search_item``.

        Each VLAN device configured under *Interfaces → Other Types → VLAN*
        becomes one entry, keyed by VLAN tag (as a string).

        The ``interfaces`` list contains the VLAN device name (e.g.
        ``em0_vlan10``).  When a device is already assigned to a logical
        interface OPNsense appends the logical name in brackets
        (``"em0_vlan10 [LAN]"``); this driver strips that annotation and
        stores the bare device name.

        Returned keys (NAPALM standard):
        ``name`` (description, or device name if no description is set),
        ``interfaces`` (list with the VLAN device name).
        """
        data = self._get("/api/interfaces/vlan_settings/search_item")
        vlans: dict[str, dict[str, Any]] = {}

        for row in data.get("rows", []):
            tag = str(row.get("tag", "")).strip()
            if not tag:
                continue

            # vlanif may be "em0_vlan10 [LAN]" when assigned to a logical interface
            vlanif_raw = str(row.get("vlanif", ""))
            vlanif = vlanif_raw.split(" [")[0].strip()

            descr = str(row.get("descr", "")).strip()
            vlans[tag] = {
                "name": descr or vlanif,
                "interfaces": [vlanif] if vlanif else [],
            }

        return vlans

    def get_bgp_neighbors(self) -> dict[str, Any]:
        """Return BGP neighbor state.

        Requires the FRR plugin (``os-frr``) to be installed on OPNsense.
        Returns an empty dict if the plugin is absent or FRR is not running.

        Calls:

        - ``GET /api/quagga/bgp/get`` — local AS number and router-id from
          the BGP configuration model.
        - ``GET /api/quagga/diagnostics/bgpneighbors`` — live neighbor state
          as returned by FRR (``vtysh -c "show bgp neighbors json"``).

        Returns a NAPALM-standard dict keyed by VRF name.  Only the default
        VRF (``"global"``) is populated; per-VRF BGP is not yet mapped.

        Each peer entry contains:
        ``local_as``, ``remote_as``, ``remote_id``, ``is_up``,
        ``is_enabled``, ``description``, ``uptime``,
        ``address_family`` (``ipv4`` and/or ``ipv6`` with prefix counters).
        """
        try:
            bgp_cfg = self._get("/api/quagga/bgp/get")
            neighbors_data = self._get("/api/quagga/diagnostics/bgpneighbors")
        except Exception as exc:
            logger.debug("BGP config fetch failed (plugin not installed?): %s", exc)
            return {}

        bgp = bgp_cfg.get("bgp", {})
        local_as_default = int(bgp.get("asnumber", 0) or 0)
        router_id = bgp.get("routerid", "")

        raw_neighbors = neighbors_data.get("response", {})
        if not isinstance(raw_neighbors, dict):
            return {}

        # FRR address-family key → NAPALM address-family key
        _AF_MAP = {
            "ipv4Unicast": "ipv4",
            "ipv6Unicast": "ipv6",
        }

        peers: dict[str, Any] = {}
        for peer_ip, nbr in raw_neighbors.items():
            if not isinstance(nbr, dict):
                continue

            is_up = nbr.get("bgpState", "").lower() == "established"
            uptime_msec = int(nbr.get("bgpTimerUpMsec", 0) or 0)
            uptime = uptime_msec // 1000 if is_up else -1

            address_family: dict[str, Any] = {}
            for frr_af, napalm_af in _AF_MAP.items():
                af = nbr.get("addressFamilyInfo", {}).get(frr_af)
                if af is not None:
                    address_family[napalm_af] = {
                        "sent_prefixes": int(af.get("sentPrefixCounter", 0) or 0),
                        "received_prefixes": int(af.get("prefixReceivedCount", 0) or 0),
                        "accepted_prefixes": int(af.get("acceptedPrefixCounter", 0) or 0),
                    }

            if not address_family:
                # FRR did not report AF info (session not yet established)
                address_family["ipv4"] = {
                    "sent_prefixes": -1,
                    "received_prefixes": -1,
                    "accepted_prefixes": -1,
                }

            peers[peer_ip] = {
                "local_as": int(nbr.get("localAs", local_as_default) or local_as_default),
                "remote_as": int(nbr.get("remoteAs", 0) or 0),
                "remote_id": nbr.get("remoteRouterId", ""),
                "is_up": is_up,
                "is_enabled": not bool(nbr.get("adminShutdown", False)),
                "description": nbr.get("nbrDesc", ""),
                "uptime": uptime,
                "address_family": address_family,
            }

        return {
            "global": {
                "router_id": router_id,
                "peers": peers,
            }
        }

    def get_config(
        self,
        retrieve: str = "all",
        full: bool = False,
        sanitized: bool = False,
        format: str = "text",
    ) -> dict[str, str]:
        """Return device configuration.

        OPNsense stores its configuration as XML.  This getter returns the
        raw XML text in the ``running`` slot.  ``startup`` mirrors ``running``
        (OPNsense applies config immediately).  The ``candidate`` slot shows
        the JSON-serialised staged routes when a candidate has been loaded,
        or an empty string otherwise.

        Calls ``GET /api/core/backup/download/this``.
        """
        configs: dict[str, str] = {"running": "", "startup": "", "candidate": ""}

        if retrieve in ("all", "running", "startup"):
            try:
                response = self._get("/api/core/backup/download/this")
                xml_text: str = (
                    response if isinstance(response, str) else str(response)
                )
                if retrieve in ("all", "running"):
                    configs["running"] = xml_text
                if retrieve in ("all", "startup"):
                    configs["startup"] = xml_text
            except Exception as exc:
                logger.warning("Failed to fetch running/startup config: %s", exc)

        if retrieve in ("all", "candidate") and self._candidate_config is not None:
            configs["candidate"] = json.dumps(self._candidate_config, indent=2)

        return configs

    # ------------------------------------------------------------------
    # Config management (static routes)
    # ------------------------------------------------------------------

    def load_merge_candidate(self, filename: str | None = None, config: str | None = None) -> None:
        """Stage a set of static-route additions as a candidate config.

        OPNsense does not offer a single generic config-push endpoint.
        This method targets the **routes** subsystem
        (``/api/routes/routes/``) and accepts a JSON list of route objects.

        The *config* parameter must be a JSON string containing a list of
        route dicts.  Each dict may contain the following keys:

        .. code-block:: json

            [
              {
                "network": "10.0.0.0/8",
                "gateway": "WAN_GW",
                "descr": "optional description",
                "disabled": "0"
              }
            ]

        ``gateway`` must be the **name** of an existing OPNsense gateway
        (not an IP address) as shown in
        *System → Gateways → Configuration*.

        :param filename: Path to a JSON file containing the route list.
        :param config: JSON string containing the route list.
        :raises MergeConfigException: if neither or both arguments are given,
            or if the JSON is malformed.
        """
        if filename is None and config is None:
            raise MergeConfigException("Provide either 'filename' or 'config'.")
        if filename is not None and config is not None:
            raise MergeConfigException("Provide either 'filename' or 'config', not both.")

        if filename is not None:
            try:
                with open(filename, "r", encoding="utf-8") as fh:
                    config = fh.read()
            except OSError as exc:
                raise MergeConfigException(f"Cannot read file {filename!r}: {exc}") from exc

        try:
            routes = json.loads(config)  # type: ignore[arg-type]
        except json.JSONDecodeError as exc:
            raise MergeConfigException(f"Invalid JSON in candidate config: {exc}") from exc

        if not isinstance(routes, list):
            raise MergeConfigException(
                "Candidate config must be a JSON array of route objects."
            )
        for i, route in enumerate(routes):
            if not isinstance(route, dict):
                raise MergeConfigException(f"Route at index {i} must be a JSON object.")
            if "network" not in route or "gateway" not in route:
                raise MergeConfigException(
                    f"Route at index {i} is missing 'network' or 'gateway'."
                )

        self._candidate_config = routes

    def compare_config(self) -> str:
        """Return a unified diff of the candidate vs the current routes.

        Fetches the live route list from ``GET /api/routes/routes/searchroute``
        and diffs it against the staged candidate.

        Returns an empty string if no candidate has been loaded.
        """
        if self._candidate_config is None:
            return ""

        current_routes = self._fetch_current_routes()
        current_text = json.dumps(current_routes, indent=2, sort_keys=True)
        candidate_text = json.dumps(self._candidate_config, indent=2, sort_keys=True)

        diff = difflib.unified_diff(
            current_text.splitlines(keepends=True),
            candidate_text.splitlines(keepends=True),
            fromfile="current",
            tofile="candidate",
        )
        return "".join(diff)

    def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
        """Apply the staged candidate routes to the device.

        Each route in the candidate is submitted via
        ``POST /api/routes/routes/addroute``.  After all routes are added,
        ``POST /api/routes/routes/reconfigure`` is called to activate them.

        The UUIDs returned by OPNsense are stored internally so that
        :meth:`rollback` can remove exactly these routes.

        :param message: Ignored (OPNsense has no commit-message concept).
        :param revert_in: Ignored (auto-rollback not supported via API).
        :raises MergeConfigException: if no candidate is staged, or if the
            API returns an error for any route.
        """
        if self._candidate_config is None:
            raise MergeConfigException("No candidate config loaded.  Call load_merge_candidate() first.")

        # Record the current backup ID so rollback() can restore exactly
        # this state after OPNsense writes the new config to disk.
        self._pre_commit_backup_id = self._get_latest_backup_id()

        try:
            for route in self._candidate_config:
                payload = {
                    "route": {
                        "network": route["network"],
                        "gateway": route["gateway"],
                        "descr": route.get("descr", ""),
                        "disabled": route.get("disabled", "0"),
                    }
                }
                self._post("/api/routes/routes/addroute", payload)

            self._post("/api/routes/routes/reconfigure")
        except RequestException as exc:
            raise MergeConfigException(f"Failed to apply route config: {exc}") from exc

        self._candidate_config = None

    def discard_config(self) -> None:
        """Discard the staged candidate config without applying it."""
        self._candidate_config = None

    def rollback(self) -> None:
        """Revert the device to the configuration state captured before the last commit.

        OPNsense automatically saves a backup of ``config.xml`` before applying
        configuration changes.  :meth:`commit_config` records the ID of the
        most recent backup at the time of the commit so that this method can
        restore exactly the pre-commit state via
        ``POST /api/core/backup/revert_backup/{backup_id}``.

        If no backup ID is available (i.e. :meth:`commit_config` was never
        called in this session, or the backup list was empty at commit time),
        the most recent backup from the device is used as a fallback.

        If no backups exist at all, this is a no-op.
        """
        backup_id = self._pre_commit_backup_id or self._get_latest_backup_id()
        if backup_id is None:
            return

        self._post(f"/api/core/backup/revert_backup/{backup_id}")
        self._pre_commit_backup_id = None

    # ------------------------------------------------------------------
    # Internal helpers (config management)
    # ------------------------------------------------------------------

    def _get_latest_backup_id(self) -> str | None:
        """Return the ID of the most recent server-side config backup, or ``None``.

        Calls ``GET /api/core/backup/backups/this``.  The response is sorted
        newest-first by OPNsense.  Returns ``None`` when no backups exist or
        the request fails.
        """
        try:
            data = self._get("/api/core/backup/backups/this")
            items = data.get("items", [])
            return items[0]["id"] if items else None
        except Exception as exc:
            logger.debug("No backup snapshots found: %s", exc)
            return None

    def _fetch_current_routes(self) -> list[dict[str, Any]]:
        """Return the current static routes from the OPNsense API.

        Calls ``GET /api/routes/routes/searchroute`` and normalises the
        response to the same keys used by :meth:`load_merge_candidate`.
        """
        data = self._get("/api/routes/routes/searchroute")
        routes: list[dict[str, Any]] = []
        for row in data.get("rows", []):
            routes.append(
                {
                    "network": row.get("network", ""),
                    "gateway": row.get("gateway", ""),
                    "descr": row.get("descr", ""),
                    "disabled": row.get("disabled", "0"),
                }
            )
        return routes

    # ------------------------------------------------------------------
    # NetOrch extensions: packages, services, updates
    # ------------------------------------------------------------------

    def get_packages(self) -> list[dict[str, Any]]:
        """Return installed OPNsense plugins.

        Calls ``GET /api/core/firmware/info`` and returns the ``plugin``
        list filtered to entries where ``installed == "1"``.  The format
        mirrors the OpenWrt NAPALM driver so NetOrch can render both
        drivers with the same UI component:
        ``{name, version, installed, description, size, source}``
        """
        info = self._get("/api/core/firmware/info")
        result: list[dict[str, Any]] = []
        for p in info.get("plugin", []):
            if p.get("installed") != "1":
                continue
            result.append({
                "name": p.get("name", ""),
                "version": p.get("version", ""),
                "installed": True,
                "description": p.get("comment", ""),
                "size": 0,
                "source": "opnsense-plugins",
            })
        return sorted(result, key=lambda x: x["name"].lower())

    def get_certificates(self) -> list[dict[str, Any]]:
        """Return certificates from the OPNsense Trust store.

        Calls ``POST /api/trust/cert/search`` and normalises each row to
        ``{name, issuer, valid_from, valid_to, in_use_by}`` (Unix
        timestamps for the validity fields). Never includes
        crt_payload/prv_payload/csr_payload -- those rows carry private key
        material and must not leave the Trust store.
        """
        try:
            data = self._post("/api/trust/cert/search")
        except Exception as exc:
            logger.warning("get_certificates() failed: %s", exc)
            return []

        def _as_int(value: Any) -> int:
            try:
                return int(value or 0)
            except (TypeError, ValueError):
                return 0

        result: list[dict[str, Any]] = []
        for row in data.get("rows", []):
            result.append({
                "name": row.get("commonname") or row.get("descr") or row.get("refid", ""),
                "issuer": row.get("%caref") or "",
                "valid_from": _as_int(row.get("valid_from")),
                "valid_to": _as_int(row.get("valid_to")),
                "in_use_by": _as_int(row.get("in_use")),
            })
        return result

    def get_ddns_status(self) -> dict[str, Any] | None:
        """Return Dynamic DNS (os-ddclient) enabled/running state.

        Calls ``GET /api/dyndns/service/status`` and
        ``GET /api/dyndns/settings/get`` -- note the API module is
        "dyndns", not "ddclient" (the service id reported by
        get_services() is "ddclient", but its REST controller lives under
        a different, historical module name).

        Returns ``{"enabled": bool, "running": bool}``, or ``None`` if the
        plugin isn't installed/reachable. Scoped to enabled/running only:
        no per-account "registered IP" comparison, since that would need a
        configured account to verify the response shape against.
        """
        try:
            status = self._get("/api/dyndns/service/status")
            settings = self._get("/api/dyndns/settings/get")
        except Exception as exc:
            logger.warning("get_ddns_status() failed: %s", exc)
            return None

        general = settings.get("ddclient", {}).get("general", {})
        return {
            "enabled": general.get("enabled") == "1",
            "running": status.get("status") == "running",
        }

    def get_radius_clients(self) -> list[dict[str, Any]]:
        """Return configured FreeRADIUS NAS clients.

        Calls ``GET /api/freeradius/client/search_client``.
        """
        try:
            data = self._get("/api/freeradius/client/search_client")
        except Exception as exc:
            logger.warning("get_radius_clients() failed: %s", exc)
            return []
        return [
            {
                "id": row.get("uuid", ""),
                "name": row.get("name", ""),
                "ip": row.get("ip", ""),
                "enabled": row.get("enabled") == "1",
            }
            for row in data.get("rows", [])
        ]

    def create_radius_client(self, name: str, ip: str, secret: str) -> dict[str, Any]:
        """Create a FreeRADIUS NAS client and apply the change.

        Calls ``POST /api/freeradius/client/add_client``, then
        ``POST /api/freeradius/service/reconfigure`` to apply -- a saved
        client has no effect on the running radiusd until reconfigured.
        The add response carries no id, so the new client's uuid is looked
        up afterwards via get_radius_clients() (matched by name) -- callers
        need it to address this client in later set_client/del_client calls.
        """
        payload = {"client": {"name": name, "ip": ip, "secret": secret}}
        result = self._post("/api/freeradius/client/add_client", payload)
        if result.get("result") != "saved":
            return {"success": False, "validations": result.get("validations", {})}
        self._post("/api/freeradius/service/reconfigure")
        created = next((c for c in self.get_radius_clients() if c["name"] == name), None)
        return {"success": True, "id": created["id"] if created else None}

    def delete_radius_client(self, client_id: str) -> dict[str, Any]:
        """Delete a FreeRADIUS NAS client by uuid and apply the change."""
        result = self._post(f"/api/freeradius/client/del_client/{client_id}")
        if result.get("result") != "deleted":
            return {"success": False}
        self._post("/api/freeradius/service/reconfigure")
        return {"success": True}

    def get_radius_users(self) -> list[dict[str, Any]]:
        """Return configured FreeRADIUS users.

        Calls ``GET /api/freeradius/user/search_user``. Never includes the
        password field.
        """
        try:
            data = self._get("/api/freeradius/user/search_user")
        except Exception as exc:
            logger.warning("get_radius_users() failed: %s", exc)
            return []
        return [
            {
                "id": row.get("uuid", ""),
                "username": row.get("username", ""),
                "enabled": row.get("enabled") == "1",
            }
            for row in data.get("rows", [])
        ]

    def create_radius_user(self, username: str, password: str) -> dict[str, Any]:
        """Create a FreeRADIUS user and apply the change.

        Calls ``POST /api/freeradius/user/add_user``, then
        ``POST /api/freeradius/service/reconfigure`` to apply. The add
        response carries no id, so the new user's uuid is looked up
        afterwards via get_radius_users() (matched by username) -- callers
        need it to address this user in later set_user/del_user calls.
        """
        payload = {"user": {"username": username, "password": password}}
        result = self._post("/api/freeradius/user/add_user", payload)
        if result.get("result") != "saved":
            return {"success": False, "validations": result.get("validations", {})}
        self._post("/api/freeradius/service/reconfigure")
        created = next((u for u in self.get_radius_users() if u["username"] == username), None)
        return {"success": True, "id": created["id"] if created else None}

    def delete_radius_user(self, user_id: str) -> dict[str, Any]:
        """Delete a FreeRADIUS user by uuid and apply the change."""
        result = self._post(f"/api/freeradius/user/del_user/{user_id}")
        if result.get("result") != "deleted":
            return {"success": False}
        self._post("/api/freeradius/service/reconfigure")
        return {"success": True}

    def get_dhcp_leases(self) -> list[dict[str, Any]]:
        """Return active DHCP leases from OPNsense.

        Tries all known DHCP backends in order:

        1. **Kea DHCPv4** (os-kea plugin) — ``GET /api/kea/leases4/search``
           Fields: ``hw-address``, ``ip-address``, ``hostname``, ``expire``
        2. **ISC DHCP** (legacy) — ``POST /api/dhcpv4/leases/searchlease``
           Fields: ``mac``, ``address``, ``hostname``, ``ends``
        3. **ARP table** fallback — ``GET /api/diagnostics/interface/get_arp``
           Provides IP only, hostname will be empty.

        Each returned entry contains:

        * ``mac``       — lower-case MAC address
        * ``ip``        — assigned IP address
        * ``hostname``  — client hostname (may be empty)
        * ``lease_end`` — Unix timestamp when the lease expires (0 if unknown)
        * ``state``     — raw state string
        """
        def _parse_kea(rows: list[dict]) -> list[dict[str, Any]]:
            result = []
            for row in rows:
                # OPNsense Kea uses "hwaddr"; standard Kea uses "hw-address"; ISC DHCP uses "mac"
                mac = (
                    row.get("hwaddr") or row.get("hw-address") or row.get("mac") or ""
                ).lower().strip()
                if not mac:
                    continue
                raw_end = row.get("expire") or row.get("ends") or 0
                try:
                    lease_end = int(raw_end)
                except (ValueError, TypeError):
                    lease_end = 0
                result.append({
                    "mac": mac,
                    "ip": (row.get("address") or row.get("ip-address") or row.get("ip") or "").strip(),
                    "hostname": (row.get("hostname") or "").strip(),
                    "lease_end": lease_end,
                    "state": str(row.get("state") or ""),
                })
            return result

        # 1. Kea DHCPv4 plugin
        try:
            data = self._get("/api/kea/leases4/search")
            rows = data.get("rows") or data.get("leases") or (data if isinstance(data, list) else [])
            leases = _parse_kea(rows)
            if leases:
                return leases
        except Exception as exc:
            logger.debug("Kea DHCP lease fetch failed: %s", exc)

        # 2. ISC DHCP (legacy)
        try:
            data = self._post(
                "/api/dhcpv4/leases/searchlease",
                {"current": 1, "rowCount": -1, "searchPhrase": "", "sort": {}},
            )
            leases = _parse_kea(data.get("rows", []))
            if leases:
                return leases
        except Exception as exc:
            logger.debug("ISC DHCP lease fetch failed: %s", exc)

        # 3. ARP table fallback (IP only, no hostname)
        try:
            arp = self.get_arp_table()
            return [
                {"mac": e["mac"], "ip": e["ip"], "hostname": "", "lease_end": 0, "state": "arp"}
                for e in arp
                if e.get("mac") and e.get("ip")
            ]
        except Exception as exc:
            logger.warning("ARP table fallback failed: %s", exc)
            return []

    def create_dhcp_reservation(self, mac: str, ip: str, hostname: str = "") -> None:
        """Create (or update) a Kea DHCPv4 static reservation (MAC → IP).

        Only the Kea backend (os-kea plugin) is supported — this driver has
        no active OPNsense environment with legacy ISC DHCP to verify a
        second code path against, unlike ``get_dhcp_leases()``'s read-only
        try-then-fallback. Callers should treat a missing/disabled Kea
        plugin as "reservations unsupported" (``RuntimeError``), not fall
        back to plain DHCP silently.

        :param mac: NIC MAC address (any common formatting; sent as-is to
            Kea's ``hw_address`` field).
        :param ip: IP address to reserve; must fall inside a subnet Kea
            already manages (``searchSubnet``), or this raises ValueError.
        :param hostname: optional hostname to record on the reservation.
        :raises ValueError: if no Kea subnet contains ``ip``.
        :raises RuntimeError: if Kea rejects the reservation (validation
            errors) or the Kea plugin isn't installed/enabled.
        """
        try:
            subnets = self._get("/api/kea/dhcpv4/searchSubnet").get("rows") or []
        except Exception as exc:
            raise RuntimeError(f"Kea DHCPv4 plugin unavailable: {exc}") from exc

        ip_obj = ip_address(ip)
        subnet_uuid = None
        for row in subnets:
            try:
                if ip_obj in ip_network(row["subnet"], strict=False):
                    subnet_uuid = row["uuid"]
                    break
            except ValueError:
                continue
        if subnet_uuid is None:
            raise ValueError(f"No Kea-managed subnet contains {ip}")

        # Idempotency: reuse an existing reservation for this IP if one
        # already exists (e.g. a retried provisioning job), rather than
        # creating a duplicate Kea rejects anyway.
        existing_uuid = None
        try:
            existing = self._post(
                "/api/kea/dhcpv4/searchReservation",
                {"current": 1, "rowCount": -1, "searchPhrase": ip},
            )
            for row in existing.get("rows") or []:
                if row.get("ip_address") == ip:
                    existing_uuid = row.get("uuid")
                    break
        except Exception as exc:
            logger.debug("Kea reservation search failed, proceeding to add: %s", exc)

        payload = {
            "reservation": {
                "subnet": subnet_uuid,
                "ip_address": ip,
                "hw_address": mac,
                "hostname": hostname,
                "description": self._NETORK_TAG,
            }
        }
        path = (
            f"/api/kea/dhcpv4/setReservation/{existing_uuid}"
            if existing_uuid
            else "/api/kea/dhcpv4/addReservation"
        )
        result = self._post(path, payload)
        if result.get("result") != "saved":
            raise RuntimeError(f"Kea rejected DHCP reservation for {ip}: {result}")

        self._post("/api/kea/service/reconfigure")

    def delete_dhcp_reservation_and_lease(self, mac: str, ip: str) -> dict[str, Any]:
        """Remove a Kea DHCPv4 static reservation and any active lease for (mac, ip).

        Combined per NetOrk's VM-deletion cleanup flow — reservation and
        lease removal are always requested together, so a single driver
        call keeps callers from having to sequence two calls themselves.

        Reservation removal follows the same search-then-act pattern as
        ``create_dhcp_reservation`` (``searchReservation`` -> ``del<X>/{uuid}``
        -> ``service/reconfigure``) and raises on failure, mirroring that
        method's "never silently no-op on the thing the caller explicitly
        asked for" contract.

        Lease removal is best-effort and non-fatal — a failure here just
        means a stale lease record lingers in Kea until its own natural
        cleanup, which is cosmetic, not a functional problem (a deleted
        reservation already prevents the client from getting the same IP
        back). Endpoint verified live against a real OPNsense instance:
        ``LeasesController`` is documented as "Abstract [non-callable]" with
        a ``del_lease($ips=null)`` action
        (https://docs.opnsense.org/development/api/core/kea.html) — the
        concrete, callable route is the ``leases4`` controller (matching
        ``search``, used above), and despite the ``$ips`` parameter name the
        IP is passed as a URL path segment, not a JSON body field — a POST
        body of ``{"ips": [ip]}`` (the natural reading of the signature)
        returns ``{"status": "error", "message": "Missing lease IP
        parameter"}``; only ``POST /api/kea/leases4/del_lease/{ip}`` works.

        :param mac: NIC MAC address of the reservation to remove.
        :param ip: IP address of the reservation/lease to remove.
        :returns: {"reservation_found", "reservation_deleted", "lease_found",
            "lease_deleted"} — all bool.
        :raises RuntimeError: Kea plugin unavailable, or Kea rejects
            deletion of a reservation that does exist.
        """
        result: dict[str, Any] = {
            "reservation_found": False,
            "reservation_deleted": False,
            "lease_found": False,
            "lease_deleted": False,
        }

        # --- Reservation ---
        try:
            existing = self._post(
                "/api/kea/dhcpv4/searchReservation",
                {"current": 1, "rowCount": -1, "searchPhrase": ip},
            )
        except Exception as exc:
            raise RuntimeError(f"Kea DHCPv4 plugin unavailable: {exc}") from exc

        reservation_uuid = None
        for row in existing.get("rows") or []:
            if row.get("ip_address") == ip:
                reservation_uuid = row.get("uuid")
                break

        if reservation_uuid:
            result["reservation_found"] = True
            del_result = self._post(f"/api/kea/dhcpv4/delReservation/{reservation_uuid}")
            if del_result.get("result") != "deleted":
                raise RuntimeError(f"Kea rejected reservation delete for {ip}: {del_result}")
            result["reservation_deleted"] = True
            self._post("/api/kea/service/reconfigure")

        # --- Lease (best-effort) ---
        try:
            leases = self._get("/api/kea/leases4/search")
            rows = leases.get("rows") or leases.get("leases") or []
            if any((row.get("address") or row.get("ip-address")) == ip for row in rows):
                result["lease_found"] = True
                del_lease = self._post(f"/api/kea/leases4/del_lease/{ip}")
                result["lease_deleted"] = bool(
                    del_lease.get("result") == "deleted" or del_lease.get("status") == "ok"
                )
        except Exception as exc:
            logger.warning("DHCP lease delete for %s failed (non-fatal): %s", ip, exc)

        return result

    # ------------------------------------------------------------------
    # DhcpServerMixin implementation (Kea DHCPv4). The diff and the apply
    # loop are generic and live in napalm_device_types.dhcp; only these
    # three methods know about Kea's REST shape.
    # ------------------------------------------------------------------

    def _kea_subnets(self) -> list[dict[str, Any]]:
        """Return Kea's managed subnets, or raise if the plugin is absent."""
        try:
            return self._get("/api/kea/dhcpv4/searchSubnet").get("rows") or []
        except Exception as exc:
            raise RuntimeError(f"Kea DHCPv4 plugin unavailable: {exc}") from exc

    def _kea_subnet_uuid_for(self, subnet: str, ip: str) -> str:
        """Resolve a reservation's target subnet to Kea's subnet UUID.

        Prefers an exact CIDR match on ``subnet``; falls back to whichever
        managed subnet contains ``ip`` when the caller left ``subnet`` empty.
        """
        rows = self._kea_subnets()

        if subnet:
            for row in rows:
                if str(row.get("subnet") or "").strip() == subnet:
                    return str(row["uuid"])
            raise ValueError(f"No Kea-managed subnet matches {subnet}")

        ip_obj = ip_address(ip)
        for row in rows:
            try:
                if ip_obj in ip_network(row["subnet"], strict=False):
                    return str(row["uuid"])
            except (ValueError, KeyError):
                continue
        raise ValueError(f"No Kea-managed subnet contains {ip}")

    def get_dhcp_reservations(self) -> list[dict[str, Any]]:
        """Return all Kea DHCPv4 static reservations, vendor-neutral.

        This is the *configured* state — ``get_dhcp_leases()`` returns what
        is actually leased. Kea's ``subnet`` field on a reservation is a
        model relation: depending on version it comes back as the related
        subnet's CIDR or as its UUID, so both are accepted and normalised to
        a CIDR here. An unresolvable relation degrades to an empty string
        rather than raising — one orphaned reservation must not make the
        whole inventory unreadable.

        :raises RuntimeError: if the Kea plugin isn't installed/enabled.
        """
        try:
            data = self._post(
                "/api/kea/dhcpv4/searchReservation",
                {"current": 1, "rowCount": -1},
            )
        except Exception as exc:
            raise RuntimeError(f"Kea DHCPv4 plugin unavailable: {exc}") from exc

        rows = data.get("rows") or []
        if not rows:
            return []

        cidr_by_uuid = {
            str(row.get("uuid") or ""): str(row.get("subnet") or "")
            for row in self._kea_subnets()
        }
        known_cidrs = set(cidr_by_uuid.values())

        reservations: list[dict[str, Any]] = []
        for row in rows:
            raw_subnet = str(row.get("subnet") or "").strip()
            if raw_subnet in known_cidrs:
                subnet = raw_subnet
            else:
                subnet = cidr_by_uuid.get(raw_subnet, "")

            reservations.append({
                "uuid": str(row.get("uuid") or ""),
                "mac": str(row.get("hw_address") or ""),
                "ip": str(row.get("ip_address") or ""),
                "hostname": str(row.get("hostname") or ""),
                "description": str(row.get("description") or ""),
                "subnet": subnet,
            })
        return reservations

    def apply_dhcp_reservation(
        self, reservation: dict[str, Any], *, uuid: str | None = None
    ) -> dict[str, Any]:
        """Create or update a single Kea DHCPv4 static reservation.

        Does not reconfigure the service — call ``commit_dhcp_reservations()``
        once after a batch, so a ruleset apply costs one daemon reload rather
        than one per reservation.

        :raises ValueError: if no Kea-managed subnet matches the reservation.
        :raises RuntimeError: if Kea rejects the write.
        """
        subnet_uuid = self._kea_subnet_uuid_for(
            str(reservation.get("subnet") or "").strip(),
            str(reservation.get("ip") or ""),
        )

        payload = {
            "reservation": {
                "subnet": subnet_uuid,
                "ip_address": reservation.get("ip", ""),
                "hw_address": reservation.get("mac", ""),
                "hostname": reservation.get("hostname", ""),
                "description": reservation.get("description", ""),
            }
        }
        path = (
            f"/api/kea/dhcpv4/setReservation/{uuid}"
            if uuid
            else "/api/kea/dhcpv4/addReservation"
        )
        result = self._post(path, payload)
        if result.get("result") != "saved":
            raise RuntimeError(
                f"Kea rejected DHCP reservation for {reservation.get('ip')}: {result}"
            )
        return {"success": True}

    def commit_dhcp_reservations(self) -> dict[str, Any]:
        """Reload Kea so pending reservation writes take effect."""
        self._post("/api/kea/service/reconfigure")
        return {"success": True}

    # ── Kea DHCPv4 subnets ────────────────────────────────────────────────

    # OPNsense renders repeatable option fields ("AsList") as comma-separated
    # strings, and the pool list as a newline-separated text block. Which of
    # the two shapes a given field uses is a property of the model, not of the
    # request, so the mapping is a constant rather than something to sniff.
    _KEA_LIST_OPTIONS = (
        "routers",
        "domain_name_servers",
        "domain_search",
        "ntp_servers",
    )
    _KEA_SCALAR_OPTIONS = ("domain_name",)

    @staticmethod
    def _kea_read_list(raw: Any) -> list[str]:
        """Read an OPNsense list field, in either shape it comes back in.

        Plain string: ``"a,b"``. Selection map: ``{"a": {"selected": 1}, ...}``
        -- which OPNsense uses for some model field types and versions. Both
        appear in the wild for the same logical field, so both are accepted
        rather than pinning the driver to one OPNsense release.
        """
        if isinstance(raw, dict):
            return [
                str(key)
                for key, meta in raw.items()
                if isinstance(meta, dict) and str(meta.get("selected", 0)) == "1"
            ]
        if isinstance(raw, list):
            return [str(item).strip() for item in raw if str(item).strip()]
        return [part.strip() for part in str(raw or "").split(",") if part.strip()]

    @staticmethod
    def _kea_read_scalar(raw: Any) -> str:
        if isinstance(raw, dict):
            selected = [
                str(key)
                for key, meta in raw.items()
                if isinstance(meta, dict) and str(meta.get("selected", 0)) == "1"
            ]
            return selected[0] if selected else ""
        return str(raw or "").strip()

    @staticmethod
    def _kea_subnet_record(detail: dict[str, Any]) -> dict[str, Any]:
        """The subnet record out of a ``getSubnet`` response.

        OPNsense wraps it under ``subnet4``; older builds used ``subnet``, and
        keeping the fallback costs nothing. Reading the wrong key costs a
        great deal: the record comes back empty, so a subnet reports no pools
        and no options at all, and on update the partial-option merge has
        nothing to preserve and blanks every option Kea autocollected --
        stranding a whole VLAN without a gateway, which is the exact failure
        that merge exists to prevent.
        """
        return detail.get("subnet4") or detail.get("subnet") or {}

    def get_dhcp_subnets(self) -> list[dict[str, Any]]:
        """Return every Kea DHCPv4 subnet with its pools and options.

        ``searchSubnet`` only carries uuid/subnet/description, so each row is
        followed by a ``getSubnet`` call for the option data. That is one
        request per subnet; a firewall serves a handful of them, so the extra
        round trips cost less than the reconfigure they help avoid.

        An option Kea does not carry is *omitted* from ``option_data`` rather
        than reported as empty. The generic diff treats an absent key as "not
        managed", so reporting ``[]`` here would make every unmanaged option
        look like a pending change and reload the daemon on every run.

        A subnet whose detail fetch fails is skipped with a log line instead
        of aborting: one broken record must not make the whole inventory
        unreadable, same rule as ``get_dhcp_reservations()``.

        :raises RuntimeError: if the Kea plugin isn't installed/enabled.
        """
        rows = self._kea_subnets()

        subnets: list[dict[str, Any]] = []
        for row in rows:
            uuid = str(row.get("uuid") or "")
            if not uuid:
                continue
            try:
                detail = self._get(f"/api/kea/dhcpv4/getSubnet/{uuid}")
            except Exception as exc:
                logger.warning("Kea subnet %s detail fetch failed, skipping: %s", uuid, exc)
                continue

            record = self._kea_subnet_record(detail)
            raw_options = record.get("option_data") or {}

            option_data: dict[str, Any] = {}
            for name in self._KEA_LIST_OPTIONS:
                values = self._kea_read_list(raw_options.get(name))
                if values:
                    option_data[name] = values
            for name in self._KEA_SCALAR_OPTIONS:
                value = self._kea_read_scalar(raw_options.get(name))
                if value:
                    option_data[name] = value

            pools = [
                line.strip()
                for line in str(record.get("pools") or "").splitlines()
                if line.strip()
            ]

            subnets.append({
                "uuid": uuid,
                "subnet": str(record.get("subnet") or row.get("subnet") or ""),
                "description": str(record.get("description") or ""),
                "pools": pools,
                "option_data": option_data,
                "match_client_id": str(record.get("match-client-id") or "0") == "1",
            })
        return subnets

    def apply_dhcp_subnet(
        self, subnet: dict[str, Any], *, uuid: str | None = None
    ) -> dict[str, Any]:
        """Create or update a single Kea DHCPv4 subnet.

        ``option_data`` is a partial update, as the mixin contract requires:
        on an update the subnet's current options are read first and only the
        named ones are replaced. Without that, managing `domain_search` alone
        would blank the `routers` Kea autocollected and strand every client on
        that VLAN without a gateway.

        Setting any option explicitly also turns ``option_data_autocollect``
        off. Left on, Kea keeps re-filling routers/DNS/NTP from the interface
        and the next diff sees a change again -- a reconfigure loop rather
        than a converged state.

        Does not reconfigure the service -- call ``commit_dhcp_subnets()``
        once after a batch.

        :raises RuntimeError: if Kea rejects the write.
        """
        desired_options = dict(subnet.get("option_data") or {})

        merged_options: dict[str, str] = {}
        if uuid and desired_options:
            try:
                current = self._get(f"/api/kea/dhcpv4/getSubnet/{uuid}")
                raw = self._kea_subnet_record(current).get("option_data") or {}
            except Exception as exc:
                raise RuntimeError(
                    f"Kea subnet {uuid} could not be read before update: {exc}"
                ) from exc
            for name in self._KEA_LIST_OPTIONS:
                values = self._kea_read_list(raw.get(name))
                if values:
                    merged_options[name] = ",".join(values)
            for name in self._KEA_SCALAR_OPTIONS:
                value = self._kea_read_scalar(raw.get(name))
                if value:
                    merged_options[name] = value

        for name, value in desired_options.items():
            merged_options[name] = ",".join(value) if isinstance(value, list) else str(value)

        record: dict[str, Any] = {"subnet": subnet.get("subnet", "")}
        if subnet.get("description") is not None:
            record["description"] = subnet.get("description") or ""
        if subnet.get("pools") is not None:
            record["pools"] = "\n".join(subnet.get("pools") or [])
        if subnet.get("match_client_id") is not None:
            record["match-client-id"] = "1" if subnet.get("match_client_id") else "0"
        if desired_options:
            record["option_data"] = merged_options
            record["option_data_autocollect"] = "0"

        path = (
            f"/api/kea/dhcpv4/setSubnet/{uuid}" if uuid else "/api/kea/dhcpv4/addSubnet"
        )
        result = self._post(path, {"subnet": record})
        if result.get("result") != "saved":
            raise RuntimeError(
                f"Kea rejected DHCP subnet {subnet.get('subnet')}: {result}"
            )
        return {"success": True}

    def commit_dhcp_subnets(self) -> dict[str, Any]:
        """Reload Kea so pending subnet writes take effect."""
        self._post("/api/kea/service/reconfigure")
        return {"success": True}

    def get_services(self) -> list[dict[str, Any]]:
        """Return running services from OPNsense.

        Calls ``GET /api/core/service/search`` and normalises the rows to
        ``{name, running, enabled, pid}`` — the same format used by the
        OpenWrt driver so the UI can render them identically.
        """
        data = self._get("/api/core/service/search")
        result: list[dict[str, Any]] = []
        for row in data.get("rows", []):
            result.append({
                "name": row.get("name") or row.get("id", ""),
                "running": bool(row.get("running", 0)),
                "enabled": True,   # OPNsense has no separate enabled/disabled state
                "pid": 0,
            })
        return sorted(result, key=lambda x: x["name"].lower())

    def manage_service(self, name: str, action: str) -> dict[str, Any]:
        """Execute a lifecycle action on an OPNsense service.

        OPNsense exposes per-plugin service endpoints at
        ``/api/<plugin>/service/<action>``.  The ``name`` parameter must
        match the service ``id`` returned by :meth:`get_services`.

        Supported actions: ``start``, ``stop``, ``restart``.
        ``enable`` and ``disable`` are not supported on OPNsense (plugins
        are enabled/disabled by installing or removing them).
        """
        import re as _re
        if not _re.match(r'^[a-zA-Z0-9_\-]+$', name):
            raise ValueError(f"Invalid service name: {name!r}")
        if action not in ('start', 'stop', 'restart'):
            return {"success": False, "output": f"Action '{action}' is not supported on OPNsense services"}
        try:
            result = self._post(f"/api/{name}/service/{action}")
            return {"success": True, "output": str(result)}
        except Exception as exc:
            logger.warning("Service action failed: %s", exc)
            return {"success": False, "output": str(exc)}

    def get_vpn_tunnels(self) -> dict[str, dict[str, Any]]:
        """Return status of all configured VPN tunnels.

        Queries IPsec, OpenVPN, and WireGuard in order and merges results
        into a single dict keyed by a unique tunnel identifier.

        Each entry follows the ``VPNTunnelDict`` schema:
        ``type``, ``local_endpoint``, ``remote_endpoint``, ``is_up``,
        ``uptime``, ``bytes_in``, ``bytes_out``, ``description``.

        * **IPsec** — ``GET /api/ipsec/sessions``
          Falls back to ``GET /api/ipsec/leases/searchPhase2`` on older
          OPNsense releases.
        * **OpenVPN** — ``GET /api/openvpn/instances/search``
          Instance status obtained from ``GET /api/openvpn/service/show``.
        * **WireGuard** — ``GET /api/wireguard/service/show``
        """
        tunnels: dict[str, dict[str, Any]] = {}

        # ── IPsec ──────────────────────────────────────────────────────────
        try:
            data = self._get("/api/ipsec/sessions")
            # OPNsense 24.x returns {"response": [...]} or a bare list
            sessions = data.get("response") if isinstance(data, dict) else data
            if isinstance(sessions, list):
                for session in sessions:
                    # Each session object may contain multiple child SAs; we
                    # report one entry per IKE peer.
                    remote_host = (
                        session.get("remote-host")
                        or session.get("remote_host")
                        or session.get("remote-id")
                        or ""
                    )
                    local_host = (
                        session.get("local-host")
                        or session.get("local_host")
                        or session.get("local-id")
                        or ""
                    )
                    name = (
                        session.get("uniqueid")
                        or session.get("con-id")
                        or session.get("name")
                        or remote_host
                        or f"ipsec-{len(tunnels)}"
                    )
                    state = str(session.get("state") or session.get("ikey-state") or "").lower()
                    is_up = state in ("established", "up", "installed")
                    uptime_raw = session.get("established") or session.get("uptime") or 0
                    try:
                        uptime = int(uptime_raw)
                    except (ValueError, TypeError):
                        uptime = 0
                    # Byte counters may live in child-sa entries
                    bytes_in = 0
                    bytes_out = 0
                    for child in session.get("child-sas", {}).values() if isinstance(session.get("child-sas"), dict) else []:
                        try:
                            bytes_in += int(child.get("bytes-in", 0) or 0)
                            bytes_out += int(child.get("bytes-out", 0) or 0)
                        except (ValueError, TypeError):
                            pass
                    description = (
                        session.get("local-id")
                        or session.get("description")
                        or ""
                    )
                    key = f"ipsec-{name}"
                    tunnels[key] = {
                        "type": "IPsec",
                        "local_endpoint": local_host,
                        "remote_endpoint": remote_host,
                        "is_up": is_up,
                        "uptime": uptime,
                        "bytes_in": bytes_in,
                        "bytes_out": bytes_out,
                        "description": description,
                    }
        except Exception as exc:
            logger.debug("IPsec session status fetch failed: %s", exc)
            # Try legacy Phase-2 leases endpoint (OPNsense < 23.x)
            try:
                data = self._post(
                    "/api/ipsec/leases/searchPhase2",
                    {"current": 1, "rowCount": -1, "searchPhrase": "", "sort": {}},
                )
                for row in data.get("rows", []):
                    name = row.get("id") or row.get("con") or f"ipsec-{len(tunnels)}"
                    key = f"ipsec-{name}"
                    state = str(row.get("state") or "").lower()
                    is_up = state in ("established", "installed", "up")
                    tunnels[key] = {
                        "type": "IPsec",
                        "local_endpoint": row.get("local-ts", ""),
                        "remote_endpoint": row.get("remote-ts", ""),
                        "is_up": is_up,
                        "uptime": 0,
                        "bytes_in": int(row.get("bytes-in", 0) or 0),
                        "bytes_out": int(row.get("bytes-out", 0) or 0),
                        "description": row.get("con", ""),
                    }
            except Exception as exc:
                logger.debug("Legacy IPsec Phase-2 lease fetch failed: %s", exc)

        # ── OpenVPN ────────────────────────────────────────────────────────
        try:
            data = self._get("/api/openvpn/instances/search")
            instances = data.get("rows", [])
            # Fetch live service status to determine up/down state
            try:
                show = self._get("/api/openvpn/service/show")
                # show is a dict of {instance_id: {status, ...}}
                status_map: dict[str, Any] = show if isinstance(show, dict) else {}
            except Exception as exc:
                logger.debug("OpenVPN service status fetch failed: %s", exc)
                status_map = {}
            for inst in instances:
                iid = inst.get("id") or inst.get("vpnid") or f"ovpn-{len(tunnels)}"
                name = inst.get("description") or inst.get("dev") or str(iid)
                key = f"openvpn-{iid}"
                svc = status_map.get(str(iid), {})
                is_up = str(svc.get("running", False)).lower() in ("true", "1", "yes")
                local_addr = inst.get("local") or inst.get("interface") or ""
                remote_addr = inst.get("server") or inst.get("remote") or ""
                tunnels[key] = {
                    "type": "SSL",
                    "local_endpoint": local_addr,
                    "remote_endpoint": remote_addr,
                    "is_up": is_up,
                    "uptime": 0,
                    "bytes_in": 0,
                    "bytes_out": 0,
                    "description": name,
                }
        except Exception as exc:
            logger.debug("OpenVPN instance fetch failed (plugin not installed?): %s", exc)

        # ── WireGuard ──────────────────────────────────────────────────────
        try:
            # /api/wireguard/service/show returns structured JSON:
            # {"total": N, "rows": [
            #   {"if":"wg0", "type":"interface", ...},
            #   {"if":"wg0", "type":"peer", "public-key":"...",
            #    "endpoint":"1.2.3.4:51820", "transfer-rx":N, "transfer-tx":N,
            #    "name":"HCQ", "peer-status":"online",
            #    "latest-handshake-age":45, "ifname":"HCQ"}, ...
            # ]}
            data = self._get("/api/wireguard/service/show")
            rows = data.get("rows", []) if isinstance(data, dict) else []

            wg_idx = 0
            for row in rows:
                if row.get("type") != "peer":
                    continue
                wg_idx += 1

                endpoint = (row.get("endpoint") or "").strip()
                if endpoint and endpoint != "(none)":
                    remote_ip = endpoint.rsplit(":", 1)[0].strip("[]")
                else:
                    remote_ip = ""

                is_up = row.get("peer-status") == "online"
                bytes_in = int(row.get("transfer-rx") or 0)
                bytes_out = int(row.get("transfer-tx") or 0)
                hs_age = row.get("latest-handshake-age")
                uptime = int(hs_age) if hs_age else 0

                description = (
                    (row.get("name") or "").strip()
                    or (row.get("ifname") or "").strip()
                    or f"wg-peer-{wg_idx}"
                )

                tunnels[f"wireguard-{wg_idx}"] = {
                    "type": "WireGuard",
                    "local_endpoint": "",
                    "remote_endpoint": remote_ip,
                    "is_up": is_up,
                    "uptime": uptime,
                    "bytes_in": bytes_in,
                    "bytes_out": bytes_out,
                    "description": description,
                }
        except Exception as exc:
            logger.debug("WireGuard status fetch failed (plugin not installed?): %s", exc)

        return tunnels

    def get_available_updates(self) -> list[dict[str, Any]]:
        """Return available firmware and package updates.

        Triggers an async update-check on OPNsense via
        ``POST /api/core/firmware/check``, then polls
        ``GET /api/core/firmware/status`` for up to 15 seconds.
        Returns a list of ``{name, current_version, new_version}`` dicts,
        or an empty list when everything is up to date or the check has
        not yet finished.
        """
        import time
        try:
            self._post("/api/core/firmware/check")
        except Exception as exc:
            logger.debug("Firmware update check trigger failed: %s", exc)

        for _ in range(5):
            time.sleep(3)
            try:
                status = self._get("/api/core/firmware/status")
                state = status.get("status", "none")
                if state in ("update", "upgrade"):
                    updates = (
                        status.get("upgrade_packages")
                        or status.get("updates")
                        or []
                    )
                    return [
                        {
                            "name": u.get("name", ""),
                            "current_version": u.get("current_version", u.get("version", "")),
                            "new_version": u.get("new_version", u.get("version", "")),
                        }
                        for u in updates
                    ]
                if state == "latest":
                    return []
            except Exception as exc:
                logger.debug("Firmware status poll failed: %s", exc)
        return []

    def get_device_warnings(self) -> list[dict[str, Any]]:
        """Return a list of warning dicts for issues detected on this device.

        Reads the cached ``GET /api/core/firmware/status`` (no network
        update trigger) to detect available package/firmware updates.
        """
        warnings: list[dict[str, Any]] = []
        try:
            status = self._get("/api/core/firmware/status")
            state = status.get("status", "none")
            if state in ("update", "upgrade"):
                upgrades = (
                    status.get("upgrade_packages")
                    or status.get("updates")
                    or []
                )
                if upgrades:
                    warnings.append({
                        "code": "updates_available",
                        "meta": {
                            "count": len(upgrades),
                            "packages": [u.get("name", "") for u in upgrades[:10]],
                        },
                    })
        except Exception as exc:
            logger.debug("Failed to check firmware status for device warnings: %s", exc)
        return warnings

    def apply_updates(self, packages: list[str]) -> dict[str, Any]:
        """Trigger a full firmware upgrade on OPNsense.

        Note: OPNsense upgrades the entire system at once rather than
        individual packages.  The ``packages`` parameter is accepted for
        API compatibility but is ignored — the full upgrade is always
        applied.

        Calls ``POST /api/core/firmware/upgrade``.
        """
        try:
            result = self._post("/api/core/firmware/upgrade")
            return {"success": True, "output": str(result)}
        except Exception as exc:
            logger.warning("Firmware upgrade failed: %s", exc)
            return {"success": False, "output": str(exc)}

    def search_packages(self, query: str) -> list[dict[str, Any]]:
        """Search available OPNsense plugins by name or description.

        Filters the full plugin list from ``GET /api/core/firmware/info``
        against *query* (case-insensitive substring match on name and
        comment).  Returns all matching plugins with an ``installed``
        flag so the UI can show which ones are already active.
        """
        q = query.lower()
        info = self._get("/api/core/firmware/info")
        result: list[dict[str, Any]] = []
        for p in info.get("plugin", []):
            name = p.get("name", "")
            comment = p.get("comment", "")
            if q in name.lower() or q in comment.lower():
                result.append({
                    "name": name,
                    "version": p.get("version", ""),
                    "installed": p.get("installed") == "1",
                    "description": comment,
                    "size": 0,
                    "source": "opnsense-plugins",
                })
        return sorted(result, key=lambda x: x["name"].lower())

    def install_package(self, name: str) -> dict[str, Any]:
        """Install an OPNsense plugin by name.

        Calls ``POST /api/core/firmware/install/{name}``.
        """
        import re as _re
        if not _re.match(r'^[a-zA-Z0-9_\-\.]+$', name):
            raise ValueError(f"Invalid package name: {name!r}")
        try:
            result = self._post(f"/api/core/firmware/install/{name}")
            return {"success": True, "output": str(result)}
        except Exception as exc:
            logger.warning("Package install failed: %s", exc)
            return {"success": False, "output": str(exc)}

    def uninstall_package(self, name: str) -> dict[str, Any]:
        """Remove an OPNsense plugin by name.

        Calls ``POST /api/core/firmware/remove/{name}``.
        """
        import re as _re
        if not _re.match(r'^[a-zA-Z0-9_\-\.]+$', name):
            raise ValueError(f"Invalid package name: {name!r}")
        try:
            result = self._post(f"/api/core/firmware/remove/{name}")
            return {"success": True, "output": str(result)}
        except Exception as exc:
            logger.warning("Package uninstall failed: %s", exc)
            return {"success": False, "output": str(exc)}

    # ── SNMP / Health ──────────────────────────────────────────────────────────

    def get_snmp_config(self):
        """Return SNMP config if the os-net-snmp plugin is installed and enabled.

        Uses GET /api/netsnmp/general/get (os-net-snmp plugin API).
        Returns None if the plugin is not installed or SNMP is disabled.
        """
        try:
            from napalm_device_types.models import SNMPConfigDict
        except ImportError:
            return None

        try:
            # Short timeout — endpoint may not exist if os-net-snmp plugin is not installed
            url = self.base_url.rstrip("/") + "/api/netsnmp/general/get"
            response = self.session.get(url, timeout=5)
            if response.status_code != 200:
                return None
            data = response.json()
        except Exception as exc:
            logger.debug("SNMP config fetch failed (plugin not installed?): %s", exc)
            return None

        if not data:
            return None

        general = data.get("general", data)
        enabled = str(general.get("enabled", "0")) == "1"
        if not enabled:
            return None

        community = general.get("community", "public") or "public"
        return SNMPConfigDict(running=True, community=community, port=161, version="2c")

    # ── Wake-on-LAN ──────────────────────────────────────────────────────────────

    def send_wake_on_lan(self, mac_address: str, interface: str = "") -> dict[str, Any]:
        """Send a Wake-on-LAN magic packet via OPNsense's os-wol plugin.

        Calls ``POST /api/wol/wol/set`` with an ephemeral (non-persisted) entry —
        omitting ``uuid`` from the payload makes OPNsense's ``WolController::setAction``
        validate and wake immediately without saving a host to config.xml. Requires
        the os-wol plugin to be installed and the target interface to have a static
        IPv4 address configured (OPNsense computes the broadcast address from the
        interface's own IP/subnet; DHCP-assigned interfaces are rejected).

        :param interface: OPNsense's *assigned* interface identifier (e.g. "lan",
            "opt1") — NOT the physical device name returned by get_interfaces()/
            get_networks() (e.g. "em0"). Required by this driver.
        :raises ValueError: if interface is not provided.
        """
        if not interface:
            raise ValueError("OPNsense requires an interface for Wake-on-LAN")
        try:
            result = self._post(
                "/api/wol/wol/set",
                {"wake": {"interface": interface, "mac": mac_address.strip()}},
            )
        except Exception as exc:
            logger.warning("Wake-on-LAN send failed for %s via %s: %s", mac_address, interface, exc)
            return {"success": False, "output": str(exc)}

        status = result.get("status")
        if status == "error":
            return {
                "success": False,
                "output": result.get("error_msg", "OPNsense rejected the Wake-on-LAN request"),
            }
        if not result:
            # Model validation (malformed MAC/interface) fails silently with an
            # empty {} response at HTTP 200 — no error_msg to surface.
            return {
                "success": False,
                "output": "OPNsense rejected the request (check interface identifier and MAC format)",
            }
        return {"success": True, "output": f"Magic packet sent to {mac_address} via {interface}"}

    def run_device_action(self, action: str) -> dict[str, Any]:
        """Execute a named action on the firewall."""
        if action == "fix_snmp":
            return self._action_fix_snmp()
        raise NotImplementedError(f"Unknown action: {action!r}")

    def _action_fix_snmp(self) -> dict[str, Any]:
        """Install os-net-snmp plugin, configure community 'public', open firewall.

        Steps:
        1. Install os-net-snmp via firmware API (idempotent)
        2. Configure via POST /api/netsnmp/general/set
        3. Start/restart the service
        4. Add a floating firewall rule allowing UDP/161 from any source
           (OPNsense default-drops traffic arriving on non-LAN interfaces
           such as WireGuard; without this rule SNMP is unreachable from
           management networks even though the daemon is running)
        5. Apply firewall and verify
        """
        lines: list = []

        # 1. Install os-net-snmp plugin ONLY if not already present.
        # Calling firmware/install on an already-installed plugin triggers an
        # async reinstall that overwrites the config with factory defaults a few
        # minutes later — causing SNMP to stop working again after the fix.
        plugin_installed = False
        try:
            chk = self._get("/api/netsnmp/general/get")
            plugin_installed = isinstance(chk, dict) and "general" in chk
        except Exception:
            pass

        if not plugin_installed:
            try:
                result = self._post("/api/core/firmware/install/os-net-snmp")
                lines.append(f"[install] {result}")
                import time as _time
                _time.sleep(5)  # wait for install to settle
            except Exception as exc:
                logger.debug("os-net-snmp install failed: %s", exc)
                lines.append(f"[install] failed: {exc}")
        else:
            lines.append("[install] os-net-snmp already installed — skipping reinstall")

        # 2. Configure SNMP
        # The listen field must contain the management IP so that snmpd binds
        # to the correct interface. Without it snmpd may not respond on any
        # non-loopback address.
        # The API expects {"<ip>": {"selected": 1}} — we use self.hostname
        # which is always the device's management IP in netOrk.
        listen_val: dict = {self.hostname: {"selected": 1}}
        try:
            self._post("/api/netsnmp/general/set", {
                "general": {
                    "enabled": "1",
                    "community": "public",
                    "contact": "netork@localhost",
                    "location": "Managed by netOrk",
                    "sysobjid": "",
                    "bindip": "",
                    "listen": listen_val,
                }
            })
            lines.append(f"[config] SNMP enabled with community 'public', listen={self.hostname}.")
        except Exception as exc:
            logger.debug("SNMP config failed: %s", exc)
            lines.append(f"[config] error: {exc}")

        # 3. Start / restart the SNMP service
        try:
            self._post("/api/netsnmp/service/restart")
            lines.append("[service] net-snmp restarted.")
        except Exception as exc:
            logger.debug("SNMP service restart failed, trying start: %s", exc)
            try:
                self._post("/api/netsnmp/service/start")
                lines.append("[service] net-snmp started.")
            except Exception as exc2:
                lines.append(f"[service] start failed: {exc2}")

        # 4. Add floating firewall rule for SNMP (UDP/161 → self)
        # OPNsense blocks traffic arriving on non-LAN interfaces by default.
        # A floating pass rule makes SNMP reachable from all management nets.
        try:
            existing = self._post("/api/firewall/filter/searchRule",
                                  {"current": 1, "rowCount": -1, "searchPhrase": "[netork] Allow SNMP"})
            if not (existing.get("rows") or []):
                self._post("/api/firewall/filter/addRule", {
                    "rule": {
                        "enabled": "1",
                        "sequence": "1",
                        "action": "pass",
                        "quick": "1",
                        "interface": "",        # floating — all interfaces
                        "direction": "in",
                        "ipprotocol": "inet",
                        "protocol": "udp",
                        "source_net": "any",
                        "source_not": "0",
                        "destination_net": "(self)",
                        "destination_not": "0",
                        "destination_port": "161",
                        "log": "0",
                        "floating": "yes",
                        "descr": "[netork] Allow SNMP",
                    }
                })
                lines.append("[firewall] Floating pass rule added for UDP/161.")
            else:
                lines.append("[firewall] Pass rule already present.")
            self._post("/api/firewall/filter/apply", {})
            lines.append("[firewall] Rules applied.")
        except Exception as exc:
            logger.debug("Firewall rule for SNMP failed: %s", exc)
            lines.append(f"[firewall] error: {exc}")

        # 5. Verify — try actual UDP/161 probe first, fall back to API config check
        success = False
        try:
            import socket as _socket
            sock = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM)
            sock.settimeout(3)
            community = b"public"
            # Minimal SNMPv2c GetRequest for sysDescr
            pdu = (b"\x30\x26\x02\x01\x01\x04"
                   + bytes([len(community)]) + community
                   + b"\xa0\x19\x02\x04\x00\x00\x00\x01"
                   + b"\x02\x01\x00\x02\x01\x00"
                   + b"\x30\x0b\x30\x09\x06\x05\x2b\x06\x01\x02\x01\x05\x00")
            sock.sendto(pdu, (self.hostname, 161))
            try:
                data, _ = sock.recvfrom(1024)
                success = len(data) > 0
                lines.append("[ok] SNMP UDP probe successful.")
            except _socket.timeout:
                lines.append("[warn] SNMP UDP probe timed out — service may be starting.")
            finally:
                sock.close()
        except Exception as exc:
            logger.debug("SNMP UDP probe failed: %s", exc)
            # Fall back to API config check
            try:
                cfg = self._get("/api/netsnmp/general/get")
                general = cfg.get("general", cfg)
                success = str(general.get("enabled", "0")) == "1" and bool(general.get("community"))
                lines.append("[ok] SNMP config active (UDP probe unavailable)." if success
                             else "[warn] SNMP config check failed.")
            except Exception:
                lines.append("[warn] Could not verify SNMP state.")

        return {"success": success, "output": "\n".join(lines)}

    def get_firewall_aliases(self) -> list[dict[str, Any]]:
        """Return all firewall aliases, sorted by type then name.

        Each entry contains:
        * ``name``        — alias name
        * ``type``        — alias type (host, network, port, url, urltable, geoip, etc.)
        * ``description`` — human-readable description
        * ``content``     — list of values (IPs, networks, ports, URLs, …)
        * ``enabled``     — bool
        * ``counters``    — optional dict with packet/byte stats if available
        """
        try:
            resp = self._get("/api/firewall/alias/searchItem?current=1&rowCount=-1")
        except Exception as exc:
            logger.warning("Failed to fetch firewall aliases: %s", exc)
            return []

        rows = resp.get("rows") or []
        result = []
        for row in rows:
            content_raw = row.get("content", "") or ""
            # OPNsense stores content as newline-separated values
            content = [v.strip() for v in content_raw.splitlines() if v.strip()]
            result.append({
                "name":        row.get("name", ""),
                "type":        row.get("type", ""),
                "description": row.get("description", "") or "",
                "content":     content,
                "enabled":     str(row.get("enabled", "1")) == "1",
                "proto":       row.get("proto", "") or "",
            })

        return sorted(result, key=lambda x: (x["type"], x["name"].lower()))

    def get_firewall_rules(self) -> list[dict[str, Any]]:
        """Return all firewall filter rules with interface labels.

        Extra fields beyond NAPALM standard:
        * ``floating``          — bool, rule applies across all interfaces
        * ``interface_label``   — human-readable interface description
        * ``is_group``          — bool, interface is an interface group
        """
        try:
            resp = self._get("/api/firewall/filter/searchRule?current=1&rowCount=-1")
        except Exception as exc:
            logger.warning("Failed to fetch firewall rules: %s", exc)
            return []

        # OPNsense provides all human-readable values via %-prefixed fields —
        # no separate lookup needed for interface labels or category names.

        # Interface group names (to distinguish groups from plain interfaces)
        group_names: set = set()
        try:
            grp = self._get("/api/ifgroups/ifgroups/searchItem?current=1&rowCount=-1")
            for g in (grp.get("rows") or []):
                n = g.get("ifname") or g.get("name") or ""
                if n:
                    group_names.add(n)
        except Exception as exc:
            logger.warning("Failed to fetch interface groups: %s", exc)

        rows = resp.get("rows") or []
        result = []
        for row in rows:
            iface = row.get("interface", "") or ""
            iface_list = [i.strip() for i in iface.split(",") if i.strip()]
            # %interface already has the resolved friendly names (e.g. "MGMT, wgadmin")
            iface_label = (row.get("%interface") or "").strip() or ", ".join(iface_list)

            explicit_floating = (
                str(row.get("floating", "0")) in ("1", "yes")
                or row.get("floating") is True
            )
            floating = explicit_floating or len(iface_list) > 1

            # %categories is the resolved category name; categories is the UUID
            category = (row.get("%categories") or row.get("category") or "").strip()

            # Use %-prefixed display values for source/destination
            source_net      = (row.get("%source_net")      or row.get("source_net")      or "any").strip()
            destination_net = (row.get("%destination_net") or row.get("destination_net") or "any").strip()

            result.append({
                "uuid":             row.get("uuid", ""),
                "sequence":         int(row.get("sequence", 0) or 0),
                "action":           row.get("action", "pass"),
                "quick":            str(row.get("quick", "1")) == "1",
                "interface":        iface,
                "interface_label":  iface_label,
                "floating":         floating,
                "is_group":         len(iface_list) == 1 and iface_list[0] in group_names,
                "direction":        row.get("direction", "in"),
                "ipprotocol":       row.get("ipprotocol", "inet"),
                "protocol":         row.get("protocol", "any") or "any",
                "source_net":       source_net,
                "source_port":      row.get("source_port", "") or "",
                "source_not":       str(row.get("source_not", "0")) == "1",
                "destination_net":  destination_net,
                "destination_port": row.get("destination_port", "") or "",
                "destination_not":  str(row.get("destination_not", "0")) == "1",
                "description":      row.get("description", "") or "",
                "log":              str(row.get("log", "0")) == "1",
                "enabled":          str(row.get("enabled", "1")) == "1",
                "category":         category,
            })

        return sorted(result, key=lambda x: (x["floating"], x["is_group"], x["interface"], x["sequence"]))

    def apply_firewall_rule(self, rule: dict[str, Any], *, uuid: str | None = None) -> dict[str, Any]:
        """Create or update a single OPNsense firewall filter rule.

        `rule` uses the vendor-neutral field names from
        ``napalm_device_types.models.FirewallRuleDict`` (see
        ``FirewallDriver.diff_firewall_rules``/``apply_firewall_ruleset``,
        the generic reconciliation engine that calls this method). This is
        the OPNsense-specific half: translating those fields into the
        ``/api/firewall/filter/addRule``/``setRule`` payload shape (string
        "1"/"0" booleans, empty ``interface`` means a floating rule — same
        payload shape as the SNMP self-provisioning rule in
        ``_action_fix_snmp``).
        """
        payload = {
            "rule": {
                "enabled": "1" if rule.get("enabled", True) else "0",
                "sequence": "1",
                "action": rule.get("action", "pass"),
                "quick": "1" if rule.get("quick", True) else "0",
                "interface": rule.get("interface", "") or "",
                "direction": rule.get("direction", "in"),
                "ipprotocol": "inet",
                "protocol": rule.get("protocol", "any"),
                "source_net": rule.get("source_net", "any") or "any",
                "source_port": rule.get("source_port", "") or "",
                "destination_net": rule.get("destination_net", "any") or "any",
                "destination_port": rule.get("destination_port", "") or "",
                "log": "1" if rule.get("log", False) else "0",
                "floating": "yes" if not rule.get("interface") else "no",
                "descr": rule.get("description", ""),
            }
        }
        path = f"/api/firewall/filter/setRule/{uuid}" if uuid else "/api/firewall/filter/addRule"
        return self._post(path, payload)

    def commit_firewall_rules(self) -> dict[str, Any]:
        """Reload the firewall filter to activate pending rule changes.

        Final step after one or more `apply_firewall_rule()` calls — same
        as the last step of `_action_fix_snmp`.
        """
        return self._post("/api/firewall/filter/apply", {})

    # ------------------------------------------------------------------
    # Hostname management
    # ------------------------------------------------------------------

    def set_hostname(self, new_hostname: str) -> None:
        """Set the system hostname on OPNsense and regenerate the web GUI certificate.

        Accepts either a bare hostname or an FQDN (``host.domain``).
        When an FQDN is passed the domain part is also updated.

        Only sends the fields that need to change — never overwrites the full
        general config object to avoid accidental data loss.

        Requires OPNsense 22.x or later with the general settings REST API.
        Raises NotImplementedError on older versions that lack this endpoint.
        """
        if "." in new_hostname:
            hostname, domain = new_hostname.split(".", 1)
        else:
            hostname = new_hostname
            domain = None

        payload: dict = {"general": {"hostname": hostname}}
        if domain:
            payload["general"]["domain"] = domain

        try:
            result = self._post("/api/core/general/set", payload)
        except Exception as exc:
            raise NotImplementedError(
                f"OPNsense at {self.hostname} does not support hostname management "
                "via REST API (requires OPNsense 22.x+). "
                "Please set the hostname manually via System → Settings → General "
                f"in the web UI. (Detail: {exc})"
            ) from exc

        logger.info("OPNsense set_hostname → %s (domain: %s): %s", hostname, domain, result)

        # ── 3. Regenerate web GUI TLS certificate ──
        self._regenerate_web_cert(hostname, domain)

    def _regenerate_web_cert(self, hostname: str, domain: str) -> None:
        """Attempt to regenerate the OPNsense web GUI self-signed certificate.

        Tries the configd helper first (OPNsense 22.x+), then falls back to the
        Trust API (older releases).  Logs a warning if neither works — the
        operator should then regenerate the cert manually in the web GUI.
        """
        fqdn = f"{hostname}.{domain}" if domain else hostname

        # Attempt 1: configd helper (most common on 22.x / 23.x / 24.x)
        try:
            self._post("/api/core/configd/generateRootCert", {})
            logger.info("OPNsense web cert regenerated via configd (fqdn: %s)", fqdn)
            return
        except Exception as exc:
            logger.debug("configd/generateRootCert not available: %s", exc)

        # Attempt 2: Trust API — create a new internal self-signed cert
        try:
            # Fetch the list of internal CAs to find the right one
            ca_data = self._post("/api/trust/ca/search", {})
            internal_ca_uuid = None
            for row in (ca_data.get("rows") or []):
                if row.get("internal") == "1" or row.get("catype") == "internal":
                    internal_ca_uuid = row.get("uuid")
                    break

            cert_payload: dict = {
                "cert": {
                    "descr": f"netork-generated for {fqdn}",
                    "caref": internal_ca_uuid or "",
                    "keytype": "RSA",
                    "keylen": "2048",
                    "digest_alg": "sha256",
                    "lifetime": "825",
                    "dn_commonname": fqdn,
                    "dn_sans": fqdn,
                }
            }
            result = self._post("/api/trust/cert/generate", cert_payload)
            logger.info("OPNsense web cert generated via Trust API: %s", result)
            return
        except Exception as exc:
            logger.debug("Trust API cert generation failed: %s", exc)

        logger.warning(
            "OPNsense: could not regenerate web GUI certificate automatically after "
            "hostname change to %s — please regenerate it manually in the web GUI "
            "(System → Trust → Certificates).",
            fqdn,
        )

    # ------------------------------------------------------------------
    # DNS host overrides
    # ------------------------------------------------------------------

    def _detect_active_dns(self) -> str | None:
        """Return 'unbound' or 'dnsmasq' depending on which service is running."""
        try:
            data = self._get("/api/core/service/search")
            for row in data.get("rows", []):
                name = (row.get("name") or "").lower()
                if name in ("unbound", "dnsmasq") and row.get("running"):
                    return name
        except Exception as exc:
            logger.warning("DNS service detection failed: %s", exc)
        return None

    def get_dns_entries(self) -> list[dict[str, Any]]:
        """Return DNS host-override entries from the active DNS service.

        Checks whether Unbound (DNS Resolver) or Dnsmasq (DNS Forwarder) is
        running and returns host overrides from whichever is active.  If neither
        is running, returns an empty list.

        Each entry::

            {
                "uuid": str,           # OPNsense-internal UUID
                "hostname": str,       # host part, e.g. "gw"
                "domain": str,         # domain part, e.g. "home.example.com"
                "fqdn": str,           # hostname.domain
                "ip": str,             # IP address
                "record_type": str,    # "A" | "AAAA" | "MX" | …
                "description": str,
                "enabled": bool,
                "service": str,        # "unbound" | "dnsmasq"
            }
        """
        service = self._detect_active_dns()
        if service == "unbound":
            return self._get_unbound_host_overrides()
        if service == "dnsmasq":
            return self._get_dnsmasq_host_overrides()
        return []

    def _get_unbound_host_overrides(self) -> list[dict[str, Any]]:
        result: list[dict[str, Any]] = []
        seen: set[tuple[str, str, str, str]] = set()
        try:
            data = self._post(
                "/api/unbound/settings/searchhostoverride",
                {"current": 1, "rowCount": -1, "searchPhrase": ""},
            )
            for row in data.get("rows", []):
                host = row.get("hostname", "") or ""
                domain = row.get("domain", "") or ""
                fqdn = f"{host}.{domain}" if host and domain else host or domain
                ip = row.get("server", "") or ""
                rr = row.get("rr", "A") or "A"
                # OPNsense searchhostoverride includes alias records alongside parent
                # records; aliases often have identical content but separate UUIDs.
                # Deduplicate by logical key to avoid inflating the zone with copies.
                key = (host.lower(), domain.lower(), ip, rr)
                if key in seen:
                    continue
                seen.add(key)
                result.append({
                    "uuid": row.get("uuid", ""),
                    "hostname": host,
                    "domain": domain,
                    "fqdn": fqdn,
                    "ip": ip,
                    "record_type": rr,
                    "description": row.get("description", "") or "",
                    "enabled": str(row.get("enabled", "1")) == "1",
                    "ptrrecord": str(row.get("ptrrecord", "1")) == "1",
                    "service": "unbound",
                })
        except Exception as exc:
            logger.warning("Failed to fetch Unbound host overrides: %s", exc)
        return sorted(result, key=lambda e: e["fqdn"].lower())

    def _get_dnsmasq_host_overrides(self) -> list[dict[str, Any]]:
        result: list[dict[str, Any]] = []
        try:
            data = self._post(
                "/api/dnsmasq/settings/searchhostoverride",
                {"current": 1, "rowCount": -1, "searchPhrase": ""},
            )
            for row in data.get("rows", []):
                host = row.get("host", "") or ""
                domain = row.get("domain", "") or ""
                fqdn = f"{host}.{domain}" if host and domain else host or domain
                result.append({
                    "uuid": row.get("uuid", ""),
                    "hostname": host,
                    "domain": domain,
                    "fqdn": fqdn,
                    "ip": row.get("ip", "") or "",
                    "record_type": "A",
                    "description": row.get("description", "") or "",
                    "enabled": str(row.get("enabled", "1")) == "1",
                    "service": "dnsmasq",
                })
        except Exception as exc:
            logger.warning("Failed to fetch Dnsmasq host overrides: %s", exc)
        return sorted(result, key=lambda e: e["fqdn"].lower())

    def set_dns_entry(self, uuid: str, service: str, data: dict[str, Any]) -> dict[str, Any]:
        """Update a single DNS host-override entry and reconfigure the service.

        ``data`` keys: hostname, domain, ip, record_type, description, enabled.
        Returns the raw API response from the set call.
        """
        if service == "unbound":
            payload = {
                "host": {
                    "enabled": "1" if data.get("enabled", True) else "0",
                    "hostname": data.get("hostname", ""),
                    "domain": data.get("domain", ""),
                    "rr": data.get("record_type", "A"),
                    "server": data.get("ip", ""),
                    "description": data.get("description", "") or "",
                    "mxprio": "",
                    "mx": "",
                }
            }
            resp = self._post(f"/api/unbound/settings/sethostoverride/{uuid}", payload)
            self._post("/api/unbound/service/reconfigure")
        elif service == "dnsmasq":
            payload = {
                "hostoverride": {
                    "enabled": "1" if data.get("enabled", True) else "0",
                    "host": data.get("hostname", ""),
                    "domain": data.get("domain", ""),
                    "ip": data.get("ip", ""),
                    "description": data.get("description", "") or "",
                }
            }
            resp = self._post(f"/api/dnsmasq/settings/sethostoverride/{uuid}", payload)
            self._post("/api/dnsmasq/service/reconfigure")
        else:
            raise NotImplementedError(f"set_dns_entry not supported for service '{service}'")
        return resp

    _NETORK_TAG = "[netork]"

    def sync_dns_zone(self, zone_name: str, records: list[dict[str, Any]]) -> None:
        """Replace all netork-managed host overrides for *zone_name* with *records*.

        records items: {"hostname": str, "ip": str, "record_type": str, "enabled": bool}
        Auto-detects whether Unbound or Dnsmasq is active.
        """
        zone_lower = zone_name.lower()
        if "in-addr.arpa" in zone_lower or "ip6.arpa" in zone_lower:
            raise ValueError(
                f"Refusing to provision reverse zone '{zone_name}' as OPNsense host overrides — "
                "PTR records must not be managed via the host override API."
            )

        service = self._detect_active_dns()
        if not service:
            raise RuntimeError("No active DNS service (Unbound or Dnsmasq) detected")

        zone_clean = zone_name.rstrip(".")

        # Remove existing netork-managed overrides for this zone
        existing = (self._get_unbound_host_overrides() if service == "unbound"
                    else self._get_dnsmasq_host_overrides())
        for entry in existing:
            if (entry["domain"].rstrip(".") == zone_clean
                    and entry["description"].startswith(self._NETORK_TAG)):
                uid = entry["uuid"]
                if uid:
                    if service == "unbound":
                        self._post(f"/api/unbound/settings/delhostoverride/{uid}")
                    else:
                        self._post(f"/api/dnsmasq/settings/delhostoverride/{uid}")
                    logger.info("Removed %s host override %s.%s (uuid %s)",
                                service, entry["hostname"], zone_clean, uid)

        # Re-add all enabled A/AAAA records
        added = 0
        for rec in records:
            if not rec.get("enabled", True):
                continue
            rtype = rec.get("record_type", "A")
            if rtype not in ("A", "AAAA"):
                continue
            if service == "unbound":
                self._post("/api/unbound/settings/addhostoverride", {
                    "host": {
                        "enabled": "1",
                        "hostname": rec.get("hostname", ""),
                        "domain": zone_clean,
                        "rr": rtype,
                        "server": rec.get("ip", ""),
                        "description": self._NETORK_TAG,
                        "ptrrecord": "1",
                        "mxprio": "",
                        "mx": "",
                    }
                })
            else:
                self._post("/api/dnsmasq/settings/addhostoverride", {
                    "hostoverride": {
                        "enabled": "1",
                        "host": rec.get("hostname", ""),
                        "domain": zone_clean,
                        "ip": rec.get("ip", ""),
                        "description": self._NETORK_TAG,
                    }
                })
            added += 1
            logger.info("Added %s host override %s.%s → %s",
                        service, rec.get("hostname"), zone_clean, rec.get("ip"))

        if service == "unbound":
            self._post("/api/unbound/service/reconfigure")
        else:
            self._post("/api/dnsmasq/service/reconfigure")
        logger.info("sync_dns_zone %s via %s: %d records provisioned", zone_clean, service, added)

    def deprovision_dns_zone(self, zone_name: str) -> None:
        """Remove all netork-managed host overrides for *zone_name*."""
        self.sync_dns_zone(zone_name, [])