diff --git a/napalm_openwrt/openwrt.py b/napalm_openwrt/openwrt.py index 2716082..87a60ee 100644 --- a/napalm_openwrt/openwrt.py +++ b/napalm_openwrt/openwrt.py @@ -303,61 +303,100 @@ class OpenWrtDriver( def _action_fix_snmp(self) -> dict[str, Any]: """Ensure snmpd is running and reachable on UDP/161. - On OpenWRT the most common reason SNMP is unreachable is a missing - firewall rule — the AP firewall default-drops everything except the - ports listed in the management zone (SSH/HTTP/HTTPS/ICMP). This - action: + On OpenWRT the most common reason SNMP is unreachable is that the + firewall management zone (typically named ``mgmt``) only allows + SSH/HTTP/HTTPS/ICMP by default and default-drops everything else. + snmpd runs but packets are rejected before reaching the process. - 1. Adds a UCI firewall rule that allows UDP/161 on all interfaces and - commits it so it persists across reboots. - 2. Reloads the firewall to apply immediately (no reboot needed). - 3. Ensures the snmpd service is enabled and running. - 4. Returns success once snmpd responds on loopback. + This action: + 1. Detects the management zone name from UCI (looks for the zone + whose named rules already allow SSH — that zone handles the + management interface where SNMP needs to be open too). + 2. Removes any wrongly-scoped SNMP rule (one without src=). + 3. Adds a named UCI rule ``allow_snmp_from_`` with + ``src=`` so it ends up in the correct nftables chain. + 4. Commits and reloads fw4 immediately (no reboot needed). + 5. Ensures snmpd is enabled and running. """ lines: list[str] = [] - # ── 1. Add persistent firewall rule via UCI ─────────────────────── - # Check if a rule for SNMP already exists to avoid duplicates. - existing = self._send_command("uci show firewall 2>/dev/null | grep -i snmp") - if "snmp" in existing.lower(): - lines.append("[firewall] SNMP rule already present in UCI — skipping add") - else: - # Find the next free rule index and add the rule - rule_out = self._send_command( - "uci add firewall rule" - " && uci set firewall.@rule[-1].name='Allow-SNMP'" - " && uci set firewall.@rule[-1].target='ACCEPT'" - " && uci set firewall.@rule[-1].proto='udp'" - " && uci set firewall.@rule[-1].dest_port='161'" - " && uci set firewall.@rule[-1].family='ipv4'" - " && uci commit firewall" - " 2>&1" - ) - lines.append(f"[firewall] Added UCI SNMP rule: {rule_out.strip()[:120] or 'ok'}") + # ── 1. Detect management zone name ─────────────────────────────── + # Find the zone whose allow-SSH rule already exists — that is the + # management zone. Falls back to "lan" if nothing more specific + # is found (on vanilla APs without a dedicated mgmt zone the lan + # zone has input=ACCEPT anyway). + fw_raw = self._send_command("uci show firewall 2>/dev/null") - # ── 2. Reload firewall to apply immediately ─────────────────────── + mgmt_zone = "lan" # safe fallback — lan zone usually has ACCEPT + for line in fw_raw.splitlines(): + # Named rule pattern: firewall.allow_ssh_from_.src='' + if ".src=" in line and "ssh" in line.lower(): + zone_val = line.split("=", 1)[-1].strip().strip("'\"") + if zone_val: + mgmt_zone = zone_val + break + + lines.append(f"[firewall] Management zone: {mgmt_zone!r}") + + # ── 2. Clean up any wrongly-scoped previous SNMP rule ──────────── + # A rule named Allow-SNMP without src= lands in the global input + # chain which is never reached for managed-zone traffic. + existing_names = [ + ln.split("=")[0].strip() + for ln in fw_raw.splitlines() + if ".name='Allow-SNMP'" in ln or ".name='allow_snmp" in ln.lower() + ] + for uci_key in existing_names: + # Check whether this rule has the correct src + src_line = next( + (l for l in fw_raw.splitlines() if uci_key.replace(".name", ".src") in l), + "", + ) + if f"='{mgmt_zone}'" not in src_line and f'="{mgmt_zone}"' not in src_line: + self._send_command(f"uci delete {uci_key.replace('.name', '')} 2>/dev/null || true") + lines.append(f"[firewall] Removed mis-scoped rule {uci_key}") + + # ── 3. Add correctly-scoped rule if not already present ────────── + named_key = f"allow_snmp_from_{mgmt_zone}" + if f"firewall.{named_key}" in fw_raw: + lines.append(f"[firewall] Rule {named_key!r} already present — skipping add") + else: + rule_out = self._send_command( + f"uci set firewall.{named_key}=rule" + f" && uci set firewall.{named_key}.name='Allow-SNMP-from-{mgmt_zone}'" + f" && uci set firewall.{named_key}.src='{mgmt_zone}'" + 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 commit firewall 2>&1" + ) + lines.append(f"[firewall] Added rule {named_key!r}: {rule_out.strip()[:80] or 'ok'}") + + # ── 4. Reload firewall ──────────────────────────────────────────── reload_out = self._send_command( "fw4 reload 2>&1 || /etc/init.d/firewall reload 2>&1 || true" ) lines.append(f"[firewall] Reload: {reload_out.strip()[:120] or 'ok'}") - # ── 3. Ensure snmpd is enabled and running ──────────────────────── + # ── 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") + 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") - # ── 4. Quick local sanity check via snmpwalk/snmpget if available ─ + # ── 6. Local probe (best-effort) ────────────────────────────────── probe = self._send_command( "snmpget -v2c -cpublic -t2 -r0 -Ov 127.0.0.1 1.3.6.1.2.1.1.1.0 2>&1" - " || snmpwalk -v2c -cpublic -t2 -r0 127.0.0.1 1.3.6.1.2.1.1.1.0 2>&1" " || echo 'snmp_client_not_available'" ) if "snmp_client_not_available" in probe: - lines.append("[probe] No local SNMP client — assuming ok (firewall rule added)") - success = True + lines.append("[probe] No local SNMP client — cannot verify locally") + success = True # firewall rule was added; remote poll will confirm else: ok_tokens = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:") success = any(t in probe for t in ok_tokens)