feat: SNMP support — get_snmp_config(), fix_snmp action

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 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
2026-06-01 13:09:57 +02:00
co-authored by Claude Sonnet 4.6
parent ae27cd5469
commit d926218eff
+134 -1
View File
@@ -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)}