modified: src/netbox_plugin_bind_provisioner/__init__.py

Bumping version to v1.0.6
- Split bind transfer command into multiple files for better code readability.
2026-02-09 16:54:17 +01:00 · 2026-02-05 12:27:28 +01:00 · 2026-02-05 12:26:55 +01:00 · 2026-01-31 09:43:19 +01:00 · 2026-01-31 09:08:42 +01:00 · 2026-01-31 09:01:04 +01:00
18 changed files with 1143 additions and 764 deletions
@@ -1,32 +1,43 @@
 # Netbox Bind Provisioner
-A NetBox plugin that implements a lightweight DNS server to automate the
+The Netbox Bind Provisioner plugin implements a lightweight DNS server inside
-provisioning and configuration of a BIND9 server based on DNS records
+Netbox and builds a bridge for BIND and other DNS Servers implementing RFC9432
-maintained in NetBox. (netbox-plugin-dns).
+to retrieve DNS Zones directly from Netbox using DNS native mechanisms.
 [![PyPi](https://img.shields.io/pypi/v/netbox-plugin-bind-provisioner)](https://pypi.org/project/netbox-plugin-bind-provisioner/)
 [![Stars Badge](https://img.shields.io/github/stars/suraxius/netbox-plugin-bind-provisioner?style=flat)](https://github.com/suraxius/netbox-plugin-bind-provisioner/stargazers)
 [![Forks Badge](https://img.shields.io/github/forks/suraxius/netbox-plugin-bind-provisioner?style=flat)](https://github.com/suraxius/netbox-plugin-bind-provisioner/network/members)
 [![Issues Badge](https://img.shields.io/github/issues/suraxius/netbox-plugin-bind-provisioner)](https://github.com/suraxius/netbox-plugin-bind-provisioner/issues)
 [![Pull Requests Badge](https://img.shields.io/github/issues-pr/suraxius/netbox-plugin-bind-provisioner)](https://github.com/suraxius/netbox-plugin-bind-provisioner/pulls)
 [![GitHub contributors](https://img.shields.io/github/contributors/suraxius/netbox-plugin-bind-provisioner?color=2b9348)](https://github.com/suraxius/netbox-plugin-bind-provisioner/graphs/contributors)
 [![License Badge](https://img.shields.io/github/license/suraxius/netbox-plugin-bind-provisioner?color=2b9348)](https://github.com/suraxius/netbox-plugin-bind-provisioner/blob/master/LICENSE)
 [![Code Style Black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)
 [![Downloads](https://static.pepy.tech/personalized-badge/netbox-plugin-bind-provisioner?period=total&left_color=BLACK&right_color=BLUE&left_text=Downloads)](https://pepy.tech/project/netbox-plugin-bind-provisioner)
 [![Downloads/Week](https://static.pepy.tech/personalized-badge/netbox-plugin-bind-provisioner?period=monthly&left_color=BLACK&right_color=BLUE&left_text=Downloads%2fMonth)](https://pepy.tech/project/netbox-plugin-bind-provisioner)
 [![Downloads/Month](https://static.pepy.tech/personalized-badge/netbox-plugin-bind-provisioner?period=weekly&left_color=BLACK&right_color=BLUE&left_text=Downloads%2fWeek)](https://pepy.tech/project/netbox-plugin-bind-provisioner)
 ## Plugin configuration
-The plugin has been reworked: instead of exporting zone files, it now provides
+While providing Zone transfers via AXFR, the Server also exposes specialized
-a minimal DNS server that is fed directly from NetBox’s DNS data. This server
+catalog zones that BIND and other RFC9432 compliant DNS Servers use to
-also exposes specialized catalog zones that BIND uses to automatically discover
+automatically discover newly created zones and remove deleted ones. The plugin
-newly created zones and remove deleted ones. The plugin supports BIND views and
+supports views and basic DNS security via TSIG.
 basic DNS security via TSIG.
-The plugin exposes one catalog zone per view. Each catalog zone is available
+The plugin exposes one catalog zone per view. Each catalog zone is made available
-under the special zone name **"catz"** and can be queried through the built-in
+under the special zone name **"catz"** and addtionally under **"[viewname].catz"**
-DNS server like any other zone.
+and may be queried through the built-in DNS server just like any other dns zone.
 For proper operation, each view requires an installed TSIG key, and the
-`bind-transfer-endpoint` must be running as a separate background service using
+`dns-transfer-endpoint` must be running as a separate background service using
 the `manage.py` command. Note that DNSSEC support will be added once BIND9
 provides a mechanism to configure it through the Catalog Zones system.
-To start the service:
+To start the service in the foreground:
 ```
-manage.py bind-transfer-endpoint --port 5354
+manage.py dns-transfer-endpoint --port 5354
 ```
 This process needs to be scheduled as a background service for the built-in DNS
 Server to work correctly. For Linux users with Systemd (Ubuntu, etc), Matt Kollross
 provides a startup unit and instructions [here](docs/install-systemd-service.md).
 When configuring the plugin, you need to provide a path on the filesystem for
 this file where the plugin can create and manage it. Please do not update it by
 hand or remove it unless you know what you are doing.
 ### Service parameters
 Parameter | Description
@@ -37,7 +48,6 @@ Parameter | Description
 ### Plugin settings
 Setting             | Description
 --------------------| ---------------------------------------------------------
 catalog_serial_file | !!!NO LONGER USED!!! Replaced by DB entry.
 tsig_keys           | Maps a TSIG Key to be used for each view.
 ## Installation guide
@@ -115,6 +125,7 @@ This guide assumes:
    ```
 5. Start listener
    This step runs the DNS endpoint used by bind to configure itself. You may want
    to write a service wrapper that runs this in the background.
    A guide for setting up a systemd service on Ubuntu is provided by Matt
@@ -124,10 +135,10 @@ This guide assumes:
    Note that `--port 5354` is optional. The listener will bind this port
    by default.
    ```
-    python3 netbox/manage.py bind-transfer-endpoint --port 5354
+    python3 netbox/manage.py dns-transfer-endpoint --port 5354
    ```
-6. Configuring a Bind9 to interact with Netbox via the bind-transfer-endpoint
+6. Configuring a Bind9 to interact with Netbox via the dns-transfer-endpoint
   endpoint. Note that its not possible to give all the correct details of the
   `options` block as it is heavily dependent on the Operating System used.
   Please dont forget to adjust as required.
@@ -15,7 +15,7 @@
    PIDFile=/var/tmp/netbox.pid
    WorkingDirectory=/opt/netbox
-    ExecStart=/opt/netbox/venv/bin/python3  /opt/netbox/netbox/manage.py bind-transfer-endpoint --port 5354
+    ExecStart=/opt/netbox/venv/bin/python3  /opt/netbox/netbox/manage.py dns-transfer-endpoint --port 5354
    Restart=on-failure
    RestartSec=30
@@ -1,11 +1,8 @@
 import logging
 from netbox.plugins import PluginConfig
 from django.db import models
 from django.utils.translation import gettext_lazy as _
 from django.dispatch import receiver
 from django.conf import settings
-__version__ = "1.0.0"
+__version__ = "1.0.6"
 logger = logging.getLogger(__name__)
@@ -32,6 +29,8 @@ class BindProvisionerConfig(PluginConfig):
                f"{self.name}: Plugin {self.verbose_name} failed to initialize due to missing settings. Terminating Netbox."
            )
        from . import signals as signals
 config = BindProvisionerConfig
 default_app_config = ".apps.NetboxBindProvisionerConfig"
@@ -1,726 +0,0 @@
 import logging
 import socketserver
 import os
 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 hashlib import sha256
 from django.core.management.base import BaseCommand
 from django.conf import settings
 from django.utils import timezone
 from netbox_dns.models import Zone, Record, View
 from netbox_dns.choices import ZoneStatusChoices, RecordStatusChoices
 from datetime import datetime
 from netbox_plugin_bind_provisioner.models import IntegerKeyValueSetting
 logger = logging.getLogger("bind-transfer-endpoint")
 class CatalogZone:
    lock = threading.Lock()
    _serial_max = 0xFFFFFFFF
    _serial_obj = None
    _previous_last_zone_update = None
    # 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.
    @classmethod
    def initSerial(cls):
        try:
            cls._serial_obj = IntegerKeyValueSetting.objects.get(
                key="catalog-zone-soa-serial"
            )
            logger.debug(
                f"Catalog zone SOA serial number {cls._serial_obj.value} loaded from database"
            )
        except IntegerKeyValueSetting.DoesNotExist:
            cls._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 {cls._serial_obj.value}"
            )
    @classmethod
    def _incrementSerial(cls):
        if 0 < cls._serial_obj.value < cls._serial_max:
            cls._serial_obj.value += 1
            cls._serial_obj.save()
        else:
            logger.warn(
                f"Failed to incremenet catalog serial {cls._serial_obj.value}. Will overflow serial back to 1"
            )
            cls._serial_obj = 1
            cls._serial_obj.save()
            logger.debug(
                f"Catalog zone SOA serial number incremented to {_serial_obj.value}"
            )
    @classmethod
    def create(cls, name, view_name):
        # Synchronize following across threads as TCP and UDP listener both use it.
        with cls.lock:
            # cls = self.__class__
            # datestamp = datetime.now().strftime("%y%m%d")
            # latest_zone = Zone.objects.filter().order_by("-last_updated").first()
            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 cls._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:
                cls._previous_last_zone_update = last_zone_update
                cls._incrementSerial()
        # 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
        )
        ptr_base = dns.name.from_text("zones", origin)
        for nb_zone in nb_zones:
            ttl = nb_zone.soa_ttl
            qname = dns.name.from_text(nb_zone.name, dns.name.root)
            # zone_id = sha256(qname.to_text().encode()).hexdigest()[0:63]
            zone_id = str(nb_zone.id)
            # Create PTR record
            ptr_name = dns.name.from_text(zone_id, ptr_base)
            assert ptr_name.is_subdomain(origin)
            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 = 3600
        rclass = dns.rdataclass.IN
        rtype = dns.rdatatype.SOA
        mname = dns.name.root  # dns.name.from_text("invalid.", origin)
        rname = dns.name.root  # dns.name.from_text("contact.master.", origin)
        serial = cls._serial_obj.value
        refresh = 60
        retry = 10
        expire = 1209600
        minimum = 86400
        # 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, 86400)
        # 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, 86400)
        # Add to node version.catz.
        txt_node = zone.find_node(version_name, create=True)
        txt_node.rdatasets.append(txt_rdataset)
        return zone
    @classmethod
    def importLegacySerialFile(cls, file_path):
        if os.path.exists(file_path):
            try:
                with open(file_path, "r") as f:
                    serial = int(f.read().strip())
                    cls._serial_obj.value = int(serial)
                    cls._serial_obj.save()
                    logger.info(
                        f"Legacy catalog serial file {file_path} imported into database and file removed. Serial is now {serial}"
                    )
                    os.remove(file_path)
            except OSError as e:
                logger.error(
                    f"Failed to import catalog zone serial from legacy file {file_path}"
                )
 class DNSBaseRequestHandler(socketserver.BaseRequestHandler):
    # getZoneFromNB rewritten
    def getZoneFromNB(self, zone_name, view_name):
        # 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
            rdata = dns.rdata.from_text(
                dns.rdataclass.IN,
                rdtype,
                record.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):
        # 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)
 class UDPRequestHandler(DNSBaseRequestHandler):
    def denyRequest(self, query, rcode: dns.rcode = dns.rcode.REFUSED):
        response = dns.message.make_response(query)
        response.set_rcode(rcode)
        wire = response.to_wire(multi=False)
        sock = self.request[1]
        sock.sendto(wire, self.client_address)
    def handle(self):
        data, sock = self.request
        peer = self.client_address[0]
        query = None
        try:
            query = dns.message.from_wire(
                data,
                keyring=self.server.keyring,
                continue_on_error=False,
                ignore_trailing=True,
            )
        except Exception as e:
            logger.error("Error parsing query: ", e)
            return
        # Create a response
        response = dns.message.make_response(query)
        # Set the Authoritative Answer flag
        response.flags |= dns.flags.AA
        question = query.question[0]
        # Get the queried record name and type:
        qname = question.name
        qtype = question.rdtype
        dname = qname.to_text().rstrip(".")
        # Only process SOA queries
        if qtype != dns.rdatatype.SOA:
            logger.warning(
                f"Request denied from {peer}: Request was not SOA (Type: {qtype})"
            )
            self.denyRequest(query)
            return
        # Identify TSIG key used
        if not query.had_tsig:
            logger.warning(f"Request denied from {peer}: No TSIG key used")
            self.denyRequest(query)
            return
        key_name = query.keyname.canonicalize().to_text()
        # Check if 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.denyRequestBadTSIG(wire, dns.rcode.BADKEY)
            return
        # Check if catalog zone
        if dname == "catz" or dname == f"{nb_view.name}.catz":
            zone = CatalogZone.create(dname, nb_view.name)
        # If generic zone, retreive from NB
        else:
            zone = self.getZoneFromNB(dname, nb_view.name)
            # When zone was not found, let client know
            if not zone:
                logger.warning(f"Zone {nb_view.name}/{dname} not found in NB")
                self.denyRequest(query)
                return
        # Retrieve the existing SOA record from the Zone
        soa_rdataset = zone.get_rdataset(zone.origin, dns.rdatatype.SOA)
        # We assume that the SOA rdataset has at least one record (it usually does).
        soa_rdata = soa_rdataset[0]  # Get the first SOA record
        # Now, create the rrset from the soa_rdata
        rrset = dns.rrset.from_rdata(zone.origin, soa_rdataset.ttl, soa_rdata)
        # 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,
                )
        # Send back the response
        response_wire = response.to_wire(max_size=512)
        sock.sendto(response_wire, self.client_address)
        logger.debug(f"{peer} SOA {nb_view.name}/{dname}")
 class TCPRequestHandler(DNSBaseRequestHandler):
    def denyRequest(self, query, rcode: dns.rcode = dns.rcode.REFUSED):
        response = dns.message.make_response(query)
        response.set_rcode(rcode)
        wire = response.to_wire(multi=False)
        length = len(wire).to_bytes(2, byteorder="big")
        self.request.sendall(length + wire)
    def handle(self):
        MAX_WIRE = 65535
        # MAX_WIRE = 2000 # For testing fragmentation
        RESERVED_TSIG = 300
        # create a var for setting the client ip in log messages
        peer = self.client_address[0]
        try:
            # Receive the entire message....
            sock = self.request  # TCP socket
            # Read 2-byte length prefix
            length_data = sock.recv(2)
            if len(length_data) < 2:
                return
            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
            # Parse the query
            query = None
            try:
                query = dns.message.from_wire(wire, keyring=self.server.keyring)
            # If TSIG signature doesnt match our key, refuse query:
            except dns.tsig.BadSignature as e:
                logger.warning(
                    f"Request denied from {peer} failed TSIG verification: {e}"
                )
                self.denyRequestBadTSIG(wire, dns.rcode.BADSIG)
                return
            # If TSIG Key used is not in our list, refuse query:
            except dns.message.UnknownTSIGKey as e:
                logger.warning(f"Request denied from {peer} with bad TSIG key: {e}")
                self.denyRequestBadTSIG(wire, dns.rcode.BADKEY)
                return
            # If there was no question in the query, refuse
            if len(query.question) != 1:
                self.denyRequest(query)
                return
            # Get the queried record name and type:
            qname = query.question[0].name
            qtype = query.question[0].rdtype
            dname = qname.to_text().rstrip(".")
            # Only process AXFR queries
            if qtype != dns.rdatatype.AXFR:
                # if qtype != dns.rdatatype.AXFR or qtype != dns.rdatatype.SOA:
                logger.warning(
                    f"Request denied from {peer}: Request was not AXFR (Type: {qtype})"
                )
                self.denyRequest(query)
                return
            # Identify TSIG key used
            if not query.had_tsig:
                logger.warning(f"Request denied from {peer}: No TSIG key used")
                self.denyRequest(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.denyRequest(query)
                return
            # Check if catalog zone
            if dname == "catz" or dname == f"{nb_view.name}.catz":
                zone = CatalogZone.create(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.denyRequest(query)
                return
            # 1. Prepare SOA + RRsets (identical to your code)
            soa_rrset = None
            rrsets = []
            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)
            if soa_rrset is None:
                logger.error(f"Zone {dname} has no SOA — aborting AXFR")
                return
            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=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()) < 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=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.request.sendall(len(wire).to_bytes(2, "big") + wire)
            # logger.debug(f"Zone transfer request for {nb_view.name}/{dname} from {peer}")
            logger.debug(f"{peer} AXFR {nb_view.name}/{dname}")
        except Exception as e:
            logger.error(f"Error handling request from {peer}: {e}")
            import traceback
            traceback.print_exc()
 class TCPDNSServer(socketserver.TCPServer):
    allow_reuse_address = True
    def __init__(self, server_address, handler_class, keyring, tsig_view_map):
        super().__init__(server_address, handler_class)
        self.keyring = keyring
        self.tsig_view_map = tsig_view_map
 class UDPDNSServer(socketserver.UDPServer):
    allow_reuse_address = True
    def __init__(self, server_address, handler_class, keyring, tsig_view_map):
        super().__init__(server_address, handler_class)
        self.keyring = keyring
        self.tsig_view_map = tsig_view_map
 class Command(BaseCommand):
    help = "Run a minimal AXFR DNS server using NetBox DNS plugin data"
    def loadSettings(self):
        self.settings = settings.PLUGINS_CONFIG.get(
            "netbox_plugin_bind_provisioner", 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 loadTSIGKeySettings(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"]
        CatalogZone.initSerial()
        # Initialize settings
        self.loadSettings()
        # self.loadCatalogZoneSettings()
        self.loadTSIGKeySettings()
        # Following is temporary. Function and invocation can be deleted once v1.0 is reached.
        serial_file = self.settings.get("catalog_serial_file", None)
        if serial_file:
            CatalogZone.importLegacySerialFile(serial_file)
        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"SOA 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"AXFR endpoint listening on {address} tcp/{port}")
        tcp_server.serve_forever()
@@ -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_plugin_bind_provisioner.service.endpoint.request_handler import UDPRequestHandler, TCPRequestHandler
 from netbox_plugin_bind_provisioner.service.endpoint.dns_server import UDPDNSServer, TCPDNSServer
 from netbox_plugin_bind_provisioner.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_plugin_bind_provisioner", 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()
@@ -0,0 +1,47 @@
 import os
 import dns.zone
 import dns.rdatatype
 import dns.rdataclass
 import dns.exception
 import netbox_dns.models
 from netbox_plugin_bind_provisioner.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}")
@@ -0,0 +1,56 @@
 import os
 import dns.zone
 import dns.rdatatype
 import dns.rdataclass
 import dns.exception
 import netbox_dns.models
 from netbox_plugin_bind_provisioner.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}'"))
@@ -0,0 +1,33 @@
 # Generated by Django 5.2.6 on 2025-12-18 14:51
 from django.db import migrations
 class Migration(migrations.Migration):
    dependencies = [
        ('netbox_plugin_bind_provisioner', '0001_initial'),
    ]
    operations = [
        migrations.AlterModelOptions(
            name='integerkeyvaluesetting',
            options={'default_permissions': ()},
        ),
        migrations.RemoveField(
            model_name='integerkeyvaluesetting',
            name='created',
        ),
        migrations.RemoveField(
            model_name='integerkeyvaluesetting',
            name='custom_field_data',
        ),
        migrations.RemoveField(
            model_name='integerkeyvaluesetting',
            name='last_updated',
        ),
        migrations.RemoveField(
            model_name='integerkeyvaluesetting',
            name='tags',
        ),
    ]
@@ -0,0 +1,26 @@
 # Generated by Django 5.2.9 on 2026-02-02 09:23
 import django.db.models.deletion
 from django.db import migrations, models
 class Migration(migrations.Migration):
    dependencies = [
        ('netbox_dns', '0030_dnsseckeytemplate_comments_dnsseckeytemplate_owner_and_more'),
        ('netbox_plugin_bind_provisioner', '0002_alter_integerkeyvaluesetting_options_and_more'),
    ]
    operations = [
        migrations.CreateModel(
            name='CatalogZoneMemberIdentifier',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False)),
                ('name', models.CharField(max_length=26, unique=True)),
                ('zone', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='bind_provisioner_models', to='netbox_dns.zone')),
            ],
            options={
                'ordering': ('name',),
            },
        ),
    ]
@@ -0,0 +1,20 @@
 # Generated by Django 5.2.9 on 2026-02-02 10:53
 import django.db.models.deletion
 from django.db import migrations, models
 class Migration(migrations.Migration):
    dependencies = [
        ('netbox_dns', '0030_dnsseckeytemplate_comments_dnsseckeytemplate_owner_and_more'),
        ('netbox_plugin_bind_provisioner', '0003_catalogzonememberidentifier'),
    ]
    operations = [
        migrations.AlterField(
            model_name='catalogzonememberidentifier',
            name='zone',
            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='bind_provisioner_catz_identifier', to='netbox_dns.zone'),
        ),
    ]
@@ -0,0 +1,20 @@
 # Generated by Django 5.2.9 on 2026-02-02 20:49
 import django.db.models.deletion
 from django.db import migrations, models
 class Migration(migrations.Migration):
    dependencies = [
        ('netbox_dns', '0030_dnsseckeytemplate_comments_dnsseckeytemplate_owner_and_more'),
        ('netbox_plugin_bind_provisioner', '0004_alter_catalogzonememberidentifier_zone'),
    ]
    operations = [
        migrations.AlterField(
            model_name='catalogzonememberidentifier',
            name='zone',
            field=models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='catz_identifier', to='netbox_dns.zone'),
        ),
    ]
@@ -2,23 +2,30 @@ from django.db import models
 import netbox.models
 import netbox_dns.models
-
+class IntegerKeyValueSetting(models.Model):
 # class CatalogZoneSerial(models.Model):
 class IntegerKeyValueSetting(netbox.models.NetBoxModel):
    # view = models.ForeignKey(netbox_dns.models.View, on_delete=models.CASCADE)
    key = models.CharField(max_length=64)
    value = models.IntegerField()
    def __str__(self):
        # return f'{self.view.name}: {self.serial}'
        return f"{self.key}: {str(self.value)}"
    class Meta:
        default_permissions = ()
-# class TSIGKey(netbox.models.NetBoxModel):
+class CatalogZoneMemberIdentifier(models.Model):
-#    view      = models.ForeignKey(netbox_dns.models.View, on_delete=models.CASCADE)
+    name = models.CharField(
-#    name      = models.CharField(max_length=256)
+        max_length=26,
-#    algorithm = models.CharField(max_length=25)
+        unique=True,
-#    secret    = models.CharField(max_length=50)
+    )
-#
+
-#    def __str__(self):
+    zone = models.OneToOneField(
-#        return f'key "{self.name}" {{ algorithm {self.algorithm}; secret "{self.secret}"; }};'
+        to=netbox_dns.models.Zone,
        on_delete=models.CASCADE,
        related_name="catz_identifier",
    )
    class Meta:
        ordering = ("name",)
    def __str__(self):
        return self.name
@@ -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_plugin_bind_provisioner.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},
    )
@@ -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
@@ -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()
@@ -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)
@@ -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}")
Author	SHA1	Message	Date
Sven Luethi	997f36264a	modified: src/netbox_plugin_bind_provisioner/__init__.py Bumping version to v1.0.6	2026-02-09 16:54:17 +01:00
Sven Luethi	19a946c880	- Split bind transfer command into multiple files for better code readability. - Rename bind-transfer-endpoint command -> dns-transfer-endpoint.	2026-02-05 12:27:28 +01:00
Sven Luethi	9afdfe869b	- Store catz member zone identifiers in database, linked to the corresponding zone object. - Catz member identifiers are created from uuid encoded in base32, lower-cased, and padding removed. - On endpoint startup, make sure all zones have a catz identifier assigned. If not, create them. - Use signals to create or update a zone's catz identifier.	2026-02-05 12:26:55 +01:00
Sven Luethi	1fae3bfe54	Adds back the changes made to support ipv6 in commit `8fba131230` but that were reverted with commit `ca51a6054d` to allow proper merging of another complex PR. Closes #11 again Also includes a small bug fix that was added with commit `6bbb2e9905` but that was also temporarily reverted with `45f9631a98`.	2026-01-31 09:43:19 +01:00
Sven Luethi	562f54d481	Merge pull request #12 from spamik/support-tcp-soa Support SOA request on TCP by merging the dns handling while leaving the transport handling in protocol specific handlers (partially). Further review pending	2026-01-31 09:08:42 +01:00
Sven Luethi	ca51a6054d	This reverts commit `8fba131230`.	2026-01-31 09:01:04 +01:00
Sven Luethi	45f9631a98	This reverts commit `6bbb2e9905`.	2026-01-31 09:00:44 +01:00
spm	0d041a0830	Handle more DNS requests via one TCP connection	2026-01-29 15:58:32 +01:00
Sven Luethi	6bbb2e9905	modified: src/netbox_plugin_bind_provisioner/management/commands/bind-transfer-endpoint.py Catches a further exception if the wrong algorithm is specified for the tsig key used.	2026-01-29 11:11:49 +01:00
Sven Luethi	8fba131230	modified: src/netbox_plugin_bind_provisioner/management/commands/bind-transfer-endpoint.py Fixes issue #11 (IPv6 support)	2026-01-29 11:10:31 +01:00
spm	a9dbbe8c88	Support SOA request on TCP - unified handling of DNS queries on TCP and UDP channels - as a result it now responds on SOA requests over TCP	2026-01-28 17:35:58 +01:00
Sven Luethi	9a2b0a2c2b	modified: src/netbox_plugin_bind_provisioner/__init__.py Bump version to v1.0.5	2026-01-09 14:46:09 +01:00
Sven Luethi	55d891f38c	modified: src/netbox_plugin_bind_provisioner/management/commands/bind-transfer-endpoint.py Slight changes to previous commit: 1. Stop converting the zone ID to string for catalog zone PTR record creation and simply use its int type. 2. Extend padding of ID in catalog zone PTR record from 6 digits to 9. Auto-incrementing of zone IDs may create larger IDs then expected if the turn-over of zones is high. Precautionary measure. 3. Fix catalog zone SOA record creation - A slight mistake was introduced in previous commit that would render the set the MNAME field to "invalid.zones.catz." rather than the intended "invalid." value. 4. Run code through black formater to stay compliant.	2026-01-09 13:36:08 +01:00
Sven Luethi	1375bbb7e3	Merge pull request #9 from jpmens/ttl_mname Use TTL=0, alter MNAME,RNAME and rename PTR owners	2026-01-09 13:19:44 +01:00
Jan-Piet Mens	723566db70	Use TTL=0, alter MNAME,RNAME and rename PTR owners I've opted for a slightly shorter name for PTR owner names The result looks like this: ``` catz. 0 IN SOA invalid. invalid. 23 60 10 1209600 0 zid-000002.zones.catz. 0 IN PTR 2.0.192.in-addr.arpa. zid-000001.zones.catz. 0 IN PTR example.net. zid-000003.zones.catz. 0 IN PTR jpmens.net. zid-000004.zones.catz. 0 IN PTR tcp53.ch. catz. 0 IN NS invalid. version.catz. 0 IN TXT "2" ```	2026-01-09 08:53:59 +01:00
Sven Luethi	e8450c90da	modified: README.md modified: src/netbox_plugin_bind_provisioner/management/commands/bind-transfer-endpoint.py Remove code and description that imported old catz serials from flat files that were used before switch to DB model. Code and docs no longer applicable so removed.	2026-01-09 02:59:36 +01:00
Sven Luethi	ae6bb3c1b6	modified: README.md	2026-01-09 02:50:58 +01:00
Sven Luthi	0028ba8d80	new file: src/netbox_plugin_bind_provisioner/migrations/0002_alter_integerkeyvaluesetting_options_and_more.py modified: src/netbox_plugin_bind_provisioner/models.py Remove the settings model from Netbox's Permission system as these values are for internal use of the plugin only.	2025-12-18 15:53:38 +01:00
Sven Luthi	78ed040495	modified: README.md Update readme	2025-12-18 15:03:10 +01:00
Sven Luthi	20064ec9bd	modified: src/netbox_plugin_bind_provisioner/__init__.py new file: src/netbox_plugin_bind_provisioner/management/commands/export-all-dns-zones.py new file: src/netbox_plugin_bind_provisioner/management/commands/export-dns-zone.py new file: src/netbox_plugin_bind_provisioner/utils.py Adds export commands for zones to bind zone files and bump version to v1.0.4	2025-12-11 15:59:21 +01:00
Sven Luethi	6ac1548eef	modified: src/netbox_plugin_bind_provisioner/__init__.py modified: src/netbox_plugin_bind_provisioner/management/commands/bind-transfer-endpoint.py Improve TXT record formating.	2025-12-01 19:22:11 +01:00
Sven Luethi	d8f4b83c2a	modified: src/netbox_plugin_bind_provisioner/__init__.py modified: src/netbox_plugin_bind_provisioner/management/commands/bind-transfer-endpoint.py Allows for TXT records to be added without quotation. This will apply the correct quoting and chunking where needed.	2025-11-29 13:36:18 +01:00
Sven Luethi	fa032b1ea3	modified: src/netbox_plugin_bind_provisioner/__init__.py modified: src/netbox_plugin_bind_provisioner/management/commands/bind-transfer-endpoint.py Fixes bug: TXT records were truncated instead of chunked when larger then 255 characters and on semicolons.	2025-11-28 18:52:00 +01:00