Merge pull request #804 from digitalocean/prefix-unique

Enforce Global Unique on Prefixes

netbox/ipam/models.py

@@ -298,8 +298,13 @@ class Prefix(CreatedUpdatedModel, CustomFieldModel):
     def get_absolute_url(self):
         return reverse('ipam:prefix', args=[self.pk])
 
-    def clean(self):
+    def duplicates(self):
+        return Prefix.objects.filter(
+            vrf=self.vrf,
+            prefix=str(self.prefix)
+        ).exclude(pk=self.pk)
 
+    def clean(self):
         # Disallow host masks
         if self.prefix:
             if self.prefix.version == 4 and self.prefix.prefixlen == 32:
@@ -311,6 +316,16 @@ class Prefix(CreatedUpdatedModel, CustomFieldModel):
                     'prefix': "Cannot create host addresses (/128) as prefixes. Create an IPv6 address instead."
                 })
 
+        if ((not self.vrf and settings.ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique)):
+            dupes = self.duplicates()
+            if dupes:
+                raise ValidationError({
+                    'prefix': "Duplicate prefix found in {}: {}".format(
+                        "VRF {}".format(self.vrf) if self.vrf else "global table",
+                        dupes.first(),
+                    )
+                })
+
     def save(self, *args, **kwargs):
         if self.prefix:
             # Clear host bits from prefix
@@ -400,22 +415,23 @@ class IPAddress(CreatedUpdatedModel, CustomFieldModel):
     def get_absolute_url(self):
         return reverse('ipam:ipaddress', args=[self.pk])
 
+    def duplicates(self):
+        return IPAddress.objects.filter(
+            vrf=self.vrf,
+            address__net_host=str(self.address.ip)
+        ).exclude(pk=self.pk)
+
     def clean(self):
 
         # Enforce unique IP space if applicable
-        if self.vrf and self.vrf.enforce_unique:
-            duplicate_ips = IPAddress.objects.filter(vrf=self.vrf, address__net_host=str(self.address.ip))\
-                .exclude(pk=self.pk)
-            if duplicate_ips:
+        if ((not self.vrf and settings.ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique)):
+            dupes = self.duplicates()
+            if dupes:
                 raise ValidationError({
-                    'address': "Duplicate IP address found in VRF {}: {}".format(self.vrf, duplicate_ips.first())
-                })
-        elif not self.vrf and settings.ENFORCE_GLOBAL_UNIQUE:
-            duplicate_ips = IPAddress.objects.filter(vrf=None, address__net_host=str(self.address.ip))\
-                .exclude(pk=self.pk)
-            if duplicate_ips:
-                raise ValidationError({
-                    'address': "Duplicate IP address found in global table: {}".format(duplicate_ips.first())
+                    'address': "Duplicate IP Address found in {}: {}".format(
+                        "VRF {}".format(self.vrf) if self.vrf else "global table",
+                        dupes.first(),
+                    )
                 })
 
     def save(self, *args, **kwargs):

netbox/ipam/tests/test_models.py

@@ -0,0 +1,72 @@
+import netaddr
+
+from django.test import TestCase, override_settings
+
+from ipam.models import IPAddress, Prefix, VRF
+from django.core.exceptions import ValidationError
+
+
+class TestPrefix(TestCase):
+
+    fixtures = [
+        'dcim',
+        'ipam'
+    ]
+
+    def test_create(self):
+        prefix = Prefix.objects.create(
+            prefix=netaddr.IPNetwork('10.1.1.0/24'),
+            status=1
+        )
+        self.assertIsNone(prefix.clean())
+
+    @override_settings(ENFORCE_GLOBAL_UNIQUE=True)
+    def test_duplicate_global(self):
+        prefix = Prefix.objects.create(
+            prefix=netaddr.IPNetwork('10.1.1.0/24'),
+            status=1
+        )
+        self.assertRaises(ValidationError, prefix.clean)
+
+    @override_settings(ENFORCE_GLOBAL_UNIQUE=True)
+    def test_duplicate_vrf(self):
+        pfx_kwargs = {
+            "prefix": netaddr.IPNetwork('10.1.1.0/24'),
+            "status": 1,
+            "vrf": VRF.objects.create(name='Test', rd='1:1'),
+        }
+        Prefix.objects.create(**pfx_kwargs)
+        dup_prefix = Prefix.objects.create(**pfx_kwargs)
+        self.assertRaises(ValidationError, dup_prefix.clean)
+
+
+class TestIPAddress(TestCase):
+
+    fixtures = [
+        'dcim',
+        'ipam'
+    ]
+
+    def test_create(self):
+        address = IPAddress.objects.create(
+            address=netaddr.IPNetwork('10.0.254.1/24'),
+        )
+        self.assertIsNone(address.clean())
+
+    @override_settings(ENFORCE_GLOBAL_UNIQUE=True)
+    def test_duplicate_global(self):
+        address = IPAddress.objects.create(
+            address=netaddr.IPNetwork('10.0.254.1/24'),
+        )
+        self.assertRaises(ValidationError, address.clean)
+
+    @override_settings(ENFORCE_GLOBAL_UNIQUE=True)
+    def test_duplicate_vrf(self):
+        pfx_kwargs = {
+            "address": netaddr.IPNetwork('10.0.254.1/24'),
+            "status": 1,
+            "vrf": VRF.objects.create(name='Test', rd='1:1'),
+        }
+        IPAddress.objects.create(**pfx_kwargs)
+        dup_address = IPAddress.objects.create(**pfx_kwargs)
+        self.assertRaises(ValidationError, dup_address.clean)