feat: restart the firewall (reboot_host)

netOrk asks a driver for reboot_host before it offers a restart, and the
OPNsense driver had none, so a firewall could not be restarted from netOrk
(netOrk #637). reboot_host asks the firmware API (POST
/api/core/firmware/reboot); a refusal raises.

.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/*

napalm_opnsense/opnsense.py

@@ -39,11 +39,16 @@ import difflib
 import json
 import logging
 import socket
+import time
 from ipaddress import ip_address, ip_network
 from typing import Any
 
 logger = logging.getLogger(__name__)
 
+#: How long refresh_available_updates waits for the firewall's update check.
+_REFRESH_POLLS = 20
+_REFRESH_INTERVAL = 3
+
 import requests
 from requests.exceptions import RequestException
 
@@ -51,6 +56,7 @@ 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(OPNsensePingMixin, FirewallDriver):
@@ -1997,45 +2003,72 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
         return tunnels
 
     def get_available_updates(self) -> list[dict[str, Any]]:
-        """Return available firmware and package updates.
+        """Return the pending firmware and package updates the firewall last found.
 
-        Triggers an async update-check on OPNsense via
-        ``POST /api/core/firmware/check``, then polls
-        ``GET /api/core/firmware/status`` for up to 15 seconds.
-        Returns a list of ``{name, current_version, new_version}`` dicts,
-        or an empty list when everything is up to date or the check has
-        not yet finished.
+        Reads the cached ``GET /api/core/firmware/status``; it triggers no check
+        (that is :meth:`refresh_available_updates`). An empty list means the last
+        check found nothing.
+
+        :raises RuntimeError: when the firewall never checked or cannot reach its
+            mirror -- never an empty list for "don't know".
         """
-        import time
-        try:
-            self._post("/api/core/firmware/check")
-        except Exception as exc:
-            logger.debug("Firmware update check trigger failed: %s", exc)
-
-        for _ in range(5):
-            time.sleep(3)
-            try:
-                status = self._get("/api/core/firmware/status")
-                state = status.get("status", "none")
-                if state in ("update", "upgrade"):
-                    updates = (
-                        status.get("upgrade_packages")
-                        or status.get("updates")
-                        or []
-                    )
-                    return [
+        status = self._checked_firmware_status()
+        if status.get("status") not in ("update", "upgrade"):
+            return []
+        return sorted(
+            (
                 {
                     "name": u.get("name", ""),
                     "current_version": u.get("current_version", u.get("version", "")),
                     "new_version": u.get("new_version", u.get("version", "")),
                 }
-                        for u in updates
-                    ]
-                if state == "latest":
-                    return []
+                for u in status.get("upgrade_packages") or status.get("updates") or []
+            ),
+            key=lambda u: u["name"],
+        )
+
+    def _checked_firmware_status(self) -> dict[str, Any]:
+        status = self._get("/api/core/firmware/status")
+        if not status.get("last_check"):
+            raise RuntimeError("The firewall has not checked for updates yet")
+        for field in ("connection", "repository"):
+            if status.get(field, "ok") != "ok":
+                raise RuntimeError(f"The firmware {field} is {status.get(field)!r}")
+        return status
+
+    def refresh_available_updates(self) -> dict[str, Any]:
+        """Run the firewall's update check (``firmware/check``) and wait for it."""
+        before = self._get("/api/core/firmware/status").get("last_check")
+        self._post("/api/core/firmware/check")
+        for _ in range(_REFRESH_POLLS):
+            time.sleep(_REFRESH_INTERVAL)
+            status = self._get("/api/core/firmware/status")
+            if status.get("last_check") and status.get("last_check") != before:
+                return {"success": True, "output": status.get("status_msg", "")}
+        return {
+            "success": False,
+            "output": f"The update check did not finish within {_REFRESH_POLLS * _REFRESH_INTERVAL} s",
+        }
+
+    def reboot_host(self) -> None:
+        """Restart the firewall through its firmware API (``HostRebootMixin``).
+
+        :raises RuntimeError: when the API refuses the request.
+        """
+        try:
+            self._post("/api/core/firmware/reboot")
         except Exception as exc:
-                logger.debug("Firmware status poll failed: %s", exc)
-        return []
+            raise RuntimeError(f"The firewall refused to reboot: {exc}") from exc
+
+    def get_host_status(self) -> dict[str, Any]:
+        """Whether the firewall needs a reboot to finish an update; it does not
+        patch itself as far as netOrk can tell."""
+        pending = self._get("/api/core/firmware/status").get("needs_reboot") == "1"
+        return {
+            "reboot_required": pending,
+            "reboot_reason": "the firmware status reports a pending reboot" if pending else None,
+            "auto_updates": None,
+        }
 
     def get_device_warnings(self) -> list[dict[str, Any]]:
         """Return a list of warning dicts for issues detected on this device.
@@ -2387,6 +2420,21 @@ class OPNsenseDriver(OPNsensePingMixin, 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.
 
@@ -2428,6 +2476,10 @@ class OPNsenseDriver(OPNsensePingMixin, 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")
@@ -2492,6 +2544,7 @@ class OPNsenseDriver(OPNsensePingMixin, 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"]))

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

tests/unit/test_updates.py

@@ -0,0 +1,140 @@
+"""Pending updates on OPNsense, read from the firmware status the firewall caches.
+
+``get_available_updates`` used to trigger ``firmware/check`` and sleep up to
+15 s; a poll could not afford that, and when the check was not done in time it
+reported "no updates". Reading now comes from the cached ``firmware/status``,
+which carries when the firewall last checked. A firewall that never checked, or
+cannot reach its mirror, raises: netOrk keeps "pending since" per package, and
+an empty list would mean nothing is pending. ``refresh_available_updates`` runs
+the check and waits for it (netOrk MVP 5).
+
+The status fields are the real ones of an OPNsense 26.7.5 firewall.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from napalm_opnsense.opnsense import OPNsenseDriver
+
+
+def _status(**over) -> dict:
+    status = {
+        "connection": "ok",
+        "repository": "ok",
+        "last_check": "Mon Oct  5 07:37:38 CEST 2026",
+        "needs_reboot": "0",
+        "upgrade_needs_reboot": "0",
+        "status": "none",
+        "status_msg": "There are no updates available on the selected mirror.",
+        "upgrade_packages": [],
+    }
+    status.update(over)
+    return status
+
+
+PENDING = [
+    {"name": "opnsense", "current_version": "26.7.5", "new_version": "26.7.6", "reason": "upgrade"},
+    {"name": "openssl", "current_version": "3.0.16", "new_version": "3.0.17", "reason": "upgrade"},
+]
+
+
+@pytest.fixture
+def driver():
+    with patch("napalm_opnsense.opnsense.requests.Session"):
+        drv = OPNsenseDriver(
+            hostname="fw", username="k", password="s", optional_args={"verify": False}
+        )
+        drv.session = MagicMock()
+        yield drv
+
+
+class TestAvailableUpdates:
+    def test_nothing_pending_after_a_check_is_an_empty_list(self, driver):
+        with (
+            patch.object(driver, "_get", return_value=_status()) as get,
+            patch.object(driver, "_post") as post,
+        ):
+            assert driver.get_available_updates() == []
+
+        get.assert_called_once_with("/api/core/firmware/status")
+        post.assert_not_called()
+
+    def test_pending_packages_from_the_cached_status(self, driver):
+        status = _status(status="update", upgrade_packages=PENDING)
+        with patch.object(driver, "_get", return_value=status):
+            updates = driver.get_available_updates()
+
+        assert updates == [
+            {"name": "openssl", "current_version": "3.0.16", "new_version": "3.0.17"},
+            {"name": "opnsense", "current_version": "26.7.5", "new_version": "26.7.6"},
+        ]
+
+    def test_a_firewall_that_never_checked_raises(self, driver):
+        with patch.object(driver, "_get", return_value=_status(last_check="")):
+            with pytest.raises(RuntimeError, match="checked"):
+                driver.get_available_updates()
+
+    @pytest.mark.parametrize("field", ["connection", "repository"])
+    def test_a_mirror_it_cannot_reach_raises(self, driver, field):
+        with patch.object(driver, "_get", return_value=_status(**{field: "error"})):
+            with pytest.raises(RuntimeError):
+                driver.get_available_updates()
+
+
+class TestRefresh:
+    def test_the_check_runs_and_is_waited_for(self, driver):
+        before = _status(last_check="Mon Oct  5 07:37:38 CEST 2026")
+        after = _status(last_check="Tue Oct  6 09:00:01 CEST 2026")
+        with (
+            patch.object(driver, "_get", side_effect=[before, before, after]),
+            patch.object(driver, "_post") as post,
+            patch("napalm_opnsense.opnsense.time.sleep"),
+        ):
+            result = driver.refresh_available_updates()
+
+        post.assert_called_once_with("/api/core/firmware/check")
+        assert result["success"] is True
+
+    def test_a_check_that_does_not_finish_in_time_says_so(self, driver):
+        with (
+            patch.object(driver, "_get", return_value=_status()),
+            patch.object(driver, "_post"),
+            patch("napalm_opnsense.opnsense.time.sleep"),
+        ):
+            result = driver.refresh_available_updates()
+
+        assert result["success"] is False
+
+
+class TestHostStatus:
+    def test_needs_reboot_comes_from_the_firmware_status(self, driver):
+        with patch.object(driver, "_get", return_value=_status(needs_reboot="1")):
+            status = driver.get_host_status()
+
+        assert status == {
+            "reboot_required": True,
+            "reboot_reason": "the firmware status reports a pending reboot",
+            "auto_updates": None,
+        }
+
+    def test_no_pending_reboot(self, driver):
+        with patch.object(driver, "_get", return_value=_status()):
+            assert driver.get_host_status()["reboot_required"] is False
+
+
+class TestRebootHost:
+    """``reboot_host`` restarts the firewall through its firmware API (netOrk #637)."""
+
+    def test_the_firmware_api_is_asked(self, driver):
+        with patch.object(driver, "_post", return_value={"status": "ok"}) as post:
+            driver.reboot_host()
+
+        post.assert_called_once_with("/api/core/firmware/reboot")
+
+    def test_a_refusal_is_raised(self, driver):
+        with patch.object(driver, "_post", side_effect=ConnectionError("403 Forbidden")):
+            with pytest.raises(RuntimeError, match="403"):
+                driver.reboot_host()