From b8977cdaa5efad7fe293d673b3c59a9a583b10a6 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Wed, 19 Aug 2026 07:24:39 +0700 Subject: [PATCH] feat(dhcp): add DhcpServerMixin for static DHCP reservations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 13 ++ napalm_device_types/__init__.py | 6 + napalm_device_types/dhcp.py | 211 +++++++++++++++++++++ napalm_device_types/firewall.py | 3 +- napalm_device_types/models.py | 34 ++++ napalm_device_types/residential_gateway.py | 3 +- tests/test_dhcp_diff_apply.py | 195 +++++++++++++++++++ 7 files changed, 463 insertions(+), 2 deletions(-) create mode 100644 napalm_device_types/dhcp.py create mode 100644 tests/test_dhcp_diff_apply.py diff --git a/README.md b/README.md index 91f82e2..a6137af 100644 --- a/README.md +++ b/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 diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index 9e148f1..a9022c8 100644 --- a/napalm_device_types/__init__.py +++ b/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", ] diff --git a/napalm_device_types/dhcp.py b/napalm_device_types/dhcp.py new file mode 100644 index 0000000..1e31a09 --- /dev/null +++ b/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)" + ) diff --git a/napalm_device_types/firewall.py b/napalm_device_types/firewall.py index 8fee538..1627de1 100644 --- a/napalm_device_types/firewall.py +++ b/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. diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py index bb2599a..c59e15c 100644 --- a/napalm_device_types/models.py +++ b/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 diff --git a/napalm_device_types/residential_gateway.py b/napalm_device_types/residential_gateway.py index ff42431..5850752 100644 --- a/napalm_device_types/residential_gateway.py +++ b/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). diff --git a/tests/test_dhcp_diff_apply.py b/tests/test_dhcp_diff_apply.py new file mode 100644 index 0000000..fff2a18 --- /dev/null +++ b/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"]