Sven Luethi
627a3a5712
export_bind_zone_file diverged from the AXFR path in four ways, each a bug: - It iterated nb_zone.records.all(), including inactive records, whereas the AXFR handler filtered Record.objects.filter(zone=nb_zone, active=True). Exported zone files could contain stale/withdrawn records. - It passed record.ttl straight to rdataset.add(), crashing with a TypeError when the NetBox DNS Record had no TTL set. The AXFR path used 'ttl = record.ttl or nb_zone.default_ttl'. - It had no TXT quoting logic, so TXT values containing spaces or semicolons were mis-tokenized by dns.rdata.from_text (a ';' was treated as the start of a comment). The AXFR path quotes and chunks TXT values for exactly this reason. - It appended a fresh rdataset to the node for every record instead of merging records that share a name and type, producing duplicate rdatasets and malformed zone files. Extract build_dns_zone(nb_zone) in utils.py containing the corrected, AXFR-proven record-building logic (active filter, TTL fallback, TXT quoting/chunking, rdataset merging via rdatasets_dict). Rewrite export_bind_zone_file to build the zone through build_dns_zone and then write it. Rewrite request_handler._get_zone_from_nb to look up the NetBox DNS Zone and delegate to build_dns_zone, keeping only the DoesNotExist/OperationalError handling that is specific to the query path. Both code paths now produce identical zone data.
100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
import dns.name
|
|
import dns.zone
|
|
import dns.rdatatype
|
|
import dns.rdataclass
|
|
import dns.rdataset
|
|
import netbox_dns.models
|
|
from .logger import get_logger
|
|
|
|
LOGGER = get_logger(__name__)
|
|
|
|
|
|
def build_dns_zone(nb_zone: netbox_dns.models.Zone) -> dns.zone.Zone:
|
|
# Build DNS zone
|
|
zone = dns.zone.Zone(nb_zone.name, dns.name.root)
|
|
zone.rdclass = dns.rdataclass.IN
|
|
|
|
nb_records = nb_zone.records.filter(active=True)
|
|
|
|
rdatasets_dict = {}
|
|
|
|
for record in nb_records:
|
|
rdtype = dns.rdatatype.from_text(record.type)
|
|
if not record.name:
|
|
name = zone.origin
|
|
elif record.name.endswith("."):
|
|
name = dns.name.from_text(record.name)
|
|
else:
|
|
name = dns.name.from_text(record.name, origin=zone.origin)
|
|
|
|
# If the record has no TTL, use the zone default
|
|
ttl = record.ttl or nb_zone.default_ttl
|
|
|
|
# Apply quoting for TXT records to stop tokanizer
|
|
# from cutting it up:
|
|
value = record.value
|
|
if rdtype == dns.rdatatype.TXT:
|
|
if value.startswith('"') and value.endswith('"'):
|
|
value = value[1:-1].replace('" "', "").replace('"', '"')
|
|
|
|
if len(value) > 255:
|
|
# This is a bug fix for netbox. If netbox allowed for
|
|
# an unquoted value to be larger then 255 characters,
|
|
# it misunderstood everything behind a ; as a comment.
|
|
chunks = [
|
|
'"{}"'.format(value[i : i + 255])
|
|
for i in range(0, len(value), 255)
|
|
]
|
|
value = " ".join(chunks)
|
|
else:
|
|
value = f'"{value}"'
|
|
|
|
rdata = dns.rdata.from_text(
|
|
dns.rdataclass.IN,
|
|
rdtype,
|
|
value,
|
|
relativize=False,
|
|
origin=zone.origin,
|
|
)
|
|
|
|
# Initialize rdataset if it doesn't exist for this name and type
|
|
if name not in rdatasets_dict:
|
|
rdatasets_dict[name] = {}
|
|
if rdtype not in rdatasets_dict[name]:
|
|
rdatasets_dict[name][rdtype] = dns.rdataset.Rdataset(
|
|
dns.rdataclass.IN, rdtype
|
|
)
|
|
|
|
# Add the rdata to the appropriate rdataset
|
|
rdatasets_dict[name][rdtype].add(rdata, ttl)
|
|
|
|
# Now, add all rdatasets to the zone
|
|
for name, rdtypes in rdatasets_dict.items():
|
|
for rdtype, rdataset in rdtypes.items():
|
|
# Ensure rdataset has the same rdclass as the zone
|
|
if rdataset.rdclass != zone.rdclass:
|
|
raise ValueError(
|
|
f"rdataset rdclass {rdataset.rdclass} does not match "
|
|
f"zone rdclass {zone.rdclass}"
|
|
)
|
|
|
|
# Check if the rdataset has any rdata before creating an RRset
|
|
if not rdataset:
|
|
LOGGER.debug(f"Skipping empty rdataset for {name} {rdtype}")
|
|
continue # Skip empty rdataset
|
|
|
|
# Replace the rdataset for the given name and type
|
|
zone.replace_rdataset(name, rdataset)
|
|
return zone
|
|
|
|
|
|
def export_bind_zone_file(nb_zone: netbox_dns.models.Zone, file_path: str):
|
|
zone = build_dns_zone(nb_zone)
|
|
|
|
# Write zone to file in BIND format
|
|
try:
|
|
with open(file_path, "w") as f:
|
|
zone.to_file(f, sorted=True)
|
|
except IOError as e:
|
|
raise IOError(f"Failed to write zone file to {file_path}: {e}")
|