From d926218eff1482eb32d28cb87761e5477970acca Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Mon, 1 Jun 2026 13:09:57 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20SNMP=20support=20=E2=80=94=20get=5Fsnmp?= =?UTF-8?q?=5Fconfig(),=20fix=5Fsnmp=20action?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_snmp_config() calls GET /api/netsnmp/general/get with 5s timeout (plugin may not be installed). fix_snmp installs os-net-snmp package, configures via POST /api/netsnmp/general/set with community 'public', restarts service. Co-Authored-By: Claude Sonnet 4.6 --- napalm_opnsense/opnsense.py | 135 +++++++++++++++++++++++++++++++++++- 1 file changed, 134 insertions(+), 1 deletion(-) diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index 52a1888..efd9a75 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -1305,7 +1305,11 @@ class OPNsenseDriver(FirewallDriver): status = self._get("/api/core/firmware/status") state = status.get("status", "none") if state in ("update", "upgrade"): - updates = status.get("updates") or [] + updates = ( + status.get("upgrade_packages") + or status.get("updates") + or [] + ) return [ { "name": u.get("name", ""), @@ -1320,6 +1324,36 @@ class OPNsenseDriver(FirewallDriver): pass return [] + def get_device_warnings(self) -> List[Dict[str, Any]]: + """Return a list of warning dicts for issues detected on this device. + + Reads the cached ``GET /api/core/firmware/status`` (no network + update trigger) to detect available package/firmware updates. + """ + warnings: List[Dict[str, Any]] = [] + try: + status = self._get("/api/core/firmware/status") + state = status.get("status", "none") + if state in ("update", "upgrade"): + upgrades = ( + status.get("upgrade_packages") + or status.get("updates") + or [] + ) + if upgrades: + warnings.append({ + "code": "updates_available", + "severity": "info", + "action": None, + "meta": { + "count": len(upgrades), + "packages": [u.get("name", "") for u in upgrades[:10]], + }, + }) + except Exception: + pass + return warnings + def apply_updates(self, packages: List[str]) -> Dict[str, Any]: """Trigger a full firmware upgrade on OPNsense. @@ -1388,3 +1422,102 @@ class OPNsenseDriver(FirewallDriver): return {"success": True, "output": str(result)} except Exception as exc: return {"success": False, "output": str(exc)} + + # ── SNMP / Health ────────────────────────────────────────────────────────── + + def get_snmp_config(self): + """Return SNMP config if the os-net-snmp plugin is installed and enabled. + + Uses GET /api/netsnmp/general/get (os-net-snmp plugin API). + Returns None if the plugin is not installed or SNMP is disabled. + """ + try: + from napalm_device_types.models import SNMPConfigDict + except ImportError: + return None + + try: + # Short timeout — endpoint may not exist if os-net-snmp plugin is not installed + url = self.base_url.rstrip("/") + "/api/netsnmp/general/get" + response = self.session.get(url, timeout=5) + if response.status_code != 200: + return None + data = response.json() + except Exception: + return None + + if not data: + return None + + general = data.get("general", data) + enabled = str(general.get("enabled", "0")) == "1" + if not enabled: + return None + + community = general.get("community", "public") or "public" + return SNMPConfigDict(running=True, community=community, port=161, version="2c") + + 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}") + + def _action_fix_snmp(self) -> Dict[str, Any]: + """Install os-net-snmp plugin, configure community 'public', start service. + + Steps: + 1. Install os-net-snmp via firmware API (idempotent — no-op if installed) + 2. Configure via POST /api/netsnmp/general/set + 3. Start/restart via POST /api/netsnmp/service/start + """ + lines: list = [] + + # 1. Install os-net-snmp plugin (POST /api/core/firmware/install/os-net-snmp) + try: + result = self._post("/api/core/firmware/install/os-net-snmp") + lines.append(f"[install] {result}") + except Exception as exc: + lines.append(f"[install] skipped or already installed: {exc}") + + # 2. Configure SNMP: enable + set community 'public' + try: + self._post("/api/netsnmp/general/set", { + "general": { + "enabled": "1", + "community": "public", + "contact": "netork@localhost", + "location": "Managed by netOrk", + "sysobjid": "", + "bindip": "", + } + }) + lines.append("[config] SNMP enabled with community 'public'.") + except Exception as exc: + lines.append(f"[config] error: {exc}") + + # 3. Start / restart the SNMP service + try: + self._post("/api/netsnmp/service/restart") + lines.append("[service] net-snmp restarted.") + except Exception: + try: + self._post("/api/netsnmp/service/start") + lines.append("[service] net-snmp started.") + except Exception as exc: + lines.append(f"[service] start failed: {exc}") + + # 4. Verify + 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")) + except Exception: + success = False + + if success: + lines.append("[ok] SNMP is active with community 'public'.") + else: + lines.append("[warn] Could not verify SNMP state via API.") + + return {"success": success, "output": "\n".join(lines)}