diff --git a/napalm_openwrt/config_mixin.py b/napalm_openwrt/config_mixin.py new file mode 100644 index 0000000..e671ba5 --- /dev/null +++ b/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 diff --git a/napalm_openwrt/interfaces_mixin.py b/napalm_openwrt/interfaces_mixin.py new file mode 100644 index 0000000..81b4966 --- /dev/null +++ b/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: 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: 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 diff --git a/napalm_openwrt/lldp_mixin.py b/napalm_openwrt/lldp_mixin.py new file mode 100644 index 0000000..3da4351 --- /dev/null +++ b/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..* + 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', '', , , ...] + 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 diff --git a/napalm_openwrt/packages_mixin.py b/napalm_openwrt/packages_mixin.py new file mode 100644 index 0000000..2281049 --- /dev/null +++ b/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} diff --git a/napalm_openwrt/routing_mixin.py b/napalm_openwrt/routing_mixin.py new file mode 100644 index 0000000..ea027e7 --- /dev/null +++ b/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 -W -s [-t ] [-I ] + + Returns ``{'success': {...}}`` or ``{'error': ''}``. + """ + 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 `` 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 -w `` (BusyBox-compatible). + Falls back to ``traceroute6`` for IPv6 destinations. + + Returns ``{'success': {hop: {'probes': {probe: {rtt, ip_address, host_name}}}}}`` + or ``{'error': ''}``. + """ + # 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} diff --git a/napalm_openwrt/vlan_mixin.py b/napalm_openwrt/vlan_mixin.py new file mode 100644 index 0000000..3a3e08f --- /dev/null +++ b/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" + ) diff --git a/napalm_openwrt/wireless_mixin.py b/napalm_openwrt/wireless_mixin.py new file mode 100644 index 0000000..e0dec18 --- /dev/null +++ b/napalm_openwrt/wireless_mixin.py @@ -0,0 +1,503 @@ +# -*- 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 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_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