# 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 ipaddress
import re
from typing import Any

from napalm.base.exceptions import (
    ConnectionClosedException,
    ConnectionException,
)
from napalm.base.netmiko_helpers import netmiko_args
from napalm_device_types import (
    AccessPointDriver,
    ByteStream,
    CommandResult,
    EnsurePackageMixin,
    FingerprintRule,
    SnmpAgentMixin,
)
from napalm_device_types.channel import open_stream_on_transport, run_on_transport
from napalm_device_types.listening import ListeningSocketsMixin
from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException

from napalm_openwrt.config_mixin import OpenWrtConfigMixin
from napalm_openwrt.interfaces_mixin import OpenWrtInterfaceMixin
from napalm_openwrt.lldp_mixin import OpenWrtLLDPMixin
from napalm_openwrt.packages_mixin import OpenWrtPackageMixin
from napalm_openwrt.routing_mixin import OpenWrtRoutingMixin
from napalm_openwrt.system_mixin import _RC_MARKER, OpenWrtSystemMixin, _split_status
from napalm_openwrt.vlan_mixin import OpenWrtVLANMixin
from napalm_openwrt.wireless_mixin import OpenWrtWirelessMixin

#: 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", [])
    ]


#: ``reboot`` detached from the session and two seconds late: the answer comes
#: back first, and a hang-up from the closing session does not stop it.
_REBOOT_DETACHED = "sh -c '(trap \"\" HUP; sleep 2; reboot) </dev/null >/dev/null 2>&1 &'"


class OpenWrtDriver(
    OpenWrtInterfaceMixin,
    OpenWrtVLANMixin,
    OpenWrtWirelessMixin,
    OpenWrtLLDPMixin,
    OpenWrtConfigMixin,
    OpenWrtSystemMixin,
    OpenWrtPackageMixin,
    OpenWrtRoutingMixin,
    ListeningSocketsMixin,
    SnmpAgentMixin,
    EnsurePackageMixin,
    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)
        # The host key to expect, and where the one seen is reported (netork#1167).
        self._host_key = HostKeyCheck(optional_args, hostname, self.port)

        # 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:
            with self._host_key.netmiko(self.netmiko_optional_args) as netmiko_kwargs:
                self.device = ConnectHandler(
                    device_type=self.NETMIKO_DEVICE_TYPE,
                    host=self.hostname,
                    username=self.username,
                    password=self.password,
                    timeout=self.timeout,
                    port=self.port,
                    **netmiko_kwargs,
                )
        except SshHostKeyMismatch:
            raise  # refused before the login; nothing was sent
        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
        self._host_key.observe(self.device.remote_conn_pre.get_transport())

    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 (OSError, 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 (OSError, EOFError) as exc:
            raise ConnectionClosedException(str(exc)) from exc

    # ------------------------------------------------------------------
    # Command channel (napalm-device-types CommandChannelMixin, via AccessPointDriver)
    # ------------------------------------------------------------------

    def _transport(self) -> Any:
        """The SSH transport netmiko already holds, for exec channels next to its shell."""
        if self.device is None:
            raise ConnectionClosedException("Not connected")
        return self.device.remote_conn_pre.get_transport()

    def _check_privilege(self, privileged: bool) -> None:
        if privileged and self.username != "root":
            raise PermissionError(
                f"{self.hostname}: OpenWrt has no sudo; {self.username!r} cannot run as root"
            )

    def run_command(
        self,
        command: str,
        *,
        privileged: bool = False,
        timeout: float = 60,
        stdin: bytes | None = None,
    ) -> CommandResult:
        """Run *command* on an SSH exec channel: no PTY, stderr apart, a real exit code.

        The login is root on OpenWrt as a rule, and there is no sudo: a
        privileged command from any other login is refused before it runs.
        """
        self._check_privilege(privileged)
        return run_on_transport(self._transport(), command, stdin=stdin, timeout=timeout)

    def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
        """Start *command* on an SSH exec channel and return a stream to it."""
        self._check_privilege(privileged)
        return open_stream_on_transport(self._transport(), command)

    def reboot_host(self) -> None:
        """Restart the AP (``HostRebootMixin``); returns once the restart is under way.

        Detached and two seconds late, so the answer comes back before the
        session goes down; a refusal is raised, not mistaken for that.

        :raises RuntimeError: for a login other than root, or a refused reboot.
        """
        try:
            result = self.run_command(_REBOOT_DETACHED, privileged=True, timeout=30)
        except PermissionError as exc:
            raise RuntimeError(str(exc)) from exc
        if result.exit_code:
            reason = result.stderr.strip() or result.stdout.strip()
            raise RuntimeError(reason or f"reboot exited with status {result.exit_code}")

    def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
        """Run the listening-socket command for ``ListeningSocketsMixin``.

        Over an SSH exec channel of its own, not the interactive shell: busybox
        ash cuts a typed line at 512 characters, and the command is longer. As
        root it names every socket's process; a login other than root has no way
        to become root here, so a privileged reading comes back empty and the
        mixin reads again without attributing.
        """
        if privileged and self.username != "root":
            return ""
        _stdin, stdout, _stderr = self.device.remote_conn_pre.exec_command(
            command, timeout=self.timeout
        )
        return stdout.read().decode("utf-8", "replace")

    @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 _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

    # ------------------------------------------------------------------
    # SNMP agent (napalm-device-types SnmpAgentMixin): the caller's values
    # ------------------------------------------------------------------

    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

    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:]}

    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] = []
        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"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"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"  uci set firewall.{broken_zone}.name='{mgmt_net}'"
                " && uci commit firewall && fw4 reload"
            )
            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"Management zone: {mgmt_zone!r}")
        elif zones:
            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("No zones configured — writing an unscoped rule")

        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:
            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)}")

        rule_cmds = [
            f"uci set firewall.{named_key}=rule",
            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}'")
        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()
        scope = ", ".join(sources) or "every source in the zone"
        lines.append(f"Wrote rule {named_key!r} for {scope}: {rule_out or 'ok'}")

        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"Reload: {reload_out or 'ok'}")
        if not fw_ok:
            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"udp/161 accept rules live in the packet filter: {live}")
        return {"ok": fw_ok and bool(live), "output": "\n".join(lines)}

    def _start_snmp_agent(self) -> dict[str, Any]:
        self._send_command(
            "/etc/init.d/snmpd enable 2>/dev/null; /etc/init.d/snmpd restart 2>/dev/null"
        )
        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",
        }