Christian Manivong
fc4658f016
CI / test (3.10) (push) Successful in 31s
CI / test (3.11) (push) Successful in 29s
CI / test (3.12) (push) Successful in 31s
CI / test (3.10) (pull_request) Successful in 31s
CI / test (3.11) (pull_request) Successful in 28s
CI / test (3.12) (pull_request) Successful in 30s
netOrk reached these through the private _get/_post (netOrk #1113); they are napalm-device-types 4.17's contracts now: - apply_updates passes the packages it is given (it dropped them) and get_update_progress reports the upgrade's state and log. - get_dns_forwards, set_dns_forward (an existing forward of the domain is changed, not doubled) and remove_dns_forward, each applied with a reconfigure; a refusal is raised. - get_ospf translates FRR's diagnostics into neutral terms. Requires napalm-device-types 4.17.0.
66 lines
2.8 KiB
Python
66 lines
2.8 KiB
Python
"""Unbound's domain forwards (napalm-device-types ``DnsForwardMixin``).
|
|
|
|
OPNsense keeps forwards and DNS-over-TLS servers in one model (``dot``); a
|
|
forward is a row of type ``forward``. Every change is applied with a
|
|
reconfigure before the method returns.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from napalm_device_types import DnsForwardMixin
|
|
|
|
_SETTINGS = "/api/unbound/settings"
|
|
_RECONFIGURE = "/api/unbound/service/reconfigure"
|
|
|
|
|
|
def _domain(name: str) -> str:
|
|
return name.strip().rstrip(".")
|
|
|
|
|
|
class OPNsenseResolverMixin(DnsForwardMixin):
|
|
"""Needs the driver's ``_post``."""
|
|
|
|
def _forward_rows(self) -> list[dict[str, Any]]:
|
|
rows = self._post(f"{_SETTINGS}/searchForward", {}).get("rows") or [] # type: ignore[attr-defined]
|
|
return [r for r in rows if r.get("type", "forward") == "forward"]
|
|
|
|
def _saved(self, result: dict[str, Any], what: str) -> None:
|
|
if result.get("result") not in ("saved", "deleted", None):
|
|
raise RuntimeError(f"Unbound refused {what}: {result.get('validations') or result}")
|
|
|
|
def get_dns_forwards(self) -> list[dict[str, Any]]:
|
|
"""Every forward Unbound has, enabled or not; DNS-over-TLS servers are not forwards."""
|
|
return [
|
|
{
|
|
"domain": _domain(r.get("domain", "")),
|
|
"server": r.get("server", ""),
|
|
"port": int(r.get("port") or 53),
|
|
"enabled": str(r.get("enabled", "1")) == "1",
|
|
}
|
|
for r in self._forward_rows()
|
|
]
|
|
|
|
def set_dns_forward(self, domain: str, server: str, port: int = 53) -> None:
|
|
"""Forward *domain* to *server*; an existing forward of the domain is changed."""
|
|
name = _domain(domain)
|
|
row = {"enabled": "1", "type": "forward", "domain": name, "server": server, "port": str(port)}
|
|
known = [r for r in self._forward_rows() if _domain(r.get("domain", "")) == name]
|
|
if known:
|
|
result = self._post(f"{_SETTINGS}/setForward/{known[0]['uuid']}", {"dot": row}) # type: ignore[attr-defined]
|
|
else:
|
|
result = self._post(f"{_SETTINGS}/addForward", {"dot": row}) # type: ignore[attr-defined]
|
|
self._saved(result or {}, f"the forward of {name}")
|
|
self._post(_RECONFIGURE) # type: ignore[attr-defined]
|
|
|
|
def remove_dns_forward(self, domain: str) -> None:
|
|
"""Delete every forward of *domain*; nothing to do when there is none."""
|
|
name = _domain(domain)
|
|
uuids = [r["uuid"] for r in self._forward_rows() if _domain(r.get("domain", "")) == name and r.get("uuid")]
|
|
if not uuids:
|
|
return
|
|
for uuid in uuids:
|
|
self._post(f"{_SETTINGS}/delForward/{uuid}") # type: ignore[attr-defined]
|
|
self._post(_RECONFIGURE) # type: ignore[attr-defined]
|