README.md

@@ -123,6 +123,13 @@ everywhere, so the command and its parse are concrete here and a driver supplies
 `_run_kernel_facts_command`. `OSDriver` does not carry it — a Windows host is an OS driver
 too, and `hasattr(driver, "get_kernel_facts")` has to stay truthful.
 
+`ServicePackagesMixin` (`get_service_packages`) is mixed in the same way: which installed
+packages each systemd service is made of — the one shipping its unit file and the one
+shipping the binary its `ExecStart` runs, by `dpkg -S` or `rpm -qf`, not the libraries
+that binary links against. An application needs it to let a statement about a service
+(the postgresql unit) reach the findings on its packages. A driver supplies
+`_run_service_packages_command`.
+
 `HostStatusMixin` (`get_host_status`) is mixed in the same way: whether a Linux host needs
 a reboot to finish an update (`/var/run/reboot-required`, `needs-restarting -r`, or a newer
 kernel of the running flavour installed), whether it patches itself (unattended-upgrades,

napalm_device_types/__init__.py

@@ -52,6 +52,7 @@ instead of being restated on every role that happens to need it:
 * :class:`~napalm_device_types.packages.PackageManagementMixin`
 * :class:`~napalm_device_types.ping_sweep.PingSweepMixin`
 * :class:`~napalm_device_types.service_clusters.ServiceClustersMixin`
+* :class:`~napalm_device_types.service_packages.ServicePackagesMixin`
 * :class:`~napalm_device_types.services.ServiceControlMixin`
 * :class:`~napalm_device_types.systemd.SystemdServicesMixin`
 * :class:`~napalm_device_types.updates.UpdateMixin`
@@ -98,6 +99,11 @@ from napalm_device_types.licenses import LicenseMixin
 from napalm_device_types.host_reboot import HostRebootMixin
 from napalm_device_types.interface_filter import InterfaceFilterMixin
 from napalm_device_types.kernel import KERNEL_FACTS_COMMAND, KernelFactsMixin, parse_kernel_facts
+from napalm_device_types.service_packages import (
+    SERVICE_PACKAGES_COMMAND,
+    ServicePackagesMixin,
+    parse_service_packages,
+)
 from napalm_device_types.lag import add_lag_interfaces
 from napalm_device_types.listening import (
     LISTENING_SOCKETS_COMMAND,
@@ -265,6 +271,9 @@ __all__ = [
     "KernelFactsMixin",
     "KERNEL_FACTS_COMMAND",
     "parse_kernel_facts",
+    "SERVICE_PACKAGES_COMMAND",
+    "ServicePackagesMixin",
+    "parse_service_packages",
     "LISTENING_SOCKETS_COMMAND",
     "ListeningSocketsMixin",
     "ListeningSocketsUnavailable",

napalm_device_types/service_packages.py

@@ -0,0 +1,183 @@
+# -*- coding: utf-8 -*-
+"""Which installed packages a systemd service is made of.
+
+A vulnerability is found on a *package*; a statement about a *service* -- the
+postgresql unit, nginx -- reaches its findings only once it is known which
+packages that service is. Here that is the package shipping the unit file and
+the one shipping the binary its ``ExecStart`` runs: ``nginx-common`` and
+``nginx-core``, ``postgresql-common`` and ``postgresql-16``. Reading it is the
+same on every systemd host, so the command and its parse live here once and a
+driver only carries the command across.
+
+**Only the unit's own packages.** The libraries its binary links against
+(``libssl3`` under nginx) are every service's at once; a statement about nginx
+is not one about libc, and following the dependencies would make it one.
+
+The command is read-only and needs no privileges. It prints ``systemctl show``
+for every loaded service unit and then asks the package manager about every
+path at once: ``dpkg -S`` once for all of them -- one call, since each loads the
+whole file list, and per path that is minutes on an ordinary host -- or
+``rpm -qf`` per path. dpkg knows a file by the path its package shipped, so each
+path is asked across the ``/usr`` merge too. The report is framed; one whose end
+is missing raises, because a list cut short must never read as services that own
+nothing.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Dict, List, Optional, Set, Tuple, TYPE_CHECKING
+
+_BEGIN = "SVCPKG_BEGIN"
+_END = "SVCPKG_END"
+_SUFFIX = ".service"
+
+#: A Debian or RPM package name.
+_PACKAGE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9+._-]*$")
+#: The binary of an ``ExecStart`` as ``systemctl show`` prints it.
+_EXEC_PATH = re.compile(r"\bpath=(/[^ ;]+)")
+
+#: Every unit file and ``ExecStart`` binary in ``$s``, each also across the
+#: ``/usr`` merge, one per line.
+_PATHS = (
+    "printf '%s\\n' \"$s\" | sed -n -e 's/^FragmentPath=\\(\\/.*\\)/\\1/p' "
+    "-e 's/^ExecStart=.*path=\\(\\/[^ ;]*\\).*/\\1/p' | sort -u "
+    "| sed -e 'p' -e 's#^/usr/#/#;t' -e 's#^/#/usr/#'"
+)
+
+#: One line, POSIX ``sh``, read-only. The frame markers are printed in two
+#: halves so that a transport which echoes the command does not show them early.
+SERVICE_PACKAGES_COMMAND = (
+    "s=$(systemctl list-units --type=service --all --no-legend --no-pager --plain 2>/dev/null "
+    "| awk '{print $1}' | tr '\\n' '\\0' | xargs -0 -r systemctl show --no-pager "
+    "-p Id -p FragmentPath -p ExecStart -- 2>/dev/null); "
+    "printf '%s%s\\n' SVCPKG_ BEGIN; echo '[units]'; printf '%s\\n\\n' \"$s\"; "
+    "if command -v dpkg >/dev/null 2>&1; then echo '[dpkg]'; "
+    f"{_PATHS} | tr '\\n' '\\0' | xargs -0 -r dpkg -S 2>/dev/null; "
+    "elif command -v rpm >/dev/null 2>&1; then echo '[rpm]'; "
+    f"{_PATHS} | while IFS= read -r f; do "
+    'printf \'%s\\t%s\\n\' "$f" "$(rpm -qf --qf \'%{NAME} \' "$f" 2>/dev/null)"; done; fi; '
+    "printf '%s%s\\n' SVCPKG_ END"
+)
+
+
+def _report(output: str) -> Dict[str, List[str]]:
+    lines = [line.rstrip() for line in output.splitlines()]
+    stripped = [line.strip() for line in lines]
+    try:
+        start = stripped.index(_BEGIN)
+        end = stripped.index(_END, start)
+    except ValueError:
+        raise ValueError("no intact service package report in the output") from None
+    sections: Dict[str, List[str]] = {}
+    current: Optional[List[str]] = None
+    for line in lines[start + 1 : end]:
+        if line in ("[units]", "[dpkg]", "[rpm]"):
+            current = sections.setdefault(line[1:-1], [])
+        elif current is not None:
+            current.append(line)
+    return sections
+
+
+def _units(lines: List[str]) -> Dict[str, Set[str]]:
+    """``systemctl show`` blocks -> ``{unit: its unit file and binary}``."""
+    units: Dict[str, Set[str]] = {}
+    unit = ""
+    paths: Set[str] = set()
+    for line in [*lines, ""]:
+        key, _, value = line.partition("=")
+        if not line.strip():
+            if unit:
+                units[unit] = paths
+            unit, paths = "", set()
+        elif key == "Id" and value.endswith(_SUFFIX):
+            unit = value[: -len(_SUFFIX)]
+        elif key == "FragmentPath" and value.startswith("/"):
+            paths.add(value)
+        elif key == "ExecStart":
+            match = _EXEC_PATH.search(value)
+            if match:
+                paths.add(match.group(1))
+    return units
+
+
+def _names(raw: str) -> Set[str]:
+    """``libfoo1:amd64, foo-bin`` -> ``{"libfoo1", "foo-bin"}``; nothing else counts."""
+    found = set()
+    for name in raw.replace(",", " ").split():
+        name = name.split(":", 1)[0]
+        if _PACKAGE.match(name):
+            found.add(name)
+    return found
+
+
+def _owners(sections: Dict[str, List[str]]) -> Dict[str, Set[str]]:
+    """``{path: the packages owning it}`` from dpkg's or rpm's answer."""
+    owners: Dict[str, Set[str]] = {}
+    for line in sections.get("dpkg", []):
+        # "pkg[, pkg2][:arch]: /path"; "diversion by x from: /path" is no owner.
+        if line.startswith("diversion by") or ": /" not in line:
+            continue
+        names, _, path = line.partition(": /")
+        owners.setdefault("/" + path.strip(), set()).update(_names(names))
+    for line in sections.get("rpm", []):
+        path, _, names = line.partition("\t")
+        if " is not owned" not in names:
+            owners.setdefault(path.strip(), set()).update(_names(names))
+    return owners
+
+
+def _across_usr_merge(path: str) -> Tuple[str, str]:
+    return (path, path[4:] if path.startswith("/usr/") else "/usr" + path)
+
+
+def parse_service_packages(output: str) -> Dict[str, List[str]]:
+    """Parse what :data:`SERVICE_PACKAGES_COMMAND` printed.
+
+    ``{unit name without ".service": its packages, sorted}``. A unit no package
+    owns -- a unit file an admin wrote -- maps to an empty list: read, and none.
+
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _report(output)
+    owners = _owners(sections)
+    return {
+        unit: sorted(
+            {name for p in paths for alt in _across_usr_merge(p) for name in owners.get(alt, ())}
+        )
+        for unit, paths in _units(sections.get("units", [])).items()
+    }
+
+
+class ServicePackagesMixin:
+    """Adds :meth:`get_service_packages` to a driver that can run a command on a systemd host.
+
+    The template form, like :class:`~napalm_device_types.kernel.KernelFactsMixin`:
+    the reading and its parse are concrete here, and a driver supplies only
+    :meth:`_run_service_packages_command` -- how a command reaches its host.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_service_packages_command(self, command: str) -> str:
+            """Run *command* on the host with ``sh`` and return what it printed."""
+            ...
+
+    def get_service_packages(self) -> Dict[str, List[str]]:
+        """
+        Returns, per systemd service, the installed packages it is made of.
+
+        The package that ships its unit file and the one that ships the binary
+        its ``ExecStart`` runs -- not the libraries that binary links against.
+
+        Example::
+
+            {
+                "nginx": ["nginx-common", "nginx-core"],
+                "postgresql@16-main": ["postgresql-16", "postgresql-common"],
+                "local-backup": [],
+            }
+
+        :raises ValueError: if the host's output carried no intact report.
+        """
+        return parse_service_packages(self._run_service_packages_command(SERVICE_PACKAGES_COMMAND))

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "4.19.0"
+version = "4.20.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_service_packages.py

@@ -0,0 +1,150 @@
+"""get_service_packages: which installed packages a systemd service is (NetOrk/netork#1236).
+
+A vulnerability is found on a *package*; a decision about a *service* -- the
+postgresql unit, nginx -- means something only once it is known which packages
+that service is. The package that ships its unit file and the one that ships the
+binary it starts (``nginx-common`` and ``nginx-core``, ``postgresql-common`` and
+``postgresql-16``) are what the unit is made of. Reading that is the same on
+every systemd host -- ``systemctl show`` and ``dpkg -S`` or ``rpm -qf`` -- so the
+command and its parse live here once, and a driver only carries the command.
+
+Only the unit's own packages: the libraries its binary links against are every
+service's at once, and a decision about nginx is not one about libc.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+
+import pytest
+
+from napalm_device_types import ServicePackagesMixin
+from napalm_device_types.service_packages import (
+    SERVICE_PACKAGES_COMMAND,
+    parse_service_packages,
+)
+
+#: ``systemctl show`` prints the properties in its own order: the name may follow its paths.
+NGINX = """FragmentPath=/lib/systemd/system/nginx.service
+ExecStart={ path=/usr/sbin/nginx ; argv[]=/usr/sbin/nginx -g daemon on; master_process on; ; ignore_errors=no ; start_time=[n/a] ; stop_time=[n/a] ; pid=0 ; code=(null) ; status=0/0 }
+Id=nginx.service
+"""
+CRON = """Id=cron.service
+FragmentPath=/usr/lib/systemd/system/cron.service
+ExecStart={ path=/usr/sbin/cron ; argv[]=/usr/sbin/cron -f $EXTRA_OPTS ; ignore_errors=no }
+"""
+
+
+def _wire(units: str, owners: str = "", *, tool: str = "dpkg", noise: str = "") -> str:
+    return f"{noise}SVCPKG_BEGIN\n[units]\n{units}\n[{tool}]\n{owners}SVCPKG_END\n"
+
+
+DPKG = """nginx-common: /lib/systemd/system/nginx.service
+nginx-core: /usr/sbin/nginx
+cron: /lib/systemd/system/cron.service
+cron: /usr/sbin/cron
+"""
+
+
+class TestParsing:
+    def test_the_unit_file_and_the_binary_name_its_packages(self):
+        assert parse_service_packages(_wire(NGINX, DPKG)) == {
+            "nginx": ["nginx-common", "nginx-core"]
+        }
+
+    def test_dpkg_is_asked_across_the_usr_merge(self):
+        """dpkg knows ``/lib/systemd/system/cron.service``; the unit says ``/usr/lib/...``."""
+        assert parse_service_packages(_wire(CRON, DPKG)) == {"cron": ["cron"]}
+
+    def test_several_units_are_kept_apart(self):
+        found = parse_service_packages(_wire(NGINX + "\n" + CRON, DPKG))
+
+        assert found == {"nginx": ["nginx-common", "nginx-core"], "cron": ["cron"]}
+
+    def test_rpm_names_its_packages_too(self):
+        units = """Id=postgresql-16.service
+FragmentPath=/usr/lib/systemd/system/postgresql-16.service
+ExecStart={ path=/usr/pgsql-16/bin/postmaster ; argv[]=/usr/pgsql-16/bin/postmaster -D ${PGDATA} }
+"""
+        owners = (
+            "/usr/lib/systemd/system/postgresql-16.service\tpostgresql16-server \n"
+            "/lib/systemd/system/postgresql-16.service\tfile /lib/systemd/system/postgresql-16.service is not owned by any package\n"
+            "/usr/pgsql-16/bin/postmaster\tpostgresql16-server \n"
+        )
+
+        found = parse_service_packages(_wire(units, owners, tool="rpm"))
+
+        assert found == {"postgresql-16": ["postgresql16-server"]}
+
+    def test_an_architecture_and_several_owners_are_read(self):
+        units = "Id=foo.service\nExecStart={ path=/usr/bin/foo ; argv[]=/usr/bin/foo }\n"
+        owners = "libfoo1:amd64, foo-bin: /usr/bin/foo\n"
+
+        assert parse_service_packages(_wire(units, owners)) == {"foo": ["foo-bin", "libfoo1"]}
+
+    def test_a_unit_no_package_owns_is_read_as_owning_none(self):
+        """A unit file an admin wrote: read, and nothing owns it -- not unknown."""
+        units = "Id=local-backup.service\nFragmentPath=/etc/systemd/system/local-backup.service\n"
+
+        assert parse_service_packages(_wire(units)) == {"local-backup": []}
+
+    def test_a_diversion_is_no_owner(self):
+        units = "Id=x.service\nExecStart={ path=/bin/sh ; argv[]=/bin/sh -c x }\n"
+        owners = "diversion by dash from: /bin/sh\ndash: /bin/sh\n"
+
+        assert parse_service_packages(_wire(units, owners)) == {"x": ["dash"]}
+
+    def test_whatever_surrounds_the_frame_is_ignored(self):
+        noise = "Last login: today\nprintf '%s%s\\n' SVCPKG_ BEGIN; ...\n"
+
+        found = parse_service_packages(_wire(NGINX, DPKG, noise=noise))
+
+        assert found == {"nginx": ["nginx-common", "nginx-core"]}
+
+    def test_a_report_cut_short_raises(self):
+        """A list cut short must never read as services that own nothing."""
+        with pytest.raises(ValueError):
+            parse_service_packages("SVCPKG_BEGIN\n[units]\n" + NGINX)
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        assert "SVCPKG_BEGIN" not in SERVICE_PACKAGES_COMMAND
+        assert "SVCPKG_END" not in SERVICE_PACKAGES_COMMAND
+
+    def test_it_writes_nothing(self):
+        for verb in ("sudo", " > ", ">>", " rm ", "install", "restart"):
+            assert verb not in SERVICE_PACKAGES_COMMAND
+
+    @pytest.mark.skipif(
+        not os.path.isdir("/run/systemd/system") or shutil.which("dpkg") is None,
+        reason="reads a systemd host with dpkg",
+    )
+    def test_it_runs_and_parses_on_this_host(self):
+        out = subprocess.run(
+            ["sh", "-c", SERVICE_PACKAGES_COMMAND], capture_output=True, text=True, timeout=60
+        ).stdout
+
+        found = parse_service_packages(out)
+
+        assert found, "a systemd host has services"
+        assert any(found.values()), "and a package owns at least one of them"
+        assert all(isinstance(p, str) and p for pkgs in found.values() for p in pkgs)
+
+
+class TestTheMixin:
+    def test_it_hands_its_command_to_the_driver_and_parses_the_answer(self):
+        class Driver(ServicePackagesMixin):
+            def __init__(self):
+                self.sent = []
+
+            def _run_service_packages_command(self, command: str) -> str:
+                self.sent.append(command)
+                return _wire(NGINX, DPKG)
+
+        driver = Driver()
+
+        assert driver.get_service_packages() == {"nginx": ["nginx-common", "nginx-core"]}
+        assert driver.sent == [SERVICE_PACKAGES_COMMAND]