Merge pull request 'feat: port forwards are a firewall reader too, and only the WAN's' (#2) from feature/port-forwards-shared into main

get_port_forwards was declared on ResidentialGatewayDriver alone, as if a
port forward were a home-router feature. A firewall forwards ports just the
same (OPNsense calls it destination NAT), and netOrk asks both: is this host
reachable from the internet, which CVEs are exposed. The declaration moves to
NatVpnMixin, where the two roles already overlap, and PortForwardDict next to
NATTranslationDict.

The contract now says what counts. Destination NAT between internal networks
and rules that only exempt traffic are not port forwards: callers read every
entry as "reachable from outside". "ANY" forwards every protocol and an
external port of 0 every port -- a whole host forwarded is the most exposed
case and must not fall out for lack of a port number.

Declaration only, under TYPE_CHECKING: nothing changes at runtime.

README.md

@@ -22,6 +22,137 @@ NAPALM's `NetworkDriver` defines a common interface for all network devices. In
 
 `napalm-device-types` sits in between: it adds one well-typed layer of abstract methods per device category, so every driver for the same category exposes the same interface.
 
+## Roles: what a device *is*
+
+A device is often several things at once. A QNAP NAS runs VMs on a Linux userland; an
+OpenMediaVault box is a NAS built on Debian. So a driver inherits **one role base per
+role its device fills**, and **the order it lists them in is the ranking**:
+
+```python
+class QnapQtsDriver(StorageDriver, HypervisorDriver, LinuxDriver):
+    ...   # primary_role_of(...) == "storage"
+```
+
+`roles_of(cls)`, `role_keys_of(cls)` and `primary_role_of(cls)` read that back. Nothing
+restates the ranking: there is no precedence table and no attribute to override.
+
+### Role bases declare; they never implement
+
+**A role base must not contain a single runtime method — not even a
+`NotImplementedError` placeholder.** Its methods are declared under `if TYPE_CHECKING`:
+
+```python
+class StorageDriver(DeviceTypeDriver):
+    """Contract, not code."""
+    ROLE: str = "storage"
+    TYPE_LABEL: str = "Storage"
+
+    if TYPE_CHECKING:                      # nothing exists at runtime
+        def get_disks(self) -> List[PhysicalDiskDict]: ...
+```
+
+This is not a style preference. A placeholder on a base class is not neutral under
+multiple inheritance: it wins the MRO against a sibling base's *working* implementation
+and silently replaces it. Adding one stub to a base is therefore a breaking change for
+every driver that mixes that base with another. It happened three times here before the
+rule existed, and each time the fix was hand-written forwarding methods in the driver.
+
+Two things follow, and both are improvements:
+
+- **`hasattr` is truthful again.** A method exists on a driver class exactly when that
+  driver implemented it, which is how netOrk asks "can this driver list disks".
+- **A method that was never implemented raises `AttributeError`, not
+  `NotImplementedError`.** Ask before calling.
+
+## Function classes: what a device *can do*
+
+Behaviour shared across roles lives in a function class, exactly once, and a role base
+is a thin bundle over them — `PackageManagementMixin`, `HealthMetricsMixin`,
+`ServiceControlMixin`, `UpdateMixin`, `NatVpnMixin`, `MacAclMixin`, `FirewallRuleMixin`,
+`DhcpServerMixin`, `PingSweepMixin`, `ConfigLifecycleMixin`, `InterfaceFilterMixin`.
+`HostRebootMixin` (`reboot_host`) is mixed into `DeviceTypeDriver` itself, since any
+device may be restartable; like the others it only declares.
+
+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
+several hooks. `ConfigLifecycleMixin.compare_config` over `_get_running_config` is the
+model. Where the base would only pass the call through, declare the method directly;
+two names for one pass-through is ceremony, not design.
+
+NAPALM's own getters (`get_facts`, `get_interfaces`, `ping`, `get_config`) are never
+wrapped in a template — they belong to NAPALM, and code outside this repo relies on
+their contract.
+
+## Design principle: generic vs. device-specific logic
+
+When adding behavior to a device-type base class, split it along one line: **would
+this exact logic work unchanged for a different vendor's driver of the same
+device-type, if that driver only implemented the same abstract methods?**
+
+- If yes, it's generic — implement it once as a **concrete** method on the
+  device-type base class (here, in this repo).
+- If no — it talks to the device itself (a specific REST endpoint, a CLI command,
+  a vendor-specific payload format) — it belongs in the concrete driver as the
+  implementation of an **abstract** method the base class declares.
+
+Concretely: matching/comparison/reconciliation algorithms, orchestration flows, and
+generic data shapes belong here. Only the actual device communication belongs in
+`vendor/napalm-<name>`.
+
+**Worked example — firewall rule diff/apply** (`FirewallDriver`):
+
+```python
+class FirewallDriver(DeviceTypeDriver):
+    # Abstract — every driver implements its own device communication.
+    def get_firewall_rules(self) -> List[FirewallRuleDict]: raise NotImplementedError
+    def apply_firewall_rule(self, rule: FirewallRuleDict, *, uuid: Optional[str] = None) -> Dict[str, Any]: raise NotImplementedError
+    def commit_firewall_rules(self) -> Dict[str, Any]: raise NotImplementedError
+
+    # Concrete — the matching/comparison/orchestration algorithm is identical
+    # for every firewall vendor, so it lives here once.
+    def diff_firewall_rules(self, desired: List[FirewallRuleDict]) -> FirewallRuleDiffDict:
+        ...  # matches self.get_firewall_rules() against `desired` by description
+
+    def apply_firewall_ruleset(self, desired: List[FirewallRuleDict]):
+        ...  # computes the diff, calls apply_firewall_rule() per change, commits
+```
+
+The same split applies to `DhcpServerMixin`: `get_dhcp_reservations`/
+`apply_dhcp_reservation`/`commit_dhcp_reservations` are abstract (Kea REST on
+OPNsense, dnsmasq/odhcpd UCI on OpenWrt), while `diff_dhcp_reservations` and
+`apply_dhcp_reservationset` are concrete — matching by normalised MAC and the
+apply-then-commit orchestration are identical for every DHCP server.
+
+A new driver (FortiGate, pfSense, …) gets `diff_firewall_rules`/
+`apply_firewall_ruleset` for free the moment it implements the three abstract
+methods — it never needs to reimplement the reconciliation logic itself.
+
+**Second worked example — ping sweeps** (`PingSweepMixin`, mixed into
+`DeviceTypeDriver`, so *every* device-type driver has it):
+
+```python
+class PingSweepMixin:
+    # Concrete — the loop, the reply parsing, the target cap and the progress
+    # reporting are the same for every device that can ping at all.
+    def ping_sweep(self, destinations, *, count=1, timeout=1, …) -> PingSweepResultDict:
+        ...  # calls NAPALM's standard ping() once per destination
+```
+
+A driver becomes a usable sweep source the moment it implements NAPALM's
+`ping()` — nothing else is required, and `driver_supports_ping(cls)` reports
+whether it did (introspection, not a hand-maintained list). A driver whose
+device offers something genuinely faster overrides `ping_sweep` and keeps the
+return shape: `napalm-opnsense` starts a batch of ping jobs over the
+diagnostics API, waits once for all of them, and reads every result with a
+single request — a per-host loop would be unusable there.
+
+This mirrors a similar split already documented on the consumer side, in NetOrk's
+`docs/ARCHITECTURE.md` ("Device Warnings — Trennung von Erkennung und
+Präsentation"): drivers return raw signals, the higher layer gives them meaning.
+Same shape of separation, different axis — device-specific vs. generic here,
+detection vs. presentation there.
+
 ## Installation
 
 ```bash
@@ -40,6 +171,13 @@ Requires Python ≥ 3.9 and NAPALM ≥ 4.0.
 | `HypervisorDriver` | Hypervisors & virtualisation platforms | Proxmox VE, VMware ESXi, KVM/libvirt |
 | `OSDriver` | General-purpose operating systems | Linux, BSD, macOS |
 | `StorageDriver` | Storage appliances & NAS/SAN | TrueNAS, Synology DSM, QNAP QTS |
+| `ResidentialGatewayDriver` | Router + firewall + AP in one box | OpenWrt, FritzBox |
+
+Mixins mixed into the classes above rather than used on their own:
+`ConfigLifecycleMixin` (config load/compare/commit/rollback), `PingSweepMixin`
+(subnet sweeps), and `DhcpServerMixin` (static DHCP reservations — mixed into
+`FirewallDriver` and `ResidentialGatewayDriver`, since both commonly run the
+DHCP server for their networks).
 
 ## Usage
 
@@ -89,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
@@ -102,7 +245,7 @@ class ProxmoxDriver(HypervisorDriver):
         # return List[VMDict]
         ...
 
-    def snapshot_create(self, name, snapshot, description="", include_memory=False):
+    def create_vm_snapshot(self, name, snapshot, description="", include_memory=False):
         ...
 ```
 

napalm_device_types/__init__.py

@@ -4,16 +4,22 @@ napalm-device-types
 
 Abstract device-type base classes for NAPALM drivers.
 
-Instead of inheriting directly from ``napalm.base.NetworkDriver``, a driver
-can inherit from one of the device-type classes defined here to gain
-type-specific abstract methods and a clearer contract::
+A driver inherits one base per role its device fills, and **the order of those
+bases is the ranking** -- the first one is what netOrk shows as the device's
+class::
 
-    from napalm_device_types import AccessPointDriver
+    from napalm_device_types import StorageDriver, HypervisorDriver
 
-    class OpenWrtDriver(AccessPointDriver):
-        ...
+    class QnapQtsDriver(StorageDriver, HypervisorDriver, LinuxDriver):
+        ...            # a NAS that also runs VMs on a Linux userland
 
-Available base classes:
+That works because a role base *declares* its methods (under ``if
+TYPE_CHECKING``) and implements none of them. Nothing exists at runtime until a
+concrete driver provides it, so no base can shadow a working implementation
+inherited from a sibling, and ``hasattr`` is a truthful answer to "can this
+driver do X".
+
+Role bases -- what a device *is*:
 
 * :class:`~napalm_device_types.access_point.AccessPointDriver`
 * :class:`~napalm_device_types.switch.SwitchDriver`
@@ -21,21 +27,52 @@ Available base classes:
 * :class:`~napalm_device_types.hypervisor.HypervisorDriver`
 * :class:`~napalm_device_types.os.OSDriver`
 * :class:`~napalm_device_types.storage.StorageDriver`
+* :class:`~napalm_device_types.phone.PhoneDriver`
+* :class:`~napalm_device_types.media.MediaDriver`
 * :class:`~napalm_device_types.residential_gateway.ResidentialGatewayDriver`
 
-Also provided:
+Function classes -- what a device *can do*. Shared behaviour lives here once
+instead of being restated on every role that happens to need it:
 
-* :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin` --
-  stand-alone mixin to reduce duplication of config lifecycle methods across
-  drivers.
+* :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin`
+* :class:`~napalm_device_types.dhcp.DhcpServerMixin`
+* :class:`~napalm_device_types.firewall_rules.FirewallRuleMixin`
+* :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.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.updates.UpdateMixin`
+
+Introspection -- :func:`~napalm_device_types.roles.roles_of`,
+:func:`~napalm_device_types.roles.role_keys_of` and
+:func:`~napalm_device_types.roles.primary_role_of`.
 """
 
 from napalm_device_types.base import DeviceTypeDriver, FingerprintRule, PortSpec
 from napalm_device_types.access_point import AccessPointDriver
 from napalm_device_types.config_lifecycle import ConfigLifecycleMixin
+from napalm_device_types.dhcp import DhcpServerMixin, normalize_cidr, normalize_mac
 from napalm_device_types.firewall import FirewallDriver
 from napalm_device_types.hypervisor import HypervisorDriver
 from napalm_device_types.os import OSDriver
+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
+from napalm_device_types.packages import PackageManagementMixin
+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.updates import UpdateMixin
 from napalm_device_types.residential_gateway import ResidentialGatewayDriver
 from napalm_device_types.storage import StorageDriver
 from napalm_device_types.switch import SwitchDriver
@@ -44,12 +81,32 @@ __all__ = [
     "AccessPointDriver",
     "ConfigLifecycleMixin",
     "DeviceTypeDriver",
+    "DhcpServerMixin",
     "FingerprintRule",
     "FirewallDriver",
+    "FirewallRuleMixin",
+    "HealthMetricsMixin",
+    "HostRebootMixin",
     "HypervisorDriver",
+    "InterfaceFilterMixin",
+    "MacAclMixin",
+    "MediaDriver",
+    "NatVpnMixin",
     "OSDriver",
+    "PackageManagementMixin",
+    "PhoneDriver",
+    "PingSweepMixin",
     "PortSpec",
     "ResidentialGatewayDriver",
+    "ServiceControlMixin",
     "StorageDriver",
     "SwitchDriver",
+    "UpdateMixin",
+    "add_lag_interfaces",
+    "driver_supports_ping",
+    "normalize_cidr",
+    "normalize_mac",
+    "primary_role_of",
+    "role_keys_of",
+    "roles_of",
 ]

napalm_device_types/access_point.py

@@ -10,30 +10,25 @@ Usage::
             ...
 """
 
-from typing import Any, Dict, List
+from typing import Any, Dict, List, TYPE_CHECKING
 from napalm_device_types.base import DeviceTypeDriver
-from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.mac_acl import MacAclMixin
+from napalm_device_types.updates import UpdateMixin
+from napalm_device_types.services import ServiceControlMixin
+from napalm_device_types.packages import PackageManagementMixin
+from napalm_device_types.health_metrics import HealthMetricsMixin
+from napalm_device_types.interface_filter import InterfaceFilterMixin
 from napalm_device_types.models import (
-    ChannelScanEntryDict,
     Dot1XConfigDict,
     FastTransitionConfigDict,
-    HealthMetricsDict,
-    MACACLDict,
     MeshConfigDict,
     MeshPeerDict,
-    PackageDict,
-    RadioStatusDict,
-    ServiceDict,
     SSIDBridgeDict,
-    SSIDDict,
-    UpdateDict,
-    WirelessClientDict,
     WirelessConfigDict,
 )
 
 
-class AccessPointDriver(DeviceTypeDriver):
-    TYPE_LABEL: str = "Access Point"
+class AccessPointDriver(MacAclMixin, UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthMetricsMixin, InterfaceFilterMixin, DeviceTypeDriver):
     """
     Abstract intermediate driver for wireless access points.
 
@@ -41,33 +36,16 @@ class AccessPointDriver(DeviceTypeDriver):
     access-point-specific operations that concrete drivers must implement.
     """
 
-    # Interfaces that carry no operational meaning on an access point and
-    # should be excluded from get_interfaces() / get_facts() interface_list.
-    _EXCLUDED_INTERFACES: frozenset = frozenset({"lo"})
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "access_point"
+    TYPE_LABEL: str = "Access Point"
+
 
-    # Interface name *prefixes* to exclude (e.g. Linux phy* are raw radio
-    # devices and have no IP/Ethernet significance at the AP level).
-    _EXCLUDED_INTERFACE_PREFIXES: tuple = ("phy",)
 
-    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
-    _SNMP_TX_ERR_IS_DROP: bool = False
 
-    @classmethod
-    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
-        return await collect_ucd_metrics(
-            snmp_get, snmp_walk,
-            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
-            if_skip=cls._SNMP_SKIP_IF,
-        )
 
-    def _filter_interfaces(self, interfaces: Dict[str, Any]) -> Dict[str, Any]:
-        """Remove loopback and radio-device (phy*) interfaces from an interface dict."""
-        return {
-            name: data
-            for name, data in interfaces.items()
-            if name not in self._EXCLUDED_INTERFACES
-            and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES)
-        }
 
     # AP-specific methods (get_wireless_clients, get_ssids, get_radio_status,
     # get_interfaces, get_vlans, …) are intentionally NOT defined here.
@@ -77,451 +55,225 @@ class AccessPointDriver(DeviceTypeDriver):
     # all standard NAPALM methods, so AccessPointDriver must come LAST to avoid
     # shadowing the mixin implementations.
 
-    def get_wireless_config(self) -> WirelessConfigDict:
-        """
-        Returns global wireless configuration parameters that apply across
-        all radios and SSIDs.
+    if TYPE_CHECKING:
 
-        * country_code (string) - ISO 3166-1 alpha-2 country code (e.g. ``"DE"``)
-        * regulatory_domain (string) - regulatory domain string (e.g. ``"ETSI"``)
-        * beacon_interval (int) - beacon interval in TUs (default 100)
-        * dtim_period (int) - DTIM period (default 2)
-        * rts_threshold (int) - RTS/CTS threshold in bytes (2347 = disabled)
-        * fragmentation_threshold (int) - fragmentation threshold in bytes
-        * short_preamble (bool) - whether short preamble is enabled
-        * wmm_enabled (bool) - whether WMM/QoS is enabled
+        def get_wireless_config(self) -> WirelessConfigDict:
+            """
+            Returns global wireless configuration parameters that apply across
+            all radios and SSIDs.
 
-        Example::
+            * country_code (string) - ISO 3166-1 alpha-2 country code (e.g. ``"DE"``)
+            * regulatory_domain (string) - regulatory domain string (e.g. ``"ETSI"``)
+            * beacon_interval (int) - beacon interval in TUs (default 100)
+            * dtim_period (int) - DTIM period (default 2)
+            * rts_threshold (int) - RTS/CTS threshold in bytes (2347 = disabled)
+            * fragmentation_threshold (int) - fragmentation threshold in bytes
+            * short_preamble (bool) - whether short preamble is enabled
+            * wmm_enabled (bool) - whether WMM/QoS is enabled
 
-            {
-                "country_code": "DE",
-                "regulatory_domain": "ETSI",
-                "beacon_interval": 100,
-                "dtim_period": 2,
-                "rts_threshold": 2347,
-                "fragmentation_threshold": 2346,
-                "short_preamble": True,
-                "wmm_enabled": True,
-            }
-        """
-        raise NotImplementedError
+            Example::
 
-    def get_fast_transition_config(self) -> Dict[str, FastTransitionConfigDict]:
-        """
-        Returns the 802.11r Fast BSS Transition (FT) configuration per SSID.
-
-        Keys are SSID names.  Each value contains:
-
-        * enabled (bool) - whether FT is active on this SSID
-        * ssid (string) - SSID name (repeated for convenience)
-        * mobility_domain (string) - 4-hex-digit Mobility Domain ID (MDID)
-        * reassociation_deadline (int) - FT reassociation deadline in TUs
-        * r0_key_lifetime (int) - PMK-R0 key lifetime in minutes
-        * r1_key_holder (string) - R1 Key Holder identifier (MAC-like string)
-        * pmk_r1_push (bool) - whether PMK-R1 is proactively pushed to neighbours
-        * over_ds (bool) - whether FT over DS (instead of FT over air) is used
-
-        Example::
-
-            {
-                "CorpWiFi": {
-                    "enabled": True,
-                    "ssid": "CorpWiFi",
-                    "mobility_domain": "a1b2",
-                    "reassociation_deadline": 1000,
-                    "r0_key_lifetime": 10000,
-                    "r1_key_holder": "00:11:22:33:44:55",
-                    "pmk_r1_push": True,
-                    "over_ds": False,
-                }
-            }
-        """
-        raise NotImplementedError
-
-    def get_mesh_config(self) -> Dict[str, MeshConfigDict]:
-        """
-        Returns the 802.11s mesh configuration per mesh interface.
-
-        Keys are mesh interface names (e.g. ``"mesh0"``).  Each value contains:
-
-        * enabled (bool) - whether the mesh interface is active
-        * radio (string) - underlying radio (e.g. ``"radio0"``)
-        * mesh_id (string) - 802.11s Mesh ID (analogous to SSID)
-        * path_metric (string) - path selection metric, e.g. ``"airtime"`` or ``"hopcount"``
-        * gate_announcements (bool) - whether gate announcements (GANN) are sent
-        * is_gate (bool) - whether this node acts as a mesh gate to the DS
-        * encryption (string) - e.g. ``"SAE"``, ``"open"``
-
-        Example::
-
-            {
-                "mesh0": {
-                    "enabled": True,
-                    "radio": "radio1",
-                    "mesh_id": "office-mesh",
-                    "path_metric": "airtime",
-                    "gate_announcements": True,
-                    "is_gate": True,
-                    "encryption": "SAE",
-                }
-            }
-        """
-        raise NotImplementedError
-
-    def get_mesh_peers(self) -> List[MeshPeerDict]:
-        """
-        Returns a list of currently active 802.11s mesh peers.
-
-        Each entry contains:
-
-        * mac (string) - peer MAC address
-        * radio (string) - radio on which the peering was established
-        * signal (int) - received signal strength in dBm
-        * tx_rate (float) - TX bitrate to peer in Mbit/s
-        * rx_rate (float) - RX bitrate from peer in Mbit/s
-        * uptime (int) - peering duration in seconds
-        * hop_count (int) - number of hops to the mesh gate (0 = this node is the gate)
-
-        Example::
-
-            [
                 {
-                    "mac": "AA:BB:CC:DD:EE:01",
-                    "radio": "radio1",
-                    "signal": -58,
-                    "tx_rate": 300.0,
-                    "rx_rate": 270.0,
-                    "uptime": 7200,
-                    "hop_count": 1,
+                    "country_code": "DE",
+                    "regulatory_domain": "ETSI",
+                    "beacon_interval": 100,
+                    "dtim_period": 2,
+                    "rts_threshold": 2347,
+                    "fragmentation_threshold": 2346,
+                    "short_preamble": True,
+                    "wmm_enabled": True,
                 }
-            ]
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_ssid_bridge_config(self) -> Dict[str, SSIDBridgeDict]:
-        """
-        Returns the Layer-2 bridging configuration for each SSID, i.e. which
-        bridge interface and VLAN each SSID is mapped to.
+        def get_fast_transition_config(self) -> Dict[str, FastTransitionConfigDict]:
+            """
+            Returns the 802.11r Fast BSS Transition (FT) configuration per SSID.
 
-        Keys are SSID names.  Each value contains:
+            Keys are SSID names.  Each value contains:
 
-        * ssid (string) - SSID name (repeated for convenience)
-        * bridge (string) - bridge interface the VAP is attached to (e.g. ``"br-lan"``, ``"br-guest"``)
-        * vlan_id (int) - 802.1Q VLAN ID (0 = untagged / no VLAN separation)
-        * tagged (bool) - whether traffic is 802.1Q-tagged on the uplink port
-        * client_isolation (bool) - whether clients on this SSID are isolated from each other
+            * enabled (bool) - whether FT is active on this SSID
+            * ssid (string) - SSID name (repeated for convenience)
+            * mobility_domain (string) - 4-hex-digit Mobility Domain ID (MDID)
+            * reassociation_deadline (int) - FT reassociation deadline in TUs
+            * r0_key_lifetime (int) - PMK-R0 key lifetime in minutes
+            * r1_key_holder (string) - R1 Key Holder identifier (MAC-like string)
+            * pmk_r1_push (bool) - whether PMK-R1 is proactively pushed to neighbours
+            * over_ds (bool) - whether FT over DS (instead of FT over air) is used
 
-        Example::
+            Example::
 
-            {
-                "CorpWiFi": {
-                    "ssid": "CorpWiFi",
-                    "bridge": "br-corp",
-                    "vlan_id": 10,
-                    "tagged": True,
-                    "client_isolation": False,
-                },
-                "GuestNet": {
-                    "ssid": "GuestNet",
-                    "bridge": "br-guest",
-                    "vlan_id": 20,
-                    "tagged": True,
-                    "client_isolation": True,
-                },
-                "IoT": {
-                    "ssid": "IoT",
-                    "bridge": "br-iot",
-                    "vlan_id": 30,
-                    "tagged": True,
-                    "client_isolation": True,
-                },
-            }
-        """
-        raise NotImplementedError
+                {
+                    "CorpWiFi": {
+                        "enabled": True,
+                        "ssid": "CorpWiFi",
+                        "mobility_domain": "a1b2",
+                        "reassociation_deadline": 1000,
+                        "r0_key_lifetime": 10000,
+                        "r1_key_holder": "00:11:22:33:44:55",
+                        "pmk_r1_push": True,
+                        "over_ds": False,
+                    }
+                }
+            """
+            ...
 
-    def get_mac_acl(self) -> Dict[str, MACACLDict]:
-        """
-        Returns the MAC-address-based access control lists configured per SSID.
+        def get_mesh_config(self) -> Dict[str, MeshConfigDict]:
+            """
+            Returns the 802.11s mesh configuration per mesh interface.
 
-        Keys are SSID names.  Each value contains:
+            Keys are mesh interface names (e.g. ``"mesh0"``).  Each value contains:
 
-        * ssid (string) - SSID name (repeated for convenience)
-        * policy (string) - ACL mode:
+            * enabled (bool) - whether the mesh interface is active
+            * radio (string) - underlying radio (e.g. ``"radio0"``)
+            * mesh_id (string) - 802.11s Mesh ID (analogous to SSID)
+            * path_metric (string) - path selection metric, e.g. ``"airtime"`` or ``"hopcount"``
+            * gate_announcements (bool) - whether gate announcements (GANN) are sent
+            * is_gate (bool) - whether this node acts as a mesh gate to the DS
+            * encryption (string) - e.g. ``"SAE"``, ``"open"``
 
-          * ``"allow"`` – whitelist: only listed MACs may associate
-          * ``"deny"``  – blacklist: listed MACs are blocked
-          * ``"disabled"`` – no MAC filtering active
+            Example::
 
-        * entries (list) - ACL entries, each with:
+                {
+                    "mesh0": {
+                        "enabled": True,
+                        "radio": "radio1",
+                        "mesh_id": "office-mesh",
+                        "path_metric": "airtime",
+                        "gate_announcements": True,
+                        "is_gate": True,
+                        "encryption": "SAE",
+                    }
+                }
+            """
+            ...
 
-          * mac (string) - MAC address (normalised, colon-separated)
-          * action (string) - ``"allow"`` or ``"deny"``
-          * description (string) - optional human-readable label
+        def get_mesh_peers(self) -> List[MeshPeerDict]:
+            """
+            Returns a list of currently active 802.11s mesh peers.
 
-        Example::
+            Each entry contains:
 
-            {
-                "CorpWiFi": {
-                    "name": "CorpWiFi",
-                    "policy": "allow",
-                    "entries": [
-                        {"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "CEO-Laptop"},
-                        {"mac": "AA:BB:CC:DD:EE:02", "action": "allow", "description": "CFO-Laptop"},
-                    ],
-                },
-                "GuestNet": {
-                    "name": "GuestNet",
-                    "policy": "deny",
-                    "entries": [
-                        {"mac": "DE:AD:BE:EF:00:01", "action": "deny", "description": "blocked device"},
-                    ],
-                },
-            }
-        """
-        raise NotImplementedError
+            * mac (string) - peer MAC address
+            * radio (string) - radio on which the peering was established
+            * signal (int) - received signal strength in dBm
+            * tx_rate (float) - TX bitrate to peer in Mbit/s
+            * rx_rate (float) - RX bitrate from peer in Mbit/s
+            * uptime (int) - peering duration in seconds
+            * hop_count (int) - number of hops to the mesh gate (0 = this node is the gate)
 
-    def get_dot1x_config(self) -> Dict[str, Dot1XConfigDict]:
-        """
-        Returns the 802.1X / WPA-Enterprise (RADIUS) configuration per SSID.
+            Example::
 
-        Keys are SSID names.  Each value contains:
+                [
+                    {
+                        "mac": "AA:BB:CC:DD:EE:01",
+                        "radio": "radio1",
+                        "signal": -58,
+                        "tx_rate": 300.0,
+                        "rx_rate": 270.0,
+                        "uptime": 7200,
+                        "hop_count": 1,
+                    }
+                ]
+            """
+            ...
 
-        * enabled (bool) - whether 802.1X authentication is active on this SSID
-        * ssid (string) - SSID name (repeated for convenience)
-        * auth_server (dict) - RADIUS authentication server:
+        def get_ssid_bridge_config(self) -> Dict[str, SSIDBridgeDict]:
+            """
+            Returns the Layer-2 bridging configuration for each SSID, i.e. which
+            bridge interface and VLAN each SSID is mapped to.
 
-          * host (string) - IP or FQDN of the RADIUS server
-          * port (int) - UDP port (default 1812)
-          * timeout (int) - request timeout in seconds
-          * retries (int) - number of retransmissions
+            Keys are SSID names.  Each value contains:
 
-        * acct_server (dict or None) - RADIUS accounting server (same keys as auth_server,
-          ``None`` if accounting is not configured)
-        * reauth_interval (int) - re-authentication interval in seconds (0 = disabled)
-        * pmksa_caching (bool) - whether PMKSA caching (opportunistic key caching) is enabled
+            * ssid (string) - SSID name (repeated for convenience)
+            * bridge (string) - bridge interface the VAP is attached to (e.g. ``"br-lan"``, ``"br-guest"``)
+            * vlan_id (int) - 802.1Q VLAN ID (0 = untagged / no VLAN separation)
+            * tagged (bool) - whether traffic is 802.1Q-tagged on the uplink port
+            * client_isolation (bool) - whether clients on this SSID are isolated from each other
 
-        Note: The RADIUS shared secret is intentionally omitted from the return
-        value for security reasons.
+            Example::
 
-        Example::
-
-            {
-                "CorpWiFi": {
-                    "enabled": True,
-                    "ssid": "CorpWiFi",
-                    "auth_server": {
-                        "host": "radius.corp.example",
-                        "port": 1812,
-                        "timeout": 5,
-                        "retries": 3,
+                {
+                    "CorpWiFi": {
+                        "ssid": "CorpWiFi",
+                        "bridge": "br-corp",
+                        "vlan_id": 10,
+                        "tagged": True,
+                        "client_isolation": False,
                     },
-                    "acct_server": {
-                        "host": "radius.corp.example",
-                        "port": 1813,
-                        "timeout": 5,
-                        "retries": 3,
+                    "GuestNet": {
+                        "ssid": "GuestNet",
+                        "bridge": "br-guest",
+                        "vlan_id": 20,
+                        "tagged": True,
+                        "client_isolation": True,
+                    },
+                    "IoT": {
+                        "ssid": "IoT",
+                        "bridge": "br-iot",
+                        "vlan_id": 30,
+                        "tagged": True,
+                        "client_isolation": True,
                     },
-                    "reauth_interval": 3600,
-                    "pmksa_caching": True,
                 }
-            }
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_packages(self) -> List[PackageDict]:
-        """
-        Returns all packages currently known to the device's package manager
-        (e.g. ``opkg`` on OpenWrt, ``apk`` on Alpine-based APs).
 
-        Each entry contains:
 
-        * name (string) - package name
-        * version (string) - installed or available version string
-        * installed (bool) - ``True`` if the package is currently installed
-        * description (string) - short package description
-        * size (int) - package size in bytes (0 if unknown)
-        * source (string) - repository / feed the package comes from
+        def get_dot1x_config(self) -> Dict[str, Dot1XConfigDict]:
+            """
+            Returns the 802.1X / WPA-Enterprise (RADIUS) configuration per SSID.
 
-        Example::
+            Keys are SSID names.  Each value contains:
+
+            * enabled (bool) - whether 802.1X authentication is active on this SSID
+            * ssid (string) - SSID name (repeated for convenience)
+            * auth_server (dict) - RADIUS authentication server:
+
+              * host (string) - IP or FQDN of the RADIUS server
+              * port (int) - UDP port (default 1812)
+              * timeout (int) - request timeout in seconds
+              * retries (int) - number of retransmissions
+
+            * acct_server (dict or None) - RADIUS accounting server (same keys as auth_server,
+              ``None`` if accounting is not configured)
+            * reauth_interval (int) - re-authentication interval in seconds (0 = disabled)
+            * pmksa_caching (bool) - whether PMKSA caching (opportunistic key caching) is enabled
+
+            Note: The RADIUS shared secret is intentionally omitted from the return
+            value for security reasons.
+
+            Example::
 
-            [
                 {
-                    "name": "luci-app-statistics",
-                    "version": "git-24.001.00000-1",
-                    "installed": True,
-                    "description": "LuCI Statistics application",
-                    "size": 20480,
-                    "source": "openwrt/packages",
-                },
-                {
-                    "name": "collectd-mod-wireless",
-                    "version": "5.12.0-24",
-                    "installed": False,
-                    "description": "Wireless statistics plugin for collectd",
-                    "size": 8192,
-                    "source": "openwrt/packages",
-                },
-            ]
-        """
-        raise NotImplementedError
+                    "CorpWiFi": {
+                        "enabled": True,
+                        "ssid": "CorpWiFi",
+                        "auth_server": {
+                            "host": "radius.corp.example",
+                            "port": 1812,
+                            "timeout": 5,
+                            "retries": 3,
+                        },
+                        "acct_server": {
+                            "host": "radius.corp.example",
+                            "port": 1813,
+                            "timeout": 5,
+                            "retries": 3,
+                        },
+                        "reauth_interval": 3600,
+                        "pmksa_caching": True,
+                    }
+                }
+            """
+            ...
 
-    def install_package(self, name: str, version: str = "") -> None:
-        """
-        Installs a package on the device.
 
-        The method blocks until the installation is complete.  After it returns
-        successfully the package is available for use without a reboot
-        (where the underlying package manager supports this).
 
-        :param name: Package name as known to the package manager.
-        :param version: Exact version to install.  An empty string (default)
-            installs the latest available version.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package name is unknown or the version does
-            not exist in any configured feed.
-        :raises RuntimeError: If the installation fails on the device side
-            (e.g. dependency conflict, disk full).
 
-        Example::
 
-            driver.install_package("luci-app-statistics")
-            driver.install_package("collectd", version="5.12.0-24")
-        """
-        raise NotImplementedError
 
-    def remove_package(self, name: str) -> None:
-        """
-        Removes an installed package from the device.
 
-        The method blocks until the removal is complete.
 
-        :param name: Package name to remove.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not currently installed.
-        :raises RuntimeError: If the removal fails on the device side
-            (e.g. other packages depend on it).
 
-        Example::
-
-            driver.remove_package("luci-app-statistics")
-        """
-        raise NotImplementedError
-
-    def get_package_config(self, name: str) -> Dict[str, Any]:
-        """
-        Returns the current configuration of an installed package as a
-        dictionary.  The structure is package-specific.
-
-        :param name: Package name.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not installed.
-
-        Example::
-
-            driver.get_package_config("luci-app-statistics")
-            # →
-            {
-                "collectd": {
-                    "enabled": True,
-                    "interval": 30,
-                },
-                "rrdtool": {
-                    "datadir": "/tmp/rrd",
-                    "stepsize": 30,
-                    "heartbeat": 60,
-                },
-            }
-        """
-        raise NotImplementedError
-
-    def get_services(self) -> List[ServiceDict]:
-        """
-        Returns the list of system services known to the device's init system.
-
-        Each entry contains:
-
-        * name (string)    - service name as registered with the init system
-        * running (bool)   - ``True`` if the service process is currently running
-        * enabled (bool)   - ``True`` if the service starts automatically at boot
-        * pid (int)        - process ID of the main service process; 0 if not running
-
-        Example::
-
-            [
-                {"name": "lldpd",   "running": True,  "enabled": True,  "pid": 2341},
-                {"name": "sshd",    "running": True,  "enabled": True,  "pid": 1198},
-                {"name": "cron",    "running": False, "enabled": False, "pid": 0},
-            ]
-        """
-        raise NotImplementedError
-
-    def manage_service(self, name: str, action: str) -> Dict[str, Any]:
-        """
-        Execute a lifecycle action on a named service.
-
-        :param name: Service name as returned by :meth:`get_services`.
-        :param action: One of ``start``, ``stop``, ``restart``, ``enable``, ``disable``.
-        :returns: ``{"success": bool, "output": str}``
-        :raises ValueError: If ``name`` or ``action`` is invalid.
-        :raises NotImplementedError: If the driver does not support service management.
-        """
-        raise NotImplementedError
-
-    def get_available_updates(self) -> List[UpdateDict]:
-        """
-        Returns the list of installed packages that have a newer version available.
-
-        Uses the local package manager cache — does not run ``opkg update`` / ``apk update``.
-
-        :returns: List of :class:`~napalm_device_types.models.UpdateDict`.
-        :raises NotImplementedError: If the driver does not support update listing.
-
-        Example::
-
-            [
-                {"name": "busybox",  "current_version": "1.36.1-1", "new_version": "1.37.0-1"},
-                {"name": "dropbear", "current_version": "2022.83-2", "new_version": "2024.86-1"},
-            ]
-        """
-        raise NotImplementedError
-
-    def apply_updates(self, packages: List[str]) -> Dict[str, Any]:
-        """
-        Upgrade one or more packages to their newest available version.
-
-        :param packages: List of package names to upgrade.
-        :returns: ``{"success": bool, "output": str}``
-        :raises NotImplementedError: If the driver does not support package upgrades.
-        """
-        raise NotImplementedError
-
-    def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
-        """
-        Writes a new configuration for an installed package.
-
-        The ``config`` dict must match the structure returned by
-        :meth:`get_package_config`.  Unknown keys are ignored or raise a
-        ``ValueError`` depending on the driver implementation.
-
-        Changes take effect immediately where the package supports live
-        reload; otherwise a package restart or device reboot may be
-        required – behaviour is driver-specific.
-
-        :param name: Package name.
-        :param config: New configuration as a nested dictionary.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not installed or the configuration
-            contains invalid values.
-        :raises RuntimeError: If the device rejects the configuration.
-
-        Example::
-
-            driver.set_package_config(
-                "luci-app-statistics",
-                {
-                    "collectd": {"enabled": True, "interval": 60},
-                    "rrdtool": {"datadir": "/tmp/rrd", "stepsize": 60, "heartbeat": 120},
-                },
-            )
-        """
-        raise NotImplementedError

napalm_device_types/base.py

@@ -12,6 +12,9 @@ from typing import NamedTuple
 
 from napalm.base import NetworkDriver
 
+from napalm_device_types.host_reboot import HostRebootMixin
+from napalm_device_types.ping_sweep import PingSweepMixin
+
 
 class FingerprintRule(NamedTuple):
     """Single pattern-matching rule for device fingerprinting.
@@ -45,13 +48,15 @@ class PortSpec(NamedTuple):
     mandatory: bool = False
 
 
-class DeviceTypeDriver(NetworkDriver):
+class DeviceTypeDriver(PingSweepMixin, HostRebootMixin, NetworkDriver):
     """Common base for all netOrk device-type drivers.
 
     Sits between napalm.base.NetworkDriver and the type-specific abstract
     classes (FirewallDriver, SwitchDriver, …). Adds the fingerprinting
-    interface consumed by the discovery subsystem; does not implement any
-    NAPALM abstract methods.
+    interface consumed by the discovery subsystem plus the generic
+    ``ping_sweep()`` from :class:`~napalm_device_types.ping_sweep.PingSweepMixin`
+    (usable by every driver that implements NAPALM's ``ping()``); does not
+    implement any NAPALM abstract methods.
 
     Override these class attributes in each concrete driver:
 
@@ -76,5 +81,15 @@ class DeviceTypeDriver(NetworkDriver):
     SSH_FINGERPRINT: list[FingerprintRule] = []
     HTTP_FINGERPRINT: list[FingerprintRule] = []
     OUI_PREFIXES: list[str] = []
+
+    #: Whether netOrk reaches this device over SSH. False for drivers that talk
+    #: to a REST API instead -- which decides whether an SSH credential is worth
+    #: asking for, and whether an SSH-shaped error message would even make sense.
+    USES_SSH: bool = True
+
+    #: How long a reboot takes before the device is worth polling again, in
+    #: seconds. Hypervisors and general-purpose OS hosts run through a full
+    #: init sequence; a switch or an access point is back in half the time.
+    REBOOT_SETTLE_SECONDS: int = 45
     # Format: "AA:BB:CC" — first 3 octets of MAC, uppercase, colon-separated.
     # A match contributes fixed weight 6.0 to the fingerprint score.

napalm_device_types/config_lifecycle.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import difflib
-from typing import ClassVar
+from typing import ClassVar, TYPE_CHECKING
 
 from napalm.base.exceptions import (
     CommandErrorException,
@@ -39,11 +39,13 @@ class ConfigLifecycleMixin:
 
     _comment_chars: ClassVar[tuple[str, ...]] = ("!", "#")
 
-    def _get_running_config(self) -> str:
-        raise NotImplementedError
+    if TYPE_CHECKING:
 
-    def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
-        raise NotImplementedError
+        def _get_running_config(self) -> str:
+            ...
+
+        def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
+            ...
 
     def load_merge_candidate(
         self, filename: str | None = None, config: str | None = None

napalm_device_types/dhcp.py

@@ -0,0 +1,400 @@
+# -*- coding: utf-8 -*-
+"""DHCP desired-state management, shared by firewalls and gateways.
+
+Covers two independent desired-state sets:
+
+* **reservations** -- static MAC -> IP bindings, matched on the MAC;
+* **subnets** -- the served ranges and their per-subnet DHCP options,
+  matched on the CIDR.
+
+Both :class:`~napalm_device_types.firewall.FirewallDriver` and
+:class:`~napalm_device_types.residential_gateway.ResidentialGatewayDriver`
+mix this in, because both device types commonly run the DHCP server for
+their networks.
+
+Only the six get/apply/commit methods are device-specific and must be
+implemented by a concrete driver; the diffs and the apply loops are
+vendor-neutral algorithms and live here -- see README.md "Design principle:
+generic vs. device-specific logic".
+
+Neither diff ever deletes. For subnets that is not just caution: removing one
+takes DHCP down for an entire VLAN.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, Iterator, List, Optional, TYPE_CHECKING
+
+from napalm_device_types.models import (
+    DhcpReservationDiffDict,
+    DhcpReservationDict,
+    DhcpReservationUpdateDict,
+    DhcpSubnetDiffDict,
+    DhcpSubnetDict,
+    DhcpSubnetUpdateDict,
+)
+
+# `mac` is the identity, so it is matched rather than compared. `uuid` is
+# assigned by the device and never part of the desired state.
+_DHCP_RESERVATION_COMPARE_FIELDS = (
+    "ip",
+    "hostname",
+    "description",
+    "subnet",
+)
+
+# `subnet` (the CIDR) is the identity, so it is matched rather than compared.
+# `uuid` is assigned by the device and never part of the desired state.
+_DHCP_SUBNET_COMPARE_FIELDS = (
+    "description",
+    "pools",
+    "option_data",
+    "match_client_id",
+)
+
+_HEX_ONLY = re.compile(r"[^0-9a-f]")
+
+
+def normalize_mac(mac: Optional[str]) -> str:
+    """Reduces a MAC address to lowercase colon-separated form.
+
+    Devices report MACs in whatever form their config store happens to use --
+    ``AA-BB-CC-DD-EE-01``, ``aabb.ccdd.ee01``, ``AABBCCDDEE01``. Reservations
+    are matched on this value, so it has to be canonical before comparison.
+
+    A value that is not 12 hex digits is returned lowercased and stripped
+    instead of raising: a malformed device response should degrade to "this
+    entry never matches" rather than abort the whole diff.
+    """
+    if not mac:
+        return ""
+
+    lowered = mac.strip().lower()
+    hex_digits = _HEX_ONLY.sub("", lowered)
+    if len(hex_digits) != 12:
+        return lowered
+
+    return ":".join(hex_digits[i : i + 2] for i in range(0, 12, 2))
+
+
+def normalize_cidr(cidr: Optional[str]) -> str:
+    """Reduces a subnet CIDR to a canonical string for matching.
+
+    Only whitespace and case are normalised -- deliberately not the network
+    address itself. Rewriting ``10.10.20.5/24`` to ``10.10.20.0/24`` would
+    make a caller's typo silently match a real subnet and then apply that
+    caller's pools and options to it.
+    """
+    return (cidr or "").strip().lower()
+
+
+def _subnet_field_differs(
+    field: str, live: DhcpSubnetDict, desired: DhcpSubnetDict
+) -> bool:
+    """Compares one subnet field, with `option_data` handled specially.
+
+    For `option_data` only the options `desired` actually names are compared;
+    see ``diff_dhcp_subnets`` for why an unmentioned option must not count as
+    a difference.
+    """
+    if field != "option_data":
+        return live.get(field) != desired.get(field)
+
+    desired_options = desired.get("option_data") or {}
+    live_options = live.get("option_data") or {}
+    return any(
+        live_options.get(option) != value for option, value in desired_options.items()
+    )
+
+
+class DhcpServerMixin:
+    """Mixin providing DHCP reservation and subnet read/diff/apply.
+
+    Concrete drivers **must** provide ``get_dhcp_reservations()``,
+    ``apply_dhcp_reservation()`` and ``commit_dhcp_reservations()``; drivers
+    that also manage subnets provide ``get_dhcp_subnets()``,
+    ``apply_dhcp_subnet()`` and ``commit_dhcp_subnets()``.
+    """
+
+    # ------------------------------------------------------------------
+    # Device-specific -- must be implemented by the concrete driver.
+    # ------------------------------------------------------------------
+
+    if TYPE_CHECKING:
+
+        def get_dhcp_reservations(self) -> List[DhcpReservationDict]:
+            """
+            Returns all static DHCP reservations currently configured on the
+            device, across all subnets.
+
+            This is the *configured* state, not the observed leases -- see
+            ``get_dhcp_leases()`` for the latter.
+
+            :raises NotImplementedError: If the driver does not support reading
+                DHCP reservations.
+            """
+            ...
+
+        def apply_dhcp_reservation(
+            self, reservation: DhcpReservationDict, *, uuid: Optional[str] = None
+        ) -> Dict[str, Any]:
+            """
+            Creates or updates a single static DHCP reservation on the device.
+
+            :param reservation: The desired reservation state, vendor-neutral.
+            :param uuid: If given, update the existing reservation with this ID
+                in-place. If ``None``, create a new one.
+            :raises NotImplementedError: If the driver does not support writing
+                DHCP reservations.
+            :raises ValueError: If `reservation` names a subnet the device does
+                not serve.
+            :raises RuntimeError: If the device rejects the write.
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+
+        def get_dhcp_subnets(self) -> List[DhcpSubnetDict]:
+            """
+            Returns every DHCPv4 subnet the device serves, with its pools and
+            per-subnet options.
+
+            :raises NotImplementedError: If the driver does not support reading
+                DHCP subnets.
+            """
+            ...
+
+        def apply_dhcp_subnet(
+            self, subnet: DhcpSubnetDict, *, uuid: Optional[str] = None
+        ) -> Dict[str, Any]:
+            """
+            Creates or updates a single DHCPv4 subnet on the device.
+
+            Implementations must treat ``subnet["option_data"]`` as a partial
+            update: an option the caller did not name is left as the device has
+            it. Managing `domain_search` alone is the common case, and it must
+            not silently drop the `routers` the server autocollected.
+
+            :param subnet: The desired subnet state, vendor-neutral.
+            :param uuid: If given, update the existing subnet with this ID
+                in-place. If ``None``, create a new one.
+            :raises NotImplementedError: If the driver does not support writing
+                DHCP subnets.
+            :raises RuntimeError: If the device rejects the write.
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+
+        def commit_dhcp_subnets(self) -> Dict[str, Any]:
+            """
+            Applies pending subnet changes (e.g. Kea's ``service/reconfigure``).
+
+            Separate from ``commit_dhcp_reservations`` even where a driver
+            implements both with the same call: the two desired-state sets are
+            applied independently, and a caller that changed only subnets should
+            not have to know which reload the vendor happens to share.
+
+            :raises NotImplementedError: If the driver does not support this.
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+
+        def commit_dhcp_reservations(self) -> Dict[str, Any]:
+            """
+            Applies pending reservation changes (e.g. Kea's ``service/reconfigure``,
+            or a dnsmasq reload).
+
+            Call once after one or more `apply_dhcp_reservation()` calls -- not
+            after every single reservation, and not at all when nothing changed:
+            on most implementations this reloads the DHCP daemon.
+
+            :raises NotImplementedError: If the driver does not support this
+                (e.g. reservations take effect immediately on write).
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+
+    # ------------------------------------------------------------------
+    # Generic, vendor-neutral algorithms.
+    # ------------------------------------------------------------------
+
+    def diff_dhcp_reservations(
+        self, desired: List[DhcpReservationDict]
+    ) -> DhcpReservationDiffDict:
+        """
+        Compares `desired` against the device's current reservations and
+        returns what would need to change to reach that state.
+
+        Matches on the normalised MAC address. A desired reservation with no
+        live counterpart becomes an "add"; a live one whose MAC matches but
+        whose other fields differ becomes an "update". Live reservations with
+        no matching desired entry are **not** reported for deletion -- a DHCP
+        server routinely carries hand-created reservations that a caller's
+        `desired` set was never meant to describe, and this method cannot tell
+        those apart from ones simply no longer wanted. Callers wanting
+        delete/cleanup semantics must implement that themselves, deliberately.
+
+        :param desired: The complete desired reservation set.
+        :returns: ``{"add": [...], "update": [{"uuid", "reservation",
+            "changed_fields"}, ...]}``.
+        """
+        live_by_mac: Dict[str, DhcpReservationDict] = {
+            normalize_mac(reservation.get("mac")): reservation
+            for reservation in self.get_dhcp_reservations()
+        }
+
+        add: List[DhcpReservationDict] = []
+        update: List[DhcpReservationUpdateDict] = []
+
+        for desired_reservation in desired:
+            live = live_by_mac.get(normalize_mac(desired_reservation.get("mac")))
+            if live is None:
+                add.append(desired_reservation)
+                continue
+
+            changed_fields = [
+                field
+                for field in _DHCP_RESERVATION_COMPARE_FIELDS
+                if live.get(field) != desired_reservation.get(field)
+            ]
+            if changed_fields:
+                update.append(
+                    {
+                        "uuid": live["uuid"],
+                        "reservation": desired_reservation,
+                        "changed_fields": changed_fields,
+                    }
+                )
+
+        return {"add": add, "update": update}
+
+    def apply_dhcp_reservationset(
+        self, desired: List[DhcpReservationDict]
+    ) -> Iterator[str]:
+        """
+        Computes the diff against `desired` and applies it, yielding one
+        human-readable progress line per change, then commits.
+
+        Unlike ``apply_firewall_ruleset``, an empty diff does **not** commit:
+        committing reloads the DHCP daemon and drops in-flight requests, which
+        is too high a price for a no-op run.
+
+        :param desired: The complete desired reservation set.
+        :yields: Progress lines, one per applied add/update, plus a final
+            commit line.
+        """
+        diff = self.diff_dhcp_reservations(desired)
+
+        for reservation in diff["add"]:
+            self.apply_dhcp_reservation(reservation)
+            yield f"[add] {reservation['mac']} -> {reservation['ip']}"
+
+        for entry in diff["update"]:
+            self.apply_dhcp_reservation(entry["reservation"], uuid=entry["uuid"])
+            fields = ", ".join(entry["changed_fields"])
+            yield (
+                f"[update] {entry['reservation']['mac']} -> "
+                f"{entry['reservation']['ip']} ({fields})"
+            )
+
+        if not diff["add"] and not diff["update"]:
+            yield "[commit] no changes"
+            return
+
+        self.commit_dhcp_reservations()
+        yield (
+            f"[commit] applied {len(diff['add'])} add(s), "
+            f"{len(diff['update'])} update(s)"
+        )
+
+    def diff_dhcp_subnets(self, desired: List[DhcpSubnetDict]) -> DhcpSubnetDiffDict:
+        """
+        Compares `desired` against the device's current subnets and returns
+        what would need to change to reach that state.
+
+        Matches on the CIDR. A desired subnet with no live counterpart becomes
+        an "add"; a live one whose CIDR matches but whose other fields differ
+        becomes an "update".
+
+        `option_data` is compared **per option**, and only over the options
+        the caller named. An option the device carries but `desired` does not
+        mention is left out of the comparison entirely, because absent means
+        "not managed" rather than "should be empty". Without that rule a
+        caller managing only `domain_search` would diff against every option
+        Kea autocollects (`routers`, `domain_name_servers`, `ntp_servers`) and
+        reconfigure the DHCP daemon on every single run.
+
+        Live subnets with no matching desired entry are **not** reported for
+        deletion, and more emphatically than for reservations: removing a
+        subnet takes DHCP down for a whole VLAN, and this method cannot tell
+        "no longer wanted" from "was never netOrk's to describe".
+
+        :param desired: The complete desired subnet set.
+        :returns: ``{"add": [...], "update": [{"uuid", "subnet",
+            "changed_fields"}, ...]}``.
+        """
+        live_by_cidr: Dict[str, DhcpSubnetDict] = {
+            normalize_cidr(live.get("subnet")): live for live in self.get_dhcp_subnets()
+        }
+
+        add: List[DhcpSubnetDict] = []
+        update: List[DhcpSubnetUpdateDict] = []
+
+        for desired_subnet in desired:
+            live = live_by_cidr.get(normalize_cidr(desired_subnet.get("subnet")))
+            if live is None:
+                add.append(desired_subnet)
+                continue
+
+            changed_fields = [
+                field
+                for field in _DHCP_SUBNET_COMPARE_FIELDS
+                if _subnet_field_differs(field, live, desired_subnet)
+            ]
+            if changed_fields:
+                update.append(
+                    {
+                        "uuid": live["uuid"],
+                        "subnet": desired_subnet,
+                        "changed_fields": changed_fields,
+                    }
+                )
+
+        return {"add": add, "update": update}
+
+    def apply_dhcp_subnetset(self, desired: List[DhcpSubnetDict]) -> Iterator[str]:
+        """
+        Computes the diff against `desired` and applies it, yielding one
+        human-readable progress line per change, then commits.
+
+        As with ``apply_dhcp_reservationset``, an empty diff does **not**
+        commit: reconfiguring the DHCP daemon is too expensive for a no-op.
+
+        :param desired: The complete desired subnet set.
+        :yields: Progress lines, one per applied add/update, plus a final
+            commit line.
+        """
+        diff = self.diff_dhcp_subnets(desired)
+
+        for subnet in diff["add"]:
+            self.apply_dhcp_subnet(subnet)
+            yield f"[add] {subnet['subnet']}"
+
+        for entry in diff["update"]:
+            self.apply_dhcp_subnet(entry["subnet"], uuid=entry["uuid"])
+            fields = ", ".join(entry["changed_fields"])
+            yield f"[update] {entry['subnet']['subnet']} ({fields})"
+
+        if not diff["add"] and not diff["update"]:
+            yield "[commit] no changes"
+            return
+
+        self.commit_dhcp_subnets()
+        yield (
+            f"[commit] applied {len(diff['add'])} add(s), "
+            f"{len(diff['update'])} update(s)"
+        )

napalm_device_types/firewall.py

@@ -10,21 +10,21 @@ Usage::
             ...
 """
 
-from typing import Any, Dict, List
+from typing import Any, ClassVar, Dict, Iterator, List, Optional, TYPE_CHECKING
 from napalm_device_types.base import DeviceTypeDriver
-from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.nat_vpn import NatVpnMixin
+from napalm_device_types.packages import PackageManagementMixin
+from napalm_device_types.health_metrics import HealthMetricsMixin
+from napalm_device_types.dhcp import DhcpServerMixin
+from napalm_device_types.firewall_rules import FirewallRuleMixin
 from napalm_device_types.models import (
-    HealthMetricsDict,
-    NATTranslationDict,
-    PackageDict,
     SecurityZoneDict,
     SessionDict,
-    VPNTunnelDict,
 )
 
 
-class FirewallDriver(DeviceTypeDriver):
-    TYPE_LABEL: str = "Firewall"
+
+class FirewallDriver(NatVpnMixin, PackageManagementMixin, HealthMetricsMixin, FirewallRuleMixin, DhcpServerMixin, DeviceTypeDriver):
     """
     Abstract intermediate driver for firewall/security devices.
 
@@ -33,268 +33,128 @@ class FirewallDriver(DeviceTypeDriver):
     that concrete drivers must implement.
     """
 
-    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "firewall"
+    TYPE_LABEL: str = "Firewall"
+
     # OPNsense reports drops in the out-error counter.
-    _SNMP_TX_ERR_IS_DROP: bool = True
+    _SNMP_TX_ERR_IS_DROP: ClassVar[bool] = True
 
-    @classmethod
-    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
-        return await collect_ucd_metrics(
-            snmp_get, snmp_walk,
-            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
-            if_skip=cls._SNMP_SKIP_IF,
-        )
 
-    def get_nat_translations(self) -> List[NATTranslationDict]:
-        """
-        Returns a list of active NAT translation entries.
+    if TYPE_CHECKING:
 
-        Each entry contains:
 
-        * protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"``
-        * inside_local (string) - original source address (IP or IP:port)
-        * inside_global (string) - translated source address (IP or IP:port)
-        * outside_local (string) - destination as seen from inside
-        * outside_global (string) - actual destination address
-        * age (float) - translation entry age in seconds
+        def get_security_zones(self) -> Dict[str, SecurityZoneDict]:
+            """
+            Returns the security zone configuration.
 
-        Example::
+            Keys are zone names.  Each value contains:
+
+            * interfaces (list of strings) - interfaces assigned to this zone
+            * policy (string) - name of the security policy applied to this zone
+            * description (string) - zone description
+
+            Example::
 
-            [
                 {
-                    "protocol": "tcp",
-                    "inside_local": "192.168.1.10:54321",
-                    "inside_global": "203.0.113.1:54321",
-                    "outside_local": "1.1.1.1:443",
-                    "outside_global": "1.1.1.1:443",
-                    "age": 120.5,
+                    "LAN": {
+                        "interfaces": ["eth0", "eth1"],
+                        "policy": "LAN-policy",
+                        "description": "Internal LAN zone",
+                    },
+                    "WAN": {
+                        "interfaces": ["eth2"],
+                        "policy": "WAN-policy",
+                        "description": "Uplink to internet",
+                    },
                 }
-            ]
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_security_zones(self) -> Dict[str, SecurityZoneDict]:
-        """
-        Returns the security zone configuration.
+        def get_sessions(self) -> List[SessionDict]:
+            """
+            Returns a list of active connection sessions (stateful flows).
 
-        Keys are zone names.  Each value contains:
+            Each entry contains:
 
-        * interfaces (list of strings) - interfaces assigned to this zone
-        * policy (string) - name of the security policy applied to this zone
-        * description (string) - zone description
+            * protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"``
+            * src_ip (string) - source IP address
+            * src_port (int) - source port (0 for ICMP)
+            * dst_ip (string) - destination IP address
+            * dst_port (int) - destination port (0 for ICMP)
+            * state (string) - session state, e.g. ``"established"``, ``"syn_sent"``
+            * age (float) - session age in seconds
 
-        Example::
+            Example::
 
-            {
-                "LAN": {
-                    "interfaces": ["eth0", "eth1"],
-                    "policy": "LAN-policy",
-                    "description": "Internal LAN zone",
-                },
-                "WAN": {
-                    "interfaces": ["eth2"],
-                    "policy": "WAN-policy",
-                    "description": "Uplink to internet",
-                },
-            }
-        """
-        raise NotImplementedError
+                [
+                    {
+                        "protocol": "tcp",
+                        "src_ip": "192.168.1.10",
+                        "src_port": 54321,
+                        "dst_ip": "1.1.1.1",
+                        "dst_port": 443,
+                        "state": "established",
+                        "age": 30.2,
+                    }
+                ]
+            """
+            ...
 
-    def get_sessions(self) -> List[SessionDict]:
-        """
-        Returns a list of active connection sessions (stateful flows).
 
-        Each entry contains:
+        def send_wake_on_lan(self, mac_address: str, interface: str = "") -> Dict[str, Any]:
+            """
+            Sends a Wake-on-LAN "magic packet" to wake a host on the network.
 
-        * protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"``
-        * src_ip (string) - source IP address
-        * src_port (int) - source port (0 for ICMP)
-        * dst_ip (string) - destination IP address
-        * dst_port (int) - destination port (0 for ICMP)
-        * state (string) - session state, e.g. ``"established"``, ``"syn_sent"``
-        * age (float) - session age in seconds
+            :param mac_address: Target host's MAC address (colon-separated,
+                case-insensitive, e.g. ``"AA:BB:CC:DD:EE:FF"``).
+            :param interface: Driver-specific interface identifier to broadcast the
+                magic packet from. Required by drivers that scope WOL per interface
+                (e.g. OPNsense); an empty string means "use the driver's default/
+                only broadcast domain." Consult the concrete driver's docstring for
+                the exact expected format.
+            :raises NotImplementedError: If the driver does not support Wake-on-LAN.
+            :raises ValueError: If ``mac_address`` is malformed, or ``interface`` is
+                required by this driver but was not provided.
 
-        Example::
+            :returns: A dict with:
 
-            [
-                {
-                    "protocol": "tcp",
-                    "src_ip": "192.168.1.10",
-                    "src_port": 54321,
-                    "dst_ip": "1.1.1.1",
-                    "dst_port": 443,
-                    "state": "established",
-                    "age": 30.2,
-                }
-            ]
-        """
-        raise NotImplementedError
+                * success (bool) - ``True`` if the magic packet was sent without error
+                * output (string) - human-readable status message
 
-    def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]:
-        """
-        Returns the status of VPN tunnels.
+            Example::
 
-        Keys are tunnel names or identifiers.  Each value contains:
+                driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
+                # → {"success": True, "output": "Magic packet sent to AA:BB:CC:DD:EE:FF via lan"}
 
-        * type (string) - tunnel type: ``"IPsec"``, ``"SSL"``, ``"GRE"``, ``"WireGuard"``
-        * local_endpoint (string) - local tunnel endpoint IP
-        * remote_endpoint (string) - remote tunnel endpoint IP
-        * is_up (bool) - whether the tunnel is operationally up
-        * uptime (int) - tunnel uptime in seconds (0 if down)
-        * bytes_in (int) - total bytes received through the tunnel
-        * bytes_out (int) - total bytes sent through the tunnel
+            .. note::
 
-        Example::
+                A driver whose ``interface`` is *not* the name :meth:`get_interfaces`
+                is keyed by must expose the name it does expect as an ``identifier``
+                key on each ``get_interfaces()`` entry. Without it a caller has no
+                way to offer a valid choice: OPNsense, for instance, keys interfaces
+                by the physical device ("em0") but wakes by the assigned name
+                ("lan"), and rejects the former. ``identifier`` is a non-standard
+                NAPALM key, so it reaches consumers through the usual passthrough
+                for extra interface data.
+            """
+            ...
 
-            {
-                "vpn-to-branch": {
-                    "type": "IPsec",
-                    "local_endpoint": "203.0.113.1",
-                    "remote_endpoint": "198.51.100.1",
-                    "is_up": True,
-                    "uptime": 86400,
-                    "bytes_in": 104857600,
-                    "bytes_out": 52428800,
-                }
-            }
-        """
-        raise NotImplementedError
 
-    def get_packages(self) -> List[PackageDict]:
-        """
-        Returns all packages / plugins currently known to the firewall's
-        package manager (e.g. ``pkg`` on pfSense/OPNsense, ``FortiGate
-        License`` add-ons, ``apt`` on Debian-based firewalls).
 
-        Each entry contains:
 
-        * name (string) - package name
-        * version (string) - installed or available version string
-        * installed (bool) - ``True`` if the package is currently installed
-        * description (string) - short package description
-        * size (int) - package size in bytes (0 if unknown)
-        * source (string) - repository / channel the package comes from
 
-        Example::
 
-            [
-                {
-                    "name": "pfBlockerNG",
-                    "version": "3.2.0_4",
-                    "installed": True,
-                    "description": "IP and DNS blocking for pfSense",
-                    "size": 2097152,
-                    "source": "pfSense-pkg",
-                },
-                {
-                    "name": "suricata",
-                    "version": "7.0.3_1",
-                    "installed": False,
-                    "description": "High-performance Network IDS/IPS",
-                    "size": 51380224,
-                    "source": "pfSense-pkg",
-                },
-            ]
-        """
-        raise NotImplementedError
+        # ------------------------------------------------------------------
+        # Firewall rule diff/apply. get_firewall_rules/apply_firewall_rule/
+        # commit_firewall_rules are abstract (device communication); everything
+        # else here is a concrete, vendor-neutral algorithm -- see README.md
+        # "Design principle: generic vs. device-specific logic".
+        # ------------------------------------------------------------------
 
-    def install_package(self, name: str, version: str = "") -> None:
-        """
-        Installs a package or plugin on the firewall.
 
-        The method blocks until the installation is complete.  Whether a
-        reboot is required afterwards depends on the device; check the vendor
-        documentation.
 
-        :param name: Package name as known to the package manager.
-        :param version: Exact version to install.  An empty string (default)
-            installs the latest available version.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package name is unknown or the requested
-            version is not available.
-        :raises RuntimeError: If the installation fails on the device side
-            (e.g. license missing, dependency conflict, disk full).
 
-        Example::
 
-            driver.install_package("pfBlockerNG")
-            driver.install_package("suricata", version="7.0.3_1")
-        """
-        raise NotImplementedError
-
-    def remove_package(self, name: str) -> None:
-        """
-        Removes an installed package or plugin from the firewall.
-
-        The method blocks until the removal is complete.
-
-        :param name: Package name to remove.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not currently installed.
-        :raises RuntimeError: If the removal fails on the device side
-            (e.g. the package is a system dependency).
-
-        Example::
-
-            driver.remove_package("pfBlockerNG")
-        """
-        raise NotImplementedError
-
-    def get_package_config(self, name: str) -> Dict[str, Any]:
-        """
-        Returns the current configuration of an installed package or plugin
-        as a dictionary.  The structure is package-specific.
-
-        :param name: Package name.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not installed.
-
-        Example::
-
-            driver.get_package_config("pfBlockerNG")
-            # →
-            {
-                "enable": True,
-                "maxmind_key": "",
-                "blocklists": [
-                    {"name": "PRI1", "action": "Deny_Both", "enabled": True},
-                    {"name": "DNSBL_ADs", "action": "Unbound", "enabled": True},
-                ],
-                "update_interval": "Once a day",
-            }
-        """
-        raise NotImplementedError
-
-    def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
-        """
-        Writes a new configuration for an installed package or plugin.
-
-        The ``config`` dict must match the structure returned by
-        :meth:`get_package_config`.  Unknown keys are ignored or raise a
-        ``ValueError`` depending on the driver implementation.
-
-        Changes take effect immediately where the package supports live
-        reload; otherwise a package restart or device reboot may be
-        required – behaviour is driver-specific.
-
-        :param name: Package name.
-        :param config: New configuration as a nested dictionary.
-        :raises NotImplementedError: If the driver does not support package management.
-        :raises ValueError: If the package is not installed or the configuration
-            contains invalid values.
-        :raises RuntimeError: If the device rejects the configuration.
-
-        Example::
-
-            driver.set_package_config(
-                "pfBlockerNG",
-                {
-                    "enable": True,
-                    "blocklists": [
-                        {"name": "PRI1", "action": "Deny_Both", "enabled": True},
-                    ],
-                    "update_interval": "Twice a day",
-                },
-            )
-        """
-        raise NotImplementedError

napalm_device_types/firewall_rules.py

@@ -0,0 +1,161 @@
+# -*- coding: utf-8 -*-
+"""Firewall rule reconciliation: generic here, device communication in the driver.
+
+The split follows the rule in this package's README: matching, diffing and the
+apply-then-commit sequence would be identical for any vendor's firewall, so they
+live here as concrete methods. Only the three hooks below touch the device, and
+a concrete driver supplies those.
+
+Declared under ``if TYPE_CHECKING``, the hooks do not exist at runtime until a
+driver implements them -- so ``hasattr`` stays an honest answer to "can this
+driver manage firewall rules", and mixing this class in can never shadow a
+working implementation inherited from elsewhere.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Iterator, List, Optional, TYPE_CHECKING
+
+from napalm_device_types.models import (
+    FirewallRuleDict,
+    FirewallRuleDiffDict,
+    FirewallRuleUpdateDict,
+)
+
+_FIREWALL_RULE_COMPARE_FIELDS = (
+    "action",
+    "interface",
+    "direction",
+    "protocol",
+    "source_net",
+    "source_port",
+    "destination_net",
+    "destination_port",
+    "log",
+    "quick",
+    "enabled",
+)
+
+
+class FirewallRuleMixin:
+    """Adds generic firewall-rule diff and apply on top of three device hooks."""
+
+    if TYPE_CHECKING:
+
+        def get_firewall_rules(self) -> List[FirewallRuleDict]:
+            """
+            Returns all firewall filter rules currently configured on the device.
+
+            `description` must be a stable, human-assigned identifier -- it is
+            the key used to match rules across calls (most firewall vendors
+            don't expose an ID a caller can pre-assign).
+
+            :raises NotImplementedError: If the driver does not support reading
+                firewall rules.
+            """
+            ...
+        def apply_firewall_rule(
+            self, rule: FirewallRuleDict, *, uuid: Optional[str] = None
+        ) -> Dict[str, Any]:
+            """
+            Creates or updates a single firewall filter rule on the device.
+
+            :param rule: The desired rule state, in vendor-neutral form.
+            :param uuid: If given, update the existing rule with this ID
+                in-place. If ``None``, create a new rule.
+            :raises NotImplementedError: If the driver does not support writing
+                firewall rules.
+            :raises ValueError: If `rule` references an alias/interface the
+                device doesn't know about.
+            :raises RuntimeError: If the device rejects the write.
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+        def commit_firewall_rules(self) -> Dict[str, Any]:
+            """
+            Applies pending firewall filter rule changes (e.g. reloads pf/pfctl,
+            or whatever the device's equivalent of "Apply Changes" is).
+
+            Call once after one or more `apply_firewall_rule()` calls -- not
+            after every single rule.
+
+            :raises NotImplementedError: If the driver does not support this
+                (e.g. rules take effect immediately on write).
+
+            :returns: A dict with at least ``{"success": bool}``.
+            """
+            ...
+
+    def diff_firewall_rules(self, desired: List[FirewallRuleDict]) -> FirewallRuleDiffDict:
+        """
+        Compares `desired` against the device's current rules and returns
+        what would need to change to reach that state.
+
+        Matches rules by `description`. A desired rule with no live
+        counterpart becomes an "add"; a live rule whose description matches
+        but whose other fields differ becomes an "update". Live rules with
+        no matching desired entry are **not** reported for deletion -- this
+        is intentionally conservative: a firewall may carry manually-created
+        or otherwise unmanaged rules that a caller's `desired` set was never
+        meant to describe, and this method has no way to distinguish those
+        from ones simply no longer wanted. Callers wanting delete/cleanup
+        semantics must implement that themselves, deliberately.
+
+        :param desired: The complete desired rule set.
+        :returns: ``{"add": [...], "update": [{"uuid", "rule",
+            "changed_fields"}, ...]}``.
+        """
+        live_by_description: Dict[str, FirewallRuleDict] = {
+            rule["description"]: rule for rule in self.get_firewall_rules()
+        }
+
+        add: List[FirewallRuleDict] = []
+        update: List[FirewallRuleUpdateDict] = []
+
+        for desired_rule in desired:
+            live_rule = live_by_description.get(desired_rule["description"])
+            if live_rule is None:
+                add.append(desired_rule)
+                continue
+
+            changed_fields = [
+                field
+                for field in _FIREWALL_RULE_COMPARE_FIELDS
+                if live_rule.get(field) != desired_rule.get(field)
+            ]
+            if changed_fields:
+                update.append(
+                    {
+                        "uuid": live_rule["uuid"],
+                        "rule": desired_rule,
+                        "changed_fields": changed_fields,
+                    }
+                )
+
+        return {"add": add, "update": update}
+    def apply_firewall_ruleset(self, desired: List[FirewallRuleDict]) -> Iterator[str]:
+        """
+        Computes the diff against `desired` and applies it, yielding one
+        human-readable progress line per change, then commits.
+
+        Intended for streaming to a caller (e.g. an SSE endpoint) that wants
+        live progress while writing to a real device.
+
+        :param desired: The complete desired rule set.
+        :yields: Progress lines, one per applied add/update, plus a final
+            commit line.
+        """
+        diff = self.diff_firewall_rules(desired)
+
+        for rule in diff["add"]:
+            self.apply_firewall_rule(rule)
+            yield f"[add] {rule['description']}"
+
+        for entry in diff["update"]:
+            self.apply_firewall_rule(entry["rule"], uuid=entry["uuid"])
+            fields = ", ".join(entry["changed_fields"])
+            yield f"[update] {entry['rule']['description']} ({fields})"
+
+        self.commit_firewall_rules()
+        yield f"[commit] applied {len(diff['add'])} add(s), {len(diff['update'])} update(s)"

napalm_device_types/health_metrics.py

@@ -0,0 +1,53 @@
+# -*- coding: utf-8 -*-
+"""SNMP health collection, shared by every device type that speaks UCD-MIB.
+
+The collection itself is generic: CPU, memory, uptime, load and per-interface
+counters come from the same standard OIDs on any UCD-MIB/IF-MIB capable device.
+Only two details vary by device type, and both are class attributes rather than
+code -- which is why this was five byte-identical copies of the same method
+before it moved here.
+
+A device type whose vendor publishes the same data under proprietary OIDs does
+**not** mix this in; ``SwitchDriver`` is the example, and its concrete drivers
+implement ``get_health_metrics`` themselves.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Awaitable, Callable, ClassVar, Dict, cast
+
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.models import HealthMetricsDict
+
+
+class HealthMetricsMixin:
+    """Adds the standard UCD-MIB/IF-MIB ``get_health_metrics`` collector."""
+
+    #: Interfaces whose counters are noise rather than signal. Loopback,
+    #: tunnels and container bridges inflate error rates without meaning.
+    _SNMP_SKIP_IF: ClassVar[re.Pattern[str]] = IF_SKIP_DEFAULT
+
+    #: Whether the device reports discards in the TX-error counter. Firewalls
+    #: do -- a dropped packet is the point, not a fault -- so counting those as
+    #: errors would make a healthy firewall look broken.
+    _SNMP_TX_ERR_IS_DROP: ClassVar[bool] = False
+
+    @classmethod
+    async def get_health_metrics(
+        cls,
+        snmp_get: Callable[..., Awaitable[Any]],
+        snmp_walk: Callable[..., Awaitable[Dict[str, Any]]],
+    ) -> HealthMetricsDict:
+        """Collect SNMP health metrics (CPU, memory, uptime, load, interfaces)."""
+        # collect_ucd_metrics is untyped shared plumbing; the shape it returns is
+        # the contract HealthMetricsDict describes.
+        return cast(
+            HealthMetricsDict,
+            await collect_ucd_metrics(
+                snmp_get,
+                snmp_walk,
+                tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+                if_skip=cls._SNMP_SKIP_IF,
+            ),
+        )

napalm_device_types/host_reboot.py

@@ -0,0 +1,30 @@
+"""Restarting the device itself.
+
+Declared under ``if TYPE_CHECKING``: a contract, not a placeholder. Only a
+driver that can actually restart its device defines ``reboot_host``, so
+``hasattr(driver, "reboot_host")`` tells a caller whether to offer it.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+
+class HostRebootMixin:
+    if TYPE_CHECKING:
+
+        def reboot_host(self) -> None:
+            """
+            Restarts the device this driver is connected to.
+
+            Returns once the device has accepted the request; the session is
+            usually gone right after. Waiting for the device to come back is
+            the caller's business (see ``REBOOT_SETTLE_SECONDS``).
+
+            A driver that manages other machines restarts its *own* host, never
+            one of them: a hypervisor restarts the hypervisor, not a VM.
+
+            :raises RuntimeError: If the device refuses, e.g. an ESXi host that
+                is not in maintenance mode while VMs are running.
+            """
+            ...

napalm_device_types/interface_filter.py

@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+"""Dropping interfaces that are noise rather than signal.
+
+An access point exposes radio PHYs and a loopback alongside its real
+interfaces. Reporting them inflates every interface count and every error rate,
+so drivers filter them out -- identically, whatever the vendor. The two class
+attributes are the customisation point.
+"""
+
+from __future__ import annotations
+
+from typing import Any, ClassVar, Dict, FrozenSet, Tuple
+
+
+class InterfaceFilterMixin:
+    """Adds :meth:`_filter_interfaces` for drivers that report raw interface dicts."""
+
+    #: Interface names dropped outright.
+    _EXCLUDED_INTERFACES: ClassVar[FrozenSet[str]] = frozenset({"lo"})
+
+    #: Name prefixes dropped -- ``phy*`` are radio devices, not links.
+    _EXCLUDED_INTERFACE_PREFIXES: ClassVar[Tuple[str, ...]] = ("phy",)
+
+    def _filter_interfaces(self, interfaces: Dict[str, Any]) -> Dict[str, Any]:
+        """Remove loopback and radio-device (phy*) interfaces from an interface dict."""
+        return {
+            name: data
+            for name, data in interfaces.items()
+            if name not in self._EXCLUDED_INTERFACES
+            and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES)
+        }

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/mac_acl.py

@@ -0,0 +1,79 @@
+# -*- coding: utf-8 -*-
+"""MAC-based access control, per SSID on an AP and per port on a switch.
+
+The reader is identical either way; only the writer is AP-specific so far.
+
+Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
+Nothing exists at runtime until a concrete driver implements it, so mixing
+this class in can never shadow a working implementation from a sibling base.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, List, TYPE_CHECKING
+
+from napalm_device_types.models import MACACLDict
+
+
+class MacAclMixin:
+    if TYPE_CHECKING:
+
+        def get_mac_acl(self) -> Dict[str, MACACLDict]:
+            """
+            Returns the MAC-address-based access control lists configured per SSID.
+
+            Keys are SSID names.  Each value contains:
+
+            * ssid (string) - SSID name (repeated for convenience)
+            * policy (string) - ACL mode:
+
+              * ``"allow"`` – whitelist: only listed MACs may associate
+              * ``"deny"``  – blacklist: listed MACs are blocked
+              * ``"disabled"`` – no MAC filtering active
+
+            * entries (list) - ACL entries, each with:
+
+              * mac (string) - MAC address (normalised, colon-separated)
+              * action (string) - ``"allow"`` or ``"deny"``
+              * description (string) - optional human-readable label
+
+            Example::
+
+                {
+                    "CorpWiFi": {
+                        "name": "CorpWiFi",
+                        "policy": "allow",
+                        "entries": [
+                            {"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "CEO-Laptop"},
+                            {"mac": "AA:BB:CC:DD:EE:02", "action": "allow", "description": "CFO-Laptop"},
+                        ],
+                    },
+                    "GuestNet": {
+                        "name": "GuestNet",
+                        "policy": "deny",
+                        "entries": [
+                            {"mac": "DE:AD:BE:EF:00:01", "action": "deny", "description": "blocked device"},
+                        ],
+                    },
+                }
+            """
+            ...
+
+        def push_mac_acl(self, ssid_name: str, mode: str, macs: List[str]) -> None:
+            """
+            Rewrites the MAC-address access control list for a single SSID.
+
+            Full-rebuild semantics: replaces whatever ACL state currently exists
+            for *ssid_name* with *mode* + *macs* — not a diff/patch.
+
+            :param ssid_name: SSID name to apply the ACL to.
+            :param mode: ``"off"`` | ``"whitelist"`` | ``"blacklist"``.
+            :param macs: MAC addresses for the active list. Ignored when ``mode == "off"``.
+            :raises NotImplementedError: If the driver does not support MAC ACL push.
+
+            Example::
+
+                driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"])
+                driver.push_mac_acl("GuestNet", "off", [])
+            """
+            ...

napalm_device_types/media.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+"""
+Abstract base class for networked media players.
+
+Usage::
+
+    from napalm_device_types import MediaDriver
+
+    class SonosDriver(MediaDriver):
+        ...
+"""
+
+from typing import TYPE_CHECKING, Any, Dict, List
+
+from napalm_device_types.base import DeviceTypeDriver
+
+
+class MediaDriver(DeviceTypeDriver):
+    """
+    Abstract intermediate driver for speakers, streamers and media renderers
+    (e.g. Sonos, Chromecast, Squeezebox, UPnP/DLNA renderers).
+
+    Like :class:`~napalm_device_types.phone.PhoneDriver`, this exists so an
+    endpoint stops being filed under ``AccessPointDriver`` for want of anywhere
+    better. A speaker has no SSIDs, no radio configuration and no AP profile; it
+    has a transport state and a volume.
+    """
+
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "media"
+    TYPE_LABEL: str = "Media"
+
+    if TYPE_CHECKING:
+
+        def get_playback_state(self) -> Dict[str, Any]:
+            """
+            Returns the transport state of the renderer.
+
+            * state (string) - ``"playing"``, ``"paused"``, ``"stopped"``, or ``"transitioning"``
+            * source (string) - input or service currently selected
+            """
+            ...
+
+        def get_volume(self) -> int:
+            """Returns the current output volume, 0-100."""
+            ...
+
+        def set_volume(self, level: int) -> None:
+            """Sets the output volume, 0-100."""
+            ...
+
+        def get_zone_info(self) -> List[Dict[str, Any]]:
+            """
+            Returns the zones or groups this renderer participates in.
+
+            Each entry contains:
+
+            * name (string) - zone/room name
+            * coordinator (bool) - whether this device leads the group
+            * members (list) - names of the other devices in the group
+            """
+            ...

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
@@ -314,6 +329,133 @@ class SessionDict(TypedDict):
     age: float
 
 
+class FirewallRuleDict(TypedDict):
+    """A single firewall filter rule, in vendor-neutral form.
+
+    `description` is the stable matching key across get_firewall_rules()/
+    diff_firewall_rules()/apply_firewall_rule() -- firewall vendors
+    generally don't expose an ID a caller can pre-assign, so the rule's
+    human description is what ties a "desired" rule to its "live"
+    counterpart. `source_net`/`source_port`/`destination_net`/
+    `destination_port` are plain strings (comma-joined by the caller if a
+    rule references multiple aliases) -- driver methods never expand or
+    split them.
+    """
+
+    uuid: str
+    description: str
+    action: str
+    interface: str
+    direction: str
+    protocol: str
+    source_net: str
+    source_port: str
+    destination_net: str
+    destination_port: str
+    enabled: bool
+    quick: bool
+    log: bool
+
+
+class FirewallRuleUpdateDict(TypedDict):
+    uuid: str
+    rule: FirewallRuleDict
+    changed_fields: List[str]
+
+
+class FirewallRuleDiffDict(TypedDict):
+    add: List[FirewallRuleDict]
+    update: List[FirewallRuleUpdateDict]
+
+
+class DhcpReservationDict(TypedDict):
+    """A single static DHCP reservation (MAC -> IP), in vendor-neutral form.
+
+    `mac` is the stable matching key across get_dhcp_reservations()/
+    diff_dhcp_reservations()/apply_dhcp_reservation() -- unlike firewall
+    rules, a reservation has a natural identity, and it is the MAC address.
+    It is compared after normalisation (see ``dhcp.normalize_mac``), so
+    devices that report ``AA-BB-CC-DD-EE-01`` still match a desired
+    ``aa:bb:cc:dd:ee:01``.
+
+    `subnet` is the CIDR the reservation lives in. It is a *compared* field,
+    not part of the key: a host that moves to another VLAN keeps its MAC, and
+    that is an update of the existing reservation rather than a second one.
+    """
+
+    uuid: str
+    mac: str
+    ip: str
+    hostname: str
+    description: str
+    subnet: str
+
+
+class DhcpReservationUpdateDict(TypedDict):
+    uuid: str
+    reservation: DhcpReservationDict
+    changed_fields: List[str]
+
+
+class DhcpReservationDiffDict(TypedDict):
+    add: List[DhcpReservationDict]
+    update: List[DhcpReservationUpdateDict]
+
+
+class DhcpOptionDataDict(TypedDict, total=False):
+    """DHCPv4 options carried by a subnet, by their RFC/Kea names.
+
+    Only options netOrk actually models are listed. `domain_search` (option
+    119) is the reason this type exists: it is the one option that cannot be
+    expressed anywhere else in the stack, and no DHCP server autocollects it.
+
+    Every field is optional and an absent key means "do not manage this
+    option" -- distinct from an empty list, which means "manage it, and the
+    desired value is empty". A driver must preserve options it was not given.
+    """
+
+    routers: List[str]
+    domain_name_servers: List[str]
+    domain_name: str
+    domain_search: List[str]
+    ntp_servers: List[str]
+
+
+class DhcpSubnetDict(TypedDict):
+    """A DHCPv4 subnet served by the device, in vendor-neutral form.
+
+    `subnet` (the CIDR) is the stable matching key, the way `mac` is for a
+    reservation. Renaming is not a thing a subnet does; changing its CIDR
+    makes it a different subnet.
+
+    `pools` are address ranges in ``"start-end"`` form, the shape both Kea
+    and ISC DHCP use.
+
+    `option_data` carries the per-subnet DHCP options. Servers that
+    autocollect some of them (Kea fills `routers`, `domain_name_servers` and
+    `ntp_servers` when ``option_data_autocollect`` is on) still never
+    autocollect `domain_name` or `domain_search`.
+    """
+
+    uuid: str
+    subnet: str
+    description: str
+    pools: List[str]
+    option_data: DhcpOptionDataDict
+    match_client_id: bool
+
+
+class DhcpSubnetUpdateDict(TypedDict):
+    uuid: str
+    subnet: DhcpSubnetDict
+    changed_fields: List[str]
+
+
+class DhcpSubnetDiffDict(TypedDict):
+    add: List[DhcpSubnetDict]
+    update: List[DhcpSubnetUpdateDict]
+
+
 class VPNTunnelDict(TypedDict):
     type: str
     local_endpoint: str
@@ -343,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
@@ -385,7 +517,7 @@ class VMNICDict(TypedDict):
 
 class VMDict(TypedDict):
     name: str
-    vmid: int
+    vmid: str
     status: str
     vcpus: int
     memory: int
@@ -395,9 +527,17 @@ class VMDict(TypedDict):
     node: str
 
 
+class VMPassthroughDict(TypedDict):
+    """A host device handed through to a VM (PCI, USB)."""
+
+    slot: str    # hypervisor's device key, e.g. "hostpci0"
+    kind: str    # "pci" or "usb"
+    config: str  # hypervisor's own description of the device
+
+
 class VMConfigDict(TypedDict):
     name: str
-    vmid: int
+    vmid: str
     vcpus: int
     memory: int
     os_type: str
@@ -406,6 +546,14 @@ class VMConfigDict(TypedDict):
     nics: List[VMNICDict]
     description: str
     tags: List[str]
+    # Hardware details not every hypervisor exposes; absent when unknown.
+    os_name: NotRequired[str]            # human-readable guest OS, e.g. "Ubuntu Linux (64-bit)"
+    cpu_type: NotRequired[str]           # e.g. "host", "kvm64"
+    sockets: NotRequired[int]
+    cores_per_socket: NotRequired[int]
+    firmware: NotRequired[str]           # "bios" or "efi"
+    machine: NotRequired[str]            # machine type / virtual hardware version
+    passthrough: NotRequired[List[VMPassthroughDict]]
 
 
 class StorageVolumeDict(TypedDict):
@@ -713,6 +861,7 @@ class NICConfigDict(TypedDict):
     vlan_tag: NotRequired[int | None]  # Access VLAN (None = untagged)
     trunk_vlan_tags: NotRequired[list[int]]  # Trunk VLAN list (alternative to vlan_tag)
     dhcp: NotRequired[bool]  # Enable DHCP (default True for first NIC, False for others)
+    mac: NotRequired[str]  # Explicit MAC address (omit to let the hypervisor auto-generate one)
 
 
 class VMProvisionResultDict(TypedDict):
@@ -730,3 +879,57 @@ class VMStatusDict(TypedDict):
     ip_address: NotRequired[str]  # Management NIC IP (absent if VM has no IP or is stopped)
     hostname: NotRequired[str]  # Hostname resolved from IP (if available)
     mac_address: NotRequired[str]  # MAC of management NIC
+
+
+class NetworkTargetDict(TypedDict):
+    """A selectable network target for a new VM's NIC (``NICConfigDict.bridge``).
+
+    Distinguishes real bridges (Linux or OVS) from SDN network segments (vnets),
+    and tells the caller whether a separate ``vlan_tag`` may be applied on top:
+    - Linux bridge: vlan_aware reflects the bridge's own ``bridge_vlan_aware`` flag.
+    - OVS bridge: always vlan_aware (OVS bridges tag per-port regardless of a
+      dedicated "VLAN aware" setting).
+    - SDN vnet: never vlan_aware — the VLAN is already fixed by the vnet's zone/tag,
+      so a NIC attached to it must not also carry a ``vlan_tag``. That fixed VLAN
+      is surfaced via ``fixed_vlan_tag`` instead, for display purposes.
+    """
+
+    name: str  # Bridge, vnet or port group name, usable directly as NICConfigDict.bridge
+    kind: str  # "bridge", "vnet" or "portgroup" (VMware: VLAN fixed like a vnet's)
+    vlan_aware: bool  # True if a NICConfigDict.vlan_tag may be set on top of this target
+    fixed_vlan_tag: NotRequired[int | None]  # vnet only: the VLAN ID already baked into it
+
+
+class StorageTargetDict(TypedDict):
+    """A selectable storage pool for a new VM's root disk (``create_vm_from_cloud_init``'s
+    ``storage`` argument), scoped to the specific node the VM will be created on —
+    a storage restricted to other cluster nodes must not appear here.
+    """
+
+    name: str  # Storage pool name, usable directly as create_vm_from_cloud_init(storage=...)
+    type: str  # Backend type: "dir", "lvmthin", "zfspool", "nfs", etc.
+    total_gb: float  # Total capacity in gigabytes
+    available_gb: float  # Free capacity in gigabytes
+
+
+# ---------------------------------------------------------------------------
+# Ping sweep  (shared across device types)
+# ---------------------------------------------------------------------------
+
+
+class PingSweepEntryDict(TypedDict):
+    """Outcome of a single ``ping`` inside a sweep (see ``PingSweepMixin``)."""
+
+    ip: str                  # destination that was probed
+    alive: bool              # True if at least one probe was answered
+    rtt_ms: Optional[float]  # average round-trip time in ms; None if unreachable
+    error: NotRequired[str]  # driver/transport error for this destination
+
+
+class PingSweepResultDict(TypedDict):
+    """Result of a ``ping_sweep()`` call."""
+
+    entries: List[PingSweepEntryDict]  # one entry per probed destination, in input order
+    scanned: int                       # destinations actually probed
+    alive_count: int                   # entries with alive=True
+    truncated: bool                    # True if targets were dropped at the sweep cap

napalm_device_types/nat_vpn.py

@@ -0,0 +1,118 @@
+# -*- coding: utf-8 -*-
+"""Address translation and VPN tunnels.
+
+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
+this class in can never shadow a working implementation from a sibling base.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, List, TYPE_CHECKING
+
+from napalm_device_types.models import NATTranslationDict, PortForwardDict, VPNTunnelDict
+
+
+class NatVpnMixin:
+    if TYPE_CHECKING:
+
+        def get_nat_translations(self) -> List[NATTranslationDict]:
+            """
+            Returns a list of active NAT translation entries.
+
+            Each entry contains:
+
+            * protocol (string) - ``"tcp"``, ``"udp"``, ``"icmp"``
+            * inside_local (string) - original source address (IP or IP:port)
+            * inside_global (string) - translated source address (IP or IP:port)
+            * outside_local (string) - destination as seen from inside
+            * outside_global (string) - actual destination address
+            * age (float) - translation entry age in seconds
+
+            Example::
+
+                [
+                    {
+                        "protocol": "tcp",
+                        "inside_local": "192.168.1.10:54321",
+                        "inside_global": "203.0.113.1:54321",
+                        "outside_local": "1.1.1.1:443",
+                        "outside_global": "1.1.1.1:443",
+                        "age": 120.5,
+                    }
+                ]
+            """
+            ...
+
+        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.
+
+            Keys are tunnel names or identifiers.  Each value contains:
+
+            * type (string) - tunnel type: ``"IPsec"``, ``"SSL"``, ``"GRE"``, ``"WireGuard"``
+            * local_endpoint (string) - local tunnel endpoint IP
+            * remote_endpoint (string) - remote tunnel endpoint IP
+            * is_up (bool) - whether the tunnel is operationally up
+            * uptime (int) - tunnel uptime in seconds (0 if down)
+            * bytes_in (int) - total bytes received through the tunnel
+            * bytes_out (int) - total bytes sent through the tunnel
+
+            Example::
+
+                {
+                    "vpn-to-branch": {
+                        "type": "IPsec",
+                        "local_endpoint": "203.0.113.1",
+                        "remote_endpoint": "198.51.100.1",
+                        "is_up": True,
+                        "uptime": 86400,
+                        "bytes_in": 104857600,
+                        "bytes_out": 52428800,
+                    }
+                }
+            """
+            ...

napalm_device_types/os.py

@@ -10,27 +10,23 @@ Usage::
             ...
 """
 
-import re
-from typing import List, Optional
+from typing import List, Optional, TYPE_CHECKING
 from napalm_device_types.base import DeviceTypeDriver
-from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.updates import UpdateMixin
+from napalm_device_types.services import ServiceControlMixin
+from napalm_device_types.packages import PackageManagementMixin
+from napalm_device_types.health_metrics import HealthMetricsMixin
 from napalm_device_types.models import (
-    ApplyUpdatesResultDict,
     CronJobDict,
     DeviceActionResultDict,
     DockerInfoDict,
-    HealthMetricsDict,
-    PackageDict,
     ProcessDict,
-    ServiceDict,
     SNMPConfigDict,
-    UpdateDict,
     UserDict,
 )
 
 
-class OSDriver(DeviceTypeDriver):
-    TYPE_LABEL: str = "OS"
+class OSDriver(UpdateMixin, ServiceControlMixin, PackageManagementMixin, HealthMetricsMixin, DeviceTypeDriver):
     """
     Abstract intermediate driver for general-purpose operating systems
     (e.g. Linux, BSD, macOS).
@@ -39,391 +35,280 @@ class OSDriver(DeviceTypeDriver):
     operations that concrete drivers must implement.
     """
 
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "linux"
+    TYPE_LABEL: str = "OS"
+
     # Interfaces matching this pattern are excluded from health-metric collection.
-    _SNMP_SKIP_IF: re.Pattern = IF_SKIP_DEFAULT
     # Set True for drivers where the out-error counter actually reports drops (e.g. OPNsense).
-    _SNMP_TX_ERR_IS_DROP: bool = False
 
-    @classmethod
-    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
-        """Collect SNMP health metrics (CPU, memory, uptime, load, interfaces).
-
-        :param snmp_get:  async callable ``(oid: str) -> Optional[str]``
-        :param snmp_walk: async callable ``(oid: str) -> Dict[str, str]``
-        """
-        return await collect_ucd_metrics(
-            snmp_get, snmp_walk,
-            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
-            if_skip=cls._SNMP_SKIP_IF,
-        )
 
     # ------------------------------------------------------------------
     # Package management
     # ------------------------------------------------------------------
 
-    def get_packages(self) -> List[PackageDict]:
-        """
-        Returns a list of all installed software packages.
+    if TYPE_CHECKING:
 
-        Each entry contains:
 
-        * name (string) - package name
-        * version (string) - installed version string
-        * installed (bool) - always ``True`` for this method
-        * description (string) - short package description
-        * size (int) - installed size in bytes; ``0`` if unavailable
-        * source (string) - package source / repository name; empty string if unavailable
 
-        Example::
 
-            [
-                {
-                    "name": "openssh-server",
-                    "version": "1:9.2p1-2+deb12u2",
-                    "installed": True,
-                    "description": "secure shell (SSH) server, for secure access from remote machines",
-                    "size": 524288,
-                    "source": "Debian",
-                },
-            ]
-        """
-        raise NotImplementedError
+        # ------------------------------------------------------------------
+        # Service management
+        # ------------------------------------------------------------------
 
-    def get_pending_updates(self) -> List[UpdateDict]:
-        """
-        Returns a list of packages that have a newer version available.
 
-        Each entry contains:
+        # ------------------------------------------------------------------
+        # Users
+        # ------------------------------------------------------------------
 
-        * name (string) - package name
-        * current_version (string) - currently installed version
-        * new_version (string) - version available in the repository
+        def get_users(self) -> List[UserDict]:
+            """
+            Returns a list of local OS user accounts.
 
-        Example::
+            Each entry contains:
 
-            [
-                {
-                    "name": "openssh-server",
-                    "current_version": "1:9.2p1-2+deb12u1",
-                    "new_version": "1:9.2p1-2+deb12u2",
-                },
-            ]
-        """
-        raise NotImplementedError
+            * username (string) - login name
+            * uid (int) - numeric user ID
+            * gid (int) - primary group ID
+            * home (string) - home directory path
+            * shell (string) - login shell path
+            * groups (list of strings) - all supplementary group names
 
-    def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
-        """
-        Upgrades the given packages to the newest available version.
+            Example::
 
-        Only packages that are already installed may be upgraded; this method
-        does **not** install new packages.  Pass an empty list to upgrade
-        **all** packages that have pending updates.
-
-        :param packages: List of package names to upgrade.  Each name must
-            match ``^[a-zA-Z0-9_\\-\\+\\.]+$``; a :exc:`ValueError` is raised
-            for any name that does not conform.
-        :returns: A dict with:
-
-            * success (bool) – ``True`` if the package manager exited without error
-            * output (string) – combined stdout / stderr from the package manager
-            * error (string, optional) – short error message when *success* is ``False``
-
-        :raises ValueError: If any package name fails the safety check.
-
-        Example::
-
-            result = driver.apply_updates(["openssh-server", "curl"])
-            # → {"success": True, "output": "Reading package lists...\\n..."}
-
-            # Upgrade everything:
-            result = driver.apply_updates([])
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # Service management
-    # ------------------------------------------------------------------
-
-    def get_services(self) -> List[ServiceDict]:
-        """
-        Returns a list of system services and their current state.
-
-        Each entry contains:
-
-        * name (string) - service unit name (without ``.service`` suffix)
-        * running (bool) - ``True`` if the service is currently active
-        * enabled (bool) - ``True`` if the service starts automatically on boot
-        * pid (int) - main process ID; ``0`` if not running
-
-        Example::
-
-            [
-                {
-                    "name": "ssh",
-                    "running": True,
-                    "enabled": True,
-                    "pid": 1234,
-                },
-                {
-                    "name": "cron",
-                    "running": True,
-                    "enabled": True,
-                    "pid": 5678,
-                },
-            ]
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # Users
-    # ------------------------------------------------------------------
-
-    def get_users(self) -> List[UserDict]:
-        """
-        Returns a list of local OS user accounts.
-
-        Each entry contains:
-
-        * username (string) - login name
-        * uid (int) - numeric user ID
-        * gid (int) - primary group ID
-        * home (string) - home directory path
-        * shell (string) - login shell path
-        * groups (list of strings) - all supplementary group names
-
-        Example::
-
-            [
-                {
-                    "username": "admin",
-                    "uid": 1000,
-                    "gid": 1000,
-                    "home": "/home/admin",
-                    "shell": "/bin/bash",
-                    "groups": ["sudo", "docker", "adm"],
-                },
-            ]
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # Processes
-    # ------------------------------------------------------------------
-
-    def get_processes(self) -> List[ProcessDict]:
-        """
-        Returns a snapshot of currently running processes.
-
-        Each entry contains:
-
-        * pid (int) - process ID
-        * ppid (int) - parent process ID
-        * user (string) - effective user name
-        * cpu (float) - CPU utilisation percentage
-        * memory (float) - RSS as a percentage of total RAM
-        * vsz (int) - virtual memory size in KiB
-        * rss (int) - resident set size in KiB
-        * tty (string) - controlling terminal; empty string if none
-        * state (string) - process state: ``"R"`` running, ``"S"`` sleeping,
-          ``"D"`` uninterruptible, ``"Z"`` zombie, ``"T"`` stopped, etc.
-        * started (string) - start time as printed by ``ps`` (e.g. ``"12:34"`` or ``"May28"``)
-        * command (string) - full command line
-
-        Example::
-
-            [
-                {
-                    "pid": 1,
-                    "ppid": 0,
-                    "user": "root",
-                    "cpu": 0.0,
-                    "memory": 0.1,
-                    "vsz": 168576,
-                    "rss": 13312,
-                    "tty": "",
-                    "state": "S",
-                    "started": "May28",
-                    "command": "/sbin/init",
-                },
-            ]
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # Cron jobs
-    # ------------------------------------------------------------------
-
-    def get_cron_jobs(self) -> List[CronJobDict]:
-        """
-        Returns scheduled cron tasks from all user crontabs and ``/etc/cron.d``.
-
-        Each entry contains:
-
-        * user (string) - owner of the crontab entry
-        * schedule (string) - five-field cron expression (e.g. ``"0 * * * *"``)
-        * command (string) - shell command to execute
-        * description (string, optional) - inline comment text if present
-
-        Example::
-
-            [
-                {
-                    "user": "root",
-                    "schedule": "0 4 * * *",
-                    "command": "/usr/local/bin/backup.sh",
-                    "description": "nightly backup",
-                },
-            ]
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # SNMP
-    # ------------------------------------------------------------------
-
-    def get_snmp_config(self) -> Optional[SNMPConfigDict]:
-        """
-        Returns the SNMP agent configuration currently active on the device,
-        or ``None`` if no SNMP daemon is running or detectable.
-
-        The returned dictionary contains:
-
-        * running (bool) - whether the SNMP daemon is currently active
-        * community (string) - the read community string (e.g. ``"public"``)
-        * port (int) - the UDP port the agent listens on (default ``161``)
-        * version (string) - highest supported SNMP version: ``"1"``, ``"2c"``, or ``"3"``
-
-        Example::
-
-            # snmpd running with community "public":
-            {
-                "running": True,
-                "community": "public",
-                "port": 161,
-                "version": "2c",
-            }
-
-            # snmpd not installed / not running:
-            None
-        """
-        raise NotImplementedError
-
-    # ------------------------------------------------------------------
-    # Docker
-    # ------------------------------------------------------------------
-
-    def get_docker_info(self) -> DockerInfoDict:
-        """
-        Returns information about the local Docker environment.
-
-        If Docker is not installed or the current user lacks access to the
-        Docker socket, returns ``{"available": False}``.  When the user has
-        no socket permission, ``permission_denied`` is additionally set to
-        ``True``.
-
-        When Docker is available the dict contains:
-
-        * available (bool) - always ``True``
-        * version (string) - Docker Engine version string
-        * containers (list) - all containers (running and stopped), each with:
-
-          * id (string) - short container ID
-          * name (string) - container name(s)
-          * image (string) - image reference
-          * image_version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
-          * command (string) - entrypoint / command string
-          * created (string) - creation timestamp string
-          * status (string) - human-readable status (e.g. ``"Up 3 hours"``)
-          * ports (string) - port mapping string
-          * state (string) - ``"running"``, ``"exited"``, ``"paused"``, etc.
-
-        * images (list) - local images, each with:
-
-          * id (string) - short image ID
-          * repository (string) - image repository
-          * tag (string) - image tag
-          * size (string) - human-readable size string (e.g. ``"187MB"``)
-          * created (string) - creation timestamp string
-          * version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
-
-        * volumes (list) - Docker volumes, each with:
-
-          * name (string) - volume name
-          * driver (string) - volume driver
-          * mountpoint (string) - host filesystem path
-          * scope (string) - ``"local"`` or ``"global"``
-
-        * networks (list) - Docker networks, each with:
-
-          * id (string) - short network ID
-          * name (string) - network name
-          * driver (string) - network driver (e.g. ``"bridge"``, ``"host"``, ``"overlay"``)
-          * scope (string) - network scope
-          * ipv6 (string) - ``"true"`` if IPv6 is enabled
-          * internal (string) - ``"true"`` if the network is internal
-
-        * outdated_images (list of strings) - image names where the local digest
-          differs from the latest remote digest; empty list if all images are
-          current or update checks could not be performed.
-
-        Example::
-
-            # Docker not installed:
-            {"available": False}
-
-            # Docker installed, no socket permission:
-            {"available": False, "permission_denied": True}
-
-            # Docker available:
-            {
-                "available": True,
-                "version": "Docker version 27.3.1, build ce12230",
-                "containers": [
+                [
                     {
-                        "id": "a1b2c3d4e5f6",
-                        "name": "my-app",
-                        "image": "nginx:latest",
-                        "image_version": "1.27.0",
-                        "command": "nginx -g 'daemon off;'",
-                        "created": "2026-05-28 10:00:00 +0000 UTC",
-                        "status": "Up 3 days",
-                        "ports": "0.0.0.0:80->80/tcp",
-                        "state": "running",
+                        "username": "admin",
+                        "uid": 1000,
+                        "gid": 1000,
+                        "home": "/home/admin",
+                        "shell": "/bin/bash",
+                        "groups": ["sudo", "docker", "adm"],
                     },
-                ],
-                "images": [...],
-                "volumes": [],
-                "networks": [...],
-                "outdated_images": ["nginx:latest"],
-            }
-        """
-        raise NotImplementedError
+                ]
+            """
+            ...
 
-    # ------------------------------------------------------------------
-    # Generic device actions
-    # ------------------------------------------------------------------
+        # ------------------------------------------------------------------
+        # Processes
+        # ------------------------------------------------------------------
 
-    def run_device_action(self, action: str) -> DeviceActionResultDict:
-        """
-        Executes a named administrative action on the device.
+        def get_processes(self) -> List[ProcessDict]:
+            """
+            Returns a snapshot of currently running processes.
 
-        This method is an extensibility point for driver-specific one-off
-        operations that do not fit any other NAPALM API method.  Each driver
-        documents the action names it supports.
+            Each entry contains:
 
-        :param action: Action identifier string (e.g. ``"fix_docker_permissions"``).
-        :raises NotImplementedError: If the driver does not implement this method.
-        :raises ValueError: If ``action`` is not a recognised action name for
-            this driver.
+            * pid (int) - process ID
+            * ppid (int) - parent process ID
+            * user (string) - effective user name
+            * cpu (float) - CPU utilisation percentage
+            * memory (float) - RSS as a percentage of total RAM
+            * vsz (int) - virtual memory size in KiB
+            * rss (int) - resident set size in KiB
+            * tty (string) - controlling terminal; empty string if none
+            * state (string) - process state: ``"R"`` running, ``"S"`` sleeping,
+              ``"D"`` uninterruptible, ``"Z"`` zombie, ``"T"`` stopped, etc.
+            * started (string) - start time as printed by ``ps`` (e.g. ``"12:34"`` or ``"May28"``)
+            * command (string) - full command line
 
-        :returns: A dict with:
+            Example::
 
-            * success (bool) – ``True`` if the action completed without error
-            * output (string) – human-readable output or status message
+                [
+                    {
+                        "pid": 1,
+                        "ppid": 0,
+                        "user": "root",
+                        "cpu": 0.0,
+                        "memory": 0.1,
+                        "vsz": 168576,
+                        "rss": 13312,
+                        "tty": "",
+                        "state": "S",
+                        "started": "May28",
+                        "command": "/sbin/init",
+                    },
+                ]
+            """
+            ...
 
-        Example::
+        # ------------------------------------------------------------------
+        # Cron jobs
+        # ------------------------------------------------------------------
 
-            result = driver.run_device_action("fix_docker_permissions")
-            # → {"success": True, "output": "Added 'pi' to the docker group. Reconnect..."}
-        """
-        raise NotImplementedError
+        def get_cron_jobs(self) -> List[CronJobDict]:
+            """
+            Returns scheduled cron tasks from all user crontabs and ``/etc/cron.d``.
+
+            Each entry contains:
+
+            * user (string) - owner of the crontab entry
+            * schedule (string) - five-field cron expression (e.g. ``"0 * * * *"``)
+            * command (string) - shell command to execute
+            * description (string, optional) - inline comment text if present
+
+            Example::
+
+                [
+                    {
+                        "user": "root",
+                        "schedule": "0 4 * * *",
+                        "command": "/usr/local/bin/backup.sh",
+                        "description": "nightly backup",
+                    },
+                ]
+            """
+            ...
+
+        # ------------------------------------------------------------------
+        # SNMP
+        # ------------------------------------------------------------------
+
+        def get_snmp_config(self) -> Optional[SNMPConfigDict]:
+            """
+            Returns the SNMP agent configuration currently active on the device,
+            or ``None`` if no SNMP daemon is running or detectable.
+
+            The returned dictionary contains:
+
+            * running (bool) - whether the SNMP daemon is currently active
+            * community (string) - the read community string (e.g. ``"public"``)
+            * port (int) - the UDP port the agent listens on (default ``161``)
+            * version (string) - highest supported SNMP version: ``"1"``, ``"2c"``, or ``"3"``
+
+            Example::
+
+                # snmpd running with community "public":
+                {
+                    "running": True,
+                    "community": "public",
+                    "port": 161,
+                    "version": "2c",
+                }
+
+                # snmpd not installed / not running:
+                None
+            """
+            ...
+
+        # ------------------------------------------------------------------
+        # Docker
+        # ------------------------------------------------------------------
+
+        def get_docker_info(self) -> DockerInfoDict:
+            """
+            Returns information about the local Docker environment.
+
+            If Docker is not installed or the current user lacks access to the
+            Docker socket, returns ``{"available": False}``.  When the user has
+            no socket permission, ``permission_denied`` is additionally set to
+            ``True``.
+
+            When Docker is available the dict contains:
+
+            * available (bool) - always ``True``
+            * version (string) - Docker Engine version string
+            * containers (list) - all containers (running and stopped), each with:
+
+              * id (string) - short container ID
+              * name (string) - container name(s)
+              * image (string) - image reference
+              * image_version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
+              * command (string) - entrypoint / command string
+              * created (string) - creation timestamp string
+              * status (string) - human-readable status (e.g. ``"Up 3 hours"``)
+              * ports (string) - port mapping string
+              * state (string) - ``"running"``, ``"exited"``, ``"paused"``, etc.
+
+            * images (list) - local images, each with:
+
+              * id (string) - short image ID
+              * repository (string) - image repository
+              * tag (string) - image tag
+              * size (string) - human-readable size string (e.g. ``"187MB"``)
+              * created (string) - creation timestamp string
+              * version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
+
+            * volumes (list) - Docker volumes, each with:
+
+              * name (string) - volume name
+              * driver (string) - volume driver
+              * mountpoint (string) - host filesystem path
+              * scope (string) - ``"local"`` or ``"global"``
+
+            * networks (list) - Docker networks, each with:
+
+              * id (string) - short network ID
+              * name (string) - network name
+              * driver (string) - network driver (e.g. ``"bridge"``, ``"host"``, ``"overlay"``)
+              * scope (string) - network scope
+              * ipv6 (string) - ``"true"`` if IPv6 is enabled
+              * internal (string) - ``"true"`` if the network is internal
+
+            * outdated_images (list of strings) - image names where the local digest
+              differs from the latest remote digest; empty list if all images are
+              current or update checks could not be performed.
+
+            Example::
+
+                # Docker not installed:
+                {"available": False}
+
+                # Docker installed, no socket permission:
+                {"available": False, "permission_denied": True}
+
+                # Docker available:
+                {
+                    "available": True,
+                    "version": "Docker version 27.3.1, build ce12230",
+                    "containers": [
+                        {
+                            "id": "a1b2c3d4e5f6",
+                            "name": "my-app",
+                            "image": "nginx:latest",
+                            "image_version": "1.27.0",
+                            "command": "nginx -g 'daemon off;'",
+                            "created": "2026-05-28 10:00:00 +0000 UTC",
+                            "status": "Up 3 days",
+                            "ports": "0.0.0.0:80->80/tcp",
+                            "state": "running",
+                        },
+                    ],
+                    "images": [...],
+                    "volumes": [],
+                    "networks": [...],
+                    "outdated_images": ["nginx:latest"],
+                }
+            """
+            ...
+
+        # ------------------------------------------------------------------
+        # Generic device actions
+        # ------------------------------------------------------------------
+
+        def run_device_action(self, action: str) -> DeviceActionResultDict:
+            """
+            Executes a named administrative action on the device.
+
+            This method is an extensibility point for driver-specific one-off
+            operations that do not fit any other NAPALM API method.  Each driver
+            documents the action names it supports.
+
+            :param action: Action identifier string (e.g. ``"fix_docker_permissions"``).
+            :raises NotImplementedError: If the driver does not implement this method.
+            :raises ValueError: If ``action`` is not a recognised action name for
+                this driver.
+
+            :returns: A dict with:
+
+                * success (bool) – ``True`` if the action completed without error
+                * output (string) – human-readable output or status message
+
+            Example::
+
+                result = driver.run_device_action("fix_docker_permissions")
+                # → {"success": True, "output": "Added 'pi' to the docker group. Reconnect..."}
+            """
+            ...

napalm_device_types/packages.py

@@ -0,0 +1,158 @@
+# -*- coding: utf-8 -*-
+"""Listing and managing installed software.
+
+Identical on every device type that has a package manager -- this was five
+byte-identical copies before it moved here. A driver whose device offers no
+package management simply implements none of it.
+
+Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
+Nothing exists at runtime until a concrete driver implements it, so mixing
+this class in can never shadow a working implementation from a sibling base.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, TYPE_CHECKING
+
+from napalm_device_types.models import PackageDict
+
+
+class PackageManagementMixin:
+    if TYPE_CHECKING:
+
+        def get_packages(self) -> List[PackageDict]:
+            """
+            Returns all packages currently known to the device's package manager
+            (e.g. ``opkg`` on OpenWrt, ``apk`` on Alpine-based APs).
+
+            Each entry contains:
+
+            * name (string) - package name
+            * version (string) - installed or available version string
+            * installed (bool) - ``True`` if the package is currently installed
+            * description (string) - short package description
+            * size (int) - package size in bytes (0 if unknown)
+            * source (string) - repository / feed the package comes from
+
+            Example::
+
+                [
+                    {
+                        "name": "luci-app-statistics",
+                        "version": "git-24.001.00000-1",
+                        "installed": True,
+                        "description": "LuCI Statistics application",
+                        "size": 20480,
+                        "source": "openwrt/packages",
+                    },
+                    {
+                        "name": "collectd-mod-wireless",
+                        "version": "5.12.0-24",
+                        "installed": False,
+                        "description": "Wireless statistics plugin for collectd",
+                        "size": 8192,
+                        "source": "openwrt/packages",
+                    },
+                ]
+            """
+            ...
+
+        def install_package(self, name: str, version: str = "") -> None:
+            """
+            Installs a package on the device.
+
+            The method blocks until the installation is complete.  After it returns
+            successfully the package is available for use without a reboot
+            (where the underlying package manager supports this).
+
+            :param name: Package name as known to the package manager.
+            :param version: Exact version to install.  An empty string (default)
+                installs the latest available version.
+            :raises NotImplementedError: If the driver does not support package management.
+            :raises ValueError: If the package name is unknown or the version does
+                not exist in any configured feed.
+            :raises RuntimeError: If the installation fails on the device side
+                (e.g. dependency conflict, disk full).
+
+            Example::
+
+                driver.install_package("luci-app-statistics")
+                driver.install_package("collectd", version="5.12.0-24")
+            """
+            ...
+
+        def uninstall_package(self, name: str) -> None:
+            """
+            Removes an installed package from the device.
+
+            The method blocks until the removal is complete.
+
+            :param name: Package name to remove.
+            :raises NotImplementedError: If the driver does not support package management.
+            :raises ValueError: If the package is not currently installed.
+            :raises RuntimeError: If the removal fails on the device side
+                (e.g. other packages depend on it).
+
+            Example::
+
+                driver.uninstall_package("luci-app-statistics")
+            """
+            ...
+
+        def get_package_config(self, name: str) -> Dict[str, Any]:
+            """
+            Returns the current configuration of an installed package as a
+            dictionary.  The structure is package-specific.
+
+            :param name: Package name.
+            :raises NotImplementedError: If the driver does not support package management.
+            :raises ValueError: If the package is not installed.
+
+            Example::
+
+                driver.get_package_config("luci-app-statistics")
+                # →
+                {
+                    "collectd": {
+                        "enabled": True,
+                        "interval": 30,
+                    },
+                    "rrdtool": {
+                        "datadir": "/tmp/rrd",
+                        "stepsize": 30,
+                        "heartbeat": 60,
+                    },
+                }
+            """
+            ...
+
+        def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
+            """
+            Writes a new configuration for an installed package.
+
+            The ``config`` dict must match the structure returned by
+            :meth:`get_package_config`.  Unknown keys are ignored or raise a
+            ``ValueError`` depending on the driver implementation.
+
+            Changes take effect immediately where the package supports live
+            reload; otherwise a package restart or device reboot may be
+            required – behaviour is driver-specific.
+
+            :param name: Package name.
+            :param config: New configuration as a nested dictionary.
+            :raises NotImplementedError: If the driver does not support package management.
+            :raises ValueError: If the package is not installed or the configuration
+                contains invalid values.
+            :raises RuntimeError: If the device rejects the configuration.
+
+            Example::
+
+                driver.set_package_config(
+                    "luci-app-statistics",
+                    {
+                        "collectd": {"enabled": True, "interval": 60},
+                        "rrdtool": {"datadir": "/tmp/rrd", "stepsize": 60, "heartbeat": 120},
+                    },
+                )
+            """
+            ...

napalm_device_types/phone.py

@@ -0,0 +1,71 @@
+# -*- coding: utf-8 -*-
+"""
+Abstract base class for IP telephony endpoints.
+
+Usage::
+
+    from napalm_device_types import PhoneDriver
+
+    class YealinkDriver(PhoneDriver):
+        ...
+"""
+
+from typing import TYPE_CHECKING, Any, Dict, List
+
+from napalm_device_types.base import DeviceTypeDriver
+
+
+class PhoneDriver(DeviceTypeDriver):
+    """
+    Abstract intermediate driver for desk phones, conference units and DECT
+    bases (e.g. Yealink T-series, Snom, Grandstream, Fanvil).
+
+    A phone is an endpoint, not infrastructure. It was worth separating from
+    ``AccessPointDriver`` — which some phone drivers used to inherit for want of
+    anywhere better — because netOrk offers access points for AP profiles,
+    wireless management and SSID drift, none of which a desk phone should appear
+    in. A WiFi-capable phone still reports its radio through its own methods; it
+    just is not an access point.
+    """
+
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "phone"
+    TYPE_LABEL: str = "Phone"
+
+    if TYPE_CHECKING:
+
+        def get_sip_accounts(self) -> List[Dict[str, Any]]:
+            """
+            Returns the SIP registrations configured on the phone.
+
+            Each entry contains:
+
+            * account (string) - display label or line key
+            * user (string) - SIP user / extension
+            * server (string) - registrar host
+            * registered (bool) - whether the registration is currently active
+
+            Example::
+
+                [
+                    {
+                        "account": "Line 1",
+                        "user": "201",
+                        "server": "pbx.example.com",
+                        "registered": True,
+                    }
+                ]
+            """
+            ...
+
+        def get_call_status(self) -> Dict[str, Any]:
+            """
+            Returns what the phone is doing right now.
+
+            * state (string) - ``"idle"``, ``"ringing"``, ``"talking"``, or ``"held"``
+            * remote (string) - remote party, empty string when idle
+            * duration (int) - seconds in the current state; ``0`` when idle
+            """
+            ...

napalm_device_types/ping_sweep.py

@@ -0,0 +1,172 @@
+# -*- coding: utf-8 -*-
+"""Generic ICMP sweep on top of the NAPALM-standard ``ping()``.
+
+Sweeping a range of addresses is orchestration, not device mechanics: the
+only vendor-specific part is how a single ``ping`` is executed, and NAPALM
+already standardises that. So the loop, the reply parsing, the target cap
+and the progress reporting live here once, and a concrete driver only has
+to implement ``ping()`` to become a usable sweep source.
+
+A driver whose device offers a *faster* sweep mechanism (a batch API, a
+single shell command that pings many hosts in parallel, an ARP-assisted
+scan) overrides :meth:`PingSweepMixin.ping_sweep` and keeps the same return
+shape — see ``napalm-opnsense`` for an example.
+
+The generic implementation is deliberately **sequential**: a NAPALM
+connection is a single session (SSH channel, HTTP client) and is not safe to
+drive from several threads at once. Callers that need many addresses covered
+quickly should either use a driver with its own parallel override or cap the
+target list (see ``PING_SWEEP_MAX_TARGETS``).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Callable, ClassVar, Dict, Iterable, List, Optional
+
+from napalm.base import NetworkDriver
+
+from napalm_device_types.models import PingSweepEntryDict, PingSweepResultDict
+
+
+def driver_supports_ping(driver_cls: type) -> bool:
+    """Whether *driver_cls* can actually execute ``ping()``.
+
+    True when the class provides its own ``ping`` implementation instead of
+    inheriting NAPALM's ``NotImplementedError`` stub. A driver that inherits
+    a working ``ping`` but cannot use it (unsupported firmware, disabled
+    service) opts out by setting ``SUPPORTS_PING = False``.
+    """
+    if getattr(driver_cls, "SUPPORTS_PING", None) is False:
+        return False
+    ping_impl = getattr(driver_cls, "ping", None)
+    if ping_impl is None:
+        return False
+    return ping_impl is not getattr(NetworkDriver, "ping", None)
+
+
+class PingSweepMixin:
+    """Adds :meth:`ping_sweep` to any driver that implements ``ping()``.
+
+    Mixed into :class:`~napalm_device_types.base.DeviceTypeDriver`, so every
+    device-type driver inherits it; drivers without a ``ping()`` of their own
+    simply report ``supports_ping() is False`` and raise from ``ping_sweep``.
+    """
+
+    #: Upper bound on destinations probed in one sweep. The sequential
+    #: default implementation costs roughly ``timeout`` seconds per silent
+    #: host, so an uncapped /24 would keep a device session busy for minutes.
+    #: Drivers with a parallel mechanism raise this.
+    PING_SWEEP_MAX_TARGETS: ClassVar[int] = 256
+
+    #: ``False`` opts a driver out of ping sweeps even though it implements
+    #: ``ping()``. ``None`` (the default) means "decide by introspection".
+    SUPPORTS_PING: ClassVar[Optional[bool]] = None
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def ping(
+            self,
+            destination: str,
+            source: str = "",
+            ttl: int = 255,
+            timeout: int = 2,
+            size: int = 100,
+            count: int = 5,
+            vrf: str = "",
+        ) -> Dict[str, Any]: ...
+
+    @classmethod
+    def supports_ping(cls) -> bool:
+        """Whether this driver class can be used as a ping-sweep source."""
+        return driver_supports_ping(cls)
+
+    def ping_sweep(
+        self,
+        destinations: Iterable[str],
+        *,
+        count: int = 1,
+        timeout: int = 1,
+        max_targets: Optional[int] = None,
+        on_progress: Optional[Callable[[int, int], None]] = None,
+        should_stop: Optional[Callable[[], bool]] = None,
+    ) -> PingSweepResultDict:
+        """Ping every address in *destinations* and report who answered.
+
+        :param destinations: IP addresses / hostnames to probe, in order.
+        :param count: probes per destination — 1 is enough for liveness.
+        :param timeout: seconds to wait for a reply per destination.
+        :param max_targets: cap for this call; defaults to
+            ``PING_SWEEP_MAX_TARGETS``. Excess destinations are dropped and
+            ``truncated`` is set in the result.
+        :param on_progress: called as ``(done, total)`` after each probe.
+        :param should_stop: polled before each probe; returning True ends the
+            sweep early (cancelled job, shutting-down worker).
+        :raises NotImplementedError: if the driver has no ``ping()``.
+        """
+        if not self.supports_ping():
+            raise NotImplementedError(
+                f"{type(self).__name__} does not implement ping(); cannot run a ping sweep"
+            )
+
+        limit = self.PING_SWEEP_MAX_TARGETS if max_targets is None else max_targets
+        targets = list(destinations)
+        truncated = len(targets) > limit
+        if truncated:
+            targets = targets[:limit]
+
+        total = len(targets)
+        entries: List[PingSweepEntryDict] = []
+        for done, destination in enumerate(targets, start=1):
+            if should_stop is not None and should_stop():
+                break
+            entries.append(self._ping_once(destination, count=count, timeout=timeout))
+            if on_progress is not None:
+                on_progress(done, total)
+
+        return {
+            "entries": entries,
+            "scanned": len(entries),
+            "alive_count": sum(1 for entry in entries if entry["alive"]),
+            "truncated": truncated,
+        }
+
+    # ── internals ────────────────────────────────────────────────────────────
+
+    def _ping_once(self, destination: str, *, count: int, timeout: int) -> PingSweepEntryDict:
+        """One probe, never raising — a dead session must not abort the sweep."""
+        try:
+            reply = self.ping(destination, count=count, timeout=timeout)
+        except Exception as exc:  # noqa: BLE001 - any driver error is just "no answer"
+            return {"ip": destination, "alive": False, "rtt_ms": None, "error": str(exc)}
+        return self._parse_ping_reply(destination, reply)
+
+    @staticmethod
+    def _parse_ping_reply(destination: str, reply: Any) -> PingSweepEntryDict:
+        """Map a NAPALM ``ping()`` reply onto a sweep entry."""
+        if not isinstance(reply, dict) or "success" not in reply:
+            error = "malformed ping reply"
+            if isinstance(reply, dict) and reply.get("error"):
+                error = str(reply["error"])
+            return {"ip": destination, "alive": False, "rtt_ms": None, "error": error}
+
+        success = reply.get("success") or {}
+        probes_sent = _as_int(success.get("probes_sent"))
+        packet_loss = _as_int(success.get("packet_loss"), default=probes_sent)
+        alive = bool(success.get("results")) or probes_sent > packet_loss
+        if not alive:
+            return {"ip": destination, "alive": False, "rtt_ms": None}
+        return {"ip": destination, "alive": True, "rtt_ms": _as_float(success.get("rtt_avg"))}
+
+
+def _as_int(value: Any, default: int = 0) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _as_float(value: Any) -> Optional[float]:
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None

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::
 
@@ -18,24 +19,21 @@ Usage::
             ...
 """
 
-from typing import Dict, List
+from typing import Dict, List, TYPE_CHECKING
 from napalm_device_types.base import DeviceTypeDriver
-from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.nat_vpn import NatVpnMixin
+from napalm_device_types.health_metrics import HealthMetricsMixin
+from napalm_device_types.dhcp import DhcpServerMixin
 from napalm_device_types.models import (
-    HealthMetricsDict,
     HostDict,
-    NATTranslationDict,
-    PortForwardDict,
     RadioStatusDict,
     SSIDDict,
-    VPNTunnelDict,
     WANStatusDict,
     WirelessClientDict,
 )
 
 
-class ResidentialGatewayDriver(DeviceTypeDriver):
-    TYPE_LABEL: str = "Gateway"
+class ResidentialGatewayDriver(NatVpnMixin, HealthMetricsMixin, DhcpServerMixin, DeviceTypeDriver):
     """
     Abstract intermediate driver for residential gateways (router + firewall + AP).
 
@@ -44,154 +42,103 @@ class ResidentialGatewayDriver(DeviceTypeDriver):
     drivers must implement.
     """
 
-    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
-    _SNMP_TX_ERR_IS_DROP: bool = False
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "residential_gateway"
+    TYPE_LABEL: str = "Gateway"
 
-    @classmethod
-    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
-        return await collect_ucd_metrics(
-            snmp_get, snmp_walk,
-            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
-            if_skip=cls._SNMP_SKIP_IF,
-        )
 
-    def get_wan_status(self) -> WANStatusDict:
-        """
-        Returns the status of the device's internet (WAN) uplink.
 
-        Contains:
+    if TYPE_CHECKING:
 
-        * connection_type (string) - e.g. ``"DSL"``, ``"Cable"``, ``"PPPoE"``, ``"DHCP"``
-        * is_connected (bool) - whether the WAN connection is currently established
-        * external_ip (string) - the public IPv4 address assigned to the WAN interface
-        * uptime (int) - seconds since the WAN connection was last (re-)established
-        * bytes_sent (int) - total bytes transmitted on the WAN interface
-        * bytes_received (int) - total bytes received on the WAN interface
-        * max_bitrate_up (int) - upstream sync rate in kbit/s
-        * max_bitrate_down (int) - downstream sync rate in kbit/s
-        * external_ipv6 (string, optional) - the public IPv6 address, if any
-        * link_status (string, optional) - physical line state, e.g. ``"Up"`` / ``"Down"``
+        def get_wan_status(self) -> WANStatusDict:
+            """
+            Returns the status of the device's internet (WAN) uplink.
 
-        Example::
+            Contains:
 
-            {
-                "connection_type": "DSL",
-                "is_connected": True,
-                "external_ip": "203.0.113.7",
-                "uptime": 345600,
-                "bytes_sent": 1234567890,
-                "bytes_received": 9876543210,
-                "max_bitrate_up": 40000,
-                "max_bitrate_down": 250000,
-                "link_status": "Up",
-            }
-        """
-        raise NotImplementedError
+            * connection_type (string) - e.g. ``"DSL"``, ``"Cable"``, ``"PPPoE"``, ``"DHCP"``
+            * is_connected (bool) - whether the WAN connection is currently established
+            * external_ip (string) - the public IPv4 address assigned to the WAN interface
+            * uptime (int) - seconds since the WAN connection was last (re-)established
+            * bytes_sent (int) - total bytes transmitted on the WAN interface
+            * bytes_received (int) - total bytes received on the WAN interface
+            * max_bitrate_up (int) - upstream sync rate in kbit/s
+            * max_bitrate_down (int) - downstream sync rate in kbit/s
+            * external_ipv6 (string, optional) - the public IPv6 address, if any
+            * link_status (string, optional) - physical line state, e.g. ``"Up"`` / ``"Down"``
 
-    def get_port_forwards(self) -> List[PortForwardDict]:
-        """
-        Returns the configured port forwarding (port mapping) rules.
+            Example::
 
-        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,
+                    "connection_type": "DSL",
+                    "is_connected": True,
+                    "external_ip": "203.0.113.7",
+                    "uptime": 345600,
+                    "bytes_sent": 1234567890,
+                    "bytes_received": 9876543210,
+                    "max_bitrate_up": 40000,
+                    "max_bitrate_down": 250000,
+                    "link_status": "Up",
                 }
-            ]
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_hosts(self) -> List[HostDict]:
-        """
-        Returns the list of hosts known to the gateway (LAN clients).
+        def get_hosts(self) -> List[HostDict]:
+            """
+            Returns the list of hosts known to the gateway (LAN clients).
 
-        Each entry contains:
+            Each entry contains:
 
-        * mac (string) - the host's MAC address
-        * ip (string) - the host's current IP address
-        * hostname (string) - the host's reported hostname (empty if unknown)
-        * interface_type (string) - how the host is connected, e.g. ``"LAN"``, ``"WLAN"``
-        * is_active (bool) - whether the host is currently online
-        * lease_time_remaining (int, optional) - remaining DHCP lease time in seconds
+            * mac (string) - the host's MAC address
+            * ip (string) - the host's current IP address
+            * hostname (string) - the host's reported hostname (empty if unknown)
+            * interface_type (string) - how the host is connected, e.g. ``"LAN"``, ``"WLAN"``
+            * is_active (bool) - whether the host is currently online
+            * lease_time_remaining (int, optional) - remaining DHCP lease time in seconds
 
-        Example::
+            Example::
 
-            [
-                {
-                    "mac": "AA:BB:CC:DD:EE:FF",
-                    "ip": "192.168.1.42",
-                    "hostname": "laptop",
-                    "interface_type": "WLAN",
-                    "is_active": True,
-                    "lease_time_remaining": 3600,
-                }
-            ]
-        """
-        raise NotImplementedError
+                [
+                    {
+                        "mac": "AA:BB:CC:DD:EE:FF",
+                        "ip": "192.168.1.42",
+                        "hostname": "laptop",
+                        "interface_type": "WLAN",
+                        "is_active": True,
+                        "lease_time_remaining": 3600,
+                    }
+                ]
+            """
+            ...
 
-    def get_nat_translations(self) -> List[NATTranslationDict]:
-        """
-        Returns a list of active NAT translation entries.
 
-        See :meth:`napalm_device_types.firewall.FirewallDriver.get_nat_translations`
-        for the entry format. Residential gateways typically derive this from
-        the active port-forwarding/NAT-PT table rather than a live connection
-        tracker; drivers that cannot provide this should return an empty list.
-        """
-        raise NotImplementedError
 
-    def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]:
-        """
-        Returns the status of VPN tunnels (e.g. WireGuard road-warrior
-        access, IPsec site-to-site).
+        def get_wireless_clients(self) -> List[WirelessClientDict]:
+            """
+            Returns the list of wireless clients currently associated with the
+            device's built-in access point(s).
 
-        See :meth:`napalm_device_types.firewall.FirewallDriver.get_vpn_tunnels`
-        for the entry format. Drivers that cannot provide this should return
-        an empty dict.
-        """
-        raise NotImplementedError
+            See :meth:`napalm_device_types.access_point.AccessPointDriver.get_wireless_clients`
+            for the entry format.
+            """
+            ...
 
-    def get_wireless_clients(self) -> List[WirelessClientDict]:
-        """
-        Returns the list of wireless clients currently associated with the
-        device's built-in access point(s).
+        def get_ssids(self) -> Dict[str, SSIDDict]:
+            """
+            Returns the configured wireless networks (SSIDs).
 
-        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_wireless_clients`
-        for the entry format.
-        """
-        raise NotImplementedError
+            See :meth:`napalm_device_types.access_point.AccessPointDriver.get_ssids`
+            for the entry format.
+            """
+            ...
 
-    def get_ssids(self) -> Dict[str, SSIDDict]:
-        """
-        Returns the configured wireless networks (SSIDs).
+        def get_radio_status(self) -> Dict[str, RadioStatusDict]:
+            """
+            Returns the status of the device's wireless radios.
 
-        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_ssids`
-        for the entry format.
-        """
-        raise NotImplementedError
-
-    def get_radio_status(self) -> Dict[str, RadioStatusDict]:
-        """
-        Returns the status of the device's wireless radios.
-
-        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_radio_status`
-        for the entry format.
-        """
-        raise NotImplementedError
+            See :meth:`napalm_device_types.access_point.AccessPointDriver.get_radio_status`
+            for the entry format.
+            """
+            ...

napalm_device_types/roles.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+"""Which roles a driver class fills, and which of them is primary.
+
+A *role* is what a device is: a firewall, a switch, a NAS. Real devices are
+often several at once -- a QNAP is storage, hypervisor and Linux host -- so a
+driver declares every role it fills by inheriting the matching base, and the
+**order of those bases is the ranking**::
+
+    class QnapQtsDriver(StorageDriver, HypervisorDriver, LinuxDriver):
+        ...        # primary_role_of(...) == "storage"
+
+That replaces the older arrangement, where an ``issubclass`` chain in netOrk
+picked a single winner by a fixed precedence and a ``DEVICE_CLASS`` attribute
+existed to overrule it. The author already states the ranking in the class
+definition; nothing needs to restate it.
+
+Whether a driver can perform a *specific operation* is a separate question with
+a separate answer: ``hasattr``. Role bases declare their methods under
+``if TYPE_CHECKING`` and implement nothing, so a method exists at runtime only
+when a concrete driver provided it.
+"""
+
+from __future__ import annotations
+
+from typing import Any, List, Optional, Type
+
+
+def _is_role_base(cls: type) -> bool:
+    """True for a class that introduces a role, not one that merely inherits one.
+
+    Keyed on ``ROLE`` appearing in the class's own ``__dict__``: a concrete
+    driver inherits the attribute but does not define it, and must not be
+    mistaken for a role base of its own.
+    """
+    role = vars(cls).get("ROLE")
+    return isinstance(role, str) and bool(role)
+
+
+def roles_of(driver_cls: type) -> List[Type[Any]]:
+    """Every role base in ``driver_cls``'s MRO, most significant first.
+
+    MRO order is inheritance order, so the list reflects exactly what the driver
+    author wrote in the class definition.
+    """
+    return [cls for cls in driver_cls.__mro__ if _is_role_base(cls)]
+
+
+def role_keys_of(driver_cls: type) -> List[str]:
+    """The stable string keys of :func:`roles_of`, e.g. ``["storage", "linux"]``."""
+    return [vars(cls)["ROLE"] for cls in roles_of(driver_cls)]
+
+
+def primary_role_of(driver_cls: type) -> Optional[str]:
+    """The driver's headline role, or ``None`` when it declares no role at all.
+
+    This is the value netOrk exposes as ``device_class``.
+    """
+    keys = role_keys_of(driver_cls)
+    return keys[0] if keys else None

napalm_device_types/services.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+"""Init-system services: listing them and starting/stopping them.
+
+Deliberately *not* the same thing as a NAS's exported shares, which live on
+``StorageServiceCapability`` as ``get_storage_services``. Those two used to
+share the name ``get_services`` with incompatible return types, which is why
+a QNAP -- storage and OS at once -- could not satisfy both.
+
+Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
+Nothing exists at runtime until a concrete driver implements it, so mixing
+this class in can never shadow a working implementation from a sibling base.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, TYPE_CHECKING
+
+from napalm_device_types.models import ServiceDict
+
+
+class ServiceControlMixin:
+    if TYPE_CHECKING:
+
+        def get_services(self) -> List[ServiceDict]:
+            """
+            Returns the list of system services known to the device's init system.
+
+            Each entry contains:
+
+            * name (string)    - service name as registered with the init system
+            * running (bool)   - ``True`` if the service process is currently running
+            * enabled (bool)   - ``True`` if the service starts automatically at boot
+            * pid (int)        - process ID of the main service process; 0 if not running
+
+            Example::
+
+                [
+                    {"name": "lldpd",   "running": True,  "enabled": True,  "pid": 2341},
+                    {"name": "sshd",    "running": True,  "enabled": True,  "pid": 1198},
+                    {"name": "cron",    "running": False, "enabled": False, "pid": 0},
+                ]
+            """
+            ...
+
+        def manage_service(self, name: str, action: str) -> Dict[str, Any]:
+            """
+            Execute a lifecycle action on a named service.
+
+            :param name: Service name as returned by :meth:`get_services`.
+            :param action: One of ``start``, ``stop``, ``restart``, ``enable``, ``disable``.
+            :returns: ``{"success": bool, "output": str}``
+            :raises ValueError: If ``name`` or ``action`` is invalid.
+            :raises NotImplementedError: If the driver does not support service management.
+            """
+            ...

napalm_device_types/switch.py

@@ -10,13 +10,13 @@ Usage::
             ...
 """
 
-from typing import Dict, List
+from typing import Any, Awaitable, Callable, Dict, List, TYPE_CHECKING
 from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types.mac_acl import MacAclMixin
 from napalm_device_types.models import (
     Dot1XPortDict,
     HealthMetricsDict,
     InterfaceConfigDict,
-    MACACLDict,
     PoESummaryDict,
     PortChannelDict,
     SpanningTreeDict,
@@ -24,8 +24,7 @@ from napalm_device_types.models import (
 )
 
 
-class SwitchDriver(DeviceTypeDriver):
-    TYPE_LABEL: str = "Switch"
+class SwitchDriver(MacAclMixin, DeviceTypeDriver):
     """
     Abstract intermediate driver for Ethernet switches.
 
@@ -34,403 +33,378 @@ class SwitchDriver(DeviceTypeDriver):
     operations that concrete drivers must implement.
     """
 
-    @classmethod
-    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
-        """Collect SNMP health metrics for this switch type.
+    #: Stable key netOrk exposes as ``device_class``. The order in which a
+    #: driver lists its role bases is the ranking; see
+    #: :func:`napalm_device_types.roles.primary_role_of`.
+    ROLE: str = "switch"
+    TYPE_LABEL: str = "Switch"
 
-        Switch vendors use proprietary OIDs — each concrete driver must
-        override this classmethod.
-        """
-        raise NotImplementedError
+    if TYPE_CHECKING:
 
-    def get_spanning_tree(self) -> Dict[str, SpanningTreeDict]:
-        """
-        Returns spanning tree status for each STP instance.
+        @classmethod
+        async def get_health_metrics(
+            cls,
+            snmp_get: Callable[..., Awaitable[Any]],
+            snmp_walk: Callable[..., Awaitable[Dict[str, Any]]],
+        ) -> HealthMetricsDict:
+            """Collect SNMP health metrics for this switch type.
 
-        Keys are STP instance identifiers (e.g. VLAN IDs for PVST,
-        ``"MST0"`` for MSTP, or ``"0"`` for a single instance).
-        Each value contains:
+            Switch vendors use proprietary OIDs — each concrete driver must
+            override this classmethod.
+            """
+            ...
 
-        * mode (string) - STP variant: ``"STP"``, ``"RSTP"``, ``"MSTP"``, ``"PVST"``
-        * root_bridge (bool) - whether this device is the root bridge
-        * root_id (string) - root bridge MAC address
-        * root_priority (int) - root bridge priority
-        * bridge_id (string) - this bridge's MAC address
-        * bridge_priority (int) - this bridge's priority
-        * interfaces (dict) - per-interface STP state:
+        def get_spanning_tree(self) -> Dict[str, SpanningTreeDict]:
+            """
+            Returns spanning tree status for each STP instance.
 
-            * role (string) - ``"root"``, ``"designated"``, ``"alternate"``, ``"backup"``
-            * state (string) - ``"forwarding"``, ``"blocking"``, ``"learning"``, ``"listening"``
-            * cost (int) - port path cost
-            * port_priority (int) - port priority
+            Keys are STP instance identifiers (e.g. VLAN IDs for PVST,
+            ``"MST0"`` for MSTP, or ``"0"`` for a single instance).
+            Each value contains:
 
-        Example::
+            * mode (string) - STP variant: ``"STP"``, ``"RSTP"``, ``"MSTP"``, ``"PVST"``
+            * root_bridge (bool) - whether this device is the root bridge
+            * root_id (string) - root bridge MAC address
+            * root_priority (int) - root bridge priority
+            * bridge_id (string) - this bridge's MAC address
+            * bridge_priority (int) - this bridge's priority
+            * interfaces (dict) - per-interface STP state:
 
-            {
-                "1": {
-                    "mode": "RSTP",
-                    "root_bridge": False,
-                    "root_id": "00:11:22:33:44:55",
-                    "root_priority": 4096,
-                    "bridge_id": "AA:BB:CC:DD:EE:FF",
-                    "bridge_priority": 32768,
-                    "interfaces": {
-                        "GigabitEthernet0/1": {
-                            "role": "root",
-                            "state": "forwarding",
-                            "cost": 4,
-                            "port_priority": 128,
+                * role (string) - ``"root"``, ``"designated"``, ``"alternate"``, ``"backup"``
+                * state (string) - ``"forwarding"``, ``"blocking"``, ``"learning"``, ``"listening"``
+                * cost (int) - port path cost
+                * port_priority (int) - port priority
+
+            Example::
+
+                {
+                    "1": {
+                        "mode": "RSTP",
+                        "root_bridge": False,
+                        "root_id": "00:11:22:33:44:55",
+                        "root_priority": 4096,
+                        "bridge_id": "AA:BB:CC:DD:EE:FF",
+                        "bridge_priority": 32768,
+                        "interfaces": {
+                            "GigabitEthernet0/1": {
+                                "role": "root",
+                                "state": "forwarding",
+                                "cost": 4,
+                                "port_priority": 128,
+                            },
+                            "GigabitEthernet0/2": {
+                                "role": "designated",
+                                "state": "forwarding",
+                                "cost": 4,
+                                "port_priority": 128,
+                            },
                         },
-                        "GigabitEthernet0/2": {
-                            "role": "designated",
-                            "state": "forwarding",
-                            "cost": 4,
-                            "port_priority": 128,
-                        },
-                    },
+                    }
                 }
-            }
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_port_channels(self) -> Dict[str, PortChannelDict]:
-        """
-        Returns port-channel (LAG) configuration and status.
+        def get_port_channels(self) -> Dict[str, PortChannelDict]:
+            """
+            Returns port-channel (LAG) configuration and status.
 
-        Keys are port-channel interface names (e.g. ``"Port-Channel1"``).
-        Each value contains:
+            Keys are port-channel interface names (e.g. ``"Port-Channel1"``).
+            Each value contains:
 
-        * members (list of strings) - names of member interfaces
-        * protocol (string) - aggregation protocol: ``"LACP"``, ``"PAgP"``, ``"static"``
-        * min_links (int) - minimum number of active members required
-        * is_up (bool) - whether the LAG is operationally up
+            * members (list of strings) - names of member interfaces
+            * protocol (string) - aggregation protocol: ``"LACP"``, ``"PAgP"``, ``"static"``
+            * min_links (int) - minimum number of active members required
+            * is_up (bool) - whether the LAG is operationally up
 
-        Example::
+            Example::
 
-            {
-                "Port-Channel1": {
-                    "members": ["GigabitEthernet0/1", "GigabitEthernet0/2"],
-                    "protocol": "LACP",
-                    "min_links": 1,
-                    "is_up": True,
+                {
+                    "Port-Channel1": {
+                        "members": ["GigabitEthernet0/1", "GigabitEthernet0/2"],
+                        "protocol": "LACP",
+                        "min_links": 1,
+                        "is_up": True,
+                    }
                 }
-            }
-        """
-        raise NotImplementedError
+            """
+            ...
 
-    def get_mac_acl(self) -> Dict[str, MACACLDict]:
-        """
-        Returns the MAC-address-based access control lists configured per port.
 
-        Keys are interface names.  Each value contains:
+        def get_dot1x_ports(self) -> Dict[str, Dot1XPortDict]:
+            """
+            Returns the 802.1X / NAC configuration per switch port.
 
-        * name (string) - interface name (repeated for convenience)
-        * policy (string) - ACL mode:
+            Keys are interface names.  Each value contains:
 
-          * ``"allow"`` – whitelist: only listed MACs may use this port
-          * ``"deny"``  – blacklist: listed MACs are blocked
-          * ``"disabled"`` – no MAC filtering active
+            * enabled (bool) - whether 802.1X is active on this port
+            * port_control (string) - authentication mode:
 
-        * entries (list) - ACL entries, each with:
+              * ``"auto"`` – port authenticates normally
+              * ``"force-authorized"`` – port always passes traffic (bypass)
+              * ``"force-unauthorized"`` – port always blocks traffic
 
-          * mac (string) - MAC address (normalised, colon-separated)
-          * action (string) - ``"allow"`` or ``"deny"``
-          * description (string) - optional human-readable label
+            * host_mode (string) - how many identities are authenticated per port:
 
-        Example::
+              * ``"single-host"`` – one device, then port is locked
+              * ``"multi-host"`` – first auth unlocks port for all devices
+              * ``"multi-domain"`` – one data + one voice device (IP phone scenario)
+              * ``"multi-auth"`` – each device authenticates individually
 
-            {
-                "GigabitEthernet0/1": {
-                    "name": "GigabitEthernet0/1",
-                    "policy": "allow",
-                    "entries": [
-                        {"mac": "AA:BB:CC:DD:EE:01", "action": "allow", "description": "printer"},
-                    ],
-                },
-                "GigabitEthernet0/2": {
-                    "name": "GigabitEthernet0/2",
-                    "policy": "disabled",
-                    "entries": [],
-                },
-            }
-        """
-        raise NotImplementedError
+            * auth_server (dict) - RADIUS authentication server (host, port, timeout, retries)
+            * acct_server (dict or None) - RADIUS accounting server, ``None`` if unused
+            * reauthentication (bool) - whether periodic re-authentication is enabled
+            * reauth_interval (int) - re-authentication interval in seconds (0 = disabled)
+            * guest_vlan (int) - VLAN ID for unauthenticated clients (0 = disabled)
+            * auth_fail_vlan (int) - VLAN ID for clients that fail authentication (0 = disabled)
 
-    def get_dot1x_config(self) -> Dict[str, Dot1XPortDict]:
-        """
-        Returns the 802.1X / NAC configuration per switch port.
+            Note: RADIUS shared secrets are intentionally omitted.
 
-        Keys are interface names.  Each value contains:
+            Example::
 
-        * enabled (bool) - whether 802.1X is active on this port
-        * port_control (string) - authentication mode:
-
-          * ``"auto"`` – port authenticates normally
-          * ``"force-authorized"`` – port always passes traffic (bypass)
-          * ``"force-unauthorized"`` – port always blocks traffic
-
-        * host_mode (string) - how many identities are authenticated per port:
-
-          * ``"single-host"`` – one device, then port is locked
-          * ``"multi-host"`` – first auth unlocks port for all devices
-          * ``"multi-domain"`` – one data + one voice device (IP phone scenario)
-          * ``"multi-auth"`` – each device authenticates individually
-
-        * auth_server (dict) - RADIUS authentication server (host, port, timeout, retries)
-        * acct_server (dict or None) - RADIUS accounting server, ``None`` if unused
-        * reauthentication (bool) - whether periodic re-authentication is enabled
-        * reauth_interval (int) - re-authentication interval in seconds (0 = disabled)
-        * guest_vlan (int) - VLAN ID for unauthenticated clients (0 = disabled)
-        * auth_fail_vlan (int) - VLAN ID for clients that fail authentication (0 = disabled)
-
-        Note: RADIUS shared secrets are intentionally omitted.
-
-        Example::
-
-            {
-                "GigabitEthernet0/1": {
-                    "enabled": True,
-                    "port_control": "auto",
-                    "host_mode": "multi-domain",
-                    "auth_server": {
-                        "host": "radius.corp.example",
-                        "port": 1812,
-                        "timeout": 5,
-                        "retries": 3,
-                    },
-                    "acct_server": None,
-                    "reauthentication": True,
-                    "reauth_interval": 3600,
-                    "guest_vlan": 99,
-                    "auth_fail_vlan": 999,
-                },
-            }
-        """
-        raise NotImplementedError
-
-    def get_poe_status(self) -> PoESummaryDict:
-        """
-        Returns the PoE status of the switch as a whole and per port.
-
-        The returned dictionary contains:
-
-        * total_power_budget (float) - total PoE power available in watts
-        * total_power_draw (float) - total PoE power currently consumed in watts
-        * ports (dict) - per-interface PoE state, keyed by interface name:
-
-          * enabled (bool) - whether PoE is configured on this port
-          * status (string) - operational state:
-
-            * ``"delivering"`` – power is being delivered to a PD
-            * ``"searching"``  – port is looking for a powered device
-            * ``"fault"``      – an error condition was detected
-            * ``"disabled"``   – PoE is administratively off
-            * ``"denied"``     – PD detected but power budget exceeded
-
-          * poe_class (string) - IEEE 802.3 class: ``"Class 0"`` … ``"Class 8"``
-            (``"unknown"`` if not yet negotiated)
-          * power_draw (float) - current power consumption in watts
-          * power_budget (float) - per-port power limit in watts
-          * voltage (float) - measured port voltage in volts
-          * current (float) - measured port current in milliamps
-
-        Example::
-
-            {
-                "total_power_budget": 740.0,
-                "total_power_draw": 43.2,
-                "ports": {
+                {
                     "GigabitEthernet0/1": {
                         "enabled": True,
-                        "status": "delivering",
-                        "poe_class": "Class 3",
-                        "power_draw": 12.4,
-                        "power_budget": 30.0,
-                        "voltage": 53.5,
-                        "current": 231.0,
+                        "port_control": "auto",
+                        "host_mode": "multi-domain",
+                        "auth_server": {
+                            "host": "radius.corp.example",
+                            "port": 1812,
+                            "timeout": 5,
+                            "retries": 3,
+                        },
+                        "acct_server": None,
+                        "reauthentication": True,
+                        "reauth_interval": 3600,
+                        "guest_vlan": 99,
+                        "auth_fail_vlan": 999,
                     },
-                    "GigabitEthernet0/2": {
+                }
+            """
+            ...
+
+        def get_poe_status(self) -> PoESummaryDict:
+            """
+            Returns the PoE status of the switch as a whole and per port.
+
+            The returned dictionary contains:
+
+            * total_power_budget (float) - total PoE power available in watts
+            * total_power_draw (float) - total PoE power currently consumed in watts
+            * ports (dict) - per-interface PoE state, keyed by interface name:
+
+              * enabled (bool) - whether PoE is configured on this port
+              * status (string) - operational state:
+
+                * ``"delivering"`` – power is being delivered to a PD
+                * ``"searching"``  – port is looking for a powered device
+                * ``"fault"``      – an error condition was detected
+                * ``"disabled"``   – PoE is administratively off
+                * ``"denied"``     – PD detected but power budget exceeded
+
+              * poe_class (string) - IEEE 802.3 class: ``"Class 0"`` … ``"Class 8"``
+                (``"unknown"`` if not yet negotiated)
+              * power_draw (float) - current power consumption in watts
+              * power_budget (float) - per-port power limit in watts
+              * voltage (float) - measured port voltage in volts
+              * current (float) - measured port current in milliamps
+
+            Example::
+
+                {
+                    "total_power_budget": 740.0,
+                    "total_power_draw": 43.2,
+                    "ports": {
+                        "GigabitEthernet0/1": {
+                            "enabled": True,
+                            "status": "delivering",
+                            "poe_class": "Class 3",
+                            "power_draw": 12.4,
+                            "power_budget": 30.0,
+                            "voltage": 53.5,
+                            "current": 231.0,
+                        },
+                        "GigabitEthernet0/2": {
+                            "enabled": True,
+                            "status": "searching",
+                            "poe_class": "unknown",
+                            "power_draw": 0.0,
+                            "power_budget": 30.0,
+                            "voltage": 0.0,
+                            "current": 0.0,
+                        },
+                        "GigabitEthernet0/3": {
+                            "enabled": False,
+                            "status": "disabled",
+                            "poe_class": "unknown",
+                            "power_draw": 0.0,
+                            "power_budget": 0.0,
+                            "voltage": 0.0,
+                            "current": 0.0,
+                        },
+                    },
+                }
+            """
+            ...
+
+        def set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None:
+            """
+            Creates or updates a VLAN on the switch.
+
+            If the VLAN does not yet exist it is created first.  Only the keys
+            present in *config* are applied; omitted keys leave the existing VLAN
+            configuration untouched.
+
+            :param vlan_id: VLAN ID (1–4094).
+            :param config: A (partial) :class:`~napalm_device_types.models.VlanConfigDict`
+                containing the fields to set.  Supported keys:
+
+                * ``name`` (str) – human-readable VLAN name
+                * ``active`` (bool) – ``True`` = active, ``False`` = suspended
+                * ``interfaces`` (list of str) – access-port names to assign to
+                  this VLAN (replaces the current membership list)
+
+            :raises NotImplementedError: If the driver does not implement this method.
+            :raises ValueError: If *vlan_id* is out of range or a field value is invalid.
+
+            Example – create VLAN 10 with a name::
+
+                driver.set_vlan(10, {"name": "Workstations", "active": True})
+
+            Example – assign ports to an existing VLAN::
+
+                driver.set_vlan(
+                    10,
+                    {"interfaces": ["GigabitEthernet0/1", "GigabitEthernet0/2"]},
+                )
+            """
+            ...
+
+        def delete_vlan(self, vlan_id: int) -> None:
+            """
+            Removes a VLAN from the switch.
+
+            All ports that were assigned to this VLAN as their access VLAN are
+            moved to the default VLAN (1) by the driver before deletion.  Trunk
+            ports that carry this VLAN will have it removed from their allowed
+            VLAN list.
+
+            :param vlan_id: VLAN ID (1–4094) to delete.
+            :raises NotImplementedError: If the driver does not implement this method.
+            :raises ValueError: If *vlan_id* is out of range or the VLAN does not
+                exist on the device.
+
+            Example::
+
+                driver.delete_vlan(10)
+            """
+            ...
+
+        def set_interface(self, interface: str, config: InterfaceConfigDict) -> None:
+            """
+            Applies configuration to a single switch interface.
+
+            Only the keys present in *config* are changed; omitted keys leave the
+            current device configuration untouched.
+
+            :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
+            :param config: A (partial) :class:`~napalm_device_types.models.InterfaceConfigDict`
+                containing the fields to update.  Supported keys:
+
+                * ``description`` (str) – human-readable port label
+                * ``enabled`` (bool) – administrative state
+                * ``speed`` (int) – link speed in Mbps; ``0`` = auto-negotiate
+                * ``duplex`` (str) – ``"full"``, ``"half"``, or ``"auto"``
+                * ``mtu`` (int) – maximum transmission unit in bytes
+                * ``mode`` (str) – ``"access"``, ``"trunk"``, or ``"routed"``
+                * ``access_vlan`` (int) – untagged VLAN; relevant when mode is ``"access"``
+                * ``voice_vlan`` (int) – voice VLAN ID (``0`` = disabled)
+                * ``trunk_vlans`` (list of int) – tagged VLANs; empty = allow all
+                * ``native_vlan`` (int) – native VLAN on trunk ports
+
+            :raises NotImplementedError: If the driver does not implement this method.
+            :raises ValueError: If *interface* does not exist or an invalid value is
+                supplied for a configuration field.
+
+            Example – convert port to access VLAN 10 and add a description::
+
+                driver.set_interface(
+                    "GigabitEthernet0/1",
+                    {
+                        "description": "Workstation port",
                         "enabled": True,
-                        "status": "searching",
-                        "poe_class": "unknown",
-                        "power_draw": 0.0,
-                        "power_budget": 30.0,
-                        "voltage": 0.0,
-                        "current": 0.0,
+                        "mode": "access",
+                        "access_vlan": 10,
                     },
-                    "GigabitEthernet0/3": {
-                        "enabled": False,
-                        "status": "disabled",
-                        "poe_class": "unknown",
-                        "power_draw": 0.0,
-                        "power_budget": 0.0,
-                        "voltage": 0.0,
-                        "current": 0.0,
+                )
+
+            Example – configure a trunk port::
+
+                driver.set_interface(
+                    "GigabitEthernet0/2",
+                    {
+                        "mode": "trunk",
+                        "trunk_vlans": [10, 20, 30],
+                        "native_vlan": 1,
                     },
-                },
-            }
-        """
-        raise NotImplementedError
+                )
+            """
+            ...
 
-    def set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None:
-        """
-        Creates or updates a VLAN on the switch.
+        def set_lag_members(self, lag_name: str, members: List[str]) -> None:
+            """
+            Sets the full member-port list of a LAG/trunk interface.
 
-        If the VLAN does not yet exist it is created first.  Only the keys
-        present in *config* are applied; omitted keys leave the existing VLAN
-        configuration untouched.
+            Diffs *members* against the LAG's current members (as reported by
+            ``get_interfaces()``'s ``lag_members`` field) and adds/removes ports
+            on the device to match. The LAG itself must already exist on the
+            device; this method only manages its membership.
 
-        :param vlan_id: VLAN ID (1–4094).
-        :param config: A (partial) :class:`~napalm_device_types.models.VlanConfigDict`
-            containing the fields to set.  Supported keys:
+            :param lag_name: Name of the LAG/trunk interface as returned by
+                ``get_interfaces()`` (e.g. ``"Trk1"``, ``"ch1"``, ``"Lag1"``).
+            :param members: Full desired list of member port names.
+            :raises NotImplementedError: If the driver does not implement this method.
+            :raises ValueError: If *lag_name* does not refer to a valid LAG.
 
-            * ``name`` (str) – human-readable VLAN name
-            * ``active`` (bool) – ``True`` = active, ``False`` = suspended
-            * ``interfaces`` (list of str) – access-port names to assign to
-              this VLAN (replaces the current membership list)
+            Example::
 
-        :raises NotImplementedError: If the driver does not implement this method.
-        :raises ValueError: If *vlan_id* is out of range or a field value is invalid.
+                driver.set_lag_members("Lag1", ["GigabitEthernet0/1", "GigabitEthernet0/2"])
+            """
+            ...
 
-        Example – create VLAN 10 with a name::
+        def set_poe_enabled(self, interface: str, enabled: bool) -> None:
+            """
+            Administratively enables or disables PoE on a single port.
 
-            driver.set_vlan(10, {"name": "Workstations", "active": True})
+            :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
+            :param enabled: ``True`` to enable PoE, ``False`` to disable it.
+            :raises NotImplementedError: If the driver does not support PoE control.
+            :raises ValueError: If the interface does not exist or does not support PoE.
 
-        Example – assign ports to an existing VLAN::
+            Example::
 
-            driver.set_vlan(
-                10,
-                {"interfaces": ["GigabitEthernet0/1", "GigabitEthernet0/2"]},
-            )
-        """
-        raise NotImplementedError
+                driver.set_poe_enabled("GigabitEthernet0/1", False)  # cut power
+                driver.set_poe_enabled("GigabitEthernet0/1", True)   # restore
+            """
+            ...
 
-    def delete_vlan(self, vlan_id: int) -> None:
-        """
-        Removes a VLAN from the switch.
+        def power_cycle_port(self, interface: str, delay: int = 5) -> None:
+            """
+            Power-cycles the PoE port: cuts power, waits ``delay`` seconds, then
+            restores power.  Useful for rebooting a hung IP camera, AP, or IP phone
+            without physical access.
 
-        All ports that were assigned to this VLAN as their access VLAN are
-        moved to the default VLAN (1) by the driver before deletion.  Trunk
-        ports that carry this VLAN will have it removed from their allowed
-        VLAN list.
+            The method blocks until the full cycle (off → wait → on) is complete.
+            After it returns the port is back in the delivering/searching state.
 
-        :param vlan_id: VLAN ID (1–4094) to delete.
-        :raises NotImplementedError: If the driver does not implement this method.
-        :raises ValueError: If *vlan_id* is out of range or the VLAN does not
-            exist on the device.
+            :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
+            :param delay: Seconds to keep the port powered off (default: 5).
+            :raises NotImplementedError: If the driver does not support PoE control.
+            :raises ValueError: If the interface does not exist, does not support PoE,
+                or PoE is administratively disabled on the port.
 
-        Example::
+            Example::
 
-            driver.delete_vlan(10)
-        """
-        raise NotImplementedError
-
-    def set_interface(self, interface: str, config: InterfaceConfigDict) -> None:
-        """
-        Applies configuration to a single switch interface.
-
-        Only the keys present in *config* are changed; omitted keys leave the
-        current device configuration untouched.
-
-        :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
-        :param config: A (partial) :class:`~napalm_device_types.models.InterfaceConfigDict`
-            containing the fields to update.  Supported keys:
-
-            * ``description`` (str) – human-readable port label
-            * ``enabled`` (bool) – administrative state
-            * ``speed`` (int) – link speed in Mbps; ``0`` = auto-negotiate
-            * ``duplex`` (str) – ``"full"``, ``"half"``, or ``"auto"``
-            * ``mtu`` (int) – maximum transmission unit in bytes
-            * ``mode`` (str) – ``"access"``, ``"trunk"``, or ``"routed"``
-            * ``access_vlan`` (int) – untagged VLAN; relevant when mode is ``"access"``
-            * ``voice_vlan`` (int) – voice VLAN ID (``0`` = disabled)
-            * ``trunk_vlans`` (list of int) – tagged VLANs; empty = allow all
-            * ``native_vlan`` (int) – native VLAN on trunk ports
-
-        :raises NotImplementedError: If the driver does not implement this method.
-        :raises ValueError: If *interface* does not exist or an invalid value is
-            supplied for a configuration field.
-
-        Example – convert port to access VLAN 10 and add a description::
-
-            driver.set_interface(
-                "GigabitEthernet0/1",
-                {
-                    "description": "Workstation port",
-                    "enabled": True,
-                    "mode": "access",
-                    "access_vlan": 10,
-                },
-            )
-
-        Example – configure a trunk port::
-
-            driver.set_interface(
-                "GigabitEthernet0/2",
-                {
-                    "mode": "trunk",
-                    "trunk_vlans": [10, 20, 30],
-                    "native_vlan": 1,
-                },
-            )
-        """
-        raise NotImplementedError
-
-    def set_lag_members(self, lag_name: str, members: List[str]) -> None:
-        """
-        Sets the full member-port list of a LAG/trunk interface.
-
-        Diffs *members* against the LAG's current members (as reported by
-        ``get_interfaces()``'s ``lag_members`` field) and adds/removes ports
-        on the device to match. The LAG itself must already exist on the
-        device; this method only manages its membership.
-
-        :param lag_name: Name of the LAG/trunk interface as returned by
-            ``get_interfaces()`` (e.g. ``"Trk1"``, ``"ch1"``, ``"Lag1"``).
-        :param members: Full desired list of member port names.
-        :raises NotImplementedError: If the driver does not implement this method.
-        :raises ValueError: If *lag_name* does not refer to a valid LAG.
-
-        Example::
-
-            driver.set_lag_members("Lag1", ["GigabitEthernet0/1", "GigabitEthernet0/2"])
-        """
-        raise NotImplementedError
-
-    def set_poe_enabled(self, interface: str, enabled: bool) -> None:
-        """
-        Administratively enables or disables PoE on a single port.
-
-        :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
-        :param enabled: ``True`` to enable PoE, ``False`` to disable it.
-        :raises NotImplementedError: If the driver does not support PoE control.
-        :raises ValueError: If the interface does not exist or does not support PoE.
-
-        Example::
-
-            driver.set_poe_enabled("GigabitEthernet0/1", False)  # cut power
-            driver.set_poe_enabled("GigabitEthernet0/1", True)   # restore
-        """
-        raise NotImplementedError
-
-    def power_cycle_port(self, interface: str, delay: int = 5) -> None:
-        """
-        Power-cycles the PoE port: cuts power, waits ``delay`` seconds, then
-        restores power.  Useful for rebooting a hung IP camera, AP, or IP phone
-        without physical access.
-
-        The method blocks until the full cycle (off → wait → on) is complete.
-        After it returns the port is back in the delivering/searching state.
-
-        :param interface: Interface name (e.g. ``"GigabitEthernet0/1"``).
-        :param delay: Seconds to keep the port powered off (default: 5).
-        :raises NotImplementedError: If the driver does not support PoE control.
-        :raises ValueError: If the interface does not exist, does not support PoE,
-            or PoE is administratively disabled on the port.
-
-        Example::
-
-            driver.power_cycle_port("GigabitEthernet0/1")        # 5 s off
-            driver.power_cycle_port("GigabitEthernet0/1", delay=15)  # 15 s off
-        """
-        raise NotImplementedError
+                driver.power_cycle_port("GigabitEthernet0/1")        # 5 s off
+                driver.power_cycle_port("GigabitEthernet0/1", delay=15)  # 15 s off
+            """
+            ...

napalm_device_types/updates.py

@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+"""Pending package updates and applying them.
+
+The two device types that offer this used to declare it under two different
+names -- ``get_available_updates`` and ``get_pending_updates`` -- with
+``napalm-linux`` carrying an alias between them so netOrk could call either.
+One name now, the one netOrk and four drivers already used.
+
+Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
+Nothing exists at runtime until a concrete driver implements it, so mixing
+this class in can never shadow a working implementation from a sibling base.
+"""
+
+from __future__ import annotations
+
+from typing import List, TYPE_CHECKING
+
+from napalm_device_types.models import ApplyUpdatesResultDict, UpdateDict
+
+
+class UpdateMixin:
+    if TYPE_CHECKING:
+
+        def get_available_updates(self) -> List[UpdateDict]:
+            """
+            Returns a list of packages that have a newer version available.
+
+            Each entry contains:
+
+            * name (string) - package name
+            * current_version (string) - currently installed version
+            * new_version (string) - version available in the repository
+
+            Example::
+
+                [
+                    {
+                        "name": "openssh-server",
+                        "current_version": "1:9.2p1-2+deb12u1",
+                        "new_version": "1:9.2p1-2+deb12u2",
+                    },
+                ]
+            """
+            ...
+
+        def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
+            """
+            Upgrades the given packages to the newest available version.
+
+            Only packages that are already installed may be upgraded; this method
+            does **not** install new packages.  Pass an empty list to upgrade
+            **all** packages that have pending updates.
+
+            :param packages: List of package names to upgrade.  Each name must
+                match ``^[a-zA-Z0-9_\\-\\+\\.]+$``; a :exc:`ValueError` is raised
+                for any name that does not conform.
+            :returns: A dict with:
+
+                * success (bool) – ``True`` if the package manager exited without error
+                * output (string) – combined stdout / stderr from the package manager
+                * error (string, optional) – short error message when *success* is ``False``
+
+            :raises ValueError: If any package name fails the safety check.
+
+            Example::
+
+                result = driver.apply_updates(["openssh-server", "curl"])
+                # → {"success": True, "output": "Reading package lists...\\n..."}
+
+                # Upgrade everything:
+                result = driver.apply_updates([])
+            """
+            ...

pyproject.toml

@@ -4,10 +4,10 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "0.5.0"
+version = "2.0.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 license = { text = "Apache-2.0" }
 authors = [
     { name = "Christian Manivong", email = "christian@manivong.de" },
@@ -31,7 +31,6 @@ classifiers = [
     "License :: OSI Approved :: Apache Software License",
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
@@ -61,6 +60,6 @@ where = ["."]
 include = ["napalm_device_types*"]
 
 [tool.mypy]
-python_version = "3.9"
+python_version = "3.10"
 strict = true
 ignore_missing_imports = true

tests/test_dhcp_diff_apply.py

@@ -0,0 +1,208 @@
+"""Tests for DhcpServerMixin's generic reservation diff/apply mechanism.
+
+diff_dhcp_reservations/apply_dhcp_reservationset are concrete methods on the
+mixin (not overridden by concrete drivers) — they only depend on the three
+abstract methods (get_dhcp_reservations/apply_dhcp_reservation/
+commit_dhcp_reservations), so a fake in-memory driver is enough to exercise
+them fully; no real device or vendor driver needed. See README.md "Design
+principle: generic vs. device-specific logic" for why this logic lives here
+and not in a vendor driver.
+"""
+
+from typing import Any, Dict, List, Optional
+
+import pytest
+from napalm_device_types import DhcpServerMixin, FirewallDriver, ResidentialGatewayDriver
+from napalm_device_types.dhcp import normalize_mac
+from napalm_device_types.models import DhcpReservationDict
+
+
+def _reservation(**overrides: Any) -> DhcpReservationDict:
+    base: DhcpReservationDict = {
+        "uuid": "",
+        "mac": "aa:bb:cc:dd:ee:01",
+        "ip": "10.10.20.50",
+        "hostname": "nas",
+        "description": "Home NAS",
+        "subnet": "10.10.20.0/24",
+    }
+    base.update(overrides)  # type: ignore[typeddict-item]
+    return base
+
+
+class _FakeDhcpServer(DhcpServerMixin):
+    """In-memory fake — no network, no Kea/UCI specifics."""
+
+    def __init__(self, live: Optional[List[DhcpReservationDict]] = None) -> None:
+        self.live = live or []
+        self.applied: List[Dict[str, Any]] = []
+        self.commits = 0
+
+    def get_dhcp_reservations(self) -> List[DhcpReservationDict]:
+        return self.live
+
+    def apply_dhcp_reservation(
+        self, reservation: DhcpReservationDict, *, uuid: Optional[str] = None
+    ) -> Dict[str, Any]:
+        self.applied.append({"reservation": reservation, "uuid": uuid})
+        return {"success": True}
+
+    def commit_dhcp_reservations(self) -> Dict[str, Any]:
+        self.commits += 1
+        return {"success": True}
+
+
+class TestNormalizeMac:
+    def test_lowercases_and_colon_separates(self):
+        assert normalize_mac("AA-BB-CC-DD-EE-01") == "aa:bb:cc:dd:ee:01"
+
+    def test_accepts_bare_hex(self):
+        assert normalize_mac("aabbccddee01") == "aa:bb:cc:dd:ee:01"
+
+    def test_accepts_cisco_dotted(self):
+        assert normalize_mac("aabb.ccdd.ee01") == "aa:bb:cc:dd:ee:01"
+
+    def test_passes_through_unparseable_value_lowercased(self):
+        # Not 12 hex digits — return something stable rather than raising, so a
+        # malformed device response degrades to "never matches" instead of
+        # aborting the whole diff.
+        assert normalize_mac("not-a-mac") == "not-a-mac"
+
+    def test_handles_none(self):
+        assert normalize_mac(None) == ""
+
+
+class TestAbstractContract:
+    """The device hooks are declared for type checkers, not implemented.
+
+    A driver that never implemented them does not carry them at runtime, so
+    `hasattr` is a truthful capability probe and the declaration cannot shadow a
+    working implementation inherited from a sibling base.
+    """
+
+    @pytest.mark.parametrize(
+        "method",
+        [
+            "get_dhcp_reservations",
+            "apply_dhcp_reservation",
+            "commit_dhcp_reservations",
+            "get_dhcp_subnets",
+            "apply_dhcp_subnet",
+            "commit_dhcp_subnets",
+        ],
+    )
+    def test_hook_absent_until_a_driver_implements_it(self, method):
+        assert not hasattr(DhcpServerMixin, method)
+
+    def test_calling_a_missing_hook_fails_loudly(self):
+        with pytest.raises(AttributeError):
+            DhcpServerMixin().get_dhcp_reservations()
+
+
+class TestMixedIntoDriverTypes:
+    """Both firewalls and residential gateways run DHCP servers, so the mixin
+    must be reachable from either base class without re-declaring it."""
+
+    def test_firewall_driver_has_dhcp_reservation_methods(self):
+        assert issubclass(FirewallDriver, DhcpServerMixin)
+
+    def test_residential_gateway_driver_has_dhcp_reservation_methods(self):
+        assert issubclass(ResidentialGatewayDriver, DhcpServerMixin)
+
+
+class TestDiff:
+    def test_empty_device_yields_all_adds(self):
+        driver = _FakeDhcpServer([])
+        diff = driver.diff_dhcp_reservations(
+            [_reservation(), _reservation(mac="aa:bb:cc:dd:ee:02")]
+        )
+
+        assert len(diff["add"]) == 2
+        assert diff["update"] == []
+
+    def test_identical_reservation_produces_no_change(self):
+        live = _reservation(uuid="u1")
+        driver = _FakeDhcpServer([live])
+
+        diff = driver.diff_dhcp_reservations([_reservation()])
+
+        assert diff == {"add": [], "update": []}
+
+    def test_changed_ip_produces_update_with_changed_fields(self):
+        driver = _FakeDhcpServer([_reservation(uuid="u1")])
+
+        diff = driver.diff_dhcp_reservations([_reservation(ip="10.10.20.51")])
+
+        assert diff["add"] == []
+        assert len(diff["update"]) == 1
+        assert diff["update"][0]["uuid"] == "u1"
+        assert diff["update"][0]["changed_fields"] == ["ip"]
+
+    def test_changed_subnet_produces_update_not_add(self):
+        # A host moved to another VLAN keeps its MAC — that is an update of the
+        # existing reservation, not a second reservation for the same MAC.
+        driver = _FakeDhcpServer([_reservation(uuid="u1")])
+
+        diff = driver.diff_dhcp_reservations(
+            [_reservation(ip="10.30.20.50", subnet="10.30.20.0/24")]
+        )
+
+        assert diff["add"] == []
+        assert diff["update"][0]["changed_fields"] == ["ip", "subnet"]
+
+    def test_mac_formatting_differences_still_match(self):
+        driver = _FakeDhcpServer([_reservation(uuid="u1", mac="AA-BB-CC-DD-EE-01")])
+
+        diff = driver.diff_dhcp_reservations([_reservation(mac="aabb.ccdd.ee01")])
+
+        assert diff == {"add": [], "update": []}
+
+    def test_unmanaged_live_reservation_is_never_deleted(self):
+        # Hand-created reservations must survive — the desired set was never
+        # meant to describe them.
+        driver = _FakeDhcpServer([_reservation(uuid="u1", mac="aa:bb:cc:dd:ee:99")])
+
+        diff = driver.diff_dhcp_reservations([_reservation()])
+
+        assert len(diff["add"]) == 1
+        assert diff["update"] == []
+        assert "delete" not in diff
+
+
+class TestApplyReservationSet:
+    def test_applies_adds_and_updates_then_commits(self):
+        driver = _FakeDhcpServer([_reservation(uuid="u1", ip="10.10.20.9")])
+
+        lines = list(
+            driver.apply_dhcp_reservationset(
+                [_reservation(), _reservation(mac="aa:bb:cc:dd:ee:02", hostname="printer")]
+            )
+        )
+
+        assert driver.commits == 1
+        assert len(driver.applied) == 2
+        # The update targets the live uuid; the add does not.
+        by_uuid = {entry["uuid"] for entry in driver.applied}
+        assert by_uuid == {"u1", None}
+        assert any(line.startswith("[add]") for line in lines)
+        assert any(line.startswith("[update]") for line in lines)
+        assert lines[-1].startswith("[commit]")
+
+    def test_progress_lines_name_the_reservation(self):
+        driver = _FakeDhcpServer([])
+
+        lines = list(driver.apply_dhcp_reservationset([_reservation()]))
+
+        assert "aa:bb:cc:dd:ee:01" in lines[0]
+        assert "10.10.20.50" in lines[0]
+
+    def test_no_changes_skips_the_commit(self):
+        # Committing means reloading the DHCP daemon (Kea `service/reconfigure`),
+        # which drops in-flight requests. A no-op diff must not cause that.
+        driver = _FakeDhcpServer([_reservation(uuid="u1")])
+
+        lines = list(driver.apply_dhcp_reservationset([_reservation()]))
+
+        assert driver.commits == 0
+        assert driver.applied == []
+        assert lines == ["[commit] no changes"]

tests/test_dhcp_subnet_diff_apply.py

@@ -0,0 +1,228 @@
+"""Tests for DhcpServerMixin's generic subnet diff/apply mechanism.
+
+Same shape as test_dhcp_diff_apply.py: diff_dhcp_subnets/apply_dhcp_subnetset
+are concrete methods that only depend on the abstract trio
+(get_dhcp_subnets/apply_dhcp_subnet/commit_dhcp_subnets), so an in-memory
+fake exercises them fully. See README.md "Design principle: generic vs.
+device-specific logic".
+
+A subnet is a heavier object than a reservation: a wrong `pools` or
+`option_data` takes a whole VLAN offline rather than one host, so the tests
+below lean on the never-delete and preserve-unmanaged-options guarantees.
+"""
+
+from typing import Any, Dict, List, Optional
+
+import pytest
+from napalm_device_types import DhcpServerMixin
+from napalm_device_types.models import DhcpSubnetDict
+
+
+def _subnet(**overrides: Any) -> DhcpSubnetDict:
+    base: DhcpSubnetDict = {
+        "uuid": "",
+        "subnet": "10.10.20.0/24",
+        "description": "Home",
+        "pools": ["10.10.20.100-10.10.20.200"],
+        "option_data": {
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+        },
+        "match_client_id": False,
+    }
+    base.update(overrides)  # type: ignore[typeddict-item]
+    return base
+
+
+class _FakeDhcpServer(DhcpServerMixin):
+    def __init__(self, live: Optional[List[DhcpSubnetDict]] = None) -> None:
+        self.live = live or []
+        self.applied: List[Dict[str, Any]] = []
+        self.commits = 0
+
+    def get_dhcp_subnets(self) -> List[DhcpSubnetDict]:
+        return self.live
+
+    def apply_dhcp_subnet(
+        self, subnet: DhcpSubnetDict, *, uuid: Optional[str] = None
+    ) -> Dict[str, Any]:
+        self.applied.append({"subnet": subnet, "uuid": uuid})
+        return {"success": True}
+
+    def commit_dhcp_subnets(self) -> Dict[str, Any]:
+        self.commits += 1
+        return {"success": True}
+
+
+# ── diff: adds ────────────────────────────────────────────────────────────────
+
+
+def test_unknown_subnet_is_an_add() -> None:
+    diff = _FakeDhcpServer().diff_dhcp_subnets([_subnet()])
+    assert len(diff["add"]) == 1
+    assert diff["add"][0]["subnet"] == "10.10.20.0/24"
+    assert diff["update"] == []
+
+
+def test_matching_subnet_with_no_changes_is_neither() -> None:
+    live = _subnet(uuid="dev-1")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet()])
+    assert diff == {"add": [], "update": []}
+
+
+# ── diff: identity is the CIDR ────────────────────────────────────────────────
+
+
+def test_subnets_are_matched_on_cidr() -> None:
+    live = _subnet(uuid="dev-1", description="renamed on the device")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet()])
+    assert diff["add"] == []
+    assert diff["update"][0]["uuid"] == "dev-1"
+    assert diff["update"][0]["changed_fields"] == ["description"]
+
+
+def test_a_different_cidr_is_a_new_subnet_not_an_update() -> None:
+    live = _subnet(uuid="dev-1", subnet="10.10.20.0/24")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet(subnet="10.10.30.0/24")])
+    assert len(diff["add"]) == 1
+    assert diff["update"] == []
+
+
+# ── diff: compared fields ─────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("description", "Office"),
+        ("pools", ["10.10.20.50-10.10.20.99"]),
+        ("match_client_id", True),
+    ],
+)
+def test_changed_field_is_reported(field: str, value: Any) -> None:
+    live = _subnet(uuid="dev-1")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet(**{field: value})])
+    assert diff["update"][0]["changed_fields"] == [field]
+
+
+def test_changed_option_is_reported_as_option_data() -> None:
+    live = _subnet(uuid="dev-1")
+    desired = _subnet(
+        option_data={
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+            "domain_search": ["home.local", "office.local"],
+        }
+    )
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([desired])
+    assert diff["update"][0]["changed_fields"] == ["option_data"]
+
+
+def test_an_unmanaged_option_on_the_device_is_not_a_change() -> None:
+    """Absent key means "not managed" -- it must not provoke an update.
+
+    Kea autocollects routers/domain_name_servers/ntp_servers. A caller that
+    only wants to set domain_search would otherwise diff-fight the server
+    forever, reloading the DHCP daemon on every run.
+    """
+    live = _subnet(
+        uuid="dev-1",
+        option_data={
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+            "ntp_servers": ["10.10.20.1"],
+        },
+    )
+    desired = _subnet(option_data={"domain_search": ["home.local"]})
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([desired])
+    assert diff["update"][0]["changed_fields"] == ["option_data"]
+
+    # ...and once it matches, it stays quiet.
+    live2 = _subnet(
+        uuid="dev-1",
+        option_data={
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+            "ntp_servers": ["10.10.20.1"],
+            "domain_search": ["home.local"],
+        },
+    )
+    assert _FakeDhcpServer([live2]).diff_dhcp_subnets([desired]) == {"add": [], "update": []}
+
+
+def test_option_order_is_significant_for_domain_search() -> None:
+    """Search order decides which zone answers an unqualified name first."""
+    live = _subnet(uuid="dev-1", option_data={"domain_search": ["a.local", "b.local"]})
+    desired = _subnet(option_data={"domain_search": ["b.local", "a.local"]})
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([desired])
+    assert diff["update"][0]["changed_fields"] == ["option_data"]
+
+
+# ── diff: never delete ────────────────────────────────────────────────────────
+
+
+def test_live_subnet_absent_from_desired_is_never_deleted() -> None:
+    """Deleting a subnet takes a whole VLAN's DHCP down. Never implicit."""
+    live = _subnet(uuid="dev-1", subnet="10.10.99.0/24")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet()])
+    assert len(diff["add"]) == 1
+    assert diff["update"] == []
+    assert "delete" not in diff
+
+
+# ── apply ─────────────────────────────────────────────────────────────────────
+
+
+def test_apply_creates_then_commits() -> None:
+    fake = _FakeDhcpServer()
+    lines = list(fake.apply_dhcp_subnetset([_subnet()]))
+    assert fake.applied[0]["uuid"] is None
+    assert fake.commits == 1
+    assert any(line.startswith("[add]") for line in lines)
+
+
+def test_apply_updates_in_place_with_the_device_uuid() -> None:
+    fake = _FakeDhcpServer([_subnet(uuid="dev-1")])
+    list(fake.apply_dhcp_subnetset([_subnet(description="Office")]))
+    assert fake.applied[0]["uuid"] == "dev-1"
+    assert fake.commits == 1
+
+
+def test_apply_does_not_commit_when_nothing_changed() -> None:
+    """A commit reconfigures the DHCP daemon -- too costly for a no-op run."""
+    fake = _FakeDhcpServer([_subnet(uuid="dev-1")])
+    lines = list(fake.apply_dhcp_subnetset([_subnet()]))
+    assert fake.commits == 0
+    assert lines == ["[commit] no changes"]
+
+
+def test_apply_names_the_subnet_in_its_progress_line() -> None:
+    fake = _FakeDhcpServer()
+    lines = list(fake.apply_dhcp_subnetset([_subnet()]))
+    assert "10.10.20.0/24" in lines[0]
+
+
+def test_apply_reports_changed_fields_on_update() -> None:
+    fake = _FakeDhcpServer([_subnet(uuid="dev-1")])
+    lines = list(fake.apply_dhcp_subnetset([_subnet(description="Office")]))
+    assert "description" in lines[0]
+
+
+# ── abstract surface ──────────────────────────────────────────────────────────
+
+
+class _BareDriver(DhcpServerMixin):
+    pass
+
+
+@pytest.mark.parametrize(
+    "method", ["get_dhcp_subnets", "apply_dhcp_subnet", "commit_dhcp_subnets"]
+)
+def test_device_specific_methods_are_absent_until_implemented(method: str) -> None:
+    """Declared under TYPE_CHECKING, so they do not exist until a driver adds them."""
+    assert not hasattr(_BareDriver, method)
+
+
+def test_calling_a_missing_device_method_fails_loudly() -> None:
+    with pytest.raises(AttributeError):
+        _BareDriver().get_dhcp_subnets()

tests/test_firewall_diff_apply.py

@@ -0,0 +1,171 @@
+"""Tests for FirewallDriver's generic diff/apply mechanism.
+
+diff_firewall_rules/apply_firewall_ruleset are concrete methods on the base
+class (not overridden by concrete drivers) — they only depend on the three
+abstract methods (get_firewall_rules/apply_firewall_rule/commit_firewall_rules),
+so a fake in-memory driver is enough to exercise them fully; no real device
+or vendor driver needed. See README.md "Design principle: generic vs.
+device-specific logic" for why this logic lives here and not in a vendor
+driver.
+"""
+
+from typing import Any, Dict, List, Optional
+
+import pytest
+from napalm_device_types import FirewallDriver
+from napalm_device_types.models import FirewallRuleDict
+
+
+def _rule(**overrides: Any) -> FirewallRuleDict:
+    base: FirewallRuleDict = {
+        "uuid": "",
+        "description": "allow_mgmt_to_fw_gui",
+        "action": "pass",
+        "interface": "lan",
+        "direction": "in",
+        "protocol": "tcp",
+        "source_net": "MGMT_NET",
+        "source_port": "",
+        "destination_net": "(self)",
+        "destination_port": "https",
+        "enabled": True,
+        "quick": True,
+        "log": False,
+    }
+    base.update(overrides)  # type: ignore[typeddict-item]
+    return base
+
+
+class _FakeFirewall(FirewallDriver):
+    """In-memory fake — no network, no OPNsense/vendor specifics."""
+
+    def __init__(self, live_rules: Optional[List[FirewallRuleDict]] = None) -> None:
+        self.live_rules = live_rules or []
+        self.applied: List[Dict[str, Any]] = []
+        self.committed = False
+
+    def get_firewall_rules(self) -> List[FirewallRuleDict]:
+        return self.live_rules
+
+    def apply_firewall_rule(
+        self, rule: FirewallRuleDict, *, uuid: Optional[str] = None
+    ) -> Dict[str, Any]:
+        self.applied.append({"rule": rule, "uuid": uuid})
+        return {"success": True}
+
+    def commit_firewall_rules(self) -> Dict[str, Any]:
+        self.committed = True
+        return {"success": True}
+
+
+class TestAbstractContract:
+    """The three device hooks are declared, never implemented, on the base.
+
+    They exist for type checkers only, so a driver that did not implement them
+    does not carry them at runtime -- which is what keeps `hasattr` truthful and
+    stops the declaration from overriding a sibling base's working version.
+    """
+
+    @pytest.mark.parametrize(
+        "method", ["get_firewall_rules", "apply_firewall_rule", "commit_firewall_rules"]
+    )
+    def test_hook_absent_until_a_driver_implements_it(self, method):
+        class _Bare(FirewallDriver):
+            def __init__(self) -> None:
+                pass
+
+        assert not hasattr(_Bare, method)
+
+    def test_calling_a_missing_hook_fails_loudly(self):
+        class _Bare(FirewallDriver):
+            def __init__(self) -> None:
+                pass
+
+        with pytest.raises(AttributeError):
+            _Bare().get_firewall_rules()
+
+
+class TestDiffFirewallRules:
+    def test_desired_rule_missing_live_is_an_add(self):
+        driver = _FakeFirewall(live_rules=[])
+        diff = driver.diff_firewall_rules([_rule()])
+
+        assert len(diff["add"]) == 1
+        assert diff["add"][0]["description"] == "allow_mgmt_to_fw_gui"
+        assert diff["update"] == []
+
+    def test_matching_rule_with_changed_field_is_an_update(self):
+        live = _rule(uuid="abc-123", action="block")
+        driver = _FakeFirewall(live_rules=[live])
+        diff = driver.diff_firewall_rules([_rule(action="pass")])
+
+        assert diff["add"] == []
+        assert len(diff["update"]) == 1
+        update = diff["update"][0]
+        assert update["uuid"] == "abc-123"
+        assert update["changed_fields"] == ["action"]
+        assert update["rule"]["action"] == "pass"
+
+    def test_identical_rule_produces_no_diff(self):
+        live = _rule(uuid="abc-123")
+        driver = _FakeFirewall(live_rules=[live])
+        diff = driver.diff_firewall_rules([_rule()])
+
+        assert diff["add"] == []
+        assert diff["update"] == []
+
+    def test_live_rule_not_in_desired_is_never_deleted(self):
+        """v1 never deletes -- rules present live but absent from `desired`
+        are simply ignored, not reported for removal."""
+        live = _rule(uuid="abc-123", description="some_unmanaged_rule")
+        driver = _FakeFirewall(live_rules=[live])
+        diff = driver.diff_firewall_rules([])
+
+        assert diff == {"add": [], "update": []}
+        assert "delete" not in diff
+
+    def test_multiple_changed_fields_all_reported(self):
+        live = _rule(uuid="abc-123", action="block", protocol="udp", log=True)
+        driver = _FakeFirewall(live_rules=[live])
+        diff = driver.diff_firewall_rules([_rule(action="pass", protocol="tcp", log=False)])
+
+        assert set(diff["update"][0]["changed_fields"]) == {"action", "protocol", "log"}
+
+
+class TestApplyFirewallRuleset:
+    def test_adds_are_applied_with_no_uuid(self):
+        driver = _FakeFirewall(live_rules=[])
+        list(driver.apply_firewall_ruleset([_rule()]))
+
+        assert len(driver.applied) == 1
+        assert driver.applied[0]["uuid"] is None
+        assert driver.applied[0]["rule"]["description"] == "allow_mgmt_to_fw_gui"
+
+    def test_updates_are_applied_with_existing_uuid(self):
+        live = _rule(uuid="abc-123", action="block")
+        driver = _FakeFirewall(live_rules=[live])
+        list(driver.apply_firewall_ruleset([_rule(action="pass")]))
+
+        assert len(driver.applied) == 1
+        assert driver.applied[0]["uuid"] == "abc-123"
+
+    def test_commits_after_applying(self):
+        driver = _FakeFirewall(live_rules=[])
+        list(driver.apply_firewall_ruleset([_rule()]))
+
+        assert driver.committed is True
+
+    def test_yields_a_progress_line_per_change(self):
+        driver = _FakeFirewall(live_rules=[])
+        lines = list(driver.apply_firewall_ruleset([_rule(), _rule(description="second_rule")]))
+
+        assert len(lines) >= 2
+        assert all(isinstance(line, str) for line in lines)
+
+    def test_no_changes_still_commits_but_applies_nothing(self):
+        live = _rule(uuid="abc-123")
+        driver = _FakeFirewall(live_rules=[live])
+        list(driver.apply_firewall_ruleset([_rule()]))
+
+        assert driver.applied == []
+        assert driver.committed is True

tests/test_host_reboot.py

@@ -0,0 +1,29 @@
+"""reboot_host: the contract for restarting the device itself.
+
+netOrk used to restart a host by sending ``/sbin/reboot`` through a driver's
+private ``_send_command``. A driver that talks to an API instead has no such
+method, and the caller swallowed the resulting AttributeError, so the reboot
+"succeeded" without happening. A declared contract lets netOrk ask first.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+from napalm_device_types import DeviceTypeDriver, HostRebootMixin
+
+
+def test_every_device_type_driver_carries_the_declaration():
+    assert issubclass(DeviceTypeDriver, HostRebootMixin)
+
+
+def test_declared_not_implemented():
+    """hasattr is netOrk's capability probe; only a driver that implements it
+    may answer True."""
+    assert not hasattr(DeviceTypeDriver, "reboot_host")
+
+
+def test_declaration_documents_the_contract():
+    source = inspect.getsource(HostRebootMixin)
+    assert "def reboot_host(self) -> None" in source
+    assert "RuntimeError" in source

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_ping_sweep.py

@@ -0,0 +1,221 @@
+"""Tests for the generic ping sweep (PingSweepMixin + driver_supports_ping)."""
+
+import pytest
+from napalm.base import NetworkDriver
+
+from napalm_device_types import DeviceTypeDriver, PingSweepMixin, driver_supports_ping
+
+
+# ── Fakes ─────────────────────────────────────────────────────────────────────
+
+
+def _napalm_ok(rtt: float, probes: int = 1) -> dict:
+    """A NAPALM-format ping() reply for a reachable destination."""
+    return {
+        "success": {
+            "probes_sent": probes,
+            "packet_loss": 0,
+            "rtt_min": rtt,
+            "rtt_avg": rtt,
+            "rtt_max": rtt,
+            "rtt_stddev": 0.0,
+            "results": [{"ip_address": "10.0.0.1", "rtt": rtt}],
+        }
+    }
+
+
+def _napalm_lost(probes: int = 1) -> dict:
+    """A NAPALM-format ping() reply where every probe was lost."""
+    return {
+        "success": {
+            "probes_sent": probes,
+            "packet_loss": probes,
+            "rtt_min": 0.0,
+            "rtt_avg": 0.0,
+            "rtt_max": 0.0,
+            "rtt_stddev": 0.0,
+            "results": [],
+        }
+    }
+
+
+class FakePingDriver(PingSweepMixin):
+    """Minimal driver exposing ping() — stands in for a real vendor driver."""
+
+    def __init__(self, replies=None, raises=None):
+        self.replies = replies or {}
+        self.raises = raises or {}
+        self.calls = []
+
+    def ping(self, destination, source="", ttl=255, timeout=2, size=100, count=5, vrf=""):
+        self.calls.append({"destination": destination, "timeout": timeout, "count": count})
+        if destination in self.raises:
+            raise self.raises[destination]
+        return self.replies.get(destination, _napalm_lost())
+
+
+class NoPingDriver(PingSweepMixin):
+    """Driver without its own ping() — inherits NAPALM's NotImplementedError stub."""
+
+    ping = NetworkDriver.ping
+
+
+class OptedOutDriver(FakePingDriver):
+    """Driver that implements ping() but declares it unusable for sweeps."""
+
+    SUPPORTS_PING = False
+
+
+# ── driver_supports_ping ──────────────────────────────────────────────────────
+
+
+def test_driver_supports_ping_false_for_unoverridden_ping():
+    assert driver_supports_ping(NoPingDriver) is False
+
+
+def test_driver_supports_ping_false_for_base_network_driver():
+    assert driver_supports_ping(NetworkDriver) is False
+
+
+def test_driver_supports_ping_true_when_overridden():
+    assert driver_supports_ping(FakePingDriver) is True
+
+
+def test_driver_supports_ping_honours_explicit_opt_out():
+    assert driver_supports_ping(OptedOutDriver) is False
+
+
+def test_driver_supports_ping_false_for_class_without_ping():
+    class Bare:
+        pass
+
+    assert driver_supports_ping(Bare) is False
+
+
+def test_device_type_driver_exposes_supports_ping_classmethod():
+    assert DeviceTypeDriver.supports_ping() is False
+    assert FakePingDriver.supports_ping() is True
+
+
+# ── ping_sweep ────────────────────────────────────────────────────────────────
+
+
+def test_ping_sweep_marks_reachable_and_unreachable_hosts():
+    driver = FakePingDriver(replies={"10.0.0.1": _napalm_ok(1.5)})
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"])
+
+    assert result["scanned"] == 2
+    assert result["alive_count"] == 1
+    assert result["truncated"] is False
+    assert result["entries"] == [
+        {"ip": "10.0.0.1", "alive": True, "rtt_ms": 1.5},
+        {"ip": "10.0.0.2", "alive": False, "rtt_ms": None},
+    ]
+
+
+def test_ping_sweep_uses_single_fast_probe_by_default():
+    driver = FakePingDriver()
+
+    driver.ping_sweep(["10.0.0.1"])
+
+    assert driver.calls == [{"destination": "10.0.0.1", "timeout": 1, "count": 1}]
+
+
+def test_ping_sweep_forwards_count_and_timeout():
+    driver = FakePingDriver()
+
+    driver.ping_sweep(["10.0.0.1"], count=3, timeout=5)
+
+    assert driver.calls == [{"destination": "10.0.0.1", "timeout": 5, "count": 3}]
+
+
+def test_ping_sweep_treats_error_reply_as_unreachable():
+    driver = FakePingDriver(replies={"10.0.0.9": {"error": "unknown host"}})
+
+    result = driver.ping_sweep(["10.0.0.9"])
+
+    assert result["entries"][0]["alive"] is False
+    assert result["entries"][0]["error"] == "unknown host"
+    assert result["alive_count"] == 0
+
+
+def test_ping_sweep_records_exception_and_continues():
+    driver = FakePingDriver(
+        replies={"10.0.0.2": _napalm_ok(2.0)},
+        raises={"10.0.0.1": RuntimeError("session closed")},
+    )
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"])
+
+    assert result["entries"][0] == {
+        "ip": "10.0.0.1",
+        "alive": False,
+        "rtt_ms": None,
+        "error": "session closed",
+    }
+    assert result["entries"][1]["alive"] is True
+    assert result["scanned"] == 2
+
+
+def test_ping_sweep_truncates_at_max_targets():
+    driver = FakePingDriver()
+
+    result = driver.ping_sweep([f"10.0.0.{i}" for i in range(1, 11)], max_targets=4)
+
+    assert result["scanned"] == 4
+    assert result["truncated"] is True
+    assert len(driver.calls) == 4
+
+
+def test_ping_sweep_respects_class_level_max_targets():
+    class SmallSweepDriver(FakePingDriver):
+        PING_SWEEP_MAX_TARGETS = 2
+
+    driver = SmallSweepDriver()
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2", "10.0.0.3"])
+
+    assert result["scanned"] == 2
+    assert result["truncated"] is True
+
+
+def test_ping_sweep_reports_progress_per_destination():
+    driver = FakePingDriver(replies={"10.0.0.1": _napalm_ok(1.0)})
+    seen = []
+
+    driver.ping_sweep(
+        ["10.0.0.1", "10.0.0.2"],
+        on_progress=lambda done, total: seen.append((done, total)),
+    )
+
+    assert seen == [(1, 2), (2, 2)]
+
+
+def test_ping_sweep_on_empty_destination_list():
+    driver = FakePingDriver()
+
+    result = driver.ping_sweep([])
+
+    assert result == {"entries": [], "scanned": 0, "alive_count": 0, "truncated": False}
+
+
+def test_ping_sweep_raises_when_driver_cannot_ping():
+    driver = NoPingDriver()
+
+    with pytest.raises(NotImplementedError):
+        driver.ping_sweep(["10.0.0.1"])
+
+
+def test_ping_sweep_stops_when_stop_requested():
+    driver = FakePingDriver()
+    calls = {"n": 0}
+
+    def _should_stop():
+        calls["n"] += 1
+        return calls["n"] > 1
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2", "10.0.0.3"], should_stop=_should_stop)
+
+    assert result["scanned"] == 1
+    assert len(driver.calls) == 1

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_role_contracts.py

@@ -0,0 +1,141 @@
+# -*- coding: utf-8 -*-
+"""The contract that makes multi-role drivers safe.
+
+A role base declares what a device of that kind can be asked for. It must not
+*implement* anything -- not even a placeholder. A ``NotImplementedError`` stub
+on a base class is not neutral under multiple inheritance: it wins the MRO
+against a sibling base's working implementation and silently replaces it. That
+failure has hit this codebase three times (OpenWrt's seven forwarding methods,
+QNAP's ``get_services``, OpenMediaVault avoiding ``StorageDriver`` altogether).
+
+These tests pin the property that prevents a fourth.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+
+from napalm_device_types import (
+    AccessPointDriver,
+    DeviceTypeDriver,
+    FirewallDriver,
+    HypervisorDriver,
+    MediaDriver,
+    OSDriver,
+    PhoneDriver,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+)
+from napalm_device_types.roles import primary_role_of, roles_of
+
+ROLE_BASES = [
+    AccessPointDriver,
+    FirewallDriver,
+    HypervisorDriver,
+    MediaDriver,
+    OSDriver,
+    PhoneDriver,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+]
+
+
+@pytest.mark.parametrize("base", ROLE_BASES, ids=lambda b: b.__name__)
+class TestRoleBasesAreContractsOnly:
+    def test_defines_no_methods_at_runtime(self, base):
+        """A role base is a declaration. Anything callable it owns can shadow a
+        sibling base, so it must own nothing callable at all."""
+        own = [
+            name
+            for name, val in vars(base).items()
+            if not name.startswith("__")
+            and (inspect.isfunction(val) or isinstance(val, (classmethod, staticmethod)))
+        ]
+        assert own == [], f"{base.__name__} implements {own}; move it to a function class"
+
+    def test_declares_a_role_key(self, base):
+        assert isinstance(vars(base).get("ROLE"), str) and vars(base)["ROLE"]
+
+    def test_has_a_real_docstring(self, base):
+        """TYPE_LABEL used to be assigned above the triple-quoted string, which
+        made it a bare expression rather than a docstring -- __doc__ was None on
+        all seven bases, killing help() and IDE hovers."""
+        assert base.__doc__ and base.__doc__.strip()
+
+
+class TestDeclaredMethodsDoNotExistAtRuntime:
+    """`hasattr` is netOrk's capability probe. It only tells the truth when a
+    declared-but-unimplemented method is genuinely absent."""
+
+    @pytest.mark.parametrize(
+        ("base", "method"),
+        [
+            (StorageDriver, "get_disks"),
+            (StorageDriver, "get_volumes"),
+            (HypervisorDriver, "get_vms"),
+            (FirewallDriver, "send_wake_on_lan"),
+            (SwitchDriver, "set_vlan"),
+            (AccessPointDriver, "get_wireless_config"),
+            (OSDriver, "get_processes"),
+            (ResidentialGatewayDriver, "get_wan_status"),
+            (PhoneDriver, "get_sip_accounts"),
+            (MediaDriver, "get_playback_state"),
+        ],
+    )
+    def test_absent_until_a_driver_implements_it(self, base, method):
+        assert not hasattr(base, method)
+
+
+class TestNoShadowingAcrossRoles:
+    """The regression test for the bug class."""
+
+    def test_role_base_listed_first_does_not_shadow_sibling(self):
+        """StorageDriver precedes the working mixin -- the order that broke QNAP."""
+
+        class WorkingPackages:
+            def get_packages(self):
+                return [{"name": "vim", "version": "9.0"}]
+
+            def get_services(self):
+                return [{"name": "sshd", "state": "running"}]
+
+        class Combined(StorageDriver, WorkingPackages):
+            pass
+
+        assert Combined.get_packages is WorkingPackages.get_packages
+        assert Combined.get_services is WorkingPackages.get_services
+
+    def test_two_role_bases_can_be_combined(self):
+        """A QNAP is NAS, hypervisor and Linux host. It must be able to say so."""
+
+        class Nas(StorageDriver, HypervisorDriver, OSDriver):
+            pass
+
+        assert [r.__name__ for r in roles_of(Nas)] == [
+            "StorageDriver",
+            "HypervisorDriver",
+            "OSDriver",
+        ]
+
+
+class TestRoleIntrospection:
+    def test_primary_role_follows_base_order(self):
+        class Nas(StorageDriver, OSDriver):
+            pass
+
+        class Host(OSDriver, StorageDriver):
+            pass
+
+        assert primary_role_of(Nas) == "storage"
+        assert primary_role_of(Host) == "linux"
+
+    def test_driver_without_a_role_has_none(self):
+        class Bare(DeviceTypeDriver):
+            pass
+
+        assert roles_of(Bare) == []
+        assert primary_role_of(Bare) is None

tests/test_send_wake_on_lan.py

@@ -0,0 +1,41 @@
+"""Tests for the FirewallDriver.send_wake_on_lan contract.
+
+``send_wake_on_lan`` is declared on ``FirewallDriver`` but implemented by very
+few firewalls. It used to exist as a ``NotImplementedError`` stub; it is now a
+``TYPE_CHECKING`` declaration, so a driver that never implemented it simply does
+not have the attribute. That is what lets ``hasattr`` answer honestly, and what
+stops the declaration from shadowing a working implementation inherited from a
+sibling base.
+"""
+
+import pytest
+from napalm_device_types import FirewallDriver
+
+
+class _BareFirewall(FirewallDriver):
+    """FirewallDriver.__init__ is NetworkDriver's, which itself raises
+    NotImplementedError — override with a no-op so tests exercise
+    send_wake_on_lan itself, not construction."""
+
+    def __init__(self):
+        pass
+
+
+def test_absent_on_a_driver_that_never_implemented_it():
+    assert not hasattr(_BareFirewall, "send_wake_on_lan")
+
+
+def test_calling_it_anyway_fails_loudly():
+    """A caller that skips the hasattr check must not get silence."""
+    with pytest.raises(AttributeError):
+        _BareFirewall().send_wake_on_lan("AA:BB:CC:DD:EE:FF")
+
+
+def test_subclass_can_implement_send_wake_on_lan():
+    class MyFirewall(_BareFirewall):
+        def send_wake_on_lan(self, mac_address: str, interface: str = ""):
+            return {"success": True, "output": f"woke {mac_address} via {interface}"}
+
+    assert hasattr(MyFirewall, "send_wake_on_lan")
+    result = MyFirewall().send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
+    assert result == {"success": True, "output": "woke AA:BB:CC:DD:EE:FF via lan"}

tests/test_vm_identifiers.py

@@ -0,0 +1,70 @@
+"""A VM's ``vmid`` is a string on every hypervisor.
+
+Proxmox numbers its guests, but VMware identifies them by UUID or MoRef
+(``"vm-42"``). An ``int`` in the contract forced netOrk to call ``int()`` on
+whatever came back, which cannot represent the second kind at all. The
+provisioning dicts already carried ``vmid`` as a string; these pin the
+read-side dicts to the same type.
+"""
+
+from __future__ import annotations
+
+from typing import get_type_hints
+
+import pytest
+from napalm_device_types.models import (
+    VMConfigDict,
+    VMDict,
+    VMProvisionResultDict,
+)
+
+
+@pytest.mark.parametrize("model", [VMDict, VMConfigDict, VMProvisionResultDict])
+def test_vmid_is_a_string(model):
+    assert get_type_hints(model)["vmid"] is str
+
+
+class TestVMConfigCarriesWhatAHardwareViewShows:
+    """netOrk's VM hardware view used to read Proxmox's raw config through the
+    driver's private API. The contract has to carry those details so a second
+    hypervisor can fill the same view -- optionally, since not every platform
+    has every one of them."""
+
+    OPTIONAL = {
+        "os_name",
+        "cpu_type",
+        "sockets",
+        "cores_per_socket",
+        "firmware",
+        "machine",
+        "passthrough",
+    }
+
+    def test_hardware_details_are_optional_fields(self):
+        assert self.OPTIONAL <= VMConfigDict.__optional_keys__
+
+    def test_contract_core_stays_required(self):
+        assert "vmid" in VMConfigDict.__required_keys__
+        assert "disks" in VMConfigDict.__required_keys__
+
+    def test_passthrough_entry_shape(self):
+        from napalm_device_types.models import VMPassthroughDict
+
+        assert get_type_hints(VMPassthroughDict) == {"slot": str, "kind": str, "config": str}
+
+
+class TestGuestAgentDeclaration:
+    """netOrk's cloud-init installed qemu-guest-agent on every new VM. A VMware
+    guest reports its IP through open-vm-tools instead; the hypervisor says
+    which, and netOrk stops hard-coding one of them."""
+
+    def test_default_is_qemu_guest_agent(self):
+        from napalm_device_types import HypervisorDriver
+
+        assert HypervisorDriver.GUEST_AGENT_PACKAGES == ("qemu-guest-agent",)
+        assert HypervisorDriver.GUEST_AGENT_RUNCMD == ("systemctl enable --now qemu-guest-agent",)
+
+    def test_attributes_are_not_methods(self):
+        from napalm_device_types import HypervisorDriver
+
+        assert not callable(vars(HypervisorDriver)["GUEST_AGENT_PACKAGES"])