From 84717ff53b5ac88c540dac7b9bde12b1d81781d0 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Mon, 5 Oct 2026 13:12:13 +0200 Subject: [PATCH] feat: start, stop, restart, enable and disable services, and list them in one round trip napalm-device-types' SystemdServicesMixin (2.2.0) now provides get_services() and manage_service(); this driver supplies only the transport (#7): - _run_service_command(): unprivileged reads and root logins run as they are -- the user is asked once per session with "id -u", so a root login on a box without sudo is not prefixed with one. With a sudo password the command goes through _sudo(); without one through "sudo -n", which fails at once instead of hanging the session on a password prompt until the read timeout. - get_services(): one round trip instead of an is-enabled and a show per unit (6.0 s -> 0.8 s on a 184-unit Ubuntu host). A host without systemd still falls back to "service --status-all". - manage_service(): when sudo wants a password netOrk does not have, the failure says how to fix it -- the same hint the apt and SNMP actions give, now one constant (_SUDO_PASSWORD_HINT) instead of two copies. OpenMediaVault and QNAP inherit this driver. OMV gets service control with it; QNAP opts out (napalm-qnap-qts), since QTS has no systemd. README: the sudo option is sudo_password, not secret; manage_service and the systemctl permissions are listed. Closes #7 --- README.md | 24 ++++++++-- napalm_linux/linux.py | 102 +++++++++++++++++++++--------------------- pyproject.toml | 2 +- tests/test_linux.py | 102 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 176 insertions(+), 54 deletions(-) diff --git a/README.md b/README.md index e3fbd61..0ceb4eb 100644 --- a/README.md +++ b/README.md @@ -48,7 +48,8 @@ with Driver( 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) + # "sudo_password": "sudo-pass", # for commands that need root; without it, + # # `sudo -n` (passwordless sudo) is tried # "debugging": True, # enable verbose logging }, ) as dev: @@ -69,6 +70,10 @@ with Driver( # Upgrade everything with pending updates result = dev.apply_updates([]) + + # Restart a service (start, stop, restart, enable, disable) + result = dev.manage_service("cron", "restart") + print(result) # {"success": True, "output": ""} ``` ## Supported NAPALM methods @@ -99,7 +104,8 @@ with Driver( | `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_services()` | ✅ | systemd, one round trip (fallback: SysV `service`) | +| `manage_service(name, action)` | ✅ | systemd: start, stop, restart, enable, disable | | `get_users()` | ✅ | `/etc/passwd` + `/etc/group` | | `get_processes()` | ✅ | `ps axo` | | `get_cron_jobs()` | ✅ | user crontabs + `/etc/cron.d/` | @@ -127,13 +133,25 @@ The SSH user needs read access to: | `/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 | +| `systemctl list-unit-files`, `systemctl show` | unprivileged | +| `systemctl start/stop/restart/enable/disable` | `root`, or `sudo` (with `sudo_password`, or passwordless) | | `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. +`get_services()` and `manage_service()` come from napalm-device-types' +`SystemdServicesMixin`; this driver supplies only the transport. An action runs as +`timeout 45 systemctl --no-ask-password -- .service`, so a unit that hangs +on its way up or down cannot hold the session, and only the exit status decides whether it +succeeded. Without a sudo password it uses `sudo -n`, which fails at once instead of +waiting for a password prompt. + +On OpenMediaVault (napalm-openmediavault inherits this driver), enabling or disabling a +unit that OMV manages itself — Samba, NFS, SSH — may be reverted the next time OMV applies +its configuration. + ## Tested distributions | Distribution | Version | Package manager | Tested | diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index e2d09bd..d195318 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -31,7 +31,13 @@ from netmiko.exceptions import ( ) from napalm.base.exceptions import ConnectionException, ConnectionClosedException from napalm.base.netmiko_helpers import netmiko_args -from napalm_device_types import FingerprintRule, KernelFactsMixin, OSDriver +from napalm_device_types import ( + FingerprintRule, + KernelFactsMixin, + OSDriver, + SystemdServicesMixin, + SystemdUnavailable, +) from napalm_device_types.models import ( ApplyUpdatesResultDict, CronJobDict, @@ -56,6 +62,13 @@ _PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"] _RC_MARKER = "__NETORK_RC=" _RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE) +#: What to do when sudo wants a password netOrk does not have. +_SUDO_PASSWORD_HINT = ( + "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." +) + # DMI field values that carry no useful information (OEM defaults, blanks) _BAD_DMI: frozenset[str] = frozenset({ "", "none", "n/a", "not specified", "not applicable", @@ -138,7 +151,7 @@ def _short_image_id(raw: str) -> str: return raw.strip().removeprefix("sha256:")[:12] -class LinuxDriver(KernelFactsMixin, OSDriver): +class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver): """NAPALM driver for generic Linux systems. Connects via SSH (netmiko ``linux`` device type) and auto-detects the @@ -292,6 +305,28 @@ class LinuxDriver(KernelFactsMixin, OSDriver): output = (raw[: last.start()] + raw[last.end():]).strip() return output, int(last.group(1)) + def _is_root(self) -> bool: + """Whether the SSH user is root, asked once per session. + + A root login on a box without sudo (an LXC container, a minimal Debian) + must not have its commands prefixed with a sudo that is not there. + """ + if getattr(self, "_root", None) is None: + self._root = self._send("id -u") == "0" + return bool(self._root) + + def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str: + """The transport for :class:`SystemdServicesMixin`. + + Without a sudo password, ``sudo -n`` fails at once where a prompt would + otherwise hang the session until the read timeout. + """ + if not privileged or self._is_root(): + return self._send(command, read_timeout=timeout) + if self._sudo_password: + return self._sudo(command, read_timeout=timeout) + return self._send(f"sudo -n {command}", read_timeout=timeout) + def _detect_pkg_manager(self) -> str | None: """Return the first package manager binary found on PATH.""" for pm in _PKG_MANAGERS: @@ -1379,50 +1414,25 @@ class LinuxDriver(KernelFactsMixin, OSDriver): return {"success": False, "output": "", "error": str(exc)} # ------------------------------------------------------------------ - # OSDriver – services (systemd) + # OSDriver – services (systemd, through SystemdServicesMixin) # ------------------------------------------------------------------ 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: + """systemd's services in one round trip; ``service --status-all`` without systemd.""" + try: + return super().get_services() + except SystemdUnavailable: 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" + def manage_service(self, name: str, action: str) -> dict[str, Any]: + """Start, stop, restart, enable or disable a systemd service. - # 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 + :raises ValueError: for an unknown action or an invalid name. + """ + result = super().manage_service(name, action) + if not result["success"] and "password is required" in result["output"]: + result["output"] = f"{result['output']}\n{_SUDO_PASSWORD_HINT}" + return result def _get_services_sysv(self) -> list[ServiceDict]: out = self._send("service --status-all 2>/dev/null") @@ -2110,11 +2120,7 @@ class LinuxDriver(KernelFactsMixin, OSDriver): 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." - ), + "output": _SUDO_PASSWORD_HINT, } lines: list[str] = [] @@ -2140,11 +2146,7 @@ class LinuxDriver(KernelFactsMixin, OSDriver): 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." - ), + "output": _SUDO_PASSWORD_HINT, } # 1. Install snmpd if missing diff --git a/pyproject.toml b/pyproject.toml index 7ac3220..ca07c17 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,7 +37,7 @@ classifiers = [ ] dependencies = [ "napalm>=4.0", - "napalm-device-types>=2.1.0", + "napalm-device-types>=2.2.0", "netmiko>=4.0.0", "paramiko>=5.0.0", # CVE-2026-44405 ] diff --git a/tests/test_linux.py b/tests/test_linux.py index a5b6a94..c09ac00 100644 --- a/tests/test_linux.py +++ b/tests/test_linux.py @@ -1180,3 +1180,105 @@ def test_get_kernel_facts_raises_on_output_without_a_report(driver): with patch.object(driver, "_send", return_value="sh: base64: not found"): with pytest.raises(ValueError): driver.get_kernel_facts() + + +# --------------------------------------------------------------------------- +# Services: listed in one round trip, controlled through systemctl (#7) +# --------------------------------------------------------------------------- + +_REPORT = ( + "SVC_BEGIN\n[files]\ncron.service enabled enabled\n[units]\n" + "MainPID=640\nId=cron.service\nNames=cron.service\nLoadState=loaded\n" + "ActiveState=active\nSubState=running\nUnitFileState=enabled\n" + "[generated]\nSVC_END\n" +) + + +class TestGetServices: + def test_one_command_lists_every_service(self, driver): + driver._device.send_command.return_value = _REPORT + + services = driver.get_services() + + assert services == [{"name": "cron", "running": True, "enabled": True, "pid": 640}] + assert driver._device.send_command.call_count == 1 + assert "systemctl show" in driver._device.send_command.call_args[0][0] + + def test_a_host_without_systemd_falls_back_to_service(self, driver): + driver._device.send_command.side_effect = [ + "SVC_BEGIN\n[no-systemd]\n[files]\n[units]\n[generated]\nSVC_END\n", + " [ + ] cron\n [ - ] rsync\n", + ] + + services = driver.get_services() + + assert {s["name"]: s["running"] for s in services} == {"cron": True, "rsync": False} + assert "service --status-all" in driver._device.send_command.call_args[0][0] + + +class TestManageService: + def _sent(self, driver) -> list[str]: + return [c[0][0] for c in driver._device.send_command.call_args_list] + + def test_as_root_the_command_runs_as_it_is(self, driver): + driver._device.send_command.side_effect = ["0", "__SVC_RC=0"] + + assert driver.manage_service("cron", "restart") == {"success": True, "output": ""} + uid, action = self._sent(driver) + assert uid == "id -u" + assert action.startswith("timeout 45 systemctl --no-ask-password restart -- cron.service") + + def test_with_a_sudo_password_it_goes_through_sudo(self, driver): + driver._sudo_password = "pw" # noqa: S105 + driver._device.send_command.side_effect = ["1000", "__SVC_RC=0"] + + assert driver.manage_service("cron", "stop")["success"] is True + action = self._sent(driver)[1] + assert action.startswith("echo pw | sudo -S") + assert "timeout 45 systemctl --no-ask-password stop -- cron.service" in action + + def test_without_one_sudo_never_waits_for_a_password(self, driver): + driver._device.send_command.side_effect = ["1000", "__SVC_RC=0"] + + driver.manage_service("cron", "enable") + + assert self._sent(driver)[1].startswith("sudo -n timeout 45 systemctl") + + def test_a_missing_sudo_password_is_explained(self, driver): + driver._device.send_command.side_effect = [ + "1000", + "sudo: a password is required\n__SVC_RC=1", + ] + + result = driver.manage_service("cron", "restart") + + assert result["success"] is False + assert "sudo password" in result["output"] + assert "NOPASSWD" in result["output"] + + def test_a_failure_keeps_systemctls_message(self, driver): + driver._device.send_command.side_effect = [ + "0", + "Failed to start nope.service: Unit nope.service not found.\n__SVC_RC=5", + ] + + result = driver.manage_service("nope", "start") + + assert result == { + "success": False, + "output": "Failed to start nope.service: Unit nope.service not found.", + } + + def test_who_the_user_is_is_asked_once(self, driver): + driver._device.send_command.side_effect = ["0", "__SVC_RC=0", "__SVC_RC=0"] + + driver.manage_service("cron", "stop") + driver.manage_service("cron", "start") + + assert self._sent(driver).count("id -u") == 1 + + def test_an_invalid_name_is_refused_before_anything_is_sent(self, driver): + with pytest.raises(ValueError): + driver.manage_service("cron; reboot", "stop") + + assert driver._device.send_command.call_count == 0