fix!: the SNMP agent takes the caller's values and opens only to its sources (0.2.0)

fix_snmp enabled os-net-snmp with community "public", contact
netork@localhost and location "Managed by netOrk", and added a
floating pass rule for UDP/161 from any source on every interface: a
firewall managed over WAN answered its SNMP tree to anyone there (#13).

The driver implements napalm-device-types 4.13's configure_snmp_agent
steps instead:
- install: os-net-snmp only when missing (a reinstall resets it);
- configure: community, contact, location and listen address from the
  caller; a DNS name is not a listen address; a refused save fails;
- firewall: one floating pass rule per allowed source, the old
  any-source rule and rules of an earlier run replaced, operators'
  rules untouched; no sources, no change;
- start: restart net-snmp.
The hand-built UDP probe goes; the caller asks the agent itself.

BREAKING CHANGE: run_device_action is removed. Requires
napalm-device-types 4.13.

Closes #13

tests/unit/test_snmp_agent.py

@@ -0,0 +1,177 @@
+"""The SNMP agent, set up through napalm-device-types' ``configure_snmp_agent``.
+
+``fix_snmp`` enabled os-net-snmp with community "public" and opened UDP/161 on
+every interface to every source (#13). The template's steps take every value
+from the caller: community, contact, location, listen address, and the sources
+the firewall lets through -- one pass rule each, and the old any-source rule
+goes when sources are given.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from napalm_opnsense.opnsense import OPNsenseDriver
+
+SETTINGS = {
+    "community": "c0mm",
+    "allowed_sources": ["10.7.224.20", "10.7.225.0/24"],
+    "contact": "ops",
+    "location": "rack 1",
+}
+
+
+class _Api:
+    """A firewall's REST API: answers by path, records every POST."""
+
+    def __init__(self, *, installed=True, rules=(), saved="saved"):
+        self.installed = installed
+        self.rules = list(rules)
+        self.saved = saved
+        self.posts: list[tuple[str, dict | None]] = []
+
+    def get(self, path):
+        if path == "/api/netsnmp/general/get":
+            return {"general": {"enabled": "0"}} if self.installed else {}
+        raise AssertionError(path)
+
+    def post(self, path, body=None):
+        self.posts.append((path, body))
+        if path == "/api/netsnmp/general/set":
+            return {"result": self.saved}
+        if path == "/api/firewall/filter/searchRule":
+            return {"rows": self.rules}
+        if path.startswith("/api/firewall/filter/addRule"):
+            return {"result": "saved", "uuid": "new"}
+        if path.startswith("/api/firewall/filter/delRule/"):
+            return {"result": "deleted"}
+        if path in ("/api/firewall/filter/apply", "/api/netsnmp/service/restart"):
+            return {"status": "ok"}
+        if path.startswith("/api/core/firmware/install/"):
+            return {"status": "ok", "msg_uuid": "x"}
+        raise AssertionError(path)
+
+    def paths(self):
+        return [p for p, _ in self.posts]
+
+
+@pytest.fixture
+def api():
+    return _Api()
+
+
+@pytest.fixture
+def driver(api, monkeypatch):
+    d = OPNsenseDriver("10.0.0.1", "key", "secret")
+    d._session = MagicMock()
+    d._get = api.get
+    d._post = api.post
+    monkeypatch.setattr("time.sleep", lambda s: None)
+    return d
+
+
+def _body(api, path):
+    return next(b for p, b in api.posts if p == path)
+
+
+class TestTheCallersValues:
+    def test_community_contact_location_and_listen(self, driver, api):
+        result = driver.configure_snmp_agent({**SETTINGS, "listen": "10.0.0.1"})
+
+        assert result["success"] is True
+        general = _body(api, "/api/netsnmp/general/set")["general"]
+        assert general["enabled"] == "1"
+        assert (general["community"], general["contact"], general["location"]) == ("c0mm", "ops", "rack 1")
+        assert general["listen"] == {"10.0.0.1": {"selected": 1}}
+
+    def test_nothing_of_an_application_is_built_in(self, driver, api):
+        driver.configure_snmp_agent({"community": "c0mm", "allowed_sources": []})
+
+        sent = repr(api.posts)
+        assert "public" not in sent and "netork" not in sent.lower()
+
+    def test_a_dns_name_is_not_a_listen_address(self, api, monkeypatch):
+        d = OPNsenseDriver("gw-01.lan", "key", "secret")
+        d._session, d._get, d._post = MagicMock(), api.get, api.post
+        monkeypatch.setattr("time.sleep", lambda s: None)
+
+        d.configure_snmp_agent({"community": "c0mm", "allowed_sources": []})
+
+        assert _body(api, "/api/netsnmp/general/set")["general"]["listen"] == {}
+
+    def test_a_rejected_configuration_fails_the_fix(self, driver, api):
+        api.saved = "failed"
+
+        result = driver.configure_snmp_agent(SETTINGS)
+
+        assert result["success"] is False
+        assert "/api/netsnmp/service/restart" not in api.paths()
+
+
+class TestTheFirewall:
+    def test_one_pass_rule_per_allowed_source(self, driver, api):
+        driver.configure_snmp_agent(SETTINGS)
+
+        rules = [b["rule"] for p, b in api.posts if p == "/api/firewall/filter/addRule"]
+        assert [r["source_net"] for r in rules] == ["10.7.224.20/32", "10.7.225.0/24"]
+        assert all(r["destination_port"] == "161" and r["protocol"] == "udp" for r in rules)
+        assert all("netork" not in r["descr"].lower() for r in rules)
+        assert "/api/firewall/filter/apply" in api.paths()
+
+    def test_the_old_any_source_rule_goes(self, api, driver):
+        api.rules = [{"uuid": "old-1", "description": "[netork] Allow SNMP", "source_net": "any"}]
+
+        driver.configure_snmp_agent(SETTINGS)
+
+        assert "/api/firewall/filter/delRule/old-1" in api.paths()
+
+    def test_rules_of_an_earlier_run_are_replaced_not_doubled(self, api, driver):
+        api.rules = [{"uuid": "r1", "description": "napalm-opnsense: SNMP from 10.7.224.20/32"}]
+
+        driver.configure_snmp_agent(SETTINGS)
+
+        assert "/api/firewall/filter/delRule/r1" in api.paths()
+        assert sum(p == "/api/firewall/filter/addRule" for p in api.paths()) == 2
+
+    def test_an_operators_rule_is_left_alone(self, api, driver):
+        api.rules = [{"uuid": "mine", "description": "SNMP from the monitoring VLAN"}]
+
+        driver.configure_snmp_agent(SETTINGS)
+
+        assert not any(p.startswith("/api/firewall/filter/delRule/") for p in api.paths())
+
+    def test_no_sources_leave_the_firewall_alone(self, driver, api):
+        result = driver.configure_snmp_agent({"community": "c0mm", "allowed_sources": []})
+
+        assert result["success"] is True
+        assert not any(p.startswith("/api/firewall/") for p in api.paths())
+
+
+class TestInstallAndStart:
+    def test_an_installed_plugin_is_not_reinstalled(self, driver, api):
+        driver.configure_snmp_agent(SETTINGS)
+
+        assert not any(p.startswith("/api/core/firmware/install/") for p in api.paths())
+        assert api.paths()[-1] == "/api/netsnmp/service/restart"
+
+    def test_a_missing_plugin_is_installed_first(self, api, driver):
+        api.installed = False
+
+        driver.configure_snmp_agent(SETTINGS)
+
+        assert api.paths()[0] == "/api/core/firmware/install/os-net-snmp"
+
+
+def test_the_old_action_is_gone():
+    assert not hasattr(OPNsenseDriver, "run_device_action")
+    assert not hasattr(OPNsenseDriver, "_action_fix_snmp")
+
+
+@pytest.mark.parametrize("community", ["two words", 'q"uote', ""])
+def test_a_community_that_is_not_one_word_is_refused(driver, api, community):
+    result = driver.configure_snmp_agent({"community": community, "allowed_sources": []})
+
+    assert result["success"] is False
+    assert "/api/netsnmp/general/set" not in api.paths()