feat: list systemd services in one round trip and control them, once for every driver

Listing a host's services and starting or stopping one is the same on every
host that runs systemd, so the command, its parse, the check of a unit name
and the reading of an action's exit status live here once, and a driver
supplies only the transport (napalm-linux#7, napalm-proxmox#6):

- SYSTEMD_SERVICES_COMMAND: one read-only POSIX sh line. list-unit-files, then
  one systemctl show over every loaded service unit (Id, Names, LoadState,
  ActiveState, SubState, UnitFileState, MainPID), and is-enabled only for
  generated units, whose boot state lives in a SysV script's rc links. Framed;
  [no-systemd] when /run/systemd/system is missing. Replaces an is-enabled and
  a show per unit: 0.8 s instead of 6 s on a 180-unit Ubuntu host.
- parse_systemd_services(): loaded units except not-found, plus installed unit
  files that are not loaded; no templates, no aliases (also not the ones older
  systemd lists as "enabled"). enabled = UnitFileState enabled or
  enabled-runtime, read from systemctl show and never from list-unit-files'
  second column, which has had a preset column after it since systemd 245.
  A report whose end is missing raises ValueError, so a list cut short never
  reads as services that went away; a host without systemd raises
  SystemdUnavailable, a NotImplementedError, so a driver can fall back.
- unit_name() / service_action_command() / parse_action_result(): template
  instances, dots, colons and \xHH escapes accepted; a bare template, a leading
  "-" and anything a shell reads refused. The action runs as
  "timeout 45 systemctl --no-ask-password <action> -- <unit>.service" with its
  exit status printed after it; only that status decides, 124 is not called
  done, and terminal colour codes are dropped. The marker also keeps the output
  from ever being empty, which a transport that retries on an empty answer
  would take as a reason to run the action twice.
- SystemdServicesMixin, in the template form: get_services() and
  manage_service() are concrete, _run_service_command(command, *, privileged,
  timeout) is the driver's hook. Mixed in by the drivers whose host runs
  systemd, not by OSDriver.

Version 2.2.0.

README.md

@@ -73,6 +73,21 @@ is a thin bundle over them — `PackageManagementMixin`, `HealthMetricsMixin`,
 `HostRebootMixin` (`reboot_host`) is mixed into `DeviceTypeDriver` itself, since any
 device may be restartable; like the others it only declares.
 
+`KernelFactsMixin` (`get_kernel_facts`) is the exception that is mixed in by a driver
+rather than by a role base: what a Linux kernel has built and loaded is read the same way
+everywhere, so the command and its parse are concrete here and a driver supplies only
+`_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.
+
+`SystemdServicesMixin` (`get_services`, `manage_service`) is mixed in the same way, by
+the drivers whose host runs systemd. Listing the services, checking a unit name and
+reading an action's exit status are the same on every such host, so they are concrete
+here, and a driver supplies only `_run_service_command(command, *, privileged, timeout)`
+— how a command reaches its host and how it gains root there. The listing is one round
+trip (`list-unit-files` plus one `systemctl show` over every loaded unit) instead of an
+`is-enabled` and a `show` per unit. A host without systemd raises `SystemdUnavailable`,
+a `NotImplementedError`, so a driver can fall back to another init system.
+
 A function class may use the **template form** — public method concrete, the
 device-specific part a `_hook` declared under `if TYPE_CHECKING` — *when the base
 genuinely does work* on the result: normalising, sorting, validating, or orchestrating
@@ -227,6 +242,11 @@ class PfSenseDriver(FirewallDriver):
     def get_vpn_tunnels(self):
         # return Dict[str, VPNTunnelDict]
         ...
+
+    def get_port_forwards(self):
+        # return List[PortForwardDict] — forwards from the WAN only, never a
+        # redirect between internal networks (shared with home gateways)
+        ...
 ```
 
 ### Hypervisor
@@ -242,6 +262,12 @@ class ProxmoxDriver(HypervisorDriver):
 
     def create_vm_snapshot(self, name, snapshot, description="", include_memory=False):
         ...
+
+    def get_vm_cpu_types(self):
+        # optional — return List[VMCpuTypeDict]: the CPU models a new VM may get
+        # on this node, each with its cpuinfo flags and whether the node can run
+        # it; the name goes to create_vm_from_cloud_init(cpu_type=...)
+        ...
 ```
 
 ### OS / Linux

napalm_device_types/__init__.py

@@ -40,11 +40,13 @@ instead of being restated on every role that happens to need it:
 * :class:`~napalm_device_types.health_metrics.HealthMetricsMixin`
 * :class:`~napalm_device_types.host_reboot.HostRebootMixin`
 * :class:`~napalm_device_types.interface_filter.InterfaceFilterMixin`
+* :class:`~napalm_device_types.kernel.KernelFactsMixin`
 * :class:`~napalm_device_types.mac_acl.MacAclMixin`
 * :class:`~napalm_device_types.nat_vpn.NatVpnMixin`
 * :class:`~napalm_device_types.packages.PackageManagementMixin`
 * :class:`~napalm_device_types.ping_sweep.PingSweepMixin`
 * :class:`~napalm_device_types.services.ServiceControlMixin`
+* :class:`~napalm_device_types.systemd.SystemdServicesMixin`
 * :class:`~napalm_device_types.updates.UpdateMixin`
 
 Introspection -- :func:`~napalm_device_types.roles.roles_of`,
@@ -63,6 +65,8 @@ from napalm_device_types.firewall_rules import FirewallRuleMixin
 from napalm_device_types.health_metrics import HealthMetricsMixin
 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.lag import add_lag_interfaces
 from napalm_device_types.mac_acl import MacAclMixin
 from napalm_device_types.media import MediaDriver
 from napalm_device_types.nat_vpn import NatVpnMixin
@@ -71,6 +75,12 @@ from napalm_device_types.phone import PhoneDriver
 from napalm_device_types.ping_sweep import PingSweepMixin, driver_supports_ping
 from napalm_device_types.roles import primary_role_of, role_keys_of, roles_of
 from napalm_device_types.services import ServiceControlMixin
+from napalm_device_types.systemd import (
+    SYSTEMD_SERVICES_COMMAND,
+    SystemdServicesMixin,
+    SystemdUnavailable,
+    parse_systemd_services,
+)
 from napalm_device_types.updates import UpdateMixin
 from napalm_device_types.residential_gateway import ResidentialGatewayDriver
 from napalm_device_types.storage import StorageDriver
@@ -93,6 +103,9 @@ __all__ = [
     "NatVpnMixin",
     "OSDriver",
     "PackageManagementMixin",
+    "KernelFactsMixin",
+    "KERNEL_FACTS_COMMAND",
+    "parse_kernel_facts",
     "PhoneDriver",
     "PingSweepMixin",
     "PortSpec",
@@ -100,7 +113,12 @@ __all__ = [
     "ServiceControlMixin",
     "StorageDriver",
     "SwitchDriver",
+    "SYSTEMD_SERVICES_COMMAND",
+    "SystemdServicesMixin",
+    "SystemdUnavailable",
+    "parse_systemd_services",
     "UpdateMixin",
+    "add_lag_interfaces",
     "driver_supports_ping",
     "normalize_cidr",
     "normalize_mac",

napalm_device_types/hypervisor.py

@@ -21,6 +21,7 @@ from napalm_device_types.models import (
     StorageTargetDict,
     StorageVolumeDict,
     VMConfigDict,
+    VMCpuTypeDict,
     VMDict,
     VMProvisionResultDict,
     VMStatusDict,
@@ -462,6 +463,7 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
             ssh_public_keys: List[str] | None = None,
             disk_resize_gb: int | None = None,
             storage: str | None = None,
+            cpu_type: str | None = None,
             download_timeout: int = 300,
             timeout: int = 180,
         ) -> VMProvisionResultDict:
@@ -503,6 +505,13 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
                   disk on (a name returned by ``get_image_storages()``). If None,
                   the driver auto-detects the first enabled, node-available storage
                   whose content includes "images".
+                cpu_type (string | None) - virtual CPU model for the VM (a name
+                  returned by ``get_vm_cpu_types()``). If None, the driver uses the
+                  entry that listing marks ``default``. A driver without
+                  ``get_vm_cpu_types()`` offers no choice and must reject any
+                  value other than None with ValueError; one that has it raises
+                  ValueError for a name it does not list or that is not
+                  ``available`` on this node, before creating anything.
                 download_timeout (int) - maximum seconds to wait for the image download
                   (skipped entirely if already cached on the hypervisor). Default 300.
                 timeout (int) - maximum seconds to wait for the remaining provisioning
@@ -618,3 +627,23 @@ class HypervisorDriver(PackageManagementMixin, HealthMetricsMixin, DeviceTypeDri
                   as ``create_vm_from_cloud_init``'s ``storage`` argument.
             """
             ...
+
+        def get_vm_cpu_types(self) -> List[VMCpuTypeDict]:
+            """
+            List the virtual CPU models a new VM may be given on the node it will
+            be created on.
+
+            Optional: a hypervisor whose VMs have no per-VM CPU model (VMware
+            sets CPU compatibility per cluster) does not implement it, and a
+            caller then offers no choice.
+
+            Every model the driver knows is listed, also those this node's CPU
+            cannot run -- marked ``available: False`` -- so a picker can show why
+            an option is missing. Exactly one entry is ``default``: the model
+            ``create_vm_from_cloud_init`` uses when ``cpu_type`` is None.
+
+            Returns:
+                List[VMCpuTypeDict] - each entry's ``name`` is directly usable as
+                  ``create_vm_from_cloud_init``'s ``cpu_type`` argument.
+            """
+            ...

napalm_device_types/kernel.py

@@ -0,0 +1,168 @@
+# -*- coding: utf-8 -*-
+"""What the running kernel has built and loaded.
+
+A kernel CVE's exploitability often hangs on code that is simply not there --
+a module that is neither loaded nor shipped, a subsystem the kernel was built
+without. Reading that is identical on every Linux host, so the command and its
+parse live here once and a driver only carries the command across: SSH,
+an API's exec endpoint, whatever it has.
+
+The command is read-only and needs no privileges. It frames its report and
+sends it gzipped and base64-encoded, for two reasons: nothing in the payload can
+then look like a shell prompt to a screen-scraping transport, and a kernel's
+build configuration (~300 kB on a distribution kernel) crosses as a fifth of
+that.
+
+What the four lists mean for a module, and why "not loaded" alone is never
+"absent": a module that is not loaded can still be loaded on demand -- by an
+attacker too, where autoloading reaches it. Only a module that is neither
+loaded, nor compiled in, nor shipped for this kernel is one it cannot have.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import gzip
+import zlib
+from typing import Dict, List, Optional, TYPE_CHECKING
+
+from napalm_device_types.models import KernelFactsDict
+
+_BEGIN = "KFACTS_BEGIN"
+_END = "KFACTS_END"
+
+#: 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.
+KERNEL_FACTS_COMMAND = (
+    "r=$(uname -r); m=/lib/modules/$r; "
+    "printf '%s%s\\n' KFACTS_ BEGIN; "
+    "{ echo '[release]'; echo \"$r\"; "
+    "if [ -r /proc/modules ]; then echo '[loaded]'; cut -d' ' -f1 /proc/modules; fi; "
+    "if [ -r $m/modules.builtin ]; then echo '[builtin]'; cat $m/modules.builtin; fi; "
+    "if [ -r $m/modules.dep ]; then echo '[available]'; cut -d: -f1 $m/modules.dep; fi; "
+    "if [ -r /boot/config-$r ]; then echo '[config]'; grep '^CONFIG_' /boot/config-$r; "
+    "elif [ -r /proc/config.gz ]; then echo '[config]'; zcat /proc/config.gz | grep '^CONFIG_'; fi; "
+    "} 2>/dev/null | gzip -c | base64; "
+    "printf '%s%s\\n' KFACTS_ END"
+)
+
+
+def module_name(raw: str) -> str:
+    """A module as the kernel names it: no path, no ``.ko`` suffix, ``_`` for ``-``.
+
+    ``kernel/net/can/can-raw.ko.zst`` and ``can_raw`` are the same module; the
+    kernel itself treats dash and underscore alike.
+    """
+    base = raw.strip().rsplit("/", 1)[-1]
+    suffix = base.find(".ko")
+    if suffix != -1:
+        base = base[:suffix]
+    return base.replace("-", "_").lower()
+
+
+def _report(output: str) -> str:
+    lines = [line.strip() for line in output.splitlines()]
+    try:
+        start = lines.index(_BEGIN)
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("no kernel facts in the output") from None
+    try:
+        packed = base64.b64decode("".join(lines[start + 1 : end]), validate=True)
+        return gzip.decompress(packed).decode()
+    except (binascii.Error, OSError, EOFError, zlib.error, UnicodeDecodeError) as exc:
+        raise ValueError(f"the kernel facts could not be decoded: {exc}") from exc
+
+
+def _sections(report: str) -> Dict[str, List[str]]:
+    sections: Dict[str, List[str]] = {}
+    current: Optional[List[str]] = None
+    for line in report.splitlines():
+        line = line.strip()
+        if line.startswith("[") and line.endswith("]"):
+            current = sections.setdefault(line[1:-1], [])
+        elif line and current is not None:
+            current.append(line)
+    return sections
+
+
+def _config(lines: List[str]) -> Dict[str, str]:
+    config: Dict[str, str] = {}
+    for line in lines:
+        option, sep, value = line.partition("=")
+        if not sep:
+            continue
+        if len(value) >= 2 and value[0] == value[-1] == '"':
+            value = value[1:-1]
+        config[option] = value
+    return config
+
+
+def parse_kernel_facts(output: str) -> KernelFactsDict:
+    """Parse what :data:`KERNEL_FACTS_COMMAND` printed.
+
+    A section the command did not print -- the file was missing or unreadable --
+    comes back ``None``, never empty.
+
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _sections(_report(output))
+
+    def names(key: str) -> Optional[List[str]]:
+        if key not in sections:
+            return None
+        return sorted({module_name(line) for line in sections[key]})
+
+    release = sections.get("release") or [""]
+    return {
+        "release": release[0],
+        "loaded": names("loaded"),
+        "builtin": names("builtin"),
+        "available": names("available"),
+        "config": _config(sections["config"]) if "config" in sections else None,
+    }
+
+
+class KernelFactsMixin:
+    """Adds :meth:`get_kernel_facts` to a driver that can run a command on a Linux host.
+
+    The template form (README, "Function classes"): the reading and its parse are
+    the same everywhere, so they are concrete here, and a driver supplies only
+    :meth:`_run_kernel_facts_command` -- how a command reaches its host. Mixed in
+    by the drivers that can, not by :class:`~napalm_device_types.os.OSDriver`:
+    a Windows host is an OS driver too and has no Linux kernel to read, and
+    ``hasattr(driver, "get_kernel_facts")`` has to stay a truthful answer.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_kernel_facts_command(self, command: str) -> str:
+            """Run *command* on the host with ``sh`` and return what it printed."""
+            ...
+
+    def get_kernel_facts(self) -> KernelFactsDict:
+        """
+        Returns what the running kernel has built and loaded.
+
+        * release (string) - ``uname -r``
+        * loaded (list or None) - loaded modules, from ``/proc/modules``
+        * builtin (list or None) - modules compiled into the kernel image
+        * available (list or None) - modules shipped for this kernel
+        * config (dict or None) - the build configuration's set options
+
+        ``None`` means the source could not be read.
+
+        Example::
+
+            {
+                "release": "6.1.0-25-amd64",
+                "loaded": ["nf_tables", "tipc"],
+                "builtin": ["tcp_cubic"],
+                "available": ["can_raw", "nf_tables", "tipc"],
+                "config": {"CONFIG_TIPC": "m", "CONFIG_HZ": "250"},
+            }
+
+        :raises ValueError: if the host's output carried no intact report.
+        """
+        return parse_kernel_facts(self._run_kernel_facts_command(KERNEL_FACTS_COMMAND))

napalm_device_types/lag.py

@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+"""Logical LAG entries for ``get_interfaces()``, built from their member ports.
+
+Some switches list only physical ports, each tagged with the trunk it belongs
+to, and never the trunk itself. Turning those tags into one row per trunk is
+the same for every vendor, so it lives here once; a driver only has to set
+``trunk_group`` on member ports and, where the device says so, pass the mode.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, List, Optional
+
+
+def _port_order(name: str) -> List[Any]:
+    return [int(p) if p.isdigit() else p for p in re.split(r"(\d+)", name)]
+
+
+def add_lag_interfaces(
+    interfaces: Dict[str, Dict[str, Any]],
+    lag_modes: Optional[Dict[str, str]] = None,
+) -> Dict[str, Dict[str, Any]]:
+    """Return *interfaces* plus one logical entry per ``trunk_group``.
+
+    The LAG entry is up/enabled if any member is, its speed is the members'
+    sum, and ``lag_members`` lists them in port order. A LAG the driver
+    already reported is left as it is. *interfaces* itself is not modified.
+
+    :param lag_modes: ``{lag_name: "lacp" | "trunk"}``. A LAG without a known
+        mode gets no ``lag_mode`` key rather than a guessed one.
+    """
+    result = dict(interfaces)
+    groups: Dict[str, List[str]] = {}
+    for name, iface in interfaces.items():
+        group = iface.get("trunk_group")
+        if group:
+            groups.setdefault(group, []).append(name)
+
+    for group, members in groups.items():
+        if group in result:
+            continue
+        members = sorted(members, key=_port_order)
+        lag: Dict[str, Any] = {
+            "is_up": any(interfaces[m].get("is_up") for m in members),
+            "is_enabled": any(interfaces[m].get("is_enabled") for m in members),
+            "description": f"LAG ({', '.join(members)})",
+            "last_flapped": -1.0,
+            "speed": sum(float(interfaces[m].get("speed") or 0) for m in members),
+            "mtu": -1,
+            "mac_address": "",
+            "lag_members": members,
+        }
+        if lag_modes and group in lag_modes:
+            lag["lag_mode"] = lag_modes[group]
+        result[group] = lag
+    return result

napalm_device_types/models.py

@@ -298,6 +298,39 @@ class NATTranslationDict(TypedDict):
     age: float
 
 
+class KernelFactsDict(TypedDict):
+    """What the running kernel has built and loaded (``KernelFactsMixin.get_kernel_facts``).
+
+    ``None`` means *could not be read*; an empty list means *read, and there is
+    nothing*. The difference is what lets a consumer say "this module cannot be
+    loaded on this kernel" rather than "we did not look".
+
+    Module names are normalised by :func:`napalm_device_types.kernel.module_name`:
+    no path, no ``.ko`` suffix, ``-`` folded to ``_``.
+    """
+
+    release: str  # uname -r
+    loaded: Optional[List[str]]  # /proc/modules
+    builtin: Optional[List[str]]  # modules.builtin -- compiled into the kernel image
+    available: Optional[List[str]]  # modules.dep -- shipped as loadable modules
+    config: Optional[Dict[str, str]]  # build configuration, set options only; quotes stripped
+
+
+class PortForwardDict(TypedDict):
+    """A port the WAN side can reach, forwarded to a host inside.
+
+    Shared by firewalls and home gateways (``NatVpnMixin.get_port_forwards``).
+    """
+
+    name: str
+    protocol: str  # "TCP" or "UDP"
+    external_port: int
+    internal_ip: str
+    internal_port: int
+    enabled: bool
+    remote_host: NotRequired[str]  # restrict forward to a specific remote source
+
+
 class SecurityZoneDict(TypedDict):
     interfaces: List[str]
     policy: str
@@ -470,16 +503,6 @@ class WANStatusDict(TypedDict):
     link_status: NotRequired[str]  # physical line state, e.g. "Up" / "Down"
 
 
-class PortForwardDict(TypedDict):
-    name: str
-    protocol: str  # "TCP" or "UDP"
-    external_port: int
-    internal_ip: str
-    internal_port: int
-    enabled: bool
-    remote_host: NotRequired[str]  # restrict forward to a specific remote source
-
-
 class HostDict(TypedDict):
     mac: str
     ip: str
@@ -907,6 +930,24 @@ class StorageTargetDict(TypedDict):
     available_gb: float  # Free capacity in gigabytes
 
 
+class VMCpuTypeDict(TypedDict):
+    """A virtual CPU model a new VM may be given (``create_vm_from_cloud_init``'s
+    ``cpu_type`` argument), judged against the specific node the VM will be
+    created on.
+
+    ``features`` uses the flag names of Linux's ``/proc/cpuinfo`` (``avx``,
+    ``avx2``, ``aes`` ...), so a caller can ask "does this model give the guest
+    AVX?" without knowing the hypervisor's model names. A model that passes the
+    host CPU through lists that CPU's own flags.
+    """
+
+    name: str  # Model name, usable directly as create_vm_from_cloud_init(cpu_type=...)
+    description: str  # One line on what the model is for, for a picker
+    features: List[str]  # cpuinfo flags the guest is guaranteed to see
+    available: bool  # False when this node's CPU cannot run the model
+    default: bool  # The model create_vm_from_cloud_init uses when cpu_type is None
+
+
 # ---------------------------------------------------------------------------
 # Ping sweep  (shared across device types)
 # ---------------------------------------------------------------------------

napalm_device_types/nat_vpn.py

@@ -1,8 +1,8 @@
 # -*- coding: utf-8 -*-
 """Address translation and VPN tunnels.
 
-A home gateway does a subset of what a firewall does, and these two readers
-are where the sets overlap exactly.
+A home gateway does a subset of what a firewall does, and these readers are
+where the sets overlap exactly.
 
 Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
 Nothing exists at runtime until a concrete driver implements it, so mixing
@@ -13,7 +13,7 @@ from __future__ import annotations
 
 from typing import Dict, List, TYPE_CHECKING
 
-from napalm_device_types.models import NATTranslationDict, VPNTunnelDict
+from napalm_device_types.models import NATTranslationDict, PortForwardDict, VPNTunnelDict
 
 
 class NatVpnMixin:
@@ -47,6 +47,46 @@ class NatVpnMixin:
             """
             ...
 
+        def get_port_forwards(self) -> List[PortForwardDict]:
+            """
+            Returns the port forwards that let traffic in from the WAN.
+
+            A port forward here means destination NAT on an interface facing
+            the internet: whoever reaches the external port is let through to
+            ``internal_ip``. A redirect between internal networks is
+            destination NAT as well, but it is **not** a port forward and must
+            be left out -- callers read every entry as "this host is reachable
+            from outside". So are rules that only exempt traffic from
+            redirection.
+
+            Each entry contains:
+
+            * name (string) - the rule's description/name
+            * protocol (string) - ``"TCP"`` or ``"UDP"``; a rule for both is
+              two entries. ``"ANY"`` forwards every protocol
+            * external_port (int) - the WAN-side port; the first of a range,
+              ``0`` for every port (a whole host forwarded)
+            * internal_ip (string) - the host the traffic is forwarded to
+            * internal_port (int) - the port on that host
+            * enabled (bool) - whether the rule is currently active
+            * remote_host (string, optional) - restricts the forward to a specific
+              remote source address; empty/absent means "any"
+
+            Example::
+
+                [
+                    {
+                        "name": "Webserver HTTPS",
+                        "protocol": "TCP",
+                        "external_port": 443,
+                        "internal_ip": "192.168.1.10",
+                        "internal_port": 443,
+                        "enabled": True,
+                    }
+                ]
+            """
+            ...
+
         def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]:
             """
             Returns the status of VPN tunnels.

napalm_device_types/residential_gateway.py

@@ -6,8 +6,9 @@ wireless access point in a single consumer device (e.g. AVM FritzBox,
 ISP-supplied DSL/cable routers). This base class merges the relevant
 subsets of :class:`~napalm_device_types.firewall.FirewallDriver` and
 :class:`~napalm_device_types.access_point.AccessPointDriver` plus
-gateway-specific operations (WAN status, port forwarding, connected
-hosts).
+gateway-specific operations (WAN status, connected hosts). Port
+forwarding is shared with firewalls, in
+:class:`~napalm_device_types.nat_vpn.NatVpnMixin`.
 
 Usage::
 
@@ -25,7 +26,6 @@ from napalm_device_types.health_metrics import HealthMetricsMixin
 from napalm_device_types.dhcp import DhcpServerMixin
 from napalm_device_types.models import (
     HostDict,
-    PortForwardDict,
     RadioStatusDict,
     SSIDDict,
     WANStatusDict,
@@ -85,36 +85,6 @@ class ResidentialGatewayDriver(NatVpnMixin, HealthMetricsMixin, DhcpServerMixin,
             """
             ...
 
-        def get_port_forwards(self) -> List[PortForwardDict]:
-            """
-            Returns the configured port forwarding (port mapping) rules.
-
-            Each entry contains:
-
-            * name (string) - the rule's description/name
-            * protocol (string) - ``"TCP"`` or ``"UDP"``
-            * external_port (int) - the WAN-side port
-            * internal_ip (string) - the LAN host the traffic is forwarded to
-            * internal_port (int) - the LAN-side port
-            * enabled (bool) - whether the rule is currently active
-            * remote_host (string, optional) - restricts the forward to a specific
-              remote source address; empty/absent means "any"
-
-            Example::
-
-                [
-                    {
-                        "name": "Webserver HTTPS",
-                        "protocol": "TCP",
-                        "external_port": 443,
-                        "internal_ip": "192.168.1.10",
-                        "internal_port": 443,
-                        "enabled": True,
-                    }
-                ]
-            """
-            ...
-
         def get_hosts(self) -> List[HostDict]:
             """
             Returns the list of hosts known to the gateway (LAN clients).

napalm_device_types/systemd.py

@@ -0,0 +1,331 @@
+# -*- coding: utf-8 -*-
+"""systemd services: listing them in one round trip, and starting and stopping them.
+
+What systemd reports about its services, and how one is started or stopped, is
+the same on every host that runs it. So the command, its parse, the check of a
+unit name and the reading of an action's exit status live here once, and a
+driver only carries a command across: SSH, an API's exec endpoint, whatever it
+has.
+
+**Listing.** One command prints the installed unit files and, for every loaded
+service unit, what ``systemctl show`` knows about it -- state, boot state and
+main PID together, instead of asking ``systemctl is-enabled`` and ``systemctl
+show`` once per unit (two hundred round trips on an ordinary Linux host). The
+report is framed, and a report whose end is missing raises: a list cut short
+must never read as services that went away.
+
+**What counts as enabled.** A unit file state of ``enabled`` or
+``enabled-runtime``. ``static`` does not: such a unit starts only when
+something else pulls it in, and calling it enabled made every one of them look
+like a service of the host. The state is read from ``UnitFileState``, never
+from a column of ``list-unit-files``, whose second column has been followed by
+a preset column since systemd 245.
+
+**Starting and stopping.** ``systemctl`` runs bounded by ``timeout`` and never
+asks for a password, and its exit status is printed after it. The marker also
+keeps the output from ever being empty, which a transport that retries on an
+empty answer would otherwise take as a reason to run the action twice.
+"""
+
+from __future__ import annotations
+
+import re
+from shlex import quote
+from typing import Any, Dict, List, Set, Tuple, TYPE_CHECKING
+
+from napalm_device_types.models import ServiceDict
+from napalm_device_types.services import ServiceControlMixin
+
+_BEGIN = "SVC_BEGIN"
+_END = "SVC_END"
+_NO_SYSTEMD = "no-systemd"
+_SUFFIX = ".service"
+
+#: What ``systemctl show`` prints per unit. It prints them in its own order.
+_PROPERTIES = "Id,Names,LoadState,ActiveState,SubState,UnitFileState,MainPID"
+
+#: Picks the units whose file state is ``generated`` out of ``systemctl show``'s
+#: output, whatever order it prints the properties in.
+_GENERATED_AWK = (
+    'awk -F= \'NF<2{id="";g=0;next} $1=="Id"{id=$2} '
+    '$1=="UnitFileState"{g=($2=="generated")} id!=""&&g{print id;id="";g=0}\''
+)
+
+#: 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.
+#: ``xargs -0`` passes escaped names such as ``foo\x2dbar.service`` unchanged.
+#: A generated unit -- the wrapper systemd makes for a SysV script -- has no unit
+#: file whose state says whether it starts at boot; ``systemctl is-enabled``
+#: asks the script's rc links instead, for those few units only.
+SYSTEMD_SERVICES_COMMAND = (
+    "printf '%s%s\\n' SVC_ BEGIN; "
+    "[ -d /run/systemd/system ] || echo '[no-systemd]'; "
+    "echo '[files]'; systemctl list-unit-files --type=service --no-legend --no-pager 2>/dev/null; "
+    "echo '[units]'; 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 "
+    f"-p {_PROPERTIES} -- 2>/dev/null); printf '%s\\n' \"$s\"; "
+    f"echo '[generated]'; printf '%s\\n' \"$s\" | {_GENERATED_AWK} | while read -r u; do "
+    'printf \'%s %s\\n\' "$u" "$(systemctl is-enabled -- "$u" 2>/dev/null)"; done; '
+    "printf '%s%s\\n' SVC_ END"
+)
+
+#: The lifecycle actions :meth:`SystemdServicesMixin.manage_service` accepts.
+SERVICE_ACTIONS = ("start", "stop", "restart", "enable", "disable")
+
+#: Seconds an action may run on the host before ``timeout`` stops waiting for
+#: it. systemd itself carries on with the job.
+ACTION_TIMEOUT = 45
+
+#: What a transport should allow for one command: the action's own bound plus
+#: the round trip around it.
+_TRANSPORT_TIMEOUT = ACTION_TIMEOUT + 15
+
+_TIMED_OUT = 124  # timeout(1)'s exit status when the time ran out
+_RC_MARKER = "__SVC_RC="
+_RC_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
+
+#: The characters systemd allows in a unit name, with ``\xHH`` for any other byte.
+_UNIT_RE = re.compile(r"(?:[A-Za-z0-9_.:@-]|\\x[0-9A-Fa-f]{2})+")
+_MAX_UNIT_LENGTH = 255
+
+#: Terminal colour codes, which systemctl adds when a transport gives it a terminal.
+_ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]")
+
+_ENABLED = frozenset({"enabled", "enabled-runtime"})
+#: Unit file states of a service that is installed but need not be loaded.
+_INSTALLED = frozenset({"enabled", "enabled-runtime", "disabled", "indirect"})
+
+
+class SystemdUnavailable(NotImplementedError):
+    """The host does not run systemd; a driver may fall back to another init system."""
+
+
+def unit_name(name: str) -> str:
+    """*name* as a service unit's name without ``.service``, or ``ValueError``.
+
+    Accepts template instances (``wg-quick@wg0``), dots (``snapd.apparmor``),
+    colons and systemd's ``\\xHH`` escapes. Refuses a bare template
+    (``getty@``), a leading ``-`` that a command would read as an option, and
+    anything a shell would read.
+    """
+    base = name[: -len(_SUFFIX)] if name.endswith(_SUFFIX) else name
+    if (
+        not _UNIT_RE.fullmatch(base)
+        or base.startswith("-")
+        or base.endswith("@")
+        or len(base) + len(_SUFFIX) > _MAX_UNIT_LENGTH
+    ):
+        raise ValueError(f"Invalid service name: {name!r}")
+    return base
+
+
+def service_action_command(name: str, action: str) -> str:
+    """The shell command that applies *action* to the service *name*.
+
+    :raises ValueError: for an unknown action or an invalid name.
+    """
+    if action not in SERVICE_ACTIONS:
+        raise ValueError(f"Invalid action {action!r}; use one of {', '.join(SERVICE_ACTIONS)}")
+    unit = quote(unit_name(name) + _SUFFIX)
+    return (
+        f"timeout {ACTION_TIMEOUT} systemctl --no-ask-password {action} -- {unit} 2>&1; "
+        f"echo {_RC_MARKER}$?"
+    )
+
+
+def parse_action_result(output: str) -> Dict[str, Any]:
+    """``{"success", "output"}`` from what :func:`service_action_command` printed.
+
+    Only the exit status decides. A job still running when ``timeout`` gave up
+    is not reported as done, and output without a status is no success.
+    """
+    output = _ANSI_RE.sub("", output)
+    statuses = _RC_RE.findall(output)
+    text = _RC_RE.sub("", output).strip()
+    if not statuses:
+        return {"success": False, "output": text or "No exit status came back from the host."}
+    status = int(statuses[-1])
+    if status == 0:
+        return {"success": True, "output": text}
+    if status == _TIMED_OUT:
+        note = f"Still running after {ACTION_TIMEOUT} s; systemd carries on with the job."
+        return {"success": False, "output": f"{text}\n{note}".strip()}
+    return {"success": False, "output": text or f"systemctl exited with status {status}."}
+
+
+def _frame(output: str) -> List[str]:
+    lines = [line.strip() for line in _ANSI_RE.sub("", output).splitlines()]
+    try:
+        start = lines.index(_BEGIN)
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("no intact systemd service report in the output") from None
+    return lines[start + 1 : end]
+
+
+def _sections(lines: List[str]) -> Dict[str, List[str]]:
+    sections: Dict[str, List[str]] = {}
+    current: List[str] = []
+    for line in lines:
+        if line.startswith("[") and line.endswith("]"):
+            current = sections.setdefault(line[1:-1], [])
+        else:
+            current.append(line)
+    return sections
+
+
+def _unit_blocks(lines: List[str]) -> List[Dict[str, str]]:
+    """``systemctl show``'s output, one dict per unit.
+
+    Units are separated by a blank line -- except where ``xargs`` split the
+    list over two runs and the blocks meet, so a key seen twice starts the next
+    unit as well.
+    """
+    blocks: List[Dict[str, str]] = []
+    current: Dict[str, str] = {}
+    for line in lines:
+        key, sep, value = line.partition("=")
+        if not sep or key in current:
+            if current:
+                blocks.append(current)
+            current = {}
+        if sep:
+            current[key] = value
+    if current:
+        blocks.append(current)
+    return blocks
+
+
+def _base(unit: str) -> str:
+    return unit[: -len(_SUFFIX)]
+
+
+def _main_pid(block: Dict[str, str]) -> int:
+    try:
+        return int(block.get("MainPID") or 0)
+    except ValueError:
+        return 0
+
+
+def _loaded(blocks: List[Dict[str, str]]) -> Tuple[Dict[str, ServiceDict], Set[str]]:
+    """The loaded services, and every name they go by (aliases included)."""
+    services: Dict[str, ServiceDict] = {}
+    names: Set[str] = set()
+    for block in blocks:
+        unit = block.get("Id", "")
+        if not unit.endswith(_SUFFIX) or block.get("LoadState") == "not-found":
+            continue
+        names.update(block.get("Names", unit).split())
+        running = block.get("ActiveState") == "active" and block.get("SubState") == "running"
+        services[_base(unit)] = {
+            "name": _base(unit),
+            "running": running,
+            "enabled": block.get("UnitFileState") in _ENABLED,
+            "pid": _main_pid(block) if running else 0,
+        }
+    return services, names
+
+
+def _installed(lines: List[str], known: Set[str]) -> Dict[str, ServiceDict]:
+    """Installed services that are not loaded: neither running nor starting now.
+
+    Templates, static units and aliases are left out -- the last also when an
+    older systemd lists an alias as ``enabled``, which is why every name a
+    loaded unit goes by is skipped.
+    """
+    services: Dict[str, ServiceDict] = {}
+    for line in lines:
+        parts = line.split()
+        if len(parts) < 2:
+            continue
+        unit, state = parts[0], parts[1]
+        if (
+            not unit.endswith(_SUFFIX)
+            or unit.endswith("@" + _SUFFIX)
+            or unit in known
+            or state not in _INSTALLED
+        ):
+            continue
+        services[_base(unit)] = {
+            "name": _base(unit),
+            "running": False,
+            "enabled": state in _ENABLED,
+            "pid": 0,
+        }
+    return services
+
+
+def _apply_generated(services: Dict[str, ServiceDict], lines: List[str]) -> None:
+    """Take a generated unit's boot state from ``is-enabled``'s answer."""
+    for line in lines:
+        parts = line.split()
+        if len(parts) == 2 and parts[0].endswith(_SUFFIX) and _base(parts[0]) in services:
+            services[_base(parts[0])]["enabled"] = parts[1] in _ENABLED
+
+
+def parse_systemd_services(output: str) -> List[ServiceDict]:
+    """Parse what :data:`SYSTEMD_SERVICES_COMMAND` printed, sorted by name.
+
+    Lists every loaded service unit but those that are not found, and every
+    installed one that is not loaded.
+
+    :raises SystemdUnavailable: when the host does not run systemd.
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _sections(_frame(output))
+    if _NO_SYSTEMD in sections:
+        raise SystemdUnavailable("the host does not run systemd")
+    loaded, known = _loaded(_unit_blocks(sections.get("units", [])))
+    _apply_generated(loaded, sections.get("generated", []))
+    merged = {**_installed(sections.get("files", []), known), **loaded}
+    return [merged[name] for name in sorted(merged)]
+
+
+class SystemdServicesMixin(ServiceControlMixin):
+    """Implements :class:`ServiceControlMixin` for a driver whose host runs systemd.
+
+    The template form (README, "Function classes"): the command, the parse,
+    the check of the name and the reading of the exit status are the same
+    everywhere, so they are concrete here, and a driver supplies only
+    :meth:`_run_service_command` -- how a command reaches its host, and how it
+    gains root there when it needs to.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
+            """Run *command* with ``sh`` on the host and return what it printed.
+
+            *privileged* commands change the system and need root; *timeout*
+            is how long the transport should wait for the output, in seconds.
+            """
+            ...
+
+    def get_services(self) -> List[ServiceDict]:
+        """
+        Returns the services systemd knows, in one round trip.
+
+        * name (string) - the unit name without ``.service``
+        * running (bool) - active and running
+        * enabled (bool) - the unit file is enabled
+        * pid (int) - the main process; 0 when not running
+
+        :raises SystemdUnavailable: if the host does not run systemd.
+        :raises ValueError: if the host's output carried no intact report.
+        """
+        output = self._run_service_command(
+            SYSTEMD_SERVICES_COMMAND, privileged=False, timeout=_TRANSPORT_TIMEOUT
+        )
+        return parse_systemd_services(output)
+
+    def manage_service(self, name: str, action: str) -> Dict[str, Any]:
+        """
+        Applies *action* (start, stop, restart, enable, disable) to the service *name*.
+
+        :returns: ``{"success": bool, "output": str}``
+        :raises ValueError: for an unknown action or an invalid name, before
+            anything is sent.
+        """
+        command = service_action_command(name, action)
+        output = self._run_service_command(command, privileged=True, timeout=_TRANSPORT_TIMEOUT)
+        return parse_action_result(output)

pyproject.toml

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

tests/test_kernel_facts.py

@@ -0,0 +1,151 @@
+"""get_kernel_facts: what the running kernel has built and loaded.
+
+A kernel CVE's preconditions ask whether a module is loaded or a build option
+set. Reading that is the same on every Linux host -- one read-only command and
+its parse -- so both live here once, and a driver only carries the command
+across (#268 in netOrk).
+"""
+
+from __future__ import annotations
+
+import base64
+import gzip
+import os
+import subprocess
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.kernel import (
+    KERNEL_FACTS_COMMAND,
+    KernelFactsMixin,
+    module_name,
+    parse_kernel_facts,
+)
+
+REPORT = """[release]
+6.1.0-25-amd64
+[loaded]
+tipc
+nf_tables
+[builtin]
+kernel/net/ipv4/tcp_cubic.ko
+kernel/drivers/char/tpm/tpm-tis.ko
+[available]
+kernel/net/tipc/tipc.ko.xz
+kernel/net/can/can-raw.ko.zst
+kernel/net/netfilter/nf_tables.ko
+[config]
+CONFIG_TIPC=m
+CONFIG_BPF_JIT=y
+CONFIG_DEFAULT_HOSTNAME="(none)"
+CONFIG_HZ=250
+"""
+
+
+def _wire(report: str, *, noise: str = "") -> str:
+    """The report as the command prints it: framed, gzipped, base64 in lines."""
+    payload = base64.encodebytes(gzip.compress(report.encode())).decode()
+    return f"{noise}KFACTS_BEGIN\n{payload}KFACTS_END\n"
+
+
+class TestParsing:
+    def test_every_section_is_read(self):
+        facts = parse_kernel_facts(_wire(REPORT))
+
+        assert facts["release"] == "6.1.0-25-amd64"
+        assert facts["loaded"] == ["nf_tables", "tipc"]
+        assert facts["builtin"] == ["tcp_cubic", "tpm_tis"]
+        assert facts["available"] == ["can_raw", "nf_tables", "tipc"]
+        assert facts["config"] == {
+            "CONFIG_TIPC": "m",
+            "CONFIG_BPF_JIT": "y",
+            "CONFIG_DEFAULT_HOSTNAME": "(none)",
+            "CONFIG_HZ": "250",
+        }
+
+    def test_a_section_never_printed_is_none_not_empty(self):
+        """``None`` is "could not read"; an empty list would claim "read it,
+        and there is nothing" -- and that is what turns a module into
+        ``not_met`` downstream."""
+        facts = parse_kernel_facts(_wire("[release]\n6.1.0\n[loaded]\n"))
+
+        assert facts["loaded"] == []
+        assert facts["builtin"] is None
+        assert facts["available"] is None
+        assert facts["config"] is None
+
+    def test_whatever_surrounds_the_frame_is_ignored(self):
+        """A screen-scraping transport may echo the command or a banner."""
+        noise = "Last login: today\nprintf '%s%s\\n' KFACTS_ BEGIN; ...\n"
+
+        assert parse_kernel_facts(_wire(REPORT, noise=noise))["release"] == "6.1.0-25-amd64"
+
+    def test_output_without_the_frame_raises(self):
+        with pytest.raises(ValueError):
+            parse_kernel_facts("sh: gzip: not found\n")
+
+    def test_a_damaged_payload_raises(self):
+        with pytest.raises(ValueError):
+            parse_kernel_facts("KFACTS_BEGIN\nnot base64 at all!\nKFACTS_END\n")
+
+
+class TestModuleNames:
+    @pytest.mark.parametrize(
+        "raw, name",
+        [
+            ("tipc", "tipc"),
+            ("kernel/net/tipc/tipc.ko", "tipc"),
+            ("kernel/net/tipc/tipc.ko.zst", "tipc"),
+            ("kernel/net/can/can-raw.ko.xz", "can_raw"),
+            ("CAN-RAW", "can_raw"),
+            ("  nf_tables  ", "nf_tables"),
+        ],
+    )
+    def test_dash_and_underscore_are_one_name(self, raw, name):
+        """The kernel treats ``-`` and ``_`` in module names as the same."""
+        assert module_name(raw) == name
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        """An echoing transport prints the command back; the markers must only
+        appear once the command has run."""
+        assert "KFACTS_BEGIN" not in KERNEL_FACTS_COMMAND
+        assert "KFACTS_END" not in KERNEL_FACTS_COMMAND
+
+    def test_it_writes_nothing(self):
+        for verb in ("modprobe", "insmod", "rmmod", "sudo", " > ", ">>"):
+            assert verb not in KERNEL_FACTS_COMMAND
+
+    @pytest.mark.skipif(os.uname().sysname != "Linux", reason="reads a Linux kernel")
+    def test_it_runs_and_parses_on_this_host(self):
+        out = subprocess.run(
+            ["sh", "-c", KERNEL_FACTS_COMMAND], capture_output=True, text=True, timeout=60
+        ).stdout
+
+        facts = parse_kernel_facts(out)
+
+        assert facts["release"] == os.uname().release
+        assert facts["loaded"] is None or all(isinstance(m, str) for m in facts["loaded"])
+
+
+class TestTheTemplate:
+    def test_a_driver_supplies_only_the_transport(self):
+        class Driver(KernelFactsMixin):
+            def _run_kernel_facts_command(self, command: str) -> str:
+                self.sent = command
+                return _wire(REPORT)
+
+        driver = Driver()
+        facts = driver.get_kernel_facts()
+
+        assert driver.sent == KERNEL_FACTS_COMMAND
+        assert facts["release"] == "6.1.0-25-amd64"
+
+    def test_not_every_os_driver_has_it(self):
+        """A Windows host is an OSDriver too, and has no Linux kernel to read:
+        ``hasattr`` has to stay a truthful answer, so the drivers that can mix
+        this in themselves."""
+        assert not issubclass(OSDriver, KernelFactsMixin)
+        assert not hasattr(OSDriver, "get_kernel_facts")

tests/test_lag_interfaces.py

@@ -0,0 +1,76 @@
+"""Tests for add_lag_interfaces — one logical row per trunk group."""
+
+from napalm_device_types import add_lag_interfaces
+
+
+def _port(is_up: bool = True, is_enabled: bool = True, speed: float = 1000.0, trunk_group: str = "") -> dict:
+    port = {
+        "is_up": is_up,
+        "is_enabled": is_enabled,
+        "description": "",
+        "last_flapped": -1.0,
+        "speed": speed,
+        "mtu": -1,
+        "mac_address": "",
+    }
+    if trunk_group:
+        port["trunk_group"] = trunk_group
+    return port
+
+
+def test_adds_one_row_per_trunk_group():
+    ifaces = {
+        "1": _port(),
+        "3": _port(trunk_group="Trk3"),
+        "4": _port(trunk_group="Trk3"),
+        "10": _port(trunk_group="Trk6"),
+        "7": _port(trunk_group="Trk6"),
+    }
+    result = add_lag_interfaces(ifaces)
+
+    assert result["Trk3"]["lag_members"] == ["3", "4"]
+    # Members in natural port order, not string order ("7" before "10").
+    assert result["Trk6"]["lag_members"] == ["7", "10"]
+    assert result["Trk6"]["description"] == "LAG (7, 10)"
+    assert "Trk1" not in result
+
+
+def test_state_is_derived_from_members():
+    ifaces = {
+        "3": _port(is_up=False, speed=1000.0, trunk_group="Trk3"),
+        "4": _port(is_up=True, speed=1000.0, trunk_group="Trk3"),
+        "6": _port(is_up=False, is_enabled=False, trunk_group="Trk6"),
+    }
+    result = add_lag_interfaces(ifaces)
+
+    assert result["Trk3"]["is_up"] is True
+    assert result["Trk3"]["is_enabled"] is True
+    assert result["Trk3"]["speed"] == 2000.0
+    assert result["Trk6"]["is_up"] is False
+    assert result["Trk6"]["is_enabled"] is False
+
+
+def test_lag_mode_only_when_known():
+    """The UI reads a missing mode as "static trunk"; guessing would mislabel LACP."""
+    ifaces = {"3": _port(trunk_group="Trk3"), "6": _port(trunk_group="Trk6")}
+    result = add_lag_interfaces(ifaces, lag_modes={"Trk3": "lacp"})
+
+    assert result["Trk3"]["lag_mode"] == "lacp"
+    assert "lag_mode" not in result["Trk6"]
+
+
+def test_keeps_a_lag_the_driver_already_reported():
+    ifaces = {
+        "3": _port(trunk_group="Trk3"),
+        "Trk3": {**_port(), "description": "uplink", "lag_members": ["3"]},
+    }
+    result = add_lag_interfaces(ifaces)
+
+    assert result["Trk3"]["description"] == "uplink"
+
+
+def test_does_not_modify_its_input():
+    ifaces = {"3": _port(trunk_group="Trk3")}
+    add_lag_interfaces(ifaces)
+
+    assert list(ifaces) == ["3"]

tests/test_port_forwards.py

@@ -0,0 +1,39 @@
+"""get_port_forwards: what the WAN side may reach inside, on any gateway.
+
+The reader used to be declared on ``ResidentialGatewayDriver`` only, as if a
+port forward were a home-router feature. A firewall forwards ports just the
+same -- OPNsense calls it destination NAT -- and the two consumers that ask
+(is this host reachable from the internet, which CVEs are exposed) need the
+answer from both. The declaration therefore lives where the two roles overlap,
+next to the NAT translations reader.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+from napalm_device_types import FirewallDriver, ResidentialGatewayDriver
+from napalm_device_types.nat_vpn import NatVpnMixin
+
+
+def test_a_firewall_and_a_gateway_share_the_declaration():
+    assert issubclass(FirewallDriver, NatVpnMixin)
+    assert issubclass(ResidentialGatewayDriver, NatVpnMixin)
+    assert "def get_port_forwards(self) -> List[PortForwardDict]" in inspect.getsource(NatVpnMixin)
+
+
+def test_it_is_declared_once():
+    assert "def get_port_forwards" not in inspect.getsource(ResidentialGatewayDriver)
+
+
+def test_absent_until_a_driver_implements_it():
+    assert not hasattr(FirewallDriver, "get_port_forwards")
+    assert not hasattr(ResidentialGatewayDriver, "get_port_forwards")
+
+
+def test_the_contract_says_what_counts():
+    """A redirect between two internal networks is destination NAT too, and
+    would make an internal host look reachable from the internet."""
+    source = inspect.getsource(NatVpnMixin)
+    assert "from the WAN" in source
+    assert "between internal networks" in source

tests/test_systemd_services.py

@@ -0,0 +1,412 @@
+"""systemd services: listing them in one round trip, and starting and stopping them.
+
+What systemd reports, and how a unit is started or stopped, is the same on
+every host that runs it -- so the command, its parse, the name check and the
+reading of the exit status live here once, and a driver only carries a command
+across (napalm-linux#7, napalm-proxmox#6).
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.systemd import (
+    ACTION_TIMEOUT,
+    SERVICE_ACTIONS,
+    SYSTEMD_SERVICES_COMMAND,
+    SystemdServicesMixin,
+    SystemdUnavailable,
+    parse_action_result,
+    parse_systemd_services,
+    service_action_command,
+    unit_name,
+)
+
+FILES = """\
+apparmor.service                             enabled         enabled
+ssh.service                                  enabled         enabled
+sshd.service                                 alias           -
+getty@.service                               enabled         enabled
+rsync.service                                disabled        enabled
+cups.service                                 indirect        enabled
+plymouth-quit.service                        static          -
+systemd-networkd-wait-online.service         enabled-runtime enabled
+nfs-server.service                           masked          enabled
+"""
+
+UNITS = """\
+MainPID=812
+Id=ssh.service
+Names=ssh.service sshd.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=enabled
+
+MainPID=0
+Id=apparmor.service
+Names=apparmor.service
+LoadState=loaded
+ActiveState=active
+SubState=exited
+UnitFileState=enabled
+
+MainPID=900
+Id=getty@tty1.service
+Names=getty@tty1.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=enabled
+
+MainPID=0
+Id=systemd-fsck@dev-disk-by\\x2dlabel-BOOT.service
+Names=systemd-fsck@dev-disk-by\\x2dlabel-BOOT.service
+LoadState=loaded
+ActiveState=inactive
+SubState=dead
+UnitFileState=static
+
+MainPID=0
+Id=display-manager.service
+Names=display-manager.service
+LoadState=not-found
+ActiveState=inactive
+SubState=dead
+UnitFileState=
+
+MainPID=0
+Id=nfs-server.service
+Names=nfs-server.service
+LoadState=masked
+ActiveState=inactive
+SubState=dead
+UnitFileState=masked
+
+MainPID=0
+Id=systemd-networkd-wait-online.service
+Names=systemd-networkd-wait-online.service
+LoadState=loaded
+ActiveState=active
+SubState=exited
+UnitFileState=enabled-runtime
+"""
+
+
+GENERATED_UNIT = """
+MainPID=0
+Id=rrdcached.service
+Names=rrdcached.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=generated
+"""
+
+
+def _wire(
+    files: str = FILES,
+    units: str = UNITS,
+    *,
+    generated: str = "",
+    noise: str = "",
+    end: bool = True,
+) -> str:
+    """The report as the command prints it, framed."""
+    tail = "SVC_END\n" if end else ""
+    return f"{noise}SVC_BEGIN\n[files]\n{files}[units]\n{units}[generated]\n{generated}{tail}"
+
+
+def _by_name(services):
+    return {s["name"]: s for s in services}
+
+
+class TestParseSystemdServices:
+    def test_a_loaded_unit_is_read_with_its_state(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["ssh"] == {"name": "ssh", "running": True, "enabled": True, "pid": 812}
+        assert services["apparmor"] == {
+            "name": "apparmor",
+            "running": False,
+            "enabled": True,
+            "pid": 0,
+        }
+
+    def test_enabled_means_enabled_now_not_merely_installed(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["systemd-networkd-wait-online"]["enabled"] is True
+        assert services[r"systemd-fsck@dev-disk-by\x2dlabel-BOOT"]["enabled"] is False
+        assert services["nfs-server"]["enabled"] is False
+
+    def test_a_generated_unit_takes_its_boot_state_from_is_enabled(self):
+        """A SysV script's unit is generated; only is-enabled knows its rc links."""
+        raw = _wire(units=UNITS + GENERATED_UNIT, generated="rrdcached.service enabled\n")
+
+        assert _by_name(parse_systemd_services(raw))["rrdcached"]["enabled"] is True
+
+    def test_a_generated_unit_is_not_enabled_unless_is_enabled_says_so(self):
+        raw = _wire(units=UNITS + GENERATED_UNIT, generated="rrdcached.service disabled\n")
+
+        assert _by_name(parse_systemd_services(raw))["rrdcached"]["enabled"] is False
+
+    def test_a_unit_that_is_not_there_is_left_out(self):
+        assert "display-manager" not in _by_name(parse_systemd_services(_wire()))
+
+    def test_an_installed_unit_that_is_not_loaded_is_listed(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["rsync"] == {"name": "rsync", "running": False, "enabled": False, "pid": 0}
+        assert services["cups"]["enabled"] is False
+
+    def test_templates_and_static_files_that_are_not_loaded_are_not(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert "getty@" not in services
+        assert "plymouth-quit" not in services
+        assert services["getty@tty1"]["running"] is True
+
+    def test_an_alias_never_appears_beside_its_unit(self):
+        assert "sshd" not in _by_name(parse_systemd_services(_wire()))
+
+    def test_an_alias_that_older_systemd_calls_enabled_does_not_either(self):
+        files = (
+            "\n".join(
+                "sshd.service enabled enabled" if line.startswith("sshd.service") else line
+                for line in FILES.splitlines()
+            )
+            + "\n"
+        )
+
+        assert "sshd" not in _by_name(parse_systemd_services(_wire(files=files)))
+
+    def test_an_escaped_name_survives(self):
+        assert r"systemd-fsck@dev-disk-by\x2dlabel-BOOT" in _by_name(
+            parse_systemd_services(_wire())
+        )
+
+    def test_blocks_run_together_are_still_told_apart(self):
+        """xargs may split the unit list across two systemctl runs."""
+        units = UNITS.replace(
+            "UnitFileState=enabled\n\nMainPID=0\nId=apparmor",
+            "UnitFileState=enabled\nMainPID=0\nId=apparmor",
+        )
+
+        services = _by_name(parse_systemd_services(_wire(units=units)))
+
+        assert services["ssh"]["pid"] == 812
+        assert services["apparmor"]["running"] is False
+
+    def test_the_list_is_sorted_by_name(self):
+        names = [s["name"] for s in parse_systemd_services(_wire())]
+
+        assert names == sorted(names)
+
+    def test_terminal_colours_in_the_report_are_dropped(self):
+        files = FILES.replace(
+            "rsync.service                                disabled",
+            "rsync.service \x1b[0;1;31mdisabled\x1b[0m",
+        )
+
+        assert "rsync" in _by_name(parse_systemd_services(_wire(files=files)))
+
+    def test_whatever_surrounds_the_frame_is_ignored(self):
+        noisy = _wire(noise="user@host:~$ systemctl ...\n") + "user@host:~$ "
+
+        assert "ssh" in _by_name(parse_systemd_services(noisy))
+
+    def test_a_cut_short_report_raises(self):
+        """A missing tail must not read as services that went away."""
+        with pytest.raises(ValueError):
+            parse_systemd_services(_wire(end=False))
+
+    def test_output_without_the_frame_raises(self):
+        with pytest.raises(ValueError):
+            parse_systemd_services("bash: systemctl: command not found\n")
+
+    def test_a_host_without_systemd_says_so(self):
+        raw = "SVC_BEGIN\n[no-systemd]\n[files]\n[units]\nSVC_END\n"
+
+        with pytest.raises(SystemdUnavailable):
+            parse_systemd_services(raw)
+
+    def test_no_systemd_is_a_not_implemented_error(self):
+        assert issubclass(SystemdUnavailable, NotImplementedError)
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        """An echoing transport must not show the end marker early."""
+        assert "SVC_END" not in SYSTEMD_SERVICES_COMMAND
+        assert "SVC_BEGIN" not in SYSTEMD_SERVICES_COMMAND
+
+    def test_it_changes_nothing(self):
+        for verb in ("start", "stop", "restart", "enable", "disable", "mask"):
+            assert f"systemctl {verb}" not in SYSTEMD_SERVICES_COMMAND
+
+    @pytest.mark.skipif(not os.path.isdir("/run/systemd/system"), reason="needs systemd")
+    def test_it_runs_and_parses_on_this_host(self):
+        out = subprocess.run(
+            ["sh", "-c", SYSTEMD_SERVICES_COMMAND], capture_output=True, text=True, timeout=60
+        ).stdout
+
+        services = _by_name(parse_systemd_services(out))
+
+        assert "systemd-journald" in services
+        assert services["systemd-journald"]["running"] is True
+
+
+class TestUnitName:
+    @pytest.mark.parametrize(
+        ("raw", "name"),
+        [
+            ("ssh", "ssh"),
+            ("ssh.service", "ssh"),
+            ("getty@tty1", "getty@tty1"),
+            ("wg-quick@wg0", "wg-quick@wg0"),
+            ("snapd.apparmor", "snapd.apparmor"),
+            ("systemd-backlight@backlight:acpi_video0", "systemd-backlight@backlight:acpi_video0"),
+            (r"systemd-fsck@dev-disk-by\x2dlabel-BOOT", r"systemd-fsck@dev-disk-by\x2dlabel-BOOT"),
+        ],
+    )
+    def test_a_unit_name_is_accepted(self, raw, name):
+        assert unit_name(raw) == name
+
+    @pytest.mark.parametrize(
+        "raw",
+        [
+            "",
+            "-x",
+            "foo@",
+            "foo@.service",
+            "a b",
+            "a;b",
+            "$(id)",
+            "a/b",
+            r"bad\x2",
+            "ssh\n",
+            "x" * 256,
+        ],
+    )
+    def test_anything_else_is_refused(self, raw):
+        with pytest.raises(ValueError):
+            unit_name(raw)
+
+
+class TestServiceActionCommand:
+    def test_the_command_is_bounded_and_never_asks(self):
+        cmd = service_action_command("getty@tty1", "restart")
+
+        assert cmd.startswith(f"timeout {ACTION_TIMEOUT} systemctl --no-ask-password restart -- ")
+        assert "getty@tty1.service" in cmd
+
+    def test_an_escaped_name_is_quoted_for_the_shell(self):
+        cmd = service_action_command(r"systemd-fsck@dev-disk-by\x2dlabel-BOOT", "stop")
+
+        assert r"'systemd-fsck@dev-disk-by\x2dlabel-BOOT.service'" in cmd
+
+    def test_its_exit_status_is_printed_after_it(self):
+        assert service_action_command("ssh", "start").endswith("; echo __SVC_RC=$?")
+
+    def test_the_actions(self):
+        assert SERVICE_ACTIONS == ("start", "stop", "restart", "enable", "disable")
+
+    def test_an_unknown_action_is_refused(self):
+        with pytest.raises(ValueError):
+            service_action_command("ssh", "mask")
+
+    def test_an_invalid_name_is_refused(self):
+        with pytest.raises(ValueError):
+            service_action_command("ssh; reboot", "stop")
+
+
+class TestParseActionResult:
+    def test_exit_status_zero_is_success(self):
+        assert parse_action_result("__SVC_RC=0\n") == {"success": True, "output": ""}
+
+    def test_what_systemctl_printed_comes_back_without_the_marker(self):
+        raw = (
+            "Created symlink /etc/systemd/system/multi-user.target.wants/cron.service.\n__SVC_RC=0"
+        )
+
+        result = parse_action_result(raw)
+
+        assert result["success"] is True
+        assert result["output"].startswith("Created symlink")
+        assert "__SVC_RC" not in result["output"]
+
+    def test_terminal_colours_are_dropped(self):
+        """systemctl colours its errors when a transport gives it a terminal."""
+        raw = (
+            "\x1b[0;1;31mFailed to restart x.service: Unit x.service not found.\x1b[0m\n"
+            "__SVC_RC=5\n"
+        )
+
+        assert parse_action_result(raw)["output"] == (
+            "Failed to restart x.service: Unit x.service not found."
+        )
+
+    def test_a_failure_keeps_its_message(self):
+        raw = "Failed to start foo.service: Unit foo.service not found.\n__SVC_RC=5\n"
+
+        assert parse_action_result(raw) == {
+            "success": False,
+            "output": "Failed to start foo.service: Unit foo.service not found.",
+        }
+
+    def test_a_job_still_running_at_the_timeout_is_not_called_done(self):
+        result = parse_action_result("__SVC_RC=124\n")
+
+        assert result["success"] is False
+        assert str(ACTION_TIMEOUT) in result["output"]
+
+    def test_no_exit_status_is_no_success(self):
+        assert parse_action_result("Connection reset\n")["success"] is False
+
+    def test_the_echoed_command_is_not_taken_for_the_status(self):
+        raw = "timeout 45 systemctl restart -- cron.service 2>&1; echo __SVC_RC=$?\n__SVC_RC=1\n"
+
+        assert parse_action_result(raw)["success"] is False
+
+
+class _Driver(SystemdServicesMixin):
+    def __init__(self, reply: str) -> None:
+        self.reply = reply
+        self.calls: list = []
+
+    def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
+        self.calls.append((command, privileged, timeout))
+        return self.reply
+
+
+class TestSystemdServicesMixin:
+    def test_listing_runs_the_command_unprivileged(self):
+        driver = _Driver(_wire())
+
+        assert "ssh" in _by_name(driver.get_services())
+        assert driver.calls == [(SYSTEMD_SERVICES_COMMAND, False, ACTION_TIMEOUT + 15)]
+
+    def test_an_action_runs_privileged_and_reports_its_outcome(self):
+        driver = _Driver("__SVC_RC=0\n")
+
+        assert driver.manage_service("cron", "restart") == {"success": True, "output": ""}
+        command, privileged, timeout = driver.calls[0]
+        assert command == service_action_command("cron", "restart")
+        assert privileged is True
+        assert timeout > ACTION_TIMEOUT
+
+    def test_an_invalid_request_is_refused_before_anything_is_sent(self):
+        driver = _Driver("__SVC_RC=0\n")
+
+        with pytest.raises(ValueError):
+            driver.manage_service("cron;reboot", "stop")
+        assert driver.calls == []
+
+    def test_not_every_os_driver_has_it(self):
+        assert not hasattr(OSDriver, "manage_service")
+        assert callable(getattr(SystemdServicesMixin, "manage_service"))

tests/test_vm_cpu_types.py

@@ -0,0 +1,58 @@
+"""A new VM's virtual CPU model can be chosen, from a list the hypervisor offers.
+
+Proxmox gives a VM created without a ``cpu`` argument the ``kvm64`` model,
+which has no AVX -- and MongoDB 5.0 and later will not start without it. Which
+model is right depends on the cluster (``host`` cannot live-migrate between
+different CPUs, ``x86-64-v3`` does not start on a CPU older than Haswell), so
+the caller chooses, from entries that say what each model provides and whether
+the node at hand can run it.
+
+The declarations sit under ``TYPE_CHECKING`` (see test_role_contracts), so the
+signature is read from the source rather than from the class.
+"""
+
+from __future__ import annotations
+
+import ast
+import inspect
+from typing import List, get_type_hints
+
+import napalm_device_types.hypervisor as hypervisor_module
+from napalm_device_types import HypervisorDriver
+from napalm_device_types.models import VMCpuTypeDict
+
+
+def _declared(name: str) -> ast.FunctionDef:
+    tree = ast.parse(inspect.getsource(hypervisor_module))
+    for node in ast.walk(tree):
+        if isinstance(node, ast.FunctionDef) and node.name == name:
+            return node
+    raise AssertionError(f"HypervisorDriver does not declare {name}()")
+
+
+class TestCreateVmTakesACpuType:
+    def test_cpu_type_is_an_optional_keyword(self):
+        fn = _declared("create_vm_from_cloud_init")
+        kwonly = {arg.arg: default for arg, default in zip(fn.args.kwonlyargs, fn.args.kw_defaults)}
+        assert "cpu_type" in kwonly
+        default = kwonly["cpu_type"]
+        assert isinstance(default, ast.Constant) and default.value is None
+
+
+class TestCpuTypeListing:
+    def test_is_declared(self):
+        assert _declared("get_vm_cpu_types").returns is not None
+
+    def test_absent_until_a_driver_implements_it(self):
+        """netOrk probes capabilities with hasattr; a hypervisor without a
+        choice of CPU model must not seem to offer one."""
+        assert not hasattr(HypervisorDriver, "get_vm_cpu_types")
+
+    def test_entry_shape(self):
+        assert get_type_hints(VMCpuTypeDict) == {
+            "name": str,
+            "description": str,
+            "features": List[str],
+            "available": bool,
+            "default": bool,
+        }