diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index 43fbc9e..702b7ae 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -36,6 +36,7 @@ per-module through their respective API controllers. from __future__ import annotations import difflib +import ipaddress import json import logging import re @@ -53,8 +54,10 @@ _REFRESH_INTERVAL = 3 import requests from requests.exceptions import RequestException -from napalm_device_types import ConfigSyncMixin, FingerprintRule, FirewallDriver +from napalm_device_types import ConfigSyncMixin, FingerprintRule, FirewallDriver, SnmpAgentMixin from napalm_device_types.models import ( + FixStepOutcomeDict, + SnmpAgentSettingsDict, ConfigSyncDict, DhcpHaDict, ListeningSocketsDict, @@ -86,7 +89,7 @@ def _carp_master(entry: dict[str, Any]) -> bool: return str(entry.get("status") or "").strip().upper() == "MASTER" -class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, FirewallDriver): +class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, SnmpAgentMixin, FirewallDriver): """NAPALM driver for OPNsense (read-only, REST API).""" VENDOR = "OPNsense" @@ -2391,157 +2394,87 @@ class OPNsenseDriver(OPNsensePingMixin, ConfigSyncMixin, FirewallDriver): } return {"success": True, "output": f"Magic packet sent to {mac_address} via {interface}"} - def run_device_action(self, action: str) -> dict[str, Any]: - """Execute a named action on the firewall.""" - if action == "fix_snmp": - return self._action_fix_snmp() - raise NotImplementedError(f"Unknown action: {action!r}") + # ------------------------------------------------------------------ + # SNMP agent (napalm-device-types SnmpAgentMixin): the caller's values + # ------------------------------------------------------------------ - def _action_fix_snmp(self) -> dict[str, Any]: - """Install os-net-snmp plugin, configure community 'public', open firewall. + #: What marks a pass rule this driver added for an allowed SNMP source; the + #: old any-source rule is recognised by its older description. + _SNMP_RULE_PREFIX = "napalm-opnsense: SNMP from " + _SNMP_RULE_OLD = "[netork] Allow SNMP" - Steps: - 1. Install os-net-snmp via firmware API (idempotent) - 2. Configure via POST /api/netsnmp/general/set - 3. Start/restart the service - 4. Add a floating firewall rule allowing UDP/161 from any source - (OPNsense default-drops traffic arriving on non-LAN interfaces - such as WireGuard; without this rule SNMP is unreachable from - management networks even though the daemon is running) - 5. Apply firewall and verify - """ - lines: list = [] - - # 1. Install os-net-snmp plugin ONLY if not already present. - # Calling firmware/install on an already-installed plugin triggers an - # async reinstall that overwrites the config with factory defaults a few - # minutes later — causing SNMP to stop working again after the fix. - plugin_installed = False + def _install_snmp_agent(self) -> FixStepOutcomeDict: + # Installing over an installed os-net-snmp starts an async reinstall that + # resets its configuration minutes later, so only when it is missing. try: chk = self._get("/api/netsnmp/general/get") - plugin_installed = isinstance(chk, dict) and "general" in chk - except Exception: - pass + except Exception: # noqa: BLE001 - the plugin's API is not there + chk = {} + if isinstance(chk, dict) and "general" in chk: + return {"ok": True, "output": "os-net-snmp already installed"} + result = self._post("/api/core/firmware/install/os-net-snmp") + time.sleep(5) # the install settles before its API answers + return {"ok": result.get("status") == "ok", "output": f"os-net-snmp install: {result}"} - if not plugin_installed: - try: - result = self._post("/api/core/firmware/install/os-net-snmp") - lines.append(f"[install] {result}") - import time as _time - _time.sleep(5) # wait for install to settle - except Exception as exc: - logger.debug("os-net-snmp install failed: %s", exc) - lines.append(f"[install] failed: {exc}") - else: - lines.append("[install] os-net-snmp already installed — skipping reinstall") - - # 2. Configure SNMP - # The listen field must contain the management IP so that snmpd binds - # to the correct interface. Without it snmpd may not respond on any - # non-loopback address. - # The API expects {"": {"selected": 1}} — we use self.hostname - # which is always the device's management IP in netOrk. - listen_val: dict = {self.hostname: {"selected": 1}} + def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict: + community = settings["community"] + if not community or any(c.isspace() or c in "\"#" for c in community): + raise ValueError("the community must be one word without quotes or #") + listen = settings.get("listen") or self.hostname try: - self._post("/api/netsnmp/general/set", { - "general": { + listen_val = {str(ipaddress.ip_address(listen)): {"selected": 1}} + except ValueError: + listen_val = {} # a DNS name binds nothing; snmpd then listens everywhere + result = self._post("/api/netsnmp/general/set", { + "general": { + "enabled": "1", + "community": community, + "contact": settings.get("contact", ""), + "location": settings.get("location", ""), + "sysobjid": "", + "bindip": "", + "listen": listen_val, + } + }) + if result.get("result") != "saved": + return {"ok": False, "output": f"configuration refused: {result}"} + return {"ok": True, "output": f"community set, listen {list(listen_val) or 'all addresses'}"} + + def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict: + sources = [str(ipaddress.ip_network(s.strip(), strict=False)) for s in settings.get("allowed_sources", [])] + if not sources: + return {"ok": True, "output": "no sources given; firewall unchanged"} + found = self._post("/api/firewall/filter/searchRule", {"current": 1, "rowCount": -1, "searchPhrase": "SNMP"}) + for row in found.get("rows") or []: + descr = str(row.get("description") or row.get("descr") or "") + if descr == self._SNMP_RULE_OLD or descr.startswith(self._SNMP_RULE_PREFIX): + self._post(f"/api/firewall/filter/delRule/{row['uuid']}") + for source in sources: + self._post("/api/firewall/filter/addRule", { + "rule": { "enabled": "1", - "community": "public", - "contact": "netork@localhost", - "location": "Managed by netOrk", - "sysobjid": "", - "bindip": "", - "listen": listen_val, + "action": "pass", + "quick": "1", + "interface": "", # floating: wherever the source comes in + "direction": "in", + "ipprotocol": "inet6" if ":" in source else "inet", + "protocol": "udp", + "source_net": source, + "source_not": "0", + "destination_net": "(self)", + "destination_not": "0", + "destination_port": "161", + "log": "0", + "floating": "yes", + "descr": f"{self._SNMP_RULE_PREFIX}{source}", } }) - lines.append(f"[config] SNMP enabled with community 'public', listen={self.hostname}.") - except Exception as exc: - logger.debug("SNMP config failed: %s", exc) - lines.append(f"[config] error: {exc}") + self._post("/api/firewall/filter/apply", {}) + return {"ok": True, "output": f"udp/161 open for {', '.join(sources)} only"} - # 3. Start / restart the SNMP service - try: - self._post("/api/netsnmp/service/restart") - lines.append("[service] net-snmp restarted.") - except Exception as exc: - logger.debug("SNMP service restart failed, trying start: %s", exc) - try: - self._post("/api/netsnmp/service/start") - lines.append("[service] net-snmp started.") - except Exception as exc2: - lines.append(f"[service] start failed: {exc2}") - - # 4. Add floating firewall rule for SNMP (UDP/161 → self) - # OPNsense blocks traffic arriving on non-LAN interfaces by default. - # A floating pass rule makes SNMP reachable from all management nets. - try: - existing = self._post("/api/firewall/filter/searchRule", - {"current": 1, "rowCount": -1, "searchPhrase": "[netork] Allow SNMP"}) - if not (existing.get("rows") or []): - self._post("/api/firewall/filter/addRule", { - "rule": { - "enabled": "1", - "sequence": "1", - "action": "pass", - "quick": "1", - "interface": "", # floating — all interfaces - "direction": "in", - "ipprotocol": "inet", - "protocol": "udp", - "source_net": "any", - "source_not": "0", - "destination_net": "(self)", - "destination_not": "0", - "destination_port": "161", - "log": "0", - "floating": "yes", - "descr": "[netork] Allow SNMP", - } - }) - lines.append("[firewall] Floating pass rule added for UDP/161.") - else: - lines.append("[firewall] Pass rule already present.") - self._post("/api/firewall/filter/apply", {}) - lines.append("[firewall] Rules applied.") - except Exception as exc: - logger.debug("Firewall rule for SNMP failed: %s", exc) - lines.append(f"[firewall] error: {exc}") - - # 5. Verify — try actual UDP/161 probe first, fall back to API config check - success = False - try: - import socket as _socket - sock = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM) - sock.settimeout(3) - community = b"public" - # Minimal SNMPv2c GetRequest for sysDescr - pdu = (b"\x30\x26\x02\x01\x01\x04" - + bytes([len(community)]) + community - + b"\xa0\x19\x02\x04\x00\x00\x00\x01" - + b"\x02\x01\x00\x02\x01\x00" - + b"\x30\x0b\x30\x09\x06\x05\x2b\x06\x01\x02\x01\x05\x00") - sock.sendto(pdu, (self.hostname, 161)) - try: - data, _ = sock.recvfrom(1024) - success = len(data) > 0 - lines.append("[ok] SNMP UDP probe successful.") - except _socket.timeout: - lines.append("[warn] SNMP UDP probe timed out — service may be starting.") - finally: - sock.close() - except Exception as exc: - logger.debug("SNMP UDP probe failed: %s", exc) - # Fall back to API config check - try: - cfg = self._get("/api/netsnmp/general/get") - general = cfg.get("general", cfg) - success = str(general.get("enabled", "0")) == "1" and bool(general.get("community")) - lines.append("[ok] SNMP config active (UDP probe unavailable)." if success - else "[warn] SNMP config check failed.") - except Exception: - lines.append("[warn] Could not verify SNMP state.") - - return {"success": success, "output": "\n".join(lines)} + def _start_snmp_agent(self) -> FixStepOutcomeDict: + result = self._post("/api/netsnmp/service/restart") + return {"ok": result.get("status") == "ok", "output": f"net-snmp restart: {result}"} def get_port_forwards(self) -> list[dict[str, Any]]: """Destination NAT on the WAN interfaces, as port forwards. diff --git a/pyproject.toml b/pyproject.toml index d24fdaa..1c8aebb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-opnsense" -version = "0.1.0" +version = "0.2.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.11.0", + "napalm_device_types>=4.13.0", "requests>=2.28.0", ] diff --git a/tests/unit/test_snmp_agent.py b/tests/unit/test_snmp_agent.py new file mode 100644 index 0000000..3d53c82 --- /dev/null +++ b/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()