feat(dhcp): add DhcpServerMixin for static DHCP reservations

Adds the generic half of DHCP reservation management: diff_dhcp_reservations
matches desired against live reservations by normalised MAC, and
apply_dhcp_reservationset walks the diff and commits once at the end.

Both are concrete here because neither is vendor-specific — only
get_dhcp_reservations/apply_dhcp_reservation/commit_dhcp_reservations touch
the device (Kea REST on OPNsense, dnsmasq/odhcpd UCI on OpenWrt).

Two deliberate choices:

- The MAC is the matching key, not a description as with firewall rules. A
  reservation has a natural identity and this is it. That also means a host
  moving to another VLAN is an update of the existing entry rather than a
  second one for the same MAC.
- An empty diff skips the commit. Committing reloads the DHCP daemon and
  drops in-flight requests, which is too high a price for a no-op run. This
  differs from apply_firewall_ruleset, which always commits.

Live reservations with no desired counterpart are never reported for
deletion — a DHCP server routinely carries hand-created entries the caller's
desired set was never meant to describe.

Mixed into FirewallDriver and ResidentialGatewayDriver: both device types
commonly run the DHCP server for their networks.

README.md

@@ -56,6 +56,12 @@ class FirewallDriver(DeviceTypeDriver):
         ...  # 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.
@@ -103,6 +109,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
 

napalm_device_types/__init__.py

@@ -28,11 +28,15 @@ Also provided:
 * :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin` --
   stand-alone mixin to reduce duplication of config lifecycle methods across
   drivers.
+* :class:`~napalm_device_types.dhcp.DhcpServerMixin` -- static DHCP
+  reservation read/diff/apply, mixed into the firewall and gateway base
+  classes.
 """
 
 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_mac
 from napalm_device_types.firewall import FirewallDriver
 from napalm_device_types.hypervisor import HypervisorDriver
 from napalm_device_types.os import OSDriver
@@ -45,6 +49,7 @@ __all__ = [
     "AccessPointDriver",
     "ConfigLifecycleMixin",
     "DeviceTypeDriver",
+    "DhcpServerMixin",
     "FingerprintRule",
     "FirewallDriver",
     "HypervisorDriver",
@@ -55,4 +60,5 @@ __all__ = [
     "StorageDriver",
     "SwitchDriver",
     "driver_supports_ping",
+    "normalize_mac",
 ]

napalm_device_types/dhcp.py

@@ -0,0 +1,211 @@
+# -*- coding: utf-8 -*-
+"""Static DHCP reservation management, shared by firewalls and gateways.
+
+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 ``get_dhcp_reservations``/``apply_dhcp_reservation``/
+``commit_dhcp_reservations`` are device-specific and must be implemented by
+a concrete driver; the diff and the apply loop are vendor-neutral algorithms
+and live here -- see README.md "Design principle: generic vs. device-specific
+logic".
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, Iterator, List, Optional
+
+from napalm_device_types.models import (
+    DhcpReservationDiffDict,
+    DhcpReservationDict,
+    DhcpReservationUpdateDict,
+)
+
+# `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",
+)
+
+_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))
+
+
+class DhcpServerMixin:
+    """Mixin providing static DHCP reservation read/diff/apply.
+
+    Concrete drivers **must** provide ``get_dhcp_reservations()``,
+    ``apply_dhcp_reservation()`` and ``commit_dhcp_reservations()``.
+    """
+
+    # ------------------------------------------------------------------
+    # Device-specific -- must be implemented by the concrete driver.
+    # ------------------------------------------------------------------
+
+    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.
+        """
+        raise NotImplementedError
+
+    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}``.
+        """
+        raise NotImplementedError
+
+    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}``.
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # 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)"
+        )

napalm_device_types/firewall.py

@@ -12,6 +12,7 @@ Usage::
 
 from typing import Any, Dict, Iterator, List, Optional
 from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types.dhcp import DhcpServerMixin
 from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
 from napalm_device_types.models import (
     FirewallRuleDict,
@@ -40,7 +41,7 @@ _FIREWALL_RULE_COMPARE_FIELDS = (
 )
 
 
-class FirewallDriver(DeviceTypeDriver):
+class FirewallDriver(DhcpServerMixin, DeviceTypeDriver):
     TYPE_LABEL: str = "Firewall"
     """
     Abstract intermediate driver for firewall/security devices.

napalm_device_types/models.py

@@ -353,6 +353,40 @@ class FirewallRuleDiffDict(TypedDict):
     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 VPNTunnelDict(TypedDict):
     type: str
     local_endpoint: str

napalm_device_types/residential_gateway.py

@@ -20,6 +20,7 @@ Usage::
 
 from typing import Dict, List
 from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types.dhcp import DhcpServerMixin
 from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
 from napalm_device_types.models import (
     HealthMetricsDict,
@@ -34,7 +35,7 @@ from napalm_device_types.models import (
 )
 
 
-class ResidentialGatewayDriver(DeviceTypeDriver):
+class ResidentialGatewayDriver(DhcpServerMixin, DeviceTypeDriver):
     TYPE_LABEL: str = "Gateway"
     """
     Abstract intermediate driver for residential gateways (router + firewall + AP).

tests/test_dhcp_diff_apply.py

@@ -0,0 +1,195 @@
+"""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:
+    def test_get_dhcp_reservations_raises_not_implemented_by_default(self):
+        with pytest.raises(NotImplementedError):
+            DhcpServerMixin().get_dhcp_reservations()
+
+    def test_apply_dhcp_reservation_raises_not_implemented_by_default(self):
+        with pytest.raises(NotImplementedError):
+            DhcpServerMixin().apply_dhcp_reservation(_reservation())
+
+    def test_commit_dhcp_reservations_raises_not_implemented_by_default(self):
+        with pytest.raises(NotImplementedError):
+            DhcpServerMixin().commit_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"]