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