Merge pull request 'feat: get_service_packages, which packages a systemd service is (NetOrk/netork#1236)' (#36) from feat/service-packages-1236 into master

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- `get_service_packages()` (napalm-device-types' `ServicePackagesMixin`, NetOrk/netork#1236): which installed packages each systemd service is made of — the one shipping its unit file and the one shipping its `ExecStart` binary. Read-only, on an exec channel under `sh`, without sudo. Requires napalm-device-types 4.20.0; version 0.12.0.
+
 ### Security
 - The SSH host key is verified (NetOrk/netork#1167). `open()` hands netmiko the key the caller expects for the port in `optional_args["ssh_host_keys"]` (as a `known_hosts` file that exists only while it connects, with `ssh_strict`), so any other key is refused before the login (`SshHostKeyMismatch`); without one netmiko accepts the first key. Every key seen goes to `optional_args["ssh_host_key_observer"]`. OpenMediaVault and QNAP get this through `LinuxDriver.open()`. Requires napalm-device-types 4.19.0; version 0.11.0.
 

napalm_linux/linux.py

@@ -47,6 +47,7 @@ from napalm_device_types import (
     OSDriver,
     PackageProxyMixin,
     ServiceClustersMixin,
+    ServicePackagesMixin,
     SnmpAgentMixin,
     SystemIdentityMixin,
     SystemdServicesMixin,
@@ -294,6 +295,7 @@ class LinuxDriver(
     EnsurePackageMixin,
     PackageProxyMixin,
     KernelFactsMixin,
+    ServicePackagesMixin,
     ListeningSocketsMixin,
     VirtualIpsMixin,
     ServiceClustersMixin,
@@ -1164,6 +1166,15 @@ class LinuxDriver(
         """The transport for ``KernelFactsMixin.get_kernel_facts``: read-only, no sudo."""
         return self._send(command, read_timeout=60)
 
+    def _run_service_packages_command(self, command: str) -> str:
+        """The transport for ``ServicePackagesMixin.get_service_packages``.
+
+        An exec channel rather than the PTY -- the command is about 1 kB -- under
+        ``sh``, which it is written for, whatever the login shell. Read-only, so
+        never through sudo.
+        """
+        return self.run_command(f"sh -c {_shlex_quote(command)}", timeout=120).stdout
+
     # ------------------------------------------------------------------
     # ListeningSocketsMixin – the transport for get_listening_sockets
     # ------------------------------------------------------------------

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.11.0"
+version = "0.12.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=4.19.0",
+    "napalm-device-types>=4.20.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_linux.py

@@ -1193,6 +1193,40 @@ def test_get_kernel_facts_raises_on_output_without_a_report(driver):
             driver.get_kernel_facts()
 
 
+# ---------------------------------------------------------------------------
+# get_service_packages -- the command and its parse live in napalm-device-types
+# ---------------------------------------------------------------------------
+
+_SVCPKG = (
+    "SVCPKG_BEGIN\n[units]\nId=nginx.service\n"
+    "FragmentPath=/lib/systemd/system/nginx.service\n"
+    "ExecStart={ path=/usr/sbin/nginx ; argv[]=/usr/sbin/nginx }\n\n"
+    "[dpkg]\nnginx-common: /lib/systemd/system/nginx.service\n"
+    "nginx-core: /usr/sbin/nginx\nSVCPKG_END\n"
+)
+
+
+def test_get_service_packages_runs_the_shared_command_under_sh_on_an_exec_channel(driver):
+    """Read-only, so never through sudo; ``sh`` because the login shell may be another."""
+    from shlex import quote
+
+    from napalm_device_types import CommandResult, ServicePackagesMixin
+    from napalm_device_types.service_packages import SERVICE_PACKAGES_COMMAND
+
+    assert isinstance(driver, ServicePackagesMixin)
+    calls = []
+
+    def run_command(command, *, privileged=False, timeout=60, stdin=None):
+        calls.append((command, privileged))
+        return CommandResult(_SVCPKG, "", 0)
+
+    driver.run_command = run_command
+
+    assert driver.get_service_packages() == {"nginx": ["nginx-common", "nginx-core"]}
+    assert calls == [(f"sh -c {quote(SERVICE_PACKAGES_COMMAND)}", False)]
+    driver._device.send_command.assert_not_called()
+
+
 # ---------------------------------------------------------------------------
 # Services: listed in one round trip, controlled through systemctl (#7)
 # ---------------------------------------------------------------------------