feat(firewall): report the gateway a filter rule policy-routes to

get_firewall_rules read searchRule but dropped the rule's gateway. A pass
rule with a gateway hands what it matches to that gateway, local
destinations included, so it does not reach a host on another internal
network. Without the field a caller judging reachability reads a rule meant
for internet traffic as a hole into every server: on the first real box,
"pass UDP IOT -> any" via WAN_GW made 61 hosts look reachable from the IoT
segment (netOrk #575).

The rule dict gains "gateway", the gateway's name or "". It is an extra
field like floating and interface_label; the generic diff compares a fixed
field list and ignores it.

.github/workflows/ci.yml

@@ -12,7 +12,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.10", "3.11", "3.12"]
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -26,6 +26,9 @@ jobs:
       - name: Install package with dev extras
         run: |
           python -m pip install --upgrade pip
+          # napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
+          # pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
+          python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
           python -m pip install -e ".[dev]"
 
       - name: Run unit tests
@@ -38,7 +41,8 @@ jobs:
           python -m build
 
       - name: Upload dist artifacts
-        uses: actions/upload-artifact@v4
+        # v4 refuses to run on Gitea ("not currently supported on GHES").
+        uses: actions/upload-artifact@v3
         with:
           name: dist-${{ matrix.python-version }}
           path: dist/*

README.md

@@ -87,9 +87,35 @@ arguments.
 | `get_bgp_neighbors` | ✅ ² | `GET /api/quagga/bgp/get` + `GET /api/quagga/diagnostics/bgpneighbors` |
 | `get_vlans` | ✅ | `GET /api/interfaces/vlan_settings/search_item` |
 | `get_mac_address_table` | ❌ | Not applicable (firewall, no L2 switching) |
+| `ping` | ✅ | `POST /api/diagnostics/ping/set` + `start` + `search_jobs` + `stop`/`remove` |
+| `ping_sweep` | ✅ ³ | same endpoints, one batch of parallel jobs at a time |
 
 > ¹ Requires the `os-lldpd` plugin. Returns empty dict if the plugin is not installed.
 > ² Requires the `os-frr` (FRR/Quagga) plugin. Returns empty dict if the plugin is not installed or FRR is not running.
+> ³ Overrides the generic per-host loop from `napalm-device-types`.
+
+### Ping
+
+OPNsense has no synchronous ping endpoint: `/api/diagnostics/ping` is a *job*
+API — create, start, read statistics, stop, remove. A single ping therefore
+costs five requests and about a second of waiting, which makes the generic
+sequential sweep from `napalm-device-types` unusable for a whole subnet.
+
+`ping_sweep` exploits what the job API does offer instead: jobs are
+independent and run on the firewall in parallel, and `search_jobs` reports all
+of them in one response. It creates and starts a batch
+(`PING_SWEEP_BATCH_SIZE`, default 32), waits once, reads every result with a
+single request, then cleans the batch up — waiting time per batch is constant
+rather than linear in hosts. `PING_SWEEP_MAX_TARGETS` (default 512) bounds the
+sweep as a whole; both are deliberately conservative, since this runs on
+production firewalls.
+
+Results are polled rather than read once: `search_jobs` signals the running
+ping with `SIGINFO` and parses whatever it has written so far, so the
+statistics line lands in the log slightly after the request that triggered it.
+
+`ttl` and `vrf` are accepted for NAPALM compatibility and ignored — the API
+has no equivalent.
 
 ## Config management
 

napalm_opnsense/opnsense.py

@@ -50,12 +50,17 @@ 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
+from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interfaces
 
-class OPNsenseDriver(FirewallDriver):
+
+class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
     """NAPALM driver for OPNsense (read-only, REST API)."""
 
     VENDOR = "OPNsense"
     DRIVER_NAME = "opnsense"
+    # Everything runs over the REST API; there is no SSH session to open.
+    USES_SSH = False
     SNMP_FINGERPRINT = [
         FingerprintRule("opnsense", weight=8.0, mandatory=True),
     ]
@@ -218,6 +223,11 @@ class OPNsenseDriver(FirewallDriver):
         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]] = {}
@@ -238,6 +248,7 @@ class OPNsenseDriver(FirewallDriver):
                 "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
@@ -1059,6 +1070,156 @@ class OPNsenseDriver(FirewallDriver):
             })
         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.
 
@@ -1208,6 +1369,403 @@ class OPNsenseDriver(FirewallDriver):
 
         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.
 
@@ -1499,8 +2057,6 @@ class OPNsenseDriver(FirewallDriver):
                 if upgrades:
                     warnings.append({
                         "code": "updates_available",
-                        "severity": "info",
-                        "action": None,
                         "meta": {
                             "count": len(upgrades),
                             "packages": [u.get("name", "") for u in upgrades[:10]],
@@ -1571,10 +2127,30 @@ class OPNsenseDriver(FirewallDriver):
         """Remove an OPNsense plugin by name.
 
         Calls ``POST /api/core/firmware/remove/{name}``.
+
+        **Plugins only, and it says so.** ``firmware/remove`` acts on the
+        plugin set, and ``get_packages`` reads the same list — so software
+        installed as a plain FreeBSD package is invisible to the one and
+        unreachable by the other. The Wazuh agent is exactly that, on a driver
+        the agent plugin lists as supported, and during a fleet-wide rollback
+        the request posted for it could never have succeeded.
+
+        Raising beats posting. A request that cannot work reports failure for
+        the wrong reason and sends whoever reads it looking in the wrong place;
+        netOrk turns ``NotImplementedError`` into a 501, which is the accurate
+        answer. Reaching plain packages would need shell access, and the
+        credentials stored for these devices are frequently API-key only —
+        a decision of its own rather than a detail of this one.
         """
         import re as _re
         if not _re.match(r'^[a-zA-Z0-9_\-\.]+$', name):
             raise ValueError(f"Invalid package name: {name!r}")
+        if not name.startswith("os-"):
+            raise NotImplementedError(
+                f"{name!r} is not an OPNsense plugin. This driver can remove plugins "
+                "(os-*) through the firmware API; a FreeBSD package needs shell access, "
+                "which is not configured for OPNsense devices."
+            )
         try:
             result = self._post(f"/api/core/firmware/remove/{name}")
             return {"success": True, "output": str(result)}
@@ -1617,6 +2193,49 @@ class OPNsenseDriver(FirewallDriver):
         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":
@@ -1769,6 +2388,21 @@ class OPNsenseDriver(FirewallDriver):
 
         return {"success": success, "output": "\n".join(lines)}
 
+    def get_port_forwards(self) -> list[dict[str, Any]]:
+        """Destination NAT on the WAN interfaces, as port forwards.
+
+        Three reads: the rules, the interface overview (a WAN is an interface
+        with an upstream gateway) and the aliases a rule may send to. The
+        filtering and resolving is in :mod:`napalm_opnsense.port_forwards`.
+        A box without the destination-NAT API (older than its MVC rework)
+        raises: an empty list would claim "nothing forwarded", which nobody
+        checked.
+        """
+        rules = self._get("/api/firewall/d_nat/search_rule?current=1&rowCount=-1").get("rows", [])
+        overview = self._get("/api/interfaces/overview/interfaces_info?current=1&rowCount=-1")
+        aliases = self._get("/api/firewall/alias/searchItem?current=1&rowCount=-1").get("rows", [])
+        return port_forwards(rules, wan_interfaces(overview), alias_index(aliases))
+
     def get_firewall_aliases(self) -> list[dict[str, Any]]:
         """Return all firewall aliases, sorted by type then name.
 
@@ -1810,6 +2444,10 @@ class OPNsenseDriver(FirewallDriver):
         * ``floating``          — bool, rule applies across all interfaces
         * ``interface_label``   — human-readable interface description
         * ``is_group``          — bool, interface is an interface group
+        * ``gateway``           — the gateway a pass rule policy-routes to, or
+          ``""``. Such a rule sends what it matches to that gateway, local
+          destinations included, so it does not reach a host on another
+          internal network.
         """
         try:
             resp = self._get("/api/firewall/filter/searchRule?current=1&rowCount=-1")
@@ -1874,10 +2512,54 @@ class OPNsenseDriver(FirewallDriver):
                 "log":              str(row.get("log", "0")) == "1",
                 "enabled":          str(row.get("enabled", "1")) == "1",
                 "category":         category,
+                "gateway":          row.get("gateway", "") or "",
             })
 
         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
     # ------------------------------------------------------------------
@@ -2023,19 +2705,32 @@ class OPNsenseDriver(FirewallDriver):
                 "/api/unbound/settings/searchhostoverride",
                 {"current": 1, "rowCount": -1, "searchPhrase": ""},
             )
-            for row in data.get("rows", []):
+            rows = data.get("rows", []) or []
+            # searchhostoverride lists alias records alongside the host override
+            # they hang off, each under its own UUID. OPNsense flags them, so
+            # skip them by that flag and report every remaining row as it is —
+            # including two rows that happen to hold the same name and address.
+            # Those are duplicates on the device, and sync_dns_zone can only
+            # clear the copies it is told about.
+            flagged = any("isAlias" in row for row in rows)
+            for row in rows:
+                if flagged and row.get("isAlias"):
+                    continue
                 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.
+                if not flagged:
+                    # Releases without the flag give nothing to tell an alias
+                    # apart from a copy, so collapsing by content stays the
+                    # safer read there.
                     key = (host.lower(), domain.lower(), ip, rr)
                     if key in seen:
                         continue
                     seen.add(key)
+                # The auto-PTR flag is "addptr" since 25.x, "ptrrecord" before.
+                ptr = row.get("addptr", row.get("ptrrecord", "1"))
                 result.append({
                     "uuid": row.get("uuid", ""),
                     "hostname": host,
@@ -2045,7 +2740,7 @@ class OPNsenseDriver(FirewallDriver):
                     "record_type": rr,
                     "description": row.get("description", "") or "",
                     "enabled": str(row.get("enabled", "1")) == "1",
-                    "ptrrecord": str(row.get("ptrrecord", "1")) == "1",
+                    "ptrrecord": str(ptr) == "1",
                     "service": "unbound",
                 })
         except Exception as exc:
@@ -2151,14 +2846,21 @@ class OPNsenseDriver(FirewallDriver):
                     logger.info("Removed %s host override %s.%s (uuid %s)",
                                 service, entry["hostname"], zone_clean, uid)
 
-        # Re-add all enabled A/AAAA records
+        # Re-add all enabled A/AAAA records, one override per logical record.
+        # Two rows for the same name is a state netOrk's own database can
+        # legitimately be in; the device must still end up with one.
         added = 0
+        pushed: set[tuple[str, str, str]] = set()
         for rec in records:
             if not rec.get("enabled", True):
                 continue
             rtype = rec.get("record_type", "A")
             if rtype not in ("A", "AAAA"):
                 continue
+            key = (str(rec.get("hostname", "")).lower(), rtype, str(rec.get("ip", "")))
+            if key in pushed:
+                continue
+            pushed.add(key)
             if service == "unbound":
                 self._post("/api/unbound/settings/addhostoverride", {
                     "host": {
@@ -2168,6 +2870,7 @@ class OPNsenseDriver(FirewallDriver):
                         "rr": rtype,
                         "server": rec.get("ip", ""),
                         "description": self._NETORK_TAG,
+                        "addptr": "1",
                         "ptrrecord": "1",
                         "mxprio": "",
                         "mx": "",

napalm_opnsense/ping_mixin.py

@@ -0,0 +1,385 @@
+# -*- 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.
+
+"""Ping support for OPNsense — NAPALM ``ping()`` plus a batched ``ping_sweep()``.
+
+OPNsense has no synchronous "ping once and tell me the result" endpoint.
+``/api/diagnostics/ping`` is a *job* API: a job is created (``set``), started
+(``start``), then runs in the background writing statistics that are read back
+via ``search_jobs``, and finally has to be stopped and removed again. A single
+ping therefore costs five API calls and roughly one second of waiting.
+
+That makes the generic sequential sweep in ``napalm-device-types`` far too slow
+here, but it also hands us something better: jobs are independent and run on
+the firewall in parallel, and ``search_jobs`` reports *all* of them in one
+response. :meth:`OPNsensePingMixin.ping_sweep` therefore works in batches —
+create and start a whole batch, wait once, read every result with a single
+request, then clean the batch up. Cost per batch is constant in waiting time
+instead of linear in hosts.
+
+``PING_SWEEP_BATCH_SIZE`` bounds how many ping processes the firewall is asked
+to run at once; ``PING_SWEEP_MAX_TARGETS`` bounds the sweep as a whole. Both
+are deliberately conservative — this runs on production firewalls.
+
+Why the results are *polled* rather than read once: ``search_jobs`` signals the
+running ping with ``SIGINFO`` and then parses whatever the process has written
+to its log so far. The statistics line therefore lands in the log slightly
+after the request that triggered it, so the first read of a healthy host can
+still show zero probes. Every job is asked repeatedly until it reports the
+requested probe count or the time budget runs out.
+
+Not supported by the API and therefore ignored: ``ttl``, ``vrf``.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from ipaddress import ip_address
+from typing import Any, Callable, Iterable, Optional
+
+logger = logging.getLogger(__name__)
+
+_PING_API = "/api/diagnostics/ping"
+
+#: Where the ping fields live when the model cannot be read. Matches what
+#: OPNsense 24/25 ship: {"ping": {"settings": {...}}}.
+_DEFAULT_PING_MODEL_PATH = ("ping", "settings")
+
+#: Seconds between two ``search_jobs`` polls while waiting for probe results.
+_POLL_INTERVAL = 0.5
+
+#: The API's ``interval`` field is in whole seconds and has a minimum of 1,
+#: so one probe takes at least this long.
+_PROBE_INTERVAL_SECONDS = 1
+
+#: Extra grace on top of the expected probe duration before giving up on a job.
+_POLL_GRACE_SECONDS = 1.0
+
+
+class OPNsensePingMixin:
+    """NAPALM ``ping()`` / ``ping_sweep()`` on the OPNsense diagnostics ping API."""
+
+    #: Ping jobs started on the firewall at the same time.
+    PING_SWEEP_BATCH_SIZE: int = 32
+
+    #: Hosts probed in a single sweep. A /24 fits; a /16 must be split by
+    #: the caller — 65k background jobs is not something to point at a firewall.
+    PING_SWEEP_MAX_TARGETS: int = 512
+
+    # Provided by the driver this mixin is mixed into.
+    _ping_model_path_cache: Optional[tuple[str, ...]] = None
+
+    # ── NAPALM ───────────────────────────────────────────────────────────────
+
+    def ping(
+        self,
+        destination: str,
+        source: str = "",
+        ttl: int = 255,
+        timeout: int = 2,
+        size: int = 100,
+        count: int = 5,
+        vrf: str = "",
+    ) -> dict[str, Any]:
+        """Ping *destination* from the firewall.
+
+        Runs the full job lifecycle and returns NAPALM's standard ping shape:
+        ``{"success": {...}}`` with probe counts and RTTs, or ``{"error": ...}``
+        if the job could not be created or its statistics could not be read.
+        """
+        try:
+            job_id = self._ping_job_create(destination, source=source, size=size)
+        except Exception as exc:  # noqa: BLE001 - reported to the caller as-is
+            return {"error": str(exc)}
+
+        try:
+            self._post(f"{_PING_API}/start/{job_id}", {})
+            rows = self._await_ping_rows([job_id], count=count, timeout=timeout)
+        except Exception as exc:  # noqa: BLE001
+            return {"error": str(exc)}
+        finally:
+            self._ping_job_cleanup([job_id])
+
+        return self._row_to_napalm_reply(destination, rows.get(job_id), count=count)
+
+    def ping_sweep(
+        self,
+        destinations: Iterable[str],
+        *,
+        count: int = 1,
+        timeout: int = 1,
+        max_targets: Optional[int] = None,
+        on_progress: Optional[Callable[[int, int], None]] = None,
+        should_stop: Optional[Callable[[], bool]] = None,
+    ) -> dict[str, Any]:
+        """Ping many destinations, one batch of parallel firewall jobs at a time.
+
+        Same contract as
+        :meth:`napalm_device_types.ping_sweep.PingSweepMixin.ping_sweep`;
+        ``should_stop`` is polled between batches rather than between hosts,
+        because a batch is the smallest unit of work here.
+        """
+        limit = self.PING_SWEEP_MAX_TARGETS if max_targets is None else max_targets
+        targets = list(destinations)
+        truncated = len(targets) > limit
+        if truncated:
+            targets = targets[:limit]
+
+        total = len(targets)
+        entries: list[dict[str, Any]] = []
+        for start in range(0, total, self.PING_SWEEP_BATCH_SIZE):
+            if should_stop is not None and should_stop():
+                break
+            batch = targets[start : start + self.PING_SWEEP_BATCH_SIZE]
+            entries.extend(self._sweep_batch(batch, count=count, timeout=timeout))
+            if on_progress is not None:
+                on_progress(len(entries), total)
+
+        return {
+            "entries": entries,
+            "scanned": len(entries),
+            "alive_count": sum(1 for entry in entries if entry["alive"]),
+            "truncated": truncated,
+        }
+
+    # ── Batch execution ──────────────────────────────────────────────────────
+
+    def _sweep_batch(
+        self, destinations: list[str], *, count: int, timeout: int
+    ) -> list[dict[str, Any]]:
+        """Create, start, harvest and remove one batch of ping jobs."""
+        job_ids: dict[str, str] = {}  # destination -> job id
+        errors: dict[str, str] = {}  # destination -> creation error
+
+        for destination in destinations:
+            try:
+                job_ids[destination] = self._ping_job_create(destination)
+            except Exception as exc:  # noqa: BLE001 - one bad host must not kill the batch
+                errors[destination] = str(exc)
+
+        rows: dict[str, dict[str, Any]] = {}
+        batch_error: str | None = None
+        try:
+            for job_id in job_ids.values():
+                self._post(f"{_PING_API}/start/{job_id}", {})
+            rows = self._await_ping_rows(list(job_ids.values()), count=count, timeout=timeout)
+        except Exception as exc:  # noqa: BLE001 - the whole batch loses its results
+            batch_error = str(exc)
+        finally:
+            self._ping_job_cleanup(list(job_ids.values()))
+
+        entries: list[dict[str, Any]] = []
+        for destination in destinations:
+            if destination in errors:
+                entries.append(self._parse_ping_reply(destination, {"error": errors[destination]}))
+            elif batch_error is not None:
+                entries.append(self._parse_ping_reply(destination, {"error": batch_error}))
+            else:
+                reply = self._row_to_napalm_reply(
+                    destination, rows.get(job_ids[destination]), count=count
+                )
+                entries.append(self._parse_ping_reply(destination, reply))
+        return entries
+
+    # ── Job lifecycle ────────────────────────────────────────────────────────
+
+    def _ping_job_create(self, destination: str, *, source: str = "", size: int = 100) -> str:
+        """Create a ping job for *destination* and return its job id."""
+        settings = {
+            "hostname": destination,
+            "fam": self._address_family(destination),
+            "packetsize": str(size),
+            "interval": str(_PROBE_INTERVAL_SECONDS),
+            "description": "netork ping sweep",
+        }
+        if source:
+            settings["source_address"] = source
+
+        response = self._post(f"{_PING_API}/set", self._wrap_in_model(settings))
+        job_id = response.get("uuid") or response.get("id")
+        if not job_id:
+            raise RuntimeError(f"ping job creation failed for {destination}: {response}")
+        return str(job_id)
+
+    def _ping_job_cleanup(self, job_ids: list[str]) -> None:
+        """Stop and remove *job_ids*, never raising — cleanup is best effort.
+
+        Every job left behind keeps a ping process and a log file in ``/tmp/ping``
+        on the firewall, so this runs even when the sweep itself failed.
+        """
+        for job_id in job_ids:
+            for action in ("stop", "remove"):
+                try:
+                    self._post(f"{_PING_API}/{action}/{job_id}", {})
+                except Exception as exc:  # noqa: BLE001
+                    logger.debug("ping job %s: %s failed: %s", job_id, action, exc)
+
+    def _await_ping_rows(
+        self, job_ids: list[str], *, count: int, timeout: int
+    ) -> dict[str, dict[str, Any]]:
+        """Poll ``search_jobs`` until every job sent *count* probes, or time is up.
+
+        Returns the last rows seen, keyed by job id — a job that never reported
+        is simply absent, which the caller reads as "no answer".
+        """
+        if not job_ids:
+            return {}
+
+        wanted = set(job_ids)
+        budget = max(timeout, 1) * max(count, 1) + _POLL_GRACE_SECONDS
+        attempts = max(1, int(budget / _POLL_INTERVAL))
+        rows: dict[str, dict[str, Any]] = {}
+
+        for attempt in range(attempts):
+            if attempt:
+                time.sleep(_POLL_INTERVAL)
+            rows = {
+                job_id: row
+                for job_id, row in self._read_ping_jobs().items()
+                if job_id in wanted
+            }
+            if wanted and all(
+                _as_int(rows.get(job_id, {}).get("send")) >= count for job_id in wanted
+            ):
+                break
+        return rows
+
+    def _read_ping_jobs(self) -> dict[str, dict[str, Any]]:
+        """All ping jobs currently known to the firewall, keyed by job id."""
+        response = self._get(f"{_PING_API}/search_jobs")
+        rows = response.get("rows") or []
+        result: dict[str, dict[str, Any]] = {}
+        for row in rows:
+            job_id = row.get("uuid") or row.get("id")
+            if job_id:
+                result[str(job_id)] = row
+        return result
+
+    def _wrap_in_model(self, settings: dict[str, Any]) -> dict[str, Any]:
+        """Nest *settings* under the keys ``set`` expects, outermost last."""
+        payload: dict[str, Any] = settings
+        for key in reversed(self._ping_model_path()):
+            payload = {key: payload}
+        return payload
+
+    def _ping_model_path(self) -> tuple[str, ...]:
+        """Keys from the model root down to the fields, e.g. ``("ping", "settings")``.
+
+        Read once from ``GET /api/diagnostics/ping/get`` rather than hardcoded,
+        so a model rename in a future release does not silently break job
+        creation. The response mirrors the model, so the path is the chain of
+        single-dict wrappers around the fields::
+
+            {"ping": {"settings": {"hostname": "", "fam": {...}, ...}}}
+
+        The descent stops at the first level that holds more than one key —
+        that is the field level, where ``fam`` is a dict too and following it
+        would land inside a form field.
+
+        Getting this wrong is expensive and quiet: OPNsense answers a job whose
+        hostname arrived at the wrong node with ``ping.settings.hostname: A
+        value is required``, and a sweep turns 254 such refusals into a network
+        that appears to hold nothing.
+        """
+        if self._ping_model_path_cache is not None:
+            return self._ping_model_path_cache
+
+        path = _DEFAULT_PING_MODEL_PATH
+        try:
+            node = self._get(f"{_PING_API}/get")
+            found: list[str] = []
+            while isinstance(node, dict) and len(node) == 1:
+                key, value = next(iter(node.items()))
+                if not isinstance(value, dict):
+                    break
+                found.append(key)
+                node = value
+            if found:
+                path = tuple(found)
+        except Exception as exc:  # noqa: BLE001 - the default is what firmware ships
+            logger.debug("Could not read ping model path, assuming %r: %s", path, exc)
+
+        self._ping_model_path_cache = path
+        return path
+
+    # ── Parsing ──────────────────────────────────────────────────────────────
+
+    @staticmethod
+    def _address_family(destination: str) -> str:
+        """``"ip"`` for IPv4 / hostnames, ``"ip6"`` for IPv6 literals."""
+        try:
+            return "ip6" if ip_address(destination).version == 6 else "ip"
+        except ValueError:
+            return "ip"
+
+    @staticmethod
+    def _row_to_napalm_reply(
+        destination: str, row: Optional[dict[str, Any]], *, count: int
+    ) -> dict[str, Any]:
+        """Turn one ``search_jobs`` row into NAPALM's ping reply shape.
+
+        A missing row, or one that never sent a probe, is reported as "all
+        probes lost" rather than as an error: from the caller's point of view
+        an unreachable host and a host the firewall never got around to
+        pinging look the same, and the sweep only asks "did it answer?".
+        """
+        if row is None:
+            return _all_lost(count)
+
+        sent = _as_int(row.get("send"))
+        received = _as_int(row.get("received"))
+        if sent == 0:
+            return _all_lost(count)
+
+        avg = _as_float(row.get("avg"))
+        return {
+            "success": {
+                "probes_sent": sent,
+                "packet_loss": max(sent - received, 0),
+                "rtt_min": _as_float(row.get("min")),
+                "rtt_avg": avg,
+                "rtt_max": _as_float(row.get("max")),
+                "rtt_stddev": _as_float(row.get("std-dev")),
+                "results": [{"ip_address": destination, "rtt": avg}] if received else [],
+            }
+        }
+
+
+def _all_lost(count: int) -> dict[str, Any]:
+    probes = max(count, 1)
+    return {
+        "success": {
+            "probes_sent": probes,
+            "packet_loss": probes,
+            "rtt_min": 0.0,
+            "rtt_avg": 0.0,
+            "rtt_max": 0.0,
+            "rtt_stddev": 0.0,
+            "results": [],
+        }
+    }
+
+
+def _as_int(value: Any, default: int = 0) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _as_float(value: Any, default: float = 0.0) -> float:
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return default

napalm_opnsense/port_forwards.py

@@ -0,0 +1,159 @@
+"""Destination NAT on the WAN, read as port forwards. Pure: no I/O.
+
+OPNsense keeps every destination-NAT rule in one list
+(``/api/firewall/d_nat/search_rule``): the forwards from the internet, and
+also redirects between internal networks, anti-lockout rules that only exempt
+traffic (``nordr``), and rules the captive portal generates
+(``is_automatic``). The contract this serves --
+``napalm_device_types.NatVpnMixin.get_port_forwards`` -- wants the first kind
+only, because callers read every entry as "this host is reachable from
+outside".
+
+What counts as the WAN is an interface with an upstream gateway, read from
+the interface overview. That catches a second uplink (an LTE backup) as well
+as the one named ``wan``.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import socket
+from typing import Any, Dict, Iterable, List, Optional, Tuple
+
+#: Port names OPNsense accepts in a rule. ``socket.getservbyname`` knows them
+#: too, but only where ``/etc/services`` exists -- a slim container has none.
+WELL_KNOWN_PORTS: Dict[str, int] = {
+    "ftp": 21,
+    "ssh": 22,
+    "telnet": 23,
+    "smtp": 25,
+    "domain": 53,
+    "dns": 53,
+    "http": 80,
+    "pop3": 110,
+    "ntp": 123,
+    "imap": 143,
+    "snmp": 161,
+    "ldap": 389,
+    "https": 443,
+    "smtps": 465,
+    "submission": 587,
+    "ldaps": 636,
+    "imaps": 993,
+    "pop3s": 995,
+    "openvpn": 1194,
+    "ms-wbt-server": 3389,
+    "rdp": 3389,
+}
+
+#: Alias types whose entries can be addresses.
+_ADDRESS_ALIASES = ("host", "network")
+
+AliasIndex = Dict[str, Tuple[str, List[str]]]
+
+
+def wan_interfaces(overview: Any) -> set:
+    """Identifiers of the interfaces that have an upstream gateway."""
+    rows = overview.get("rows", []) if isinstance(overview, dict) else overview or []
+    return {r["identifier"] for r in rows if r.get("identifier") and r.get("gateways")}
+
+
+def alias_index(rows: Iterable[dict]) -> AliasIndex:
+    """``name -> (type, entries)`` from the alias list."""
+    return {
+        r["name"]: (r.get("type", ""), [e.strip() for e in str(r.get("content", "")).splitlines() if e.strip()])
+        for r in rows
+        if r.get("name")
+    }
+
+
+def _as_address(value: str) -> Optional[str]:
+    try:
+        return str(ipaddress.ip_address(value))
+    except ValueError:
+        return None
+
+
+def _addresses(target: str, aliases: AliasIndex) -> List[str]:
+    """The addresses a rule sends to: a literal, or a host/network alias's.
+
+    Anything else -- an interface address, a name resolved by DNS -- gives no
+    address, and the rule is left out rather than put on a guessed host.
+    """
+    literal = _as_address(target)
+    if literal:
+        return [literal]
+    kind, entries = aliases.get(target, ("", []))
+    if kind not in _ADDRESS_ALIASES:
+        return []
+    return [a for a in (_as_address(e) for e in entries) if a]
+
+
+def _port(value: Any, protocol: str, aliases: AliasIndex) -> Optional[int]:
+    """A rule's port as a number: literal, start of a range, alias or name.
+
+    No port at all means every port, which the contract writes as 0.
+    """
+    text = str(value).strip()
+    if not text:
+        return 0
+    first = text.replace(":", "-").split("-")[0].strip()
+    if first.isdigit():
+        return int(first)
+    kind, entries = aliases.get(text, ("", []))
+    if kind == "port" and entries:
+        return _port(entries[0], protocol, aliases)
+    try:
+        return socket.getservbyname(text, protocol)
+    except OSError:
+        return WELL_KNOWN_PORTS.get(text.lower())
+
+
+def _faces_the_wan(rule: dict, wan: set) -> bool:
+    if rule.get("nordr") == "1" or rule.get("is_automatic"):
+        return False
+    return bool(set(str(rule.get("interface", "")).split(",")) & wan)
+
+
+def _remote_host(rule: dict) -> Optional[str]:
+    """A source restriction; an inverted one ("all but X") restricts nothing."""
+    source = str(rule.get("source.network") or "").strip()
+    if source in ("", "any") or rule.get("source.not") == "1":
+        return None
+    return source
+
+
+def _forwards_of(rule: dict, aliases: AliasIndex) -> List[dict]:
+    protocols = [p.upper() for p in str(rule.get("protocol") or "any").split("/") if p]
+    target = str(rule.get("target", "")).strip()
+    remote = _remote_host(rule)
+    result: List[dict] = []
+    for protocol in protocols:
+        external = _port(rule.get("destination.port", ""), protocol.lower(), aliases)
+        if external is None:
+            continue
+        local = rule.get("local-port")
+        internal = _port(local, protocol.lower(), aliases) if str(local or "").strip() else external
+        name = rule.get("descr") or f"{protocol} {external} -> {target}"
+        for address in _addresses(target, aliases):
+            entry = {
+                "name": name,
+                "protocol": protocol,
+                "external_port": external,
+                "internal_ip": address,
+                "internal_port": internal if internal is not None else external,
+                "enabled": rule.get("disabled") != "1",
+            }
+            if remote:
+                entry["remote_host"] = remote
+            result.append(entry)
+    return result
+
+
+def port_forwards(rules: Iterable[dict], wan: set, aliases: AliasIndex) -> List[dict]:
+    """The destination-NAT rules on a WAN interface, as ``PortForwardDict`` entries."""
+    result: List[dict] = []
+    for rule in rules:
+        if _faces_the_wan(rule, wan):
+            result.extend(_forwards_of(rule, aliases))
+    return result

pyproject.toml

@@ -8,7 +8,7 @@ version = "0.1.0"
 description = "NAPALM driver for OPNsense (read-only via REST API)."
 readme = "README.md"
 license = { text = "Apache-2.0" }
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 authors = [
     { name = "Christian Manivong" },
 ]
@@ -16,7 +16,6 @@ classifiers = [
     "Topic :: Utilities",
     "License :: OSI Approved :: Apache Software License",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",

tests/unit/test_firewall_rules.py

@@ -0,0 +1,78 @@
+"""Filter rules as ``get_firewall_rules`` reports them.
+
+A pass rule with a gateway is policy routing: OPNsense hands what it matches to
+that gateway, local destinations included. A caller judging which network can
+reach which host has to know that, or a rule meant for internet traffic reads
+as a hole into every server (netOrk #575: on the first real box, "pass UDP
+IOT -> any" via WAN_GW made 61 hosts look reachable from the IoT segment).
+
+The row shape follows that box's ``/api/firewall/filter/searchRule``.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from napalm_opnsense.opnsense import OPNsenseDriver
+
+
+@pytest.fixture
+def driver():
+    with patch("napalm_opnsense.opnsense.requests.Session"):
+        yield OPNsenseDriver(
+            hostname="opnsense.example.com",
+            username="api_key",
+            password="api_secret",
+            optional_args={"verify": False},
+        )
+
+
+def _row(**over) -> dict:
+    """One rule row as the API returns it; booleans are "0"/"1" strings."""
+    row = {
+        "uuid": "u1",
+        "sequence": "1",
+        "action": "pass",
+        "quick": "1",
+        "interface": "opt3",
+        "%interface": "IOT",
+        "direction": "in",
+        "ipprotocol": "inet",
+        "protocol": "UDP",
+        "%source_net": "HOME_OFFICE_IOT_NET",
+        "%destination_net": "any",
+        "destination_port": "",
+        "description": "reolink_udp_long_state_timeout",
+        "enabled": "1",
+        "gateway": "WAN_GW",
+    }
+    row.update(over)
+    return row
+
+
+def _rules(driver, *rows):
+    def fake_get(path):
+        return {"rows": list(rows)} if "searchRule" in path else {"rows": []}
+
+    with patch.object(driver, "_get", side_effect=fake_get):
+        return driver.get_firewall_rules()
+
+
+def test_a_policy_routed_rule_reports_its_gateway(driver):
+    [rule] = _rules(driver, _row(gateway="WAN_GW"))
+    assert rule["gateway"] == "WAN_GW"
+
+
+def test_a_rule_that_routes_normally_reports_no_gateway(driver):
+    [rule] = _rules(driver, _row(gateway=""))
+    assert rule["gateway"] == ""
+
+
+def test_a_row_without_the_field_reports_no_gateway(driver):
+    # Older firmware, or a rule type that never carries one.
+    row = _row()
+    del row["gateway"]
+    [rule] = _rules(driver, row)
+    assert rule["gateway"] == ""

tests/unit/test_ping.py

@@ -0,0 +1,382 @@
+"""Unit tests for OPNsense ping / ping_sweep — no real device required.
+
+The OPNsense ping API is job-based: a job is created (``set``), started
+(``start``), produces statistics that are read back via ``search_jobs``, and
+has to be stopped and removed again. These tests drive that whole lifecycle
+against a fake ``_get`` / ``_post``.
+"""
+
+import pytest
+from unittest.mock import MagicMock, patch
+
+from napalm_opnsense.opnsense import OPNsenseDriver
+
+
+# ---------------------------------------------------------------------------
+# Fake API
+# ---------------------------------------------------------------------------
+
+
+class FakePingAPI:
+    """Minimal stand-in for the OPNsense diagnostics ping controller."""
+
+    def __init__(self, stats=None, model_path=("ping", "settings")):
+        #: hostname -> row returned by search_jobs
+        self.stats = stats or {}
+        #: Keys from the model root down to the field level. A real OPNsense
+        #: nests them two deep ({"ping": {"settings": {...}}}); the tests used
+        #: to assume one, which is why nothing caught the payload going to the
+        #: wrong node.
+        self.model_path = tuple(model_path)
+        self.created = []      # payloads passed to set
+        self.started = []      # job ids passed to start
+        self.stopped = []      # job ids passed to stop
+        self.removed = []      # job ids passed to remove
+        self.jobs = {}         # job id -> hostname
+        self.search_calls = 0
+        self._next_id = 0
+
+    def get(self, path):
+        if path == "/api/diagnostics/ping/get":
+            fields = {
+                "hostname": "",
+                "fam": {"ip": {"value": "IPv4", "selected": 1}},
+                "source_address": "",
+                "packetsize": "",
+                "description": "",
+            }
+            return self._nest(fields)
+        if path == "/api/diagnostics/ping/search_jobs":
+            self.search_calls += 1
+            return {"rows": [self._row(jid, host) for jid, host in self.jobs.items()]}
+        raise AssertionError(f"unexpected GET {path}")
+
+    def post(self, path, data=None):
+        if path == "/api/diagnostics/ping/set":
+            self.created.append(data)
+            self._next_id += 1
+            job_id = f"job-{self._next_id}"
+            self.jobs[job_id] = self.fields(data or {}).get("hostname", "")
+            return {"result": "saved", "uuid": job_id}
+        for action, sink in (("start", self.started), ("stop", self.stopped)):
+            if path.startswith(f"/api/diagnostics/ping/{action}/"):
+                sink.append(path.rsplit("/", 1)[1])
+                return {"status": "ok"}
+        if path.startswith("/api/diagnostics/ping/remove/"):
+            job_id = path.rsplit("/", 1)[1]
+            self.removed.append(job_id)
+            self.jobs.pop(job_id, None)
+            return {"status": "ok"}
+        raise AssertionError(f"unexpected POST {path}")
+
+    def _nest(self, fields):
+        """Wrap *fields* in the model path, innermost first."""
+        node = fields
+        for key in reversed(self.model_path):
+            node = {key: node}
+        return node
+
+    def fields(self, payload):
+        """The field level of a posted payload, or {} if it went to the wrong node."""
+        node = payload
+        for key in self.model_path:
+            if not isinstance(node, dict) or key not in node:
+                return {}
+            node = node[key]
+        return node if isinstance(node, dict) else {}
+
+    def _row(self, job_id, hostname):
+        row = {"id": job_id, "hostname": hostname, "status": "running"}
+        row.update(self.stats.get(hostname, {"send": 0, "received": 0}))
+        return row
+
+
+def _driver(fake):
+    """An OPNsenseDriver wired to *fake* instead of a real firewall."""
+    with patch("napalm_opnsense.opnsense.requests.Session"):
+        drv = OPNsenseDriver(hostname="fw", username="k", password="s")
+    drv.session = MagicMock()
+    drv._get = fake.get
+    drv._post = fake.post
+    return drv
+
+
+def alive_row(rtt=1.5, send=1, received=1):
+    return {
+        "send": send,
+        "received": received,
+        "loss": "0.0 %",
+        "min": rtt,
+        "avg": rtt,
+        "max": rtt,
+        "std-dev": 0.0,
+        "last_error": None,
+    }
+
+
+def dead_row(send=1):
+    return {
+        "send": send,
+        "received": 0,
+        "loss": "100.0 %",
+        "min": 0.0,
+        "avg": 0.0,
+        "max": 0.0,
+        "std-dev": 0.0,
+        "last_error": None,
+    }
+
+
+@pytest.fixture
+def api():
+    return FakePingAPI()
+
+
+@pytest.fixture
+def driver(api):
+    """Driver whose HTTP layer is replaced by the fake ping API."""
+    with patch("napalm_opnsense.opnsense.requests.Session"):
+        drv = OPNsenseDriver(
+            hostname="fw.example.com",
+            username="api_key",
+            password="api_secret",
+            optional_args={"verify": False},
+        )
+    drv.session = MagicMock()
+    drv._get = api.get
+    drv._post = api.post
+    with patch("napalm_opnsense.ping_mixin.time.sleep"):
+        yield drv
+
+
+# ---------------------------------------------------------------------------
+# ping()
+# ---------------------------------------------------------------------------
+
+
+def test_ping_reachable_host_returns_napalm_success(driver, api):
+    api.stats["10.0.0.1"] = alive_row(rtt=2.5, send=2, received=2)
+
+    result = driver.ping("10.0.0.1", count=2)
+
+    assert result["success"]["probes_sent"] == 2
+    assert result["success"]["packet_loss"] == 0
+    assert result["success"]["rtt_avg"] == 2.5
+    assert result["success"]["results"] == [{"ip_address": "10.0.0.1", "rtt": 2.5}]
+
+
+def test_ping_unreachable_host_reports_full_packet_loss(driver, api):
+    api.stats["10.0.0.9"] = dead_row(send=2)
+
+    result = driver.ping("10.0.0.9", count=2)
+
+    assert result["success"]["probes_sent"] == 2
+    assert result["success"]["packet_loss"] == 2
+    assert result["success"]["results"] == []
+
+
+def test_ping_creates_job_with_destination_and_packetsize(driver, api):
+    api.stats["10.0.0.1"] = alive_row()
+
+    driver.ping("10.0.0.1", size=64, source="10.0.0.254")
+
+    assert api.fields(api.created[0]) == {
+        "hostname": "10.0.0.1",
+        "fam": "ip",
+        "packetsize": "64",
+        "interval": "1",
+        "source_address": "10.0.0.254",
+        "description": "netork ping sweep",
+    }
+
+
+def test_ping_uses_ipv6_family_for_v6_destination(driver, api):
+    api.stats["2001:db8::1"] = alive_row()
+
+    driver.ping("2001:db8::1")
+
+    assert api.fields(api.created[0])["fam"] == "ip6"
+
+
+def test_ping_starts_stops_and_removes_the_job(driver, api):
+    api.stats["10.0.0.1"] = alive_row()
+
+    driver.ping("10.0.0.1")
+
+    assert api.started == ["job-1"]
+    assert api.stopped == ["job-1"]
+    assert api.removed == ["job-1"]
+
+
+def test_ping_removes_job_even_when_reading_stats_fails(driver, api):
+    def _boom(path):
+        if path == "/api/diagnostics/ping/search_jobs":
+            raise RuntimeError("API down")
+        return api.get(path)
+
+    driver._get = _boom
+
+    result = driver.ping("10.0.0.1")
+
+    assert "error" in result
+    assert api.removed == ["job-1"]
+
+
+def test_ping_returns_error_when_job_creation_fails(driver, api):
+    api.post = lambda path, data=None: {"result": "failed", "validations": {"hostname": "bad"}}
+    driver._post = api.post
+
+    result = driver.ping("nope")
+
+    assert "error" in result
+    assert "hostname" in result["error"]
+
+
+def test_ping_settings_go_to_the_node_the_api_asks_for():
+    """A real OPNsense nests the ping model two deep — {"ping": {"settings":
+    {...}}} — and rejects a job whose hostname arrives anywhere else with
+    "ping.settings.hostname: A value is required". Posting the fields one level
+    too high made every probe fail, which a sweep reported as a silent network."""
+    fake = FakePingAPI()
+    fake.stats["10.0.0.1"] = alive_row()
+    drv = _driver(fake)
+
+    with patch("napalm_opnsense.ping_mixin.time.sleep"):
+        reply = drv.ping("10.0.0.1")
+
+    assert fake.created[0] == {
+        "ping": {
+            "settings": {
+                "hostname": "10.0.0.1",
+                "fam": "ip",
+                "packetsize": "100",
+                "interval": "1",
+                "description": "netork ping sweep",
+            }
+        }
+    }
+    assert reply["success"]["probes_sent"] == 1
+
+
+def test_a_model_that_is_only_one_level_deep_still_works():
+    """Older firmware exposes the fields directly under one node. The path is
+    read from the API rather than assumed, so both shapes work."""
+    fake = FakePingAPI(model_path=("settings",))
+    fake.stats["10.0.0.1"] = alive_row()
+    drv = _driver(fake)
+
+    with patch("napalm_opnsense.ping_mixin.time.sleep"):
+        drv.ping("10.0.0.1")
+
+    assert list(fake.created[0]) == ["settings"]
+    assert fake.created[0]["settings"]["hostname"] == "10.0.0.1"
+
+
+# ---------------------------------------------------------------------------
+# ping_sweep()
+# ---------------------------------------------------------------------------
+
+
+def test_ping_sweep_probes_all_destinations_in_parallel_batches(driver, api):
+    api.stats["10.0.0.1"] = alive_row(rtt=1.0)
+    api.stats["10.0.0.3"] = alive_row(rtt=3.0)
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2", "10.0.0.3"])
+
+    assert result["scanned"] == 3
+    assert result["alive_count"] == 2
+    assert result["entries"] == [
+        {"ip": "10.0.0.1", "alive": True, "rtt_ms": 1.0},
+        {"ip": "10.0.0.2", "alive": False, "rtt_ms": None},
+        {"ip": "10.0.0.3", "alive": True, "rtt_ms": 3.0},
+    ]
+
+
+def test_ping_sweep_starts_one_job_per_destination_and_cleans_up(driver, api):
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"])
+
+    assert len(api.created) == 2
+    assert sorted(api.started) == ["job-1", "job-2"]
+    assert sorted(api.removed) == ["job-1", "job-2"]
+    assert result["scanned"] == 2
+
+
+def test_ping_sweep_reads_stats_once_per_batch_not_once_per_host(driver, api):
+    driver.PING_SWEEP_BATCH_SIZE = 8
+    api.stats.update({f"10.0.0.{i}": alive_row() for i in range(1, 5)})
+
+    driver.ping_sweep([f"10.0.0.{i}" for i in range(1, 5)])
+
+    assert api.search_calls == 1
+
+
+def test_ping_sweep_splits_into_batches(driver, api):
+    driver.PING_SWEEP_BATCH_SIZE = 2
+    api.stats.update({f"10.0.0.{i}": alive_row() for i in range(1, 6)})
+
+    result = driver.ping_sweep([f"10.0.0.{i}" for i in range(1, 6)])
+
+    assert api.search_calls == 3  # 2 + 2 + 1
+    assert result["scanned"] == 5
+
+
+def test_ping_sweep_truncates_at_max_targets(driver, api):
+    result = driver.ping_sweep([f"10.0.0.{i}" for i in range(1, 11)], max_targets=3)
+
+    assert result["scanned"] == 3
+    assert result["truncated"] is True
+    assert len(api.created) == 3
+
+
+def test_ping_sweep_stops_between_batches_when_asked(driver, api):
+    driver.PING_SWEEP_BATCH_SIZE = 2
+    calls = {"n": 0}
+
+    def _should_stop():
+        calls["n"] += 1
+        return calls["n"] > 1
+
+    result = driver.ping_sweep(
+        [f"10.0.0.{i}" for i in range(1, 6)],
+        should_stop=_should_stop,
+    )
+
+    assert result["scanned"] == 2
+
+
+def test_ping_sweep_reports_progress_per_batch(driver, api):
+    driver.PING_SWEEP_BATCH_SIZE = 2
+    seen = []
+
+    driver.ping_sweep(
+        [f"10.0.0.{i}" for i in range(1, 6)],
+        on_progress=lambda done, total: seen.append((done, total)),
+    )
+
+    assert seen == [(2, 5), (4, 5), (5, 5)]
+
+
+def test_ping_sweep_marks_hosts_as_errored_when_a_batch_fails(driver, api):
+    def _boom(path):
+        if path == "/api/diagnostics/ping/search_jobs":
+            raise RuntimeError("API down")
+        return api.get(path)
+
+    driver._get = _boom
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"])
+
+    assert result["alive_count"] == 0
+    assert all(entry["error"] == "API down" for entry in result["entries"])
+    assert sorted(api.removed) == ["job-1", "job-2"]
+
+
+def test_ping_sweep_on_empty_list_touches_no_api(driver, api):
+    result = driver.ping_sweep([])
+
+    assert result == {"entries": [], "scanned": 0, "alive_count": 0, "truncated": False}
+    assert api.created == []
+
+
+def test_opnsense_driver_reports_ping_support():
+    assert OPNsenseDriver.supports_ping() is True

tests/unit/test_port_forwards.py

@@ -0,0 +1,217 @@
+"""Destination NAT on the WAN, read as port forwards.
+
+OPNsense lists every destination-NAT rule in one place: the forwards from the
+internet, but also redirects between internal networks, anti-lockout rules
+that only exempt traffic, and rules the captive portal generates. The
+contract (``NatVpnMixin.get_port_forwards``) wants the first kind only: a
+caller reads each entry as "this host is reachable from outside". On the
+first real box (OPNsense 26.7, 2026-10-03) that was 2 of 22 rules.
+
+The row shapes below follow that box's ``/api/firewall/d_nat/search_rule``,
+``/api/interfaces/overview/interfaces_info`` and alias list, with the
+addresses replaced.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from napalm_opnsense.opnsense import OPNsenseDriver
+from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interfaces
+
+INTERFACES = [
+    {"identifier": "lan", "description": "MGMT", "gateways": []},
+    {"identifier": "wan", "description": "WAN", "gateways": ["192.0.2.1"]},
+    {"identifier": "opt6", "description": "WAN4G", "gateways": ["198.51.100.1"]},
+    {"identifier": "opt9", "description": "HOMEOFFICE", "gateways": []},
+    {"identifier": "", "description": "Unassigned Interface"},
+]
+
+ALIASES = [
+    {"name": "proxy_01", "type": "host", "content": "172.22.50.2"},
+    {"name": "web_pair", "type": "host", "content": "10.0.0.5\n10.0.0.6"},
+    {"name": "by_name", "type": "host", "content": "web.example.com"},
+    {"name": "postgres", "type": "port", "content": "5432"},
+]
+
+
+def _rule(**over) -> dict:
+    """One row as the API returns it; booleans are "0"/"1" strings."""
+    row = {
+        "uuid": "u",
+        "disabled": "0",
+        "nordr": "0",
+        "interface": "wan",
+        "ipprotocol": "inet",
+        "protocol": "tcp",
+        "source.network": "any",
+        "source.not": "0",
+        "destination.network": "wanip",
+        "destination.not": "0",
+        "destination.port": "443",
+        "target": "proxy_01",
+        "local-port": "",
+        "descr": "Reverse proxy HTTPS",
+    }
+    row.update(over)
+    return row
+
+
+def _read(*rows: dict) -> list[dict]:
+    return port_forwards(list(rows), wan_interfaces(INTERFACES), alias_index(ALIASES))
+
+
+class TestWhichInterfacesFaceTheInternet:
+    def test_an_interface_with_an_upstream_gateway(self):
+        assert wan_interfaces(INTERFACES) == {"wan", "opt6"}
+
+    def test_the_overview_may_come_wrapped_in_rows(self):
+        assert wan_interfaces({"rows": INTERFACES}) == {"wan", "opt6"}
+
+
+class TestWhatCounts:
+    def test_a_forward_on_the_wan(self):
+        assert _read(_rule()) == [
+            {
+                "name": "Reverse proxy HTTPS",
+                "protocol": "TCP",
+                "external_port": 443,
+                "internal_ip": "172.22.50.2",
+                "internal_port": 443,
+                "enabled": True,
+            }
+        ]
+
+    def test_a_second_wan_counts_too(self):
+        assert len(_read(_rule(interface="opt6"))) == 1
+
+    def test_a_rule_on_several_interfaces_counts_when_one_is_a_wan(self):
+        assert len(_read(_rule(interface="opt9,wan"))) == 1
+
+    def test_a_redirect_between_internal_networks_does_not(self):
+        """gw: HTTPS from HOMEOFFICE to a NAS name, sent to the proxy."""
+        assert _read(_rule(interface="opt9", **{"destination.network": "HOST_NAS"})) == []
+
+    def test_an_exemption_from_redirection_does_not(self):
+        """The anti-lockout rules: ``nordr`` means "do not redirect"."""
+        assert _read(_rule(nordr="1")) == []
+
+    def test_a_generated_rule_does_not(self):
+        assert _read(_rule(is_automatic=True)) == []
+
+    def test_a_disabled_forward_is_listed_as_disabled(self):
+        (entry,) = _read(_rule(disabled="1"))
+        assert entry["enabled"] is False
+
+
+class TestWhereItGoes:
+    def test_a_literal_address(self):
+        (entry,) = _read(_rule(target="10.0.0.9"))
+        assert entry["internal_ip"] == "10.0.0.9"
+
+    def test_an_alias_with_two_hosts_is_two_forwards(self):
+        assert [e["internal_ip"] for e in _read(_rule(target="web_pair"))] == ["10.0.0.5", "10.0.0.6"]
+
+    def test_an_alias_that_names_a_host_by_dns_is_skipped(self):
+        """No address to put the forward on; guessing one would be worse."""
+        assert _read(_rule(target="by_name")) == []
+
+    def test_an_interface_address_is_skipped(self):
+        assert _read(_rule(target="opt5ip")) == []
+
+    def test_an_ipv6_target(self):
+        (entry,) = _read(_rule(ipprotocol="inet6", target="2001:db8::5"))
+        assert entry["internal_ip"] == "2001:db8::5"
+
+
+class TestPorts:
+    @pytest.mark.parametrize(
+        ("value", "expected"),
+        [("443", 443), ("https", 443), ("http", 80), ("postgres", 5432), ("8000-8010", 8000), ("8000:8010", 8000)],
+    )
+    def test_the_external_port(self, value, expected):
+        (entry,) = _read(_rule(**{"destination.port": value}))
+        assert entry["external_port"] == expected
+
+    def test_the_internal_port_defaults_to_the_external_one(self):
+        (entry,) = _read(_rule(**{"destination.port": "80"}))
+        assert entry["internal_port"] == 80
+
+    def test_a_different_internal_port_may_come_as_a_number(self):
+        (entry,) = _read(_rule(**{"destination.port": "80", "local-port": 9000}))
+        assert entry["internal_port"] == 9000
+
+    def test_an_unknown_port_name_skips_the_rule(self):
+        assert _read(_rule(**{"destination.port": "no-such-service"})) == []
+
+    def test_a_whole_host_forwarded_is_kept(self):
+        """The most exposed case of all: every protocol, every port."""
+        (entry,) = _read(_rule(protocol="any", **{"destination.port": ""}))
+        assert (entry["protocol"], entry["external_port"], entry["internal_port"]) == ("ANY", 0, 0)
+
+    def test_every_port_of_one_protocol(self):
+        (entry,) = _read(_rule(**{"destination.port": ""}))
+        assert (entry["protocol"], entry["external_port"]) == ("TCP", 0)
+
+    def test_tcp_and_udp_are_two_forwards(self):
+        assert [e["protocol"] for e in _read(_rule(protocol="tcp/udp"))] == ["TCP", "UDP"]
+
+
+class TestNameAndSource:
+    def test_without_a_description_the_rule_is_named_after_what_it_does(self):
+        (entry,) = _read(_rule(descr=""))
+        assert entry["name"] == "TCP 443 -> proxy_01"
+
+    def test_a_source_restriction_is_the_remote_host(self):
+        (entry,) = _read(_rule(**{"source.network": "203.0.113.7"}))
+        assert entry["remote_host"] == "203.0.113.7"
+
+    def test_any_source_has_no_remote_host(self):
+        (entry,) = _read(_rule())
+        assert "remote_host" not in entry
+
+    def test_an_inverted_source_has_no_remote_host(self):
+        """"Everyone but X" is as good as anyone for whether it is reachable."""
+        (entry,) = _read(_rule(**{"source.network": "203.0.113.7", "source.not": "1"}))
+        assert "remote_host" not in entry
+
+
+class TestTheDriverMethod:
+    @pytest.fixture
+    def driver(self):
+        with patch("napalm_opnsense.opnsense.requests.Session"):
+            drv = OPNsenseDriver(
+                hostname="opnsense.example.com",
+                username="key",
+                password="secret",
+                optional_args={"verify": False},
+            )
+            drv.session = MagicMock()
+            yield drv
+
+    def test_it_reads_rules_interfaces_and_aliases(self, driver):
+        seen = []
+
+        def fake_get(path):
+            seen.append(path.split("?")[0])
+            if path.startswith("/api/firewall/d_nat/search_rule"):
+                return {"rows": [_rule()]}
+            if path.startswith("/api/interfaces/overview/interfaces_info"):
+                return {"rows": INTERFACES}
+            if path.startswith("/api/firewall/alias/searchItem"):
+                return {"rows": ALIASES}
+            raise AssertionError(path)
+
+        driver._get = fake_get
+        assert [e["internal_ip"] for e in driver.get_port_forwards()] == ["172.22.50.2"]
+        assert seen == [
+            "/api/firewall/d_nat/search_rule",
+            "/api/interfaces/overview/interfaces_info",
+            "/api/firewall/alias/searchItem",
+        ]
+
+    def test_netork_can_tell_it_is_there(self):
+        """netOrk asks ``hasattr`` before it calls."""
+        assert hasattr(OPNsenseDriver, "get_port_forwards")