Share zone-building logic between AXFR and zone-file export

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.

netbox_dns_bridge/request_handler.py

@@ -17,6 +17,7 @@ from django.conf import settings
 from netbox_dns.models import Zone, Record, View
 from netbox_dns_bridge import catalog_zone_manager as catzm
 from .models import SeenTransferClients
+from .utils import build_dns_zone
 
 LOGGER = get_logger(__name__)
 SETTINGS = settings.PLUGINS_CONFIG["netbox_dns_bridge"]
@@ -37,83 +38,7 @@ class DNSBaseRequestHandler(socketserver.BaseRequestHandler):
                 view__name=view_name,
                 active=True,
             )
-            # Build DNS zone
+            return build_dns_zone(nb_zone)
-            zone = dns.zone.Zone(zone_name, dns.name.root)
-            zone.rdclass = dns.rdataclass.IN
-
-            nb_records = Record.objects.filter(zone=nb_zone, 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
-
         except Zone.DoesNotExist:
             return None
         except OperationalError as e:

netbox_dns_bridge/utils.py

@@ -1,34 +1,99 @@
+import dns.name
 import dns.zone
 import dns.rdatatype
 import dns.rdataclass
-import dns.exception
+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)
-    # Create dnspython zone object
-    dp_zone = dns.zone.Zone(origin=dns.name.from_text(nb_zone.name))
-
-    for record in nb_zone.records.all():
-        rname = record.name or "@"
-        rdclass = dns.rdataclass.IN
-        rdatatype = dns.rdatatype.from_text(record.type)
-
-        # Create rdata object
-        rdata = dns.rdata.from_text(rdclass, rdatatype, record.value)
-
-        # Create rdataset
-        rdataset = dns.rdataset.Rdataset(rdclass, rdatatype)
-        rdataset.add(rdata, record.ttl)
-
-        # Add to zone
-        node = dp_zone.find_node(rname, create=True)
-        node.rdatasets.append(rdataset)
 
     # Write zone to file in BIND format
     try:
         with open(file_path, "w") as f:
-            dp_zone.to_file(f, sorted=True)
+            zone.to_file(f, sorted=True)
     except IOError as e:
         raise IOError(f"Failed to write zone file to {file_path}: {e}")