Compare commits

...

2 commits

Author SHA1 Message Date
78448d2bde only unescape txt and spf records
All checks were successful
check code / check-code (push) Successful in 30s
2024-02-29 22:17:45 +01:00
ff1a47cf8c update logging 2024-02-29 22:14:00 +01:00

View file

@ -92,12 +92,12 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
def _escape_semicolon(self, value: str) -> str: def _escape_semicolon(self, value: str) -> str:
fixed = value.replace(";", "\\;") fixed = value.replace(";", "\\;")
self.log.debug(f"in='{value}', fixed='{fixed}'") self.log.debug(f"in='{value}', escaped='{fixed}'")
return fixed return fixed
def _unescape_semicolon(self, value: str) -> str: def _unescape_semicolon(self, value: str) -> str:
fixed = value.replace("\\\\", "\\").replace("\\;", ";") fixed = value.replace("\\\\", "\\").replace("\\;", ";")
self.log.debug(f"in='{value}', fixed='{fixed}'") self.log.debug(f"in='{value}', unescaped='{fixed}'")
return fixed return fixed
def _get_nb_view(self, view: str | None | Literal[False]) -> dict[str, int | str]: def _get_nb_view(self, view: str | None | Literal[False]) -> dict[str, int | str]:
@ -339,6 +339,11 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
return True 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: def _apply(self, plan: octodns.provider.plan.Plan) -> None:
"""apply the changes to the NetBox DNS zone. """apply the changes to the NetBox DNS zone.
@ -357,17 +362,15 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
new_changeset = self._format_changeset(change.new) new_changeset = self._format_changeset(change.new)
for record in new_changeset: for record in new_changeset:
nb_record: pynetbox.core.response.Record = ( self.log.debug(f"ADD {change.new._type} {rcd_name} {record}")
self.api.plugins.netbox_dns.records.create( self.api.plugins.netbox_dns.records.create(
zone=nb_zone.id, zone=nb_zone.id,
name=rcd_name, name=rcd_name,
type=change.new._type, type=change.new._type,
ttl=change.new.ttl, ttl=change.new.ttl,
value=self._unescape_semicolon(record), value=self._unescape_for_netbox(change.new._type, record),
disable_ptr=self.disable_ptr, disable_ptr=self.disable_ptr,
) )
)
self.log.debug(f"ADD {nb_record.type} {nb_record.name} {nb_record.value}")
case octodns.record.Delete(): case octodns.record.Delete():
nb_records: pynetbox.core.response.RecordSet = ( nb_records: pynetbox.core.response.RecordSet = (
@ -411,19 +414,19 @@ class NetBoxDNSProvider(octodns.provider.base.BaseProvider):
) )
nb_record.delete() nb_record.delete()
if nb_record.value in to_update: if nb_record.value in to_update:
nb_record.ttl = change.new.ttl
self.log.debug( self.log.debug(
f"MODIFY {nb_record.type} {nb_record.name} {nb_record.value}" f"MODIFY {nb_record.type} {nb_record.name} {nb_record.value}"
) )
nb_record.ttl = change.new.ttl
nb_record.save() nb_record.save()
for record in to_create: 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( nb_record = self.api.plugins.netbox_dns.records.create(
zone=nb_zone.id, zone=nb_zone.id,
name=rcd_name, name=rcd_name,
type=change.new._type, type=change.new._type,
ttl=change.new.ttl, ttl=change.new.ttl,
value=self._unescape_semicolon(record), value=self._unescape_for_netbox(change.new._type, record),
disable_ptr=self.disable_ptr, disable_ptr=self.disable_ptr,
) )
self.log.debug(f"ADD {nb_record.type} {nb_record.name} {nb_record.value}")