napalm_openwrt/openwrt.py

@@ -19,6 +19,7 @@ Netmiko device_type: ``linux``
 
 from __future__ import annotations
 
+import ipaddress
 import re
 import socket
 from typing import Any
@@ -26,7 +27,12 @@ from typing import Any
 from netmiko import ConnectHandler
 from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
 
-from napalm_device_types import AccessPointDriver, FingerprintRule
+from napalm_device_types import (
+    AccessPointDriver,
+    EnsurePackageMixin,
+    FingerprintRule,
+    SnmpAgentMixin,
+)
 from napalm_device_types.listening import ListeningSocketsMixin
 from napalm.base.exceptions import (
     ConnectionException,
@@ -39,11 +45,23 @@ from napalm_openwrt.wireless_mixin import OpenWrtWirelessMixin
 from napalm_openwrt.lldp_mixin import OpenWrtLLDPMixin
 from napalm_openwrt.vlan_mixin import OpenWrtVLANMixin
 from napalm_openwrt.config_mixin import OpenWrtConfigMixin
-from napalm_openwrt.system_mixin import OpenWrtSystemMixin
+from napalm_openwrt.system_mixin import _RC_MARKER, OpenWrtSystemMixin, _split_status
 from napalm_openwrt.packages_mixin import OpenWrtPackageMixin
 from napalm_openwrt.routing_mixin import OpenWrtRoutingMixin
 
 
+#: What may stand in a UCI value written in single quotes: no whitespace, quote or #.
+_SNMP_WORD = re.compile(r"^[^\s'\"#]+$")
+
+
+def _sources(settings: dict) -> list[str]:
+    """The allowed sources as networks, or ValueError."""
+    return [
+        str(ipaddress.ip_network(x.strip(), strict=False))
+        for x in settings.get("allowed_sources", [])
+    ]
+
+
 class OpenWrtDriver(
     OpenWrtInterfaceMixin,
     OpenWrtVLANMixin,
@@ -54,6 +72,8 @@ class OpenWrtDriver(
     OpenWrtPackageMixin,
     OpenWrtRoutingMixin,
     ListeningSocketsMixin,
+    SnmpAgentMixin,
+    EnsurePackageMixin,
     AccessPointDriver,
 ):
     """NAPALM driver for OpenWrt routers and access-points."""
@@ -358,18 +378,6 @@ class OpenWrtDriver(
     # Device actions
     # ------------------------------------------------------------------
 
-    def run_device_action(self, action: str) -> dict[str, Any]:
-        """Execute a named action on the device."""
-        if action == "install_lldpd":
-            return self._action_install_lldpd()
-        if action == "install_auc":
-            return self._action_install_auc()
-        if action == "install_coreutils_base64":
-            return self._action_install_coreutils_base64()
-        if action == "fix_snmp":
-            return self._action_fix_snmp()
-        raise NotImplementedError(f"Unknown action: {action!r}")
-
     def _mgmt_address(self) -> str:
         """Return the device-side address of our own SSH session.
 
@@ -487,43 +495,83 @@ class OpenWrtDriver(
                 return int(line)
         return 0
 
-    def _action_fix_snmp(self) -> dict[str, Any]:
+    # ------------------------------------------------------------------
-        """Ensure snmpd is running and reachable on UDP/161.
+    # SNMP agent (napalm-device-types SnmpAgentMixin): the caller's values
+    # ------------------------------------------------------------------
 
-        On OpenWrt the usual reason SNMP stays unreachable is the firewall:
+    def get_management_peer(self) -> str | None:
-        snmpd listens, but the zone covering the management interface
+        """The address the device sees the SSH connection come from (``SSH_CONNECTION``)."""
-        default-drops everything except SSH/HTTP/HTTPS/ICMP.  A rule only
+        parts = self._send_command("echo $SSH_CONNECTION").split()
-        lands in the right nftables chain when its ``src`` names the zone
+        try:
-        that actually owns the management network.
+            return str(ipaddress.ip_address(parts[0])) if parts else None
+        except ValueError:
+            return None
 
-        This action:
+    def _rc_step(self, command: str, done: str) -> dict[str, Any]:
+        output, status = _split_status(self._send_command(f"{command} 2>&1; echo {_RC_MARKER}$?"))
+        return {"ok": status == 0, "output": done if status == 0 else output.strip()[-300:]}
 
-        1. Resolves the management address from ``$SSH_CONNECTION``, maps it
+    def _install_snmp_agent(self) -> dict[str, Any]:
-           to its ``network`` section and from there to the owning firewall
+        if self._send_command("[ -x /usr/sbin/snmpd ] && echo present").strip() == "present":
-           zone — no guessing from rule names.
+            return {"ok": True, "output": "snmpd already installed"}
-        2. Aborts with a diagnosis when that zone has no ``option name``:
+        if self._pm_type() == "apk":
-           fw4 skips such a section, so any rule written against it is dead
+            return self._rc_step("apk add snmpd", "snmpd installed")
-           on arrival.  Naming the zone is left to the operator because an
+        return self._rc_step("opkg update >/dev/null 2>&1; opkg install snmpd", "snmpd installed")
-           inert zone becoming active changes what the AP filters.
+
-        3. Removes stale SNMP rules from earlier versions (highest anonymous
+    def _write_snmp_agent_config(self, settings: dict) -> dict[str, Any]:
-           index first) and commits the deletion.
+        """Community, sources, contact and location in UCI's snmpd config.
-        4. Writes the ``allow_snmp_from_<zone>`` rule in full every run, so a
+
-           half-written rule from a previous attempt gets repaired.
+        One ``com2sec`` (``com2sec6``) per allowed source, all mapped to the
-        5. Reloads the firewall and fails on any reload complaint.
+        read-only security name the stock groups use. Sections of an earlier run
-        6. Verifies on the device that snmpd listens and that a udp/161
+        (``snmp_src*``) go first. Without sources every address may ask; with only
-           accept rule is live in the packet filter.
+        IPv4 sources, IPv6 answers localhost only.
+        """
+        community = settings["community"]
+        if not _SNMP_WORD.match(community):
+            raise ValueError("the community must be one word without quotes or #")
+        sources = _sources(settings)
+        v4 = [x for x in sources if ":" not in x] or (["127.0.0.1/32"] if sources else ["default"])
+        v6 = [x for x in sources if ":" in x] or (["::1/128"] if sources else ["default"])
+        old = self._parse_uci_sections(self._send_command("uci show snmpd 2>/dev/null"))
+        stale = sorted((key for key in old if key.startswith("snmp_src")), reverse=True)
+        cmds = [f"uci -q delete snmpd.{key}" for key in stale]
+        for kind, first, nets in (("com2sec", "public", v4), ("com2sec6", "public6", v6)):
+            for i, net in enumerate(nets):
+                key = first if i == 0 else f"snmp_src{'6' if kind.endswith('6') else ''}_{i}"
+                cmds += [
+                    f"uci set snmpd.{key}={kind}",
+                    f"uci set snmpd.{key}.secname='ro'",
+                    f"uci set snmpd.{key}.source='{net}'",
+                    f"uci set snmpd.{key}.community='{community}'",
+                ]
+        for key, option in (("location", "sysLocation"), ("contact", "sysContact")):
+            value = settings.get(key, "")
+            if value:
+                if "'" in value or "\n" in value:
+                    raise ValueError(f"{key} must be one line without '")
+                cmds.append(f"uci set snmpd.@system[0].{option}='{value}'")
+        cmds.append("uci commit snmpd")
+        done = f"community set for {', '.join(sources) or 'every address'}"
+        return self._rc_step("; ".join(cmds), done)
+
+    def _allow_snmp_sources(self, settings: dict) -> dict[str, Any]:
+        """A udp/161 accept rule in the zone that owns the management address.
+
+        The zone is resolved from ``$SSH_CONNECTION`` → its ``network`` section →
+        the zone listing it, not guessed from rule names. A zone without ``option
+        name`` is skipped by fw4, so the step fails with how to fix it instead of
+        writing a dead rule. Stale SNMP rules go (highest anonymous index first);
+        the rule is written in full every run and limited to the allowed sources
+        (``src_ip``); the reload and the live packet filter decide success.
         """
         lines: list[str] = []
-
+        sources = _sources(settings)
-        # ── 1. Resolve the zone that owns the management address ─────────
+        fw_sections = self._parse_uci_sections(self._send_command("uci show firewall 2>/dev/null"))
-        fw_raw = self._send_command("uci show firewall 2>/dev/null")
-        fw_sections = self._parse_uci_sections(fw_raw)
         zones, nameless = self._firewall_zones(fw_sections)
-
         address = self._mgmt_address()
         net_sections = self._parse_uci_sections(self._send_command("uci show network 2>/dev/null"))
         mgmt_net = self._mgmt_network(net_sections, address)
-        lines.append(f"[firewall] Management address {address} on network {mgmt_net or '<unknown>'}")
+        lines.append(f"Management address {address} on network {mgmt_net or '<unknown>'}")
 
         broken_zone = (
             self._zone_for_network(fw_sections, {key: key for key in nameless}, mgmt_net)
@@ -532,50 +580,45 @@ class OpenWrtDriver(
         )
         if broken_zone:
             lines.append(
-                f"[firewall] Zone {broken_zone} owns network {mgmt_net!r} but has no "
+                f"Zone {broken_zone} owns network {mgmt_net!r} but has no 'name' option — "
-                f"'name' option — fw4 skips the section and drops every rule that "
+                "fw4 skips the section and drops every rule that references it. "
-                f"references it. Fix on the device, then re-run:"
+                "Fix on the device, then re-run:"
             )
             lines.append(
-                f"[firewall]   uci set firewall.{broken_zone}.name='{mgmt_net}' "
+                f"  uci set firewall.{broken_zone}.name='{mgmt_net}'"
-                f"&& uci commit firewall && fw4 reload"
+                " && uci commit firewall && fw4 reload"
             )
-            return {"success": False, "output": "\n".join(lines)}
+            return {"ok": False, "output": "\n".join(lines)}
 
         mgmt_zone = self._zone_for_network(fw_sections, zones, mgmt_net) if mgmt_net else ""
         if not mgmt_zone and "lan" in zones:
             mgmt_zone = "lan"
         if not mgmt_zone and len(zones) == 1:
             mgmt_zone = next(iter(zones))
-
         if mgmt_zone:
-            lines.append(f"[firewall] Management zone: {mgmt_zone!r}")
+            lines.append(f"Management zone: {mgmt_zone!r}")
         elif zones:
-            lines.append(
+            present = ", ".join(sorted(zones))
-                f"[firewall] No zone covers the management network — zones present: "
+            lines.append(f"No zone covers the management network — zones present: {present}")
-                f"{', '.join(sorted(zones))}"
+            return {"ok": False, "output": "\n".join(lines)}
-            )
-            return {"success": False, "output": "\n".join(lines)}
         else:
-            lines.append("[firewall] No zones configured — writing an unscoped rule")
+            lines.append("No zones configured — writing an unscoped rule")
 
-        # ── 2. Drop stale SNMP rules from earlier versions of this action ─
         named_key = f"allow_snmp_from_{mgmt_zone}" if mgmt_zone else "allow_snmp"
         stale = self._stale_snmp_rules(fw_sections, keep=named_key)
         if stale:
-            self._send_command(
+            deletes = "; ".join(f"uci -q delete firewall.{key}" for key in stale)
-                "; ".join(f"uci -q delete firewall.{key}" for key in stale)
+            self._send_command(f"{deletes}; uci commit firewall")
-                + "; uci commit firewall"
+            lines.append(f"Removed stale SNMP rule(s): {', '.join(stale)}")
-            )
-            lines.append(f"[firewall] Removed stale SNMP rule(s): {', '.join(stale)}")
 
-        # ── 3. Write the rule in full — repairs a half-written one ────────
         rule_cmds = [
             f"uci set firewall.{named_key}=rule",
-            f"uci set firewall.{named_key}.name='Allow-SNMP-netOrk'",
+            f"uci set firewall.{named_key}.name='Allow-SNMP'",
             f"uci set firewall.{named_key}.target='ACCEPT'",
             f"uci set firewall.{named_key}.proto='udp'",
             f"uci set firewall.{named_key}.dest_port='161'",
+            f"uci -q delete firewall.{named_key}.src_ip",
+            *(f"uci add_list firewall.{named_key}.src_ip='{net}'" for net in sources),
         ]
         if mgmt_zone:
             rule_cmds.append(f"uci set firewall.{named_key}.src='{mgmt_zone}'")
@@ -583,86 +626,33 @@ class OpenWrtDriver(
             rule_cmds.append(f"uci -q delete firewall.{named_key}.src")
         rule_cmds.append("uci commit firewall")
         rule_out = self._send_command("; ".join(rule_cmds) + " 2>&1").strip()
-        lines.append(f"[firewall] Wrote rule {named_key!r}: {rule_out or 'ok'}")
+        scope = ", ".join(sources) or "every source in the zone"
+        lines.append(f"Wrote rule {named_key!r} for {scope}: {rule_out or 'ok'}")
 
-        # ── 4. Reload the firewall, and believe what it says ──────────────
         reload_out = self._send_command(
             "fw4 reload 2>&1 || /etc/init.d/firewall reload 2>&1"
         ).strip()
         fw_ok = not self._fw_reload_failed(reload_out)
-        lines.append(f"[firewall] Reload: {reload_out or 'ok'}")
+        lines.append(f"Reload: {reload_out or 'ok'}")
         if not fw_ok:
-            lines.append("[firewall] Reload reported invalid sections — ruleset not applied")
+            lines.append("Reload reported invalid sections — ruleset not applied")
-
-        # ── 5. Ensure snmpd is enabled and running ────────────────────────
-        status = self._send_command("/etc/init.d/snmpd status 2>/dev/null")
-        if "running" not in status.lower() and "active" not in status.lower():
-            self._send_command(
-                "/etc/init.d/snmpd enable 2>/dev/null;"
-                " /etc/init.d/snmpd start 2>/dev/null"
-            )
-            lines.append("[snmpd] Service started and enabled")
-        else:
-            lines.append("[snmpd] Service already running")
-
-        # ── 6. Verify on the device instead of assuming success ───────────
-        listening = self._grep_count(
-            self._send_command("ss -lun 2>/dev/null | grep -c ':161'")
-        )
-        lines.append(f"[probe] snmpd listening on udp/161: {'yes' if listening else 'no'}")
-
         live = self._grep_count(
             self._send_command(
                 "{ nft list ruleset 2>/dev/null || iptables-save 2>/dev/null; }"
                 " | grep -c 'dport 161'"
             )
         )
-        lines.append(f"[probe] udp/161 accept rules live in the packet filter: {live}")
+        lines.append(f"udp/161 accept rules live in the packet filter: {live}")
+        return {"ok": fw_ok and bool(live), "output": "\n".join(lines)}
 
-        success = fw_ok and bool(listening) and bool(live)
+    def _start_snmp_agent(self) -> dict[str, Any]:
-        return {"success": success, "output": "\n".join(lines)}
-
-    def _action_install_coreutils_base64(self) -> dict[str, Any]:
-        """Install coreutils-base64 via the device package manager."""
-        pm = self._pm_type()
-        if pm == "apk":
-            raw = self._send_command("apk add coreutils-base64 2>&1")
-        else:
-            self._send_command("opkg update 2>&1")
-            raw = self._send_command("opkg install coreutils-base64 2>&1")
-        out = self._clean_pkg_output(raw)
-        low = out.lower()
-        success = not any(kw in low for kw in ("error:", "failed", "not found", "unknown package"))
-        return {"success": success, "output": out}
-
-    def _action_install_auc(self) -> dict[str, Any]:
-        """Install the attended sysupgrade client (auc) via opkg."""
-        self._send_command("opkg update 2>&1")
-        raw = self._send_command("opkg install auc 2>&1")
-        out = self._clean_pkg_output(raw)
-        low = out.lower()
-        success = not any(kw in low for kw in ("error:", "failed", "not found", "unknown package"))
-        return {"success": success, "output": out}
-
-    def _action_install_lldpd(self) -> dict[str, Any]:
-        """Install lldpd, add eth0 to its interface list and start the service."""
-        pm = self._pm_type()
-        if pm == "apk":
-            raw = self._send_command("apk add lldpd 2>&1")
-        else:
-            self._send_command("opkg update 2>&1")
-            raw = self._send_command("opkg install lldpd 2>&1")
-        out = self._clean_pkg_output(raw)
-
-        # Configure the correct management interface and enable the service
-        self._lldpd_fix_interface()
-
-        # Enable and start the service
         self._send_command(
-            "/etc/init.d/lldpd enable 2>/dev/null; "
+            "/etc/init.d/snmpd enable 2>/dev/null; /etc/init.d/snmpd restart 2>/dev/null"
-            "/etc/init.d/lldpd start 2>/dev/null"
         )
-
+        listening = self._grep_count(self._send_command("ss -lun 2>/dev/null | grep -c ':161'"))
-        low = out.lower()
+        return {
-        success = not any(kw in low for kw in ("error:", "failed", "not found", "unknown package"))
+            "ok": bool(listening),
-        return {"success": success, "output": out}
+            "output": "snmpd enabled and listening on udp/161"
+            if listening
+            else "snmpd is not listening on udp/161",
+        }

napalm_openwrt/system_mixin.py

@@ -14,7 +14,6 @@
 from __future__ import annotations
 
 import re
-import time as _time
 from typing import Any
 
 
@@ -258,156 +257,6 @@ class OpenWrtSystemMixin:
 
         return SNMPConfigDict(running=True, community=community, port=161, version="2c")
 
-    def _action_fix_snmp(self) -> dict[str, Any]:
-        """Install and configure snmpd on OpenWrt.
-
-        Installs snmpd-nossl (the daemon) and luci-app-snmpd (UCI schema +
-        proper procd init script).  Configures community 'public' via UCI.
-        """
-        lines: list = []
-
-        # 1. Install packages — snmpd-nossl (daemon) + luci-app-snmpd (UCI init)
-        pm = self._pm_type()
-        if pm == "apk":
-            raw = self._send_command("apk add snmpd-nossl luci-app-snmpd 2>&1")
-        else:
-            self._send_command("opkg update 2>/dev/null || true")
-            raw = self._send_command("opkg install snmpd-nossl luci-app-snmpd 2>&1")
-        out = self._clean_pkg_output(raw)
-        low = out.lower()
-        installed = not any(kw in low for kw in ("error:", "failed"))
-        lines.append(f"[install] {out[-300:]}")
-
-        if not installed and "already installed" not in low:
-            return {"success": False, "output": "\n".join(lines)}
-
-        # 2. Configure via UCI — modify the existing default sections only.
-        #    Do NOT create new named sections (causes duplicate directives in
-        #    the generated /var/run/snmpd.conf which crashes snmpd).
-        #    Also remove any stale named sections from previous fix attempts.
-        # The init script reads these UCI field names to generate /var/run/snmpd.conf:
-        #   agent:  agentaddress
-        #   com2sec: secname, source, community
-        #   group:  group (name!), version, secname
-        #   view:   viewname (not name!), type, oid
-        #   access: group, version, level, prefix, read, write, notify
-        # Default luci-app-snmpd schema uses different field names for group/view/access,
-        # so we patch all required fields explicitly.
-        uci_cmds = [
-            # Remove any stale named sections from previous runs
-            "uci -q delete snmpd.agent",
-            "uci -q delete snmpd.public",
-            # agent
-            "uci set snmpd.@agent[0].agentaddress='161'",
-            # com2sec
-            "uci set snmpd.@com2sec[0].secname='ro'",
-            "uci set snmpd.@com2sec[0].source='0.0.0.0/0'",
-            "uci set snmpd.@com2sec[0].community='public'",
-            # group — init script reads field 'group' (not 'name')
-            "uci set snmpd.@group[0].group='rogroup'",
-            "uci set snmpd.@group[0].version='v2c'",
-            "uci set snmpd.@group[0].secname='ro'",
-            # view — init script reads field 'viewname' (not 'name')
-            "uci set snmpd.@view[0].viewname='all'",
-            "uci set snmpd.@view[0].type='included'",
-            "uci set snmpd.@view[0].oid='.1'",
-            # access — init script needs write + notify or it returns early
-            "uci set snmpd.@access[0].group='rogroup'",
-            "uci set snmpd.@access[0].context='none'",
-            "uci set snmpd.@access[0].version='v2c'",
-            "uci set snmpd.@access[0].level='noAuthNoPriv'",
-            "uci set snmpd.@access[0].prefix='exact'",
-            "uci set snmpd.@access[0].read='all'",
-            "uci set snmpd.@access[0].write='none'",
-            "uci set snmpd.@access[0].notify='none'",
-            "uci commit snmpd",
-        ]
-        for cmd in uci_cmds:
-            self._send_command(f"{cmd} 2>/dev/null || true")
-        lines.append("[config] Configured snmpd via UCI (all required fields set).")
-
-        # 3. Firewall: allow UDP 161 from the management zone
-        try:
-            # Use `ss` (always available on OpenWrt) to find the IP of the current SSH client
-            raw_conn = self._send_command(
-                "ss -tnp 2>/dev/null | awk '/ESTAB.*:22/{print $5}' | head -1 | sed 's/:[0-9]*$//'"
-            ).strip()
-            # Strip IPv6-mapped prefix if present
-            if raw_conn.startswith("::ffff:"):
-                raw_conn = raw_conn[7:]
-            peer_ip = raw_conn.splitlines()[-1].strip() if raw_conn else ""
-
-            if peer_ip and peer_ip not in ("", "0.0.0.0", "::"):
-                # Determine which firewall zone owns the interface that routes to peer_ip.
-                # Walk uci firewall zones: find the zone whose associated network interface
-                # has an address in the same /24 as peer_ip.
-                peer_prefix = peer_ip.rsplit(".", 1)[0] if "." in peer_ip else ""
-                fw_out = self._send_command("uci show firewall 2>/dev/null || true")
-                src_zone = "*"
-                # Build map zone_name → zone_uci_key
-                zone_map: dict[str, str] = {}
-                for fw_line in fw_out.splitlines():
-                    m = re.match(r"firewall\.(\w+)\.name='([^']+)'", fw_line.strip())
-                    if m:
-                        zone_map[m.group(2)] = m.group(1)
-                # For each zone find its network, then the interface IP
-                for zone_name, uci_key in zone_map.items():
-                    net_line = self._send_command(
-                        f"uci -q get firewall.{uci_key}.network 2>/dev/null || true"
-                    ).strip()
-                    if not net_line:
-                        continue
-                    for net_name in net_line.split():
-                        iface_ip = self._send_command(
-                            f"uci -q get network.{net_name}.ipaddr 2>/dev/null || true"
-                        ).strip()
-                        if iface_ip and "." in iface_ip:
-                            iface_prefix = iface_ip.split("/")[0].rsplit(".", 1)[0]
-                            if peer_prefix and iface_prefix == peer_prefix:
-                                src_zone = zone_name
-                                break
-                    if src_zone != "*":
-                        break
-
-                self._send_command(
-                    f"uci -q delete firewall.snmp_netork 2>/dev/null; "
-                    f"uci set firewall.snmp_netork=rule; "
-                    f"uci set firewall.snmp_netork.name='Allow-SNMP-from-mgmt'; "
-                    f"uci set firewall.snmp_netork.src='{src_zone}'; "
-                    f"uci set firewall.snmp_netork.dest_port='161'; "
-                    f"uci set firewall.snmp_netork.proto='udp'; "
-                    f"uci set firewall.snmp_netork.target='ACCEPT'; "
-                    f"uci commit firewall; "
-                    f"/etc/init.d/firewall reload 2>/dev/null || true"
-                )
-                lines.append(f"[firewall] Added UDP:161 allow rule (src zone: {src_zone}).")
-            else:
-                lines.append("[firewall] Could not detect peer IP via ss — skipping firewall step.")
-        except Exception as exc:
-            lines.append(f"[firewall] skipped — {exc}")
-
-        # 4. Break any crash-loop, then start cleanly
-        self._send_command("/etc/init.d/snmpd stop 2>/dev/null; true")
-        _time.sleep(2)
-        self._send_command("pkill -9 snmpd 2>/dev/null; true")  # kill crash-loop zombie
-        _time.sleep(3)
-        self._send_command("/etc/init.d/snmpd enable 2>/dev/null; true")
-        self._send_command("/etc/init.d/snmpd start 2>/dev/null; true")
-        _time.sleep(4)
-        lines.append("[service] snmpd started via procd.")
-
-        # 5. Check if snmpd is now active (no local snmpget on OpenWrt by default)
-        status = self._send_command(
-            "/etc/init.d/snmpd running 2>/dev/null && echo active || echo inactive"
-        ).strip()
-        success = status == "active"
-        if success:
-            lines.append("[ok] snmpd is active.")
-        else:
-            lines.append(f"[warn] snmpd status: {status}")
-
-        return {"success": success, "output": "\n".join(lines)}
-
     def get_users(self) -> dict[str, Any]:
         """Return users configured on the device.
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-openwrt"
-version = "0.1.0"
+version = "0.2.0"
 description = "NAPALM driver for OpenWrt routers/access-points"
 readme = "README.md"
 license = { text = "Apache-2.0" }
@@ -26,7 +26,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0.0",
-    "napalm_device_types>=2.5.0",
+    "napalm_device_types>=4.13.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
     "netaddr",

tests/unit/test_driver.py

@@ -1488,6 +1488,12 @@ class _FakeShell:
 
     def __call__(self, cmd, **kw):
         self.issued.append(cmd)
+        if "/usr/sbin/snmpd" in cmd:
+            return "present"
+        if "__OWRT_RC" in cmd:
+            return "ok\n__OWRT_RC=0"
+        if cmd.startswith("uci show snmpd"):
+            return ""
         if cmd.startswith("uci show firewall"):
             return self.firewall
         if cmd.startswith("uci show network"):
@@ -1507,8 +1513,11 @@ class _FakeShell:
         return ""
 
 
+_SETTINGS = {"community": "c0mm", "allowed_sources": []}
+
+
 class TestFixSnmpZoneDetection:
-    """_action_fix_snmp() must resolve the real zone that owns the mgmt address."""
+    """The firewall step must resolve the real zone that owns the mgmt address."""
 
     def test_static_mgmt_address_selects_owning_zone(self, driver):
         """10.10.0.5 lives on network 'mgmt' → zone 'mgmt', not the 'lan' fallback."""
@@ -1518,7 +1527,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="10.10.0.1 51234 10.10.0.5 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "Management zone: 'mgmt'" in result["output"]
         assert any("firewall.allow_snmp_from_mgmt.src='mgmt'" in c for c in shell.issued)
 
@@ -1530,7 +1539,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "Management zone: 'lan'" in result["output"]
 
     def test_dhcp_mgmt_address_resolved_via_l3_device(self, driver):
@@ -1542,7 +1551,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="10.10.0.1 51234 10.10.0.5 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "Management zone: 'lan'" in result["output"]
 
     def test_nameless_zone_is_reported_and_action_fails(self, driver):
@@ -1553,7 +1562,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
         assert "@zone[0]" in result["output"]
         assert "name" in result["output"]
@@ -1566,7 +1575,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         assert not any("allow_snmp" in c for c in shell.issued)
 
 
@@ -1586,7 +1595,7 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         assert any("firewall.allow_snmp_from_lan.src='lan'" in c for c in shell.issued)
 
     def test_stale_rules_are_deleted_highest_index_first(self, driver):
@@ -1605,7 +1614,7 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         deletes = [c for c in shell.issued if "delete" in c and "@rule" in c]
         joined = " ".join(deletes)
         assert joined.index("@rule[2]") < joined.index("@rule[1]")
@@ -1623,7 +1632,7 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         delete_idx = next(i for i, c in enumerate(shell.issued) if "snmp_netork" in c and "delete" in c)
         assert any("uci commit firewall" in c for c in shell.issued[delete_idx:])
 
@@ -1641,7 +1650,7 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         assert any("snmp_netork" in c and "delete" in c for c in shell.issued)
 
 
@@ -1663,7 +1672,7 @@ class TestFixSnmpReloadVerification:
             reload_out=self.FW4_ZONE_ERROR,
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
 
     def test_reload_output_is_not_truncated(self, driver):
@@ -1675,7 +1684,7 @@ class TestFixSnmpReloadVerification:
             reload_out=self.FW4_ZONE_ERROR,
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "references unknown zone" in result["output"]
 
     def test_missing_live_rule_fails_the_action(self, driver):
@@ -1687,7 +1696,7 @@ class TestFixSnmpReloadVerification:
             nft_hits="0",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
 
     def test_snmpd_not_listening_fails_the_action(self, driver):
@@ -1698,7 +1707,7 @@ class TestFixSnmpReloadVerification:
             listen_hits="0",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
 
     def test_fully_healthy_run_succeeds(self, driver):
@@ -1708,7 +1717,7 @@ class TestFixSnmpReloadVerification:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is True
 
 

tests/unit/test_snmp_agent.py

@@ -0,0 +1,96 @@
+"""configure_snmp_agent's values: community, sources, contact, location (netOrk #1105).
+
+The old fix_snmp set no community at all and opened the management zone to every
+source; the rule was named after an application. Every value is the caller's now.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_openwrt import OpenWrtDriver
+from tests.unit.test_driver import UCI_FIREWALL_HEALTHY, UCI_NETWORK_STATIC, _FakeShell
+
+SETTINGS = {
+    "community": "c0mm",
+    "allowed_sources": ["10.7.224.20", "fd00::20"],
+    "contact": "ops",
+    "location": "rack 1",
+}
+
+
+@pytest.fixture
+def shell():
+    return _FakeShell(
+        firewall=UCI_FIREWALL_HEALTHY,
+        network=UCI_NETWORK_STATIC,
+        ssh_connection="10.7.224.20 51234 192.168.1.1 22",
+    )
+
+
+@pytest.fixture
+def owrt(shell):
+    d = OpenWrtDriver("ap", "root", "pw")
+    d._send_command = shell
+    return d
+
+
+def _joined(shell):
+    return " ".join(shell.issued)
+
+
+def test_the_steps_run_in_order_and_succeed(owrt):
+    result = owrt.configure_snmp_agent(SETTINGS)
+
+    assert result["success"] is True
+    assert [s["step"] for s in result["steps"]] == ["install", "configure", "firewall", "start"]
+
+
+def test_community_per_source_in_uci(owrt, shell):
+    owrt.configure_snmp_agent(SETTINGS)
+
+    uci = _joined(shell)
+    assert "uci set snmpd.public.source='10.7.224.20/32'" in uci
+    assert "uci set snmpd.public.community='c0mm'" in uci
+    assert "uci set snmpd.public6.source='fd00::20/128'" in uci
+    assert "snmpd.@system[0].sysLocation='rack 1'" in uci
+    assert "uci commit snmpd" in uci
+
+
+def test_without_sources_every_address_may_ask(owrt, shell):
+    owrt.configure_snmp_agent({"community": "c0mm", "allowed_sources": []})
+
+    assert "uci set snmpd.public.source='default'" in _joined(shell)
+
+
+def test_ipv4_sources_only_keep_ipv6_to_localhost(owrt, shell):
+    owrt.configure_snmp_agent({"community": "c0mm", "allowed_sources": ["10.7.224.20"]})
+
+    assert "uci set snmpd.public6.source='::1/128'" in _joined(shell)
+
+
+def test_the_firewall_rule_is_limited_to_the_sources_and_neutrally_named(owrt, shell):
+    owrt.configure_snmp_agent(SETTINGS)
+
+    uci = _joined(shell)
+    assert "uci add_list firewall.allow_snmp_from_lan.src_ip='10.7.224.20/32'" in uci
+    assert "firewall.allow_snmp_from_lan.name='Allow-SNMP'" in uci
+    assert "netork" not in uci.lower()
+
+
+def test_the_management_peer_is_the_ssh_client(owrt):
+    assert owrt.get_management_peer() == "10.7.224.20"
+
+
+@pytest.mark.parametrize("community", ["two words", "it's", ""])
+def test_a_community_that_would_break_uci_is_refused(owrt, shell, community):
+    result = owrt.configure_snmp_agent({"community": community, "allowed_sources": []})
+
+    assert result["success"] is False
+    assert "uci commit snmpd" not in _joined(shell)
+
+
+def test_the_old_actions_are_gone():
+    gone = ("run_device_action", "_action_fix_snmp", "_action_install_lldpd", "_action_install_auc")
+    for name in gone:
+        assert not hasattr(OpenWrtDriver, name), name