Compare commits
No commits in common. "78448d2bde85fda1718c64e6cd8983b4fcb20fe0" and "6b06cada38751648b9365fd27b850aa2b6d250a8" have entirely different histories.
78448d2bde
...
6b06cada38
3 changed files with 21 additions and 139 deletions
|
|
@ -1,4 +1,5 @@
|
|||
import logging
|
||||
import re
|
||||
from typing import Any, Literal
|
||||
|
||||
import dns.rdata
|
||||
|
|
@ -90,16 +91,6 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
|
|||
|
||||
return absolute_value
|
||||
|
||||
def _escape_semicolon(self, value: str) -> str:
|
||||
fixed = value.replace(";", "\\;")
|
||||
self.log.debug(f"in='{value}', escaped='{fixed}'")
|
||||
return fixed
|
||||
|
||||
def _unescape_semicolon(self, value: str) -> str:
|
||||
fixed = value.replace("\\\\", "\\").replace("\\;", ";")
|
||||
self.log.debug(f"in='{value}', unescaped='{fixed}'")
|
||||
return fixed
|
||||
|
||||
def _get_nb_view(self, view: str | None | Literal[False]) -> dict[str, int | str]:
|
||||
"""get the correct netbox view when requested
|
||||
|
||||
|
|
@ -205,7 +196,7 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
|
|||
}
|
||||
|
||||
case "SPF" | "TXT":
|
||||
value = self._escape_semicolon(rcd_value)
|
||||
value = re.sub(r"\\*;", "\\;", rcd_value)
|
||||
|
||||
case "SRV":
|
||||
value = {
|
||||
|
|
@ -245,8 +236,8 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
|
|||
zone_id=nb_zone.id, status="active"
|
||||
)
|
||||
for nb_record in nb_records:
|
||||
rcd_name: str = "" if nb_record.name == "@" else nb_record.name
|
||||
rcd_value: str = nb_record.zone.name if nb_record.value == "@" else nb_record.value
|
||||
rcd_name: str = nb_record.name if nb_record.name != "@" else ""
|
||||
rcd_value: str = nb_record.value if nb_record.value != "@" else nb_record.zone.name
|
||||
rcd_type: str = nb_record.type
|
||||
rcd_ttl: int = nb_record.ttl or nb_zone.default_ttl
|
||||
if nb_record.type == "NS":
|
||||
|
|
@ -258,10 +249,11 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
|
|||
"ttl": rcd_ttl,
|
||||
"values": [],
|
||||
}
|
||||
self.log.debug(f"working on record={rcd_data} -> {rcd_value}")
|
||||
|
||||
self.log.debug(f"record data={rcd_data}")
|
||||
|
||||
try:
|
||||
rcd_rdata = self._format_rdata(rcd_type, rcd_value)
|
||||
rcd_rdata = self._format_rdata(nb_record, rcd_value)
|
||||
except NotImplementedError:
|
||||
continue
|
||||
|
||||
|
|
@ -270,8 +262,6 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
|
|||
|
||||
records[(rcd_name, rcd_type)]["values"].append(rcd_rdata)
|
||||
|
||||
self.log.debug(f"record data={records[(rcd_name, rcd_type)]}")
|
||||
|
||||
return list(records.values())
|
||||
|
||||
def populate(
|
||||
|
|
@ -339,11 +329,6 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
|
|||
|
||||
return True
|
||||
|
||||
def _unescape_for_netbox(self, rcd_type: str, value: str) -> str:
|
||||
if rcd_type not in ["TXT", "SPF"]:
|
||||
return value
|
||||
return self._unescape_semicolon(value)
|
||||
|
||||
def _apply(self, plan: octodns.provider.plan.Plan) -> None:
|
||||
"""apply the changes to the NetBox DNS zone.
|
||||
|
||||
|
|
@ -362,15 +347,17 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
|
|||
|
||||
new_changeset = self._format_changeset(change.new)
|
||||
for record in new_changeset:
|
||||
self.log.debug(f"ADD {change.new._type} {rcd_name} {record}")
|
||||
self.api.plugins.netbox_dns.records.create(
|
||||
zone=nb_zone.id,
|
||||
name=rcd_name,
|
||||
type=change.new._type,
|
||||
ttl=change.new.ttl,
|
||||
value=self._unescape_for_netbox(change.new._type, record),
|
||||
disable_ptr=self.disable_ptr,
|
||||
nb_record: pynetbox.core.response.Record = (
|
||||
self.api.plugins.netbox_dns.records.create(
|
||||
zone=nb_zone.id,
|
||||
name=rcd_name,
|
||||
type=change.new._type,
|
||||
ttl=change.new.ttl,
|
||||
value=re.sub(r"\\*;", ";", record),
|
||||
disable_ptr=self.disable_ptr,
|
||||
)
|
||||
)
|
||||
self.log.debug(f"ADD {nb_record.type} {nb_record.name} {nb_record.value}")
|
||||
|
||||
case octodns.record.Delete():
|
||||
nb_records: pynetbox.core.response.RecordSet = (
|
||||
|
|
@ -414,19 +401,19 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
|
|||
)
|
||||
nb_record.delete()
|
||||
if nb_record.value in to_update:
|
||||
nb_record.ttl = change.new.ttl
|
||||
self.log.debug(
|
||||
f"MODIFY {nb_record.type} {nb_record.name} {nb_record.value}"
|
||||
)
|
||||
nb_record.ttl = change.new.ttl
|
||||
nb_record.save()
|
||||
|
||||
for record in to_create:
|
||||
self.log.debug(f"ADD {change.new._type} {rcd_name} {record}")
|
||||
nb_record = self.api.plugins.netbox_dns.records.create(
|
||||
zone=nb_zone.id,
|
||||
name=rcd_name,
|
||||
type=change.new._type,
|
||||
ttl=change.new.ttl,
|
||||
value=self._unescape_for_netbox(change.new._type, record),
|
||||
value=re.sub(r"\\*;", ";", record),
|
||||
disable_ptr=self.disable_ptr,
|
||||
)
|
||||
self.log.debug(f"ADD {nb_record.type} {nb_record.name} {nb_record.value}")
|
||||
|
|
|
|||
|
|
@ -1,75 +0,0 @@
|
|||
from octodns_netbox_dns import NetBoxDNSProvider
|
||||
|
||||
|
||||
DEFAULT_CONFIG = {
|
||||
"id": 1,
|
||||
"url": "https://localhost:8000",
|
||||
"token": "",
|
||||
"view": False,
|
||||
"replace_duplicates": False,
|
||||
"make_absolute": True,
|
||||
}
|
||||
|
||||
|
||||
def test_escape1():
|
||||
nbdns = NetBoxDNSProvider(**DEFAULT_CONFIG)
|
||||
rcd_value = r"v=TLSRPTv1; rua=mailto:tlsrpt@example.com"
|
||||
value = nbdns._escape_semicolon(rcd_value)
|
||||
|
||||
assert value == r"v=TLSRPTv1\; rua=mailto:tlsrpt@example.com"
|
||||
|
||||
|
||||
def test_escape2():
|
||||
nbdns = NetBoxDNSProvider(**DEFAULT_CONFIG)
|
||||
rcd_value = r"v=TLSRPTv1\; rua=mailto:tlsrpt@example.com"
|
||||
value = nbdns._escape_semicolon(rcd_value)
|
||||
|
||||
assert value == r"v=TLSRPTv1\\; rua=mailto:tlsrpt@example.com"
|
||||
|
||||
|
||||
def test_escape3():
|
||||
nbdns = NetBoxDNSProvider(**DEFAULT_CONFIG)
|
||||
rcd_value = r"t=y\;o=~\;"
|
||||
value = nbdns._escape_semicolon(rcd_value)
|
||||
|
||||
assert value == r"t=y\\;o=~\\;"
|
||||
|
||||
|
||||
def test_escape4():
|
||||
nbdns = NetBoxDNSProvider(**DEFAULT_CONFIG)
|
||||
rcd_value = r"t=y;o=~;"
|
||||
value = nbdns._escape_semicolon(rcd_value)
|
||||
|
||||
assert value == r"t=y\;o=~\;"
|
||||
|
||||
|
||||
def test_unescape1():
|
||||
nbdns = NetBoxDNSProvider(**DEFAULT_CONFIG)
|
||||
rcd_value = r"v=TLSRPTv1\; rua=mailto:tlsrpt@example.com"
|
||||
value = nbdns._unescape_semicolon(rcd_value)
|
||||
|
||||
assert value == r"v=TLSRPTv1; rua=mailto:tlsrpt@example.com"
|
||||
|
||||
|
||||
def test_unescape2():
|
||||
nbdns = NetBoxDNSProvider(**DEFAULT_CONFIG)
|
||||
rcd_value = r"v=TLSRPTv1\\; rua=mailto:tlsrpt@example.com"
|
||||
value = nbdns._unescape_semicolon(rcd_value)
|
||||
|
||||
assert value == r"v=TLSRPTv1; rua=mailto:tlsrpt@example.com"
|
||||
|
||||
|
||||
def test_unescape3():
|
||||
nbdns = NetBoxDNSProvider(**DEFAULT_CONFIG)
|
||||
rcd_value = r"t=y\\;o=~\;"
|
||||
value = nbdns._unescape_semicolon(rcd_value)
|
||||
|
||||
assert value == r"t=y;o=~;"
|
||||
|
||||
|
||||
def test_unescape4():
|
||||
nbdns = NetBoxDNSProvider(**DEFAULT_CONFIG)
|
||||
rcd_value = r"t=y;o=~;"
|
||||
value = nbdns._unescape_semicolon(rcd_value)
|
||||
|
||||
assert value == r"t=y;o=~;"
|
||||
|
|
@ -56,37 +56,7 @@ def test_txt2():
|
|||
rcd_value = r"v=TLSRPTv1\; rua=mailto:tlsrpt@example.com"
|
||||
value = nbdns._format_rdata(rcd_type, rcd_value)
|
||||
|
||||
assert value == r"v=TLSRPTv1\\; rua=mailto:tlsrpt@example.com"
|
||||
|
||||
|
||||
def test_txt3():
|
||||
nbdns = NetBoxDNSProvider(**DEFAULT_CONFIG)
|
||||
rcd_type = "TXT"
|
||||
rcd_value = r"v=DKIM1; k=rsa; p=/0f+sikE+k9ZKbn1BJu0/soWht/+Zd/nc/+Gy//mQ1B5sCKYKgAmYTSWkxRjFzkc6KAQhi+/IzaFogEV050wcscdC8Rc8lAQzDUFrMs2ZZK1vFtkwIDAQAB"
|
||||
value = nbdns._format_rdata(rcd_type, rcd_value)
|
||||
|
||||
assert (
|
||||
value
|
||||
== r"v=DKIM1\; k=rsa\; p=/0f+sikE+k9ZKbn1BJu0/soWht/+Zd/nc/+Gy//mQ1B5sCKYKgAmYTSWkxRjFzkc6KAQhi+/IzaFogEV050wcscdC8Rc8lAQzDUFrMs2ZZK1vFtkwIDAQAB"
|
||||
)
|
||||
|
||||
|
||||
def test_txt4():
|
||||
nbdns = NetBoxDNSProvider(**DEFAULT_CONFIG)
|
||||
rcd_type = "TXT"
|
||||
rcd_value = r"t=y\;o=~\;"
|
||||
value = nbdns._format_rdata(rcd_type, rcd_value)
|
||||
|
||||
assert value == r"t=y\\;o=~\\;"
|
||||
|
||||
|
||||
def test_txt5():
|
||||
nbdns = NetBoxDNSProvider(**DEFAULT_CONFIG)
|
||||
rcd_type = "TXT"
|
||||
rcd_value = r"t=y;o=~;"
|
||||
value = nbdns._format_rdata(rcd_type, rcd_value)
|
||||
|
||||
assert value == r"t=y\;o=~\;"
|
||||
assert value == r"v=TLSRPTv1\; rua=mailto:tlsrpt@example.com"
|
||||
|
||||
|
||||
def test_srv():
|
||||
|
|
|
|||
Loading…
Reference in a new issue