From fc4658f016115d6cb87e91812f218506708a3a7c Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sun, 18 Oct 2026 15:04:03 +0200 Subject: [PATCH] feat: public methods for the upgrade, Unbound forwards and OSPF (0.4.0) netOrk reached these through the private _get/_post (netOrk #1113); they are napalm-device-types 4.17's contracts now: - apply_updates passes the packages it is given (it dropped them) and get_update_progress reports the upgrade's state and log. - get_dns_forwards, set_dns_forward (an existing forward of the domain is changed, not doubled) and remove_dns_forward, each applied with a reconfigure; a refusal is raised. - get_ospf translates FRR's diagnostics into neutral terms. Requires napalm-device-types 4.17.0. --- napalm_opnsense/opnsense.py | 35 ++++-- napalm_opnsense/ospf_mixin.py | 41 +++++++ napalm_opnsense/resolver_mixin.py | 65 +++++++++++ pyproject.toml | 4 +- tests/unit/test_public_contracts.py | 173 ++++++++++++++++++++++++++++ 5 files changed, 307 insertions(+), 11 deletions(-) create mode 100644 napalm_opnsense/ospf_mixin.py create mode 100644 napalm_opnsense/resolver_mixin.py create mode 100644 tests/unit/test_public_contracts.py diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index b722e88..d38b5b6 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -68,6 +68,8 @@ from napalm.base.exceptions import ConnectionException, ConnectionClosedExceptio from napalm_opnsense.ha import config_sync, dhcp_ha from napalm_opnsense.listening import parse_socket_statistics, wireguard_listen_ports from napalm_opnsense.ping_mixin import OPNsensePingMixin +from napalm_opnsense.ospf_mixin import OPNsenseOspfMixin +from napalm_opnsense.resolver_mixin import OPNsenseResolverMixin from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interfaces from napalm_opnsense.virtual_ips import carp_addresses @@ -89,7 +91,14 @@ def _carp_master(entry: dict[str, Any]) -> bool: return str(entry.get("status") or "").strip().upper() == "MASTER" -class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, SnmpAgentMixin, FirewallDriver): +class OPNsenseDriver( + OPNsensePingMixin, + ConfigSyncMixin, + OPNsenseResolverMixin, + OPNsenseOspfMixin, + SnmpAgentMixin, + FirewallDriver, +): """NAPALM driver for OPNsense (read-only, REST API).""" VENDOR = "OPNsense" @@ -2197,22 +2206,30 @@ class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, SnmpAgentMixin, Firewal } def apply_updates(self, packages: list[str]) -> dict[str, Any]: - """Trigger a full firmware upgrade on OPNsense. + """Start the firmware upgrade of *packages*, or of everything for none. - Note: OPNsense upgrades the entire system at once rather than - individual packages. The ``packages`` parameter is accepted for - API compatibility but is ignored — the full upgrade is always - applied. - - Calls ``POST /api/core/firmware/upgrade``. + Returns once the firewall has taken the request; the upgrade runs on + and :meth:`get_update_progress` follows it. Calls + ``POST /api/core/firmware/upgrade``. """ try: - result = self._post("/api/core/firmware/upgrade") + if packages: + result = self._post("/api/core/firmware/upgrade", {"upgrade_packages": list(packages)}) + else: + result = self._post("/api/core/firmware/upgrade") return {"success": True, "output": str(result)} except Exception as exc: logger.warning("Firmware upgrade failed: %s", exc) return {"success": False, "output": str(exc)} + def get_update_progress(self) -> dict[str, Any]: + """Where the firmware upgrade stands, and its whole log + (``GET /api/core/firmware/upgradestatus``).""" + st = self._get("/api/core/firmware/upgradestatus") + status = st.get("status") or "" + state = {"running": "running", "done": "done", "": "idle"}.get(status, "failed") + return {"state": state, "log": st.get("log") or ""} + def search_packages(self, query: str) -> list[dict[str, Any]]: """Search available OPNsense plugins by name or description. diff --git a/napalm_opnsense/ospf_mixin.py b/napalm_opnsense/ospf_mixin.py new file mode 100644 index 0000000..c2f7195 --- /dev/null +++ b/napalm_opnsense/ospf_mixin.py @@ -0,0 +1,41 @@ +"""FRR's OSPF state in neutral terms (napalm-device-types ``OspfMixin``). + +Reads the os-frr plugin's diagnostics (``/api/quagga/diagnostics/...``); the +field names are FRR's. Whether an interface without neighbours or without +authentication is a problem is the caller's to decide. +""" + +from __future__ import annotations + +from typing import Any + +from napalm_device_types import OspfMixin + +_DIAG = "/api/quagga/diagnostics" + + +def _interface(ifc: dict[str, Any]) -> dict[str, Any]: + return { + "address": ifc.get("ipAddress") or "", + "prefix_length": int(ifc.get("ipAddressPrefixlen") or 0), + "area": ifc.get("area") or "", + "network_type": ifc.get("networkType") or "", + "authentication": ifc.get("authentication") or "", + "neighbors": int(ifc.get("nbrCount") or 0), + "neighbors_adjacent": int(ifc.get("nbrAdjacentCount") or 0), + "cost": int(ifc.get("cost") or 0), + } + + +class OPNsenseOspfMixin(OspfMixin): + """Needs the driver's ``_get``.""" + + def get_ospf(self) -> dict[str, Any]: + """Router ID, areas and OSPF per interface; empty when OSPF does not run.""" + overview = self._get(f"{_DIAG}/ospfoverview").get("response") or {} # type: ignore[attr-defined] + interfaces = (self._get(f"{_DIAG}/ospfinterface").get("response") or {}).get("interfaces") or {} # type: ignore[attr-defined] + return { + "router_id": overview.get("routerId") or None, + "areas": list((overview.get("areas") or {}).keys()), + "interfaces": {name: _interface(ifc) for name, ifc in interfaces.items()}, + } diff --git a/napalm_opnsense/resolver_mixin.py b/napalm_opnsense/resolver_mixin.py new file mode 100644 index 0000000..0d044e8 --- /dev/null +++ b/napalm_opnsense/resolver_mixin.py @@ -0,0 +1,65 @@ +"""Unbound's domain forwards (napalm-device-types ``DnsForwardMixin``). + +OPNsense keeps forwards and DNS-over-TLS servers in one model (``dot``); a +forward is a row of type ``forward``. Every change is applied with a +reconfigure before the method returns. +""" + +from __future__ import annotations + +from typing import Any + +from napalm_device_types import DnsForwardMixin + +_SETTINGS = "/api/unbound/settings" +_RECONFIGURE = "/api/unbound/service/reconfigure" + + +def _domain(name: str) -> str: + return name.strip().rstrip(".") + + +class OPNsenseResolverMixin(DnsForwardMixin): + """Needs the driver's ``_post``.""" + + def _forward_rows(self) -> list[dict[str, Any]]: + rows = self._post(f"{_SETTINGS}/searchForward", {}).get("rows") or [] # type: ignore[attr-defined] + return [r for r in rows if r.get("type", "forward") == "forward"] + + def _saved(self, result: dict[str, Any], what: str) -> None: + if result.get("result") not in ("saved", "deleted", None): + raise RuntimeError(f"Unbound refused {what}: {result.get('validations') or result}") + + def get_dns_forwards(self) -> list[dict[str, Any]]: + """Every forward Unbound has, enabled or not; DNS-over-TLS servers are not forwards.""" + return [ + { + "domain": _domain(r.get("domain", "")), + "server": r.get("server", ""), + "port": int(r.get("port") or 53), + "enabled": str(r.get("enabled", "1")) == "1", + } + for r in self._forward_rows() + ] + + def set_dns_forward(self, domain: str, server: str, port: int = 53) -> None: + """Forward *domain* to *server*; an existing forward of the domain is changed.""" + name = _domain(domain) + row = {"enabled": "1", "type": "forward", "domain": name, "server": server, "port": str(port)} + known = [r for r in self._forward_rows() if _domain(r.get("domain", "")) == name] + if known: + result = self._post(f"{_SETTINGS}/setForward/{known[0]['uuid']}", {"dot": row}) # type: ignore[attr-defined] + else: + result = self._post(f"{_SETTINGS}/addForward", {"dot": row}) # type: ignore[attr-defined] + self._saved(result or {}, f"the forward of {name}") + self._post(_RECONFIGURE) # type: ignore[attr-defined] + + def remove_dns_forward(self, domain: str) -> None: + """Delete every forward of *domain*; nothing to do when there is none.""" + name = _domain(domain) + uuids = [r["uuid"] for r in self._forward_rows() if _domain(r.get("domain", "")) == name and r.get("uuid")] + if not uuids: + return + for uuid in uuids: + self._post(f"{_SETTINGS}/delForward/{uuid}") # type: ignore[attr-defined] + self._post(_RECONFIGURE) # type: ignore[attr-defined] diff --git a/pyproject.toml b/pyproject.toml index d602780..3ea938a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-opnsense" -version = "0.3.0" +version = "0.4.0" description = "NAPALM driver for OPNsense (read-only via REST API)." readme = "README.md" license = { text = "Apache-2.0" } @@ -24,7 +24,7 @@ classifiers = [ ] dependencies = [ "napalm>=4.0.0", - "napalm_device_types>=4.13.0", + "napalm_device_types>=4.17.0", "requests>=2.28.0", ] diff --git a/tests/unit/test_public_contracts.py b/tests/unit/test_public_contracts.py new file mode 100644 index 0000000..def5728 --- /dev/null +++ b/tests/unit/test_public_contracts.py @@ -0,0 +1,173 @@ +"""What netOrk reached through ``_get``/``_post``, as public methods (netOrk #1113). + +netOrk followed the firmware upgrade, edited Unbound's forwards and read FRR's +OSPF state through this driver's private REST helpers. The contracts are +napalm-device-types 4.17's; the OPNsense API calls and FRR's field names stay +here. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from napalm_opnsense.opnsense import OPNsenseDriver + + +@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 TestUpdateProgress: + @pytest.mark.parametrize( + "status, state", + [("running", "running"), ("done", "done"), ("error", "failed"), ("", "idle"), (None, "idle")], + ) + def test_the_upgrade_status_as_a_state(self, driver, status, state): + st = {"status": status, "log": "***GOT REQUEST TO UPGRADE***\nfetching\n"} + with patch.object(driver, "_get", return_value=st) as get: + progress = driver.get_update_progress() + + get.assert_called_once_with("/api/core/firmware/upgradestatus") + assert progress == {"state": state, "log": "***GOT REQUEST TO UPGRADE***\nfetching\n"} + + def test_the_packages_go_with_the_upgrade(self, driver): + """netOrk named the packages it upgrades; ``apply_updates`` dropped them.""" + with patch.object(driver, "_post", return_value={"status": "ok"}) as post: + result = driver.apply_updates(["opnsense", "openssl"]) + + post.assert_called_once_with( + "/api/core/firmware/upgrade", {"upgrade_packages": ["opnsense", "openssl"]} + ) + assert result["success"] is True + + def test_no_packages_is_the_whole_upgrade(self, driver): + with patch.object(driver, "_post", return_value={"status": "ok"}) as post: + driver.apply_updates([]) + + post.assert_called_once_with("/api/core/firmware/upgrade") + + +ROWS = [ + {"uuid": "u1", "enabled": "1", "type": "forward", "domain": "lab.example", "server": "10.0.0.53", "port": "53"}, + {"uuid": "u2", "enabled": "0", "type": "forward", "domain": "old.example", "server": "10.0.0.9", "port": ""}, + {"uuid": "u3", "enabled": "1", "type": "dot", "domain": "", "server": "9.9.9.9", "port": "853"}, +] + + +class TestDnsForwards: + def _post(self, calls): + def post(path, data=None): + calls.append((path, data)) + if path.endswith("searchForward"): + return {"rows": ROWS} + return {"result": "saved", "status": "ok"} + + return post + + def test_the_forwards_without_dns_over_tls(self, driver): + calls = [] + with patch.object(driver, "_post", side_effect=self._post(calls)): + forwards = driver.get_dns_forwards() + + assert forwards == [ + {"domain": "lab.example", "server": "10.0.0.53", "port": 53, "enabled": True}, + {"domain": "old.example", "server": "10.0.0.9", "port": 53, "enabled": False}, + ] + + def test_a_new_domain_is_added_and_applied(self, driver): + calls = [] + with patch.object(driver, "_post", side_effect=self._post(calls)): + driver.set_dns_forward("new.example.", "10.0.0.53") + + paths = [p for p, _ in calls] + assert paths[-2:] == ["/api/unbound/settings/addForward", "/api/unbound/service/reconfigure"] + added = calls[-2][1]["dot"] + assert added == {"enabled": "1", "type": "forward", "domain": "new.example", "server": "10.0.0.53", "port": "53"} + + def test_a_known_domain_is_changed_not_doubled(self, driver): + calls = [] + with patch.object(driver, "_post", side_effect=self._post(calls)): + driver.set_dns_forward("lab.example", "10.0.0.54", port=5353) + + paths = [p for p, _ in calls] + assert "/api/unbound/settings/setForward/u1" in paths + assert "/api/unbound/settings/addForward" not in paths + + def test_a_refusal_is_raised(self, driver): + def post(path, data=None): + if path.endswith("searchForward"): + return {"rows": []} + return {"result": "failed", "validations": {"dot.server": "invalid"}} + + with patch.object(driver, "_post", side_effect=post), pytest.raises(RuntimeError, match="invalid"): + driver.set_dns_forward("x.example", "not-an-ip") + + def test_removing_deletes_every_forward_of_the_domain(self, driver): + calls = [] + with patch.object(driver, "_post", side_effect=self._post(calls)): + driver.remove_dns_forward("lab.example.") + + paths = [p for p, _ in calls] + assert paths[-2:] == ["/api/unbound/settings/delForward/u1", "/api/unbound/service/reconfigure"] + + def test_removing_an_unknown_domain_changes_nothing(self, driver): + calls = [] + with patch.object(driver, "_post", side_effect=self._post(calls)): + driver.remove_dns_forward("nothing.example") + + assert [p for p, _ in calls] == ["/api/unbound/settings/searchForward"] + + +OVERVIEW = {"response": {"routerId": "10.255.0.1", "areas": {"0.0.0.0": {}, "0.0.0.1": {}}}} +INTERFACES = { + "response": { + "interfaces": { + "wg0": { + "ipAddress": "10.255.0.1", + "ipAddressPrefixlen": 24, + "area": "0.0.0.0", + "networkType": "POINTOPOINT", + "authentication": "MessageDigest", + "nbrCount": 2, + "nbrAdjacentCount": 1, + "cost": 10, + } + } + } +} + + +class TestOspf: + def test_frr_in_neutral_terms(self, driver): + def get(path): + return OVERVIEW if path.endswith("ospfoverview") else INTERFACES + + with patch.object(driver, "_get", side_effect=get): + ospf = driver.get_ospf() + + assert ospf == { + "router_id": "10.255.0.1", + "areas": ["0.0.0.0", "0.0.0.1"], + "interfaces": { + "wg0": { + "address": "10.255.0.1", + "prefix_length": 24, + "area": "0.0.0.0", + "network_type": "POINTOPOINT", + "authentication": "MessageDigest", + "neighbors": 2, + "neighbors_adjacent": 1, + "cost": 10, + } + }, + } + + def test_no_ospf_running(self, driver): + with patch.object(driver, "_get", return_value={"response": {}}): + assert driver.get_ospf() == {"router_id": None, "areas": [], "interfaces": {}} -- 2.54.0