feat: a new VM's CPU model can be chosen, from a list the hypervisor offers

create_vm_from_cloud_init takes cpu_type. Proxmox gives a VM created without
one the kvm64 model, which has no AVX, so MongoDB 5.0 and later do not start
there, and netOrk's graylog role failed on every VM it provisioned
(netork#494). 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.

get_vm_cpu_types() is the new, optional listing behind that choice. Each
VMCpuTypeDict names the model, says what it is for, lists the /proc/cpuinfo
flags the guest gets (so a caller can ask "does this give AVX?" without
knowing model names), whether the node at hand can run it, and which one is
the default. A hypervisor whose VMs have no per-VM CPU model (VMware sets CPU
compatibility per cluster) does not implement it and must reject any
cpu_type other than None.

Both declarations sit under TYPE_CHECKING like the rest of the contract, so
hasattr stays a truthful capability probe; the tests read the signature from
the source.

README.md

@@ -227,6 +227,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 +247,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

@@ -63,6 +63,7 @@ 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.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
@@ -101,6 +102,7 @@ __all__ = [
     "StorageDriver",
     "SwitchDriver",
     "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/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,21 @@ class NATTranslationDict(TypedDict):
     age: float
 
 
+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 +485,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 +912,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).

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_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,
+        }