Merge pull request #240 from peteeckel/feature/arpa-network

Extend the database model with a CIDR field for .arpa zones

netbox_dns/fields/__init__.py

@@ -0,0 +1 @@
+from .network import *

netbox_dns/fields/network.py

@@ -0,0 +1,108 @@
+from django.db import models
+from django.db.models import Lookup
+
+from netaddr import AddrFormatError, IPNetwork
+
+
+class NetContains(Lookup):
+    lookup_name = "net_contains"
+
+    def as_sql(self, qn, connection):
+        lhs, lhs_params = self.process_lhs(qn, connection)
+        rhs, rhs_params = self.process_rhs(qn, connection)
+        params = lhs_params + rhs_params
+        return "%s >> %s" % (lhs, rhs), params
+
+
+class NetContainsOrEquals(Lookup):
+    lookup_name = "net_contains_or_equals"
+
+    def as_sql(self, qn, connection):
+        lhs, lhs_params = self.process_lhs(qn, connection)
+        rhs, rhs_params = self.process_rhs(qn, connection)
+        params = lhs_params + rhs_params
+        return "%s >>= %s" % (lhs, rhs), params
+
+
+class NetContained(Lookup):
+    lookup_name = "net_contained"
+
+    def as_sql(self, qn, connection):
+        lhs, lhs_params = self.process_lhs(qn, connection)
+        rhs, rhs_params = self.process_rhs(qn, connection)
+        params = lhs_params + rhs_params
+        return "%s << %s" % (lhs, rhs), params
+
+
+class NetContainedOrEqual(Lookup):
+    lookup_name = "net_contained_or_equal"
+
+    def as_sql(self, qn, connection):
+        lhs, lhs_params = self.process_lhs(qn, connection)
+        rhs, rhs_params = self.process_rhs(qn, connection)
+        params = lhs_params + rhs_params
+        return "%s <<= %s" % (lhs, rhs), params
+
+
+class NetworkFormField(models.Field):
+    def to_python(self, value):
+        if not value:
+            return None
+
+        if isinstance(value, IPNetwork):
+            return value
+
+        try:
+            ip_network = IPNetwork(value)
+        except AddrFormatError as exc:
+            raise ValidationError(exc)
+
+        return ip_network
+
+
+class NetworkField(models.Field):
+    description = "IPv4/v6 network associated with a reverse lookup zone"
+
+    def python_type(self):
+        return IPNetwork
+
+    def from_db_value(self, value, expression, connection):
+        return self.to_python(value)
+
+    def to_python(self, value):
+        if not value:
+            return value
+
+        try:
+            ip_network = IPNetwork(value)
+        except (AddressFormatError, TypeError, ValueError) as exc:
+            raise ValidationError(exc)
+
+        return ip_network
+
+    def get_prep_value(self, value):
+        if not value:
+            return None
+
+        if isinstance(value, list):
+            return [str(self.to_python(v)) for v in value]
+
+        return str(self.to_python(value))
+
+    def form_class(self):
+        return NetworkFormField
+
+    def formfield(self, **kwargs):
+        defaults = {"form_class": self.form_class()}
+        defaults.update(kwargs)
+
+        return super().formfield(**defaults)
+
+    def db_type(self, connection):
+        return "cidr"
+
+
+NetworkField.register_lookup(NetContains)
+NetworkField.register_lookup(NetContained)
+NetworkField.register_lookup(NetContainsOrEquals)
+NetworkField.register_lookup(NetContainedOrEqual)

netbox_dns/graphql/__init__.py

@@ -1,3 +1,14 @@
+import graphene
+from graphene_django.converter import convert_django_field
+
+from netbox_dns.fields import NetworkField
+
+
+@convert_django_field.register(NetworkField)
+def convert_field_to_string(field, registry=None):
+    return graphene.String(description=field.help_text, required=not field.null)
+
+
 from .schema import *
 
 from .view import *

netbox_dns/migrations/0018_zone_arpa_network.py

@@ -0,0 +1,52 @@
+# Generated by Django 4.0.8 on 2022-10-26 18:55
+
+from netaddr import IPNetwork, AddrFormatError
+
+from django.db import migrations
+import netbox_dns.fields.network
+
+
+def update_zone_arpa_network(apps, schema_editor):
+    Zone = apps.get_model("netbox_dns", "Zone")
+
+    for zone in Zone.objects.filter(name__endswith=".arpa"):
+        name = zone.name
+
+        if name.endswith(".in-addr.arpa"):
+            address = ".".join(reversed(name.replace(".in-addr.arpa", "").split(".")))
+            mask = len(address.split(".")) * 8
+
+            try:
+                zone.arpa_network = IPNetwork(f"{address}/{mask}")
+            except AddrFormatError:
+                uone.arpa_network = None
+
+        elif name.endswith("ip6.arpa"):
+            address = "".join(reversed(name.replace(".ip6.arpa", "").split(".")))
+            mask = len(address)
+            address = address + "0" * (32 - mask)
+
+            try:
+                zone.arpa_network = IPNetwork(
+                    f"{':'.join([(address[i:i+4]) for i in range(0, mask, 4)])}::/{mask*4}"
+                )
+            except AddrFormatError:
+                zone.arpa_network = None
+
+        zone.save()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("netbox_dns", "0017_alter_record_ttl"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="zone",
+            name="arpa_network",
+            field=netbox_dns.fields.network.NetworkField(blank=True, null=True),
+        ),
+        migrations.RunPython(update_zone_arpa_network),
+    ]

netbox_dns/models.py

@@ -7,6 +7,8 @@
 from dns import rdata, rdatatype, rdataclass
 from dns.rdtypes.ANY import SOA
 
+from netaddr import IPNetwork, AddrFormatError, IPAddress
+
 from django.core.validators import (
     MinValueValidator,
     MaxValueValidator,
@@ -28,6 +30,8 @@
 
 from netbox.models import NetBoxModel
 
+from netbox_dns.fields import NetworkField
+
 
 class NameServer(NetBoxModel):
     name = models.CharField(
@@ -169,6 +173,12 @@ class Zone(NetBoxModel):
         max_length=200,
         blank=True,
     )
+    arpa_network = NetworkField(
+        verbose_name="ARPA Network",
+        help_text="Network related to a reverse lookup zone (.arpa)",
+        blank=True,
+        null=True,
+    )
 
     objects = ZoneManager()
 
@@ -213,6 +223,20 @@ def get_absolute_url(self):
     def get_status_class(self):
         return self.CSS_CLASSES.get(self.status)
 
+    @property
+    def is_active(self):
+        return self.status in Zone.ACTIVE_STATUS_LIST
+
+    @property
+    def is_reverse_zone(self):
+        return self.name.endswith(".arpa")
+
+    @property
+    def view_filter(self):
+        if self.view is None:
+            return Q(view__isnull=True)
+        return Q(view=self.view)
+
     def update_soa_record(self):
         soa_name = "@"
         soa_ttl = self.soa_ttl
@@ -331,11 +355,32 @@ def update_serial(self):
         self.last_updated = datetime.now()
         self.save()
 
-    def parent_zones(self):
-        zone_fields = self.name.split(".")
-        return [
-            f'{".".join(zone_fields[length:])}' for length in range(1, len(zone_fields))
-        ]
+    def get_network(self):
+        name = self.name.rstrip(".")
+
+        if name.endswith(".in-addr.arpa"):
+            address = ".".join(reversed(name.replace(".in-addr.arpa", "").split(".")))
+            mask = len(address.split(".")) * 8
+
+            try:
+                return IPNetwork(f"{address}/{mask}")
+            except AddrFormatError:
+                return None
+
+        elif name.endswith("ip6.arpa"):
+            address = "".join(reversed(name.replace(".ip6.arpa", "").split(".")))
+            mask = len(address)
+            address = address + "0" * (32 - mask)
+
+            try:
+                return IPNetwork(
+                    f"{':'.join([(address[i:i+4]) for i in range(0, mask, 4)])}::/{mask*4}"
+                )
+            except AddrFormatError:
+                return None
+
+        else:
+            return None
 
     def check_name_conflict(self):
         if self.view is None:
@@ -366,33 +411,35 @@ def save(self, *args, **kwargs):
         new_zone = self.pk is None
         if not new_zone:
             old_zone = Zone.objects.get(pk=self.pk)
-            renamed_zone = old_zone.name != self.name
-            changed_view = old_zone.view != self.view
-            changed_status = old_zone.status != self.status
-        else:
-            renamed_zone = False
-            changed_view = False
-            changed_status = False
+
+        name_changed = not new_zone and old_zone.name != self.name
+        view_changed = not new_zone and old_zone.view != self.view
+        status_changed = not new_zone and old_zone.status != self.status
 
         if self.soa_serial_auto:
             self.soa_serial = self.get_auto_serial()
 
+        if self.is_reverse_zone:
+            self.arpa_network = self.get_network()
+
         super().save(*args, **kwargs)
 
         if (
-            new_zone or renamed_zone or changed_view or changed_status
-        ) and self.name.endswith(".arpa"):
+            new_zone or name_changed or view_changed or status_changed
+        ) and self.is_reverse_zone:
+            zones = Zone.objects.filter(
+                self.view_filter,
+                arpa_network__net_contains_or_equals=self.arpa_network,
+            )
             address_records = Record.objects.filter(
-                Q(ptr_record__isnull=True)
-                | Q(ptr_record__zone__name__in=self.parent_zones())
-                | Q(ptr_record__zone__name=self.name),
+                Q(ptr_record__isnull=True) | Q(ptr_record__zone__in=zones),
                 type__in=(RecordTypeChoices.A, RecordTypeChoices.AAAA),
                 disable_ptr=False,
             )
             for record in address_records:
                 record.update_ptr_record()
 
-        elif renamed_zone or changed_view or changed_status:
+        elif name_changed or view_changed or status_changed:
             for record in self.record_set.filter(
                 type__in=(RecordTypeChoices.A, RecordTypeChoices.AAAA)
             ):
@@ -592,26 +639,11 @@ def is_active(self):
             and self.zone.status in Zone.ACTIVE_STATUS_LIST
         )
 
-    def ptr_zone(self, view=None):
-        address = ipaddress.ip_address(self.value)
-        if address.version == 4:
-            lengths = range(1, 4)
-        else:
-            lengths = range(16, 32)
-
-        zone_names = [
-            ".".join(address.reverse_pointer.split(".")[length:]) for length in lengths
-        ]
-
-        if view is None:
-            view = self.zone.view
-
-        if view is None:
-            ptr_zone_filter = Q(name__in=zone_names, view__isnull=True)
-        else:
-            ptr_zone_filter = Q(name__in=zone_names, view_id=view.pk)
+    def ptr_zone(self):
+        ptr_zones = Zone.objects.filter(
+            self.zone.view_filter, arpa_network__net_contains=self.value
+        ).order_by(Length("name").desc())
 
-        ptr_zones = Zone.objects.filter(ptr_zone_filter).order_by(Length("name").desc())
         if len(ptr_zones):
             return ptr_zones[0]
 

netbox_dns/tests/record/test_auto_ptr.py

@@ -69,7 +69,8 @@ def setUpTestData(cls):
                 soa_mname=cls.nameserver,
             ),
         ]
-        Zone.objects.bulk_create(cls.zones)
+        for zone in cls.zones:
+            zone.save()
 
     def test_create_ipv4_ptr(self):
         f_zone = self.zones[0]

netbox_dns/tests/view/test_auto_ptr.py

@@ -99,7 +99,8 @@ def setUpTestData(cls):
                 view=cls.views[1],
             ),
         ]
-        Zone.objects.bulk_create(cls.zones)
+        for zone in cls.zones:
+            zone.save()
 
     def test_create_ipv4_ptr_no_view(self):
         f_zone = self.zones[0]

netbox_dns/tests/view/test_zone.py

@@ -96,7 +96,8 @@ def setUpTestData(cls):
                 view=cls.views[1],
             ),
         ]
-        Zone.objects.bulk_create(cls.zones)
+        for zone in cls.zones:
+            zone.save()
 
     def test_ipv4_add_view_to_zone_new_ptr_added(self):
         f_zone = self.zones[0]

netbox_dns/tests/zone/test_auto_soa_serial.py

@@ -53,7 +53,8 @@ def setUpTestData(cls):
                 soa_serial_auto=False,
             ),
         ]
-        Zone.objects.bulk_create(cls.zones)
+        for zone in cls.zones:
+            zone.save()
 
     def test_soa_serial_auto(self):
         zone = self.zones[0]