modified: netbox_dns_bridge/catalog_zone_manager.py

modified:   netbox_dns_bridge/dns_server.py
modified:   netbox_dns_bridge/management/commands/dns-export-zone.py
modified:   netbox_dns_bridge/management/commands/dns-export-zones.py
modified:   netbox_dns_bridge/management/commands/dns-settings.py
modified:   netbox_dns_bridge/management/commands/dns-transfer-endpoint.py
modified:   netbox_dns_bridge/migrations/0001_initial.py
modified:   netbox_dns_bridge/models.py
modified:   netbox_dns_bridge/request_handler.py
modified:   netbox_dns_bridge/signals.py
modified:   netbox_dns_bridge/utils.py
Black code formating corrected

netbox_dns_bridge/catalog_zone_manager.py

@@ -23,19 +23,19 @@ _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(
+        _SERIAL_OBJ = IntegerKeyValueSetting.objects.get(key="catalog-zone-soa-serial")
-            key="catalog-zone-soa-serial"
-        )
         logger.debug(
             f"Catalog zone SOA serial number {_SERIAL_OBJ.value} loaded from database"
         )
@@ -47,18 +47,20 @@ def _init_serial() -> None:
             f"Catalog zone SOA serial number was not set in the database. Set to {_SERIAL_OBJ.value}"
         )
 
+
 def _set_serial(new_serial: int) -> bool:
     global _SERIAL_OBJ
 
-    if (0 < new_serial < _SERIAL_MAX):
+    if 0 < new_serial < _SERIAL_MAX:
         _SERIAL_OBJ.value = new_serial
         _SERIAL_OBJ.save()
         return True
     else:
         return False
 
+
 def _increment_serial() -> None:
-    if not _set_serial( _SERIAL_OBJ.value + 1 ):
+    if not _set_serial(_SERIAL_OBJ.value + 1):
         logger.warning(
             f"Failed to incremenet catalog serial {_SERIAL_OBJ.value}. Will overflow serial back to 1"
         )
@@ -67,29 +69,32 @@ def _increment_serial() -> None:
             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(
+    existing_zone_ids = CatalogZoneMemberIdentifier.objects.values_list(
-       "zone_id", flat=True
+        "zone_id", flat=True
-   )
+    )
 
-   missing_zones = Zone.objects.exclude(id__in=existing_zone_ids)
+    missing_zones = Zone.objects.exclude(id__in=existing_zone_ids)
 
-   new_objects = [
+    new_objects = [
-       CatalogZoneMemberIdentifier(
+        CatalogZoneMemberIdentifier(
-           zone=zone,
+            zone=zone,
-           name=_generate_member_identifier(),
+            name=_generate_member_identifier(),
-       )
+        )
-       for zone in missing_zones
+        for zone in missing_zones
-   ]
+    ]
 
-   for identifier in new_objects:
+    for identifier in new_objects:
-       logger.debug(f"Zone {identifier.zone} has no catz member identifier. Creating...")
+        logger.debug(
+            f"Zone {identifier.zone} has no catz member identifier. Creating..."
+        )
 
-   CatalogZoneMemberIdentifier.objects.bulk_create(
+    CatalogZoneMemberIdentifier.objects.bulk_create(
-       new_objects,
+        new_objects,
-       ignore_conflicts=False,
+        ignore_conflicts=False,
-   )
+    )
 
 
 def create_zone(name, view_name) -> dns.zone.Zone:
@@ -134,14 +139,14 @@ def create_zone(name, view_name) -> dns.zone.Zone:
         qname = dns.name.from_text(nb_zone.name, dns.name.root)
 
         # Create PTR record
-        #p_name = f"zid-{nb_zone.id:09d}"
+        # 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()
         )
@@ -196,9 +201,7 @@ def create_zone(name, view_name) -> dns.zone.Zone:
 
     # NS record for catz.
     ns_name = dns.name.from_text("invalid", dns.name.root)
-    ns_rdata = dns.rdata.from_text(
+    ns_rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.NS, str(ns_name))
-        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)
 
@@ -207,9 +210,7 @@ def create_zone(name, view_name) -> dns.zone.Zone:
     ns_node.rdatasets.append(ns_rdataset)
 
     # TXT record for version.catz.
-    version_name = dns.name.from_text(
+    version_name = dns.name.from_text("version", origin)  # relative to origin "catz."
-        "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)
@@ -221,8 +222,10 @@ def create_zone(name, view_name) -> dns.zone.Zone:
 
     return zone
 
+
 def _generate_member_identifier() -> None:
-    return b32encode(uuid4().bytes)[0:26].lower().decode('UTF-8')
+    return b32encode(uuid4().bytes)[0:26].lower().decode("UTF-8")
+
 
 def update_member_identifier(zone: Zone) -> None:
     CatalogZoneMemberIdentifier.objects.update_or_create(

netbox_dns_bridge/dns_server.py

@@ -4,17 +4,13 @@ import socket
 
 logger = logging.getLogger(__name__)
 
+
 class DNSAddressMixin:
     def _resolve_address(self, server_address, socktype, proto):
         host, port = server_address
 
         infos = socket.getaddrinfo(
-            host,
+            host, port, socket.AF_UNSPEC, socktype, proto, socket.AI_PASSIVE
-            port,
-            socket.AF_UNSPEC,
-            socktype,
-            proto,
-            socket.AI_PASSIVE
         )
 
         family, _, _, _, sockaddr = infos[0]
@@ -27,9 +23,7 @@ class TCPDNSServer(DNSAddressMixin, socketserver.TCPServer):
 
     def __init__(self, server_address, handler_class, keyring, tsig_view_map):
         sockaddr = self._resolve_address(
-            server_address,
+            server_address, socket.SOCK_STREAM, socket.IPPROTO_TCP
-            socket.SOCK_STREAM,
-            socket.IPPROTO_TCP
         )
 
         super().__init__(sockaddr, handler_class)
@@ -43,9 +37,7 @@ class UDPDNSServer(DNSAddressMixin, socketserver.UDPServer):
 
     def __init__(self, server_address, handler_class, keyring, tsig_view_map):
         sockaddr = self._resolve_address(
-            server_address,
+            server_address, socket.SOCK_DGRAM, socket.IPPROTO_UDP
-            socket.SOCK_DGRAM,
-            socket.IPPROTO_UDP
         )
 
         super().__init__(sockaddr, handler_class)

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

@@ -7,11 +7,14 @@ 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"
+            "--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"
@@ -20,16 +23,15 @@ class Command(BaseCommand):
             "--file", type=str, help="Path of the zone file to be written"
         )
 
-
     def handle(self, *args, **options):
-        #if len(options) < 2:
+        # if len(options) < 2:
         #    print("export-zone <zone name> <file path>")
         #    sys.exit(1)
 
         # Load parameters
-        view_name = options['view']
+        view_name = options["view"]
-        zone_name = options['zone']
+        zone_name = options["zone"]
-        file_path = options['file']
+        file_path = options["file"]
 
         if not view_name:
             raise CommandError("No --view parameter given")
@@ -40,7 +42,9 @@ class Command(BaseCommand):
 
         try:
             # Load the zone from NetBox DNS
-            nb_zone = netbox_dns.models.Zone.objects.get(view__name=view_name, name=zone_name)
+            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)
@@ -51,6 +55,6 @@ class Command(BaseCommand):
         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}'"))
+        self.stdout.write(
-
+            self.style.SUCCESS(f"Zone '{zone_name}' exported to '{file_path}'")
-
+        )

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

@@ -7,16 +7,19 @@ 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"
+            "--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']
+        base_path = options["path"]
 
         if not base_path:
             raise CommandError("No --path parameter given")
@@ -42,6 +45,8 @@ class Command(BaseCommand):
             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}'"))
+                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/dns-settings.py

@@ -6,9 +6,13 @@ class Command(BaseCommand):
     help = "Get or set integer key-value settings"
 
     def add_arguments(self, parser):
-        parser.add_argument("action", choices=["get", "set", "list"], help="Action to perform")
+        parser.add_argument(
+            "action", choices=["get", "set", "list"], help="Action to perform"
+        )
         parser.add_argument("key", nargs="?", help="Setting key")
-        parser.add_argument("value", nargs="?", type=int, help="Value to set (required for 'set')")
+        parser.add_argument(
+            "value", nargs="?", type=int, help="Value to set (required for 'set')"
+        )
 
     def handle(self, *args, **options):
         action = options["action"]

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

@@ -21,13 +21,12 @@ from netbox_dns_bridge.models import IntegerKeyValueSetting
 
 logger = logging.getLogger(__name__)
 
+
 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(
+        self.settings = settings.PLUGINS_CONFIG.get("netbox_dns_bridge", None)
-            "netbox_dns_bridge", None
-        )
         if not self.settings:
             raise RuntimeError(
                 "Command failed to initialize due to missing settings. Terminating Netbox."

netbox_dns_bridge/migrations/0001_initial.py

@@ -9,30 +9,50 @@ class Migration(migrations.Migration):
     initial = True
 
     dependencies = [
-        ('netbox_dns', '0030_dnsseckeytemplate_comments_dnsseckeytemplate_owner_and_more'),
+        (
+            "netbox_dns",
+            "0030_dnsseckeytemplate_comments_dnsseckeytemplate_owner_and_more",
+        ),
     ]
 
     operations = [
         migrations.CreateModel(
-            name='IntegerKeyValueSetting',
+            name="IntegerKeyValueSetting",
             fields=[
-                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False)),
+                (
-                ('key', models.CharField(max_length=64)),
+                    "id",
-                ('value', models.IntegerField()),
+                    models.BigAutoField(
+                        auto_created=True, primary_key=True, serialize=False
+                    ),
+                ),
+                ("key", models.CharField(max_length=64)),
+                ("value", models.IntegerField()),
             ],
             options={
-                'default_permissions': (),
+                "default_permissions": (),
             },
         ),
         migrations.CreateModel(
-            name='CatalogZoneMemberIdentifier',
+            name="CatalogZoneMemberIdentifier",
             fields=[
-                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False)),
+                (
-                ('name', models.CharField(max_length=26, unique=True)),
+                    "id",
-                ('zone', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='catz_identifier', to='netbox_dns.zone')),
+                    models.BigAutoField(
+                        auto_created=True, primary_key=True, serialize=False
+                    ),
+                ),
+                ("name", models.CharField(max_length=26, unique=True)),
+                (
+                    "zone",
+                    models.OneToOneField(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="catz_identifier",
+                        to="netbox_dns.zone",
+                    ),
+                ),
             ],
             options={
-                'ordering': ('name',),
+                "ordering": ("name",),
             },
         ),
     ]

netbox_dns_bridge/models.py

@@ -2,6 +2,7 @@ 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()
@@ -12,6 +13,7 @@ class IntegerKeyValueSetting(models.Model):
     class Meta:
         default_permissions = ()
 
+
 class CatalogZoneMemberIdentifier(models.Model):
     name = models.CharField(
         max_length=26,

netbox_dns_bridge/request_handler.py

@@ -114,7 +114,7 @@ class DNSBaseRequestHandler(socketserver.BaseRequestHandler):
                 zone.replace_rdataset(name, rdataset)
         return zone
 
-    def _denyRequestBadTSIG(self, wire, tsig_error: dns.rcode) -> None:
+    def _deny_request_bad_tsig(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
@@ -131,7 +131,7 @@ class DNSBaseRequestHandler(socketserver.BaseRequestHandler):
                 keyname=query.keyname,
                 tsig_error=tsig_error,
             )
-        self._denyRequest(query)
+        self._deny_request(query)
 
     def _send_response(self, data) -> None:
         raise NotImplementedError
@@ -273,19 +273,17 @@ class DNSBaseRequestHandler(socketserver.BaseRequestHandler):
                 wire,
                 keyring=self.server.keyring,
                 continue_on_error=False,
-                ignore_trailing=True
+                ignore_trailing=True,
             )
 
         except dns.tsig.BadSignature as e:
-            logger.warning(
+            logger.warning(f"Request denied from {peer} failed TSIG verification: {e}")
-                f"Request denied from {peer} failed TSIG verification: {e}"
+            self._deny_request_bad_tsig(wire, dns.rcode.BADSIG)
-            )
-            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)
+            self._deny_request_bad_tsig(wire, dns.rcode.BADKEY)
             return
 
         except Exception as e:
@@ -366,6 +364,7 @@ class UDPRequestHandler(DNSBaseRequestHandler):
         except Exception as e:
             logger.error(f"Error handling request from {peer}: {e}")
             import traceback
+
             traceback.print_exc()
 
 
@@ -409,5 +408,5 @@ class TCPRequestHandler(DNSBaseRequestHandler):
         except Exception as e:
             logger.error(f"Error handling request from {peer}: {e}")
             import traceback
-            traceback.print_exc()
 
+            traceback.print_exc()

netbox_dns_bridge/signals.py

@@ -3,6 +3,7 @@ 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):
     """
@@ -10,17 +11,13 @@ def zone_pre_save(sender, instance, **kwargs):
     """
     if instance.pk:
         try:
-            instance._old_name = (
+            instance._old_name = sender.objects.only("name").get(pk=instance.pk).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):
     """

netbox_dns_bridge/utils.py

@@ -5,6 +5,7 @@ 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
@@ -32,4 +33,3 @@ def export_bind_zone_file(nb_zone: netbox_dns.models.Zone, file_path: str):
             dp_zone.to_file(f, sorted=True)
     except IOError as e:
         raise IOError(f"Failed to write zone file to {file_path}: {e}")
-