diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index e97a254..ab8af25 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -1541,24 +1541,29 @@ class OPNsenseDriver(FirewallDriver): raise NotImplementedError(f"Unknown action: {action!r}") def _action_fix_snmp(self) -> dict[str, Any]: - """Install os-net-snmp plugin, configure community 'public', start service. + """Install os-net-snmp plugin, configure community 'public', open firewall. Steps: - 1. Install os-net-snmp via firmware API (idempotent — no-op if installed) + 1. Install os-net-snmp via firmware API (idempotent) 2. Configure via POST /api/netsnmp/general/set - 3. Start/restart via POST /api/netsnmp/service/start + 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 (POST /api/core/firmware/install/os-net-snmp) + # 1. Install os-net-snmp plugin try: result = self._post("/api/core/firmware/install/os-net-snmp") lines.append(f"[install] {result}") except Exception as exc: logger.debug("os-net-snmp install skipped or failed: %s", exc) - lines.append(f"[install] skipped or already installed: {exc}") + lines.append(f"[install] already installed or skipped") - # 2. Configure SNMP: enable + set community 'public' + # 2. Configure SNMP try: self._post("/api/netsnmp/general/set", { "general": { @@ -1585,10 +1590,45 @@ class OPNsenseDriver(FirewallDriver): self._post("/api/netsnmp/service/start") lines.append("[service] net-snmp started.") except Exception as exc2: - logger.debug("SNMP service start failed: %s", exc2) lines.append(f"[service] start failed: {exc2}") - # 4. Verify + # 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 SNMP config try: cfg = self._get("/api/netsnmp/general/get") general = cfg.get("general", cfg) @@ -1597,11 +1637,7 @@ class OPNsenseDriver(FirewallDriver): logger.debug("SNMP verification failed: %s", exc) success = False - if success: - lines.append("[ok] SNMP is active with community 'public'.") - else: - lines.append("[warn] Could not verify SNMP state via API.") - + lines.append("[ok] SNMP active." if success else "[warn] Could not verify SNMP state.") return {"success": success, "output": "\n".join(lines)} def get_firewall_aliases(self) -> list[dict[str, Any]]: