Renaming Project and changing license

netbox_dns_bridge/__init__.py

@@ -0,0 +1,36 @@
+import logging
+from netbox.plugins import PluginConfig
+from django.conf import settings
+
+__version__ = "1.0.8"
+
+logger = logging.getLogger(__name__)
+
+logging.basicConfig(
+    level=logging.DEBUG,  # Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",  # Log format
+    datefmt="%Y-%m-%d %H:%M:%S",  # Date format for timestamps
+)
+
+
+class DNSBridgeConfig(PluginConfig):
+    name = "netbox_dns_bridge"
+    verbose_name = "Netbox DNS Bridge"
+    description = ""
+    version = __version__
+    author = "Sven Luethi"
+    author_email = "dev@sven.luethi.co"
+    base_url = "dns-bridge"
+
+    def ready(self):
+        self.settings = settings.PLUGINS_CONFIG.get(self.name, None)
+        if not self.settings:
+            raise RuntimeError(
+                f"{self.name}: Plugin {self.verbose_name} failed to initialize due to missing settings. Terminating Netbox."
+            )
+
+        from . import signals as signals
+
+
+config = DNSBridgeConfig
+default_app_config = ".apps.NetboxDNSBridgeConfig"

netbox_dns_bridge/management/commands/dns-transfer-endpoint.py

@@ -0,0 +1,117 @@
+import logging
+import threading
+import dns.query
+import dns.message
+import dns.tsigkeyring
+import dns.name
+import dns.zone
+import dns.rdatatype
+import dns.rdataclass
+import dns.rdtypes
+import dns.exception
+import dns.renderer
+
+from django.core.management.base import BaseCommand
+from django.conf import settings
+from netbox_dns.models import View
+from netbox_dns_bridge.service.endpoint.request_handler import UDPRequestHandler, TCPRequestHandler
+from netbox_dns_bridge.service.endpoint.dns_server import UDPDNSServer, TCPDNSServer
+from netbox_dns_bridge.service.endpoint import catalog_zone_manager as catzm
+
+logger = logging.getLogger("dns-transfer-endpoint")
+
+class Command(BaseCommand):
+    help = "Run a minimal AXFR DNS server using data from NetBox DNS plugin"
+
+    def load_settings(self):
+        self.settings = settings.PLUGINS_CONFIG.get(
+            "netbox_dns_bridge", None
+        )
+        if not self.settings:
+            raise RuntimeError(
+                "Command failed to initialize due to missing settings. Terminating Netbox."
+            )
+
+        self.tsig_keys = self.settings.get("tsig_keys", None)
+        if not self.tsig_keys:
+            raise RuntimeError("tsig_keys variable not set in plugin settings.")
+
+    # Load TSIG keys and map them to views
+    def load_tsig_key_settings(self):
+        self.keyring = {}
+        self.tsig_view_map = {}
+
+        for view_name, data in self.tsig_keys.items():
+            raw_key_name = data.get("keyname")
+            secret = data.get("secret")
+            algorithm_str = data.get("algorithm", "hmac-sha256")
+
+            if not raw_key_name or not secret:
+                logger.error(
+                    f"Skipping TSIG key for view {view_name}: missing keyname or secret."
+                )
+                continue
+
+            try:
+                nb_view = View.objects.get(name=view_name)
+            except View.DoesNotExist:
+                logger.error(
+                    f"Skipping TSIG key {raw_key_name}: view '{view_name}' not found."
+                )
+                continue
+
+            # Normalize key name to absolute DNS name
+            key_name_obj = dns.name.from_text(raw_key_name, origin=None).canonicalize()
+            if not key_name_obj.is_absolute():
+                key_name_obj = key_name_obj.concatenate(dns.name.root)
+            key_name_str = key_name_obj.to_text()  # Will always include trailing do
+
+            self.keyring[key_name_obj] = dns.tsig.Key(
+                name=key_name_obj, secret=secret, algorithm=algorithm_str
+            )
+            self.tsig_view_map[key_name_str] = nb_view
+            logger.debug(f"Loaded TSIG key: {key_name_str} view: {nb_view.name}")
+
+        if not self.keyring:
+            msg = "No TSIG keys found in database."
+            logger.critical(msg)
+            raise RuntimeError(msg)
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--port", type=int, default=5354, help="Port number to listen on"
+        )
+        parser.add_argument(
+            "--address", type=str, default="0.0.0.0", help="IP to bind to"
+        )
+
+    def handle(self, *args, **options):
+        # Load parameters
+        port = options["port"]
+        address = options["address"]
+        catzm.init()
+
+        # Initialize settings
+        self.load_settings()
+        self.load_tsig_key_settings()
+
+        udp_server = UDPDNSServer(
+            (address, port), UDPRequestHandler, self.keyring, self.tsig_view_map
+        )
+
+        tcp_server = TCPDNSServer(
+            (address, port), TCPRequestHandler, self.keyring, self.tsig_view_map
+        )
+
+        def run_udp_server(server):
+            logger.debug(f"Query endpoint listening on {address} udp/{port}")
+            server.serve_forever()
+
+        udp_thread = threading.Thread(
+            target=run_udp_server, args=(udp_server,), daemon=True
+        )
+
+        udp_thread.start()
+
+        logger.debug(f"Query endpoint listening on {address} tcp/{port}")
+        tcp_server.serve_forever()

netbox_dns_bridge/management/commands/export-all-dns-zones.py

@@ -0,0 +1,47 @@
+import os
+import dns.zone
+import dns.rdatatype
+import dns.rdataclass
+import dns.exception
+import netbox_dns.models
+from netbox_dns_bridge.utils import export_bind_zone_file
+from django.core.management.base import BaseCommand, CommandError
+
+class Command(BaseCommand):
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--path", type=str, help="Path to the directory where zone files are to be written"
+        )
+
+    def handle(self, *args, **options):
+        # Load parameters
+        base_path = options['path']
+
+        if not base_path:
+            raise CommandError("No --path parameter given")
+
+        # Ensure base_path exists
+        os.makedirs(base_path, exist_ok=True)
+
+        # Iterate through all zones
+        nb_zones = netbox_dns.models.Zone.objects.prefetch_related("view").all()
+        if not nb_zones.exists():
+            self.stdout.write(self.style.WARNING("No zones found in NetBox DNS."))
+            return
+
+        for zone in nb_zones:
+            # Determine view directory
+            view_name = zone.view.name
+            view_dir = os.path.join(base_path, view_name)
+            os.makedirs(view_dir, exist_ok=True)
+
+            # Zone file path
+            file_path = os.path.join(view_dir, zone.name)
+
+            try:
+                # Call existing export function
+                export_bind_zone_file(zone, file_path=file_path)
+                self.stdout.write(self.style.SUCCESS(f"Exported zone '{zone.name}' to '{file_path}'"))
+            except Exception as e:
+                self.stderr.write(f"Failed to export zone '{zone.name}': {e}")

netbox_dns_bridge/management/commands/export-dns-zone.py

@@ -0,0 +1,56 @@
+import os
+import dns.zone
+import dns.rdatatype
+import dns.rdataclass
+import dns.exception
+import netbox_dns.models
+from netbox_dns_bridge.utils import export_bind_zone_file
+from django.core.management.base import BaseCommand, CommandError
+
+class Command(BaseCommand):
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--view", type=str, help="The name of the view the Zone to be exported is in"
+        )
+        parser.add_argument(
+            "--zone", type=str, help="The FQDN of the Zone to be exported"
+        )
+        parser.add_argument(
+            "--file", type=str, help="Path of the zone file to be written"
+        )
+
+
+    def handle(self, *args, **options):
+        #if len(options) < 2:
+        #    print("export-zone <zone name> <file path>")
+        #    sys.exit(1)
+
+        # Load parameters
+        view_name = options['view']
+        zone_name = options['zone']
+        file_path = options['file']
+
+        if not view_name:
+            raise CommandError("No --view parameter given")
+        elif not zone_name:
+            raise CommandError("No --zone parameter given")
+        elif not file_path:
+            raise CommandError("No --file parameter given")
+
+        try:
+            # Load the zone from NetBox DNS
+            nb_zone = netbox_dns.models.Zone.objects.get(view__name=view_name, name=zone_name)
+
+            if nb_zone:
+                export_bind_zone_file(nb_zone, file_path=file_path)
+            else:
+                print("Zone not found in Netbox. Aborting")
+                sys.exit(1)
+
+        except Exception as e:
+            raise CommandError(f"Failed to export zone: {e}")
+
+        self.stdout.write(self.style.SUCCESS(f"Zone '{zone_name}' exported to '{file_path}'"))
+
+

netbox_dns_bridge/models.py

@@ -0,0 +1,31 @@
+from django.db import models
+import netbox.models
+import netbox_dns.models
+
+class IntegerKeyValueSetting(models.Model):
+    key = models.CharField(max_length=64)
+    value = models.IntegerField()
+
+    def __str__(self):
+        return f"{self.key}: {str(self.value)}"
+
+    class Meta:
+        default_permissions = ()
+
+class CatalogZoneMemberIdentifier(models.Model):
+    name = models.CharField(
+        max_length=26,
+        unique=True,
+    )
+
+    zone = models.OneToOneField(
+        to=netbox_dns.models.Zone,
+        on_delete=models.CASCADE,
+        related_name="catz_identifier",
+    )
+
+    class Meta:
+        ordering = ("name",)
+
+    def __str__(self):
+        return self.name

netbox_dns_bridge/service/endpoint/catalog_zone_manager.py

@@ -0,0 +1,227 @@
+import logging
+import threading
+import dns.query
+import dns.message
+import dns.tsigkeyring
+import dns.name
+import dns.zone
+import dns.rdatatype
+import dns.rdataclass
+import dns.rdtypes
+import dns.exception
+import dns.renderer
+from netbox_dns.models import Zone
+from netbox_dns.choices import ZoneStatusChoices
+from netbox_dns_bridge.models import IntegerKeyValueSetting, CatalogZoneMemberIdentifier
+from uuid import uuid4
+from base64 import b32encode
+
+logger = logging.getLogger("dns-transfer-endpoint")
+
+_LOCK = threading.Lock()
+_SERIAL_MAX = 0xFFFFFFFF
+_SERIAL_OBJ = None
+_PREVIOUS_LAST_ZONE_UPDATE = None
+
+def init() -> None:
+    _init_serial()
+    _create_missing_member_identifiers()
+
+# Following function loads the last serial from the DB. No return value
+# but it sets the setting "catalog_serial" or terminates the plugin on failure.
+def _init_serial() -> None:
+    global _SERIAL_OBJ
+
+    try:
+        _SERIAL_OBJ = IntegerKeyValueSetting.objects.get(
+            key="catalog-zone-soa-serial"
+        )
+        logger.debug(
+            f"Catalog zone SOA serial number {_SERIAL_OBJ.value} loaded from database"
+        )
+    except IntegerKeyValueSetting.DoesNotExist:
+        _SERIAL_OBJ = IntegerKeyValueSetting.objects.create(
+            key="catalog-zone-soa-serial", value=1
+        )
+        logger.debug(
+            f"Catalog zone SOA serial number was not set in the database. Set to {_SERIAL_OBJ.value}"
+        )
+
+def _increment_serial() -> None:
+    global _SERIAL_OBJ
+
+    if 0 < _SERIAL_OBJ.value < _SERIAL_MAX:
+        _SERIAL_OBJ.value += 1
+        _SERIAL_OBJ.save()
+    else:
+        logger.warning(
+            f"Failed to incremenet catalog serial {_SERIAL_OBJ.value}. Will overflow serial back to 1"
+        )
+        _SERIAL_OBJ.value = 1
+        _SERIAL_OBJ.save()
+        logger.debug(
+            f"Catalog zone SOA serial number incremented to {_SERIAL_OBJ.value}"
+        )
+
+# If a zone has no catz identifier yet, create it:
+def _create_missing_member_identifiers() -> None:
+   existing_zone_ids = CatalogZoneMemberIdentifier.objects.values_list(
+       "zone_id", flat=True
+   )
+
+   missing_zones = Zone.objects.exclude(id__in=existing_zone_ids)
+
+   new_objects = [
+       CatalogZoneMemberIdentifier(
+           zone=zone,
+           name=_generate_member_identifier(),
+       )
+       for zone in missing_zones
+   ]
+
+   for identifier in new_objects:
+       logger.debug(f"Zone {identifier.zone} has no catz member identifier. Creating...")
+
+   CatalogZoneMemberIdentifier.objects.bulk_create(
+       new_objects,
+       ignore_conflicts=False,
+   )
+
+
+def create_zone(name, view_name) -> dns.zone.Zone:
+    global _PREVIOUS_LAST_ZONE_UPDATE
+    # Synchronize following across threads as TCP and UDP listener both use it.
+    with _LOCK:
+        latest_zone = (
+            Zone.objects.filter(status=ZoneStatusChoices.STATUS_ACTIVE)
+            .order_by("-last_updated")
+            .first()
+        )
+
+        last_zone_update = getattr(latest_zone, "last_updated", None)
+
+        # Check if there was a zone updated since last call
+        # If no zone was found previously then this will be false since (None != None) = False
+        if _PREVIOUS_LAST_ZONE_UPDATE != last_zone_update:
+            if last_zone_update is not None:
+                logger.debug(
+                    f"Zone {latest_zone.name} was updated in view {latest_zone.view.name}"
+                )
+            # Setting previous last zone update for next iteration:
+            _PREVIOUS_LAST_ZONE_UPDATE = last_zone_update
+            _increment_serial()
+
+    # Zone origin
+    origin = dns.name.from_text(name, dns.name.root)
+
+    # Create a new empty zone
+    zone = dns.zone.Zone(origin)
+    zone.rdclass = dns.rdataclass.IN
+
+    # get zones from netbox
+    nb_zones = Zone.objects.filter(
+        view__name=view_name, status=ZoneStatusChoices.STATUS_ACTIVE
+    ).select_related("catz_identifier")
+
+    ptr_base = dns.name.from_text("zones", origin)
+
+    for nb_zone in nb_zones:
+        ttl = 0
+        qname = dns.name.from_text(nb_zone.name, dns.name.root)
+
+        # Create PTR record
+        #p_name = f"zid-{nb_zone.id:09d}"
+        p_name = nb_zone.catz_identifier.name
+
+        ptr_name = dns.name.from_text(p_name, ptr_base)
+        if not ptr_name.is_subdomain(origin):
+            raise ValueError(
+                "Catalog zone member identifier {ptr_name.to_text()} not a subdomain"
+                )
+        rdata = dns.rdata.from_text(
+            dns.rdataclass.IN, dns.rdatatype.PTR, qname.to_text()
+        )
+        rdataset = zone.find_rdataset(ptr_name, dns.rdatatype.PTR, create=True)
+        rdataset.add(rdata, ttl)
+
+        # Configure DNSSec Policy for member Zone if DNSSec is enabled
+        if nb_zone.dnssec_policy:
+            # Configure policy
+            rid = dns.name.from_text("group", ptr_name)
+            policy_name = nb_zone.dnssec_policy.name.rstrip(" ")
+            group_name = f"dnssec-policy-{policy_name}"
+            rdata = dns.rdata.from_text(
+                dns.rdataclass.IN, dns.rdatatype.TXT, group_name
+            )
+            rdataset = zone.find_rdataset(rid, dns.rdatatype.TXT, create=True)
+            rdataset.add(rdata, ttl)
+
+        ## Configure dnssec status for member zone
+        # status = str(1 if nb_zone.dnssec_policy else 0)
+        # rid = dns.name.from_text("enabled.dnssec.ext", ptr_name)
+        # rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.TXT, status)
+        # rdataset = zone.find_rdataset(rid, dns.rdatatype.TXT, create=True)
+        # rdataset.add(rdata, ttl)
+
+    # SOA Record components
+    ttl = 0
+    rclass = dns.rdataclass.IN
+    rtype = dns.rdatatype.SOA
+    mname = dns.name.from_text("invalid", dns.name.root)
+    rname = dns.name.from_text("invalid", dns.name.root)
+    serial = _SERIAL_OBJ.value
+    refresh = 60
+    retry = 10
+    expire = 1209600
+    minimum = 0
+
+    # Create SOA rdata object
+    soa_rdata = dns.rdata.from_text(
+        rclass,
+        rtype,
+        f"{mname} {rname} {serial} {refresh} {retry} {expire} {minimum}",
+    )
+
+    # Create Rdataset and add the RDATA to it
+    soa_rdataset = dns.rdataset.Rdataset(rclass, rtype)
+    soa_rdataset.add(soa_rdata, ttl)
+
+    # Add to the origin node in the zone
+    node = zone.find_node(origin, create=True)
+    node.rdatasets.append(soa_rdataset)
+
+    # NS record for catz.
+    ns_name = dns.name.from_text("invalid", dns.name.root)
+    ns_rdata = dns.rdata.from_text(
+        dns.rdataclass.IN, dns.rdatatype.NS, str(ns_name)
+    )
+    ns_rdataset = dns.rdataset.Rdataset(dns.rdataclass.IN, dns.rdatatype.NS)
+    ns_rdataset.add(ns_rdata, 0)
+
+    # Add to node (catz. is the origin)
+    ns_node = zone.find_node(origin, create=True)
+    ns_node.rdatasets.append(ns_rdataset)
+
+    # TXT record for version.catz.
+    version_name = dns.name.from_text(
+        "version", origin
+    )  # relative to origin "catz."
+    txt_rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.TXT, '"2"')
+
+    txt_rdataset = dns.rdataset.Rdataset(dns.rdataclass.IN, dns.rdatatype.TXT)
+    txt_rdataset.add(txt_rdata, 0)
+
+    # Add to node version.catz.
+    txt_node = zone.find_node(version_name, create=True)
+    txt_node.rdatasets.append(txt_rdataset)
+
+    return zone
+
+def _generate_member_identifier() -> None:
+    return b32encode(uuid4().bytes)[0:26].lower().decode('UTF-8')
+
+def update_member_identifier(zone: Zone) -> None:
+    CatalogZoneMemberIdentifier.objects.update_or_create(
+        zone=zone,
+        defaults={"name": _generate_member_identifier},
+    )

netbox_dns_bridge/service/endpoint/dns_server.py

@@ -0,0 +1,54 @@
+import logging
+import socketserver
+import socket
+
+logger = logging.getLogger("dns-transfer-endpoint")
+
+class DNSAddressMixin:
+    def _resolve_address(self, server_address, socktype, proto):
+        host, port = server_address
+
+        infos = socket.getaddrinfo(
+            host,
+            port,
+            socket.AF_UNSPEC,
+            socktype,
+            proto,
+            socket.AI_PASSIVE
+        )
+
+        family, _, _, _, sockaddr = infos[0]
+        self.address_family = family
+        return sockaddr
+
+
+class TCPDNSServer(DNSAddressMixin, socketserver.TCPServer):
+    allow_reuse_address = True
+
+    def __init__(self, server_address, handler_class, keyring, tsig_view_map):
+        sockaddr = self._resolve_address(
+            server_address,
+            socket.SOCK_STREAM,
+            socket.IPPROTO_TCP
+        )
+
+        super().__init__(sockaddr, handler_class)
+
+        self.keyring = keyring
+        self.tsig_view_map = tsig_view_map
+
+
+class UDPDNSServer(DNSAddressMixin, socketserver.UDPServer):
+    allow_reuse_address = True
+
+    def __init__(self, server_address, handler_class, keyring, tsig_view_map):
+        sockaddr = self._resolve_address(
+            server_address,
+            socket.SOCK_DGRAM,
+            socket.IPPROTO_UDP
+        )
+
+        super().__init__(sockaddr, handler_class)
+
+        self.keyring = keyring
+        self.tsig_view_map = tsig_view_map

netbox_dns_bridge/service/endpoint/request_handler.py

@@ -0,0 +1,413 @@
+import logging
+import socketserver
+import socket
+import dns.query
+import dns.message
+import dns.tsigkeyring
+import dns.name
+import dns.zone
+import dns.rdatatype
+import dns.rdataclass
+import dns.rdtypes
+import dns.exception
+import dns.renderer
+from netbox_dns.models import Zone, Record
+from netbox_dns.choices import ZoneStatusChoices, RecordStatusChoices
+from . import catalog_zone_manager as catzm
+
+logger = logging.getLogger("dns-transfer-endpoint")
+
+
+class DNSBaseRequestHandler(socketserver.BaseRequestHandler):
+    def __init__(self, request, client_address, server) -> None:
+        self.MAX_WIRE = 65535
+        self.RESERVED_TSIG = 300
+        super().__init__(request, client_address, server)
+
+    # _getZoneFromNB rewritten
+    def _getZoneFromNB(self, zone_name, view_name) -> dns.zone.Zone:
+        # Find the zone
+        try:
+            nb_zone = Zone.objects.get(
+                name=zone_name,
+                view__name=view_name,
+                status=ZoneStatusChoices.STATUS_ACTIVE,
+            )
+        except Zone.DoesNotExist:
+            return None
+
+        # Build DNS zone
+        zone = dns.zone.Zone(zone_name, dns.name.root)
+        zone.rdclass = dns.rdataclass.IN
+
+        nb_records = Record.objects.filter(
+            zone=nb_zone, status=RecordStatusChoices.STATUS_ACTIVE
+        )
+
+        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 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 _denyRequestBadTSIG(self, wire, tsig_error: dns.rcode) -> None:
+        # Use empty keyring to parse TSIG without validating
+        query = dns.message.from_wire(
+            wire, keyring={}, ignore_trailing=True, continue_on_error=True
+        )
+
+        # Make a response
+        response = dns.message.make_response(query)
+        response.set_rcode(dns.rcode.REFUSED)
+
+        if query.had_tsig:
+            # Add TSIG with BADKEY, but do not sign (no MAC)
+            response.use_tsig(
+                keyring={},  # empty; we're not signing
+                keyname=query.keyname,
+                tsig_error=tsig_error,
+            )
+        self.denyRequest(query)
+
+    def _send_response(self, data) -> None:
+        raise NotImplementedError
+
+    def _deny_request(self, query, rcode: dns.rcode = dns.rcode.REFUSED) -> None:
+        response = dns.message.make_response(query)
+        response.set_rcode(rcode)
+        wire = response.to_wire(multi=False)
+        self._send_response(wire)
+
+    def _handle_soa_request(self, query, soa_rrset, zone, peer, nb_view, dname) -> None:
+        # We assume that the SOA rdataset has at least one record (it usually does).
+        soa_rdata = soa_rrset[0]  # Get the first SOA record
+
+        # Now, create the rrset from the soa_rdata
+        rrset = dns.rrset.from_rdata(zone.origin, soa_rrset.ttl, soa_rdata)
+
+        response = dns.message.make_response(query)
+        # Set the Authoritative Answer flag
+        response.flags |= dns.flags.AA
+
+        # Append the rrset to the response's answer section
+        response.answer.append(rrset)
+
+        # TSIG response
+        if query.had_tsig:
+            # If key was found in DB
+            if query.keyname in self.server.keyring:
+                response.use_tsig(
+                    self.server.keyring, keyname=query.keyname, original_id=query.id
+                )
+            else:
+                # If key is unknown
+                response.set_rcode(dns.rcode.REFUSED)
+                # Generate new key for keyname provided
+                # keyname = query.keyname
+                # b64 = base64.b64encode(urandom(32)).decode("ascii")
+                # newkey = dns.tsigkeyring.from_text({ keyname: b64 })
+                response.use_tsig(
+                    self.server.keyring,
+                    keyname=query.keyname,
+                    tsig_error=dns.rcode.BADKEY,
+                )
+
+        data = response.to_wire(max_size=512)
+        self._send_response(data)
+        logger.debug(f"{peer} SOA {nb_view.name}/{dname}")
+
+    def _handle_axfr_request(self, query, zone, peer, nb_view, dname) -> None:
+        rrsets = []
+        soa_rrset = None
+        for name, rdataset in zone.iterate_rdatasets():
+            if not name.is_absolute():
+                name = name.concatenate(zone.origin)
+            rrset = dns.rrset.from_rdata_list(name, rdataset.ttl, rdataset)
+            if rdataset.rdtype == dns.rdatatype.SOA and soa_rrset is None:
+                soa_rrset = rrset
+            else:
+                rrsets.append(rrset)
+
+        rrsets.insert(0, soa_rrset)  # Opening SOA
+        rrsets.append(soa_rrset)  # Closing SOA
+
+        # 2. Create a Renderer for the first message
+        r = dns.renderer.Renderer(
+            id=query.id,
+            flags=(dns.flags.QR | dns.flags.AA),
+            max_size=self.MAX_WIRE,
+            origin=None,
+        )
+        r.add_question(
+            query.question[0].name,
+            query.question[0].rdtype,
+            query.question[0].rdclass,
+        )
+
+        # 3. Loop through RRsets
+        tsig_ctx = None
+        for rrset in rrsets:
+            # logger.debug(f"Iterating over rrset {rrset}. tsig_ctx: {tsig_ctx}")
+            try:
+                # logger.debug(f"Adding {rrset} to renderer object")
+                r.add_rrset(dns.renderer.ANSWER, rrset)
+                if r.max_size - len(r.output.getvalue()) < self.RESERVED_TSIG:
+                    raise dns.exception.TooBig("TSIG wont fit")
+            except dns.exception.TooBig:
+                # TSIG chain previous message
+                r.write_header()
+                tsig_ctx = r.add_multi_tsig(
+                    ctx=tsig_ctx,
+                    secret=self.server.keyring[query.keyname],
+                    keyname=query.keyname,
+                    fudge=300,
+                    id=query.id,
+                    tsig_error=0,
+                    other_data=b"",
+                    request_mac=r.mac if tsig_ctx else query.mac,
+                )
+                wire = r.get_wire()
+                self.request.sendall(len(wire).to_bytes(2, "big") + wire)
+
+                # Start new renderer
+                r = dns.renderer.Renderer(
+                    id=query.id,
+                    flags=(dns.flags.QR | dns.flags.AA),
+                    max_size=self.MAX_WIRE,
+                    origin=None,
+                )
+                r.add_question(
+                    query.question[0].name,
+                    query.question[0].rdtype,
+                    query.question[0].rdclass,
+                )
+                r.add_rrset(dns.renderer.ANSWER, rrset)
+
+        # 4. Final message with terminating TSIG
+        r.write_header()
+        # logger.debug(f"Final message. tsig_ctx: {tsig_ctx}")
+        tsig_ctx = r.add_multi_tsig(
+            ctx=tsig_ctx,
+            secret=self.server.keyring[query.keyname],
+            keyname=query.keyname,
+            fudge=300,
+            id=query.id,
+            tsig_error=0,
+            other_data=b"",
+            # request_mac=r.mac if tsig_ctx else None,
+            request_mac=r.mac if tsig_ctx else query.mac,
+        )
+        wire = r.get_wire()
+        self._send_response(wire)
+
+        logger.debug(f"{peer} AXFR {nb_view.name}/{dname}")
+
+    def _handle_dns_query(self, wire) -> None:
+        peer = self.client_address[0]
+        try:
+            query = dns.message.from_wire(
+                wire,
+                keyring=self.server.keyring,
+                continue_on_error=False,
+                ignore_trailing=True
+            )
+
+        except dns.tsig.BadSignature as e:
+            logger.warning(
+                f"Request denied from {peer} failed TSIG verification: {e}"
+            )
+            self._denyRequestBadTSIG(wire, dns.rcode.BADSIG)
+            return
+
+        except (dns.message.UnknownTSIGKey, dns.tsig.BadAlgorithm) as e:
+            logger.warning(f"Request denied from {peer} with bad TSIG key: {e}")
+            self._denyRequestBadTSIG(wire, dns.rcode.BADKEY)
+            return
+
+        except Exception as e:
+            logger.error("Error parsing query: ", e)
+            return
+
+        # If there was no question in the query, refuse
+        if len(query.question) != 1:
+            self._deny_request(query)
+            return
+
+        question = query.question[0]
+        qname = question.name
+        qtype = question.rdtype
+        dname = qname.to_text().rstrip(".")
+
+        # Only process AXFR/SOA queris
+        if qtype not in (dns.rdatatype.AXFR, dns.rdatatype.SOA):
+            logger.warning(
+                f"Request denied from {peer}: Request was not AXFR or SOA (Type: {qtype})"
+            )
+            self._deny_request(query)
+            return
+
+        # Identify TSIG key used
+        if not query.had_tsig:
+            logger.warning(f"Request denied from {peer}: No TSIG key used")
+            self._deny_request(query)
+            return
+
+        key_name = query.keyname.canonicalize().to_text()
+
+        # Check if the key matches a view
+        nb_view = self.server.tsig_view_map.get(key_name)
+        if not nb_view:
+            logger.warning(
+                f"Request denied from {peer}: {key_name} does not match a view"
+            )
+            self._deny_request(query)
+            return
+
+        # Check if catalog zone
+        if dname == "catz" or dname == f"{nb_view.name}.catz":
+            zone = catzm.create_zone(dname, nb_view.name)
+        else:
+            zone = self._getZoneFromNB(dname, nb_view.name)
+        # When zone was not found, let client know
+        if not zone:
+            logger.warning(f"Zone {dname} not found in view {nb_view.name}")
+            self._deny_request(query)
+            return
+
+        # Retrieve the existing SOA record from the Zone
+        soa_rrset = zone.get_rdataset(zone.origin, dns.rdatatype.SOA)
+        if soa_rrset is None:
+            logger.error(f"Zone {dname} has no SOA — aborting")
+            return
+
+        if qtype == dns.rdatatype.SOA:
+            self._handle_soa_request(query, soa_rrset, zone, peer, nb_view, dname)
+        elif qtype == dns.rdatatype.AXFR:
+            self._handle_axfr_request(query, zone, peer, nb_view, dname)
+
+
+class UDPRequestHandler(DNSBaseRequestHandler):
+    def __init__(self, request, client_address, server) -> None:
+        super().__init__(request, client_address, server)
+
+    def _send_response(self, data) -> None:
+        sock = self.request[1]
+        sock.sendto(data, self.client_address)
+
+    def handle(self) -> None:
+        data, sock = self.request
+        peer = self.client_address[0]
+        try:
+            self._handle_dns_query(data)
+        except Exception as e:
+            logger.error(f"Error handling request from {peer}: {e}")
+            import traceback
+            traceback.print_exc()
+
+
+class TCPRequestHandler(DNSBaseRequestHandler):
+    def __init__(self, request, client_address, server) -> None:
+        super().__init__(request, client_address, server)
+
+    def _send_response(self, data) -> None:
+        length = len(data).to_bytes(2, byteorder="big")
+        self.request.sendall(length + data)
+
+    def handle(self) -> None:
+        peer = self.client_address[0]
+        sock = self.request  # TCP socket
+        sock.settimeout(10.0)  # Default 10 second timeout for inactivity
+        try:
+            while True:
+                # Read 2-byte length prefix
+                length_data = sock.recv(2)
+                if not length_data:
+                    return  # connection closed by peer
+                if len(length_data) < 2:
+                    # Incomplete length data, wait for more or fail
+                    while len(length_data) < 2:
+                        chunk = sock.recv(2 - len(length_data))
+                        if not chunk:
+                            return
+                        length_data += chunk
+
+                length = int.from_bytes(length_data, byteorder="big")
+                wire = b""
+                while len(wire) < length:
+                    chunk = sock.recv(length - len(wire))
+                    if not chunk:
+                        return  # connection closed
+                    wire += chunk
+                self._handle_dns_query(wire)
+
+        except socket.timeout:
+            logger.debug(f"Connection from {peer} timed out")
+        except Exception as e:
+            logger.error(f"Error handling request from {peer}: {e}")
+            import traceback
+            traceback.print_exc()
+

netbox_dns_bridge/signals.py

@@ -0,0 +1,40 @@
+from django.db.models.signals import pre_save, post_save
+from django.dispatch import receiver
+from netbox_dns.models import Zone
+from .service.endpoint import catalog_zone_manager as catzm
+
+@receiver(pre_save, sender=Zone)
+def zone_pre_save(sender, instance, **kwargs):
+    """
+    Cache the old name so post_save can see if it changed.
+    """
+    if instance.pk:
+        try:
+            instance._old_name = (
+                sender.objects
+                .only("name")
+                .get(pk=instance.pk)
+                .name
+            )
+        except sender.DoesNotExist:
+            instance._old_name = None
+    else:
+        instance._old_name = None
+
+@receiver(post_save, sender=Zone)
+def sync_catalog_zone_identifier(sender, instance, created, **kwargs):
+    """
+    Ensure CatalogZoneMemberIdentifier exists for each Zone
+    and keep its identifier in sync.
+    """
+    if created:
+        # Create the related object on Zone creation
+        catzm.update_member_identifier(instance)
+    else:
+        old_name = getattr(instance, "_old_name", None)
+
+        # If the name did not change, do nothing
+        if old_name == instance.name:
+            return
+
+        catzm.update_member_identifier(instance)

netbox_dns_bridge/utils.py

@@ -0,0 +1,35 @@
+import os
+import dns.zone
+import dns.rdatatype
+import dns.rdataclass
+import dns.exception
+import netbox_dns.models
+
+def export_bind_zone_file(nb_zone: netbox_dns.models.Zone, file_path: str):
+
+    # 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, record.ttl)
+        rdataset.add(rdata)
+
+        # 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)
+    except IOError as e:
+        raise IOError(f"Failed to write zone file to {file_path}: {e}")
+