Christian ManivongandClaude Sonnet 4.6
e3a0a9e4b3
Ergänzt DRIVER_NAME, HTTP_FINGERPRINT, SNMP_FINGERPRINT, SSH_FINGERPRINT, PORT_SPECS und SNMP_OBJECT_ID_PREFIX gemäß docs/DISCOVERY_FINGERPRINTING.md. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
466 lines
19 KiB
Python
466 lines
19 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, FingerprintRule
|
|
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"
|
|
DRIVER_NAME = "openwrt"
|
|
SSH_FINGERPRINT = [
|
|
FingerprintRule("dropbear", weight=10.0, mandatory=True),
|
|
]
|
|
HTTP_FINGERPRINT = [
|
|
FingerprintRule("luci", weight=8.0, mandatory=True),
|
|
FingerprintRule("lua configuration", weight=6.0),
|
|
FingerprintRule("openwrt", weight=6.0),
|
|
]
|
|
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)
|
|
|
|
# Hardware vendor from board_name (e.g. "sophos,ap100" → "Sophos")
|
|
# Fall back to "OpenWrt" when board_name is unavailable or has no comma.
|
|
board_name_raw = self._send_command("cat /tmp/sysinfo/board_name 2>/dev/null").strip()
|
|
if board_name_raw and "," in board_name_raw and not board_name_raw.startswith("cat:"):
|
|
facts["vendor"] = board_name_raw.split(",")[0].strip().title()
|
|
else:
|
|
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 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.
|
|
|
|
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=<zone>).
|
|
3. Adds a named UCI rule ``allow_snmp_from_<zone>`` with
|
|
``src=<zone>`` 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. 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")
|
|
|
|
mgmt_zone = "lan" # safe fallback — lan zone usually has ACCEPT
|
|
for line in fw_raw.splitlines():
|
|
# Named rule pattern: firewall.allow_ssh_from_<zone>.src='<zone>'
|
|
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'}")
|
|
|
|
# ── 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. 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"
|
|
" || echo 'snmp_client_not_available'"
|
|
)
|
|
if "snmp_client_not_available" in probe:
|
|
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)
|
|
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}
|