diff --git a/napalm_openwrt/openwrt.py b/napalm_openwrt/openwrt.py index fa11ef1..739ab47 100644 --- a/napalm_openwrt/openwrt.py +++ b/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: - snmpd listens, but the zone covering the management interface - default-drops everything except SSH/HTTP/HTTPS/ICMP. A rule only - lands in the right nftables chain when its ``src`` names the zone - that actually owns the management network. + def get_management_peer(self) -> str | None: + """The address the device sees the SSH connection come from (``SSH_CONNECTION``).""" + parts = self._send_command("echo $SSH_CONNECTION").split() + try: + 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 - to its ``network`` section and from there to the owning firewall - zone — no guessing from rule names. - 2. Aborts with a diagnosis when that zone has no ``option name``: - fw4 skips such a section, so any rule written against it is dead - on arrival. Naming the zone is left to the operator because an - inert zone becoming active changes what the AP filters. - 3. Removes stale SNMP rules from earlier versions (highest anonymous - index first) and commits the deletion. - 4. Writes the ``allow_snmp_from_`` rule in full every run, so a - half-written rule from a previous attempt gets repaired. - 5. Reloads the firewall and fails on any reload complaint. - 6. Verifies on the device that snmpd listens and that a udp/161 - accept rule is live in the packet filter. + def _install_snmp_agent(self) -> dict[str, Any]: + if self._send_command("[ -x /usr/sbin/snmpd ] && echo present").strip() == "present": + return {"ok": True, "output": "snmpd already installed"} + if self._pm_type() == "apk": + return self._rc_step("apk add snmpd", "snmpd installed") + return self._rc_step("opkg update >/dev/null 2>&1; opkg install snmpd", "snmpd installed") + + def _write_snmp_agent_config(self, settings: dict) -> dict[str, Any]: + """Community, sources, contact and location in UCI's snmpd config. + + One ``com2sec`` (``com2sec6``) per allowed source, all mapped to the + read-only security name the stock groups use. Sections of an earlier run + (``snmp_src*``) go first. Without sources every address may ask; with only + 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] = [] - - # ── 1. Resolve the zone that owns the management address ───────── - fw_raw = self._send_command("uci show firewall 2>/dev/null") - fw_sections = self._parse_uci_sections(fw_raw) + sources = _sources(settings) + fw_sections = self._parse_uci_sections(self._send_command("uci show firewall 2>/dev/null")) 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 ''}") + lines.append(f"Management address {address} on network {mgmt_net or ''}") 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"'name' option — fw4 skips the section and drops every rule that " - f"references it. Fix on the device, then re-run:" + f"Zone {broken_zone} owns network {mgmt_net!r} but has no 'name' option — " + "fw4 skips the section and drops every rule that references it. " + "Fix on the device, then re-run:" ) lines.append( - f"[firewall] uci set firewall.{broken_zone}.name='{mgmt_net}' " - f"&& uci commit firewall && fw4 reload" + f" uci set firewall.{broken_zone}.name='{mgmt_net}'" + " && 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( - f"[firewall] No zone covers the management network — zones present: " - f"{', '.join(sorted(zones))}" - ) - return {"success": False, "output": "\n".join(lines)} + present = ", ".join(sorted(zones)) + lines.append(f"No zone covers the management network — zones present: {present}") + return {"ok": 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( - "; ".join(f"uci -q delete firewall.{key}" for key in stale) - + "; uci commit firewall" - ) - lines.append(f"[firewall] Removed stale SNMP rule(s): {', '.join(stale)}") + deletes = "; ".join(f"uci -q delete firewall.{key}" for key in stale) + self._send_command(f"{deletes}; uci commit firewall") + lines.append(f"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") - - # ── 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'}") - + lines.append("Reload reported invalid sections — ruleset not applied") 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) - 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 + def _start_snmp_agent(self) -> dict[str, Any]: self._send_command( - "/etc/init.d/lldpd enable 2>/dev/null; " - "/etc/init.d/lldpd start 2>/dev/null" + "/etc/init.d/snmpd enable 2>/dev/null; /etc/init.d/snmpd restart 2>/dev/null" ) - - low = out.lower() - success = not any(kw in low for kw in ("error:", "failed", "not found", "unknown package")) - return {"success": success, "output": out} + listening = self._grep_count(self._send_command("ss -lun 2>/dev/null | grep -c ':161'")) + return { + "ok": bool(listening), + "output": "snmpd enabled and listening on udp/161" + if listening + else "snmpd is not listening on udp/161", + } diff --git a/napalm_openwrt/system_mixin.py b/napalm_openwrt/system_mixin.py index 6694940..be007f5 100644 --- a/napalm_openwrt/system_mixin.py +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index 79e52af..a4e9d6a 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 7b24938..fd6a793 100644 --- a/tests/unit/test_driver.py +++ b/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 diff --git a/tests/unit/test_snmp_agent.py b/tests/unit/test_snmp_agent.py new file mode 100644 index 0000000..a56bb4a --- /dev/null +++ b/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