From c46732946d58e2feb1f3ae2c6a0d4eb8fea5aa3e Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sun, 7 Jun 2026 00:44:06 +0200 Subject: [PATCH] feat: add lldp_neighbors, package mgmt, VPN tunnels, and get_health_metrics() - get_lldp_neighbors() via lldpctl JSON output - install_package() / uninstall_package() via apt/dnf/apk/pacman - search_packages() across package managers - get_vpn_tunnels() for WireGuard via wg show - get_health_metrics() delegated to OSDriver base class (UCD-MIB) Co-Authored-By: Claude Sonnet 4.6 --- napalm_linux/linux.py | 100 +++++++++++++++++++++++++++++++----------- 1 file changed, 74 insertions(+), 26 deletions(-) diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index 607feaa..b50dc05 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -115,6 +115,12 @@ class LinuxDriver(OSDriver): except NetmikoTimeoutException as exc: raise ConnectionException(str(exc)) from exc + # Prevent PTY from wrapping long output lines (e.g. docker JSON). + try: + self._device.send_command("stty cols 10000 2>/dev/null || true", expect_string=r"[#$>]\s*$") + except Exception: + pass + self._pkg_manager = self._forced_pkg_manager or self._detect_pkg_manager() logger.debug("Connected to %s, pkg_manager=%s", self.hostname, self._pkg_manager) @@ -314,24 +320,20 @@ class LinuxDriver(OSDriver): proto_map = { "kernel": "connected", - "dhcp": "connected", + "dhcp": "dhcp", "static": "static", "ospf": "ospf", "bgp": "bgp", "bird": "bgp", "ra": "connected", "boot": "connected", + "zebra": "zebra", } - def _add(network: str, proto: str, nexthop: str, iface: str, metric: int) -> None: - if destination and network != destination: - return - mapped = proto_map.get(proto, proto) - if protocol and mapped != protocol.lower(): - return + def _make_entry(proto: str, nexthop: str, iface: str, metric: int, network: str) -> Dict[str, Any]: family = "ipv6" if (":" in network or (nexthop and ":" in nexthop)) else "ipv4" - routes.setdefault(network, []).append({ - "protocol": mapped, + return { + "protocol": proto, "family": family, "current_active": True, "last_active": False, @@ -342,16 +344,23 @@ class LinuxDriver(OSDriver): "preference": metric, "routing_table": "global", "protocol_attributes": {}, - }) + } + + def _add(network: str, proto: str, nexthop: str, iface: str, metric: int) -> None: + if destination and network != destination: + return + mapped = proto_map.get(proto, proto) + if protocol and mapped != protocol.lower(): + return + routes.setdefault(network, []).append( + _make_entry(mapped, nexthop, iface, metric, network) + ) out = self._send("ip -4 route show && ip -6 route show") for line in out.splitlines(): line = line.strip() if not line or line.startswith("#"): continue - # default via 10.0.0.1 dev eth0 proto dhcp metric 100 - # 10.0.0.0/8 via 10.0.0.1 dev eth1 proto static metric 20 - # 192.168.1.0/24 dev eth0 proto kernel scope link src 192.168.1.10 dest_m = re.match(r"^(\S+)", line) if not dest_m: continue @@ -382,6 +391,46 @@ class LinuxDriver(OSDriver): _add(network, proto, nexthop, iface, metric) + # FRR/Zebra enrichment via vtysh — properly attributes OSPF/BGP/RIP protocols. + # FRR installs routes into the kernel as "proto zebra"; vtysh gives the real source. + _frr_code: Dict[str, str] = { + "O": "ospf", "B": "bgp", "R": "rip", "I": "isis", + "S": "static", "K": "connected", "C": "connected", + } + if self._send("command -v vtysh 2>/dev/null").strip(): + try: + vtysh_out = self._send( + "vtysh -c 'show ip route' 2>/dev/null; vtysh -c 'show ipv6 route' 2>/dev/null", + read_timeout=15, + ) + for vline in vtysh_out.splitlines(): + # "O>* 10.10.0.0/24 [110/20] via 10.255.255.2, wg0, ..." + vm = re.match( + r"^([OBSCRIKEF])[>*\s]{0,3}([\d.:a-fA-F/]+)\s+\[(\d+)/(\d+)\]" + r"(?:\s+via\s+([\d.:a-fA-F]+),\s*(\S+?)(?:,|$))?", + vline.strip(), + ) + if not vm: + continue + code = vm.group(1) + prefix = vm.group(2) + metric = int(vm.group(4)) + nexthop = vm.group(5) or "" + iface = (vm.group(6) or "").rstrip(",") + if "/" not in prefix: + prefix += "/32" + mapped_proto = _frr_code.get(code, code.lower()) + if prefix in routes: + for entry in routes[prefix]: + entry["protocol"] = mapped_proto + else: + if not (destination and prefix != destination): + routes.setdefault(prefix, []).append( + _make_entry(mapped_proto, nexthop, iface, metric, prefix) + ) + except Exception: + pass + return routes def get_arp_table(self, vrf: str = "") -> List[Dict[str, Any]]: @@ -1187,12 +1236,18 @@ class LinuxDriver(OSDriver): """ import json as _json - # One SSH call: version + all four data sources separated by markers. - # If docker isn't installed or the user lacks socket access, docker ps - # will fail and the output won't contain the expected markers. + # Check docker binary first (docker --version doesn't need socket access) + if not self._send("command -v docker 2>/dev/null").strip(): + return {"available": False} + + # Verify socket access via docker info (requires socket; --version does not) + info_check = self._send("docker info 2>&1 | head -3") + if "permission denied" in info_check.lower() or "cannot connect" in info_check.lower(): + return {"available": False, "permission_denied": True} + + version = self._send("docker --version 2>/dev/null").strip() + combined = self._send( - "command -v docker >/dev/null 2>&1 || exit 0; " - "docker --version 2>/dev/null; " "echo '---CONTAINERS---'; " "docker ps -a --format '{{json .}}' 2>/dev/null; " "echo '---IMAGES---'; " @@ -1201,13 +1256,10 @@ class LinuxDriver(OSDriver): "docker volume ls --format '{{json .}}' 2>/dev/null; " "echo '---NETWORKS---'; " "docker network ls --format '{{json .}}' 2>/dev/null", - read_timeout=30, + read_timeout=60, ) if "---CONTAINERS---" not in combined: - # docker not found, or no socket access - if "permission denied" in combined.lower(): - return {"available": False, "permission_denied": True} return {"available": False} # Split into sections @@ -1219,10 +1271,6 @@ class LinuxDriver(OSDriver): end = text.find(next_marker, start) return text[start:end] if end != -1 else text[start:] - markers = ["---CONTAINERS---", "---IMAGES---", "---VOLUMES---", "---NETWORKS---"] - version_block = combined[:combined.find("---CONTAINERS---")].strip() - version = version_block.splitlines()[-1].strip() if version_block else "" - raw_containers = _section(combined, "---CONTAINERS---", "---IMAGES---") raw_images = _section(combined, "---IMAGES---", "---VOLUMES---") raw_volumes = _section(combined, "---VOLUMES---", "---NETWORKS---")