|
|
@@ -9,16 +9,18 @@ from django.core.exceptions import ValidationError
|
|
|
from django.core.files.storage import default_storage
|
|
|
from django.core.validators import MaxValueValidator, MinValueValidator
|
|
|
from django.db import models, router, transaction
|
|
|
-from django.db.models import F, ProtectedError, prefetch_related_objects
|
|
|
+from django.db.models import Count, F, ProtectedError, prefetch_related_objects
|
|
|
from django.db.models.functions import Lower
|
|
|
from django.db.models.signals import post_save
|
|
|
from django.urls import reverse
|
|
|
from django.utils.safestring import mark_safe
|
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
+from netaddr import EUI, AddrFormatError
|
|
|
|
|
|
from dcim.choices import *
|
|
|
from dcim.constants import *
|
|
|
-from dcim.fields import MACAddressField
|
|
|
+from dcim.fields import MACAddressField, mac_unix_expanded_uppercase
|
|
|
+from dcim.querysets import MACPrefixQuerySet
|
|
|
from dcim.utils import create_port_mappings, update_interface_bridges, update_interface_parents
|
|
|
from extras.models import ConfigContextModel, CustomField
|
|
|
from extras.querysets import ConfigContextModelQuerySet
|
|
|
@@ -42,6 +44,7 @@ __all__ = (
|
|
|
'DeviceRole',
|
|
|
'DeviceType',
|
|
|
'MACAddress',
|
|
|
+ 'MACPrefix',
|
|
|
'Manufacturer',
|
|
|
'Platform',
|
|
|
'VirtualChassis',
|
|
|
@@ -1559,3 +1562,219 @@ class MACAddress(PrimaryModel):
|
|
|
raise ValidationError(
|
|
|
_("Cannot reassign MAC Address while it is designated as the primary MAC for an object")
|
|
|
)
|
|
|
+
|
|
|
+ # Enforce uniqueness within any MAC prefix which requires it
|
|
|
+ if self.mac_address and (
|
|
|
+ mac_prefix := self.get_mac_prefixes().filter(enforce_unique=True).order_by('-prefix_length').first()
|
|
|
+ ):
|
|
|
+ duplicates = MACAddress.objects.filter(mac_address=self.mac_address).exclude(pk=self.pk)
|
|
|
+ if duplicates.exists():
|
|
|
+ raise ValidationError({
|
|
|
+ 'mac_address': _(
|
|
|
+ "Duplicate MAC address found within MAC prefix {mac_prefix}, which enforces uniqueness"
|
|
|
+ ).format(mac_prefix=mac_prefix)
|
|
|
+ })
|
|
|
+
|
|
|
+ def get_mac_prefixes(self):
|
|
|
+ """
|
|
|
+ Return all MACPrefixes which contain this MAC address.
|
|
|
+ """
|
|
|
+ return MACPrefix.objects.filter(prefix__lte=self.mac_address, _last_address__gte=self.mac_address)
|
|
|
+
|
|
|
+ @cached_property
|
|
|
+ def mac_prefix(self):
|
|
|
+ """
|
|
|
+ Return the most specific MACPrefix which contains this MAC address (if any).
|
|
|
+ """
|
|
|
+ if not self.mac_address:
|
|
|
+ return None
|
|
|
+ return self.get_mac_prefixes().order_by('-prefix_length').first()
|
|
|
+
|
|
|
+
|
|
|
+class MACPrefix(PrimaryModel):
|
|
|
+ """
|
|
|
+ A block of MAC addresses which share a common prefix, such as an organizationally unique identifier (OUI).
|
|
|
+ """
|
|
|
+ prefix = MACAddressField(
|
|
|
+ verbose_name=_('prefix'),
|
|
|
+ help_text=_('The first MAC address in the block')
|
|
|
+ )
|
|
|
+ prefix_length = models.PositiveSmallIntegerField(
|
|
|
+ verbose_name=_('prefix length'),
|
|
|
+ default=MACPREFIX_LENGTH_DEFAULT,
|
|
|
+ validators=(
|
|
|
+ MinValueValidator(MACPREFIX_LENGTH_MIN),
|
|
|
+ MaxValueValidator(MACPREFIX_LENGTH_MAX),
|
|
|
+ ),
|
|
|
+ help_text=_('Number of leading bits shared by all MAC addresses in the block (e.g. 24 for an OUI)')
|
|
|
+ )
|
|
|
+ name = models.CharField(
|
|
|
+ verbose_name=_('name'),
|
|
|
+ max_length=100,
|
|
|
+ blank=True,
|
|
|
+ help_text=_('For example, the organization to which the block is assigned')
|
|
|
+ )
|
|
|
+ enforce_unique = models.BooleanField(
|
|
|
+ verbose_name=_('enforce unique'),
|
|
|
+ default=False,
|
|
|
+ help_text=_('Prevent duplicate MAC addresses within this prefix')
|
|
|
+ )
|
|
|
+
|
|
|
+ # Cached last MAC address in the block, for finding the MAC addresses and prefixes it contains
|
|
|
+ _last_address = MACAddressField(
|
|
|
+ blank=True,
|
|
|
+ editable=False
|
|
|
+ )
|
|
|
+
|
|
|
+ objects = MACPrefixQuerySet.as_manager()
|
|
|
+
|
|
|
+ clone_fields = ('prefix_length', 'enforce_unique', 'description')
|
|
|
+
|
|
|
+ class Meta:
|
|
|
+ ordering = ('prefix', 'prefix_length', 'pk')
|
|
|
+ indexes = (
|
|
|
+ models.Index(fields=('prefix', 'prefix_length', 'id')), # Default ordering
|
|
|
+ models.Index(fields=('prefix', '_last_address')),
|
|
|
+ )
|
|
|
+ constraints = (
|
|
|
+ models.UniqueConstraint(
|
|
|
+ fields=('prefix', 'prefix_length'),
|
|
|
+ name='%(app_label)s_%(class)s_unique_prefix_prefix_length',
|
|
|
+ violation_error_message=_("A MAC prefix with this prefix and length already exists.")
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ verbose_name = _('MAC prefix')
|
|
|
+ verbose_name_plural = _('MAC prefixes')
|
|
|
+
|
|
|
+ def __str__(self):
|
|
|
+ return f'{self.prefix}/{self.prefix_length}'
|
|
|
+
|
|
|
+ def _get_bounds(self):
|
|
|
+ """
|
|
|
+ Return the first and last MAC addresses in the block as integers, or None if the prefix or its length is
|
|
|
+ invalid.
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ prefix = int(EUI(self.prefix, version=48))
|
|
|
+ prefix_length = int(self.prefix_length)
|
|
|
+ except (AddrFormatError, TypeError, ValueError):
|
|
|
+ return None
|
|
|
+ if not MACPREFIX_LENGTH_MIN <= prefix_length <= MACPREFIX_LENGTH_MAX:
|
|
|
+ return None
|
|
|
+ host_mask = (1 << (48 - prefix_length)) - 1
|
|
|
+ return prefix & ~host_mask, prefix | host_mask
|
|
|
+
|
|
|
+ @property
|
|
|
+ def first_address(self):
|
|
|
+ """
|
|
|
+ Return the first MAC address in the block (the prefix with any host bits cleared).
|
|
|
+ """
|
|
|
+ if bounds := self._get_bounds():
|
|
|
+ return EUI(bounds[0], version=48, dialect=mac_unix_expanded_uppercase)
|
|
|
+ return None
|
|
|
+
|
|
|
+ @property
|
|
|
+ def last_address(self):
|
|
|
+ """
|
|
|
+ Return the last MAC address in the block.
|
|
|
+ """
|
|
|
+ if bounds := self._get_bounds():
|
|
|
+ return EUI(bounds[1], version=48, dialect=mac_unix_expanded_uppercase)
|
|
|
+ return None
|
|
|
+
|
|
|
+ @property
|
|
|
+ def size(self):
|
|
|
+ """
|
|
|
+ Return the number of MAC addresses in the block.
|
|
|
+ """
|
|
|
+ return 2 ** (48 - self.prefix_length)
|
|
|
+
|
|
|
+ def clean(self):
|
|
|
+ super().clean()
|
|
|
+
|
|
|
+ # Uniqueness cannot be enforced if the prefix already contains duplicate MAC addresses
|
|
|
+ if self.enforce_unique and self.first_address is not None:
|
|
|
+ duplicate = self.get_mac_addresses().values('mac_address').annotate(
|
|
|
+ count=Count('pk')
|
|
|
+ ).filter(count__gt=1).order_by('mac_address').first()
|
|
|
+ if duplicate:
|
|
|
+ raise ValidationError({
|
|
|
+ 'enforce_unique': _(
|
|
|
+ "Cannot enforce uniqueness: duplicate MAC address {mac_address} exists within this prefix"
|
|
|
+ ).format(mac_address=duplicate['mac_address'])
|
|
|
+ })
|
|
|
+
|
|
|
+ def save(self, *args, **kwargs):
|
|
|
+ # Clear any host bits from the prefix, and cache the last address in the block
|
|
|
+ self.prefix = self.first_address
|
|
|
+ self._last_address = self.last_address
|
|
|
+
|
|
|
+ super().save(*args, **kwargs)
|
|
|
+
|
|
|
+ def get_mac_addresses(self):
|
|
|
+ """
|
|
|
+ Return all MACAddresses within this prefix.
|
|
|
+ """
|
|
|
+ return MACAddress.objects.filter(mac_address__gte=self.first_address, mac_address__lte=self.last_address)
|
|
|
+
|
|
|
+ def get_child_prefixes(self):
|
|
|
+ """
|
|
|
+ Return all MACPrefixes within this prefix.
|
|
|
+ """
|
|
|
+ return MACPrefix.objects.filter(
|
|
|
+ prefix__gte=self.first_address,
|
|
|
+ _last_address__lte=self.last_address,
|
|
|
+ prefix_length__gt=self.prefix_length,
|
|
|
+ )
|
|
|
+
|
|
|
+ @cached_property
|
|
|
+ def mac_address_count(self):
|
|
|
+ """
|
|
|
+ Return the number of distinct MAC addresses within the prefix. (May be annotated by
|
|
|
+ MACPrefixQuerySet.annotate_mac_address_count().)
|
|
|
+ """
|
|
|
+ return self.get_mac_addresses().order_by().values('mac_address').distinct().count()
|
|
|
+
|
|
|
+ @property
|
|
|
+ def utilization(self):
|
|
|
+ """
|
|
|
+ Return the percentage of MAC addresses within the prefix which exist in NetBox.
|
|
|
+ """
|
|
|
+ return self.mac_address_count / self.size * 100
|
|
|
+
|
|
|
+ def _iter_occupied(self, batch_size=5000):
|
|
|
+ """
|
|
|
+ Yield the distinct MAC addresses within the prefix which exist in NetBox as integers, in ascending order.
|
|
|
+ Addresses are fetched in batches, so that stopping early stops reading from the database.
|
|
|
+ """
|
|
|
+ resume = self.first_address
|
|
|
+ while True:
|
|
|
+ mac_addresses = list(
|
|
|
+ MACAddress.objects.filter(mac_address__gte=resume, mac_address__lte=self.last_address)
|
|
|
+ .order_by('mac_address')
|
|
|
+ .values_list('mac_address', flat=True)
|
|
|
+ .distinct()[:batch_size]
|
|
|
+ )
|
|
|
+ for mac_address in mac_addresses:
|
|
|
+ yield int(mac_address)
|
|
|
+ if len(mac_addresses) < batch_size or mac_addresses[-1] == self.last_address:
|
|
|
+ return
|
|
|
+ resume = EUI(int(mac_addresses[-1]) + 1, version=48)
|
|
|
+
|
|
|
+ def iter_available_mac_addresses(self):
|
|
|
+ """
|
|
|
+ Yield the MAC addresses within the prefix which do not exist in NetBox, in ascending order.
|
|
|
+ """
|
|
|
+ candidate = int(self.first_address)
|
|
|
+ for occupied in self._iter_occupied():
|
|
|
+ for value in range(candidate, occupied):
|
|
|
+ yield EUI(value, version=48, dialect=mac_unix_expanded_uppercase)
|
|
|
+ candidate = occupied + 1
|
|
|
+ for value in range(candidate, int(self.last_address) + 1):
|
|
|
+ yield EUI(value, version=48, dialect=mac_unix_expanded_uppercase)
|
|
|
+
|
|
|
+ def get_first_available_mac_address(self):
|
|
|
+ """
|
|
|
+ Return the first MAC address within the prefix which does not exist in NetBox (or None).
|
|
|
+ """
|
|
|
+ return next(self.iter_available_mac_addresses(), None)
|