Christian Manivong
571d9abb5f
Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM driver uses (E, F, I, UP, B at line length 100, Python 3.10). What ruff cannot fix comes in the next commit.
182 lines
6.5 KiB
Python
182 lines
6.5 KiB
Python
"""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()
|