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"]