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
 
@@ -1998,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
+        Reads the cached ``GET /api/core/firmware/status``; it triggers no check
-        ``POST /api/core/firmware/check``, then polls
+        (that is :meth:`refresh_available_updates`). An empty list means the last
-        ``GET /api/core/firmware/status`` for up to 15 seconds.
+        check found nothing.
-        Returns a list of ``{name, current_version, new_version}`` dicts,
+
-        or an empty list when everything is up to date or the check has
+        :raises RuntimeError: when the firewall never checked or cannot reach its
-        not yet finished.
+            mirror -- never an empty list for "don't know".
         """
-        import time
+        status = self._checked_firmware_status()
-        try:
+        if status.get("status") not in ("update", "upgrade"):
-            self._post("/api/core/firmware/check")
+            return []
-        except Exception as exc:
+        return sorted(
-            logger.debug("Firmware update check trigger failed: %s", exc)
+            (
-
-        for _ in range(5):
-            time.sleep(3)
-            try:
-                status = self._get("/api/core/firmware/status")
-                state = status.get("status", "none")
-                if state in ("update", "upgrade"):
-                    updates = (
-                        status.get("upgrade_packages")
-                        or status.get("updates")
-                        or []
-                    )
-                    return [
                 {
                     "name": u.get("name", ""),
                     "current_version": u.get("current_version", u.get("version", "")),
                     "new_version": u.get("new_version", u.get("version", "")),
                 }
-                        for u in updates
+                for u in status.get("upgrade_packages") or status.get("updates") or []
-                    ]
+            ),
-                if state == "latest":
+            key=lambda u: u["name"],
-                    return []
+        )
+
+    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)
+            raise RuntimeError(f"The firewall refused to reboot: {exc}") from exc
-        return []
+
+    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.
@@ -2444,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")
@@ -2508,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"]))

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