diff --git a/netbox/ipam/models.py b/netbox/ipam/models.py index fa42d68f1..fc404dbda 100644 --- a/netbox/ipam/models.py +++ b/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): diff --git a/netbox/ipam/tests/__init__.py b/netbox/ipam/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/netbox/ipam/tests/test_models.py b/netbox/ipam/tests/test_models.py new file mode 100644 index 000000000..49ca3b09b --- /dev/null +++ b/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)