Improved DNS record creation for IPAM coupling

* Removed the middleware code as it turned out to be too limited for the purpose. More
  specifically, it is not used for model-level changes to objects, so it doesn't work
  with Custom Scripts etc. (see issue #105)
* All checks and operations are now done in signal handlers for ipam.IPAddress that are
  part of NetBox DNS. Future releases (dropping support for NetBox < 3.7) could be
  using CUSTOM_VALIDATORS and PROTECTION_RULES instead.
* Testing was extended to cover model-level actions and API level actions to IPAddress
  objects.
* Improved DNS record permission checks.

Open issue:

* Although actions with missing permissions to delete DNS records work properly in the
  API and the GUI, the respective tests have to be skipped as they fail in the context
  of unit testing. Potential reasons include a bug in the test framework. This does not
  have any functional implications and just affects tests for the deletion of IPAddress
  objects with missing NetBox DNS (object) permissions.

netbox_dns/__init__.py

@@ -20,7 +20,6 @@ class DNSConfig(PluginConfig):
     version = __version__
     author = "Peter Eckel"
     author_email = "[email protected]"
-    middleware = ["netbox_dns.middleware.IpamCouplingMiddleware"]
     required_settings = []
     default_settings = {
         "zone_default_ttl": 86400,

netbox_dns/models/__init__.py

@@ -4,3 +4,5 @@
 from .view import *
 from .contact import *
 from .registrar import *
+
+from netbox_dns.signals import ipam_coupling

netbox_dns/signals/ipam_coupling.py

@@ -0,0 +1,149 @@
+from django.dispatch import receiver
+from django.db.models.signals import pre_save, post_save, pre_delete
+from django.core.exceptions import ValidationError, PermissionDenied
+from rest_framework.exceptions import PermissionDenied as APIPermissionDenied
+
+from netbox.signals import post_clean
+from netbox.context import current_request
+from ipam.models import IPAddress
+
+from netbox_dns.models import Zone
+from netbox_dns.utilities.ipam_coupling import (
+    ipaddress_cf_data,
+    get_address_record,
+    new_address_record,
+    update_address_record,
+    check_permission,
+    dns_changed,
+    DNSPermissionDenied,
+)
+
+try:
+    # NetBox 3.5.0 - 3.5.7, 3.5.9+
+    from extras.plugins import get_plugin_config
+except ImportError:
+    # NetBox 3.5.8
+    from extras.plugins.utils import get_plugin_config
+
+
+@receiver(post_clean, sender=IPAddress)
+def ip_address_check_permissions_save(instance, **kwargs):
+    if not get_plugin_config("netbox_dns", "feature_ipam_coupling"):
+        return
+
+    request = current_request.get()
+    if request is None:
+        return
+
+    try:
+        if instance.id is None:
+            record = new_address_record(instance)
+            if record is not None:
+                record.full_clean()
+                check_permission(request, "netbox_dns.add_record", record)
+
+        else:
+            if not dns_changed(IPAddress.objects.get(pk=instance.id), instance):
+                return
+
+            record = get_address_record(instance)
+            if record is not None:
+                name, zone_id = ipaddress_cf_data(instance)
+                if zone_id is not None:
+                    update_address_record(record, instance)
+                    record.full_clean()
+                    check_permission(request, "netbox_dns.change_record", record)
+                else:
+                    check_permission(request, "netbox_dns.delete_record", record)
+
+            else:
+                record = new_address_record(instance)
+                if record is not None:
+                    record.full_clean()
+                    check_permission(request, "netbox_dns.add_record", record)
+
+    except ValidationError as exc:
+        if hasattr(exc, "error_dict"):
+            value = exc.error_dict.pop("name", None)
+            if value is not None:
+                exc.error_dict["cf_ipaddress_dns_record_name"] = value
+
+            value = exc.error_dict.pop("value", None)
+            if value is not None:
+                exc.error_dict["cf_ipaddress_dns_record_name"] = value
+
+        raise ValidationError(exc)
+
+    except DNSPermissionDenied as exc:
+        raise ValidationError(exc)
+
+
+@receiver(pre_delete, sender=IPAddress)
+def ip_address_delete_address_record(instance, **kwargs):
+    if not get_plugin_config("netbox_dns", "feature_ipam_coupling"):
+        return
+
+    request = current_request.get()
+    if request is not None:
+        try:
+            for record in instance.netbox_dns_records.all():
+                check_permission(request, "netbox_dns.delete_record", record)
+
+        except DNSPermissionDenied as exc:
+            if request.path_info.startswith("/api/"):
+                raise APIPermissionDenied(exc) from None
+
+            raise PermissionDenied(exc) from None
+
+    for record in instance.netbox_dns_records.all():
+        record.delete()
+
+
+#
+# Update DNS related fields according to the contents of the IPAM-DNS
+# coupling custom fields.
+#
+@receiver(pre_save, sender=IPAddress)
+def ip_address_update_dns_information(instance, **kwargs):
+    if not get_plugin_config("netbox_dns", "feature_ipam_coupling"):
+        return
+
+    name, zone_id = ipaddress_cf_data(instance)
+
+    if zone_id is not None:
+        instance.dns_name = f"{name}.{Zone.objects.get(pk=zone_id).name}"
+    else:
+        instance.dns_name = ""
+        instance.custom_field_data["ipaddress_dns_record_name"] = None
+        instance.custom_field_data["ipaddress_dns_zone_id"] = None
+
+
+#
+# Handle DNS record operation after IPAddress has been created or modified
+#
+@receiver(post_save, sender=IPAddress)
+def ip_address_update_address_record(instance, **kwargs):
+    if not get_plugin_config("netbox_dns", "feature_ipam_coupling"):
+        return
+
+    name, zone_id = ipaddress_cf_data(instance)
+
+    if zone_id is None:
+        #
+        # Name/Zone CF data has been removed: Remove the DNS address record
+        #
+        for record in instance.netbox_dns_records.all():
+            record.delete()
+
+    else:
+        #
+        # Name/Zone CF data is present: Check for a DNS address record and add
+        # or modify it as necessary
+        #
+        record = get_address_record(instance)
+        if record is None:
+            record = new_address_record(instance)
+        else:
+            update_address_record(record, instance)
+
+        record.save()

netbox_dns/tables/zone.py

@@ -4,7 +4,6 @@
     ChoiceFieldColumn,
     NetBoxTable,
     TagColumn,
-    ActionsColumn,
 )
 from tenancy.tables import TenancyColumnsMixin