Christian Manivong
fd972a427e
AccessPointDriver used to declare get_services, manage_service, get_available_updates, apply_updates, get_packages, install_package and remove_package as NotImplementedError stubs. Those stubs preceded OpenWrtSystemMixin and OpenWrtPackageMixin in the MRO and shadowed their working implementations, so this driver carried a forwarder for each one purely to delegate past the base. napalm-device-types v1.0 makes role bases declaration-only, so nothing shadows anything and all seven forwarders are dead weight. Every method now resolves straight to the mixin that implements it. remove_package went with them: the base now calls the method by the name the package mixin already used, uninstall_package, so the name adapter is gone too.
651 lines
25 KiB
Python
651 lines
25 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(
|
|
OpenWrtInterfaceMixin,
|
|
OpenWrtVLANMixin,
|
|
OpenWrtWirelessMixin,
|
|
OpenWrtLLDPMixin,
|
|
OpenWrtConfigMixin,
|
|
OpenWrtSystemMixin,
|
|
OpenWrtPackageMixin,
|
|
OpenWrtRoutingMixin,
|
|
AccessPointDriver,
|
|
):
|
|
"""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)
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# UCI parsing helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
_UCI_TOKEN_RE = re.compile(r"'([^']*)'|\"([^\"]*)\"|(\S+)")
|
|
|
|
@classmethod
|
|
def _uci_tokens(cls, value: str) -> list[str]:
|
|
"""Split a ``uci show`` value into its (possibly quoted) tokens.
|
|
|
|
List options are printed as ``opt='a' 'b'``; scalars as ``opt='a'``.
|
|
"""
|
|
return [a or b or c for a, b, c in cls._UCI_TOKEN_RE.findall(value)]
|
|
|
|
@classmethod
|
|
def _uci_scalar(cls, value: str) -> str:
|
|
"""Return the first token of a ``uci show`` value, unquoted."""
|
|
tokens = cls._uci_tokens(value)
|
|
return tokens[0] if tokens else ""
|
|
|
|
@classmethod
|
|
def _parse_uci_sections(cls, raw: str) -> dict[str, dict[str, str]]:
|
|
"""Parse ``uci show <pkg>`` into ``{section: {"_type": t, opt: raw_value}}``.
|
|
|
|
Option values are kept verbatim so that list options survive; use
|
|
:meth:`_uci_scalar` / :meth:`_uci_tokens` to read them.
|
|
"""
|
|
sections: dict[str, dict[str, str]] = {}
|
|
for line in raw.splitlines():
|
|
line = line.strip()
|
|
if not line or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
parts = key.split(".")
|
|
if len(parts) == 2: # firewall.@zone[0]=zone
|
|
sections.setdefault(parts[1], {})["_type"] = cls._uci_scalar(value)
|
|
elif len(parts) >= 3: # firewall.@zone[0].name='lan'
|
|
sections.setdefault(parts[1], {})[parts[2]] = value.strip()
|
|
return sections
|
|
|
|
@staticmethod
|
|
def _uci_delete_order(section: str) -> tuple[int, int]:
|
|
"""Sort key that deletes named sections first, then anonymous descending.
|
|
|
|
``uci delete firewall.@rule[1]`` renumbers every later ``@rule[n]``,
|
|
so a batch of deletes only stays correct when the highest index goes
|
|
first.
|
|
"""
|
|
match = re.fullmatch(r"@[\w-]+\[(-?\d+)\]", section)
|
|
return (1, -int(match.group(1))) if match else (0, 0)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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.
|
|
|
|
``$SSH_CONNECTION`` is ``<client ip> <client port> <server ip>
|
|
<server port>`` — the third field is the address the device is
|
|
managed on, which is what the firewall rule has to cover. Falls
|
|
back to the address NAPALM connected to.
|
|
"""
|
|
parts = self._send_command("echo $SSH_CONNECTION").split()
|
|
return parts[2] if len(parts) >= 3 else self.hostname
|
|
|
|
def _l3_device_for_address(self, address: str) -> str:
|
|
"""Return the interface holding ``address`` (``br-lan``, ``eth0.9``, …)."""
|
|
raw = self._send_command("ip -o -4 addr show 2>/dev/null")
|
|
for line in raw.splitlines():
|
|
parts = line.split()
|
|
if len(parts) >= 4 and parts[2] == "inet" and parts[3].split("/")[0] == address:
|
|
return parts[1]
|
|
return ""
|
|
|
|
def _mgmt_network(self, net_sections: dict[str, dict[str, str]], address: str) -> str:
|
|
"""Return the ``network`` section that carries the management address."""
|
|
for key, opts in net_sections.items():
|
|
if opts.get("_type") != "interface":
|
|
continue
|
|
if self._uci_scalar(opts.get("ipaddr", "")) == address:
|
|
return key
|
|
# DHCP-addressed management interface: address → L3 device → section
|
|
device = self._l3_device_for_address(address)
|
|
if device:
|
|
for key, opts in net_sections.items():
|
|
if opts.get("_type") != "interface":
|
|
continue
|
|
if self._uci_scalar(opts.get("device", "")) == device:
|
|
return key
|
|
return ""
|
|
|
|
@classmethod
|
|
def _firewall_zones(
|
|
cls, fw_sections: dict[str, dict[str, str]]
|
|
) -> tuple[dict[str, str], list[str]]:
|
|
"""Split zone sections into ``{name: section}`` and the nameless ones.
|
|
|
|
A ``config zone`` without ``option name`` is invalid for fw4: the
|
|
section is skipped and every rule whose ``src`` points at it is
|
|
dropped along with it.
|
|
"""
|
|
named: dict[str, str] = {}
|
|
nameless: list[str] = []
|
|
for key, opts in fw_sections.items():
|
|
if opts.get("_type") != "zone":
|
|
continue
|
|
name = cls._uci_scalar(opts.get("name", ""))
|
|
if name:
|
|
named[name] = key
|
|
else:
|
|
nameless.append(key)
|
|
return named, sorted(nameless)
|
|
|
|
@classmethod
|
|
def _zone_for_network(
|
|
cls,
|
|
fw_sections: dict[str, dict[str, str]],
|
|
candidates: dict[str, str],
|
|
network: str,
|
|
) -> str:
|
|
"""Return the label of the candidate zone whose ``network`` list holds ``network``.
|
|
|
|
``candidates`` maps a label (zone name, or section key for zones that
|
|
have none) to the UCI section it lives in.
|
|
"""
|
|
for label, section in candidates.items():
|
|
if network in cls._uci_tokens(fw_sections.get(section, {}).get("network", "")):
|
|
return label
|
|
return ""
|
|
|
|
@classmethod
|
|
def _stale_snmp_rules(
|
|
cls, fw_sections: dict[str, dict[str, str]], keep: str
|
|
) -> list[str]:
|
|
"""Return every SNMP rule section except ``keep``, in delete-safe order."""
|
|
hits = [
|
|
key
|
|
for key, opts in fw_sections.items()
|
|
if opts.get("_type") == "rule"
|
|
and key != keep
|
|
and (
|
|
"snmp" in cls._uci_scalar(opts.get("name", "")).lower()
|
|
or "snmp" in key.lower()
|
|
or cls._uci_scalar(opts.get("dest_port", "")) == "161"
|
|
)
|
|
]
|
|
return sorted(hits, key=cls._uci_delete_order)
|
|
|
|
_FW_RELOAD_ERRORS = (
|
|
"mandatory but not set",
|
|
"skipped due to invalid",
|
|
"references unknown",
|
|
"is not a valid",
|
|
"error:",
|
|
)
|
|
|
|
@classmethod
|
|
def _fw_reload_failed(cls, output: str) -> bool:
|
|
"""True when fw4/fw3 refused part of the ruleset."""
|
|
low = output.lower()
|
|
return any(token in low for token in cls._FW_RELOAD_ERRORS)
|
|
|
|
@staticmethod
|
|
def _grep_count(output: str) -> int:
|
|
"""Read the count printed by ``grep -c``; 0 when the command failed."""
|
|
for line in reversed(output.strip().splitlines()):
|
|
line = line.strip()
|
|
if line.isdigit():
|
|
return int(line)
|
|
return 0
|
|
|
|
def _action_fix_snmp(self) -> dict[str, Any]:
|
|
"""Ensure snmpd is running and reachable on UDP/161.
|
|
|
|
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.
|
|
|
|
This action:
|
|
|
|
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_<zone>`` 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.
|
|
"""
|
|
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)
|
|
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>'}")
|
|
|
|
broken_zone = (
|
|
self._zone_for_network(fw_sections, {key: key for key in nameless}, mgmt_net)
|
|
if mgmt_net
|
|
else ""
|
|
)
|
|
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:"
|
|
)
|
|
lines.append(
|
|
f"[firewall] uci set firewall.{broken_zone}.name='{mgmt_net}' "
|
|
f"&& uci commit firewall && fw4 reload"
|
|
)
|
|
return {"success": 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}")
|
|
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)}
|
|
else:
|
|
lines.append("[firewall] 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)}")
|
|
|
|
# ── 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}.target='ACCEPT'",
|
|
f"uci set firewall.{named_key}.proto='udp'",
|
|
f"uci set firewall.{named_key}.dest_port='161'",
|
|
]
|
|
if mgmt_zone:
|
|
rule_cmds.append(f"uci set firewall.{named_key}.src='{mgmt_zone}'")
|
|
else:
|
|
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'}")
|
|
|
|
# ── 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'}")
|
|
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'}")
|
|
|
|
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}")
|
|
|
|
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
|
|
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}
|