diff --git a/docs/images/IPAMCouplingCustomFields.png b/docs/images/IPAMCouplingCustomFields.png
new file mode 100644
index 00000000..6feae287
Binary files /dev/null and b/docs/images/IPAMCouplingCustomFields.png differ
diff --git a/docs/images/IPAMCouplingRecordDetailView.png b/docs/images/IPAMCouplingRecordDetailView.png
new file mode 100644
index 00000000..381e8927
Binary files /dev/null and b/docs/images/IPAMCouplingRecordDetailView.png differ
diff --git a/docs/images/IPAMCouplingRelatedAddressRecord.png b/docs/images/IPAMCouplingRelatedAddressRecord.png
new file mode 100644
index 00000000..0bbad3f9
Binary files /dev/null and b/docs/images/IPAMCouplingRelatedAddressRecord.png differ
diff --git a/docs/images/IPAMCouplingRelatedPointerRecord.png b/docs/images/IPAMCouplingRelatedPointerRecord.png
new file mode 100644
index 00000000..6a8929d8
Binary files /dev/null and b/docs/images/IPAMCouplingRelatedPointerRecord.png differ
diff --git a/docs/using_netbox_dns.md b/docs/using_netbox_dns.md
index d185bab9..41031b28 100644
--- a/docs/using_netbox_dns.md
+++ b/docs/using_netbox_dns.md
@@ -426,4 +426,69 @@ The NetBox detail view for tenants shows a table of NetBox DNS objects assigned
-The colums of the table on the left side are clickable and link to filtered lists showing the related views, nameservers, zones and records.
\ No newline at end of file
+The colums of the table on the left side are clickable and link to filtered lists showing the related views, nameservers, zones and records.
+
+## IPAM Coupling
+
+Starting with NetBox DNS 0.20.0, a new experimental feature providing coupling between NetBox DNS and NetBox IPAM data is available. This feature can be used to link IP addresses in IPAM to NetBox DNS address records. The old IPAM integration feature was dropped in favour of the new and improved functionality.
+
+Thanks to Jean Benoît for this contribution!
+
+### Enabling IPAM Coupling
+
+The new experimental feature needs to be enabled in the NetBox configuration file by setting its flag:
+
+```
+PLUGINS_CONFIG = {
+ 'netbox_dns': {
+ ...
+ 'feature_ipam_coupling': True,
+ ...
+ },
+}
+```
+
+In addition, two custom fields on `ipam.IPAddress` objects are required for the feature to work. These custom fields can be created using the Django management command `setup_coupling`:
+
+```
+/opt/netbox/netbox/manage.py setup_coupling
+```
+
+In order to remove the custom fields and all related data, the same command can be used with the option `--remove`.
+
+After these steps, a restart of NetBox is required.
+
+### Using IPAM Coupling
+
+With the new custom fields it is possible to automatically generate a DNS address record for an IP address. To do this, define a name for the record in the 'Name' custom field and select a zone in the 'Zone' custom in the DNS group.
+
+
+
+When the IP address is saved, NetBox DNS now automatically creates a managed address record for it in the selected zone, using the name from the 'Name' custom field. The 'DNS Name' field for the IP address is set to the FQDN of the resulting address record.
+
+The IP address is now linked to the address record in the following ways:
+
+* When one of the custom fields for the IP address is updated, the DNS record is updated as well. This includes changing the name as well as moving it to a different DNS zone
+* When the IP address is deleted, the managed DNS record is deleted as well
+* When the DNS zone is renamed, the 'DNS Name' for the IP address is updated to reflect the zone's new name
+* When the DNS zone is deleted, the address record is deleted and the connection from the IP address object is cleared
+
+### Additional Information for IP Addresses and DNS Records
+
+When a link between an IP address and a DNS address record is present, there are some additional panes in the IPAM IP address and NetBox DNS record view, as well as in the detail views for NetBox DNS managed records.
+
+#### IP Address Information
+
+If a DNS address record is linked to an IP address, the detail view for the IP address contains an additional pane showing that address record.
+
+
+
+If NetBox DNS also created a PTR record for the linked DNS address record, the detail view for the IP address contains an a second additional pane showing that pointer record.
+
+
+
+#### DNS Record Information
+
+The detail views for the address and pointer records created for coupled IP addresses include a link to that IP address, which can be used to navigate to the address.
+
+
diff --git a/netbox_dns/__init__.py b/netbox_dns/__init__.py
index d49b3f38..92582360 100644
--- a/netbox_dns/__init__.py
+++ b/netbox_dns/__init__.py
@@ -1,4 +1,7 @@
from extras.plugins import PluginConfig
+import logging
+
+logger = logging.getLogger("netbox.config")
__version__ = "0.19.4"
@@ -11,6 +14,7 @@ class DNSConfig(PluginConfig):
version = __version__
author = "Peter Eckel"
author_email = "pe-netbox-plugin-dns@hindenburgring.com"
+ middleware = ["netbox_dns.middleware.IpamCouplingMiddleware"]
required_settings = []
default_settings = {
"zone_default_ttl": 86400,
@@ -20,7 +24,7 @@ class DNSConfig(PluginConfig):
"zone_soa_retry": 7200,
"zone_soa_expire": 2592000,
"zone_soa_minimum": 3600,
- "feature_ipam_integration": False,
+ "feature_ipam_coupling": False,
"tolerate_underscores_in_hostnames": False,
"tolerate_leading_underscore_types": [
"TXT",
@@ -32,5 +36,32 @@ class DNSConfig(PluginConfig):
}
base_url = "netbox-dns"
+ def ready(self):
+ #
+ # Check if required custom field exist for IPAM coupling
+ #
+ if self.default_settings["feature_ipam_coupling"]:
+ from extras.models import CustomField
+ from ipam.models import IPAddress
+ from django.contrib.contenttypes.models import ContentType
+
+ objtype = ContentType.objects.get_for_model(IPAddress)
+ required_cf = ("ipaddress_dns_record_name", "ipaddress_dns_zone_id")
+
+ if CustomField.objects.filter(
+ name__in=required_cf, content_types=objtype
+ ).count() < len(required_cf):
+ logger.warning(
+ "'feature_ipam_coupling' is enabled, but the required custom"
+ " fields for IPAM-DNS coupling are missing. Please run the"
+ " Django management command 'setup_coupling' to create the"
+ " custom fields."
+ )
+
+ super().ready()
+
+#
+# Initialize plugin config
+#
config = DNSConfig
diff --git a/netbox_dns/api/serializers.py b/netbox_dns/api/serializers.py
index 518f5116..78395b07 100644
--- a/netbox_dns/api/serializers.py
+++ b/netbox_dns/api/serializers.py
@@ -1,6 +1,7 @@
from rest_framework import serializers
from netbox.api.serializers import NetBoxModelSerializer
+from ipam.api.nested_serializers import NestedIPAddressSerializer
from tenancy.api.nested_serializers import NestedTenantSerializer
from netbox_dns.api.nested_serializers import (
@@ -168,6 +169,13 @@ class RecordSerializer(NetBoxModelSerializer):
required=False,
read_only=True,
)
+ ipam_ip_address = NestedIPAddressSerializer(
+ many=False,
+ read_only=True,
+ required=False,
+ allow_null=True,
+ help_text="IPAddress linked to the record",
+ )
tenant = NestedTenantSerializer(required=False, allow_null=True)
class Meta:
@@ -193,4 +201,5 @@ class Meta:
"active",
"custom_fields",
"tenant",
+ "ipam_ip_address",
)
diff --git a/netbox_dns/management/commands/setup_coupling.py b/netbox_dns/management/commands/setup_coupling.py
new file mode 100644
index 00000000..d2531c08
--- /dev/null
+++ b/netbox_dns/management/commands/setup_coupling.py
@@ -0,0 +1,75 @@
+from django.core.management.base import BaseCommand, CommandError
+
+from django.contrib.contenttypes.models import ContentType
+from extras.models import CustomField
+from extras.choices import CustomFieldTypeChoices
+from ipam.models import IPAddress
+from netbox_dns.models import Record, Zone
+
+
+class Command(BaseCommand):
+ help = "Setup IPAddress custom fields needed for IPAM-DNS coupling"
+
+ def add_arguments(self, parser):
+ parser.add_argument(
+ "--remove", action="store_true", help="Remove custom fields"
+ )
+
+ def handle(self, *model_names, **options):
+ ipaddress_object_type = ContentType.objects.get_for_model(IPAddress)
+ zone_object_type = ContentType.objects.get_for_model(Zone)
+ record_object_type = ContentType.objects.get_for_model(Record)
+ customfields = ("ipaddress_dns_record_name", "ipaddress_dns_zone_id")
+
+ if options["remove"]:
+ for cf in customfields:
+ try:
+ CustomField.objects.get(
+ name=cf, content_types=ipaddress_object_type
+ ).delete()
+ except:
+ self.stderr.write(f"Custom field '{cf}' does not exist!")
+ else:
+ self.stdout.write(f"Custom field '{cf}' removed")
+
+ else:
+ msg = ""
+ for cf in customfields:
+ try:
+ CustomField.objects.get(
+ name=cf, content_types=ipaddress_object_type
+ )
+ except:
+ pass
+ else:
+ msg += f"custom fields '{cf}' already exists, "
+ if msg != "":
+ raise CommandError(
+ "\n".join(
+ (
+ "Can't setup IPAM-DNS coupling:",
+ msg,
+ "Remove them with NetBox command:",
+ "python manage.py setup_coupling --remove",
+ )
+ )
+ )
+
+ cf_name = CustomField.objects.create(
+ name="ipaddress_dns_record_name",
+ label="Name",
+ type=CustomFieldTypeChoices.TYPE_TEXT,
+ required=False,
+ group_name="DNS",
+ )
+ cf_name.content_types.set([ipaddress_object_type])
+ cf_zone = CustomField.objects.create(
+ name="ipaddress_dns_zone_id",
+ label="Zone",
+ type=CustomFieldTypeChoices.TYPE_OBJECT,
+ object_type=zone_object_type,
+ required=False,
+ group_name="DNS",
+ )
+ cf_zone.content_types.set([ipaddress_object_type])
+ self.stdout.write(f"Custom fields for IPAM-DNS coupling added")
diff --git a/netbox_dns/middleware.py b/netbox_dns/middleware.py
new file mode 100644
index 00000000..032e0d15
--- /dev/null
+++ b/netbox_dns/middleware.py
@@ -0,0 +1,190 @@
+from django.db import transaction
+from django.db.models import signals
+from django.core.exceptions import MiddlewareNotUsed, PermissionDenied
+
+from ipam.models import IPAddress
+from extras.plugins import get_plugin_config
+from netbox_dns.models import Zone, Record, RecordTypeChoices
+from utilities.exceptions import PermissionsViolation, AbortRequest
+from utilities.permissions import resolve_permission
+
+
+class Action:
+ def __init__(self, request):
+ self.request = request
+
+ #
+ # Check permission to create DNS record before IP address creation
+ # NB: If IP address is created *before* DNS record is allowed it's too late
+ # → permission check must be done at pre-save, and an exception
+ # must be raised to prevent IP creation.
+ #
+ def pre_save(self, sender, **kwargs):
+ if kwargs.get("update_fields"):
+ return
+
+ ip_address = kwargs.get("instance")
+ name = ip_address.custom_field_data.get("ipaddress_dns_record_name")
+ zone_id = ip_address.custom_field_data.get("ipaddress_dns_zone_id")
+
+ # Handle new IPAddress objects only; name and zone must both be defined
+ if ip_address.id is None and name is not None and zone_id is not None:
+ zone = Zone.objects.get(id=zone_id)
+ type = (
+ RecordTypeChoices.AAAA
+ if ip_address.family == 6
+ else RecordTypeChoices.A
+ )
+ value = str(ip_address.address.ip)
+
+ # Create a DNS record *without saving* in order to check permissions
+ record = Record(name=name, zone=zone, type=type, value=value)
+ user = self.request.user
+ check_record_permission(user, record, "netbox_dns.add_record")
+
+ #
+ # Handle DNS record operation after IPAddress has been created or modified
+ #
+ def post_save(self, sender, **kwargs):
+ # Do not process specific field update (eg. dns_hostname modify)
+ if kwargs.get("update_fields"):
+ return
+
+ ip_address = kwargs.get("instance")
+ user = self.request.user
+ name = ip_address.custom_field_data.get("ipaddress_dns_record_name")
+ zone_id = ip_address.custom_field_data.get("ipaddress_dns_zone_id")
+ zone = Zone.objects.get(id=zone_id) if zone_id is not None else None
+
+ # Clear the other field if one is empty, which is inconsistent
+ if name is None:
+ zone = None
+ elif zone is None:
+ name = None
+
+ # Delete the DNS record because name and zone have been removed
+ if zone is None:
+ # Find the record pointing to this IP Address
+ for record in ip_address.netbox_dns_records.all():
+ # If permission ok, clear all fields related to DNS
+ check_record_permission(user, record, "netbox_dns.delete_record")
+
+ ip_address.dns_name = ""
+ ip_address.custom_field_data["ipaddress_dns_record_name"] = ""
+ ip_address.save(update_fields=["custom_field_data", "dns_name"])
+
+ record.delete()
+
+ # Modify or add the DNS record
+ else:
+ type = (
+ RecordTypeChoices.AAAA
+ if ip_address.family == 6
+ else RecordTypeChoices.A
+ )
+
+ # If DNS record already point to this IP, modify it
+ record = ip_address.netbox_dns_records.first()
+ if record is not None:
+ record.name = name
+ record.zone = zone
+ record.value = str(ip_address.address.ip)
+ record.type = type
+
+ check_record_permission(user, record, "netbox_dns.change_record")
+ record.save()
+
+ else:
+ # Create a new record
+ record = Record(
+ name=name,
+ zone=zone,
+ type=type,
+ value=str(ip_address.address.ip),
+ ipam_ip_address=ip_address,
+ managed=True,
+ )
+
+ check_record_permission(
+ user, record, "netbox_dns.add_record", commit=True
+ )
+
+ # Update the dns_name field with FQDN
+ ip_address.dns_name = record.fqdn.rstrip(".")
+ ip_address.save(update_fields=["dns_name"])
+
+ #
+ # Delete DNS record before deleting IP address
+ #
+ def pre_delete(self, sender, **kwargs):
+ ip_address = kwargs.get("instance")
+
+ for record in ip_address.netbox_dns_records.all():
+ user = self.request.user
+ check_record_permission(user, record, "netbox_dns.delete_record")
+
+ record.delete()
+
+
+#
+# Filter through permissions. Simulate adding the record in the "add" case.
+# NB: Side-effect if "commit" is set to True → the DNS record is created.
+# This is necessary to avoid the cascading effects of PTR creation.
+#
+def check_record_permission(user, record, perm, commit=False):
+ # Check that the user has been granted the required permission(s).
+ action = resolve_permission(perm)[1]
+
+ if not user.has_perm(perm):
+ raise PermissionDenied()
+
+ try:
+ with transaction.atomic():
+ # Save record when adding
+ # Rollback is done at the end of the transaction, unless committed
+
+ if action == "add":
+ record.save()
+
+ # Update the view's QuerySet to filter only the permitted objects
+ queryset = Record.objects.restrict(user, action)
+ # Check that record conforms to permissions
+ # → must be included in the restricted queryset
+ if not queryset.filter(pk=record.pk).exists():
+ raise PermissionDenied()
+
+ if not commit:
+ raise AbortRequest("Normal Exit")
+
+ # Catch "Normal Exit" without modification, rollback transaction
+ except AbortRequest as exc:
+ pass
+
+ except Exception as exc:
+ raise exc
+
+
+class IpamCouplingMiddleware:
+ def __init__(self, get_response):
+ if not get_plugin_config("netbox_dns", "feature_ipam_coupling"):
+ raise MiddlewareNotUsed
+
+ self.get_response = get_response
+
+ def __call__(self, request):
+ # connect signals to actions
+ action = Action(request)
+ connections = [
+ (signals.pre_save, action.pre_save),
+ (signals.post_save, action.post_save),
+ (signals.pre_delete, action.pre_delete),
+ ]
+ for signal, receiver in connections:
+ signal.connect(receiver, sender=IPAddress)
+
+ response = self.get_response(request)
+
+ for signal, receiver in connections:
+ signal.disconnect(receiver)
+
+ return response
diff --git a/netbox_dns/migrations/0025_ipam_coupling_cf.py b/netbox_dns/migrations/0025_ipam_coupling_cf.py
new file mode 100644
index 00000000..a9436fa0
--- /dev/null
+++ b/netbox_dns/migrations/0025_ipam_coupling_cf.py
@@ -0,0 +1,22 @@
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+ dependencies = [
+ ("ipam", "0066_iprange_mark_utilized"),
+ ("netbox_dns", "0024_tenancy"),
+ ]
+ operations = [
+ migrations.AddField(
+ model_name="record",
+ name="ipam_ip_address",
+ field=models.ForeignKey(
+ blank=True,
+ null=True,
+ on_delete=django.db.models.deletion.CASCADE,
+ related_name="netbox_dns_records",
+ to="ipam.ipaddress",
+ ),
+ ),
+ ]
diff --git a/netbox_dns/models.py b/netbox_dns/models.py
index fcab52bb..9d0e731d 100644
--- a/netbox_dns/models.py
+++ b/netbox_dns/models.py
@@ -9,7 +9,7 @@
from dns.rdtypes.ANY import SOA
from dns.exception import DNSException
-from netaddr import IPNetwork, AddrFormatError, IPAddress
+from ipam.models import IPAddress
from django.core.validators import (
MinValueValidator,
@@ -556,6 +556,17 @@ def save(self, *args, **kwargs):
):
record.update_ptr_record()
+ # Fix name in IP Address when zone name is changed
+ if (
+ get_plugin_config("netbox_dns", "feature_ipam_coupling")
+ and name_changed
+ ):
+ for ip in IPAddress.objects.filter(
+ custom_field_data__ipaddress_dns_zone_id=self.pk
+ ):
+ ip.dns_name = f'{ip.custom_field_data["ipaddress_dns_record_name"]}.{self.name}'
+ ip.save(update_fields=["dns_name"])
+
self.update_soa_record()
def delete(self, *args, **kwargs):
@@ -570,6 +581,16 @@ def delete(self, *args, **kwargs):
for record in Record.objects.filter(ptr_record__in=ptr_records)
]
+ if get_plugin_config("netbox_dns", "feature_ipam_coupling"):
+ # Remove coupling from IPAddress to DNS record when zone is deleted
+ for ip in IPAddress.objects.filter(
+ custom_field_data__ipaddress_dns_zone_id=self.pk
+ ):
+ ip.dns_name = ""
+ ip.custom_field_data["ipaddress_dns_record_name"] = ""
+ ip.custom_field_data["ipaddress_dns_zone_id"] = None
+ ip.save(update_fields=["dns_name", "custom_field_data"])
+
super().delete(*args, **kwargs)
for record in Record.objects.filter(pk__in=update_records):
@@ -729,7 +750,14 @@ class Record(NetBoxModel):
blank=True,
null=True,
)
-
+ ipam_ip_address = models.ForeignKey(
+ verbose_name="IPAM IP Address",
+ to="ipam.IPAddress",
+ on_delete=models.CASCADE,
+ related_name="netbox_dns_records",
+ blank=True,
+ null=True,
+ )
objects = RecordManager()
raw_objects = RestrictedQuerySet.as_manager()
diff --git a/netbox_dns/tables/record.py b/netbox_dns/tables/record.py
index f59fde5d..0552790d 100644
--- a/netbox_dns/tables/record.py
+++ b/netbox_dns/tables/record.py
@@ -92,6 +92,10 @@ class ManagedRecordTable(RecordBaseTable):
verbose_name="Address Record",
linkify=True,
)
+ ipam_ip_address = tables.Column(
+ verbose_name="IPAM IP Address",
+ linkify=True,
+ )
actions = ActionsColumn(actions=("changelog",))
class Meta(NetBoxTable.Meta):
@@ -104,6 +108,7 @@ class Meta(NetBoxTable.Meta):
"value",
"unicode_value",
"address_record",
+ "ipam_ip_address",
"active",
)
default_columns = (
diff --git a/netbox_dns/template_content.py b/netbox_dns/template_content.py
index 35975177..bbdb6494 100644
--- a/netbox_dns/template_content.py
+++ b/netbox_dns/template_content.py
@@ -4,28 +4,35 @@
from extras.plugins import PluginTemplateExtension
from netbox_dns.models import Record, RecordTypeChoices, Zone, View, NameServer
-from netbox_dns.tables import RelatedRecordTable, RelatedZoneTable
+from netbox_dns.tables import RelatedRecordTable
class RelatedDNSRecords(PluginTemplateExtension):
model = "ipam.ipaddress"
def right_page(self):
- obj = self.context.get("object")
-
- address_records = Record.objects.filter(
- ip_address=obj.address.ip,
- type__in=(RecordTypeChoices.A, RecordTypeChoices.AAAA),
- )
- pointer_records = Record.objects.filter(
- ip_address=obj.address.ip, type=RecordTypeChoices.PTR
- )
- address_record_table = RelatedRecordTable(
- data=address_records,
- )
- pointer_record_table = RelatedRecordTable(
- data=pointer_records,
- )
+ ip_address = self.context.get("object")
+
+ address_records = ip_address.netbox_dns_records.all()
+ pointer_records = [
+ address_record.ptr_record
+ for address_record in address_records
+ if address_record.ptr_record is not None
+ ]
+
+ if address_records:
+ address_record_table = RelatedRecordTable(
+ data=address_records,
+ )
+ else:
+ address_record_table = None
+
+ if pointer_records:
+ pointer_record_table = RelatedRecordTable(
+ data=pointer_records,
+ )
+ else:
+ pointer_record_table = None
return self.render(
"netbox_dns/record/related.html",
@@ -36,31 +43,6 @@ def right_page(self):
)
-class RelatedDNSPointerZones(PluginTemplateExtension):
- model = "ipam.prefix"
-
- def full_width_page(self):
- obj = self.context.get("object")
-
- pointer_zones = (
- Zone.objects.filter(
- arpa_network__net_contains_or_equals=obj.prefix
- ).order_by(Length("name").desc())[:1]
- | Zone.objects.filter(arpa_network__net_contained=obj.prefix)
- ).order_by("name")
-
- pointer_zone_table = RelatedZoneTable(
- data=pointer_zones,
- )
-
- return self.render(
- "netbox_dns/zone/related.html",
- extra_context={
- "related_pointer_zones": pointer_zone_table,
- },
- )
-
-
class RelatedDNSObjects(PluginTemplateExtension):
model = "tenancy.tenant"
@@ -96,5 +78,5 @@ def left_page(self):
template_extensions = [RelatedDNSObjects]
-if get_plugin_config("netbox_dns", "feature_ipam_integration"):
- template_extensions += [RelatedDNSRecords, RelatedDNSPointerZones]
+if get_plugin_config("netbox_dns", "feature_ipam_coupling"):
+ template_extensions.append(RelatedDNSRecords)
diff --git a/netbox_dns/templates/netbox_dns/record.html b/netbox_dns/templates/netbox_dns/record.html
index bc943c8d..e221924e 100644
--- a/netbox_dns/templates/netbox_dns/record.html
+++ b/netbox_dns/templates/netbox_dns/record.html
@@ -98,6 +98,18 @@
| Address Record |
{{ object.address_record }} |
+
+ {% if object.address_record.ipam_ip_address %}
+
+ | IPAM IP Address |
+ {{ object.address_record.ipam_ip_address }} |
+
+ {% endif %}
+ {% endif %}
+ {% if object.ipam_ip_address %}
+
+ | IPAM IP Address |
+ {{ object.ipam_ip_address }} |
{% endif %}
{% if not object.managed %}
diff --git a/netbox_dns/templates/netbox_dns/record/related.html b/netbox_dns/templates/netbox_dns/record/related.html
index dcf5ce4e..b4cdf7f0 100644
--- a/netbox_dns/templates/netbox_dns/record/related.html
+++ b/netbox_dns/templates/netbox_dns/record/related.html
@@ -2,9 +2,9 @@
{% if perms.netbox_dns.view_record %}
{% if related_address_records %}
-{% include 'inc/panel_table.html' with table=related_address_records heading='Related DNS Address Records' %}
+{% include 'inc/panel_table.html' with table=related_address_records heading='Related DNS Address Record' %}
{% endif %}
{% if related_pointer_records %}
-{% include 'inc/panel_table.html' with table=related_pointer_records heading='Related DNS Pointer Records' %}
+{% include 'inc/panel_table.html' with table=related_pointer_records heading='Related DNS Pointer Record' %}
{% endif %}
{% endif %}
diff --git a/netbox_dns/templates/netbox_dns/zone/related.html b/netbox_dns/templates/netbox_dns/zone/related.html
deleted file mode 100644
index c615a8f3..00000000
--- a/netbox_dns/templates/netbox_dns/zone/related.html
+++ /dev/null
@@ -1,7 +0,0 @@
-{% load perms %}
-
-{% if perms.netbox_dns.view_zone %}
-{% if related_pointer_zones %}
-{% include 'inc/panel_table.html' with table=related_pointer_zones heading='Related DNS Pointer Zones' %}
-{% endif %}
-{% endif %}
diff --git a/netbox_dns/tests/coupling/__init__.py b/netbox_dns/tests/coupling/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/netbox_dns/tests/coupling/test_ip_dns_coupling.py b/netbox_dns/tests/coupling/test_ip_dns_coupling.py
new file mode 100644
index 00000000..c0fc40da
--- /dev/null
+++ b/netbox_dns/tests/coupling/test_ip_dns_coupling.py
@@ -0,0 +1,422 @@
+from django.urls import reverse
+from django.test import override_settings
+from django.contrib.contenttypes.models import ContentType
+from django.core import management
+
+from rest_framework import status
+from utilities.testing import APITestCase
+
+from extras.models import CustomField
+from extras.choices import CustomFieldTypeChoices
+from ipam.models import IPAddress, Prefix
+from netaddr import IPNetwork
+from netbox_dns.models import Record, Zone, NameServer, RecordTypeChoices
+
+
+class IPAddressDNSRecordCouplingTest(APITestCase):
+ network = "10.0.0.0/24"
+ ns = "ns1.example.com"
+ zone_data = {
+ "default_ttl": 86400,
+ "soa_rname": "hostmaster.example.com",
+ "soa_refresh": 172800,
+ "soa_retry": 7200,
+ "soa_expire": 2592000,
+ "soa_ttl": 86400,
+ "soa_minimum": 3600,
+ "soa_serial": 1,
+ }
+
+ @classmethod
+ def setUpTestData(cls):
+ # Test data
+ cls.nameserver = NameServer.objects.create(name=cls.ns)
+ cls.zone = Zone.objects.create(
+ name="zone1.example.com", **cls.zone_data, soa_mname=cls.nameserver
+ )
+ cls.zone2 = Zone.objects.create(
+ name="zone2.example.com", **cls.zone_data, soa_mname=cls.nameserver
+ )
+ cls.prefix = Prefix.objects.create(prefix=cls.network)
+
+ #
+ # Add the required custom fields
+ #
+ management.call_command("setup_coupling")
+
+ @override_settings(PLUGINS_CONFIG={"netbox_dns": {"feature_ipam_coupling": True}})
+ def test_create_ip(self):
+ zone = self.zone
+ name = "test-create"
+ addr = "10.0.0.25/24"
+
+ # Grant permissions to user
+ self.add_permissions("ipam.add_ipaddress")
+ self.add_permissions("netbox_dns.add_record")
+
+ url = reverse("ipam-api:ipaddress-list")
+ data = {
+ "address": addr,
+ "custom_fields": {
+ "ipaddress_dns_zone_id": zone.id,
+ "ipaddress_dns_record_name": name,
+ },
+ }
+ response = self.client.post(url, data, format="json", **self.header)
+
+ self.assertTrue(status.is_success(response.status_code))
+
+ # Check if "record" has been created, is managed and has correct name and zone
+ ip_id = response.data["id"]
+ record = Record.objects.get(ipam_ip_address=ip_id)
+ self.assertEqual(record.name, name)
+ self.assertEqual(record.zone.id, zone.id)
+ self.assertTrue(record.managed)
+ # Check value of dns_name
+ ip_address = IPAddress.objects.get(id=ip_id)
+ self.assertEqual(ip_address.dns_name, f"{name}.{zone.name}")
+
+ @override_settings(PLUGINS_CONFIG={"netbox_dns": {"feature_ipam_coupling": True}})
+ def test_create_ip_existing_dns_record(self):
+ zone = self.zone
+ name = "test-create-ip-existing-dns-record"
+ addr = "10.0.0.25/24"
+
+ # Create DNS record
+ record = Record.objects.create(
+ name=name,
+ zone=zone,
+ type=RecordTypeChoices.A,
+ value=str(IPNetwork(addr).ip),
+ )
+
+ # Grant permissions to user
+ self.add_permissions("ipam.add_ipaddress")
+ self.add_permissions("netbox_dns.add_record")
+
+ url = reverse("ipam-api:ipaddress-list")
+ data = {
+ "address": addr,
+ "custom_fields": {
+ "ipaddress_dns_zone_id": zone.id,
+ "ipaddress_dns_record_name": name,
+ },
+ }
+ response = self.client.post(url, data, format="json", **self.header)
+
+ self.assertTrue(status.is_success(response.status_code))
+
+ # Check if "record" has been linked to and is now managed
+ ip_address = IPAddress.objects.get(id=response.data["id"])
+ record = Record.objects.get(ipam_ip_address=ip_address)
+ self.assertTrue(record.managed)
+ self.assertEqual(record.name, name)
+ self.assertEqual(record.zone, zone)
+ # Check value of dns_name
+ self.assertEqual(ip_address.dns_name, f"{name}.{zone.name}")
+
+ @override_settings(PLUGINS_CONFIG={"netbox_dns": {"feature_ipam_coupling": True}})
+ def test_create_ip_missing_dns_permission(self):
+ zone = self.zone
+ name = "test-create-ip-missing-dns-perm"
+ addr = "10.0.0.26/24"
+
+ # Grant only IPAddress add permission to user,
+ # and *no* DNS record add permission
+ self.add_permissions("ipam.add_ipaddress")
+
+ url = reverse("ipam-api:ipaddress-list")
+ data = {
+ "address": addr,
+ "custom_fields": {
+ "ipaddress_dns_zone_id": zone.id,
+ "ipaddress_dns_record_name": name,
+ },
+ }
+ response = self.client.post(url, data, format="json", **self.header)
+
+ # Should be denied
+ self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+ # No IP address should have been created
+ self.assertFalse(IPAddress.objects.filter(address=addr).exists())
+ # No DNS Record should have been created
+ self.assertFalse(Record.objects.filter(name=name, zone_id=zone.id).exists())
+
+ @override_settings(PLUGINS_CONFIG={"netbox_dns": {"feature_ipam_coupling": True}})
+ def test_delete_ip(self):
+ zone = self.zone
+ name = "test-delete-ip"
+ addr = IPNetwork("10.0.0.27/24")
+ # Grant delete on both IP address and DNS record
+ self.add_permissions("ipam.delete_ipaddress")
+ self.add_permissions("netbox_dns.delete_record")
+
+ # Create DNS record and IP Address
+ A = RecordTypeChoices.A
+ s_addr = str(addr.ip)
+ ip_address = IPAddress.objects.create(
+ address=addr,
+ dns_name=f"{name}.{zone.name}",
+ custom_field_data={
+ "ipaddress_dns_record_name": name,
+ "ipaddress_dns_zone_id": zone.id,
+ },
+ )
+ record = Record.objects.create(
+ name=name,
+ zone=zone,
+ type=A,
+ value=s_addr,
+ ipam_ip_address=ip_address,
+ )
+
+ # Delete address
+ url = reverse("ipam-api:ipaddress-list") + str(ip_address.id) + "/"
+ response = self.client.delete(url, **self.header)
+ # Check response
+ self.assertTrue(status.is_success(response.status_code))
+ # Check if DNS record has been deleted
+ self.assertFalse(Record.objects.filter(id=record.id).exists())
+ # Check if IP address has been deleted
+ self.assertFalse(IPAddress.objects.filter(id=ip_address.id).exists())
+
+ @override_settings(PLUGINS_CONFIG={"netbox_dns": {"feature_ipam_coupling": True}})
+ def test_modify_name_existing_ip(self):
+ addr = IPNetwork("10.0.0.27/24")
+ zone = self.zone
+ name = "test-modify-name-existing-ip"
+ newname = "newname"
+ zone2 = self.zone2
+
+ # Grant permissions to user
+ self.add_permissions("ipam.change_ipaddress")
+ self.add_permissions("netbox_dns.change_record")
+
+ # Create IP Address
+ ip_address = IPAddress.objects.create(
+ address=addr,
+ dns_name=f"{name}.{zone.name}",
+ custom_field_data={
+ "ipaddress_dns_record_name": name,
+ "ipaddress_dns_zone_id": zone.id,
+ },
+ )
+ # Create DNS record
+ A = RecordTypeChoices.A
+ s_addr = str(addr.ip)
+ record = Record.objects.create(
+ name=name,
+ zone=zone,
+ type=A,
+ value=s_addr,
+ ipam_ip_address=ip_address,
+ )
+
+ # Change name and zone
+ url = reverse("ipam-api:ipaddress-list") + str(ip_address.id) + "/"
+ data = {
+ "custom_fields": {
+ "ipaddress_dns_record_name": newname,
+ "ipaddress_dns_zone_id": zone2.id,
+ }
+ }
+ response = self.client.patch(url, data, format="json", **self.header)
+
+ # Check response
+ self.assertTrue(status.is_success(response.status_code))
+
+ # Check if ip still point to a record and if record matches
+ ip_address = IPAddress.objects.get(id=ip_address.id)
+ record_query = Record.objects.filter(ipam_ip_address=ip_address)
+ self.assertEqual(record_query.count(), 1)
+ self.assertEqual(record_query[0].name, newname)
+ self.assertEqual(record_query[0].zone, zone2)
+ # Check value of dns_name
+ self.assertEqual(ip_address.dns_name, f"{newname}.{zone2.name}")
+
+ @override_settings(PLUGINS_CONFIG={"netbox_dns": {"feature_ipam_coupling": True}})
+ def test_modify_name_existing_ip_missing_dns_permission(self):
+ addr = IPNetwork("10.0.0.27/24")
+ zone = self.zone
+ name = "test-modify-name-existing-ip-no-perm"
+
+ newname = "newname"
+ zone2 = self.zone2
+
+ # Grant permissions to user
+ self.add_permissions("ipam.change_ipaddress")
+
+ # Create IP Address
+ ip_address = IPAddress.objects.create(
+ address=addr,
+ dns_name=f"{name}.{zone.name}",
+ custom_field_data={
+ "ipaddress_dns_record_name": name,
+ "ipaddress_dns_zone_id": zone.id,
+ },
+ )
+ # Create DNS record
+ A = RecordTypeChoices.A
+ s_addr = str(addr.ip)
+ record = Record.objects.create(
+ name=name,
+ zone=zone,
+ type=A,
+ value=s_addr,
+ ipam_ip_address=ip_address,
+ )
+
+ # Change name and zone
+ url = reverse("ipam-api:ipaddress-list") + str(ip_address.id) + "/"
+ data = {
+ "custom_fields": {
+ "ipaddress_dns_record_name": newname,
+ "ipaddress_dns_zone_id": zone2.id,
+ }
+ }
+ response = self.client.patch(url, data, format="json", **self.header)
+
+ # Check response
+ self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+ # Check for no changes
+ ip_address = IPAddress.objects.get(id=ip_address.id)
+ record_query = Record.objects.filter(ipam_ip_address=ip_address)
+ self.assertEqual(record_query.count(), 1)
+ self.assertEqual(record_query[0].name, name)
+ self.assertEqual(record_query[0].zone, zone)
+ # Check value of dns_name
+ self.assertEqual(ip_address.dns_name, f"{name}.{zone.name}")
+
+ @override_settings(PLUGINS_CONFIG={"netbox_dns": {"feature_ipam_coupling": True}})
+ def test_clear_name_existing_ip(self):
+ addr = IPNetwork("10.0.0.28/24")
+ zone = self.zone
+ name = "test-clear-name-existing-ip"
+
+ # Grant permissions to user
+ self.add_permissions("ipam.change_ipaddress")
+ self.add_permissions("netbox_dns.delete_record")
+
+ # Create IP Address
+ ip_address = IPAddress.objects.create(
+ address=addr,
+ dns_name=f"{name}.{zone.name}",
+ custom_field_data={
+ "ipaddress_dns_record_name": name,
+ "ipaddress_dns_zone_id": zone.id,
+ },
+ )
+ # Create DNS record
+ A = RecordTypeChoices.A
+ s_addr = str(addr.ip)
+ record = Record.objects.create(
+ name=name,
+ zone=zone,
+ type=A,
+ value=s_addr,
+ ipam_ip_address=ip_address,
+ )
+
+ url = reverse("ipam-api:ipaddress-list") + str(ip_address.id) + "/"
+ data = {"custom_fields": {"ipaddress_dns_zone_id": None}}
+ response = self.client.patch(url, data, format="json", **self.header)
+
+ # Check response
+ self.assertTrue(status.is_success(response.status_code))
+ # Check if record has been deleted
+ self.assertFalse(Record.objects.filter(ipam_ip_address=ip_address).exists())
+ # Re-read IPAddress object
+ ip_address = IPAddress.objects.get(id=ip_address.id)
+ # Check if dns_name is empty
+ self.assertEqual(ip_address.dns_name, "")
+ cf_name = ip_address.custom_field_data.get("ipaddress_dns_record_name")
+ self.assertEqual(cf_name, "")
+
+ @override_settings(PLUGINS_CONFIG={"netbox_dns": {"feature_ipam_coupling": True}})
+ def test_rename_zone_existing_ip(self):
+ addr = IPNetwork("10.0.0.29/24")
+ zone = self.zone
+ name = "test-rename-zone-existing-ip"
+ new_zone_name = "newzone.example.com"
+
+ # Grant permissions to user
+ self.add_permissions("ipam.change_ipaddress")
+ self.add_permissions("netbox_dns.change_zone")
+
+ # Create IP Address
+ ip_address = IPAddress.objects.create(
+ address=addr,
+ dns_name=f"{name}.{zone.name}",
+ custom_field_data={
+ "ipaddress_dns_record_name": name,
+ "ipaddress_dns_zone_id": zone.id,
+ },
+ )
+ # Create DNS record
+ A = RecordTypeChoices.A
+ s_addr = str(addr.ip)
+ record = Record.objects.create(
+ name=name,
+ zone=zone,
+ type=A,
+ value=s_addr,
+ ipam_ip_address=ip_address,
+ )
+
+ url = reverse("plugins-api:netbox_dns-api:zone-list") + str(zone.id) + "/"
+ data = {"name": new_zone_name}
+ response = self.client.patch(url, data, format="json", **self.header)
+
+ # Check response
+ self.assertTrue(status.is_success(response.status_code))
+ # Re-read IPAddress object
+ ip_address = IPAddress.objects.get(id=ip_address.id)
+ # Check if dns_name has correct value
+ self.assertEqual(ip_address.dns_name, f"{name}.{new_zone_name}")
+ # Check if record zone has correct name
+ record_query = Record.objects.filter(ipam_ip_address=ip_address)
+ self.assertEqual(record_query.count(), 1)
+ self.assertEqual(record_query[0].zone.name, new_zone_name)
+
+ @override_settings(PLUGINS_CONFIG={"netbox_dns": {"feature_ipam_coupling": True}})
+ def test_delete_zone_existing_ip(self):
+ addr = IPNetwork("10.0.0.30/24")
+ zone = self.zone
+ name = "test-delete-zone-existing-ip"
+
+ # Create IP Address
+ ip_address = IPAddress.objects.create(
+ address=addr,
+ dns_name=f"{name}.{zone.name}",
+ custom_field_data={
+ "ipaddress_dns_record_name": name,
+ "ipaddress_dns_zone_id": zone.id,
+ },
+ )
+ # Create DNS record
+ A = RecordTypeChoices.A
+ s_addr = str(addr.ip)
+ record = Record.objects.create(
+ name=name, zone=zone, type=A, value=s_addr, ipam_ip_address=ip_address
+ )
+
+ # Grant permissions to user
+ self.add_permissions("ipam.change_ipaddress")
+ self.add_permissions("netbox_dns.delete_zone")
+ self.add_permissions("netbox_dns.delete_record")
+
+ url = reverse("plugins-api:netbox_dns-api:zone-list") + str(zone.id) + "/"
+ response = self.client.delete(url, **self.header)
+
+ # Check response
+ self.assertTrue(status.is_success(response.status_code))
+ # Check if record has been deleted
+ self.assertFalse(Record.objects.filter(id=record.id).exists())
+ # Re-read IPAddress object
+ ip_address = IPAddress.objects.get(id=ip_address.id)
+ # Check if dns_name is empty
+ self.assertEqual(ip_address.dns_name, "")
+ # Check if custom field "ipaddress_dns_record_name" is empty
+ cf_name = ip_address.custom_field_data.get("ipaddress_dns_record_name")
+ self.assertEqual(cf_name, "")