Christian ManivongandClaude Sonnet 4.6
e05d878f94
The method was called but missing — always raised AttributeError, so Fix SNMP did nothing on OpenWRT devices. OpenWRT's default firewall (fw4) policy-drops everything except the ports explicitly listed in the management zone (SSH/HTTP/HTTPS/ICMP). SNMP (UDP/161) is not in that list, which is why snmpd runs but is unreachable from outside the device. Fix adds a persistent UCI firewall rule for UDP/161, reloads fw4 immediately, verifies snmpd is running, and probes locally if SNMP client tools are available. Returns success so callers can clear the snmp_no_data warning. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
412 lines
16 KiB
Python
412 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""NAPALM driver for OpenWrt routers and access-points.
|
|
|
|
Communicates via SSH. The device must be running OpenWrt 19.07 or newer.
|
|
Netmiko device_type: ``linux``
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import socket
|
|
from typing import Any
|
|
|
|
from netmiko import ConnectHandler
|
|
from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
|
|
|
|
from napalm_device_types import AccessPointDriver
|
|
from napalm.base.exceptions import (
|
|
ConnectionException,
|
|
ConnectionClosedException,
|
|
)
|
|
from napalm.base.netmiko_helpers import netmiko_args
|
|
|
|
from napalm_openwrt.interfaces_mixin import OpenWrtInterfaceMixin
|
|
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.packages_mixin import OpenWrtPackageMixin
|
|
from napalm_openwrt.routing_mixin import OpenWrtRoutingMixin
|
|
|
|
|
|
class OpenWrtDriver(
|
|
AccessPointDriver,
|
|
OpenWrtInterfaceMixin,
|
|
OpenWrtWirelessMixin,
|
|
OpenWrtLLDPMixin,
|
|
OpenWrtVLANMixin,
|
|
OpenWrtConfigMixin,
|
|
OpenWrtSystemMixin,
|
|
OpenWrtPackageMixin,
|
|
OpenWrtRoutingMixin,
|
|
):
|
|
"""NAPALM driver for OpenWrt routers and access-points."""
|
|
|
|
VENDOR = "OpenWrt"
|
|
NETMIKO_DEVICE_TYPE = "linux"
|
|
|
|
def __init__(
|
|
self,
|
|
hostname: str,
|
|
username: str,
|
|
password: str,
|
|
timeout: int = 60,
|
|
optional_args: dict[str, Any] | None = None,
|
|
) -> None:
|
|
self.hostname = hostname
|
|
self.username = username
|
|
self.password = password
|
|
self.timeout = timeout
|
|
self.device: ConnectHandler | None = None
|
|
|
|
if optional_args is None:
|
|
optional_args = {}
|
|
|
|
self.port = optional_args.pop("port", 22)
|
|
self.netmiko_optional_args = netmiko_args(optional_args)
|
|
|
|
# Config management state
|
|
self._candidate_config: str | None = None
|
|
self._candidate_mode: str | None = None
|
|
self._backup_config: str | None = None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Connection management
|
|
# ------------------------------------------------------------------
|
|
|
|
def open(self) -> None:
|
|
"""Open an SSH connection to the device."""
|
|
try:
|
|
self.device = ConnectHandler(
|
|
device_type=self.NETMIKO_DEVICE_TYPE,
|
|
host=self.hostname,
|
|
username=self.username,
|
|
password=self.password,
|
|
timeout=self.timeout,
|
|
port=self.port,
|
|
**self.netmiko_optional_args,
|
|
)
|
|
except NetmikoTimeoutException as exc:
|
|
raise ConnectionException(
|
|
f"Cannot connect to {self.hostname}: {exc}"
|
|
) from exc
|
|
except NetmikoAuthenticationException as exc:
|
|
raise ConnectionException(
|
|
f"Authentication failed for {self.hostname}: {exc}"
|
|
) from exc
|
|
|
|
def close(self) -> None:
|
|
"""Close the SSH connection."""
|
|
if self.device:
|
|
self.device.disconnect()
|
|
self.device = None
|
|
|
|
def is_alive(self) -> dict[str, bool]:
|
|
"""Return connection liveness."""
|
|
if self.device is None:
|
|
return {"is_alive": False}
|
|
try:
|
|
return {"is_alive": self.device.remote_conn.transport.is_active()}
|
|
except (socket.error, EOFError, AttributeError):
|
|
return {"is_alive": False}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def _send_command(self, command: str | list[str]) -> str:
|
|
"""Send a shell command (or list of fallback commands) to the device.
|
|
|
|
When a list is supplied, commands are tried in order and the first
|
|
one that does not return an error indicator is returned.
|
|
"""
|
|
def _do_send(cmd: str) -> str:
|
|
return self.device.send_command(
|
|
cmd,
|
|
read_timeout=self.timeout,
|
|
).strip()
|
|
|
|
try:
|
|
if isinstance(command, list):
|
|
output = ""
|
|
for cmd in command:
|
|
output = _do_send(cmd)
|
|
if not output.startswith(("sh: ", "ash: ", "-ash: ", "command not found")):
|
|
return output
|
|
return output
|
|
return _do_send(command)
|
|
except (socket.error, EOFError) as exc:
|
|
raise ConnectionClosedException(str(exc)) from exc
|
|
|
|
@staticmethod
|
|
def _parse_openwrt_release(output: str) -> dict[str, str]:
|
|
"""Parse ``/etc/openwrt_release`` key=value pairs."""
|
|
result: dict[str, str] = {}
|
|
for line in output.splitlines():
|
|
m = re.match(r'^(\w+)=["\']?([^"\']*)["\']?$', line.strip())
|
|
if m:
|
|
result[m.group(1)] = m.group(2)
|
|
return result
|
|
|
|
@staticmethod
|
|
def _parse_uptime_seconds(uptime_str: str) -> float:
|
|
"""Convert ``/proc/uptime`` first field (seconds.hundredths) to float."""
|
|
try:
|
|
return float(uptime_str.split()[0])
|
|
except (IndexError, ValueError):
|
|
return 0.0
|
|
|
|
# ------------------------------------------------------------------
|
|
# NAPALM getters kept in driver
|
|
# ------------------------------------------------------------------
|
|
|
|
def get_facts(self) -> dict[str, Any]:
|
|
"""Return a dictionary of general device facts.
|
|
|
|
Retrieves data from:
|
|
* ``/etc/openwrt_release``
|
|
* ``/proc/uptime``
|
|
* ``uname -a``
|
|
* ``ip link show`` (interface count)
|
|
"""
|
|
facts: dict[str, Any] = {}
|
|
|
|
# OS version
|
|
release_out = self._send_command("cat /etc/openwrt_release")
|
|
release = self._parse_openwrt_release(release_out)
|
|
|
|
facts["vendor"] = "OpenWrt"
|
|
|
|
# Prefer the device-tree model (e.g. "Sophos AP100") over the SoC target
|
|
model_raw = self._send_command("cat /tmp/sysinfo/model 2>/dev/null")
|
|
if model_raw and not model_raw.startswith("cat: "):
|
|
facts["model"] = model_raw.strip().replace("\x00", "")
|
|
else:
|
|
facts["model"] = (
|
|
release.get("DISTRIB_TARGET", "").replace("/", " ")
|
|
)
|
|
|
|
facts["os_version"] = release.get(
|
|
"DISTRIB_DESCRIPTION",
|
|
release.get("DISTRIB_RELEASE", "unknown"),
|
|
)
|
|
facts["serial_number"] = ""
|
|
facts["uptime"] = self._parse_uptime_seconds(
|
|
self._send_command("cat /proc/uptime")
|
|
)
|
|
|
|
# hostname and FQDN
|
|
uname_out = self._send_command("uname -a")
|
|
uname_parts = uname_out.split()
|
|
if len(uname_parts) >= 2:
|
|
facts["hostname"] = uname_parts[1]
|
|
facts["fqdn"] = uname_parts[1]
|
|
else:
|
|
facts["hostname"] = "unknown"
|
|
facts["fqdn"] = "unknown"
|
|
|
|
# Interface count
|
|
interface_list = self._get_interface_list()
|
|
facts["interface_list"] = interface_list
|
|
facts["number_of_interfaces"] = len(interface_list)
|
|
|
|
return facts
|
|
|
|
def get_config(
|
|
self,
|
|
retrieve: str = "all",
|
|
full: bool = False,
|
|
sanitized: bool = False,
|
|
format: str = "text",
|
|
) -> dict[str, str]:
|
|
"""Return the device configuration via ``uci export``.
|
|
|
|
OpenWrt does not have a distinct startup/candidate config concept.
|
|
``running`` and ``startup`` both return ``uci export`` output.
|
|
``candidate`` is always empty.
|
|
"""
|
|
configs = {"running": "", "startup": "", "candidate": ""}
|
|
|
|
if retrieve in ("all", "running"):
|
|
configs["running"] = self._send_command("uci export")
|
|
|
|
if retrieve in ("all", "startup"):
|
|
configs["startup"] = self._send_command("uci export")
|
|
|
|
return configs
|
|
|
|
def cli(
|
|
self, commands: list[str], encoding: str = "text"
|
|
) -> dict[str, str]:
|
|
"""Execute a list of CLI commands and return their output."""
|
|
return {
|
|
cmd: self._send_command(cmd)
|
|
for cmd in commands
|
|
}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Service management (forwarded to mixins because AccessPointDriver
|
|
# defines NotImplementedError stubs for these)
|
|
# ------------------------------------------------------------------
|
|
|
|
def get_services(self) -> list[dict[str, Any]]:
|
|
return OpenWrtSystemMixin.get_services(self)
|
|
|
|
def manage_service(self, name: str, action: str) -> dict[str, Any]:
|
|
return OpenWrtSystemMixin.manage_service(self, name, action)
|
|
|
|
def get_available_updates(self) -> list[dict[str, Any]]:
|
|
return OpenWrtSystemMixin.get_available_updates(self)
|
|
|
|
def apply_updates(self, packages: list[str]) -> dict[str, Any]:
|
|
return OpenWrtSystemMixin.apply_updates(self, packages)
|
|
|
|
def get_packages(self) -> list[dict[str, Any]]:
|
|
return OpenWrtPackageMixin.get_packages(self)
|
|
|
|
def install_package(self, name: str) -> dict[str, Any]:
|
|
return OpenWrtPackageMixin.install_package(self, name)
|
|
|
|
def remove_package(self, name: str) -> dict[str, Any]:
|
|
return OpenWrtPackageMixin.uninstall_package(self, name)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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 _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:
|
|
|
|
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.
|
|
"""
|
|
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'}")
|
|
|
|
# ── 2. Reload firewall to apply immediately ───────────────────────
|
|
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 ────────────────────────
|
|
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")
|
|
|
|
# ── 4. Quick local sanity check via snmpwalk/snmpget if available ─
|
|
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
|
|
else:
|
|
ok_tokens = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:")
|
|
success = any(t in probe for t in ok_tokens)
|
|
lines.append(f"[probe] {'ok' if success else 'FAILED'}: {probe.strip()[:120]}")
|
|
|
|
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/lldpd start 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}
|