fix: mixin classes must precede AccessPointDriver in OpenWrtDriver MRO

NetworkDriver (parent of AccessPointDriver) raises NotImplementedError for all
standard NAPALM methods. With AccessPointDriver listed first, get_interfaces()
and get_vlans() from the mixins were shadowed and always raised NotImplementedError
(empty message) — causing all AP polls to report 0 interfaces and 0 VLANs.

Fix: reorder inheritance so mixins come before AccessPointDriver. The standard
NAPALM method stubs in NetworkDriver are now shadowed by the mixin implementations.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

napalm_openwrt/config_mixin.py

@@ -0,0 +1,101 @@
+# -*- 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.
+
+from __future__ import annotations
+
+from napalm.base.exceptions import (
+    CommandErrorException,
+    MergeConfigException,
+    ReplaceConfigException,
+)
+from napalm_device_types import ConfigLifecycleMixin
+
+
+class OpenWrtConfigMixin(ConfigLifecycleMixin):
+    """Mixin providing configuration-management NAPALM methods for OpenWrt.
+
+    Reuses the standard lifecycle (load/compare/discard) from
+    :class:`ConfigLifecycleMixin` and adds OpenWrt-specific
+    ``commit_config`` and ``rollback``.
+    """
+
+    _comment_chars = ("#",)
+
+    def _get_running_config(self) -> str:
+        return self._send_command("uci export")
+
+    def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
+        """Apply the staged candidate configuration and commit it.
+
+        **Merge mode**: each UCI command line is sent to the device shell, then
+        ``uci commit`` is called to persist the changes.
+
+        **Replace mode**: the candidate is piped through ``uci import`` and
+        then ``uci commit`` is called for every affected package.
+
+        :raises MergeConfigException: if no candidate is staged or if
+            commands are rejected.
+        :raises ReplaceConfigException: same, for replace candidates.
+        """
+        if self._candidate_config is None:
+            raise MergeConfigException("No candidate configuration is staged.")
+
+        ex_cls = ReplaceConfigException if self._candidate_mode == "replace" else MergeConfigException
+
+        self._backup_config = self._send_command("uci export")
+
+        errors: list[str] = []
+        try:
+            if self._candidate_mode == "merge":
+                for line in self._candidate_config.splitlines():
+                    stripped = line.strip()
+                    if not stripped or stripped.startswith("#"):
+                        continue
+                    out = self._send_command(stripped)
+                    if out and ("uci: " in out.lower() or "error" in out.lower()):
+                        errors.append(f"  {stripped!r}: {out}")
+                self._send_command("uci commit")
+            else:
+                escaped = self._candidate_config.replace("'", "'\\''")
+                self._send_command(f"printf '%s' '{escaped}' > /tmp/napalm_candidate.uci")
+                out = self._send_command("uci import < /tmp/napalm_candidate.uci && uci commit")
+                self._send_command("rm -f /tmp/napalm_candidate.uci")
+                if out and "error" in out.lower():
+                    errors.append(out)
+        except Exception as exc:
+            raise ex_cls(str(exc)) from exc
+
+        if errors:
+            raise ex_cls("The following commands were rejected:\n" + "\n".join(errors))
+
+        self._candidate_config = None
+        self._candidate_mode = None
+
+    def rollback(self) -> None:
+        """Restore the UCI configuration to the state before the last :meth:`commit_config`.
+
+        Pipes the saved backup through ``uci import`` and then commits.
+
+        :raises CommandErrorException: if no backup is available.
+        """
+        if self._backup_config is None:
+            raise CommandErrorException(
+                "No backup configuration available – commit_config has not been called in this session."
+            )
+
+        escaped = self._backup_config.replace("'", "'\\''")
+        self._send_command(f"printf '%s' '{escaped}' > /tmp/napalm_rollback.uci")
+        self._send_command("uci import < /tmp/napalm_rollback.uci && uci commit")
+        self._send_command("rm -f /tmp/napalm_rollback.uci")
+
+        self._backup_config = None

napalm_openwrt/interfaces_mixin.py

@@ -0,0 +1,379 @@
+# -*- 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.
+
+"""Interface-related NAPALM getters for OpenWrt devices."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+import netaddr
+from napalm.base import helpers as napalm_helpers
+
+
+class OpenWrtInterfaceMixin:
+    """Mixin providing interface-related NAPALM getters."""
+
+    def _get_interface_list(self) -> list[str]:
+        """Return a sorted list of interface names from ``ip link show``."""
+        output = self._send_command("ip link show")
+        interfaces = []
+        for line in output.splitlines():
+            m = re.match(r"^\d+:\s+(\S+?)[@:]", line)
+            if m:
+                name = m.group(1)
+                if (
+                    name not in self._EXCLUDED_INTERFACES
+                    and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES)
+                ):
+                    interfaces.append(name)
+        return sorted(set(interfaces))
+
+    def get_interfaces(self) -> dict[str, Any]:
+        """Return interface details, excluding loopback and raw radio (phy*) interfaces."""
+        output = self._send_command("ip link show")
+        return self._filter_interfaces(self._parse_ip_link(output))
+
+    @staticmethod
+    def _parse_ip_link(output: str) -> dict[str, Any]:
+        """Parse ``ip link show`` output into NAPALM interface dicts."""
+        interfaces: dict[str, Any] = {}
+        current: str | None = None
+
+        for line in output.splitlines():
+            # New interface block: "2: eth0: <FLAGS> mtu 1500 ..."
+            m = re.match(
+                r"^\d+:\s+(\S+?)[@:].*<([^>]*)>.*\bmtu\s+(\d+).*\bstate\s+(\S+)",
+                line,
+            )
+            if m:
+                name = m.group(1)
+                flags = m.group(2).upper()
+                mtu = int(m.group(3))
+                state = m.group(4).upper()
+
+                is_up = state in ("UP", "UNKNOWN") and "UP" in flags.split(",")
+                is_enabled = "UP" in flags.split(",")
+
+                interfaces[name] = {
+                    "is_up": is_up,
+                    "is_enabled": is_enabled,
+                    "description": "",
+                    "last_flapped": -1.0,
+                    "speed": -1.0,
+                    "mtu": mtu,
+                    "mac_address": "",
+                }
+                current = name
+                continue
+
+            # MAC address line: "    link/ether aa:bb:cc:dd:ee:ff ..."
+            if current and "link/ether" in line:
+                m2 = re.search(r"link/ether\s+(\S+)", line)
+                if m2:
+                    try:
+                        interfaces[current]["mac_address"] = napalm_helpers.mac(m2.group(1))
+                    except Exception:
+                        interfaces[current]["mac_address"] = m2.group(1)
+
+        return interfaces
+
+    def get_interfaces_ip(self) -> dict[str, Any]:
+        """Return all configured IP addresses grouped by interface.
+
+        Uses ``ip addr show``.
+
+        Example output::
+
+            2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 ...
+                inet 192.168.1.1/24 brd 192.168.1.255 scope global eth0
+                inet6 fd00::1/64 scope global
+        """
+        output = self._send_command("ip addr show")
+        interfaces_ip: dict[str, Any] = {}
+        current_iface: str | None = None
+
+        for line in output.splitlines():
+            # Interface line
+            m = re.match(r"^\d+:\s+(\S+?)[@:]", line)
+            if m:
+                current_iface = m.group(1)
+                continue
+
+            if current_iface is None:
+                continue
+
+            # IPv4
+            m = re.match(r"^\s+inet\s+(\S+)", line)
+            if m:
+                cidr = m.group(1)
+                try:
+                    ip_net = netaddr.IPNetwork(cidr)
+                except (netaddr.AddrFormatError, ValueError):
+                    continue
+                if current_iface not in interfaces_ip:
+                    interfaces_ip[current_iface] = {}
+                interfaces_ip[current_iface].setdefault("ipv4", {})[str(ip_net.ip)] = {
+                    "prefix_length": ip_net.prefixlen
+                }
+                continue
+
+            # IPv6
+            m = re.match(r"^\s+inet6\s+(\S+)", line)
+            if m:
+                cidr = m.group(1)
+                try:
+                    ip_net = netaddr.IPNetwork(cidr)
+                except (netaddr.AddrFormatError, ValueError):
+                    continue
+                if current_iface not in interfaces_ip:
+                    interfaces_ip[current_iface] = {}
+                interfaces_ip[current_iface].setdefault("ipv6", {})[str(ip_net.ip)] = {
+                    "prefix_length": ip_net.prefixlen
+                }
+
+        return interfaces_ip
+
+    def get_interfaces_counters(self) -> dict[str, Any]:
+        """Return per-interface packet and byte counters from ``/proc/net/dev``.
+
+        ``/proc/net/dev`` columns (Receive | Transmit)::
+
+            face |bytes packets errs drop fifo frame compressed multicast| \
+                  bytes packets errs drop fifo colls carrier compressed
+        """
+        output = self._send_command("cat /proc/net/dev")
+        counters: dict[str, Any] = {}
+
+        for line in output.splitlines():
+            # Skip header lines
+            if "|" in line or "Inter" in line:
+                continue
+            line_s = line.strip()
+            if not line_s:
+                continue
+
+            parts = line_s.replace(":", " ").split()
+            if len(parts) < 17:
+                continue
+
+            iface = parts[0]
+            try:
+                counters[iface] = {
+                    "tx_errors": int(parts[10]),
+                    "rx_errors": int(parts[3]),
+                    "tx_discards": int(parts[11]),
+                    "rx_discards": int(parts[4]),
+                    "tx_octets": int(parts[9]),
+                    "rx_octets": int(parts[1]),
+                    "tx_unicast_packets": int(parts[10 - 1]),  # packets field
+                    "rx_unicast_packets": int(parts[2]),
+                    "tx_multicast_packets": 0,
+                    "rx_multicast_packets": int(parts[8]),
+                    "tx_broadcast_packets": 0,
+                    "rx_broadcast_packets": 0,
+                }
+            except (IndexError, ValueError):
+                continue
+
+        return counters
+
+    def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
+        """Return the ARP/neighbour table.
+
+        Uses ``ip neigh show`` (preferred) which produces::
+
+            192.168.1.100 dev br-lan lladdr aa:bb:cc:dd:ee:ff REACHABLE
+            192.168.1.1 dev br-lan lladdr 00:11:22:33:44:55 STALE
+        """
+        output = self._send_command(["ip neigh show", "cat /proc/net/arp"])
+        arp_table = []
+
+        # ip neigh show format
+        for line in output.splitlines():
+            line_s = line.strip()
+            if not line_s:
+                continue
+
+            # Skip incomplete/failed entries
+            if "FAILED" in line_s or "INCOMPLETE" in line_s:
+                continue
+
+            # ip neigh show: "192.168.1.100 dev br-lan lladdr aa:bb:cc:dd:ee:ff REACHABLE"
+            m = re.match(
+                r"^(\S+)\s+dev\s+(\S+)\s+lladdr\s+(\S+)",
+                line_s,
+                re.I,
+            )
+            if m:
+                ip_addr = m.group(1)
+                interface = m.group(2)
+                mac_raw = m.group(3)
+
+                try:
+                    netaddr.IPAddress(ip_addr)
+                except (netaddr.AddrFormatError, ValueError):
+                    continue
+
+                try:
+                    mac_addr = napalm_helpers.mac(mac_raw)
+                except Exception:
+                    mac_addr = mac_raw
+
+                arp_table.append(
+                    {
+                        "interface": interface,
+                        "mac": mac_addr,
+                        "ip": ip_addr,
+                        "age": 0.0,
+                    }
+                )
+                continue
+
+            # /proc/net/arp fallback: "IP address HW type Flags HW address Mask Device"
+            # skip header
+            if line_s.startswith("IP address"):
+                continue
+            parts = line_s.split()
+            if len(parts) >= 6:
+                ip_addr = parts[0]
+                mac_raw = parts[3]
+                interface = parts[5]
+
+                try:
+                    netaddr.IPAddress(ip_addr)
+                except (netaddr.AddrFormatError, ValueError):
+                    continue
+
+                if mac_raw in ("00:00:00:00:00:00", ""):
+                    continue
+
+                try:
+                    mac_addr = napalm_helpers.mac(mac_raw)
+                except Exception:
+                    mac_addr = mac_raw
+
+                arp_table.append(
+                    {
+                        "interface": interface,
+                        "mac": mac_addr,
+                        "ip": ip_addr,
+                        "age": 0.0,
+                    }
+                )
+
+        return arp_table
+
+    def get_mac_address_table(self) -> list[dict[str, Any]]:
+        """Return the bridge forwarding database (MAC address table).
+
+        Uses ``bridge fdb show`` which produces::
+
+            aa:bb:cc:dd:ee:ff dev br-lan master br-lan permanent
+            11:22:33:44:55:66 dev eth0.1 vlan 1 master br-lan
+        """
+        output = self._send_command("bridge fdb show")
+        mac_table = []
+
+        for line in output.splitlines():
+            line_s = line.strip()
+            if not line_s:
+                continue
+
+            m = re.match(r"^(\S+)\s+dev\s+(\S+)", line_s)
+            if not m:
+                continue
+
+            mac_raw = m.group(1)
+            interface = m.group(2)
+
+            # Skip broadcast/multicast self-entries that are always present
+            if mac_raw.lower() in ("ff:ff:ff:ff:ff:ff", "33:33:00:00:00:01"):
+                continue
+
+            static = "permanent" in line_s or "static" in line_s
+
+            # Extract VLAN if present: "vlan 10"
+            vlan = 0
+            vlan_m = re.search(r"\bvlan\s+(\d+)", line_s)
+            if vlan_m:
+                vlan = int(vlan_m.group(1))
+
+            try:
+                mac_addr = napalm_helpers.mac(mac_raw)
+            except Exception:
+                mac_addr = mac_raw
+
+            mac_table.append(
+                {
+                    "mac": mac_addr,
+                    "interface": interface,
+                    "vlan": vlan,
+                    "static": static,
+                    "active": True,
+                    "moves": None,
+                    "last_move": None,
+                }
+            )
+
+        return mac_table
+
+    def get_ipv6_neighbors_table(self) -> list[dict[str, Any]]:
+        """Return the IPv6 neighbour table from ``ip -6 neigh show``.
+
+        Example output::
+
+            2001:db8::1 dev eth0 lladdr aa:bb:cc:dd:ee:ff REACHABLE
+            fe80::1 dev br-lan lladdr 11:22:33:44:55:66 STALE
+        """
+        output = self._send_command("ip -6 neigh show")
+        table = []
+
+        for line in output.splitlines():
+            line_s = line.strip()
+            if not line_s or "FAILED" in line_s or "INCOMPLETE" in line_s:
+                continue
+
+            m = re.match(
+                r"^(\S+)\s+dev\s+(\S+)\s+lladdr\s+(\S+)\s+(\S+)",
+                line_s,
+                re.I,
+            )
+            if not m:
+                continue
+
+            ip_addr = m.group(1)
+            interface = m.group(2)
+            mac_raw = m.group(3)
+            state = m.group(4)
+
+            try:
+                netaddr.IPAddress(ip_addr, version=6)
+            except (netaddr.AddrFormatError, ValueError):
+                continue
+
+            try:
+                mac_addr = napalm_helpers.mac(mac_raw)
+            except Exception:
+                mac_addr = mac_raw
+
+            table.append({
+                "interface": interface,
+                "mac": mac_addr,
+                "ip": ip_addr,
+                "age": -1.0,
+                "state": state,
+            })
+
+        return table

napalm_openwrt/lldp_mixin.py

@@ -0,0 +1,131 @@
+# -*- 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.
+
+from __future__ import annotations
+
+from typing import Any
+
+
+class OpenWrtLLDPMixin:
+    """Mixin providing LLDP NAPALM getters."""
+
+    def get_lldp_neighbors(self) -> dict[str, list[dict[str, Any]]]:
+        """Return LLDP neighbors (requires ``lldpd`` to be installed on the device).
+
+        Uses ``lldpctl -f keyvalue`` output::
+
+            lldp.eth0.port.ifname=eth1
+            lldp.eth0.chassis.name=router-core
+        """
+        neighbors: dict[str, list[dict[str, Any]]] = {}
+        for row in self._get_lldp_table():
+            neighbors.setdefault(row["local_port"], []).append(
+                {"hostname": row["system_name"], "port": row["port_id"]}
+            )
+        return neighbors
+
+    def _lldpd_fix_interface(self) -> None:
+        """Ensure lldpd UCI config uses the actual management interface.
+
+        Detects the interface carrying the default route (e.g. br-lan.10 on a
+        VLAN trunk) and replaces any stale/wrong interface entry in the lldpd
+        UCI config.  Idempotent — only writes if the config needs updating.
+        """
+        mgmt_iface = self._send_command(
+            "ip route show default 2>/dev/null | head -1 | grep -oE 'dev [^ ]+' | awk '{print $2}'"
+        ).strip()
+        if not mgmt_iface:
+            mgmt_iface = "br-lan"
+        # LLDP is a L2 protocol — it must run on the bridge, not a VLAN subinterface.
+        # Strip .VID suffix so br-ap.10 → br-ap.
+        if "." in mgmt_iface:
+            mgmt_iface = mgmt_iface.rsplit(".", 1)[0]
+
+        current = self._send_command("uci get lldpd.config.interface 2>/dev/null").strip()
+        if mgmt_iface in current:
+            return  # already correct
+
+        # Reset the interface list to just the management interface
+        self._send_command(
+            "uci delete lldpd.config.interface 2>/dev/null; "
+            f"uci add_list lldpd.config.interface='{mgmt_iface}' 2>/dev/null; "
+            "uci commit lldpd 2>/dev/null; "
+            "/etc/init.d/lldpd restart 2>/dev/null"
+        )
+
+    def _get_lldp_table(self) -> list[dict[str, Any]]:
+        """Parse ``lldpctl -f keyvalue`` into a list of row dicts.
+
+        Ensures ``lldpd`` is enabled and running before querying; if it was
+        not already running the daemon needs time to discover neighbors so
+        the first call after a fresh install will return an empty list.
+        """
+        # Ensure lldpd is running on the correct management interface
+        self._send_command(
+            "pgrep lldpd >/dev/null 2>&1 || "
+            "(/etc/init.d/lldpd enable 2>/dev/null; /etc/init.d/lldpd start 2>/dev/null)"
+        )
+        self._lldpd_fix_interface()
+        output = self._send_command("lldpctl -f keyvalue")
+        rows: list[dict[str, Any]] = []
+
+        # Group by local interface prefix: lldp.<iface>.*
+        entries: dict[str, dict[str, str]] = {}
+        for line in output.splitlines():
+            line_s = line.strip()
+            if "=" not in line_s:
+                continue
+            key, _, value = line_s.partition("=")
+            parts = key.split(".")
+            # parts: ['lldp', '<iface>', <category>, <field>, ...]
+            if len(parts) < 3 or parts[0] != "lldp":
+                continue
+            iface = parts[1]
+            subkey = ".".join(parts[2:])
+            entries.setdefault(iface, {})[subkey] = value
+
+        for iface, data in entries.items():
+            rows.append(
+                {
+                    "local_port": iface,
+                    "remote_chassis_id": data.get("chassis.mac", data.get("chassis.id.value", "")),
+                    "port_id": data.get("port.ifname", data.get("port.id.value", "")),
+                    "mgmt_address": data.get("chassis.mgmt-ip", ""),
+                    "port_description": data.get("port.descr", ""),
+                    "system_name": data.get("chassis.name", ""),
+                }
+            )
+
+        return rows
+
+    def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, list[dict[str, Any]]]:
+        """Return detailed LLDP neighbor info."""
+        details: dict[str, list[dict[str, Any]]] = {}
+
+        for row in self._get_lldp_table():
+            if interface and row["local_port"] != interface:
+                continue
+            details.setdefault(row["local_port"], []).append(
+                {
+                    "parent_interface": "",
+                    "remote_port": row["port_id"],
+                    "remote_port_description": row["port_description"],
+                    "remote_chassis_id": row["remote_chassis_id"],
+                    "remote_system_name": row["system_name"],
+                    "remote_system_description": "",
+                    "remote_system_capab": [],
+                    "remote_system_enable_capab": [],
+                }
+            )
+
+        return details

napalm_openwrt/packages_mixin.py

@@ -0,0 +1,220 @@
+# -*- 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.
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+
+class OpenWrtPackageMixin:
+    """Mixin providing package-management NAPALM methods."""
+
+    # Package management  (opkg <= OpenWrt 23 / apk >= OpenWrt 24)
+    # ------------------------------------------------------------------
+
+    def get_packages(self) -> list[dict[str, Any]]:
+        """Return installed packages from the device's package manager.
+
+        Automatically detects whether to use ``apk`` (OpenWrt 24+, Alpine
+        APK) or ``opkg`` (older OpenWrt releases).  Returns one entry per
+        installed package.
+        """
+        pm = self._send_command("command -v apk 2>/dev/null || echo __no_apk__").strip()
+        if "__no_apk__" not in pm and pm:
+            return self._get_packages_apk()
+        return self._get_packages_opkg()
+
+    def _get_packages_opkg(self) -> list[dict[str, Any]]:
+        """Parse ``opkg status`` (dpkg-style stanzas)."""
+        out = self._send_command("opkg status")
+        packages: list[dict[str, Any]] = []
+        stanza: dict[str, str] = {}
+        for raw in out.splitlines():
+            line = raw.rstrip()
+            if line == "":
+                if stanza.get("Package"):
+                    packages.append(self._opkg_stanza_to_dict(stanza))
+                stanza = {}
+            elif line[:1] in (" ", "\t"):
+                # Continuation of previous field (e.g. multi-line Description)
+                last_key = list(stanza)[-1] if stanza else None
+                if last_key:
+                    stanza[last_key] += " " + line.strip()
+            elif ":" in line:
+                key, _, val = line.partition(":")
+                stanza[key.strip()] = val.strip()
+        if stanza.get("Package"):
+            packages.append(self._opkg_stanza_to_dict(stanza))
+        return sorted(packages, key=lambda p: p["name"].lower())
+
+    @staticmethod
+    def _opkg_stanza_to_dict(stanza: dict[str, str]) -> dict[str, Any]:
+        status = stanza.get("Status", "")
+        try:
+            size = int(stanza.get("Installed-Size", 0) or 0)
+        except ValueError:
+            size = 0
+        return {
+            "name": stanza["Package"],
+            "version": stanza.get("Version", ""),
+            "installed": "installed" in status.lower(),
+            "description": stanza.get("Description", ""),
+            "size": size,
+            "source": stanza.get("Section", ""),
+        }
+
+    def _get_packages_apk(self) -> list[dict[str, Any]]:
+        """Parse ``apk list --installed`` output.
+
+        Line format::
+
+            busybox-1.37.0-r0 x86_64 {busybox} (GPL-2.0-only) [installed]
+            kmod-nft-bridge-6.6.75-r0 mips_24kc {kmod-nft-bridge} (GPL-2.0-only) [installed]
+        """
+        out = self._send_command("apk list --installed 2>/dev/null")
+        packages: list[dict[str, Any]] = []
+        for line in out.splitlines():
+            line = line.strip()
+            if not line or "[installed]" not in line:
+                continue
+            # Split name from version: version always starts with a digit after '-'
+            m = re.match(r"^(.*?)-(\d\S*)\s+\S+\s+\{(\S+)\}", line)
+            if m:
+                name, version, origin = m.group(1), m.group(2), m.group(3)
+            else:
+                # Minimal fallback: first token only
+                token = line.split()[0]
+                vm = re.search(r"-(\d\S*)$", token)
+                name = token[: vm.start()] if vm else token
+                version = vm.group(1) if vm else ""
+                origin = ""
+            packages.append({
+                "name": name,
+                "version": version,
+                "installed": True,
+                "description": "",
+                "size": 0,
+                "source": origin,
+            })
+        return sorted(packages, key=lambda p: p["name"].lower())
+
+    def _pm_type(self) -> str:
+        """Return ``'apk'`` if device has apk (OpenWrt 24+), otherwise ``'opkg'``."""
+        out = self._send_command("command -v apk 2>/dev/null || echo __no_apk__").strip()
+        return "apk" if ("__no_apk__" not in out and out) else "opkg"
+
+    def search_packages(self, query: str) -> list[dict[str, Any]]:
+        """Search available packages matching *query* (name or description).
+
+        Runs ``opkg update`` / ``apk update`` first to ensure the package
+        index is populated (OpenWrt stores it in RAM and loses it on reboot).
+        """
+        import shlex
+        safe_q = shlex.quote(query)
+        if self._pm_type() == "apk":
+            # Refresh index (no-ops if already current, safe to run every time)
+            self._send_command("apk update 2>/dev/null || true")
+            out = self._send_command(f"apk search {safe_q} 2>/dev/null")
+            installed = {p["name"] for p in self._get_packages_apk()}
+            packages: list[dict[str, Any]] = []
+            for line in out.splitlines():
+                line = line.strip()
+                if not line:
+                    continue
+                m = re.match(r"^(.*?)-(\d\S*)(?:\s+(.*))?$", line)
+                if m:
+                    name, version, description = m.group(1), m.group(2), (m.group(3) or "")
+                else:
+                    name, version, description = line, "", ""
+                packages.append({
+                    "name": name,
+                    "version": version,
+                    "installed": name in installed,
+                    "description": description,
+                    "size": 0,
+                    "source": "",
+                })
+        else:
+            # opkg lists live in /var/opkg-lists/ (RAM) — cleared on reboot
+            self._send_command("opkg update 2>/dev/null || true")
+            out = self._send_command(f"opkg list 2>/dev/null | grep -i {safe_q}")
+            installed = {p["name"] for p in self._get_packages_opkg()}
+            packages = []
+            for line in out.splitlines():
+                line = line.strip()
+                if not line:
+                    continue
+                parts = line.split(" - ", 2)
+                name = parts[0].strip()
+                version = parts[1].strip() if len(parts) > 1 else ""
+                description = parts[2].strip() if len(parts) > 2 else ""
+                packages.append({
+                    "name": name,
+                    "version": version,
+                    "installed": name in installed,
+                    "description": description,
+                    "size": 0,
+                    "source": "",
+                })
+        return packages
+
+    @staticmethod
+    def _clean_pkg_output(raw: str) -> str:
+        """Strip ANSI/VT100 escape sequences and progress-bar lines."""
+        # Strip CSI sequences (\x1b[...X), OSC, charset designations, and
+        # 2-byte DEC private sequences like ESC 7 (cursor save) / ESC 8 (restore)
+        cleaned = re.sub(
+            r'\x1b(?:\[[0-9;?]*[a-zA-Z]|\][^\x07]*\x07|[()][0-9A-Za-z]|[\x30-\x7e])',
+            '', raw,
+        )
+        # After stripping cursor-save/restore sequences, apk progress updates
+        # end up concatenated on a single line.  Strip those inline patterns.
+        cleaned = re.sub(r'\s*\d{1,3}%\s*#*', ' ', cleaned)
+        lines = []
+        for segment in cleaned.split('\n'):
+            # \r overwrites the line; keep only the portion after the last \r
+            part = segment.split('\r')[-1].strip()
+            if not part:
+                continue
+            # Drop pure progress-bar lines (only #, spaces, digits, %)
+            if re.match(r'^[#\s\d%]*$', part):
+                continue
+            lines.append(part)
+        return '\n'.join(lines)
+
+    def install_package(self, name: str) -> dict[str, Any]:
+        """Install a package by name. Returns ``{"success": bool, "output": str}``."""
+        import shlex
+        safe_name = shlex.quote(name)
+        if self._pm_type() == "apk":
+            raw = self._send_command(f"apk add {safe_name} 2>&1")
+        else:
+            raw = self._send_command(f"opkg install {safe_name} 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 uninstall_package(self, name: str) -> dict[str, Any]:
+        """Remove a package by name. Returns ``{"success": bool, "output": str}``."""
+        import shlex
+        safe_name = shlex.quote(name)
+        if self._pm_type() == "apk":
+            raw = self._send_command(f"apk del {safe_name} 2>&1")
+        else:
+            raw = self._send_command(f"opkg remove {safe_name} 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}

napalm_openwrt/routing_mixin.py

@@ -0,0 +1,309 @@
+# -*- 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.
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+import netaddr
+
+
+class OpenWrtRoutingMixin:
+    """Mixin providing routing-related NAPALM methods (ping, traceroute, route)."""
+
+    def ping(
+        self,
+        destination: str,
+        source: str = "",
+        ttl: int = 255,
+        timeout: int = 2,
+        size: int = 56,
+        count: int = 5,
+        vrf: str = "",
+        source_interface: str = "",
+    ) -> dict[str, Any]:
+        """Execute ping on the device and return statistics.
+
+        Builds a ``ping`` command with standard BusyBox/iputils flags::
+
+            ping -c <count> -W <timeout> -s <size> [-t <ttl>] [-I <iface>] <dest>
+
+        Returns ``{'success': {...}}`` or ``{'error': '<message>'}``.
+        """
+        cmd_parts = ["ping", "-c", str(count), "-W", str(timeout), "-s", str(size)]
+        if ttl != 255:
+            cmd_parts += ["-t", str(ttl)]
+        if source_interface:
+            cmd_parts += ["-I", source_interface]
+        elif source:
+            cmd_parts += ["-I", source]
+        cmd_parts.append(destination)
+
+        output = self._send_command(" ".join(cmd_parts))
+
+        # Check for hard failure before parsing
+        if re.search(r"unknown host|bad address|Network unreachable|not reachable", output, re.I):
+            m = re.search(r"(unknown host.*|bad address.*|Network unreachable)", output, re.I)
+            return {"error": m.group(0) if m else output.strip()}
+
+        return self._parse_ping_output(output, destination)
+
+    @staticmethod
+    def _parse_ping_output(output: str, destination: str) -> dict[str, Any]:
+        """Parse BusyBox/iputils ping output into NAPALM format."""
+        # "2 packets transmitted, 2 packets received, 0% packet loss"
+        summary_m = re.search(
+            r"(\d+)\s+packets?\s+transmitted.*?(\d+)\s+(?:packets?\s+)?received.*?(\d+)%\s+packet\s+loss",
+            output,
+            re.S | re.I,
+        )
+        if not summary_m:
+            return {"error": output.strip() or f"No response from {destination}"}
+
+        sent = int(summary_m.group(1))
+        received = int(summary_m.group(2))
+        loss = sent - received
+
+        # "round-trip min/avg/max = 6.987/7.055/7.123 ms"  (BusyBox)
+        # "rtt min/avg/max/mdev = 6.987/7.055/7.123/0.094 ms" (iputils)
+        rtt_m = re.search(
+            r"(?:round-trip|rtt)\s+min/avg/max(?:/(?:mdev|stddev))?\s*=\s*([\d.]+)/([\d.]+)/([\d.]+)(?:/([\d.]+))?",
+            output,
+            re.I,
+        )
+        rtt_min = rtt_avg = rtt_max = rtt_stddev = 0.0
+        if rtt_m:
+            rtt_min = float(rtt_m.group(1))
+            rtt_avg = float(rtt_m.group(2))
+            rtt_max = float(rtt_m.group(3))
+            rtt_stddev = float(rtt_m.group(4)) if rtt_m.group(4) else 0.0
+
+        # Individual probe results
+        results = []
+        for m in re.finditer(
+            r"(\d+)\s+bytes\s+from\s+(\S+?):\s+(?:icmp_seq|seq)=\d+\s+.*?time=([\d.]+)\s*ms",
+            output,
+            re.I,
+        ):
+            ip = m.group(2).rstrip(":")
+            results.append({"ip_address": ip, "rtt": float(m.group(3))})
+
+        return {
+            "success": {
+                "probes_sent": sent,
+                "packet_loss": loss,
+                "rtt_min": rtt_min,
+                "rtt_avg": rtt_avg,
+                "rtt_max": rtt_max,
+                "rtt_stddev": rtt_stddev,
+                "results": results,
+            }
+        }
+
+    def get_route_to(
+        self, destination: str = "", protocol: str = "", longer: bool = False
+    ) -> dict[str, list[dict[str, Any]]]:
+        """Return routes to *destination* from the kernel routing table.
+
+        Uses ``ip route show`` (optionally filtered by prefix/match) and
+        ``ip route get <dest>`` for the best-path lookup.
+
+        Protocol filter is applied post-parse (kernel proto names:
+        ``kernel``, ``static``, ``dhcp``, ``bird``, ``zebra``, …).
+
+        Example ``ip route show`` output::
+
+            default via 192.168.1.1 dev br-wan proto dhcp src 203.0.113.1 metric 100
+            192.168.1.0/24 dev br-lan proto kernel scope link src 192.168.1.1
+        """
+        if destination:
+            cmd = f"ip route show {'match ' if longer else ''}{destination}"
+        else:
+            cmd = "ip route show"
+
+        output = self._send_command(cmd)
+        routes: dict[str, list[dict[str, Any]]] = {}
+
+        for line in output.splitlines():
+            line_s = line.strip()
+            if not line_s:
+                continue
+
+            # Determine the prefix
+            # "default via ..." → prefix = "0.0.0.0/0"
+            # "192.168.1.0/24 dev ..." → prefix as-is
+            if line_s.startswith("default"):
+                prefix = "0.0.0.0/0"
+                rest = line_s[len("default"):].strip()
+            else:
+                parts = line_s.split()
+                prefix = parts[0]
+                rest = " ".join(parts[1:])
+
+            # Extract fields
+            next_hop = ""
+            outgoing_iface = ""
+            proto_raw = "kernel"
+            metric = 0
+
+            m = re.search(r"\bvia\s+(\S+)", rest)
+            if m:
+                next_hop = m.group(1)
+
+            m = re.search(r"\bdev\s+(\S+)", rest)
+            if m:
+                outgoing_iface = m.group(1)
+
+            m = re.search(r"\bproto\s+(\S+)", rest)
+            if m:
+                proto_raw = m.group(1)
+
+            m = re.search(r"\bmetric\s+(\d+)", rest)
+            if m:
+                metric = int(m.group(1))
+
+            # Map proto to NAPALM-style name
+            proto_map = {
+                "kernel": "connected",
+                "static": "static",
+                "dhcp": "static",
+                "bird": "bgp",
+                "zebra": "ospf",
+            }
+            napalm_proto = proto_map.get(proto_raw.lower(), proto_raw)
+
+            if protocol and napalm_proto.lower() != protocol.lower():
+                continue
+
+            entry = {
+                "protocol": napalm_proto,
+                "current_active": True,
+                "last_active": True,
+                "age": 0,
+                "next_hop": next_hop,
+                "outgoing_interface": outgoing_iface,
+                "selected_next_hop": True,
+                "preference": metric,
+                "inactive_reason": "",
+                "routing_table": "default",
+                "protocol_attributes": {},
+            }
+            routes.setdefault(prefix, []).append(entry)
+
+        return routes
+
+    def traceroute(
+        self,
+        destination: str,
+        source: str = "",
+        ttl: int = 30,
+        timeout: int = 3,
+        vrf: str = "",
+    ) -> dict[str, Any]:
+        """Execute traceroute on the device.
+
+        Uses ``traceroute -m <ttl> -w <timeout> <dest>`` (BusyBox-compatible).
+        Falls back to ``traceroute6`` for IPv6 destinations.
+
+        Returns ``{'success': {hop: {'probes': {probe: {rtt, ip_address, host_name}}}}}``
+        or ``{'error': '<message>'}``.
+        """
+        # Detect IPv6 destination
+        try:
+            is_ipv6 = netaddr.IPAddress(destination).version == 6
+        except (netaddr.AddrFormatError, ValueError):
+            is_ipv6 = ":" in destination
+
+        cmd_base = "traceroute6" if is_ipv6 else "traceroute"
+        cmd_parts = [cmd_base, "-m", str(ttl), "-w", str(timeout)]
+        if source:
+            cmd_parts += ["-s", source]
+        cmd_parts.append(destination)
+
+        output = self._send_command(" ".join(cmd_parts))
+
+        if re.search(r"unknown host|bad address|not reachable|cannot resolve", output, re.I):
+            m = re.search(r"(unknown host.*|bad address.*|cannot resolve.*)", output, re.I)
+            return {"error": m.group(0) if m else output.strip()}
+
+        return self._parse_traceroute_output(output)
+
+    @staticmethod
+    def _parse_traceroute_output(output: str) -> dict[str, Any]:
+        """Parse BusyBox traceroute output into NAPALM format.
+
+        Example lines::
+
+            1  192.168.1.1 (192.168.1.1)  1.123 ms  1.456 ms  1.789 ms
+            2  * * *
+        """
+        hops: dict[int, dict[str, Any]] = {}
+
+        for line in output.splitlines():
+            line_s = line.strip()
+            # Hop line starts with an integer
+            m = re.match(r"^(\d+)\s+(.*)", line_s)
+            if not m:
+                continue
+
+            hop_id = int(m.group(1))
+            rest = m.group(2).strip()
+
+            # All-star line: no response
+            if re.match(r"^\*[\s*]*$", rest):
+                hops[hop_id] = {
+                    "probes": {
+                        1: {"rtt": -1.0, "ip_address": "*", "host_name": "*"},
+                        2: {"rtt": -1.0, "ip_address": "*", "host_name": "*"},
+                        3: {"rtt": -1.0, "ip_address": "*", "host_name": "*"},
+                    }
+                }
+                continue
+
+            # Extract host/IP and RTT values
+            # Format: "hostname (ip)  1.1 ms  2.2 ms  3.3 ms"
+            # or:     "ip  1.1 ms  2.2 ms  3.3 ms"
+            host_m = re.match(r"^(\S+)\s+\((\S+)\)", rest)
+            if host_m:
+                host_name = host_m.group(1)
+                ip_address = host_m.group(2)
+            else:
+                # IP only
+                ip_m = re.match(r"^(\d[\d.]+|[0-9a-f:]+)", rest)
+                if ip_m:
+                    ip_address = ip_m.group(1)
+                    host_name = ip_address
+                else:
+                    continue
+
+            rtt_values = [float(x) for x in re.findall(r"([\d.]+)\s+ms", rest)]
+
+            probes: dict[int, dict[str, Any]] = {}
+            for i, rtt in enumerate(rtt_values[:3], start=1):
+                probes[i] = {
+                    "rtt": rtt,
+                    "ip_address": ip_address,
+                    "host_name": host_name,
+                }
+            # Fill missing probes with star entries
+            for i in range(len(rtt_values) + 1, 4):
+                probes[i] = {"rtt": -1.0, "ip_address": "*", "host_name": "*"}
+
+            if probes:
+                hops[hop_id] = {"probes": probes}
+
+        if not hops:
+            return {"error": output.strip() or "No traceroute output received"}
+
+        return {"success": hops}

napalm_openwrt/system_mixin.py

@@ -0,0 +1,732 @@
+# -*- 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.
+
+from __future__ import annotations
+
+import re
+import time as _time
+from typing import Any
+
+
+class OpenWrtSystemMixin:
+    """Mixin providing system-level NAPALM getters (environment, NTP, SNMP, users, services, updates, device actions)."""
+
+    def get_environment(self) -> dict[str, Any]:
+        """Return device environment data (CPU, memory).
+
+        CPU usage from ``/proc/stat`` (two samples 1 second apart via ``awk``).
+        Memory from ``/proc/meminfo``.
+        """
+        cpu_out = self._send_command(
+            "awk '/^cpu /{idle1=$5; total1=$2+$3+$4+$5+$6+$7+$8} END{print (1-(idle1/total1))*100}' /proc/stat"
+        )
+        mem_out = self._send_command("cat /proc/meminfo")
+
+        cpu_pct = 0.0
+        try:
+            cpu_pct = float(cpu_out.strip())
+        except (ValueError, AttributeError):
+            pass
+
+        mem_total = 0
+        mem_available = 0
+        for line in mem_out.splitlines():
+            if line.startswith("MemTotal:"):
+                try:
+                    mem_total = int(line.split()[1])
+                except (IndexError, ValueError):
+                    pass
+            elif line.startswith("MemAvailable:"):
+                try:
+                    mem_available = int(line.split()[1])
+                except (IndexError, ValueError):
+                    pass
+
+        return {
+            "fans": {},
+            "temperature": {},
+            "power": {},
+            "cpu": {0: {"%usage": round(cpu_pct, 1)}},
+            "memory": {
+                "available_ram": mem_available * 1024,
+                "used_ram": (mem_total - mem_available) * 1024,
+            },
+        }
+
+    def get_system_config(self) -> dict[str, Any]:
+        """Return system-level configuration from UCI.
+
+        Reads ``uci show system`` and ``uci show dropbear`` to collect:
+
+        * hostname (str)
+        * timezone (str) — POSIX TZ string, e.g. ``"CET-1CEST,M3.5.0,M10.5.0/3"``
+        * zonename (str) — human-readable name, e.g. ``"Europe/Berlin"``
+        * ntp_servers (list[str])
+        * dropbear_port (int) — SSH port
+        * dropbear_password_auth (bool) — whether password login is allowed
+        * dropbear_root_password_auth (bool)
+        """
+        sys_out = self._send_command("uci show system 2>/dev/null || true")
+        db_out = self._send_command("uci show dropbear 2>/dev/null || true")
+
+        sys_cfg: dict[str, str] = {}
+        for line in sys_out.splitlines():
+            m = re.match(r"system\.@system\[0\]\.(\w+)='([^']*)'", line.strip())
+            if m:
+                sys_cfg[m.group(1)] = m.group(2)
+
+        # NTP server list: all servers on a single line, space-separated quoted values
+        # e.g. system.ntp.server='0.openwrt.pool.ntp.org' '1.openwrt.pool.ntp.org' ...
+        ntp_servers: list[str] = []
+        for line in sys_out.splitlines():
+            if re.match(r"system\.ntp\.server=", line.strip()):
+                ntp_servers = re.findall(r"'([^']+)'", line)
+                break
+
+        # Dropbear settings
+        db_cfg: dict[str, str] = {}
+        for line in db_out.splitlines():
+            # May be @dropbear[0] or named section
+            m = re.match(r"dropbear\.[@\w]+\.(\w+)='([^']*)'", line.strip())
+            if m:
+                db_cfg.setdefault(m.group(1), m.group(2))
+
+        try:
+            ssh_port = int(db_cfg.get("Port", "22"))
+        except (ValueError, TypeError):
+            ssh_port = 22
+
+        def _bool_uci(val: str, default: bool = True) -> bool:
+            return val.lower() not in ("0", "off", "false", "no") if val else default
+
+        return {
+            "hostname": sys_cfg.get("hostname", ""),
+            "timezone": sys_cfg.get("timezone", ""),
+            "zonename": sys_cfg.get("zonename", ""),
+            "ntp_servers": ntp_servers,
+            "dropbear_port": ssh_port,
+            "dropbear_password_auth": _bool_uci(db_cfg.get("PasswordAuth", "on")),
+            "dropbear_root_password_auth": _bool_uci(db_cfg.get("RootPasswordAuth", "on")),
+        }
+
+    def get_snmp_information(self) -> dict[str, Any]:
+        """Return SNMP configuration from ``uci show snmpd``.
+
+        UCI example::
+
+            snmpd.@com2sec[0].community='public'
+            snmpd.@com2sec[0].secname='public'
+            snmpd.@system[0].sysContact='root@localhost'
+            snmpd.@system[0].sysLocation='Unknown'
+        """
+        uci_out = self._send_command("uci show snmpd")
+
+        contact = ""
+        location = ""
+        chassis_id = ""
+        community: dict[str, Any] = {}
+
+        # Track com2sec entries by index
+        com2sec: dict[str, dict[str, str]] = {}
+
+        for line in uci_out.splitlines():
+            line_s = line.strip()
+            m = re.match(r"snmpd\.@com2sec\[(\d+)\]\.(\w+)='([^']*)'", line_s)
+            if m:
+                idx, key, val = m.group(1), m.group(2), m.group(3)
+                com2sec.setdefault(idx, {})[key] = val
+                continue
+            m = re.match(r"snmpd\.@system\[0\]\.sys(\w+)='([^']*)'", line_s)
+            if m:
+                key, val = m.group(1).lower(), m.group(2)
+                if key == "contact":
+                    contact = val
+                elif key == "location":
+                    location = val
+                elif key == "name":
+                    chassis_id = val
+
+        for entry in com2sec.values():
+            name = entry.get("community", entry.get("secname", ""))
+            if not name:
+                continue
+            # OpenWrt snmpd doesn't distinguish rw/ro per community via UCI by default
+            mode = "ro"
+            if entry.get("secname", "").lower() in ("private", "readwrite", "rw"):
+                mode = "rw"
+            community[name] = {
+                "acl": entry.get("source", "N/A"),
+                "mode": mode,
+            }
+
+        return {
+            "chassis_id": chassis_id,
+            "community": community,
+            "contact": contact,
+            "location": location,
+        }
+
+    def get_snmp_config(self):
+        """Return SNMP agent config if snmpd is installed and running on OpenWrt."""
+        try:
+            from napalm_device_types.models import SNMPConfigDict
+        except ImportError:
+            return None
+
+        running = (
+            self._send_command(
+                "/etc/init.d/snmpd running 2>/dev/null && echo active || echo inactive"
+            ).strip() == "active"
+        )
+        if not running:
+            return None
+
+        community = "public"
+        try:
+            # UCI config (set by luci-app-snmpd)
+            uci_comm = self._send_command(
+                "uci -q get snmpd.public.community 2>/dev/null || "
+                "uci -q get snmpd.@com2sec[0].community 2>/dev/null || echo ''"
+            ).strip()
+            if uci_comm:
+                community = uci_comm
+        except Exception:
+            pass
+
+        return SNMPConfigDict(running=True, community=community, port=161, version="2c")
+
+    def _action_fix_snmp(self) -> dict[str, Any]:
+        """Install and configure snmpd on OpenWrt.
+
+        Installs snmpd-nossl (the daemon) and luci-app-snmpd (UCI schema +
+        proper procd init script).  Configures community 'public' via UCI.
+        """
+        lines: list = []
+
+        # 1. Install packages — snmpd-nossl (daemon) + luci-app-snmpd (UCI init)
+        pm = self._pm_type()
+        if pm == "apk":
+            raw = self._send_command("apk add snmpd-nossl luci-app-snmpd 2>&1")
+        else:
+            self._send_command("opkg update 2>/dev/null || true")
+            raw = self._send_command("opkg install snmpd-nossl luci-app-snmpd 2>&1")
+        out = self._clean_pkg_output(raw)
+        low = out.lower()
+        installed = not any(kw in low for kw in ("error:", "failed"))
+        lines.append(f"[install] {out[-300:]}")
+
+        if not installed and "already installed" not in low:
+            return {"success": False, "output": "\n".join(lines)}
+
+        # 2. Configure via UCI — modify the existing default sections only.
+        #    Do NOT create new named sections (causes duplicate directives in
+        #    the generated /var/run/snmpd.conf which crashes snmpd).
+        #    Also remove any stale named sections from previous fix attempts.
+        # The init script reads these UCI field names to generate /var/run/snmpd.conf:
+        #   agent:  agentaddress
+        #   com2sec: secname, source, community
+        #   group:  group (name!), version, secname
+        #   view:   viewname (not name!), type, oid
+        #   access: group, version, level, prefix, read, write, notify
+        # Default luci-app-snmpd schema uses different field names for group/view/access,
+        # so we patch all required fields explicitly.
+        uci_cmds = [
+            # Remove any stale named sections from previous runs
+            "uci -q delete snmpd.agent",
+            "uci -q delete snmpd.public",
+            # agent
+            "uci set snmpd.@agent[0].agentaddress='161'",
+            # com2sec
+            "uci set snmpd.@com2sec[0].secname='ro'",
+            "uci set snmpd.@com2sec[0].source='0.0.0.0/0'",
+            "uci set snmpd.@com2sec[0].community='public'",
+            # group — init script reads field 'group' (not 'name')
+            "uci set snmpd.@group[0].group='rogroup'",
+            "uci set snmpd.@group[0].version='v2c'",
+            "uci set snmpd.@group[0].secname='ro'",
+            # view — init script reads field 'viewname' (not 'name')
+            "uci set snmpd.@view[0].viewname='all'",
+            "uci set snmpd.@view[0].type='included'",
+            "uci set snmpd.@view[0].oid='.1'",
+            # access — init script needs write + notify or it returns early
+            "uci set snmpd.@access[0].group='rogroup'",
+            "uci set snmpd.@access[0].context='none'",
+            "uci set snmpd.@access[0].version='v2c'",
+            "uci set snmpd.@access[0].level='noAuthNoPriv'",
+            "uci set snmpd.@access[0].prefix='exact'",
+            "uci set snmpd.@access[0].read='all'",
+            "uci set snmpd.@access[0].write='none'",
+            "uci set snmpd.@access[0].notify='none'",
+            "uci commit snmpd",
+        ]
+        for cmd in uci_cmds:
+            self._send_command(f"{cmd} 2>/dev/null || true")
+        lines.append("[config] Configured snmpd via UCI (all required fields set).")
+
+        # 3. Firewall: allow UDP 161 from the management zone
+        try:
+            # Use `ss` (always available on OpenWrt) to find the IP of the current SSH client
+            raw_conn = self._send_command(
+                "ss -tnp 2>/dev/null | awk '/ESTAB.*:22/{print $5}' | head -1 | sed 's/:[0-9]*$//'"
+            ).strip()
+            # Strip IPv6-mapped prefix if present
+            if raw_conn.startswith("::ffff:"):
+                raw_conn = raw_conn[7:]
+            peer_ip = raw_conn.splitlines()[-1].strip() if raw_conn else ""
+
+            if peer_ip and peer_ip not in ("", "0.0.0.0", "::"):
+                # Determine which firewall zone owns the interface that routes to peer_ip.
+                # Walk uci firewall zones: find the zone whose associated network interface
+                # has an address in the same /24 as peer_ip.
+                peer_prefix = peer_ip.rsplit(".", 1)[0] if "." in peer_ip else ""
+                fw_out = self._send_command("uci show firewall 2>/dev/null || true")
+                src_zone = "*"
+                # Build map zone_name → zone_uci_key
+                zone_map: dict[str, str] = {}
+                for fw_line in fw_out.splitlines():
+                    m = re.match(r"firewall\.(\w+)\.name='([^']+)'", fw_line.strip())
+                    if m:
+                        zone_map[m.group(2)] = m.group(1)
+                # For each zone find its network, then the interface IP
+                for zone_name, uci_key in zone_map.items():
+                    net_line = self._send_command(
+                        f"uci -q get firewall.{uci_key}.network 2>/dev/null || true"
+                    ).strip()
+                    if not net_line:
+                        continue
+                    for net_name in net_line.split():
+                        iface_ip = self._send_command(
+                            f"uci -q get network.{net_name}.ipaddr 2>/dev/null || true"
+                        ).strip()
+                        if iface_ip and "." in iface_ip:
+                            iface_prefix = iface_ip.split("/")[0].rsplit(".", 1)[0]
+                            if peer_prefix and iface_prefix == peer_prefix:
+                                src_zone = zone_name
+                                break
+                    if src_zone != "*":
+                        break
+
+                self._send_command(
+                    f"uci -q delete firewall.snmp_netork 2>/dev/null; "
+                    f"uci set firewall.snmp_netork=rule; "
+                    f"uci set firewall.snmp_netork.name='Allow-SNMP-from-mgmt'; "
+                    f"uci set firewall.snmp_netork.src='{src_zone}'; "
+                    f"uci set firewall.snmp_netork.dest_port='161'; "
+                    f"uci set firewall.snmp_netork.proto='udp'; "
+                    f"uci set firewall.snmp_netork.target='ACCEPT'; "
+                    f"uci commit firewall; "
+                    f"/etc/init.d/firewall reload 2>/dev/null || true"
+                )
+                lines.append(f"[firewall] Added UDP:161 allow rule (src zone: {src_zone}).")
+            else:
+                lines.append("[firewall] Could not detect peer IP via ss — skipping firewall step.")
+        except Exception as exc:
+            lines.append(f"[firewall] skipped — {exc}")
+
+        # 4. Break any crash-loop, then start cleanly
+        self._send_command("/etc/init.d/snmpd stop 2>/dev/null; true")
+        _time.sleep(2)
+        self._send_command("pkill -9 snmpd 2>/dev/null; true")  # kill crash-loop zombie
+        _time.sleep(3)
+        self._send_command("/etc/init.d/snmpd enable 2>/dev/null; true")
+        self._send_command("/etc/init.d/snmpd start 2>/dev/null; true")
+        _time.sleep(4)
+        lines.append("[service] snmpd started via procd.")
+
+        # 5. Check if snmpd is now active (no local snmpget on OpenWrt by default)
+        status = self._send_command(
+            "/etc/init.d/snmpd running 2>/dev/null && echo active || echo inactive"
+        ).strip()
+        success = status == "active"
+        if success:
+            lines.append("[ok] snmpd is active.")
+        else:
+            lines.append(f"[warn] snmpd status: {status}")
+
+        return {"success": success, "output": "\n".join(lines)}
+
+    def get_users(self) -> dict[str, Any]:
+        """Return users configured on the device.
+
+        Parses ``/etc/passwd`` for accounts with a valid login shell.
+        SSH public keys are read from ``~<user>/.ssh/authorized_keys``
+        (Dropbear also stores root keys at ``/etc/dropbear/authorized_keys``).
+
+        Level mapping:
+        - UID 0 (root) → 15 (full access)
+        - all other users → 1
+        """
+        passwd_out = self._send_command("cat /etc/passwd")
+        # Root authorized_keys locations on OpenWrt
+        root_keys_out = self._send_command(
+            ["cat /root/.ssh/authorized_keys", "cat /etc/dropbear/authorized_keys"]
+        )
+
+        users: dict[str, Any] = {}
+        valid_shells = {"/bin/sh", "/bin/ash", "/bin/bash", "/usr/bin/fish"}
+
+        for line in passwd_out.splitlines():
+            parts = line.strip().split(":")
+            if len(parts) < 7:
+                continue
+            username, password_hash, uid_str, _, _, home, shell = (
+                parts[0], parts[1], parts[2], parts[3], parts[4], parts[5], parts[6],
+            )
+            if shell not in valid_shells:
+                continue
+            try:
+                uid = int(uid_str)
+            except ValueError:
+                continue
+
+            level = 15 if uid == 0 else 1
+
+            # Collect SSH keys for this user
+            sshkeys: list[str] = []
+            if uid == 0:
+                for line_k in root_keys_out.splitlines():
+                    line_k = line_k.strip()
+                    if line_k and not line_k.startswith("#"):
+                        sshkeys.append(line_k)
+            else:
+                # Try reading per-user authorized_keys
+                keys_out = self._send_command(f"cat {home}/.ssh/authorized_keys 2>/dev/null")
+                for line_k in keys_out.splitlines():
+                    line_k = line_k.strip()
+                    if line_k and not line_k.startswith("#"):
+                        sshkeys.append(line_k)
+
+            users[username] = {
+                "level": level,
+                "password": password_hash,
+                "sshkeys": sshkeys,
+            }
+
+        return users
+
+    def get_services(self) -> list[dict[str, Any]]:
+        """Return all system services with their running and enabled state.
+
+        Uses ``ubus call service list`` for running/PID info and
+        ``/etc/rc.d/S*`` symlinks for enabled-at-boot state.
+        """
+        import json as _json
+
+        # -- enabled set: names from /etc/rc.d/S<priority><name> symlinks ----
+        rc_out = self._send_command(
+            "ls /etc/rc.d/ 2>/dev/null | grep '^S' | sed 's/^S[0-9]*//'"
+        )
+        enabled: set[str] = {s.strip() for s in rc_out.splitlines() if s.strip()}
+
+        # -- running info from procd via ubus ---------------------------------
+        ubus_raw = self._send_command("ubus call service list 2>/dev/null")
+        ubus_data: dict = {}
+        try:
+            ubus_data = _json.loads(ubus_raw)
+        except (ValueError, TypeError):
+            pass
+
+        # Build index from ubus data
+        service_map: dict[str, dict] = {}
+        for svc_name, svc_info in ubus_data.items():
+            if not isinstance(svc_info, dict):
+                continue
+            instances = svc_info.get("instances", {})
+            running = any(
+                inst.get("running", False)
+                for inst in instances.values()
+                if isinstance(inst, dict)
+            )
+            pid = next(
+                (
+                    inst.get("pid", 0)
+                    for inst in instances.values()
+                    if isinstance(inst, dict) and inst.get("running")
+                ),
+                0,
+            )
+            service_map[svc_name] = {"running": running, "pid": pid}
+
+        # -- all init scripts -------------------------------------------------
+        init_raw = self._send_command("ls -1 /etc/init.d/ 2>/dev/null")
+        init_scripts: set[str] = {s.strip() for s in init_raw.splitlines() if s.strip()}
+
+        # Merge: all known services (from init.d + ubus)
+        all_names = init_scripts | set(service_map.keys())
+        # Exclude procd internal pseudo-service
+        all_names.discard("")
+
+        result: list[dict[str, Any]] = []
+        for name in sorted(all_names):
+            info = service_map.get(name, {})
+            result.append({
+                "name": name,
+                "running": info.get("running", False),
+                "enabled": name in enabled,
+                "pid": info.get("pid", 0),
+            })
+
+        return result
+
+    def manage_service(self, name: str, action: str) -> dict[str, Any]:
+        """Execute a lifecycle action (start/stop/restart/enable/disable) on a service."""
+        import re as _re
+        if not _re.match(r'^[a-zA-Z0-9_\-]+$', name):
+            raise ValueError(f"Invalid service name: {name!r}")
+        if action not in ('start', 'stop', 'restart', 'enable', 'disable'):
+            raise ValueError(f"Invalid action: {action!r}")
+        output = self._send_command(f"/etc/init.d/{name} {action} 2>&1")
+        return {"success": True, "output": output}
+
+    def get_device_warnings(self) -> list[dict[str, Any]]:
+        """Return a list of warning dicts for issues detected on this device.
+
+        Currently detects:
+        - lldpd not installed (LLDP neighbor discovery unavailable)
+        - package updates available (uses local package cache, no network call)
+        - update notifications not configured (auc not installed, opkg only)
+        """
+        warnings: list[dict[str, Any]] = []
+
+        # 1. LLDP daemon
+        lldpd_path = self._send_command("which lldpd 2>/dev/null").strip()
+        if not lldpd_path:
+            warnings.append({
+                "code": "lldpd_not_installed",
+                "severity": "warning",
+                "action": "install_lldpd",
+            })
+
+        pm = self._pm_type()
+
+        # 2. Package updates available (local cache only – no opkg update)
+        def _pkg_name(line: str, pm_type: str) -> str:
+            """Extract just the package name from an upgradable line.
+
+            apk:  'luci-app-firewall-26.143~abc < 26.151~def' → 'luci-app-firewall'
+            opkg: 'luci-app-firewall - 1.0 - 1.1'             → 'luci-app-firewall'
+            """
+            import re as _re
+            if pm_type == "apk":
+                # Strip trailing ' < ...' then remove the version suffix (-\d...)
+                name_ver = line.split(" ")[0]
+                m = _re.match(r'^(.*?)-\d', name_ver)
+                return m.group(1) if m else name_ver
+            else:
+                return line.split(" - ")[0].strip()
+
+        try:
+            if pm == "apk":
+                raw_upg = self._send_command("apk version 2>/dev/null | grep '<'")
+            else:
+                raw_upg = self._send_command("opkg list-upgradable 2>/dev/null")
+            upgradable = [ln.strip() for ln in raw_upg.splitlines() if ln.strip()]
+        except Exception:
+            upgradable = []
+
+        if upgradable:
+            pkg_names = [_pkg_name(ln, pm) for ln in upgradable]
+            warnings.append({
+                "code": "updates_available",
+                "severity": "info",
+                "action": None,
+                "meta": {
+                    "count": len(upgradable),
+                    "packages": pkg_names[:10],
+                },
+            })
+
+        # 3. Attended sysupgrade client not installed (opkg systems only)
+        if pm == "opkg":
+            auc_path = self._send_command("which auc 2>/dev/null").strip()
+            if not auc_path:
+                warnings.append({
+                    "code": "update_notifications_disabled",
+                    "severity": "warning",
+                    "action": "install_auc",
+                })
+
+        # 4. base64 not available — needed for efficient config apply
+        b64_path = self._send_command("command -v base64 2>/dev/null").strip()
+        if not b64_path:
+            warnings.append({
+                "code": "no_base64",
+                "severity": "warning",
+                "action": "install_coreutils_base64",
+            })
+
+        return warnings
+
+    def get_available_updates(self) -> list[dict[str, Any]]:
+        """Return list of upgradable packages from the local package manager cache."""
+        import re as _re
+        pm = self._pm_type()
+        updates: list[dict[str, Any]] = []
+
+        if pm == "apk":
+            # Output format: "pkgname-current_ver < new_ver"
+            raw = self._send_command("apk version 2>/dev/null | grep '<'")
+            for line in raw.splitlines():
+                line = line.strip()
+                m = _re.match(r'^(.+)-(\d\S*)\s+<\s+(\S+)', line)
+                if m:
+                    updates.append({
+                        "name": m.group(1),
+                        "current_version": m.group(2),
+                        "new_version": m.group(3),
+                    })
+        else:
+            # opkg output: "pkgname - current_ver - new_ver"
+            raw = self._send_command("opkg list-upgradable 2>/dev/null")
+            for line in raw.splitlines():
+                parts = [p.strip() for p in line.split(" - ")]
+                if len(parts) == 3:
+                    updates.append({
+                        "name": parts[0],
+                        "current_version": parts[1],
+                        "new_version": parts[2],
+                    })
+
+        return sorted(updates, key=lambda u: u["name"])
+
+    def apply_updates(self, packages: list[str]) -> dict[str, Any]:
+        """Upgrade the given packages using the device's package manager."""
+        import re as _re
+        for pkg in packages:
+            if not _re.match(r'^[a-zA-Z0-9_\-\+\.]+$', pkg):
+                raise ValueError(f"Invalid package name: {pkg!r}")
+        pm = self._pm_type()
+        pkg_args = " ".join(packages)
+        if pm == "apk":
+            cmd = f"apk upgrade {pkg_args} 2>&1"
+        else:
+            cmd = f"opkg upgrade {pkg_args} 2>&1"
+        output = self._send_command(cmd)
+        return {"success": True, "output": output}
+
+    # ------------------------------------------------------------------
+    # NTP
+    # ------------------------------------------------------------------
+
+    def get_ntp_servers(self) -> dict[str, Any]:
+        """Return configured NTP servers from ``uci show system``.
+
+        UCI example::
+
+            system.ntp.server='0.openwrt.pool.ntp.org 1.openwrt.pool.ntp.org'
+        """
+        uci_out = self._send_command("uci show system")
+        servers: dict[str, Any] = {}
+
+        for line in uci_out.splitlines():
+            # Handles both list and single-value UCI representations
+            m = re.match(r"system\.ntp\.server(?:\[\d+\])?='([^']*)'", line.strip())
+            if m:
+                for srv in m.group(1).split():
+                    srv = srv.strip()
+                    if srv:
+                        servers[srv] = {}
+
+        return servers
+
+    def get_ntp_peers(self) -> dict[str, Any]:
+        """Return NTP peers from ``uci show system``.
+
+        OpenWrt's busybox ntpd does not differentiate peers from servers;
+        the same UCI ``ntp.server`` list is returned.
+        """
+        return self.get_ntp_servers()
+
+    def get_ntp_stats(self) -> list[dict[str, Any]]:
+        """Return NTP synchronisation statistics.
+
+        Tries ``ntpq -pn`` first (ntpd), then ``chronyc sources -v`` (chrony).
+        Returns an empty list when neither tool is available.
+
+        ``ntpq -pn`` example line::
+
+            *188.114.101.4   188.114.100.1    4 u  107  256  377  164.228 -13.866   2.695
+
+        ``chronyc sources -v`` example line::
+
+            ^* 192.168.1.1                   2   6    17     8  +2345us[  0ns] +/-   15ms
+        """
+        ntpq_out = self._send_command("ntpq -pn")
+        if ntpq_out and not ntpq_out.startswith(("ntpq: ", "sh: ", "ash: ", "command not found")):
+            return self._parse_ntpq(ntpq_out)
+
+        chrony_out = self._send_command("chronyc sources -v")
+        if chrony_out and not chrony_out.startswith(("sh: ", "ash: ", "command not found")):
+            return self._parse_chronyc(chrony_out)
+
+        return []
+
+    @staticmethod
+    def _parse_ntpq(output: str) -> list[dict[str, Any]]:
+        """Parse ``ntpq -pn`` tabular output."""
+        stats = []
+        for line in output.splitlines():
+            line_s = line.strip()
+            if not line_s or line_s.startswith(("remote", "=")):
+                continue
+            # First char is the tally code (* = synchronized, + = candidate, etc.)
+            tally = line_s[0] if line_s[0] in "* +-x.o#" else " "
+            parts = line_s[1:].split()
+            if len(parts) < 10:
+                continue
+            try:
+                stats.append({
+                    "remote": parts[0],
+                    "referenceid": parts[1],
+                    "synchronized": tally == "*",
+                    "stratum": int(parts[2]),
+                    "type": parts[3],
+                    "when": parts[4],
+                    "hostpoll": int(parts[5]),
+                    "reachability": int(parts[6], 8),  # octal
+                    "delay": float(parts[7]),
+                    "offset": float(parts[8]),
+                    "jitter": float(parts[9]),
+                })
+            except (ValueError, IndexError):
+                continue
+        return stats
+
+    @staticmethod
+    def _parse_chronyc(output: str) -> list[dict[str, Any]]:
+        """Parse ``chronyc sources -v`` tabular output."""
+        stats = []
+        for line in output.splitlines():
+            line_s = line.strip()
+            # Data lines start with ^* ^+ ^- ^?
+            m = re.match(r"^(\^[*+\-?])\s+(\S+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\S+)\s+(.*)", line_s)
+            if not m:
+                continue
+            tally = m.group(1)
+            try:
+                stats.append({
+                    "remote": m.group(2),
+                    "referenceid": "",
+                    "synchronized": tally == "^*",
+                    "stratum": int(m.group(3)),
+                    "type": "u",
+                    "when": m.group(6),
+                    "hostpoll": int(m.group(4)),
+                    "reachability": int(m.group(5), 8) if re.match(r"^[0-7]+$", m.group(5)) else 0,
+                    "delay": 0.0,
+                    "offset": 0.0,
+                    "jitter": 0.0,
+                })
+            except (ValueError, IndexError):
+                continue
+        return stats

napalm_openwrt/vlan_mixin.py

@@ -0,0 +1,221 @@
+# -*- 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.
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+
+class OpenWrtVLANMixin:
+    """Mixin providing VLAN and network-instance NAPALM getters."""
+
+    def get_vlans(self) -> dict[str, Any]:
+        """Return VLAN information with proper tagged/untagged separation.
+
+        Uses ``bridge vlan show`` (DSA-based OpenWrt ≥21.02) for VLAN/port
+        membership and ``uci show network`` for VLAN names.
+
+        A port marked *PVID Egress Untagged* is an untagged member.
+        All other VLAN memberships for the same port are tagged.
+
+        Also detects legacy 802.1q sub-interfaces (``eth0.10`` etc.) from
+        ``ip link show``.  The parent interface (``eth0``) is added as a
+        tagged member for every such VLAN.
+        """
+        bridge_out = self._send_command("bridge vlan show")
+        uci_out = self._send_command("uci show network")
+
+        # vlan_id -> {name, tagged: [], untagged: []}
+        vlans: dict[str, Any] = {}
+        current_port: str | None = None
+
+        for line in bridge_out.splitlines():
+            line_s = line.strip()
+            if not line_s or line_s.lower().startswith("port"):
+                continue
+
+            # Port line: "eth0              1 PVID Egress Untagged"
+            m = re.match(r"^(\S+)\s+(\d+)(.*)", line)
+            if m:
+                current_port = m.group(1)
+                vlan_id = str(int(m.group(2)))
+                flags = m.group(3).upper()
+                vlans.setdefault(vlan_id, {"name": "", "tagged": [], "untagged": []})
+                if "PVID" in flags or "UNTAGGED" in flags:
+                    if current_port not in vlans[vlan_id]["untagged"]:
+                        vlans[vlan_id]["untagged"].append(current_port)
+                else:
+                    if current_port not in vlans[vlan_id]["tagged"]:
+                        vlans[vlan_id]["tagged"].append(current_port)
+                continue
+
+            # Continuation line with only a VLAN ID (tagged for current_port)
+            m = re.match(r"^(\d+)", line_s)
+            if m and current_port:
+                vlan_id = str(int(m.group(1)))
+                vlans.setdefault(vlan_id, {"name": "", "tagged": [], "untagged": []})
+                if current_port not in vlans[vlan_id]["tagged"]:
+                    vlans[vlan_id]["tagged"].append(current_port)
+
+        # Enrich with UCI VLAN names from explicit bridge-vlan sections
+        uci_entries: dict[str, dict[str, str]] = {}
+        for line in uci_out.splitlines():
+            m = re.match(r"network\.@bridge-vlan\[(\d+)\]\.(\w+)='([^']*)'", line.strip())
+            if m:
+                idx, key, value = m.group(1), m.group(2), m.group(3)
+                uci_entries.setdefault(idx, {})[key] = value
+
+        for entry in uci_entries.values():
+            if "vlan" in entry and "name" in entry:
+                vlan_id = str(int(entry["vlan"]))
+                if vlan_id in vlans:
+                    vlans[vlan_id]["name"] = entry["name"]
+
+        # Also derive VLAN names from UCI network interface sections that
+        # reference subinterfaces like eth0.N or br-ap.N:
+        #   network.guest.device='eth0.8'    -> VLAN 8 name = "guest"
+        #   network.ap_v8.device='br-ap.8'   -> VLAN 8 name = "ap_v8"
+        # Only fills in names that are still empty after bridge-vlan lookup.
+        for line in uci_out.splitlines():
+            m = re.match(r"network\.(\w+)\.device='[\w-]+\.(\d+)'", line.strip())
+            if m:
+                section_name, vid_str = m.group(1), m.group(2)
+                vlan_id = str(int(vid_str))
+                if vlan_id in vlans and not vlans[vlan_id]["name"]:
+                    vlans[vlan_id]["name"] = section_name
+
+        # Also detect VLAN sub-interfaces (eth0.10, br-ap.8, ...) from ip link show.
+        # The sub-interface is the untagged egress point; its parent is tagged.
+        link_out = self._send_command("ip link show")
+        for line in link_out.splitlines():
+            lm = re.match(r"^\d+:\s+(\S+?)[@:]", line)
+            if not lm:
+                continue
+            iface = lm.group(1)
+            vm = re.match(r"^([\w-]+)\.(\d+)$", iface)  # allow hyphens (br-ap)
+            if not vm:
+                continue
+            parent = vm.group(1)          # e.g. "eth0" or "br-ap"
+            vlan_id = str(int(vm.group(2)))  # e.g. "10"
+            vlans.setdefault(vlan_id, {"name": "", "tagged": [], "untagged": []})
+            # sub-interface itself -> untagged egress
+            if iface not in vlans[vlan_id]["untagged"] and iface not in vlans[vlan_id]["tagged"]:
+                vlans[vlan_id]["untagged"].append(iface)
+            # parent -> tagged trunk
+            if parent not in vlans[vlan_id]["tagged"] and parent not in vlans[vlan_id]["untagged"]:
+                vlans[vlan_id]["tagged"].append(parent)
+
+        return vlans
+
+    def delete_vlan(self, vlan_id: int) -> None:
+        """Remove a VLAN from the device by deleting its UCI bridge-vlan section.
+
+        Finds the ``network.@bridge-vlan[N]`` section whose ``.vlan`` matches
+        *vlan_id*, deletes it and commits.  If no matching section is found the
+        method is a no-op (the VLAN may only exist as an eth0.N sub-interface,
+        which cannot be deleted via UCI alone).
+
+        :param vlan_id: VLAN ID to remove.
+        :raises ValueError: If *vlan_id* is out of the valid range.
+        """
+        if not 1 <= vlan_id <= 4094:
+            raise ValueError(f"VLAN ID {vlan_id} is out of range (1-4094)")
+
+        uci_out = self.cli(["uci show network"]).get("uci show network", "")
+        idx = None
+        for line in uci_out.splitlines():
+            m = re.match(r"network\.@bridge-vlan\[(\d+)\]\.vlan='(\d+)'", line.strip())
+            if m and int(m.group(2)) == vlan_id:
+                idx = m.group(1)
+                break
+
+        if idx is None:
+            # No explicit bridge-vlan section -- nothing to delete via UCI
+            return
+
+        self.cli([
+            f"uci delete network.@bridge-vlan[{idx}]",
+            "uci commit network",
+            "/etc/init.d/network reload",
+        ])
+
+    def get_network_instances(self, name: str = "") -> dict[str, Any]:
+        """Return network instances (Linux network namespaces + default).
+
+        The ``default`` instance contains all interfaces not assigned to a
+        named namespace.  Named namespaces are discovered via ``ip netns list``.
+
+        Example::
+
+            {
+                'default': {
+                    'name': 'default',
+                    'type': 'DEFAULT_INSTANCE',
+                    'state': {'route_distinguisher': None},
+                    'interfaces': {'interface': {'br-lan': {}, 'eth0': {}}}
+                }
+            }
+        """
+        netns_out = self._send_command("ip netns list")
+        iface_list = self._get_interface_list()
+
+        instances: dict[str, Any] = {}
+
+        # Named namespaces
+        netns_names: list[str] = []
+        for line in netns_out.splitlines():
+            line_s = line.strip()
+            if not line_s:
+                continue
+            # "myns (id: 3)" or just "myns"
+            ns_name = line_s.split()[0]
+            netns_names.append(ns_name)
+
+            # Interfaces inside the namespace
+            ns_ifaces_out = self._send_command(f"ip netns exec {ns_name} ip link show")
+            ns_ifaces: dict[str, Any] = {}
+            for iline in ns_ifaces_out.splitlines():
+                im = re.match(r"^\d+:\s+(\S+?)[@:]", iline)
+                if im and im.group(1) != "lo":
+                    ns_ifaces[im.group(1)] = {}
+
+            instances[ns_name] = {
+                "name": ns_name,
+                "type": "L3VRF",
+                "state": {"route_distinguisher": None},
+                "interfaces": {"interface": ns_ifaces},
+            }
+
+        # Default instance: interfaces NOT in any named namespace
+        # (on most OpenWrt devices there are no named namespaces)
+        default_ifaces = {iface: {} for iface in iface_list}
+        instances["default"] = {
+            "name": "default",
+            "type": "DEFAULT_INSTANCE",
+            "state": {"route_distinguisher": None},
+            "interfaces": {"interface": default_ifaces},
+        }
+
+        if name:
+            return {k: v for k, v in instances.items() if k == name}
+
+        return instances
+
+    def set_hostname(self, new_hostname: str) -> None:
+        """Set the system hostname via UCI and reload the system service."""
+        self._send_command(
+            f"uci set system.@system[0].hostname='{new_hostname}' && "
+            f"uci commit system && "
+            f"/etc/init.d/system reload"
+        )

napalm_openwrt/wireless_mixin.py

@@ -0,0 +1,666 @@
+# -*- 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.
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+
+class OpenWrtWirelessMixin:
+    """Mixin providing wireless/radio NAPALM getters."""
+
+    def get_ssids(self) -> dict[str, Any]:
+        """Return configured SSIDs from UCI wireless configuration.
+
+        Parses ``uci show wireless`` for ``wifi-iface`` entries and enriches
+        each entry with:
+
+        * ``band`` — human-readable frequency band ("2.4 GHz", "5 GHz", "6 GHz")
+          derived from the radio's ``band`` or ``hwmode`` UCI key.
+        * ``encryption`` — human-readable security mode ("WPA2-PSK", "Open", …).
+
+        When the same SSID name is broadcast on multiple radios, the keys in
+        the returned dict are disambiguated as ``"ssid (2.4 GHz)"`` /
+        ``"ssid (5 GHz)"``.
+        """
+        uci_out = self._send_command("uci show wireless")
+
+        # Collect radio band info: radio0 → "2g", radio1 → "5g", …
+        radio_bands: dict[str, str] = {}
+        iface_entries: dict[str, dict[str, str]] = {}
+
+        # First pass: identify named sections that are wifi-iface types and
+        # collect radio band info.
+        named_iface_sections: set = set()
+        for line in uci_out.splitlines():
+            line_s = line.strip()
+            # named wifi-iface declaration: wireless.managed_family_2g=wifi-iface
+            nm = re.match(r"wireless\.(\w+)=wifi-iface", line_s)
+            if nm:
+                named_iface_sections.add(nm.group(1))
+                continue
+            # radio device config: wireless.radio0.band='2g'
+            rm = re.match(r"wireless\.(radio\d+)\.(band|hwmode)='([^']*)'", line_s)
+            if rm:
+                radio, key, val = rm.group(1), rm.group(2), rm.group(3)
+                if key == "band" or radio not in radio_bands:
+                    radio_bands[radio] = val
+
+        # Second pass: collect iface properties (both anonymous and named sections)
+        for line in uci_out.splitlines():
+            line_s = line.strip()
+            # radio device config (already handled above)
+            rm = re.match(r"wireless\.(radio\d+)\.(band|hwmode)='([^']*)'", line_s)
+            if rm:
+                radio, key, val = rm.group(1), rm.group(2), rm.group(3)
+                # Prefer 'band' over 'hwmode' when both present
+                if key == "band" or radio not in radio_bands:
+                    radio_bands[radio] = val
+                continue
+            # anonymous wifi-iface values: wireless.@wifi-iface[0].ssid='MyNet'
+            im = re.match(r"wireless\.@wifi-iface\[(\d+)\]\.(\w+)='([^']*)'", line_s)
+            if im:
+                idx, key, val = im.group(1), im.group(2), im.group(3)
+                iface_entries.setdefault(idx, {})[key] = val
+                continue
+            # named wifi-iface values: wireless.managed_family_2g.ssid='manivong'
+            nm = re.match(r"wireless\.(\w+)\.(\w+)='([^']*)'", line_s)
+            if nm and nm.group(1) in named_iface_sections:
+                section, key, val = nm.group(1), nm.group(2), nm.group(3)
+                iface_entries.setdefault(section, {})[key] = val
+
+        def _band_label(radio: str) -> str:
+            raw = radio_bands.get(radio, "").lower()
+            if raw in ("2g", "11g", "b", "g", "bg", "bgn", "b/g", "b/g/n"):
+                return "2.4 GHz"
+            if raw in ("5g", "11a", "a", "ac", "ax5", "a/n", "a/n/ac"):
+                return "5 GHz"
+            if raw in ("6g", "ax6"):
+                return "6 GHz"
+            return ""
+
+        _ENC_MAP = {
+            "": "Open", "none": "Open", "0": "Open",
+            "wep": "WEP", "wep-open": "WEP (Open)", "wep-shared": "WEP (Shared)",
+            "psk": "WPA-PSK",
+            "psk+ccmp": "WPA-PSK",
+            "psk-mixed": "WPA/WPA2-PSK",
+            "psk2": "WPA2-PSK",
+            "psk2+ccmp": "WPA2-PSK",
+            "psk2+aes": "WPA2-PSK",
+            "psk3": "WPA3-SAE",
+            "psk2+psk3": "WPA2/WPA3",
+            "sae": "WPA3-SAE",
+            "sae-mixed": "WPA2/WPA3",
+            "wpa": "WPA-Enterprise",
+            "wpa2": "WPA2-Enterprise",
+            "wpa3": "WPA3-Enterprise",
+            "ccmp": "WPA2-PSK",
+        }
+
+        def _enc_label(enc_raw: str) -> str:
+            return _ENC_MAP.get(enc_raw.lower(), enc_raw.upper() or "Open")
+
+        # Build network→vlan_id map from UCI network config.
+        # A wifi-iface has option network='ap_7'; the corresponding UCI network
+        # interface has either an explicit vid ('7') or a bridge device whose
+        # name encodes the VLAN, e.g. br-ap.7 → VLAN 7.
+        def _vlan_from_device(dev: str) -> int | None:
+            m = re.search(r"\.(\d+)$", dev)
+            if m:
+                return int(m.group(1))
+            return None
+
+        net_vlan: dict[str, int] = {}
+        try:
+            net_out = self._send_command("uci show network 2>/dev/null || true")
+            net_entries: dict[str, dict[str, str]] = {}
+            for line in net_out.splitlines():
+                line_s = line.strip()
+                m = re.match(r"network\.(\w+)\.(\w+)='([^']*)'", line_s)
+                if m:
+                    iface, key, val = m.group(1), m.group(2), m.group(3)
+                    net_entries.setdefault(iface, {})[key] = val
+            for iface, props in net_entries.items():
+                vid_str = props.get("vid") or props.get("vlan")
+                if vid_str and vid_str.isdigit():
+                    net_vlan[iface] = int(vid_str)
+                    continue
+                dev = props.get("device", "")
+                vlan = _vlan_from_device(dev)
+                if vlan is not None:
+                    net_vlan[iface] = vlan
+        except Exception:
+            pass  # Non-fatal: VLAN info is optional enrichment
+
+        # Build result; group entries with the same SSID name, merging bands
+        result: dict[str, Any] = {}
+        # Intermediate: ssid -> list of bands seen
+        ssid_bands: dict[str, list[str]] = {}
+        for entry in iface_entries.values():
+            ssid = entry.get("ssid")
+            if not ssid:
+                continue
+            radio = entry.get("device", "")
+            band = _band_label(radio)
+            disabled = entry.get("disabled", "0") == "1"
+            enc_raw = entry.get("encryption", "") or ""
+            encryption = _enc_label(enc_raw)
+            hidden = entry.get("hidden", "0") == "1"
+            network_name = entry.get("network", "")
+            vlan_id: int | None = net_vlan.get(network_name)
+            ft_enabled = entry.get("ieee80211r", "0") == "1"
+            ft_mobility_domain = entry.get("mobility_domain", "")
+            ft_over_ds = entry.get("ft_over_ds", "1") == "1"
+            client_isolation = entry.get("isolate", "0") == "1"
+            _max_raw = entry.get("maxassoc")
+            max_clients: int | None = int(_max_raw) if _max_raw and str(_max_raw).isdigit() else None
+            _disassoc_raw = entry.get("disassoc_low_ack")
+            disassoc_low_ack: bool | None = (_disassoc_raw == "1") if _disassoc_raw is not None else None
+            _max_inact_raw = entry.get("max_inactivity")
+            max_inactivity: int | None = int(_max_inact_raw) if _max_inact_raw and str(_max_inact_raw).isdigit() else None
+            key: str = entry.get("key", "") or ""
+
+            if ssid in result:
+                # Merge: append band if not already present
+                if band and band not in ssid_bands[ssid]:
+                    ssid_bands[ssid].append(band)
+                    # Keep alphabetical order so 2.4 GHz comes before 5 GHz
+                    ssid_bands[ssid].sort()
+                    result[ssid]["band"] = " + ".join(ssid_bands[ssid])
+                    result[ssid]["bands_list"] = list(ssid_bands[ssid])
+                # If one radio is enabled, the SSID counts as enabled
+                if not disabled:
+                    result[ssid]["enabled"] = True
+                # Keep vlan_id if not yet set
+                if result[ssid].get("vlan_id") is None and vlan_id is not None:
+                    result[ssid]["vlan_id"] = vlan_id
+                # FT: if any radio has ieee80211r enabled, mark the SSID as FT-enabled
+                if ft_enabled:
+                    result[ssid]["ieee80211r"] = True
+                    result[ssid]["mobility_domain"] = ft_mobility_domain
+                    result[ssid]["ft_over_ds"] = ft_over_ds
+                # Client isolation: if any iface has it, mark True
+                if client_isolation:
+                    result[ssid]["client_isolation"] = True
+                # Max clients: keep first non-None value
+                if max_clients is not None and result[ssid].get("max_clients") is None:
+                    result[ssid]["max_clients"] = max_clients
+                # disassoc_low_ack / max_inactivity: keep first explicit value
+                if disassoc_low_ack is not None and result[ssid].get("disassoc_low_ack") is None:
+                    result[ssid]["disassoc_low_ack"] = disassoc_low_ack
+                if max_inactivity is not None and result[ssid].get("max_inactivity") is None:
+                    result[ssid]["max_inactivity"] = max_inactivity
+                # key: keep first non-empty value seen
+                if key and not result[ssid].get("key"):
+                    result[ssid]["key"] = key
+            else:
+                ssid_bands[ssid] = [band] if band else []
+                result[ssid] = {
+                    "enabled": not disabled,
+                    "radio": radio,
+                    "band": band,
+                    "bands_list": list(ssid_bands[ssid]),
+                    "bssid": "",
+                    "encryption": encryption,
+                    "encryption_uci": enc_raw,
+                    "hidden": hidden,
+                    "client_isolation": client_isolation,
+                    "max_clients": max_clients,
+                    "clients": 0,
+                    "vlan_id": vlan_id,
+                    "ieee80211r": ft_enabled,
+                    "mobility_domain": ft_mobility_domain,
+                    "ft_over_ds": ft_over_ds,
+                    "disassoc_low_ack": disassoc_low_ack,
+                    "max_inactivity": max_inactivity,
+                    "key": key,
+                }
+        return result
+
+    def get_wireless_clients(self) -> list[dict[str, Any]]:
+        """Return currently associated wireless clients from all AP interfaces.
+
+        Uses ``iw dev`` to discover AP-mode interfaces and then
+        ``iw dev <iface> station dump`` to collect per-client statistics.
+        """
+        from napalm_device_types.models import WirelessClientDict
+
+        # Step 1: discover interfaces and their SSIDs / radio mappings
+        iw_out = self._send_command("iw dev 2>/dev/null || true")
+
+        iface_info: dict[str, dict[str, str]] = {}
+        current_phy: str = ""
+        current_iface: str = ""
+
+        for line in iw_out.splitlines():
+            stripped = line.strip()
+            phy_m = re.match(r"^phy#(\d+)$", stripped)
+            if phy_m:
+                current_phy = f"radio{phy_m.group(1)}"
+                current_iface = ""
+                continue
+
+            iface_m = re.match(r"^Interface\s+(\S+)$", stripped)
+            if iface_m:
+                current_iface = iface_m.group(1)
+                iface_info[current_iface] = {"ssid": "", "radio": current_phy, "type": ""}
+                continue
+
+            if not current_iface:
+                continue
+
+            ssid_m = re.match(r"^ssid\s+(.+)$", stripped)
+            if ssid_m:
+                iface_info[current_iface]["ssid"] = ssid_m.group(1)
+                continue
+
+            type_m = re.match(r"^type\s+(\S+)$", stripped)
+            if type_m:
+                iface_info[current_iface]["type"] = type_m.group(1)
+                continue
+
+            # channel 6 (2437 MHz), width: 20 MHz, ...
+            chan_m = re.match(r"^channel\s+\d+\s+\((\d+)\s+MHz\)", stripped)
+            if chan_m:
+                try:
+                    freq = int(chan_m.group(1))
+                    if freq < 3000:
+                        iface_info[current_iface]["band"] = "2.4 GHz"
+                    elif freq < 6000:
+                        iface_info[current_iface]["band"] = "5 GHz"
+                    else:
+                        iface_info[current_iface]["band"] = "6 GHz"
+                except ValueError:
+                    pass
+
+        # Filter to AP-mode interfaces only
+        ap_ifaces = {
+            name: info
+            for name, info in iface_info.items()
+            if info.get("type", "").upper() in ("AP", "AP/VLAN")
+        }
+
+        if not ap_ifaces:
+            return []
+
+        # Step 2: fetch station dumps for all AP interfaces in one SSH call
+        dump_cmd = " ; ".join(
+            f"echo '=== {name} ===' && iw dev {name} station dump 2>/dev/null || true"
+            for name in ap_ifaces
+        )
+        station_out = self._send_command(dump_cmd)
+
+        # Step 3: parse station dump output
+        results: list[dict[str, Any]] = []
+        active_iface: str = ""
+        current_station: dict[str, Any] | None = None
+
+        def _flush() -> None:
+            if current_station and current_station.get("mac"):
+                info = ap_ifaces.get(active_iface, {})
+                results.append(WirelessClientDict(
+                    mac=current_station["mac"],
+                    ssid=info.get("ssid", ""),
+                    radio=info.get("band") or info.get("radio", ""),
+                    signal=current_station.get("signal", 0),
+                    noise=0,
+                    tx_rate=current_station.get("tx_rate", 0.0),
+                    rx_rate=current_station.get("rx_rate", 0.0),
+                    uptime=current_station.get("uptime", 0),
+                ))
+
+        for line in station_out.splitlines():
+            stripped = line.strip()
+
+            # Section header injected above: === wlan0 ===
+            hdr_m = re.match(r"^=== (\S+) ===$", stripped)
+            if hdr_m:
+                _flush()
+                active_iface = hdr_m.group(1)
+                current_station = None
+                continue
+
+            # Station aa:bb:cc:dd:ee:ff (on wlan0)
+            sta_m = re.match(r"^Station\s+([\da-fA-F:]{17})\s+\(", stripped)
+            if sta_m:
+                _flush()
+                current_station = {"mac": sta_m.group(1)}
+                continue
+
+            if current_station is None:
+                continue
+
+            # signal:  -65 dBm  (may be "signal:  -65 [-65] dBm")
+            sig_m = re.match(r"^signal:\s+([-\d]+)", stripped)
+            if sig_m:
+                try:
+                    current_station["signal"] = int(sig_m.group(1))
+                except ValueError:
+                    pass
+                continue
+
+            # tx bitrate:	54.0 MBit/s
+            tx_m = re.match(r"^tx bitrate:\s+([\d.]+)", stripped)
+            if tx_m:
+                try:
+                    current_station["tx_rate"] = float(tx_m.group(1))
+                except ValueError:
+                    pass
+                continue
+
+            # rx bitrate:	72.2 MBit/s
+            rx_m = re.match(r"^rx bitrate:\s+([\d.]+)", stripped)
+            if rx_m:
+                try:
+                    current_station["rx_rate"] = float(rx_m.group(1))
+                except ValueError:
+                    pass
+                continue
+
+            # connected time:	3600 seconds
+            uptime_m = re.match(r"^connected time:\s+(\d+)", stripped)
+            if uptime_m:
+                try:
+                    current_station["uptime"] = int(uptime_m.group(1))
+                except ValueError:
+                    pass
+
+        _flush()
+        return results
+
+    def get_channel_scan(self, mode: str = "active") -> dict[str, Any]:
+        """Return channel scan results for all AP-mode interfaces.
+
+        Uses ``iw dev <iface> scan`` (active) or
+        ``iw dev <iface> scan passive`` (passive) on each AP interface.
+
+        Keys are interface names (e.g. ``"wlan0"``).  Values are lists of
+        :class:`~napalm_device_types.models.ChannelScanEntryDict`.
+        """
+        from napalm_device_types.models import ChannelScanEntryDict
+
+        # Step 1: discover interfaces and filter to AP mode only
+        iw_out = self._send_command("iw dev 2>/dev/null || true")
+        ap_ifaces: list[str] = []
+        current_iface: str = ""
+        current_type: str = ""
+        iface_order: list[tuple[str, str]] = []  # (iface, type)
+
+        for line in iw_out.splitlines():
+            stripped = line.strip()
+            im = re.match(r"^Interface\s+(\S+)$", stripped)
+            if im:
+                if current_iface:
+                    iface_order.append((current_iface, current_type))
+                current_iface = im.group(1)
+                current_type = ""
+                continue
+            tm = re.match(r"^type\s+(\S+)$", stripped)
+            if tm and current_iface:
+                current_type = tm.group(1)
+
+        if current_iface:
+            iface_order.append((current_iface, current_type))
+
+        ap_ifaces = [
+            name for name, itype in iface_order
+            if itype.upper() in ("AP", "AP/VLAN")
+        ]
+
+        if not ap_ifaces:
+            return {}
+
+        # Step 2: run scan on each AP interface
+        passive_flag = " passive" if mode == "passive" else ""
+        result: dict[str, list[Any]] = {}
+
+        for iface in ap_ifaces:
+            scan_cmd = f"iw dev {iface} scan{passive_flag} 2>/dev/null || true"
+            scan_out = self._send_command(scan_cmd)
+            entries = self._parse_iw_scan(scan_out, iface)
+            result[iface] = entries
+
+        return result
+
+    @staticmethod
+    def _parse_iw_scan(scan_out: str, iface: str) -> list[Any]:
+        """Parse ``iw dev <iface> scan`` output into ChannelScanEntryDict list."""
+        from napalm_device_types.models import ChannelScanEntryDict
+
+        entries: list[Any] = []
+        current: dict[str, Any] = {}
+
+        def _freq_to_band(freq: int) -> str:
+            if freq < 3000:
+                return "2.4GHz"
+            if freq < 5950:
+                return "5GHz"
+            return "6GHz"
+
+        def _freq_to_channel(freq: int) -> int:
+            if 2412 <= freq <= 2484:
+                if freq == 2484:
+                    return 14
+                return (freq - 2407) // 5
+            if 5160 <= freq <= 5885:
+                return (freq - 5000) // 5
+            if 5955 <= freq <= 7115:
+                return (freq - 5950) // 5 + 1
+            return 0
+
+        def _flush() -> None:
+            if current.get("bssid"):
+                freq = current.get("frequency", 0)
+                ch = current.get("channel") or _freq_to_channel(freq)
+                width = current.get("channel_width", 0)
+                # If width is still 0 try to derive from HT/VHT/HE info
+                if width == 0 and current.get("_ht_width"):
+                    width = current["_ht_width"]
+                entries.append(ChannelScanEntryDict(
+                    bssid=current["bssid"],
+                    ssid=current.get("ssid", ""),
+                    frequency=freq,
+                    channel=ch,
+                    signal_dbm=current.get("signal_dbm", 0),
+                    channel_width=width,
+                    band=_freq_to_band(freq),
+                ))
+
+        for line in scan_out.splitlines():
+            stripped = line.strip()
+
+            # New BSS block: "BSS aa:bb:cc:dd:ee:ff(on wlan0)"
+            bss_m = re.match(r"^BSS\s+([\da-fA-F:]{17})\(", stripped)
+            if bss_m:
+                _flush()
+                current = {"bssid": bss_m.group(1).lower()}
+                continue
+
+            # freq: 2437
+            freq_m = re.match(r"^freq:\s+(\d+)$", stripped)
+            if freq_m:
+                current["frequency"] = int(freq_m.group(1))
+                continue
+
+            # signal: -72.00 dBm
+            sig_m = re.match(r"^signal:\s+([-\d.]+)\s+dBm", stripped)
+            if sig_m:
+                current["signal_dbm"] = int(float(sig_m.group(1)))
+                continue
+
+            # SSID: NeighborNet  (may be empty for hidden)
+            ssid_m = re.match(r"^SSID:\s*(.*)?$", stripped)
+            if ssid_m:
+                current["ssid"] = ssid_m.group(1).strip()
+                continue
+
+            # HT operation primary channel: 6
+            ht_ch_m = re.match(r"^\*\s+primary channel:\s+(\d+)", stripped)
+            if ht_ch_m:
+                current["channel"] = int(ht_ch_m.group(1))
+                continue
+
+            # HT STA channel width: 20 MHz
+            ht_w_m = re.match(r"^\*\s+STA channel width:\s+(\d+)\s+MHz", stripped)
+            if ht_w_m:
+                current["_ht_width"] = int(ht_w_m.group(1))
+                continue
+
+            # VHT channel width: 80 MHz
+            vht_w_m = re.match(r"^\*\s+channel width:\s+(\d+)\s+MHz", stripped)
+            if vht_w_m:
+                current["channel_width"] = int(vht_w_m.group(1))
+                continue
+
+        _flush()
+        return entries
+
+    def push_radio_channel(self, radio: str, channel: int) -> None:
+        """Set a new channel on *radio* via UCI and restart the wireless stack.
+
+        Uses ``wifi`` (full down/up cycle) rather than ``wifi reload``.
+        ``wifi reload`` only re-applies the running config without changing
+        the physical channel on many hardware/driver combinations; a full
+        restart is required for channel changes to take effect.
+
+        :param radio: UCI radio name, e.g. ``"radio0"``.
+        :param channel: Channel number to set; ``0`` sets UCI ``auto``.
+        """
+        ch_val = "auto" if channel == 0 else str(channel)
+        self._send_command(f"uci set wireless.{radio}.channel={ch_val}")
+        self._send_command("uci commit wireless")
+        self._send_command("wifi")
+
+    def get_radio_status(self) -> dict[str, Any]:
+        """Return radio status from UCI and iwinfo.
+
+        Combines ``uci show wireless`` for static config with ``iwinfo``
+        output for runtime channel/frequency and tx-power data.
+
+        Returns a dict keyed by radio name (e.g. ``"radio0"``) with:
+
+        * enabled (bool)
+        * band (str) — ``"2.4GHz"``, ``"5GHz"``, ``"6GHz"``
+        * channel (int) — 0 means auto
+        * channel_width (int) — channel bandwidth in MHz (0 if unknown)
+        * tx_power (int) — TX power in dBm (0 if unknown)
+        * frequency (float) — centre frequency in MHz (0 if unknown)
+        * htmode (str) — e.g. ``"HT20"``, ``"VHT80"``, ``"HE80"``
+        * country (str) — regulatory country code, e.g. ``"DE"``
+        """
+        from napalm_device_types.models import RadioStatusDict
+
+        uci_out = self._send_command("uci show wireless")
+        radios: dict[str, dict[str, str]] = {}
+
+        for line in uci_out.splitlines():
+            # wifi-device section: wireless.radio0.band='2g'
+            m = re.match(r"wireless\.(radio\d+)\.(\w+)='([^']*)'", line.strip())
+            if m:
+                radio, key, value = m.group(1), m.group(2), m.group(3)
+                radios.setdefault(radio, {})[key] = value
+
+        result: dict[str, Any] = {}
+        for radio, cfg in sorted(radios.items()):
+            band_raw = cfg.get("band", cfg.get("hwmode", ""))
+            # Normalise band: '2g'/'11g' → '2.4GHz', '5g'/'11a' → '5GHz', '6g' → '6GHz'
+            if band_raw in ("2g", "11g", "b", "g", "bg", "bgn"):
+                band = "2.4GHz"
+            elif band_raw in ("5g", "11a", "a", "ac", "ax5"):
+                band = "5GHz"
+            elif band_raw in ("6g", "ax6"):
+                band = "6GHz"
+            else:
+                band = band_raw or "unknown"
+
+            try:
+                channel = int(cfg.get("channel", 0))
+            except (ValueError, TypeError):
+                channel = 0  # 'auto'
+
+            try:
+                tx_power = int(cfg.get("txpower", 0))
+            except (ValueError, TypeError):
+                tx_power = 0
+
+            disabled = cfg.get("disabled", "0") == "1"
+            htmode = cfg.get("htmode", "")
+            country = cfg.get("country", "")
+
+            # Derive channel_width from htmode string (e.g. VHT80 → 80 MHz)
+            _HTMODE_WIDTH = {
+                "HT20": 20, "HT40": 40,
+                "VHT20": 20, "VHT40": 40, "VHT80": 80, "VHT80+80": 80, "VHT160": 160,
+                "HE20": 20, "HE40": 40, "HE80": 80, "HE160": 160,
+                "EHT20": 20, "EHT40": 40, "EHT80": 80, "EHT160": 160, "EHT320": 320,
+            }
+            channel_width = _HTMODE_WIDTH.get(htmode.upper(), 0)
+
+            result[radio] = {
+                **RadioStatusDict(
+                    enabled=not disabled,
+                    band=band,
+                    channel=channel,
+                    channel_width=channel_width,
+                    tx_power=tx_power,
+                    frequency=0.0,    # enriched below via iwinfo
+                ),
+                "htmode": htmode,
+                "country": country,
+            }
+
+        # Enrich with iwinfo runtime data (channel, frequency, tx_power, channel_width)
+        # iwinfo groups output per interface; we need to map interface → radio.
+        # "phy0-ap0  ESSID: "MyNet"" → radio0
+        # "          Tx-Power: 23 dBm"
+        # "          Channel: 44 (5.220 GHz), Width: 80 MHz"
+        try:
+            iwinfo_out = self._send_command("iwinfo 2>/dev/null || true")
+        except Exception:
+            iwinfo_out = ""
+
+        current_radio: str | None = None
+        for line in iwinfo_out.splitlines():
+            # Interface header line: "phy0-ap0  ESSID: ..."
+            iface_m = re.match(r"^(\S+)\s+ESSID:", line)
+            if iface_m:
+                iface_name = iface_m.group(1)
+                phy_m = re.match(r"^phy(\d+)", iface_name)
+                if phy_m:
+                    current_radio = f"radio{phy_m.group(1)}"
+                else:
+                    current_radio = None
+                continue
+
+            if current_radio is None or current_radio not in result:
+                continue
+
+            # Channel and frequency: "Channel: 44 (5.220 GHz), Width: 80 MHz"
+            ch_m = re.search(r"Channel:\s+(\d+)\s+\(([\d.]+)\s+GHz\)", line)
+            if ch_m:
+                result[current_radio]["channel"] = int(ch_m.group(1))
+                result[current_radio]["frequency"] = float(ch_m.group(2)) * 1000
+
+            # Width (MHz): "Width: 80 MHz" or ", Width: 80 MHz"
+            width_m = re.search(r"Width:\s+(\d+)\s+MHz", line)
+            if width_m:
+                result[current_radio]["channel_width"] = int(width_m.group(1))
+
+            # Tx-Power: "Tx-Power: 23 dBm"
+            pwr_m = re.search(r"Tx-Power:\s+(\d+)\s+dBm", line)
+            if pwr_m:
+                result[current_radio]["tx_power"] = int(pwr_m.group(1))
+
+        return result

tests/unit/test_driver.py

@@ -833,3 +833,269 @@ class TestGetNetworkInstances:
         )
         for inst in driver.get_network_instances().values():
             assert set(inst.keys()) >= {"name", "type", "state", "interfaces"}
+
+
+# ---------------------------------------------------------------------------
+# Sample data for get_channel_scan()
+# ---------------------------------------------------------------------------
+
+IW_DEV_OUT = """\
+phy#0
+\tInterface wlan0
+\t\tifindex 4
+\t\twdev 0x1
+\t\taddr aa:bb:cc:dd:ee:f0
+\t\tssid HomeNet
+\t\ttype AP
+\t\tchannel 6 (2437 MHz), width: 20 MHz, center1: 2437 MHz
+phy#1
+\tInterface wlan1
+\t\tifindex 5
+\t\twdev 0x2
+\t\taddr aa:bb:cc:dd:ee:f1
+\t\tssid HomeNet
+\t\ttype AP
+\t\tchannel 36 (5180 MHz), width: 80 MHz, center1: 5210 MHz
+\tInterface wlan2
+\t\tifindex 6
+\t\twdev 0x3
+\t\taddr aa:bb:cc:dd:ee:f2
+\t\tssid GuestNet
+\t\ttype managed
+"""
+
+# Minimal iw scan output for wlan0 (2.4 GHz AP, ch 6 + ch 1)
+IW_SCAN_WLAN0 = """\
+BSS aa:bb:cc:dd:01:ff(on wlan0)
+\tfreq: 2437
+\tsignal: -72.00 dBm
+\tSSID: NeighborNet1
+\tHT operation:
+\t\t * primary channel: 6
+\t\t * secondary channel offset: no secondary
+\t\t * STA channel width: 20 MHz
+BSS aa:bb:cc:dd:02:ff(on wlan0)
+\tfreq: 2412
+\tsignal: -85.00 dBm
+\tSSID: FarAway
+\tHT operation:
+\t\t * primary channel: 1
+\t\t * secondary channel offset: no secondary
+\t\t * STA channel width: 20 MHz
+BSS aa:bb:cc:dd:03:ff(on wlan0)
+\tfreq: 2452
+\tsignal: -90.00 dBm
+\tSSID: Hidden
+\tHT operation:
+\t\t * primary channel: 9
+\t\t * secondary channel offset: no secondary
+\t\t * STA channel width: 20 MHz
+"""
+
+# Minimal iw scan output for wlan1 (5 GHz AP, ch 36 + 44)
+IW_SCAN_WLAN1 = """\
+BSS 11:22:33:44:01:ff(on wlan1)
+\tfreq: 5180
+\tsignal: -58.00 dBm
+\tSSID: Corp5G
+\tVHT operation:
+\t\t * channel width: 80 MHz
+\t\t * center freq segment 1: 42
+\t\t * center freq segment 2: 0
+BSS 11:22:33:44:02:ff(on wlan1)
+\tfreq: 5220
+\tsignal: -77.00 dBm
+\tSSID: Office5G
+\tVHT operation:
+\t\t * channel width: 80 MHz
+"""
+
+
+class TestGetChannelScan:
+    """Tests for OpenWrtWirelessMixin.get_channel_scan()."""
+
+    def _make_send(self, mode: str = "active"):
+        """Return a _send_command mock routing iw dev / iw scan output."""
+        def _send(cmd, **kw):
+            if cmd.strip().startswith("iw dev") and "scan" not in cmd:
+                return IW_DEV_OUT
+            if "wlan0" in cmd and "scan" in cmd:
+                return IW_SCAN_WLAN0
+            if "wlan1" in cmd and "scan" in cmd:
+                return IW_SCAN_WLAN1
+            return ""
+        return _send
+
+    def test_returns_dict(self, driver):
+        driver._send_command = self._make_send()
+        result = driver.get_channel_scan()
+        assert isinstance(result, dict)
+
+    def test_only_ap_interfaces_included(self, driver):
+        """wlan2 is managed (client) mode — must not appear in results."""
+        driver._send_command = self._make_send()
+        result = driver.get_channel_scan()
+        assert "wlan0" in result
+        assert "wlan1" in result
+        assert "wlan2" not in result
+
+    def test_wlan0_entry_count(self, driver):
+        driver._send_command = self._make_send()
+        result = driver.get_channel_scan()
+        assert len(result["wlan0"]) == 3
+
+    def test_wlan0_entry_required_keys(self, driver):
+        driver._send_command = self._make_send()
+        entry = driver.get_channel_scan()["wlan0"][0]
+        assert set(entry.keys()) >= {
+            "bssid", "ssid", "frequency", "channel", "signal_dbm",
+            "channel_width", "band",
+        }
+
+    def test_wlan0_first_entry_values(self, driver):
+        driver._send_command = self._make_send()
+        entry = driver.get_channel_scan()["wlan0"][0]
+        assert entry["bssid"] == "aa:bb:cc:dd:01:ff"
+        assert entry["ssid"] == "NeighborNet1"
+        assert entry["frequency"] == 2437
+        assert entry["channel"] == 6
+        assert entry["signal_dbm"] == -72
+        assert entry["channel_width"] == 20
+        assert entry["band"] == "2.4GHz"
+
+    def test_wlan0_ch1_entry(self, driver):
+        driver._send_command = self._make_send()
+        entries = driver.get_channel_scan()["wlan0"]
+        ch1 = next(e for e in entries if e["channel"] == 1)
+        assert ch1["signal_dbm"] == -85
+        assert ch1["band"] == "2.4GHz"
+
+    def test_wlan0_hidden_ssid(self, driver):
+        """Empty SSID (hidden) should be stored as empty string."""
+        driver._send_command = self._make_send()
+        entries = driver.get_channel_scan()["wlan0"]
+        ch9 = next(e for e in entries if e["channel"] == 9)
+        assert ch9["ssid"] == "Hidden"
+
+    def test_wlan1_band_is_5ghz(self, driver):
+        driver._send_command = self._make_send()
+        entries = driver.get_channel_scan()["wlan1"]
+        assert all(e["band"] == "5GHz" for e in entries)
+
+    def test_wlan1_80mhz_width(self, driver):
+        driver._send_command = self._make_send()
+        entry = driver.get_channel_scan()["wlan1"][0]
+        assert entry["channel_width"] == 80
+
+    def test_passive_mode_uses_passive_flag(self, driver):
+        """With mode='passive', the scan command must contain 'passive'."""
+        issued_cmds: list[str] = []
+
+        def _send(cmd, **kw):
+            issued_cmds.append(cmd)
+            if cmd.strip().startswith("iw dev") and "scan" not in cmd:
+                return IW_DEV_OUT
+            return ""
+
+        driver._send_command = _send
+        driver.get_channel_scan(mode="passive")
+        scan_cmds = [c for c in issued_cmds if "scan" in c]
+        assert all("passive" in c for c in scan_cmds)
+
+    def test_active_mode_no_passive_flag(self, driver):
+        """With mode='active' (default), scan command must NOT contain 'passive'."""
+        issued_cmds: list[str] = []
+
+        def _send(cmd, **kw):
+            issued_cmds.append(cmd)
+            if cmd.strip().startswith("iw dev") and "scan" not in cmd:
+                return IW_DEV_OUT
+            return ""
+
+        driver._send_command = _send
+        driver.get_channel_scan(mode="active")
+        scan_cmds = [c for c in issued_cmds if "scan" in c]
+        assert all("passive" not in c for c in scan_cmds)
+
+    def test_no_ap_interfaces_returns_empty(self, driver):
+        """If no AP-mode interfaces exist, return empty dict."""
+        no_ap = IW_DEV_OUT.replace("type AP", "type managed")
+        driver._send_command = lambda cmd, **kw: (
+            no_ap if "iw dev" in cmd and "scan" not in cmd else ""
+        )
+        result = driver.get_channel_scan()
+        assert result == {}
+
+    def test_empty_iw_output_returns_empty(self, driver):
+        driver._send_command = lambda cmd, **kw: ""
+        result = driver.get_channel_scan()
+        assert result == {}
+
+    def test_6ghz_band_detected(self, driver):
+        iw_dev_6g = """\
+phy#0
+\tInterface wlan0
+\t\taddr aa:bb:cc:dd:ee:f0
+\t\tssid Net6G
+\t\ttype AP
+"""
+        scan_6g = """\
+BSS aa:bb:cc:dd:01:ff(on wlan0)
+\tfreq: 5975
+\tsignal: -65.00 dBm
+\tSSID: Neighbor6G
+"""
+
+        def _send(cmd, **kw):
+            if "iw dev" in cmd and "scan" not in cmd:
+                return iw_dev_6g
+            if "scan" in cmd:
+                return scan_6g
+            return ""
+
+        driver._send_command = _send
+        entries = driver.get_channel_scan().get("wlan0", [])
+        assert len(entries) == 1
+        assert entries[0]["band"] == "6GHz"
+        assert entries[0]["frequency"] == 5975
+
+
+class TestPushRadioChannel:
+    """Tests for OpenWrtWirelessMixin.push_radio_channel()."""
+
+    def test_issues_uci_set(self, driver):
+        """Must call 'uci set wireless.radio0.channel=11'."""
+        issued: list[str] = []
+        driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
+        driver.push_radio_channel("radio0", 11)
+        assert any("uci set wireless.radio0.channel=11" in c for c in issued)
+
+    def test_issues_uci_commit(self, driver):
+        issued: list[str] = []
+        driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
+        driver.push_radio_channel("radio0", 11)
+        assert any("uci commit wireless" in c for c in issued)
+
+    def test_issues_wifi_restart(self, driver):
+        """Must call 'wifi' (full restart) — NOT just 'wifi reload'."""
+        issued: list[str] = []
+        driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
+        driver.push_radio_channel("radio1", 36)
+        # 'wifi' alone is a full restart; 'wifi reload' is insufficient for channel changes
+        assert any(c.strip() == "wifi" for c in issued)
+
+    def test_auto_channel_zero(self, driver):
+        """channel=0 should set 'auto' in UCI."""
+        issued: list[str] = []
+        driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
+        driver.push_radio_channel("radio0", 0)
+        assert any("channel=auto" in c or "channel='auto'" in c for c in issued)
+
+    def test_commit_before_restart(self, driver):
+        """UCI commit must happen before wifi restart."""
+        issued: list[str] = []
+        driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
+        driver.push_radio_channel("radio0", 6)
+        commit_idx = next(i for i, c in enumerate(issued) if "commit" in c)
+        wifi_idx = next(i for i, c in enumerate(issued) if c.strip() == "wifi")
+        assert commit_idx < wifi_idx