feat: ServicePackagesMixin, which packages a systemd service is (NetOrk/netork#1236)

get_service_packages() returns, per systemd service, the installed packages
it is made of: the one shipping its unit file and the one shipping the
binary its ExecStart runs -- not the libraries that binary links against.
One read-only command: systemctl show for every loaded service unit, then
one dpkg -S for all paths (each also across the /usr merge), or rpm -qf per
path. The report is framed; one cut short raises. A driver supplies
_run_service_packages_command, as for KernelFactsMixin.

Version 4.20.0.

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]