From 8ba95a07092742442aae639b72ece026ee250e86 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Thu, 20 Aug 2026 07:21:55 +0700 Subject: [PATCH] feat(dhcp): implement subnet get/apply/commit against Kea DHCPv4 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part of netork#85. Fills in the three device-specific methods the new DhcpServerMixin subnet layer expects. searchSubnet only carries uuid/subnet/description, so get_dhcp_subnets follows each row with getSubnet for the option data. That is one request per subnet; a firewall serves a handful, so the round trips cost less than the reconfigure they help avoid. An option Kea does not carry is omitted rather than reported as empty, because the generic diff reads an absent key as "not managed" — reporting [] would make every unmanaged option look like a pending change. apply_dhcp_subnet honours the mixin's partial-update contract: on an update it reads the subnet's current options first and replaces only the named ones. Without that, managing domain_search alone would blank the routers Kea autocollected and strand every client on that VLAN without a gateway. Setting any option also forces option_data_autocollect off — left on, Kea keeps re-filling routers/DNS/NTP and the next diff sees a change again, which is a reconfigure loop rather than a converged state. OPNsense renders repeatable option fields as comma-separated strings in some versions and as a selection map in others, for the same logical field. Both shapes are accepted rather than pinning the driver to one release. Pools are a newline-separated text block. A subnet whose detail fetch fails is skipped with a log line instead of aborting, same rule as get_dhcp_reservations: one broken record must not make the whole inventory unreadable. 16 new tests. Not yet verified against a live device — no reachable OPNsense at the time of writing, same caveat the reservation support shipped with. --- napalm_opnsense/opnsense.py | 175 +++++++++++++++++++++++++++ tests/unit/test_driver.py | 234 ++++++++++++++++++++++++++++++++++++ 2 files changed, 409 insertions(+) diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index a47b4a7..89dae3d 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -1568,6 +1568,181 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver): self._post("/api/kea/service/reconfigure") return {"success": True} + # ── Kea DHCPv4 subnets ──────────────────────────────────────────────── + + # OPNsense renders repeatable option fields ("AsList") as comma-separated + # strings, and the pool list as a newline-separated text block. Which of + # the two shapes a given field uses is a property of the model, not of the + # request, so the mapping is a constant rather than something to sniff. + _KEA_LIST_OPTIONS = ( + "routers", + "domain_name_servers", + "domain_search", + "ntp_servers", + ) + _KEA_SCALAR_OPTIONS = ("domain_name",) + + @staticmethod + def _kea_read_list(raw: Any) -> list[str]: + """Read an OPNsense list field, in either shape it comes back in. + + Plain string: ``"a,b"``. Selection map: ``{"a": {"selected": 1}, ...}`` + -- which OPNsense uses for some model field types and versions. Both + appear in the wild for the same logical field, so both are accepted + rather than pinning the driver to one OPNsense release. + """ + if isinstance(raw, dict): + return [ + str(key) + for key, meta in raw.items() + if isinstance(meta, dict) and str(meta.get("selected", 0)) == "1" + ] + if isinstance(raw, list): + return [str(item).strip() for item in raw if str(item).strip()] + return [part.strip() for part in str(raw or "").split(",") if part.strip()] + + @staticmethod + def _kea_read_scalar(raw: Any) -> str: + if isinstance(raw, dict): + selected = [ + str(key) + for key, meta in raw.items() + if isinstance(meta, dict) and str(meta.get("selected", 0)) == "1" + ] + return selected[0] if selected else "" + return str(raw or "").strip() + + def get_dhcp_subnets(self) -> list[dict[str, Any]]: + """Return every Kea DHCPv4 subnet with its pools and options. + + ``searchSubnet`` only carries uuid/subnet/description, so each row is + followed by a ``getSubnet`` call for the option data. That is one + request per subnet; a firewall serves a handful of them, so the extra + round trips cost less than the reconfigure they help avoid. + + An option Kea does not carry is *omitted* from ``option_data`` rather + than reported as empty. The generic diff treats an absent key as "not + managed", so reporting ``[]`` here would make every unmanaged option + look like a pending change and reload the daemon on every run. + + A subnet whose detail fetch fails is skipped with a log line instead + of aborting: one broken record must not make the whole inventory + unreadable, same rule as ``get_dhcp_reservations()``. + + :raises RuntimeError: if the Kea plugin isn't installed/enabled. + """ + rows = self._kea_subnets() + + subnets: list[dict[str, Any]] = [] + for row in rows: + uuid = str(row.get("uuid") or "") + if not uuid: + continue + try: + detail = self._get(f"/api/kea/dhcpv4/getSubnet/{uuid}") + except Exception as exc: + logger.warning("Kea subnet %s detail fetch failed, skipping: %s", uuid, exc) + continue + + record = detail.get("subnet") or {} + raw_options = record.get("option_data") or {} + + option_data: dict[str, Any] = {} + for name in self._KEA_LIST_OPTIONS: + values = self._kea_read_list(raw_options.get(name)) + if values: + option_data[name] = values + for name in self._KEA_SCALAR_OPTIONS: + value = self._kea_read_scalar(raw_options.get(name)) + if value: + option_data[name] = value + + pools = [ + line.strip() + for line in str(record.get("pools") or "").splitlines() + if line.strip() + ] + + subnets.append({ + "uuid": uuid, + "subnet": str(record.get("subnet") or row.get("subnet") or ""), + "description": str(record.get("description") or ""), + "pools": pools, + "option_data": option_data, + "match_client_id": str(record.get("match-client-id") or "0") == "1", + }) + return subnets + + def apply_dhcp_subnet( + self, subnet: dict[str, Any], *, uuid: str | None = None + ) -> dict[str, Any]: + """Create or update a single Kea DHCPv4 subnet. + + ``option_data`` is a partial update, as the mixin contract requires: + on an update the subnet's current options are read first and only the + named ones are replaced. Without that, managing `domain_search` alone + would blank the `routers` Kea autocollected and strand every client on + that VLAN without a gateway. + + Setting any option explicitly also turns ``option_data_autocollect`` + off. Left on, Kea keeps re-filling routers/DNS/NTP from the interface + and the next diff sees a change again -- a reconfigure loop rather + than a converged state. + + Does not reconfigure the service -- call ``commit_dhcp_subnets()`` + once after a batch. + + :raises RuntimeError: if Kea rejects the write. + """ + desired_options = dict(subnet.get("option_data") or {}) + + merged_options: dict[str, str] = {} + if uuid and desired_options: + try: + current = self._get(f"/api/kea/dhcpv4/getSubnet/{uuid}") + raw = (current.get("subnet") or {}).get("option_data") or {} + except Exception as exc: + raise RuntimeError( + f"Kea subnet {uuid} could not be read before update: {exc}" + ) from exc + for name in self._KEA_LIST_OPTIONS: + values = self._kea_read_list(raw.get(name)) + if values: + merged_options[name] = ",".join(values) + for name in self._KEA_SCALAR_OPTIONS: + value = self._kea_read_scalar(raw.get(name)) + if value: + merged_options[name] = value + + for name, value in desired_options.items(): + merged_options[name] = ",".join(value) if isinstance(value, list) else str(value) + + record: dict[str, Any] = {"subnet": subnet.get("subnet", "")} + if subnet.get("description") is not None: + record["description"] = subnet.get("description") or "" + if subnet.get("pools") is not None: + record["pools"] = "\n".join(subnet.get("pools") or []) + if subnet.get("match_client_id") is not None: + record["match-client-id"] = "1" if subnet.get("match_client_id") else "0" + if desired_options: + record["option_data"] = merged_options + record["option_data_autocollect"] = "0" + + path = ( + f"/api/kea/dhcpv4/setSubnet/{uuid}" if uuid else "/api/kea/dhcpv4/addSubnet" + ) + result = self._post(path, {"subnet": record}) + if result.get("result") != "saved": + raise RuntimeError( + f"Kea rejected DHCP subnet {subnet.get('subnet')}: {result}" + ) + return {"success": True} + + def commit_dhcp_subnets(self) -> dict[str, Any]: + """Reload Kea so pending subnet writes take effect.""" + self._post("/api/kea/service/reconfigure") + return {"success": True} + def get_services(self) -> list[dict[str, Any]]: """Return running services from OPNsense. diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 0932749..906b789 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -2127,3 +2127,237 @@ class TestCommitDhcpReservations: assert calls == ["/api/kea/service/reconfigure"] assert result == {"success": True} + + +# ── Kea DHCPv4 subnets ──────────────────────────────────────────────────────── + +KEA_SUBNET_SEARCH_RESPONSE = { + "rows": [ + {"uuid": "sub-home", "subnet": "10.10.20.0/24", "description": "Home"}, + {"uuid": "sub-guest", "subnet": "10.10.40.0/24", "description": "Guest"}, + ] +} + +# getSubnet wraps the record and, unlike searchSubnet, carries the options. +# OPNsense renders AsList fields as comma-separated strings. +KEA_SUBNET_DETAIL_RESPONSE = { + "subnet": { + "subnet": "10.10.20.0/24", + "description": "Home", + "pools": "10.10.20.100-10.10.20.200", + "match-client-id": "1", + "option_data": { + "routers": "10.10.20.1", + "domain_name_servers": "10.10.20.1,10.10.20.2", + "domain_name": "home.local", + "domain_search": "home.local,office.local", + "ntp_servers": "", + }, + } +} + + +class TestGetDhcpSubnets: + def _wire(self, driver, detail=None): + driver._get = lambda path: ( + KEA_SUBNET_SEARCH_RESPONSE + if "searchSubnet" in path + else (detail or KEA_SUBNET_DETAIL_RESPONSE) + ) + + def test_maps_kea_subnet_to_vendor_neutral_dict(self, driver): + self._wire(driver) + + result = driver.get_dhcp_subnets() + + assert result[0] == { + "uuid": "sub-home", + "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", "10.10.20.2"], + "domain_name": "home.local", + "domain_search": ["home.local", "office.local"], + }, + "match_client_id": True, + } + + def test_fetches_detail_per_subnet(self, driver): + paths = [] + driver._get = lambda path: ( + paths.append(path) + or (KEA_SUBNET_SEARCH_RESPONSE if "searchSubnet" in path else KEA_SUBNET_DETAIL_RESPONSE) + ) + + driver.get_dhcp_subnets() + + assert "/api/kea/dhcpv4/getSubnet/sub-home" in paths + assert "/api/kea/dhcpv4/getSubnet/sub-guest" in paths + + def test_empty_option_is_omitted_not_reported_as_empty(self, driver): + """An option Kea does not carry must read as "unset", not as [].""" + self._wire(driver) + result = driver.get_dhcp_subnets() + assert "ntp_servers" not in result[0]["option_data"] + + def test_multiline_pools_become_a_list(self, driver): + detail = { + "subnet": { + "subnet": "10.10.20.0/24", + "description": "", + "pools": "10.10.20.100-10.10.20.150\n10.10.20.180-10.10.20.200", + "option_data": {}, + } + } + self._wire(driver, detail) + assert driver.get_dhcp_subnets()[0]["pools"] == [ + "10.10.20.100-10.10.20.150", + "10.10.20.180-10.10.20.200", + ] + + def test_selection_map_shape_is_accepted(self, driver): + """Some OPNsense versions render list fields as a selection map.""" + detail = { + "subnet": { + "subnet": "10.10.20.0/24", + "description": "", + "pools": "", + "option_data": { + "domain_search": { + "home.local": {"value": "home.local", "selected": 1}, + "office.local": {"value": "office.local", "selected": 0}, + } + }, + } + } + self._wire(driver, detail) + assert driver.get_dhcp_subnets()[0]["option_data"]["domain_search"] == ["home.local"] + + def test_a_subnet_whose_detail_fails_is_skipped_not_fatal(self, driver): + def _get(path): + if "searchSubnet" in path: + return KEA_SUBNET_SEARCH_RESPONSE + if "sub-home" in path: + raise RuntimeError("boom") + return KEA_SUBNET_DETAIL_RESPONSE + + driver._get = _get + result = driver.get_dhcp_subnets() + assert [s["uuid"] for s in result] == ["sub-guest"] + + def test_missing_plugin_raises(self, driver): + def _get(path): + raise RuntimeError("404") + + driver._get = _get + with pytest.raises(RuntimeError, match="Kea DHCPv4 plugin unavailable"): + driver.get_dhcp_subnets() + + +class TestApplyDhcpSubnet: + def test_update_uses_set_endpoint_with_uuid(self, driver): + calls = [] + driver._get = lambda path: KEA_SUBNET_DETAIL_RESPONSE + driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"} + + driver.apply_dhcp_subnet( + {"subnet": "10.10.20.0/24", "option_data": {"domain_search": ["a.local"]}}, + uuid="sub-home", + ) + + assert calls[0][0] == "/api/kea/dhcpv4/setSubnet/sub-home" + + def test_create_uses_add_endpoint(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"} + + driver.apply_dhcp_subnet({"subnet": "10.10.99.0/24", "option_data": {}}) + + assert calls[0][0] == "/api/kea/dhcpv4/addSubnet" + + def test_unnamed_options_are_preserved_on_update(self, driver): + """The whole point of the partial-update contract. + + Kea autocollects routers/domain_name_servers. Setting only + domain_search must not blank them, or every managed subnet loses its + gateway the first time netOrk touches it. + """ + calls = [] + driver._get = lambda path: KEA_SUBNET_DETAIL_RESPONSE + driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"} + + driver.apply_dhcp_subnet( + {"subnet": "10.10.20.0/24", "option_data": {"domain_search": ["new.local"]}}, + uuid="sub-home", + ) + + sent = calls[0][1]["subnet"]["option_data"] + assert sent["domain_search"] == "new.local" + assert sent["routers"] == "10.10.20.1" + assert sent["domain_name_servers"] == "10.10.20.1,10.10.20.2" + + def test_lists_are_serialised_comma_separated(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"} + + driver.apply_dhcp_subnet( + { + "subnet": "10.10.99.0/24", + "option_data": {"domain_search": ["a.local", "b.local"]}, + } + ) + + assert calls[0][1]["subnet"]["option_data"]["domain_search"] == "a.local,b.local" + + def test_pools_are_serialised_newline_separated(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"} + + driver.apply_dhcp_subnet( + { + "subnet": "10.10.99.0/24", + "pools": ["10.10.99.10-10.10.99.20", "10.10.99.30-10.10.99.40"], + "option_data": {}, + } + ) + + assert ( + calls[0][1]["subnet"]["pools"] + == "10.10.99.10-10.10.99.20\n10.10.99.30-10.10.99.40" + ) + + def test_explicit_option_data_disables_autocollect(self, driver): + """Kea would otherwise re-fill routers/DNS/NTP and fight the desired state.""" + calls = [] + driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"} + + driver.apply_dhcp_subnet( + {"subnet": "10.10.99.0/24", "option_data": {"domain_search": ["a.local"]}} + ) + + assert calls[0][1]["subnet"]["option_data_autocollect"] == "0" + + def test_no_options_leaves_autocollect_alone(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append((path, data)) or {"result": "saved"} + + driver.apply_dhcp_subnet({"subnet": "10.10.99.0/24", "option_data": {}}) + + assert "option_data_autocollect" not in calls[0][1]["subnet"] + + def test_rejection_raises(self, driver): + driver._post = lambda path, data=None: {"result": "failed", "validations": {"x": "bad"}} + + with pytest.raises(RuntimeError, match="10.10.99.0/24"): + driver.apply_dhcp_subnet({"subnet": "10.10.99.0/24", "option_data": {}}) + + +class TestCommitDhcpSubnets: + def test_reconfigures_kea(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append(path) or {} + + assert driver.commit_dhcp_subnets() == {"success": True} + assert calls == ["/api/kea/service/reconfigure"]