From 5d193ba7e872fd44f94da17495f7398239830d92 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sat, 22 Aug 2026 15:44:56 +0700 Subject: [PATCH] fix(ping): post the settings where the API expects them, not one node above MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every ping against a live OPNsense failed: ping job creation failed for 10.30.0.1: {'result': 'failed', 'validations': {'ping.settings.hostname': 'A value is required.'}} _ping_model_node read GET /api/diagnostics/ping/get and took the single dict-valued key as the node to post under. The real model nests two levels: {"ping": {"settings": {"hostname": "", "fam": {"ip": {...}, "ip6": {...}}, "source_address": "", "packetsize": "", ...}}} so the helper answered "ping" and the job was created with the fields sitting where the settings node belongs. The hostname never arrived, and the firewall said so on every single call. _ping_model_path walks the whole chain of single-dict wrappers and stops at the first level holding more than one key — the field level, where fam is a dict too and one more step would land inside a form field. _wrap_in_model nests the settings accordingly, so a one-level model keeps working and the default, for when /get cannot be read, is what current firmware ships. The tests missed this because FakePingAPI answered /get with a one-level model and read the posted payload back through the same assumption: the fake agreed with the code about a shape neither of them shares with a device. It now speaks what an OPNsense speaks, reads the payload through the model path, and a second test keeps the one-level case covered. Verified against a live firewall: the job is accepted ({"result": "ok"}) and 10.30.0.1 answers 3 of 3 at 0.116 ms. Closes #3 --- napalm_opnsense/ping_mixin.py | 65 ++++++++++++++------ tests/unit/test_ping.py | 109 ++++++++++++++++++++++++++-------- 2 files changed, 130 insertions(+), 44 deletions(-) diff --git a/napalm_opnsense/ping_mixin.py b/napalm_opnsense/ping_mixin.py index 1c411fd..72bf555 100644 --- a/napalm_opnsense/ping_mixin.py +++ b/napalm_opnsense/ping_mixin.py @@ -52,6 +52,10 @@ 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 @@ -74,7 +78,7 @@ class OPNsensePingMixin: PING_SWEEP_MAX_TARGETS: int = 512 # Provided by the driver this mixin is mixed into. - _ping_model_node_cache: Optional[str] = None + _ping_model_path_cache: Optional[tuple[str, ...]] = None # ── NAPALM ─────────────────────────────────────────────────────────────── @@ -202,7 +206,7 @@ class OPNsensePingMixin: if source: settings["source_address"] = source - response = self._post(f"{_PING_API}/set", {self._ping_model_node(): settings}) + 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}") @@ -262,27 +266,52 @@ class OPNsensePingMixin: result[str(job_id)] = row return result - def _ping_model_node(self) -> str: - """Name of the model's root node, i.e. the key ``set`` expects. + 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 - Read once from ``GET /api/diagnostics/ping/get`` instead of hardcoded, - so a model rename in a future OPNsense release does not silently break - job creation. Falls back to ``"settings"``. + 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_node_cache is not None: - return self._ping_model_node_cache + if self._ping_model_path_cache is not None: + return self._ping_model_path_cache - node = "settings" + path = _DEFAULT_PING_MODEL_PATH try: - response = self._get(f"{_PING_API}/get") - candidates = [key for key, value in response.items() if isinstance(value, dict)] - if len(candidates) == 1: - node = candidates[0] - except Exception as exc: # noqa: BLE001 - the default is a safe guess - logger.debug("Could not read ping model node, assuming %r: %s", node, exc) + 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_node_cache = node - return node + self._ping_model_path_cache = path + return path # ── Parsing ────────────────────────────────────────────────────────────── diff --git a/tests/unit/test_ping.py b/tests/unit/test_ping.py index 5e6e6e1..666c581 100644 --- a/tests/unit/test_ping.py +++ b/tests/unit/test_ping.py @@ -20,10 +20,14 @@ from napalm_opnsense.opnsense import OPNsenseDriver class FakePingAPI: """Minimal stand-in for the OPNsense diagnostics ping controller.""" - def __init__(self, stats=None, model_node="settings"): + def __init__(self, stats=None, model_path=("ping", "settings")): #: hostname -> row returned by search_jobs self.stats = stats or {} - self.model_node = model_node + #: 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 @@ -34,7 +38,14 @@ class FakePingAPI: def get(self, path): if path == "/api/diagnostics/ping/get": - return {self.model_node: {"hostname": "", "fam": "ip"}} + 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()]} @@ -45,8 +56,7 @@ class FakePingAPI: self.created.append(data) self._next_id += 1 job_id = f"job-{self._next_id}" - settings = (data or {}).get(self.model_node, {}) - self.jobs[job_id] = settings.get("hostname", "") + 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}/"): @@ -59,12 +69,38 @@ class FakePingAPI: 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, @@ -144,18 +180,14 @@ def test_ping_creates_job_with_destination_and_packetsize(driver, api): driver.ping("10.0.0.1", size=64, source="10.0.0.254") - assert api.created == [ - { - "settings": { - "hostname": "10.0.0.1", - "fam": "ip", - "packetsize": "64", - "interval": "1", - "source_address": "10.0.0.254", - "description": "netork ping sweep", - } - } - ] + 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): @@ -163,7 +195,7 @@ def test_ping_uses_ipv6_family_for_v6_destination(driver, api): driver.ping("2001:db8::1") - assert api.created[0]["settings"]["fam"] == "ip6" + assert api.fields(api.created[0])["fam"] == "ip6" def test_ping_starts_stops_and_removes_the_job(driver, api): @@ -200,19 +232,44 @@ def test_ping_returns_error_when_job_creation_fails(driver, api): assert "hostname" in result["error"] -def test_ping_honours_the_model_root_node_reported_by_the_api(): - fake = FakePingAPI(model_node="ping") +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() - 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 + 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 "ping" in fake.created[0] + assert list(fake.created[0]) == ["settings"] + assert fake.created[0]["settings"]["hostname"] == "10.0.0.1" # ---------------------------------------------------------------------------