diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index a9022c8..08f2bf0 100644 --- a/napalm_device_types/__init__.py +++ b/napalm_device_types/__init__.py @@ -36,7 +36,7 @@ Also provided: 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.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 @@ -60,5 +60,6 @@ __all__ = [ "StorageDriver", "SwitchDriver", "driver_supports_ping", + "normalize_cidr", "normalize_mac", ] diff --git a/napalm_device_types/dhcp.py b/napalm_device_types/dhcp.py index 1e31a09..8c002db 100644 --- a/napalm_device_types/dhcp.py +++ b/napalm_device_types/dhcp.py @@ -1,16 +1,24 @@ # -*- coding: utf-8 -*- -"""Static DHCP reservation management, shared by firewalls and gateways. +"""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 ``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". +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 @@ -22,6 +30,9 @@ 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 @@ -33,6 +44,15 @@ _DHCP_RESERVATION_COMPARE_FIELDS = ( "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]") @@ -58,11 +78,43 @@ def normalize_mac(mac: Optional[str]) -> str: 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 static DHCP reservation read/diff/apply. + """Mixin providing DHCP reservation and subnet read/diff/apply. Concrete drivers **must** provide ``get_dhcp_reservations()``, - ``apply_dhcp_reservation()`` and ``commit_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()``. """ # ------------------------------------------------------------------ @@ -101,6 +153,53 @@ class DhcpServerMixin: """ raise NotImplementedError + 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. + """ + raise NotImplementedError + + 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}``. + """ + raise NotImplementedError + + 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}``. + """ + raise NotImplementedError + def commit_dhcp_reservations(self) -> Dict[str, Any]: """ Applies pending reservation changes (e.g. Kea's ``service/reconfigure``, @@ -209,3 +308,91 @@ class DhcpServerMixin: 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)" + ) diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py index c59e15c..3e3a9fa 100644 --- a/napalm_device_types/models.py +++ b/napalm_device_types/models.py @@ -387,6 +387,60 @@ class DhcpReservationDiffDict(TypedDict): 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 diff --git a/tests/test_dhcp_subnet_diff_apply.py b/tests/test_dhcp_subnet_diff_apply.py new file mode 100644 index 0000000..8f9c863 --- /dev/null +++ b/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( + "call", + [ + lambda d: d.get_dhcp_subnets(), + lambda d: d.apply_dhcp_subnet({}), # type: ignore[typeddict-item] + lambda d: d.commit_dhcp_subnets(), + ], +) +def test_device_specific_methods_raise_not_implemented(call: Any) -> None: + with pytest.raises(NotImplementedError): + call(_BareDriver())