From cfaf8b9157db535a5289a656188628842c781ddb Mon Sep 17 00:00:00 2001 From: Zach Moody Date: Mon, 16 Jan 2017 16:28:04 -0600 Subject: [PATCH 1/4] added duplicates() method to IPAddress and Prefix model managers. refactored condition on IPAddress and Prefix clean method to use new manager method. --- netbox/ipam/models.py | 48 +++++++++++++++++++++++++++++++------------ 1 file changed, 35 insertions(+), 13 deletions(-) diff --git a/netbox/ipam/models.py b/netbox/ipam/models.py index fa42d68f1..62faacb18 100644 --- a/netbox/ipam/models.py +++ b/netbox/ipam/models.py @@ -262,6 +262,12 @@ class PrefixQuerySet(NullsFirstQuerySet): return queryset return filter(lambda p: p.depth <= limit, queryset) + def duplicates(self, prefix): + return self.filter( + vrf=prefix.vrf, + prefix=str(prefix) + ).exclude(pk=prefix.pk) + class Prefix(CreatedUpdatedModel, CustomFieldModel): """ @@ -299,7 +305,6 @@ class Prefix(CreatedUpdatedModel, CustomFieldModel): return reverse('ipam:prefix', args=[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 = Prefix.objects.duplicates(self) + 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 @@ -361,6 +376,12 @@ class IPAddressManager(models.Manager): qs = super(IPAddressManager, self).get_queryset() return qs.annotate(host=RawSQL('INET(HOST(ipam_ipaddress.address))', [])).order_by('family', 'host') + def duplicates(self, ip_obj): + return self.filter( + vrf=ip_obj.vrf, + address__net_host=str(ip_obj.address.ip) + ).exclude(pk=ip_obj.pk) + class IPAddress(CreatedUpdatedModel, CustomFieldModel): """ @@ -401,21 +422,22 @@ class IPAddress(CreatedUpdatedModel, CustomFieldModel): return reverse('ipam:ipaddress', args=[self.pk]) def clean(self): + if ((not self.vrf and settings.ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique)): + dupes = IPAddress.objects.duplicates(self) + if dupes: + raise ValidationError({ + 'address': "Duplicate IP address found in global table: {}".format(dupes.first()) + }) # 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 = IPAddress.objects.duplicates(self) + 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): From eedec192ba0a21c09f1108b61cd590a0bb679820 Mon Sep 17 00:00:00 2001 From: Zach Moody Date: Mon, 16 Jan 2017 16:28:25 -0600 Subject: [PATCH 2/4] Added model tests for duplicate prefix and IPs. --- netbox/ipam/tests/__init__.py | 0 netbox/ipam/tests/test_models.py | 72 ++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+) create mode 100644 netbox/ipam/tests/__init__.py create mode 100644 netbox/ipam/tests/test_models.py 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) From 485a21f13ee73c25ccd99a029a4f3616eee35bc2 Mon Sep 17 00:00:00 2001 From: Zach Moody Date: Mon, 16 Jan 2017 16:52:03 -0600 Subject: [PATCH 3/4] cleaned up IPAddress clean() to be more like Prefix's --- netbox/ipam/models.py | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/netbox/ipam/models.py b/netbox/ipam/models.py index 62faacb18..8e45277ab 100644 --- a/netbox/ipam/models.py +++ b/netbox/ipam/models.py @@ -422,19 +422,13 @@ class IPAddress(CreatedUpdatedModel, CustomFieldModel): return reverse('ipam:ipaddress', args=[self.pk]) def clean(self): - if ((not self.vrf and settings.ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique)): - dupes = IPAddress.objects.duplicates(self) - if dupes: - raise ValidationError({ - 'address': "Duplicate IP address found in global table: {}".format(dupes.first()) - }) # Enforce unique IP space if applicable if ((not self.vrf and settings.ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique)): dupes = IPAddress.objects.duplicates(self) if dupes: raise ValidationError({ - 'address': "Duplicate IP address found in {}: {}".format( + 'address': "Duplicate IP Address found in {}: {}".format( "VRF {}".format(self.vrf) if self.vrf else "global table", dupes.first(), ) From edf29e7b9b38821650dbba6a762b188a44948558 Mon Sep 17 00:00:00 2001 From: Zach Moody Date: Mon, 16 Jan 2017 18:14:34 -0600 Subject: [PATCH 4/4] moved duplicates() method to model instead of manager. --- netbox/ipam/models.py | 28 ++++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/netbox/ipam/models.py b/netbox/ipam/models.py index 8e45277ab..fc404dbda 100644 --- a/netbox/ipam/models.py +++ b/netbox/ipam/models.py @@ -262,12 +262,6 @@ class PrefixQuerySet(NullsFirstQuerySet): return queryset return filter(lambda p: p.depth <= limit, queryset) - def duplicates(self, prefix): - return self.filter( - vrf=prefix.vrf, - prefix=str(prefix) - ).exclude(pk=prefix.pk) - class Prefix(CreatedUpdatedModel, CustomFieldModel): """ @@ -304,6 +298,12 @@ class Prefix(CreatedUpdatedModel, CustomFieldModel): def get_absolute_url(self): return reverse('ipam:prefix', args=[self.pk]) + 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: @@ -317,7 +317,7 @@ class Prefix(CreatedUpdatedModel, CustomFieldModel): }) if ((not self.vrf and settings.ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique)): - dupes = Prefix.objects.duplicates(self) + dupes = self.duplicates() if dupes: raise ValidationError({ 'prefix': "Duplicate prefix found in {}: {}".format( @@ -376,12 +376,6 @@ class IPAddressManager(models.Manager): qs = super(IPAddressManager, self).get_queryset() return qs.annotate(host=RawSQL('INET(HOST(ipam_ipaddress.address))', [])).order_by('family', 'host') - def duplicates(self, ip_obj): - return self.filter( - vrf=ip_obj.vrf, - address__net_host=str(ip_obj.address.ip) - ).exclude(pk=ip_obj.pk) - class IPAddress(CreatedUpdatedModel, CustomFieldModel): """ @@ -421,11 +415,17 @@ 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 ((not self.vrf and settings.ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique)): - dupes = IPAddress.objects.duplicates(self) + dupes = self.duplicates() if dupes: raise ValidationError({ 'address': "Duplicate IP Address found in {}: {}".format(