From 2712389818ef80ea8ca070e024a821d63e1ac106 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sat, 6 Jun 2026 15:50:02 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20add=20get=5Froute=5Fto()=20via=20ip=20r?= =?UTF-8?q?oute=20show;=20family-Feld=20f=C3=BCr=20IPv4/IPv6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Initial commit mit bestehendem Code inkl. neuem get_route_to(): - Parsed ip -4 route show und ip -6 route show - Protokoll-Map: kernel/dhcp/ra/boot→connected, static→static, ospf→ospf, bgp→bgp - family-Feld aus Netzadresse oder Next-Hop (: = ipv6) - default/default6 → 0.0.0.0/0 / ::/0; Host-Routen ohne Prefix bekommen /32 Co-Authored-By: Claude Sonnet 4.6 --- .gitignore | 44 + CHANGELOG.md | 28 + README.md | 169 ++++ napalm_linux/__init__.py | 5 + napalm_linux/linux.py | 1847 ++++++++++++++++++++++++++++++++++++++ pyproject.toml | 65 ++ tests/__init__.py | 1 + tests/test_linux.py | 341 +++++++ 8 files changed, 2500 insertions(+) create mode 100644 .gitignore create mode 100644 CHANGELOG.md create mode 100644 README.md create mode 100644 napalm_linux/__init__.py create mode 100644 napalm_linux/linux.py create mode 100644 pyproject.toml create mode 100644 tests/__init__.py create mode 100644 tests/test_linux.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..1d5fe4c --- /dev/null +++ b/.gitignore @@ -0,0 +1,44 @@ +# Python +__pycache__/ +*.py[cod] +*.pyo +*.pyd +*.so +*.egg +*.egg-info/ +dist/ +build/ +.eggs/ +wheels/ + +# Virtual environments +.venv/ +venv/ +env/ +.env + +# Packaging +*.tar.gz +*.whl +MANIFEST + +# Testing +.pytest_cache/ +.coverage +.coverage.* +htmlcov/ +coverage.xml + +# Type checking +.mypy_cache/ +.ruff_cache/ + +# IDEs +.vscode/ +.idea/ +*.swp +*~ + +# OS +.DS_Store +Thumbs.db diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..68a1932 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,28 @@ +# Changelog + +All notable changes to this project will be documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## [Unreleased] + +## [0.1.0] – 2026-05-29 + +### Added + +- Initial release +- `LinuxDriver` inheriting from `napalm_device_types.OSDriver` +- SSH transport via Netmiko `linux` device type +- Auto-detection of package manager: `apt`, `dnf`, `yum`, `apk`, `pacman` +- Standard NAPALM methods: `get_facts()`, `get_interfaces()`, `get_interfaces_ip()`, + `get_arp_table()`, `get_config()`, `ping()` +- OS-specific methods: `get_packages()`, `get_pending_updates()`, `apply_updates()`, + `get_services()`, `get_users()`, `get_processes()`, `get_cron_jobs()` +- `apply_updates()` validates all package names against `^[a-zA-Z0-9_\-\+\.]+$` + to prevent shell injection before passing them to the package manager +- SysV fallback for `get_services()` on non-systemd systems +- Unit tests with mocked SSH connections + +[Unreleased]: https://github.com/chrismanivong/napalm-linux/compare/v0.1.0...HEAD +[0.1.0]: https://github.com/chrismanivong/napalm-linux/releases/tag/v0.1.0 diff --git a/README.md b/README.md new file mode 100644 index 0000000..e3fbd61 --- /dev/null +++ b/README.md @@ -0,0 +1,169 @@ +# napalm-linux + +NAPALM driver for **generic Linux systems** — Debian, Ubuntu, RHEL, Rocky, Alpine, Arch and any +other distribution reachable via SSH. + +Connects over SSH using [Netmiko](https://github.com/ktbyers/netmiko) (`linux` device type) and +**automatically detects** the installed package manager (`apt`, `dnf`, `yum`, `apk`, `pacman`). + +## Requirements + +| Dependency | Minimum version | +|---|---| +| Python | 3.9 | +| NAPALM | 4.0 | +| Netmiko | 4.0 | +| napalm-device-types | 0.2.0 | + +## Installation + +```bash +pip install napalm napalm-linux +``` + +Or from source: + +```bash +git clone https://github.com/chrismanivong/napalm-linux +pip install -e napalm-linux/ +``` + +When working in the NetOrk monorepo, install both packages as editable: + +```bash +pip install -e vendor/napalm-device-types/ -e vendor/napalm-linux/ +``` + +## Quick start + +```python +from napalm import get_network_driver + +Driver = get_network_driver("linux") + +with Driver( + "10.0.0.5", + "admin", + "s3cr3t", + optional_args={ + # "port": 22, + # "pkg_manager": "apt", # force package manager; auto-detected by default + # "secret": "sudo-pass", # password for sudo / enable (defaults to login password) + # "debugging": True, # enable verbose logging + }, +) as dev: + facts = dev.get_facts() + print(facts) + + # OS-specific methods (from napalm-device-types OSDriver) + packages = dev.get_packages() + updates = dev.get_pending_updates() + services = dev.get_services() + users = dev.get_users() + procs = dev.get_processes() + jobs = dev.get_cron_jobs() + + # Upgrade specific packages + result = dev.apply_updates(["openssh-server", "curl"]) + print(result) # {"success": True, "output": "..."} + + # Upgrade everything with pending updates + result = dev.apply_updates([]) +``` + +## Supported NAPALM methods + +### Standard NAPALM + +| Method | Supported | Notes | +|---|---|---| +| `open()` / `close()` | ✅ | SSH via netmiko `linux` | +| `is_alive()` | ✅ | | +| `get_facts()` | ✅ | DMI / `/proc/uptime` / `ip link` | +| `get_interfaces()` | ✅ | `ip link show` | +| `get_interfaces_ip()` | ✅ | `ip addr show` | +| `get_arp_table()` | ✅ | `ip neigh show` | +| `get_config()` | ✅ | Returns `ip addr` + `ip route` output | +| `ping()` | ✅ | Executes `ping` on the remote host | +| `load_merge_candidate()` | ❌ | Not applicable for generic Linux | +| `load_replace_candidate()` | ❌ | Not applicable for generic Linux | +| `compare_config()` | ❌ | Not applicable for generic Linux | +| `commit_config()` | ❌ | Not applicable for generic Linux | +| `discard_config()` | ❌ | Not applicable for generic Linux | +| `rollback()` | ❌ | Not applicable for generic Linux | + +### OSDriver extensions (napalm-device-types) + +| Method | Supported | Package managers | +|---|---|---| +| `get_packages()` | ✅ | apt, dnf, yum, apk, pacman | +| `get_pending_updates()` | ✅ | apt, dnf, yum, apk, pacman | +| `apply_updates(packages)` | ✅ | apt, dnf, yum, apk, pacman | +| `get_services()` | ✅ | systemd (fallback: SysV `service`) | +| `get_users()` | ✅ | `/etc/passwd` + `/etc/group` | +| `get_processes()` | ✅ | `ps axo` | +| `get_cron_jobs()` | ✅ | user crontabs + `/etc/cron.d/` | + +## Package manager auto-detection + +The driver probes for each binary in order via `command -v`: + +``` +apt → dnf → yum → apk → pacman +``` + +Force a specific package manager: + +```python +optional_args={"pkg_manager": "dnf"} +``` + +## SSH user permissions + +The SSH user needs read access to: + +| Data | Required permission | +|---|---| +| `/etc/passwd`, `/etc/group` | world-readable (default) | +| `/proc/uptime`, `/sys/class/dmi/…` | world-readable (default) | +| User crontabs (`/var/spool/cron/…`) | `root` or `sudo` required | +| `systemctl is-enabled ` | unprivileged on most distros | +| `apt list --upgradable` | may require `apt-get update` (root) | +| `dnf check-update` / `yum check-update` | unprivileged, but slower without cache | + +For full functionality it is recommended to run as `root` or grant passwordless `sudo` for +the above commands. + +## Tested distributions + +| Distribution | Version | Package manager | Tested | +|---|---|---|---| +| Debian | 12 (Bookworm) | apt | ✅ | +| Ubuntu | 22.04 LTS | apt | ✅ | +| Rocky Linux | 9 | dnf | planned | +| Alpine Linux | 3.19 | apk | planned | +| Arch Linux | rolling | pacman | planned | + +Contributions for additional distributions and versions are welcome. + +## Development + +```bash +# Create venv +python -m venv .venv +source .venv/bin/activate + +# Install in editable mode with dev dependencies +pip install -e ../napalm-device-types/ -e ".[dev]" + +# Run tests +pytest tests/ -v + +# Lint / format +ruff check napalm_linux/ +ruff format napalm_linux/ +``` + +## License + +Apache 2.0 diff --git a/napalm_linux/__init__.py b/napalm_linux/__init__.py new file mode 100644 index 0000000..d145689 --- /dev/null +++ b/napalm_linux/__init__.py @@ -0,0 +1,5 @@ +"""napalm-linux – NAPALM driver for generic Linux systems.""" + +from napalm_linux.linux import LinuxDriver + +__all__ = ["LinuxDriver"] diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py new file mode 100644 index 0000000..607feaa --- /dev/null +++ b/napalm_linux/linux.py @@ -0,0 +1,1847 @@ +# -*- coding: utf-8 -*- +# Licensed under the Apache License, Version 2.0 + +"""NAPALM driver for generic Linux systems. + +Connects via SSH using netmiko (device_type ``linux``) and supports +auto-detection of the installed package manager: + + * apt — Debian, Ubuntu, Raspberry Pi OS, … + * dnf — RHEL 8+, Rocky Linux, AlmaLinux, Fedora + * yum — RHEL 7, CentOS 7 + * apk — Alpine Linux + * pacman — Arch Linux, Manjaro + +A specific package manager can be forced with +``optional_args={"pkg_manager": "apt"}``. +""" + +import logging +import re +import socket +from shlex import quote as _shlex_quote +from typing import Any, Dict, List, Optional + +from netmiko import ConnectHandler +from netmiko.exceptions import ( + NetmikoAuthenticationException, + NetmikoTimeoutException, +) +from napalm.base.exceptions import ConnectionException, ConnectionClosedException +from napalm.base.netmiko_helpers import netmiko_args +from napalm_device_types import OSDriver +from napalm_device_types.models import ( + ApplyUpdatesResultDict, + CronJobDict, + DeviceActionResultDict, + DockerInfoDict, + PackageDict, + ProcessDict, + ServiceDict, + SNMPConfigDict, + UpdateDict, + UserDict, +) + +logger = logging.getLogger("napalm_linux") + +# Package managers in detection order +_PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"] + + +class LinuxDriver(OSDriver): + """NAPALM driver for generic Linux systems. + + Connects via SSH (netmiko ``linux`` device type) and auto-detects the + package manager unless overridden by ``optional_args["pkg_manager"]``. + """ + + VENDOR = "Linux" + NETMIKO_DEVICE_TYPE = "linux" + + def __init__( + self, + hostname: str, + username: str, + password: str, + timeout: int = 60, + optional_args: Optional[Dict] = None, + ) -> None: + self.hostname = hostname + self.username = username + self.password = password + self.timeout = timeout + + if optional_args is None: + optional_args = {} + + self.port: int = optional_args.get("port", 22) + self._forced_pkg_manager: Optional[str] = optional_args.get("pkg_manager") + self._secret: str = optional_args.get("secret", password) + # Optional sudo password for privilege escalation (e.g. apt-get update) + self._sudo_password: Optional[str] = optional_args.get("sudo_password") + + if optional_args.get("debugging"): + logger.setLevel(logging.DEBUG) + + self.netmiko_optional_args = netmiko_args(optional_args) + # port is passed explicitly in open() — remove it from netmiko_optional_args + # to avoid "multiple values for keyword argument 'port'" + self.netmiko_optional_args.pop("port", None) + + # Runtime state + self._device: Optional[ConnectHandler] = None + self._pkg_manager: Optional[str] = None # set after open() + + # ------------------------------------------------------------------ + # Connection management + # ------------------------------------------------------------------ + + def open(self) -> None: + """Open the SSH connection and detect the package manager.""" + try: + self._device = ConnectHandler( + device_type=self.NETMIKO_DEVICE_TYPE, + host=self.hostname, + username=self.username, + password=self.password, + port=self.port, + secret=self._secret, + timeout=self.timeout, + **self.netmiko_optional_args, + ) + except NetmikoAuthenticationException as exc: + raise ConnectionException(str(exc)) from exc + except NetmikoTimeoutException as exc: + raise ConnectionException(str(exc)) from exc + + 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) + + def close(self) -> None: + """Close the SSH connection.""" + if self._device: + try: + self._device.disconnect() + except Exception: + pass + self._device = None + self._pkg_manager = None + + def is_alive(self) -> Dict[str, bool]: + if self._device: + try: + return {"is_alive": self._device.remote_conn.transport.is_active()} + except (AttributeError, socket.error, EOFError): + return {"is_alive": False} + return {"is_alive": False} + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _send(self, command: str, read_timeout: float = 100) -> str: + """Send a command and return stripped output.""" + if not self._device: + raise ConnectionClosedException("Not connected") + return self._device.send_command( + command, + read_timeout=read_timeout, + cmd_verify=False, + expect_string=r'[#$\>]\s*$', + ).strip() + + def _sudo(self, command: str, read_timeout: float = 100) -> str: + """Run *command* via sudo, feeding the password via stdin (-S). + + Falls back to plain execution when no sudo password is configured. + """ + if self._sudo_password: + wrapped = f'echo {_shlex_quote(self._sudo_password)} | sudo -S -p "" {command}' + return self._send(wrapped, read_timeout=read_timeout) + return self._send(f'sudo {command}', read_timeout=read_timeout) + + def _detect_pkg_manager(self) -> Optional[str]: + """Return the first package manager binary found on PATH.""" + for pm in _PKG_MANAGERS: + result = self._send(f"command -v {pm} 2>/dev/null") + if result: + return pm + return None + + # ------------------------------------------------------------------ + # Standard NAPALM – read-only + # ------------------------------------------------------------------ + + def get_facts(self) -> Dict[str, Any]: + """Return basic system facts.""" + hostname = self._send("hostname -s 2>/dev/null || hostname") + fqdn = self._send("hostname -f 2>/dev/null || hostname") + os_version = self._send( + "cat /etc/os-release 2>/dev/null | grep '^PRETTY_NAME' | cut -d= -f2 | tr -d '\"'" + ) or self._send("uname -r") + kernel = self._send("uname -r") + uptime_secs = self._parse_uptime() + serial = self._send( + "cat /sys/class/dmi/id/product_serial 2>/dev/null || echo ''" + ) + model = self._send( + "cat /sys/class/dmi/id/product_name 2>/dev/null || echo ''" + ) + + # Interface list + iface_out = self._send("ip -o link show | awk -F': ' '{print $2}' | cut -d@ -f1") + interface_list = [i.strip() for i in iface_out.splitlines() if i.strip() and i.strip() != "lo"] + + return { + "hostname": hostname, + "fqdn": fqdn, + "vendor": self.VENDOR, + "model": model, + "serial_number": serial, + "os_version": os_version, + "uptime": uptime_secs, + "interface_list": interface_list, + } + + def _parse_uptime(self) -> int: + """Return uptime in seconds from ``/proc/uptime``.""" + raw = self._send("cat /proc/uptime 2>/dev/null") + try: + return int(float(raw.split()[0])) + except (IndexError, ValueError): + return 0 + + def get_lldp_neighbors(self) -> Dict[str, List[Dict[str, Any]]]: + """Return LLDP neighbors if lldpd is installed and currently running. + + Uses ``lldpctl -f keyvalue``. Returns an empty dict when lldpd is + absent or stopped — does NOT attempt to start the daemon. + """ + # Check lldpctl is available + if not self._send("command -v lldpctl 2>/dev/null").strip(): + return {} + + # Check lldpd is active (systemd or fallback to pgrep) + running = self._send( + "systemctl is-active lldpd 2>/dev/null || " + "service lldpd status 2>/dev/null | grep -q running && echo active || " + "pgrep -x lldpd >/dev/null 2>&1 && echo active || true" + ).strip() + if "active" not in running: + return {} + + output = self._send("lldpctl -f keyvalue 2>/dev/null || true") + neighbors: Dict[str, List[Dict[str, Any]]] = {} + entries: Dict[str, Dict[str, str]] = {} + + for line in output.splitlines(): + line = line.strip() + if "=" not in line: + continue + key, _, value = line.partition("=") + parts = key.split(".") + 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(): + neighbors.setdefault(iface, []).append({ + "hostname": data.get("chassis.name", ""), + "port": data.get("port.ifname", data.get("port.id.value", "")), + }) + + return neighbors + + def get_interfaces(self) -> Dict[str, Any]: + """Return interface operational data.""" + interfaces: Dict[str, Any] = {} + + # ip -o link show: one line per interface + link_out = self._send("ip -o link show") + for line in link_out.splitlines(): + # 2: eth0: mtu 1500 ... state UP + m = re.match(r"^\d+:\s+(\S+?)(?:@\S+)?:\s+<([^>]*)>.*mtu\s+(\d+).*state\s+(\S+)", line) + if not m: + continue + name, flags, mtu, state = m.group(1), m.group(2), int(m.group(3)), m.group(4) + mac_m = re.search(r"link/ether\s+([\da-f:]+)", line) + mac = mac_m.group(1) if mac_m else "" + is_up = "UP" in flags.split(",") or state == "UP" + interfaces[name] = { + "is_up": is_up, + "is_enabled": "UP" in flags.split(","), + "description": "", + "last_flapped": -1.0, + "speed": -1.0, + "mtu": mtu, + "mac_address": mac, + } + + return interfaces + + def get_interfaces_ip(self) -> Dict[str, Any]: + """Return IP addresses per interface.""" + result: Dict[str, Any] = {} + + addr_out = self._send("ip -o addr show") + for line in addr_out.splitlines(): + # 2: eth0 inet 192.168.1.10/24 brd ... + m = re.match(r"^\d+:\s+(\S+)\s+(inet6?)\s+([\da-f.:]+)/(\d+)", line) + if not m: + continue + iface, family, addr, prefix = m.group(1), m.group(2), m.group(3), int(m.group(4)) + af = "ipv4" if family == "inet" else "ipv6" + result.setdefault(iface, {"ipv4": {}, "ipv6": {}}) + result[iface][af][addr] = {"prefix_length": prefix} + + return result + + def get_route_to( + self, + destination: str = "", + protocol: str = "", + longer: bool = False, + ) -> Dict[str, List[Dict[str, Any]]]: + """Return the routing table via ``ip route show``. + + OSPF routes (from FRR/Quagga) are included via ``ip route show proto ospf`` + if any are present. The result is keyed by network prefix. + """ + routes: Dict[str, List[Dict[str, Any]]] = {} + + proto_map = { + "kernel": "connected", + "dhcp": "connected", + "static": "static", + "ospf": "ospf", + "bgp": "bgp", + "bird": "bgp", + "ra": "connected", + "boot": "connected", + } + + 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 + family = "ipv6" if (":" in network or (nexthop and ":" in nexthop)) else "ipv4" + routes.setdefault(network, []).append({ + "protocol": mapped, + "family": family, + "current_active": True, + "last_active": False, + "age": -1, + "next_hop": nexthop, + "outgoing_interface": iface, + "selected_next_hop": True, + "preference": metric, + "routing_table": "global", + "protocol_attributes": {}, + }) + + 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 + raw_dest = dest_m.group(1) + network = "0.0.0.0/0" if raw_dest == "default" else ("::/0" if raw_dest == "default6" else raw_dest) + if "/" not in network: + network += "/32" + + nexthop = "" + nh_m = re.search(r"via\s+(\S+)", line) + if nh_m: + nexthop = nh_m.group(1) + + iface = "" + dev_m = re.search(r"dev\s+(\S+)", line) + if dev_m: + iface = dev_m.group(1) + + proto = "kernel" + proto_m = re.search(r"proto\s+(\S+)", line) + if proto_m: + proto = proto_m.group(1) + + metric = 0 + metric_m = re.search(r"metric\s+(\d+)", line) + if metric_m: + metric = int(metric_m.group(1)) + + _add(network, proto, nexthop, iface, metric) + + return routes + + def get_arp_table(self, vrf: str = "") -> List[Dict[str, Any]]: + """Return the ARP/neighbour table.""" + entries = [] + neigh_out = self._send("ip -4 neigh show") + for line in neigh_out.splitlines(): + # 192.168.1.1 dev eth0 lladdr aa:bb:cc:dd:ee:ff REACHABLE + m = re.match( + r"^([\d.]+)\s+dev\s+(\S+)\s+lladdr\s+([\da-f:]+)\s+(\S+)", line + ) + if not m: + continue + entries.append({ + "interface": m.group(2), + "mac": m.group(3), + "ip": m.group(1), + "age": 0.0, + }) + return entries + + def get_config( + self, retrieve: str = "all", full: bool = False, sanitized: bool = False + ) -> Dict[str, Any]: + """Return minimal config representation (network interfaces only).""" + running = self._send("ip addr show && ip route show") + return {"running": running, "startup": "", "candidate": ""} + + # ------------------------------------------------------------------ + # Config management – not applicable for generic Linux + # ------------------------------------------------------------------ + + def load_merge_candidate(self, filename: str = None, config: str = None) -> None: # type: ignore[override] + raise NotImplementedError("Config management is not supported for Linux hosts") + + def load_replace_candidate(self, filename: str = None, config: str = None) -> None: # type: ignore[override] + raise NotImplementedError("Config management is not supported for Linux hosts") + + def compare_config(self) -> str: + raise NotImplementedError("Config management is not supported for Linux hosts") + + def commit_config(self, message: str = "") -> None: + raise NotImplementedError("Config management is not supported for Linux hosts") + + def discard_config(self) -> None: + raise NotImplementedError("Config management is not supported for Linux hosts") + + def rollback(self) -> None: + raise NotImplementedError("Config management is not supported for Linux hosts") + + # ------------------------------------------------------------------ + # Optional NAPALM methods + # ------------------------------------------------------------------ + + def ping( + self, + destination: str, + source: str = "", + ttl: int = 255, + timeout: int = 2, + size: int = 100, + count: int = 5, + vrf: str = "", + ) -> Dict[str, Any]: + """Execute ping from the remote host.""" + src_opt = f"-I {source}" if source else "" + cmd = f"ping -c {count} -W {timeout} -s {size} -t {ttl} {src_opt} {destination} 2>&1" + output = self._send(cmd) + + # Parse summary line: "5 packets transmitted, 5 received, 0% packet loss" + m = re.search( + r"(\d+) packets transmitted,\s*(\d+) received,\s*([\d.]+)% packet loss", + output, + ) + if not m: + return {"error": output} + + sent, received = int(m.group(1)), int(m.group(2)) + + # Parse rtt line: "rtt min/avg/max/mdev = 0.123/0.456/0.789/0.100 ms" + rtt_m = re.search( + r"rtt .* = ([\d.]+)/([\d.]+)/([\d.]+)/([\d.]+) ms", output + ) + + results = [] + for line in output.splitlines(): + icmp_m = re.search( + r"bytes from ([\d.]+).*icmp_seq=\d+ ttl=(\d+) time=([\d.]+) ms", line + ) + if icmp_m: + results.append({ + "ip_address": icmp_m.group(1), + "rtt": float(icmp_m.group(3)), + }) + + return { + "success": { + "probes_sent": sent, + "packet_loss": sent - received, + "rtt_min": float(rtt_m.group(1)) if rtt_m else 0.0, + "rtt_avg": float(rtt_m.group(2)) if rtt_m else 0.0, + "rtt_max": float(rtt_m.group(3)) if rtt_m else 0.0, + "rtt_stddev": float(rtt_m.group(4)) if rtt_m else 0.0, + "results": results, + } + } + + # ------------------------------------------------------------------ + # OSDriver – package management + # ------------------------------------------------------------------ + + def get_packages(self) -> List[PackageDict]: + if self._pkg_manager == "apt": + return self._get_packages_apt() + if self._pkg_manager in ("dnf", "yum"): + return self._get_packages_rpm() + if self._pkg_manager == "apk": + return self._get_packages_apk() + if self._pkg_manager == "pacman": + return self._get_packages_pacman() + raise NotImplementedError( + f"Package manager '{self._pkg_manager}' is not supported" + ) + + def _get_packages_apt(self) -> List[PackageDict]: + out = self._send( + "dpkg-query -W -f='${Package}\\t${Version}\\t${Installed-Size}\\t${binary:Summary}\\n' 2>/dev/null" + ) + packages: List[PackageDict] = [] + for line in out.splitlines(): + parts = line.split("\t", 3) + if len(parts) < 2: + continue + name = parts[0].strip() + version = parts[1].strip() + size = int(parts[2].strip()) * 1024 if len(parts) > 2 and parts[2].strip().isdigit() else 0 + description = parts[3].strip() if len(parts) > 3 else "" + packages.append({ + "name": name, + "version": version, + "installed": True, + "description": description, + "size": size, + "source": "apt", + }) + return packages + + def _get_packages_rpm(self) -> List[PackageDict]: + out = self._send( + "rpm -qa --queryformat '%{NAME}\\t%{VERSION}-%{RELEASE}\\t%{SIZE}\\t%{SUMMARY}\\n' 2>/dev/null" + ) + packages: List[PackageDict] = [] + for line in out.splitlines(): + parts = line.split("\t", 3) + if len(parts) < 2: + continue + packages.append({ + "name": parts[0].strip(), + "version": parts[1].strip(), + "installed": True, + "description": parts[3].strip() if len(parts) > 3 else "", + "size": int(parts[2].strip()) if len(parts) > 2 and parts[2].strip().isdigit() else 0, + "source": self._pkg_manager or "rpm", + }) + return packages + + def _get_packages_apk(self) -> List[PackageDict]: + out = self._send("apk info -v 2>/dev/null") + packages: List[PackageDict] = [] + for line in out.splitlines(): + # openssh-9.3_p2-r4 OpenSSH + m = re.match(r"^(\S+)-(\d[\S]*)\s*(.*)", line) + if not m: + continue + packages.append({ + "name": m.group(1), + "version": m.group(2), + "installed": True, + "description": m.group(3).strip(), + "size": 0, + "source": "apk", + }) + return packages + + def _get_packages_pacman(self) -> List[PackageDict]: + out = self._send("pacman -Q 2>/dev/null") + packages: List[PackageDict] = [] + for line in out.splitlines(): + parts = line.split(None, 1) + if len(parts) < 2: + continue + packages.append({ + "name": parts[0], + "version": parts[1], + "installed": True, + "description": "", + "size": 0, + "source": "pacman", + }) + return packages + + def search_packages(self, query: str) -> List[Dict[str, Any]]: + """Search available (installable) packages matching *query*.""" + from shlex import quote as _q + safe_q = _q(query) + installed = {p["name"] for p in self.get_packages()} + packages: List[Dict[str, Any]] = [] + + if self._pkg_manager == "apt": + out = self._send(f"apt-cache search {safe_q} 2>/dev/null") + versions: Dict[str, str] = {} + ver_out = self._send(f"apt-cache show {safe_q} 2>/dev/null | grep -E '^(Package|Version):' || true") + cur_pkg = "" + for line in ver_out.splitlines(): + if line.startswith("Package:"): + cur_pkg = line.split(":", 1)[1].strip() + elif line.startswith("Version:") and cur_pkg: + versions[cur_pkg] = line.split(":", 1)[1].strip() + for line in out.splitlines(): + if " - " not in line: + continue + name, _, description = line.partition(" - ") + name = name.strip() + packages.append({ + "name": name, + "version": versions.get(name, ""), + "installed": name in installed, + "description": description.strip(), + "size": 0, + "source": "apt", + }) + + elif self._pkg_manager in ("dnf", "yum"): + cmd = "dnf" if self._pkg_manager == "dnf" else "yum" + out = self._send(f"{cmd} search {safe_q} 2>/dev/null || true") + for line in out.splitlines(): + if " : " not in line: + continue + pkg_ver, _, description = line.partition(" : ") + name = pkg_ver.split(".")[0].strip() + version = "" + packages.append({ + "name": name, + "version": version, + "installed": name in installed, + "description": description.strip(), + "size": 0, + "source": self._pkg_manager or "rpm", + }) + + elif self._pkg_manager == "apk": + out = self._send(f"apk search {safe_q} 2>/dev/null") + 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": "apk", + }) + + elif self._pkg_manager == "pacman": + out = self._send(f"pacman -Ss {safe_q} 2>/dev/null || true") + lines = out.splitlines() + i = 0 + while i < len(lines): + line = lines[i].strip() + if "/" in line and " " in line: + parts = line.split() + name_ver = parts[0].split("/")[-1] if "/" in parts[0] else parts[0] + name_parts = name_ver.rsplit(" ", 1) + name = name_parts[0] + version = parts[1] if len(parts) > 1 else "" + description = lines[i + 1].strip() if i + 1 < len(lines) else "" + packages.append({ + "name": name, + "version": version, + "installed": name in installed, + "description": description, + "size": 0, + "source": "pacman", + }) + i += 2 + continue + i += 1 + + return packages + + def install_package(self, name: str) -> Dict[str, Any]: + """Install a package by name. Returns ``{"success": bool, "output": str}``.""" + from shlex import quote as _q + safe = _q(name) + pm = self._pkg_manager + if pm == "apt": + raw = self._sudo(f"DEBIAN_FRONTEND=noninteractive apt-get install -y {safe} 2>&1 || true") + elif pm in ("dnf", "yum"): + raw = self._sudo(f"{pm} install -y {safe} 2>&1 || true") + elif pm == "apk": + raw = self._sudo(f"apk add {safe} 2>&1 || true") + elif pm == "pacman": + raw = self._sudo(f"pacman -S --noconfirm {safe} 2>&1 || true") + else: + return {"success": False, "output": f"Unsupported package manager: {pm}"} + low = raw.lower() + success = not any(kw in low for kw in ("error:", "failed", "no packages", "not found", "unable to locate", "no match")) + return {"success": success, "output": raw.strip()} + + def uninstall_package(self, name: str) -> Dict[str, Any]: + """Remove a package by name. Returns ``{"success": bool, "output": str}``.""" + from shlex import quote as _q + safe = _q(name) + pm = self._pkg_manager + if pm == "apt": + raw = self._sudo(f"DEBIAN_FRONTEND=noninteractive apt-get remove -y {safe} 2>&1 || true") + elif pm in ("dnf", "yum"): + raw = self._sudo(f"{pm} remove -y {safe} 2>&1 || true") + elif pm == "apk": + raw = self._sudo(f"apk del {safe} 2>&1 || true") + elif pm == "pacman": + raw = self._sudo(f"pacman -R --noconfirm {safe} 2>&1 || true") + else: + return {"success": False, "output": f"Unsupported package manager: {pm}"} + low = raw.lower() + success = not any(kw in low for kw in ("error:", "failed", "not found", "is not installed", "no packages")) + return {"success": success, "output": raw.strip()} + + def get_pending_updates(self) -> List[UpdateDict]: + if self._pkg_manager == "apt": + return self._get_updates_apt() + if self._pkg_manager in ("dnf", "yum"): + return self._get_updates_rpm() + if self._pkg_manager == "apk": + return self._get_updates_apk() + if self._pkg_manager == "pacman": + return self._get_updates_pacman() + raise NotImplementedError( + f"Package manager '{self._pkg_manager}' is not supported" + ) + + def get_available_updates(self) -> List[UpdateDict]: + """Alias for get_pending_updates(); called by the netork API backend.""" + return self.get_pending_updates() + + def get_device_warnings(self) -> List[Dict[str, Any]]: + """Return warning dicts for issues detected on this device. + + Currently detects: + - package updates available (uses local package cache) + """ + warnings: List[Dict[str, Any]] = [] + try: + updates = self.get_available_updates() + except Exception as exc: + logger.warning("get_device_warnings: get_available_updates() failed: %s", exc) + updates = [] + if updates: + warnings.append({ + "code": "updates_available", + "severity": "warning", + "action": None, + "meta": { + "count": len(updates), + "packages": [u.get("name", "") for u in updates], + }, + }) + return warnings + + def _get_updates_apt(self) -> List[UpdateDict]: + # apt list --upgradable does not need root; avoid sudo so it works even + # without a configured sudo password. + out = self._send( + "LC_ALL=C apt list --upgradable 2>/dev/null | grep -v '^Listing'", + read_timeout=60, + ) + # Join wrapped lines: netmiko's 80-col pseudo-TTY causes long apt lines to + # break; continuation lines start with a space. + raw_lines: List[str] = [] + for line in out.splitlines(): + if line.startswith(" ") and raw_lines: + raw_lines[-1] += line.strip() + else: + raw_lines.append(line) + updates: List[UpdateDict] = [] + for line in raw_lines: + # openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1] + m = re.match( + r"^(\S+)/\S+\s+(\S+)\s+\S+\s+\[upgradable from:\s+(\S+)\]", line + ) + if m: + updates.append({ + "name": m.group(1), + "current_version": m.group(3), + "new_version": m.group(2), + }) + return updates + + def _get_updates_rpm(self) -> List[UpdateDict]: + cmd = "dnf check-update --quiet 2>/dev/null" if self._pkg_manager == "dnf" else "yum check-update -q 2>/dev/null" + out = self._sudo(cmd) + updates: List[UpdateDict] = [] + for line in out.splitlines(): + parts = line.split() + if len(parts) >= 2 and not line.startswith(" ") and "." in parts[0]: + name_arch = parts[0] + name = name_arch.rsplit(".", 1)[0] if "." in name_arch else name_arch + updates.append({ + "name": name, + "current_version": "", + "new_version": parts[1], + }) + return updates + + def _get_updates_apk(self) -> List[UpdateDict]: + out = self._send("apk version -l '<' 2>/dev/null") + updates: List[UpdateDict] = [] + for line in out.splitlines(): + # openssh-9.3_p2-r3 < 9.3_p2-r4 + m = re.match(r"^(\S+)-(\S+)\s+<\s+(\S+)", line) + if m: + updates.append({ + "name": m.group(1), + "current_version": m.group(2), + "new_version": m.group(3), + }) + return updates + + def _get_updates_pacman(self) -> List[UpdateDict]: + out = self._send("pacman -Qu 2>/dev/null") + updates: List[UpdateDict] = [] + for line in out.splitlines(): + # openssh 9.3p2-1 -> 9.4p1-1 + m = re.match(r"^(\S+)\s+(\S+)\s+->\s+(\S+)", line) + if m: + updates.append({ + "name": m.group(1), + "current_version": m.group(2), + "new_version": m.group(3), + }) + return updates + + # ------------------------------------------------------------------ + # OSDriver – apply updates + # ------------------------------------------------------------------ + + # Allowlist for package names – same pattern used by napalm-proxmox + _PKG_NAME_RE = re.compile(r'^[a-zA-Z0-9_\-\+\.]+$') + + def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict: + """Upgrade *packages* (or all pending updates when the list is empty). + + Package names are validated against ``^[a-zA-Z0-9_\\-\\+\\.]+$`` before + being passed to the package manager to prevent shell injection. + """ + for pkg in packages: + if not self._PKG_NAME_RE.match(pkg): + raise ValueError(f"Invalid package name: {pkg!r}") + + if self._pkg_manager == "apt": + return self._apply_updates_apt(packages) + if self._pkg_manager in ("dnf", "yum"): + return self._apply_updates_rpm(packages) + if self._pkg_manager == "apk": + return self._apply_updates_apk(packages) + if self._pkg_manager == "pacman": + return self._apply_updates_pacman(packages) + raise NotImplementedError( + f"Package manager '{self._pkg_manager}' is not supported" + ) + + def _apply_updates_apt(self, packages: List[str]) -> ApplyUpdatesResultDict: + pkg_args = " ".join(packages) if packages else "--with-new-pkgs" + cmd = ( + "DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y " + f"{pkg_args} 2>&1" + if packages else + "DEBIAN_FRONTEND=noninteractive apt-get upgrade -y 2>&1" + ) + try: + output = self._sudo(cmd, read_timeout=600) + success = not re.search(r'^E:', output, re.MULTILINE) + result: ApplyUpdatesResultDict = {"success": success, "output": output} + if not success: + m = re.search(r'^E:.*', output, re.MULTILINE) + result["error"] = m.group(0) if m else "apt-get exited with errors" + return result + except Exception as exc: + return {"success": False, "output": "", "error": str(exc)} + + def _apply_updates_rpm(self, packages: List[str]) -> ApplyUpdatesResultDict: + bin_ = self._pkg_manager # "dnf" or "yum" + if packages: + pkg_args = " ".join(packages) + cmd = f"{bin_} upgrade -y {pkg_args} 2>&1" + else: + cmd = f"{bin_} upgrade -y 2>&1" + try: + output = self._sudo(cmd, read_timeout=600) + # dnf/yum signal failure via "Error:" lines or non-zero exit; + # since we can't check the exit code directly, look for error markers. + success = not re.search(r'^Error:', output, re.MULTILINE | re.IGNORECASE) + result: ApplyUpdatesResultDict = {"success": success, "output": output} + if not success: + m = re.search(r'^Error:.*', output, re.MULTILINE | re.IGNORECASE) + result["error"] = m.group(0) if m else f"{bin_} exited with errors" + return result + except Exception as exc: + return {"success": False, "output": "", "error": str(exc)} + + def _apply_updates_apk(self, packages: List[str]) -> ApplyUpdatesResultDict: + if packages: + pkg_args = " ".join(packages) + cmd = f"apk upgrade {pkg_args} 2>&1" + else: + cmd = "apk upgrade 2>&1" + try: + output = self._sudo(cmd, read_timeout=300) + success = "ERROR" not in output.upper().split("\n")[0] if output else True + result: ApplyUpdatesResultDict = {"success": success, "output": output} + if not success: + result["error"] = "apk upgrade reported an error" + return result + except Exception as exc: + return {"success": False, "output": "", "error": str(exc)} + + def _apply_updates_pacman(self, packages: List[str]) -> ApplyUpdatesResultDict: + if packages: + pkg_args = " ".join(packages) + cmd = f"pacman --noconfirm -S {pkg_args} 2>&1" + else: + cmd = "pacman --noconfirm -Syu 2>&1" + try: + output = self._sudo(cmd, read_timeout=300) + success = "error" not in output.lower() + result: ApplyUpdatesResultDict = {"success": success, "output": output} + if not success: + result["error"] = "pacman reported an error" + return result + except Exception as exc: + return {"success": False, "output": "", "error": str(exc)} + + # ------------------------------------------------------------------ + # OSDriver – services (systemd) + # ------------------------------------------------------------------ + + def get_services(self) -> List[ServiceDict]: + """Return systemd service units (falls back to service --status-all on SysV).""" + out = self._send( + "systemctl list-units --type=service --all --no-legend --no-pager " + "--plain 2>/dev/null" + ) + if not out: + return self._get_services_sysv() + + services: List[ServiceDict] = [] + for line in out.splitlines(): + # ssh.service loaded active running OpenBSD Secure Shell server + parts = line.split(None, 4) + if len(parts) < 4: + continue + unit, load, active, sub = parts[0], parts[1], parts[2], parts[3] + name = unit.removesuffix(".service") + running = active == "active" and sub == "running" + enabled_out = self._send( + f"systemctl is-enabled {unit} 2>/dev/null" + ) + enabled = enabled_out.strip() == "enabled" + + # Retrieve main PID for running services + pid = 0 + if running: + pid_out = self._send( + f"systemctl show -p MainPID --value {unit} 2>/dev/null" + ) + try: + pid = int(pid_out.strip()) + except ValueError: + pid = 0 + + services.append({ + "name": name, + "running": running, + "enabled": enabled, + "pid": pid, + }) + return services + + def _get_services_sysv(self) -> List[ServiceDict]: + out = self._send("service --status-all 2>/dev/null") + services: List[ServiceDict] = [] + for line in out.splitlines(): + m = re.match(r"^\s*\[\s*([+\-?])\s*\]\s+(\S+)", line) + if not m: + continue + services.append({ + "name": m.group(2), + "running": m.group(1) == "+", + "enabled": False, + "pid": 0, + }) + return services + + # ------------------------------------------------------------------ + # OSDriver – users + # ------------------------------------------------------------------ + + def get_users(self) -> List[UserDict]: + """Return local user accounts from /etc/passwd plus supplementary groups.""" + passwd_out = self._send("getent passwd 2>/dev/null || cat /etc/passwd") + groups_out = self._send("getent group 2>/dev/null || cat /etc/group") + + # Build uid→[group] map from /etc/group + uid_to_groups: Dict[int, List[str]] = {} + for line in groups_out.splitlines(): + parts = line.split(":") + if len(parts) < 4: + continue + gname = parts[0] + members = [m.strip() for m in parts[3].split(",") if m.strip()] + for member in members: + # We'll convert username→uid below; collect by username first + uid_to_groups.setdefault(-1, []) # placeholder + + # Simpler: collect username→groups, then join with passwd + username_to_groups: Dict[str, List[str]] = {} + for line in groups_out.splitlines(): + parts = line.split(":") + if len(parts) < 4: + continue + gname = parts[0] + members = [m.strip() for m in parts[3].split(",") if m.strip()] + for member in members: + username_to_groups.setdefault(member, []).append(gname) + + users: List[UserDict] = [] + for line in passwd_out.splitlines(): + parts = line.split(":") + if len(parts) < 7: + continue + username, _, uid_s, gid_s, _, home, shell = parts[:7] + try: + uid, gid = int(uid_s), int(gid_s) + except ValueError: + continue + users.append({ + "username": username, + "uid": uid, + "gid": gid, + "home": home, + "shell": shell, + "groups": username_to_groups.get(username, []), + }) + return users + + # ------------------------------------------------------------------ + # OSDriver – processes + # ------------------------------------------------------------------ + + def get_processes(self) -> List[ProcessDict]: + """Return running processes via ``ps axo``.""" + out = self._send( + "ps axo pid,ppid,user:20,pcpu,pmem,vsz,rss,tty,stat,lstart,args " + "--no-headers 2>/dev/null" + ) + processes: List[ProcessDict] = [] + for line in out.splitlines(): + parts = line.split(None, 10) + if len(parts) < 11: + continue + try: + pid = int(parts[0]) + ppid = int(parts[1]) + user = parts[2] + cpu = float(parts[3]) + mem = float(parts[4]) + vsz = int(parts[5]) + rss = int(parts[6]) + tty = parts[7] if parts[7] != "?" else "" + state = parts[8][0] if parts[8] else "?" + # lstart is 5 tokens: "Mon May 27 12:34:56 2024" → parts[9..13] + # args starts at parts[14] but we merged from 10 onward + # With --no-headers and ps axo, lstart takes 5 parts + # Rebuild: parts[9] is start, args is parts[10] + started = parts[9] + command = parts[10] + except (ValueError, IndexError): + continue + processes.append({ + "pid": pid, + "ppid": ppid, + "user": user, + "cpu": cpu, + "memory": mem, + "vsz": vsz, + "rss": rss, + "tty": tty, + "state": state, + "started": started, + "command": command, + }) + return processes + + # ------------------------------------------------------------------ + # OSDriver – cron jobs + # ------------------------------------------------------------------ + + def get_cron_jobs(self) -> List[CronJobDict]: + """Return cron entries from user crontabs and /etc/cron.d.""" + jobs: List[CronJobDict] = [] + + # /etc/cron.d/* — system-wide cron fragments (include user field) + cron_d_files = self._send("ls /etc/cron.d/ 2>/dev/null").splitlines() + for fname in cron_d_files: + fname = fname.strip() + if not fname: + continue + content = self._send(f"cat /etc/cron.d/{fname} 2>/dev/null") + for line in content.splitlines(): + job = self._parse_cron_line(line, source_user="root", has_user_field=True) + if job: + jobs.append(job) + + # Per-user crontabs from /var/spool/cron/crontabs (Debian) or /var/spool/cron (RHEL) + for spool_dir in ("/var/spool/cron/crontabs", "/var/spool/cron"): + ls_out = self._send(f"ls {spool_dir} 2>/dev/null") + for uname in ls_out.splitlines(): + uname = uname.strip() + if not uname: + continue + content = self._send(f"cat {spool_dir}/{uname} 2>/dev/null") + for line in content.splitlines(): + job = self._parse_cron_line(line, source_user=uname, has_user_field=False) + if job: + jobs.append(job) + + return jobs + + @staticmethod + def _parse_cron_line( + line: str, source_user: str, has_user_field: bool + ) -> Optional[CronJobDict]: + """Parse a single crontab line; returns ``None`` for comments/blanks.""" + stripped = line.strip() + # Remove trailing comment + comment = "" + if "#" in stripped: + idx = stripped.index("#") + comment = stripped[idx + 1:].strip() + stripped = stripped[:idx].strip() + + if not stripped or stripped.startswith("@") or stripped.startswith("MAILTO"): + return None + + parts = stripped.split(None, 6 if has_user_field else 5) + expected = 6 if has_user_field else 5 + if len(parts) < expected: + return None + + schedule = " ".join(parts[:5]) + if has_user_field: + user = parts[5] + command = parts[6] if len(parts) > 6 else "" + else: + user = source_user + command = parts[5] if len(parts) > 5 else "" + + job: CronJobDict = { + "user": user, + "schedule": schedule, + "command": command, + } + if comment: + job["description"] = comment + return job + + # ------------------------------------------------------------------ + # Docker + # ------------------------------------------------------------------ + + def get_docker_info(self) -> DockerInfoDict: + """Return information about the local Docker environment. + + Uses a single SSH call to collect all Docker data at once, eliminating + per-section round-trip overhead. Labels from ``docker images`` are used + directly for the OCI version field — no separate ``docker image inspect`` + needed. + + Returns a dict with keys: + - ``available`` (bool) — False if docker is not installed/accessible + - ``version`` (str) — Docker Engine version string + - ``containers`` (list) — list of container dicts + - ``images`` (list) — list of image dicts + - ``volumes`` (list) — list of volume dicts + - ``networks`` (list) — list of network dicts + """ + 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. + 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---'; " + "docker images --format '{{json .}}' 2>/dev/null; " + "echo '---VOLUMES---'; " + "docker volume ls --format '{{json .}}' 2>/dev/null; " + "echo '---NETWORKS---'; " + "docker network ls --format '{{json .}}' 2>/dev/null", + read_timeout=30, + ) + + 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 + def _section(text: str, marker: str, next_marker: str) -> str: + start = text.find(marker) + if start == -1: + return "" + start += len(marker) + 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---") + raw_networks = _section(combined, "---NETWORKS---", "\x00") # sentinel + + def _parse_labels(raw: Any) -> Dict[str, str]: + """Parse Docker labels — may be a dict (JSON map) or comma-sep string.""" + if isinstance(raw, dict): + return {str(k): str(v) for k, v in raw.items()} + if isinstance(raw, str) and raw: + result: Dict[str, str] = {} + for part in raw.split(","): + if "=" in part: + k, _, v = part.partition("=") + result[k.strip()] = v.strip() + return result + return {} + + # Containers + containers: List[Dict[str, Any]] = [] + for line in raw_containers.splitlines(): + line = line.strip() + if not line: + continue + try: + obj = _json.loads(line) + labels = _parse_labels(obj.get("Labels", "")) + containers.append({ + "id": obj.get("ID", ""), + "name": obj.get("Names", ""), + "image": obj.get("Image", ""), + "image_version": labels.get("org.opencontainers.image.version", ""), + "command": obj.get("Command", ""), + "created": obj.get("CreatedAt", ""), + "status": obj.get("Status", ""), + "ports": obj.get("Ports", ""), + "state": obj.get("State", ""), + "compose_project": labels.get("com.docker.compose.project", ""), + "compose_service": labels.get("com.docker.compose.service", ""), + "compose_file": labels.get("com.docker.compose.project.config_files", ""), + }) + except Exception: + pass + + # Images — OCI version comes from Labels, no separate inspect needed + images: List[Dict[str, Any]] = [] + for line in raw_images.splitlines(): + line = line.strip() + if not line: + continue + try: + obj = _json.loads(line) + labels = _parse_labels(obj.get("Labels", "")) + images.append({ + "id": obj.get("ID", ""), + "repository": obj.get("Repository", ""), + "tag": obj.get("Tag", ""), + "size": obj.get("Size", ""), + "created": obj.get("CreatedAt", ""), + "version": labels.get("org.opencontainers.image.version", ""), + }) + except Exception: + pass + + # Volumes + volumes: List[Dict[str, Any]] = [] + for line in raw_volumes.splitlines(): + line = line.strip() + if not line: + continue + try: + obj = _json.loads(line) + volumes.append({ + "name": obj.get("Name", ""), + "driver": obj.get("Driver", ""), + "mountpoint": obj.get("Mountpoint", ""), + "scope": obj.get("Scope", ""), + }) + except Exception: + pass + + # Networks + networks: List[Dict[str, Any]] = [] + for line in raw_networks.splitlines(): + line = line.strip() + if not line: + continue + try: + obj = _json.loads(line) + networks.append({ + "id": obj.get("ID", ""), + "name": obj.get("Name", ""), + "driver": obj.get("Driver", ""), + "scope": obj.get("Scope", ""), + "ipv6": obj.get("IPv6", ""), + "internal": obj.get("Internal", ""), + }) + except Exception: + pass + + return { + "available": True, + "version": version, + "containers": containers, + "images": images, + "volumes": volumes, + "networks": networks, + "outdated_images": [], # populated by separate check_docker_outdated task + } + + def get_docker_outdated(self, containers: List[Dict]) -> List[str]: + """Check registry for available updates for all container images. + + Runs ``docker buildx imagetools inspect`` (metadata-only, no download) + for each unique image referenced by a container. Intended to be called + from a separate Celery task on a long interval (e.g. every 3 hours) so + it never blocks the main device poll. + + Returns a list of image references that have a newer digest available. + """ + outdated_images: List[str] = [] + candidate_images: List[str] = list({ + c["image"] for c in containers + if c.get("image") + and "@sha256:" not in c.get("image", "") # skip digest-pinned + }) + for img_name in candidate_images: + try: + local_raw = self._send( + f"docker inspect {img_name!r} --format '{{{{index .RepoDigests 0}}}}' 2>/dev/null", + read_timeout=5, + ).strip() + if not local_raw or "@" not in local_raw: + continue # locally built or not yet pulled + local_digest = local_raw.split("@", 1)[1] + + remote_full = self._send( + f"docker buildx imagetools inspect {img_name!r} 2>&1", + read_timeout=30, + ).strip() + if ("429" in remote_full + or "Too Many Requests" in remote_full + or "toomanyrequests" in remote_full): + logger.warning( + "Docker Hub rate limit hit for %s — run " + "'docker login' on the device to avoid this", + img_name, + ) + continue + remote_digest = "" + for _line in remote_full.splitlines(): + _ls = _line.strip() + if _ls.startswith("Digest:"): + remote_digest = _ls[7:].strip() + break + if not remote_digest or not remote_digest.startswith("sha256:"): + continue + if local_digest != remote_digest: + outdated_images.append(img_name) + except Exception as exc: + logger.warning("image update check for %s: %s", img_name, exc) + return outdated_images + + def reconstruct_docker_run(self, container_id: str) -> Optional[Dict]: + """Return the information needed to recreate a standalone container. + + Parses ``docker inspect`` JSON and returns a dict with: + - ``name`` — container name (without leading slash) + - ``image`` — current image reference + - ``run_args`` — list of CLI args for ``docker run`` (without image/cmd) + - ``cmd`` — command override (may be empty list) + - ``entrypoint`` — entrypoint override (may be empty list) + + Returns None if the container does not exist or inspect fails. + """ + import json as _json + import shlex as _shlex + + raw = self._send( + f"docker inspect {_shlex.quote(container_id)} 2>/dev/null", + read_timeout=10, + ).strip() + if not raw: + return None + try: + data = _json.loads(raw) + except Exception: + return None + if not data: + return None + c = data[0] + + name = c.get("Name", "").lstrip("/") + cfg = c.get("Config", {}) + hcfg = c.get("HostConfig", {}) + net_settings = c.get("NetworkSettings", {}) + + args: List[str] = ["--name", name] + + # Restart policy + rp = hcfg.get("RestartPolicy", {}) + rp_name = rp.get("Name", "no") + if rp_name and rp_name != "no": + max_retry = rp.get("MaximumRetryCount", 0) + if rp_name == "on-failure" and max_retry: + args += ["--restart", f"on-failure:{max_retry}"] + else: + args += ["--restart", rp_name] + + # Hostname + hostname = cfg.get("Hostname", "") + if hostname and hostname != name[:12]: + args += ["--hostname", hostname] + + # Environment (skip vars that look like Docker-injected metadata) + _skip_prefixes = ("PATH=", "HOME=", "TERM=", "HOSTNAME=") + for env in cfg.get("Env") or []: + if not any(env.startswith(p) for p in _skip_prefixes): + args += ["-e", env] + + # Volume binds + for bind in hcfg.get("Binds") or []: + args += ["-v", bind] + + # Port bindings + for container_port, host_bindings in (hcfg.get("PortBindings") or {}).items(): + for hb in (host_bindings or []): + host_ip = hb.get("HostIp", "") + host_port = hb.get("HostPort", "") + if host_ip: + args += ["-p", f"{host_ip}:{host_port}:{container_port}"] + else: + args += ["-p", f"{host_port}:{container_port}"] + + # Network mode + net_mode = hcfg.get("NetworkMode", "default") + if net_mode not in ("default", "bridge"): + args += ["--network", net_mode] + else: + # Check for custom networks from NetworkSettings + for net_name in (net_settings.get("Networks") or {}): + if net_name not in ("bridge", "host", "none"): + args += ["--network", net_name] + break + + # Privileged + if hcfg.get("Privileged"): + args.append("--privileged") + + # Cap-add + for cap in hcfg.get("CapAdd") or []: + args += ["--cap-add", cap] + + # Devices + for dev in hcfg.get("Devices") or []: + host_p = dev.get("PathOnHost", "") + ctr_p = dev.get("PathInContainer", "") + perms = dev.get("CgroupPermissions", "rwm") + if host_p: + args += ["--device", f"{host_p}:{ctr_p}:{perms}"] + + # Extra hosts + for eh in hcfg.get("ExtraHosts") or []: + args += ["--add-host", eh] + + # DNS + for dns in hcfg.get("Dns") or []: + args += ["--dns", dns] + + # Labels (skip Docker-internal labels) + _skip_label_prefixes = ("com.docker.compose.", "org.opencontainers.") + for k, v in (cfg.get("Labels") or {}).items(): + if not any(k.startswith(p) for p in _skip_label_prefixes): + args += ["--label", f"{k}={v}"] + + # Detach always + args.append("-d") + + return { + "name": name, + "image": cfg.get("Image", ""), + "run_args": args, + "cmd": cfg.get("Cmd") or [], + "entrypoint": cfg.get("Entrypoint") or [], + } + + # ── Device actions ──────────────────────────────────────────────────────── + + def get_snmp_config(self) -> Optional[SNMPConfigDict]: + """Return SNMP agent config if snmpd is installed and running.""" + try: + running = ( + self._send("systemctl is-active snmpd 2>/dev/null || true").strip() + == "active" + ) + if not running: + return None + + # Parse community string from snmpd.conf + community = "public" + port = 161 + try: + conf = self._send( + "grep -E '^[[:space:]]*(ro|rw)?community' /etc/snmp/snmpd.conf 2>/dev/null" + " | head -5" + ) + for line in conf.splitlines(): + parts = line.split() + if not parts: + continue + kw = parts[0].lower() + if kw in ("rocommunity", "rwcommunity", "rocommunity6", "rwcommunity6"): + if len(parts) >= 2: + community = parts[1] + break + elif kw == "com2sec" and len(parts) >= 4: + # com2sec notConfigUser default + community = parts[3] + break + except Exception: + pass + + # Detect port override + try: + port_line = self._send( + "grep -E '^agentAddress' /etc/snmp/snmpd.conf 2>/dev/null | head -1" + ).strip() + if port_line: + m = re.search(r':(\d+)', port_line) + if m: + port = int(m.group(1)) + except Exception: + pass + + return SNMPConfigDict(running=True, community=community, port=port, version="2c") + except Exception as exc: + logger.debug("get_snmp_config() failed: %s", exc) + return None + + def run_device_action(self, action: str) -> DeviceActionResultDict: + """Execute a named action on the device.""" + if action == "fix_docker_permissions": + return self._action_fix_docker_permissions() + if action == "fix_snmp": + return self._action_fix_snmp() + raise NotImplementedError(f"Unknown action: {action!r}") + + def _action_fix_snmp(self) -> DeviceActionResultDict: + """Install, configure and start snmpd with community 'public'.""" + lines: list[str] = [] + + # 0. Verify sudo access before attempting anything + sudo_check = self._send("sudo -n true 2>&1 || echo __SUDO_NEEDS_PW__") + if "__SUDO_NEEDS_PW__" in sudo_check or "password is required" in sudo_check.lower(): + if not self._sudo_password: + return { + "success": False, + "output": ( + "sudo requires a password on this device but none is configured in netOrk. " + "Please add the sudo password to a Credential Profile assigned to this device, " + "or configure passwordless sudo (NOPASSWD) for this user." + ), + } + + # 1. Install snmpd if missing + pkg_mgr = self._detect_pkg_manager() + if not pkg_mgr: + return {"success": False, "output": "Package manager not detected — cannot install snmpd."} + + # Install both snmpd (daemon) and snmp (client tools incl. snmpget for probing) + install_cmd: dict[str, str] = { + "apt": "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1", + "dnf": "dnf install -y net-snmp net-snmp-utils 2>&1", + "yum": "yum install -y net-snmp net-snmp-utils 2>&1", + "apk": "apk add --no-cache net-snmp net-snmp-tools 2>&1", + "pacman": "pacman -Sy --noconfirm net-snmp 2>&1", + } + cmd = install_cmd.get(pkg_mgr) + if cmd: + out = self._sudo(cmd, read_timeout=120) + lines.append(f"[install] {out.strip()[-200:]}") + + # 2. Determine the IP netOrk is connecting from by checking the established SSH connection + netork_ip = "" + try: + # ss shows the remote peer of the current SSH connection + raw = self._send( + "ss -tnp 2>/dev/null | awk '/sshd/{print $5}' | head -1 | cut -d: -f1" + ).strip() + if raw and raw not in ("", "0.0.0.0", "::", "127.0.0.1"): + netork_ip = raw + except Exception: + pass + + # Write snmpd.conf: + # 1. Write to /tmp (no sudo needed, avoids stdin conflict with sudo -S) + # 2. sudo mv to /etc/snmp/snmpd.conf + # agentAddress udp:161 overrides Debian's localhost-only default. + import base64 as _b64 + conf_str = ( + "agentAddress udp:161\n" + "rocommunity public\n" + "sysLocation Managed by netOrk\n" + "sysContact netork@localhost\n" + ) + conf_b64 = _b64.b64encode(conf_str.encode()).decode() + self._send(f"echo {conf_b64} | base64 -d > /tmp/netork_snmpd.conf") + self._sudo("mv /tmp/netork_snmpd.conf /etc/snmp/snmpd.conf && chown root:root /etc/snmp/snmpd.conf && chmod 644 /etc/snmp/snmpd.conf") + verify = self._send("cat /etc/snmp/snmpd.conf 2>/dev/null").strip() + if "agentAddress" in verify and "rocommunity" in verify: + lines.append("[config] Wrote /etc/snmp/snmpd.conf — agentAddress udp:161, rocommunity public.") + else: + lines.append(f"[warn] snmpd.conf write may have failed: {verify[:100]}") + + # 3. Open firewall for SNMP (UDP 161) — restrict to netOrk's source IP + if netork_ip: + try: + ufw = self._send("command -v ufw 2>/dev/null").strip() + ipt = self._send("command -v iptables 2>/dev/null").strip() + if ufw: + # Expand to /24 so all containers in the same Docker network can probe + parts = netork_ip.rsplit(".", 1) + subnet = f"{parts[0]}.0/24" if len(parts) == 2 else netork_ip + fw_out = self._sudo( + f"ufw allow from {subnet} to any port 161 proto udp 2>&1", read_timeout=10 + ) + lines.append(f"[firewall/ufw] {fw_out.strip()[:200]}") + elif ipt: + fw_out = self._sudo( + f"iptables -C INPUT -s {netork_ip} -p udp --dport 161 -j ACCEPT 2>/dev/null" + f" || iptables -I INPUT -s {netork_ip} -p udp --dport 161 -j ACCEPT", + read_timeout=10, + ) + lines.append(f"[firewall/iptables] rule added for {netork_ip}:161/udp") + except Exception as exc: + lines.append(f"[firewall] skipped — {exc}") + + # 4. Restart snmpd. + # - Redirect all output to /dev/null so netmiko's prompt detection is + # never confused by service status messages. + # - Append "; echo __OK__" so there is always a known token to wait for. + import time as _time + # Stop any running snmpd (systemctl-managed or apt-started orphan) + self._sudo("systemctl stop snmpd >/dev/null 2>&1; echo s1", read_timeout=15) + self._sudo("pkill -9 snmpd >/dev/null 2>&1; echo s2", read_timeout=10) + _time.sleep(2) + # Enable and start fresh + self._sudo("systemctl enable snmpd >/dev/null 2>&1; echo s3", read_timeout=15) + self._sudo("systemctl start snmpd >/dev/null 2>&1; echo s4", read_timeout=20) + _time.sleep(2) + lines.append("[service] snmpd restarted.") + + # 5. Verify snmpd responds via local SNMP probe (sysDescr.0). + # Success requires actual SNMP data types in the output, not just + # the absence of error keywords. + _time.sleep(2) + probe_out = self._send( + "snmpget -v2c -cpublic -t2 -r0 -Ov 127.0.0.1 1.3.6.1.2.1.1.1.0 2>&1 || true" + ).strip() + # snmpget returns lines like "STRING: Linux ..." or "Timeticks: (n) ..." + _snmp_types = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:") + success = any(t in probe_out for t in _snmp_types) + if success: + lines.append(f"[ok] SNMP probe successful — community 'public' is working.") + else: + lines.append(f"[warn] SNMP probe failed — output: {probe_out[:200]}") + + return {"success": success, "output": "\n".join(lines)} + + def _action_fix_docker_permissions(self) -> Dict[str, Any]: + """Add the SSH user to the 'docker' group via sudo usermod.""" + user = self._send("whoami 2>/dev/null || id -un").strip().splitlines()[-1].strip() + out = self._sudo(f"usermod -aG docker {user}") + low = out.lower() + success = not any(kw in low for kw in ("error", "invalid", "no such", "command not found")) + if not out.strip(): + out = f"Added {user!r} to the docker group. Reconnect or run a new poll to verify." + return {"success": success, "output": out} + + def get_vpn_tunnels(self) -> Dict[str, Any]: + """Return WireGuard status via ``wg show all dump`` (requires root/sudo). + + Falls back to interface-level data from ``ip link`` + ``/proc/net/dev`` + when root access is unavailable. + + Full data keyed by ``wireguard--`` (one entry per peer). + Fallback keyed by ``wireguard-`` (one entry per WireGuard interface). + """ + import time as _time + + tunnels: Dict[str, Any] = {} + + # ── Attempt 1: wg show all dump via sudo ───────────────────────────── + # Write output to a fixed temp file to preserve literal tab characters. + # PTY output processing expands tabs to spaces, breaking split("\t"). + import hashlib as _hashlib + _tmp = "/tmp/.netork_wg_" + _hashlib.md5(self.hostname.encode()).hexdigest()[:8] + if self._sudo_password: + self._sudo(f"wg show all dump > {_tmp} 2>&1 || true") + else: + self._send(f"sudo -n wg show all dump > {_tmp} 2>&1 || true") + raw = self._send(f"cat {_tmp} 2>/dev/null || true; rm -f {_tmp}").strip() + + _perm_errors = ( + "operation not permitted", "permission denied", + "a password is required", "a terminal is required", + "command not found", "not found", "no such file", + ) + has_peer_data = raw and not any(e in raw.lower() for e in _perm_errors) + + if has_peer_data: + # Build listen-port map from interface lines (5 tab-separated fields) + listen_ports: Dict[str, str] = {} + for line in raw.splitlines(): + parts = line.split("\t") + if len(parts) == 5: + iface, _priv, _pub, port, _fwmark = parts + listen_ports[iface.strip()] = port.strip() + + now = int(_time.time()) + for line in raw.splitlines(): + parts = line.split("\t") + if len(parts) != 9: + continue + (iface, pubkey, _psk, endpoint, allowed_ips, + latest_hs, rx_bytes, tx_bytes, _keepalive) = parts + + iface = iface.strip() + pubkey = pubkey.strip() + endpoint = endpoint.strip() + + remote_ip = "" + if endpoint and endpoint != "(none)": + remote_ip = endpoint.rsplit(":", 1)[0].strip("[]") + + try: + hs_ts = int(latest_hs) + except (ValueError, TypeError): + hs_ts = 0 + + # "up" if last handshake within 3 min (WireGuard re-handshake every 2 min) + is_up = hs_ts > 0 and (now - hs_ts) < 180 + + try: + bytes_in = int(rx_bytes) + except (ValueError, TypeError): + bytes_in = 0 + try: + bytes_out = int(tx_bytes) + except (ValueError, TypeError): + bytes_out = 0 + + local_port = listen_ports.get(iface, "") + local_ep = f":{local_port}" if local_port and local_port != "0" else "" + + key = f"wireguard-{iface}-{pubkey[:8]}" + tunnels[key] = { + "type": "WireGuard", + "local_endpoint": local_ep, + "remote_endpoint": remote_ip, + "is_up": is_up, + "uptime": (now - hs_ts) if hs_ts > 0 else 0, + "bytes_in": bytes_in, + "bytes_out": bytes_out, + "description": f"{iface} — peer {pubkey[:16]}…", + "public_key": pubkey, + "allowed_ips": allowed_ips.strip(), + "interface": iface, + } + return tunnels + + # ── Fallback: interface-level data without root ────────────────────── + # ip -j link show type wireguard → list of WireGuard interface objects + ip_raw = self._send("ip -j link show type wireguard 2>/dev/null || true").strip() + if not ip_raw or ip_raw.startswith("[") is False: + # Try stripping shell noise before the JSON + start = ip_raw.find("[") + ip_raw = ip_raw[start:] if start != -1 else "" + + if not ip_raw: + return tunnels + + try: + import json as _json + iface_list = _json.loads(ip_raw) + except Exception: + return tunnels + + # /proc/net/dev for total RX/TX bytes per interface + proc_dev = self._send("cat /proc/net/dev 2>/dev/null || true") + proc_bytes: Dict[str, tuple] = {} + for line in proc_dev.splitlines()[2:]: + line = line.strip() + if ":" not in line: + continue + iface_name, rest = line.split(":", 1) + fields = rest.split() + try: + proc_bytes[iface_name.strip()] = (int(fields[0]), int(fields[8])) + except (IndexError, ValueError): + pass + + for iface_obj in iface_list: + iface = iface_obj.get("ifname", "") + if not iface: + continue + flags = iface_obj.get("flags", []) + is_up = "UP" in flags and "LOWER_UP" in flags + rx, tx = proc_bytes.get(iface, (0, 0)) + tunnels[f"wireguard-{iface}"] = { + "type": "WireGuard", + "local_endpoint": "", + "remote_endpoint": "", + "is_up": is_up, + "uptime": 0, + "bytes_in": rx, + "bytes_out": tx, + "description": f"{iface} (peer data requires root/sudo)", + "interface": iface, + } + + return tunnels diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..3dc3410 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,65 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "napalm-linux" +version = "0.1.0" +description = "NAPALM driver for generic Linux systems via SSH" +readme = "README.md" +requires-python = ">=3.9" +license = { text = "Apache-2.0" } +authors = [ + { name = "Christian Manivong", email = "christian@manivong.de" }, +] +keywords = [ + "napalm", + "network", + "automation", + "linux", + "ssh", + "driver", +] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Intended Audience :: System Administrators", + "License :: OSI Approved :: Apache Software License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: System :: Networking", + "Topic :: System :: Systems Administration", + "Typing :: Typed", +] +dependencies = [ + "napalm>=4.0", + "napalm-device-types>=0.3.0", + "netmiko>=4.0.0", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-cov", + "black", + "ruff", + "mypy", +] + +[project.entry-points."napalm.drivers"] +linux = "napalm_linux:LinuxDriver" + +[project.urls] +Repository = "https://github.com/chrismanivong/napalm-linux" + +[tool.setuptools.packages.find] +where = ["."] +include = ["napalm_linux*"] + +[tool.ruff] +line-length = 100 +target-version = "py39" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..255f2c7 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for napalm-linux driver.""" diff --git a/tests/test_linux.py b/tests/test_linux.py new file mode 100644 index 0000000..e3e7f00 --- /dev/null +++ b/tests/test_linux.py @@ -0,0 +1,341 @@ +"""Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed).""" + +import pytest +from unittest.mock import MagicMock, patch +from napalm_linux.linux import LinuxDriver + + +# --------------------------------------------------------------------------- +# Fixture – driver without a real connection +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def driver(): + """Return a LinuxDriver instance with netmiko mocked out.""" + d = LinuxDriver.__new__(LinuxDriver) + d.hostname = "testhost" + d.username = "user" + d.password = "pass" # noqa: S106 + d.timeout = 60 + d.port = 22 + d._secret = "pass" # noqa: S105 + d._forced_pkg_manager = None + d._pkg_manager = "apt" + d.netmiko_optional_args = {} + d._device = MagicMock() + return d + + +def _mock_send(driver_fixture, output: str): + """Patch _send to return *output* for any command.""" + driver_fixture._device.send_command.return_value = output + + +# --------------------------------------------------------------------------- +# _parse_cron_line +# --------------------------------------------------------------------------- + + +class TestParseCronLine: + def test_regular_user_cron(self): + line = "0 4 * * * /usr/local/bin/backup.sh # nightly backup" + job = LinuxDriver._parse_cron_line(line, source_user="root", has_user_field=False) + assert job is not None + assert job["user"] == "root" + assert job["schedule"] == "0 4 * * *" + assert job["command"] == "/usr/local/bin/backup.sh" + assert job.get("description") == "nightly backup" + + def test_cron_d_with_user_field(self): + line = "*/5 * * * * www-data /usr/bin/php /var/www/cron.php" + job = LinuxDriver._parse_cron_line(line, source_user="root", has_user_field=True) + assert job is not None + assert job["user"] == "www-data" + assert job["schedule"] == "*/5 * * * *" + assert "/usr/bin/php" in job["command"] + + def test_comment_line_returns_none(self): + assert LinuxDriver._parse_cron_line("# this is a comment", "root", False) is None + + def test_blank_line_returns_none(self): + assert LinuxDriver._parse_cron_line(" ", "root", False) is None + + def test_mailto_returns_none(self): + assert LinuxDriver._parse_cron_line("MAILTO=root", "root", False) is None + + +# --------------------------------------------------------------------------- +# get_interfaces (parsing) +# --------------------------------------------------------------------------- + + +IP_LINK_OUTPUT = """\ +1: lo: mtu 65536 qdisc noqueue state UNKNOWN mode DEFAULT group default qlen 1000\\ link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00 +2: eth0: mtu 1500 qdisc mq state UP mode DEFAULT group default qlen 1000\\ link/ether aa:bb:cc:dd:ee:ff brd ff:ff:ff:ff:ff:ff +3: eth1: mtu 1500 qdisc noop state DOWN mode DEFAULT group default qlen 1000\\ link/ether 11:22:33:44:55:66 brd ff:ff:ff:ff:ff:ff +""" + + +def test_get_interfaces_parses_state(driver): + with patch.object(driver, "_send", return_value=IP_LINK_OUTPUT): + result = driver.get_interfaces() + assert "eth0" in result + assert result["eth0"]["is_up"] is True + assert result["eth0"]["mtu"] == 1500 + assert result["eth0"]["mac_address"] == "aa:bb:cc:dd:ee:ff" + assert "eth1" in result + assert result["eth1"]["is_up"] is False + + +# --------------------------------------------------------------------------- +# _parse_uptime +# --------------------------------------------------------------------------- + + +def test_parse_uptime(driver): + with patch.object(driver, "_send", return_value="86400.12 1234.56"): + assert driver._parse_uptime() == 86400 + + +def test_parse_uptime_invalid(driver): + with patch.object(driver, "_send", return_value=""): + assert driver._parse_uptime() == 0 + + +# --------------------------------------------------------------------------- +# get_packages (apt) +# --------------------------------------------------------------------------- + + +APT_PKG_OUTPUT = ( + "openssh-server\t1:9.2p1-2+deb12u2\t512\tsecure shell server\n" + "curl\t7.88.1-10+deb12u5\t1024\tcommand line tool for transferring data\n" +) + + +def test_get_packages_apt(driver): + driver._pkg_manager = "apt" + with patch.object(driver, "_send", return_value=APT_PKG_OUTPUT): + pkgs = driver.get_packages() + assert len(pkgs) == 2 + assert pkgs[0]["name"] == "openssh-server" + assert pkgs[0]["version"] == "1:9.2p1-2+deb12u2" + assert pkgs[0]["installed"] is True + assert pkgs[0]["source"] == "apt" + + +# --------------------------------------------------------------------------- +# get_pending_updates (apt) +# --------------------------------------------------------------------------- + + +APT_UPGRADABLE = ( + "Listing... Done\n" + "openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1]\n" + "curl/stable 7.88.1-10+deb12u6 amd64 [upgradable from: 7.88.1-10+deb12u5]\n" +) + + +def test_get_pending_updates_apt(driver): + driver._pkg_manager = "apt" + with patch.object(driver, "_send", side_effect=["", APT_UPGRADABLE]): + updates = driver.get_pending_updates() + assert len(updates) == 2 + assert updates[0]["name"] == "openssh-server" + assert updates[0]["current_version"] == "1:9.2p1-2+deb12u1" + assert updates[0]["new_version"] == "1:9.2p1-2+deb12u2" + + +# --------------------------------------------------------------------------- +# get_users +# --------------------------------------------------------------------------- + + +PASSWD_OUT = ( + "root:x:0:0:root:/root:/bin/bash\n" + "admin:x:1000:1000:Admin User:/home/admin:/bin/bash\n" + "daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin\n" +) + +GROUP_OUT = ( + "sudo:x:27:admin\n" + "docker:x:999:admin\n" + "adm:x:4:admin\n" +) + + +def test_get_users(driver): + with patch.object(driver, "_send", side_effect=[PASSWD_OUT, GROUP_OUT]): + users = driver.get_users() + + admin = next(u for u in users if u["username"] == "admin") + assert admin["uid"] == 1000 + assert admin["gid"] == 1000 + assert admin["home"] == "/home/admin" + assert admin["shell"] == "/bin/bash" + assert set(admin["groups"]) == {"sudo", "docker", "adm"} + + +# --------------------------------------------------------------------------- +# ping parsing +# --------------------------------------------------------------------------- + + +PING_OUTPUT = """\ +PING 8.8.8.8 (8.8.8.8) 100(128) bytes of data. +108 bytes from 8.8.8.8: icmp_seq=1 ttl=118 time=12.3 ms +108 bytes from 8.8.8.8: icmp_seq=2 ttl=118 time=11.9 ms +108 bytes from 8.8.8.8: icmp_seq=3 ttl=118 time=12.1 ms + +--- 8.8.8.8 ping statistics --- +3 packets transmitted, 3 received, 0% packet loss, time 2003ms +rtt min/avg/max/mdev = 11.900/12.100/12.300/0.163 ms +""" + + +def test_ping_parses_output(driver): + with patch.object(driver, "_send", return_value=PING_OUTPUT): + result = driver.ping("8.8.8.8", count=3) + assert "success" in result + assert result["success"]["probes_sent"] == 3 + assert result["success"]["packet_loss"] == 0 + assert result["success"]["rtt_avg"] == pytest.approx(12.1) + + +# --------------------------------------------------------------------------- +# apply_updates – input validation +# --------------------------------------------------------------------------- + + +class TestApplyUpdatesValidation: + def test_valid_package_names_accepted(self, driver): + driver._pkg_manager = "apt" + apt_output = "Reading package lists... Done\nThe following packages will be upgraded:\n openssh-server\n1 upgraded." + with patch.object(driver, "_send", return_value=apt_output): + result = driver.apply_updates(["openssh-server", "curl", "lib32-foo+bar.so"]) + assert result["success"] is True + + def test_invalid_package_name_raises(self, driver): + with pytest.raises(ValueError, match="Invalid package name"): + driver.apply_updates(["open;ssh"]) + + def test_shell_injection_blocked(self, driver): + with pytest.raises(ValueError, match="Invalid package name"): + driver.apply_updates(["pkg && rm -rf /"]) + + def test_space_in_name_blocked(self, driver): + with pytest.raises(ValueError, match="Invalid package name"): + driver.apply_updates(["my package"]) + + +# --------------------------------------------------------------------------- +# apply_updates – apt +# --------------------------------------------------------------------------- + + +APT_UPGRADE_SUCCESS = ( + "Reading package lists... Done\n" + "Building dependency tree... Done\n" + "The following packages will be upgraded:\n" + " openssh-server\n" + "1 upgraded, 0 newly installed, 0 to remove and 0 not upgraded.\n" +) + +APT_UPGRADE_ERROR = ( + "Reading package lists... Done\n" + "E: Unable to lock the administration directory (/var/lib/dpkg/), " + "is another process using it?\n" +) + + +def test_apply_updates_apt_success(driver): + driver._pkg_manager = "apt" + with patch.object(driver, "_send", return_value=APT_UPGRADE_SUCCESS): + result = driver.apply_updates(["openssh-server"]) + assert result["success"] is True + assert "openssh-server" in result["output"] + assert "error" not in result + + +def test_apply_updates_apt_error(driver): + driver._pkg_manager = "apt" + with patch.object(driver, "_send", return_value=APT_UPGRADE_ERROR): + result = driver.apply_updates(["openssh-server"]) + assert result["success"] is False + assert "error" in result + assert result["error"].startswith("E:") + + +def test_apply_updates_apt_all_packages(driver): + """Empty list should upgrade everything (no package name args).""" + driver._pkg_manager = "apt" + sent_commands = [] + + def capture_send(cmd): + sent_commands.append(cmd) + return APT_UPGRADE_SUCCESS + + with patch.object(driver, "_send", side_effect=capture_send): + result = driver.apply_updates([]) + + assert result["success"] is True + # Should use 'apt-get upgrade' without specific package args + assert any("upgrade" in cmd and "install" not in cmd for cmd in sent_commands) + + +# --------------------------------------------------------------------------- +# apply_updates – dnf +# --------------------------------------------------------------------------- + + +DNF_UPGRADE_SUCCESS = ( + "Last metadata expiration check: 0:01:23 ago.\n" + "Dependencies resolved.\n" + "Upgraded:\n openssh-server-9.4p1-1.el9.x86_64\n" + "Complete!\n" +) + +DNF_UPGRADE_ERROR = ( + "Last metadata expiration check: 0:01:23 ago.\n" + "Error: No match for argument: nonexistent-pkg\n" +) + + +def test_apply_updates_dnf_success(driver): + driver._pkg_manager = "dnf" + with patch.object(driver, "_send", return_value=DNF_UPGRADE_SUCCESS): + result = driver.apply_updates(["openssh-server"]) + assert result["success"] is True + + +def test_apply_updates_dnf_error(driver): + driver._pkg_manager = "dnf" + with patch.object(driver, "_send", return_value=DNF_UPGRADE_ERROR): + result = driver.apply_updates(["nonexistent-pkg"]) + assert result["success"] is False + assert "error" in result + + +# --------------------------------------------------------------------------- +# apply_updates – exception path +# --------------------------------------------------------------------------- + + +def test_apply_updates_ssh_exception_returns_failure(driver): + driver._pkg_manager = "apt" + with patch.object(driver, "_send", side_effect=RuntimeError("SSH timeout")): + result = driver.apply_updates(["curl"]) + assert result["success"] is False + assert "SSH timeout" in result.get("error", "") + + +# --------------------------------------------------------------------------- +# apply_updates – unsupported package manager +# --------------------------------------------------------------------------- + + +def test_apply_updates_unsupported_pm_raises(driver): + driver._pkg_manager = "zypper" + with pytest.raises(NotImplementedError): + driver.apply_updates(["curl"])