Merge pull request 'feat(firewall): report the gateway a filter rule policy-routes to' (#7) from feat/firewall-rule-gateway into master

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/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")