import netaddr from django.contrib.contenttypes.fields import GenericForeignKey from django.contrib.contenttypes.models import ContentType from django.contrib.postgres.indexes import GistIndex from django.core.exceptions import ValidationError from django.db import models from django.db.models import F from django.db.models.functions import Cast from django.utils.functional import cached_property from django.utils.translation import gettext_lazy as _ from dcim.models.mixins import CachedScopeMixin from ipam.choices import * from ipam.constants import * from ipam.fields import IPAddressField, IPNetworkField from ipam.lookups import Host from ipam.managers import IPAddressManager from ipam.querysets import IPRangeQuerySet, PrefixQuerySet from ipam.validators import DNSValidator from netbox.config import get_config from netbox.models import OrganizationalModel, PrimaryModel from netbox.models.features import ContactsMixin __all__ = ( 'RIR', 'Aggregate', 'IPAddress', 'IPRange', 'Prefix', 'Role', ) class GetAvailablePrefixesMixin: def get_available_prefixes(self): """ Return all available prefixes within this Aggregate or Prefix as an IPSet. """ params = { 'prefix__net_contained': str(self.prefix) } if hasattr(self, 'vrf'): params['vrf'] = self.vrf child_prefixes = Prefix.objects.filter(**params).values_list('prefix', flat=True) return netaddr.IPSet(self.prefix) - netaddr.IPSet(child_prefixes) def get_first_available_prefix(self): """ Return the first available child prefix within the prefix (or None). """ available_prefixes = self.get_available_prefixes() if not available_prefixes: return None return available_prefixes.iter_cidrs()[0] class RIR(OrganizationalModel): """ A Regional Internet Registry (RIR) is responsible for the allocation of a large portion of the global IP address space. This can be an organization like ARIN or RIPE, or a governing standard such as RFC 1918. """ is_private = models.BooleanField( default=False, verbose_name=_('private'), help_text=_('IP space managed by this RIR is considered private') ) class Meta: ordering = ('name',) verbose_name = _('RIR') verbose_name_plural = _('RIRs') class Aggregate(ContactsMixin, GetAvailablePrefixesMixin, PrimaryModel): """ An aggregate exists at the root level of the IP address space hierarchy in NetBox. Aggregates are used to organize the hierarchy and track the overall utilization of available address space. Each Aggregate is assigned to a RIR. """ prefix = IPNetworkField( help_text=_("IPv4 or IPv6 network") ) rir = models.ForeignKey( to='ipam.RIR', on_delete=models.PROTECT, related_name='aggregates', verbose_name=_('RIR'), help_text=_("Regional Internet Registry responsible for this IP space") ) tenant = models.ForeignKey( to='tenancy.Tenant', on_delete=models.PROTECT, related_name='aggregates', blank=True, null=True ) date_added = models.DateField( verbose_name=_('date added'), blank=True, null=True ) clone_fields = ( 'rir', 'tenant', 'date_added', 'description', ) prerequisite_models = ( 'ipam.RIR', ) class Meta: ordering = ('prefix', 'pk') # prefix may be non-unique indexes = ( models.Index(fields=('prefix', 'id')), # Default ordering ) verbose_name = _('aggregate') verbose_name_plural = _('aggregates') def __str__(self): return str(self.prefix) def clean(self): super().clean() if self.prefix: # /0 masks are not acceptable if self.prefix.prefixlen == 0: raise ValidationError({ 'prefix': _("Cannot create aggregate with /0 mask.") }) # Ensure that the aggregate being added is not covered by an existing aggregate covering_aggregates = Aggregate.objects.filter( prefix__net_contains_or_equals=str(self.prefix) ) if self.pk: covering_aggregates = covering_aggregates.exclude(pk=self.pk) if covering_aggregates: raise ValidationError({ 'prefix': _( "Aggregates cannot overlap. {prefix} is already covered by an existing aggregate ({aggregate})." ).format( prefix=self.prefix, aggregate=covering_aggregates[0] ) }) # Ensure that the aggregate being added does not cover an existing aggregate covered_aggregates = Aggregate.objects.filter(prefix__net_contained=str(self.prefix)) if self.pk: covered_aggregates = covered_aggregates.exclude(pk=self.pk) if covered_aggregates: raise ValidationError({ 'prefix': _( "Prefixes cannot overlap aggregates. {prefix} covers an existing aggregate ({aggregate})." ).format( prefix=self.prefix, aggregate=covered_aggregates[0] ) }) @property def family(self): if not self.prefix: return None if isinstance(self.prefix, str): return netaddr.IPNetwork(self.prefix).version return self.prefix.version @property def ipv6_full(self): if self.prefix and self.prefix.version == 6: return netaddr.IPAddress(self.prefix).format(netaddr.ipv6_full) return None def get_child_prefixes(self): """ Return all Prefixes within this Aggregate """ return Prefix.objects.filter(prefix__net_contained=str(self.prefix)) def get_utilization(self): """ Determine the prefix utilization of the aggregate and return it as a percentage. """ queryset = Prefix.objects.filter(prefix__net_contained_or_equal=str(self.prefix)) child_prefixes = netaddr.IPSet([p.prefix for p in queryset]) utilization = float(child_prefixes.size) / self.prefix.size * 100 return min(utilization, 100) class Role(OrganizationalModel): """ A Role represents the functional role of a Prefix or VLAN; for example, "Customer," "Infrastructure," or "Management." """ weight = models.PositiveSmallIntegerField( verbose_name=_('weight'), default=1000 ) class Meta: ordering = ('weight', 'name') indexes = ( models.Index(fields=('weight', 'name')), # Default ordering ) verbose_name = _('role') verbose_name_plural = _('roles') def __str__(self): return self.name class Prefix(ContactsMixin, GetAvailablePrefixesMixin, CachedScopeMixin, PrimaryModel): """ A Prefix represents an IPv4 or IPv6 network, including mask length. Prefixes can optionally be scoped to certain areas and/or assigned to VRFs. A Prefix must be assigned a status and may optionally be assigned a used-define Role. A Prefix can also be assigned to a VLAN where appropriate. """ prefix = IPNetworkField( verbose_name=_('prefix'), help_text=_('IPv4 or IPv6 network with mask') ) vrf = models.ForeignKey( to='ipam.VRF', on_delete=models.PROTECT, related_name='prefixes', blank=True, null=True, verbose_name=_('VRF') ) tenant = models.ForeignKey( to='tenancy.Tenant', on_delete=models.PROTECT, related_name='prefixes', blank=True, null=True ) vlan = models.ForeignKey( to='ipam.VLAN', on_delete=models.PROTECT, related_name='prefixes', blank=True, null=True ) status = models.CharField( max_length=50, choices=PrefixStatusChoices, default=PrefixStatusChoices.STATUS_ACTIVE, verbose_name=_('status'), help_text=_('Operational status of this prefix') ) role = models.ForeignKey( to='ipam.Role', on_delete=models.SET_NULL, related_name='prefixes', blank=True, null=True, help_text=_('The primary function of this prefix') ) is_pool = models.BooleanField( verbose_name=_('is a pool'), default=False, help_text=_('All IP addresses within this prefix are considered usable') ) mark_utilized = models.BooleanField( verbose_name=_('mark utilized'), default=False, help_text=_("Treat as fully utilized") ) # Cached depth & child counts _depth = models.PositiveSmallIntegerField( default=0, editable=False ) _children = models.PositiveBigIntegerField( default=0, editable=False ) objects = PrefixQuerySet.as_manager() clone_fields = ( 'scope_type', 'scope_id', 'vrf', 'tenant', 'vlan', 'status', 'role', 'is_pool', 'mark_utilized', 'description', ) class Meta: ordering = (F('vrf').asc(nulls_first=True), 'prefix', 'pk') # (vrf, prefix) may be non-unique verbose_name = _('prefix') verbose_name_plural = _('prefixes') indexes = ( models.Index(fields=('scope_type', 'scope_id')), GistIndex(fields=['prefix'], name='ipam_prefix_gist_idx', opclasses=['inet_ops']), ) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # Cache the original prefix and VRF so we can check if they have changed on post_save self._prefix = self.__dict__.get('prefix') self._vrf_id = self.__dict__.get('vrf_id') def __str__(self): return str(self.prefix) def clean(self): super().clean() if self.prefix: # /0 masks are not acceptable if self.prefix.prefixlen == 0: raise ValidationError({ 'prefix': _("Cannot create prefix with /0 mask.") }) # Enforce unique IP space (if applicable) if (self.vrf is None and get_config().ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique): duplicate_prefixes = self.get_duplicates() if duplicate_prefixes: table = _("VRF {vrf}").format(vrf=self.vrf) if self.vrf else _("global table") raise ValidationError({ 'prefix': _("Duplicate prefix found in {table}: {prefix}").format( table=table, prefix=duplicate_prefixes.first(), ) }) def save(self, *args, **kwargs): if isinstance(self.prefix, netaddr.IPNetwork): # Clear host bits from prefix self.prefix = self.prefix.cidr # Cache objects associated with the terminating object (for filtering) self.cache_related_objects() super().save(*args, **kwargs) @property def family(self): if not self.prefix: return None if isinstance(self.prefix, str): return netaddr.IPNetwork(self.prefix).version return self.prefix.version @property def mask_length(self): if not self.prefix: return None if isinstance(self.prefix, str): return netaddr.IPNetwork(self.prefix).prefixlen return self.prefix.prefixlen @property def ipv6_full(self): if self.prefix and self.prefix.version == 6: return netaddr.IPAddress(self.prefix).format(netaddr.ipv6_full) return None @property def depth(self): return self._depth @property def children(self): return self._children def _set_prefix_length(self, value): """ Expose the IPNetwork object's prefixlen attribute on the parent model so that it can be manipulated directly, e.g. for bulk editing. """ if self.prefix is not None: self.prefix.prefixlen = value prefix_length = property(fset=_set_prefix_length) def get_status_color(self): return PrefixStatusChoices.colors.get(self.status) @cached_property def aggregate(self): """ Return the containing Aggregate for this Prefix, if any. """ try: return Aggregate.objects.get(prefix__net_contains_or_equals=str(self.prefix)) except Aggregate.DoesNotExist: return None def get_parents(self, include_self=False): """ Return all containing Prefixes in the hierarchy. """ lookup = 'net_contains_or_equals' if include_self else 'net_contains' return Prefix.objects.filter(**{ 'vrf_id': self.vrf_id, f'prefix__{lookup}': self.prefix }) def get_children(self, include_self=False): """ Return all covered Prefixes in the hierarchy. """ lookup = 'net_contained_or_equal' if include_self else 'net_contained' return Prefix.objects.filter(**{ 'vrf_id': self.vrf_id, f'prefix__{lookup}': self.prefix }) def get_duplicates(self): return Prefix.objects.filter(vrf=self.vrf, prefix=str(self.prefix)).exclude(pk=self.pk) def get_child_prefixes(self): """ Return all Prefixes within this Prefix and VRF. If this Prefix is a container in the global table, return child Prefixes belonging to any VRF. """ if self.vrf is None and self.status == PrefixStatusChoices.STATUS_CONTAINER: return Prefix.objects.filter(prefix__net_contained=str(self.prefix)) return Prefix.objects.filter(prefix__net_contained=str(self.prefix), vrf=self.vrf) @property def usable_ip_bounds(self): """ Return the first and last IPs considered usable for available-IP calculations. Pools and IPv4 /31-/32 / IPv6 /127-/128 are fully usable; otherwise IPv4 excludes network and broadcast, IPv6 excludes the subnet-router anycast address. """ network = netaddr.IPNetwork(self.prefix) family = network.version first = network.first last = network.last mask_length = network.prefixlen if ( self.is_pool or (family == 4 and mask_length >= 31) or (family == 6 and mask_length >= 127) ): return ( netaddr.IPAddress(first, version=family), netaddr.IPAddress(last, version=family), ) if family == 4: return ( netaddr.IPAddress(first + 1, version=family), netaddr.IPAddress(last - 1, version=family), ) return ( netaddr.IPAddress(first + 1, version=family), netaddr.IPAddress(last, version=family), ) @property def usable_size(self): """ The number of usable host addresses within the prefix (excludes reserved addresses). """ first_ip, last_ip = self.usable_ip_bounds return int(last_ip) - int(first_ip) + 1 def get_child_ranges(self, **kwargs): """ Return all IPRanges within this Prefix and VRF. """ # A host BETWEEN over the prefix span uses the ipam_iprange_*_host btree indexes. prefix = netaddr.IPNetwork(self.prefix) bounds = ( netaddr.IPAddress(prefix.first, version=prefix.version), netaddr.IPAddress(prefix.last, version=prefix.version), ) return IPRange.objects.filter( vrf=self.vrf, start_address__host_between=bounds, end_address__host_between=bounds, **kwargs ) def get_child_ips(self): """ Return all IPAddresses within this Prefix and VRF. If this Prefix is a container in the global table, return child IPAddresses belonging to any VRF. """ # A host BETWEEN over the prefix span is index-sargable without the <<= containment recheck. prefix = netaddr.IPNetwork(self.prefix) bounds = ( netaddr.IPAddress(prefix.first, version=prefix.version), netaddr.IPAddress(prefix.last, version=prefix.version), ) if self.vrf is None and self.status == PrefixStatusChoices.STATUS_CONTAINER: return IPAddress.objects.filter(address__host_between=bounds) return IPAddress.objects.filter(address__host_between=bounds, vrf=self.vrf) def get_available_ips(self): """ Return all available IPs within this prefix as an IPSet. """ return netaddr.IPSet( cidr for start, end in self._available_intervals() for cidr in netaddr.iprange_to_cidrs(start, end) ) def iter_available_ips(self): """ Yield the available IPs within this prefix as netaddr.IPAddress objects, in ascending order. Unlike get_available_ips(), consumption is lazy: stopping early stops reading from the database. """ for start, end in self._available_intervals(): yield from netaddr.iter_iprange(start, end) def get_available_ip_count(self): """ Return the number of available IPs within the prefix. """ first_ip, last_ip = self.usable_ip_bounds usable_size = int(last_ip) - int(first_ip) + 1 populated_intervals = self.get_child_ranges(mark_populated=True).get_intervals(first_ip, last_ip) populated_count = sum(int(end) - int(start) + 1 for start, end in populated_intervals) # Populated ranges already cover the usable span; skip the child-IP count entirely. if populated_count >= usable_size: return 0 child_ip_count = ( self.get_child_ips() .filter(address__host_between=(first_ip, last_ip)) .count_distinct_hosts(exclude_intervals=populated_intervals) ) return max(usable_size - populated_count - child_ip_count, 0) def get_ip_usage_summary(self): """ Return the available IP count and utilization together as a dict, sharing a single distinct-host scan. Intended for detail views rendering both values; list views should call get_utilization() alone, which is cheaper per row. """ # Marked-utilized and container utilization need no host scan; delegate. if self.mark_utilized or self.status == PrefixStatusChoices.STATUS_CONTAINER: return { 'available_ip_count': self.get_available_ip_count(), 'utilization': self.get_utilization(), } first_ip, last_ip = self.usable_ip_bounds usable_size = int(last_ip) - int(first_ip) + 1 populated_intervals = self.get_child_ranges(mark_populated=True).get_intervals(first_ip, last_ip) utilized_intervals = self.get_child_ranges(mark_utilized=True).get_intervals() counts = self.get_child_ips().count_distinct_hosts_pair( bounds=(first_ip, last_ip), bounded_exclude=populated_intervals, total_exclude=utilized_intervals, ) populated_count = sum(int(end) - int(start) + 1 for start, end in populated_intervals) utilized_range_count = sum(int(end) - int(start) + 1 for start, end in utilized_intervals) prefix_size = self._get_utilization_denominator() return { 'available_ip_count': max(usable_size - populated_count - counts['bounded'], 0), 'utilization': min(float(utilized_range_count + counts['total']) / prefix_size * 100, 100), } def get_first_available_ip(self): """ Return the first available IP within the prefix (or None). """ first_ip, last_ip = self.usable_ip_bounds populated_intervals = self.get_child_ranges(mark_populated=True).get_intervals(first_ip, last_ip) first_available_ip = self.get_child_ips().first_available_host( first_ip, last_ip, exclude_intervals=populated_intervals, ) if first_available_ip is None: return None return f'{first_available_ip}/{self.prefix.prefixlen}' def get_utilization(self): """ Determine the utilization of the prefix and return it as a percentage. For Prefixes with a status of "container", calculate utilization based on child prefixes. For all others, count child IP addresses. """ if self.mark_utilized: return 100 if self.status == PrefixStatusChoices.STATUS_CONTAINER: queryset = Prefix.objects.filter( prefix__net_contained=str(self.prefix), vrf=self.vrf ) child_prefixes = netaddr.IPSet([p.prefix for p in queryset]) utilization = float(child_prefixes.size) / self.prefix.size * 100 else: prefix_size = self._get_utilization_denominator() utilized_intervals = self.get_child_ranges(mark_utilized=True).get_intervals() utilized_range_count = sum(int(end) - int(start) + 1 for start, end in utilized_intervals) # Utilized ranges already saturate the prefix; skip the child-IP count. if utilized_range_count >= prefix_size: return 100 child_ip_count = self.get_child_ips().count_distinct_hosts( exclude_intervals=utilized_intervals, ) utilization = float(utilized_range_count + child_ip_count) / prefix_size * 100 return min(utilization, 100) def _available_intervals(self): """ Yield the available (start, end) host intervals within the prefix. """ first_ip, last_ip = self.usable_ip_bounds populated_intervals = self.get_child_ranges(mark_populated=True).get_intervals(first_ip, last_ip) return self.get_child_ips().available_intervals( first_ip, last_ip, exclude_intervals=populated_intervals, ) def _get_utilization_denominator(self): """ The address count utilization is measured against (IPv4 non-pool prefixes exclude the network and broadcast addresses; IPv6 uses the full prefix size). """ prefix_size = self.prefix.size if self.prefix.version == 4 and self.prefix.prefixlen < 31 and not self.is_pool: return prefix_size - 2 return prefix_size class IPRange(ContactsMixin, PrimaryModel): """ A range of IP addresses, defined by start and end addresses. """ start_address = IPAddressField( verbose_name=_('start address'), help_text=_('IPv4 or IPv6 address (with mask)') ) end_address = IPAddressField( verbose_name=_('end address'), help_text=_('IPv4 or IPv6 address (with mask)') ) size = models.PositiveIntegerField( verbose_name=_('size'), editable=False ) vrf = models.ForeignKey( to='ipam.VRF', on_delete=models.PROTECT, related_name='ip_ranges', blank=True, null=True, verbose_name=_('VRF') ) tenant = models.ForeignKey( to='tenancy.Tenant', on_delete=models.PROTECT, related_name='ip_ranges', blank=True, null=True ) status = models.CharField( verbose_name=_('status'), max_length=50, choices=IPRangeStatusChoices, default=IPRangeStatusChoices.STATUS_ACTIVE, help_text=_('Operational status of this range') ) role = models.ForeignKey( to='ipam.Role', on_delete=models.SET_NULL, related_name='ip_ranges', blank=True, null=True, help_text=_('The primary function of this range') ) mark_populated = models.BooleanField( verbose_name=_('mark populated'), default=False, help_text=_("Prevent the creation of IP addresses within this range") ) mark_utilized = models.BooleanField( verbose_name=_('mark utilized'), default=False, help_text=_("Report space as fully utilized") ) objects = IPRangeQuerySet.as_manager() clone_fields = ( 'vrf', 'tenant', 'status', 'role', 'description', 'mark_populated', 'mark_utilized', ) class Meta: ordering = (F('vrf').asc(nulls_first=True), 'start_address', 'pk') # (vrf, start_address) may be non-unique indexes = ( models.Index( Cast(Host('start_address'), output_field=IPAddressField()), name='ipam_iprange_start_host', ), models.Index( Cast(Host('end_address'), output_field=IPAddressField()), name='ipam_iprange_end_host', ), ) verbose_name = _('IP range') verbose_name_plural = _('IP ranges') def __str__(self): return self.name def clean(self): super().clean() if self.start_address and self.end_address: # Check that start & end IP versions match if self.start_address.version != self.end_address.version: raise ValidationError({ 'end_address': _("Starting and ending IP address versions must match") }) # Check that the start & end IP prefix lengths match if self.start_address.prefixlen != self.end_address.prefixlen: raise ValidationError({ 'end_address': _("Starting and ending IP address masks must match") }) # Equal start/end addresses are permitted for single-address ranges. # Use .ip (host portion) rather than the IPNetwork object to avoid comparing prefix lengths again. if self.end_address.ip < self.start_address.ip: raise ValidationError({ 'end_address': _( "Ending address must be greater than or equal to the starting address ({start_address})" ).format(start_address=self.start_address) }) # Check for overlapping ranges overlapping_ranges = ( IPRange.objects.exclude(pk=self.pk) .filter(vrf=self.vrf) .filter( # Starts inside Q( start_address__host__inet__gte=self.start_address.ip, start_address__host__inet__lte=self.end_address.ip, ) | # Ends inside Q( end_address__host__inet__gte=self.start_address.ip, end_address__host__inet__lte=self.end_address.ip, ) | # Starts & ends outside Q( start_address__host__inet__lte=self.start_address.ip, end_address__host__inet__gte=self.end_address.ip, ) ) ) if overlapping_ranges.exists(): raise ValidationError( _("Defined addresses overlap with range {overlapping_range} in VRF {vrf}").format( overlapping_range=overlapping_ranges.first(), vrf=self.vrf )) # Validate maximum size MAX_SIZE = 2 ** 32 - 1 if int(self.end_address.ip - self.start_address.ip) + 1 > MAX_SIZE: raise ValidationError( _("Defined range exceeds maximum supported size ({max_size})").format(max_size=MAX_SIZE) ) def save(self, *args, **kwargs): # Record the range's size (number of IP addresses) self.size = int(self.end_address.ip - self.start_address.ip) + 1 super().save(*args, **kwargs) @property def family(self): if not self.start_address: return None if isinstance(self.start_address, str): return netaddr.IPAddress(self.start_address.split('/')[0]).version return self.start_address.version @property def range(self): return netaddr.IPRange(self.start_address.ip, self.end_address.ip) @property def mask_length(self): return self.start_address.prefixlen if self.start_address else None @cached_property def name(self): """ Return an efficient string representation of the IP range. """ # Single-address ranges render both endpoints to stay distinct from IPAddress rows. if self.start_address.ip == self.end_address.ip: return f'{self.start_address.ip}-{self.end_address.ip}/{self.start_address.prefixlen}' separator = ':' if self.family == 6 else '.' start_chunks = str(self.start_address.ip).split(separator) end_chunks = str(self.end_address.ip).split(separator) base_chunks = [] for a, b in zip(start_chunks, end_chunks): if a == b: base_chunks.append(a) base_str = separator.join(base_chunks) start_str = separator.join(start_chunks[len(base_chunks):]) end_str = separator.join(end_chunks[len(base_chunks):]) return f'{base_str}{separator}{start_str}-{end_str}/{self.start_address.prefixlen}' def _set_prefix_length(self, value): """ Expose the IPRange object's prefixlen attribute on the parent model so that it can be manipulated directly, e.g. for bulk editing. """ self.start_address.prefixlen = value self.end_address.prefixlen = value prefix_length = property(fset=_set_prefix_length) def get_status_color(self): return IPRangeStatusChoices.colors.get(self.status) @cached_property def first_available_ip(self): """ Return the first available IP within the range (or None). """ return self.get_first_available_ip() @property def utilization(self): """ Determine the utilization of the range and return it as a percentage. """ if self.mark_utilized: return 100 return min(float(self._occupied_host_count) / self.size * 100, 100) def get_child_ips(self): """ Return all IPAddresses within this IPRange and VRF. """ return IPAddress.objects.filter( vrf=self.vrf, address__host_between=(self.start_address.ip, self.end_address.ip), ) def get_available_ips(self): """ Return all available IPs within this range as an IPSet. """ return netaddr.IPSet( cidr for start, end in self._available_intervals() for cidr in netaddr.iprange_to_cidrs(start, end) ) def iter_available_ips(self): """ Yield the available IPs within this range as netaddr.IPAddress objects, in ascending order. Unlike get_available_ips(), consumption is lazy: stopping early stops reading from the database. """ for start, end in self._available_intervals(): yield from netaddr.iter_iprange(start, end) def get_available_ip_count(self): """ Return the number of available IPs within the range. """ if self.mark_populated: return 0 return max(self.size - self._occupied_host_count, 0) def get_first_available_ip(self): """ Return the first available IP within the range (or None). """ if self.mark_populated: return None first_available_ip = self.get_child_ips().first_available_host( self.start_address.ip, self.end_address.ip, ) if first_available_ip is None: return None return f'{first_available_ip}/{self.start_address.prefixlen}' def _available_intervals(self): """ Yield the available (start, end) host intervals within the range. """ if self.mark_populated: return iter(()) return self.get_child_ips().available_intervals( self.start_address.ip, self.end_address.ip, ) @cached_property def _occupied_host_count(self): """ The number of distinct occupied hosts within the range, cached for the lifetime of the instance. """ return self.get_child_ips().count_distinct_hosts() class IPAddress(ContactsMixin, PrimaryModel): """ An IPAddress represents an individual IPv4 or IPv6 address and its mask. The mask length should match what is configured in the real world. (Typically, only loopback interfaces are configured with /32 or /128 masks.) Like Prefixes, IPAddresses can optionally be assigned to a VRF. An IPAddress can optionally be assigned to an Interface. Interfaces can have zero or more IPAddresses assigned to them. An IPAddress can also optionally point to a NAT inside IP, designating itself as a NAT outside IP. This is useful, for example, when mapping public addresses to private addresses. When an Interface has been assigned an IPAddress which has a NAT outside IP, that Interface's Device can use either the inside or outside IP as its primary IP. """ address = IPAddressField( verbose_name=_('address'), help_text=_('IPv4 or IPv6 address (with mask)') ) vrf = models.ForeignKey( to='ipam.VRF', on_delete=models.PROTECT, related_name='ip_addresses', blank=True, null=True, verbose_name=_('VRF') ) tenant = models.ForeignKey( to='tenancy.Tenant', on_delete=models.PROTECT, related_name='ip_addresses', blank=True, null=True ) status = models.CharField( verbose_name=_('status'), max_length=50, choices=IPAddressStatusChoices, default=IPAddressStatusChoices.STATUS_ACTIVE, help_text=_('The operational status of this IP') ) role = models.CharField( verbose_name=_('role'), max_length=50, choices=IPAddressRoleChoices, blank=True, null=True, help_text=_('The functional role of this IP') ) assigned_object_type = models.ForeignKey( to='contenttypes.ContentType', on_delete=models.PROTECT, related_name='+', blank=True, null=True ) assigned_object_id = models.PositiveBigIntegerField( blank=True, null=True ) assigned_object = GenericForeignKey( ct_field='assigned_object_type', fk_field='assigned_object_id' ) nat_inside = models.ForeignKey( to='self', on_delete=models.SET_NULL, related_name='nat_outside', blank=True, null=True, verbose_name=_('NAT (inside)'), help_text=_('The IP for which this address is the "outside" IP') ) dns_name = models.CharField( max_length=255, blank=True, validators=[DNSValidator], verbose_name=_('DNS name'), help_text=_('Hostname or FQDN (not case-sensitive)') ) objects = IPAddressManager() clone_fields = ( 'vrf', 'tenant', 'status', 'role', 'dns_name', 'description', ) class Meta: ordering = ('address', 'pk') # address may be non-unique indexes = ( models.Index(fields=('address', 'id')), # Default ordering (see IPAddressManager). The primary key must be included so that the index can # satisfy the ordering outright; without it PostgreSQL falls back to an incremental sort, which # measurably slows deep pagination. models.Index( Cast(Host('address'), output_field=IPAddressField()), F('id'), name='ipam_ipaddress_host' ), models.Index(fields=('assigned_object_type', 'assigned_object_id')), ) verbose_name = _('IP address') verbose_name_plural = _('IP addresses') def __str__(self): return str(self.address) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # Denote the original assigned object (if any) for validation in clean() self._original_assigned_object_id = self.__dict__.get('assigned_object_id') self._original_assigned_object_type_id = self.__dict__.get('assigned_object_type_id') @property def ipv6_full(self): if self.address and self.address.version == 6: return netaddr.IPAddress(self.address).format(netaddr.ipv6_full) return None def get_duplicates(self): return IPAddress.objects.filter( vrf=self.vrf, address__net_host=str(self.address.ip) ).exclude(pk=self.pk) def get_next_available_ip(self): """ Return the next available IP address within this IP's network (if any) """ if self.address and self.address.broadcast: start_ip = self.address.ip + 1 end_ip = self.address.broadcast - 1 if start_ip <= end_ip: available_ips = netaddr.IPSet(netaddr.IPRange(start_ip, end_ip)) available_ips -= netaddr.IPSet([ address.ip for address in IPAddress.objects.filter( vrf=self.vrf, address__gt=self.address, address__net_contained_or_equal=self.address.cidr ).values_list('address', flat=True) ]) if available_ips: return next(iter(available_ips)) return None def get_related_ips(self): """ Return all IPAddresses belonging to the same VRF. """ return IPAddress.objects.exclude(address=str(self.address)).filter( vrf=self.vrf, address__net_contained_or_equal=str(self.address) ) def clean(self): super().clean() if self.address: # /0 masks are not acceptable if self.address.prefixlen == 0: raise ValidationError({ 'address': _("Cannot create IP address with /0 mask.") }) # Do not allow assigning a network ID or broadcast address to an interface. if self.assigned_object: if self.address.ip == self.address.network: msg = _("{ip} is a network ID, which may not be assigned to an interface.").format( ip=self.address.ip ) if self.address.version == 4 and self.address.prefixlen not in (31, 32): raise ValidationError(msg) if self.address.version == 6 and self.address.prefixlen not in (127, 128): raise ValidationError(msg) if ( self.address.version == 4 and self.address.ip == self.address.broadcast and self.address.prefixlen not in (31, 32) ): msg = _("{ip} is a broadcast address, which may not be assigned to an interface.").format( ip=self.address.ip ) raise ValidationError(msg) # Enforce unique IP space (if applicable) if (self.vrf is None and get_config().ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique): duplicate_ips = self.get_duplicates() if duplicate_ips and ( self.role not in IPADDRESS_ROLES_NONUNIQUE or any(dip.role not in IPADDRESS_ROLES_NONUNIQUE for dip in duplicate_ips) ): table = _("VRF {vrf}").format(vrf=self.vrf) if self.vrf else _("global table") raise ValidationError({ 'address': _("Duplicate IP address found in {table}: {ipaddress}").format( table=table, ipaddress=duplicate_ips.first(), ) }) # Disallow the creation of IPAddresses within an IPRange with mark_populated=True parent_range_qs = IPRange.objects.filter( start_address__host__inet__lte=self.address.ip, end_address__host__inet__gte=self.address.ip, vrf=self.vrf, mark_populated=True, ) if not self.pk and (parent_range := parent_range_qs.first()): raise ValidationError({ 'address': _( "Cannot create IP address {ip} inside range {range}." ).format(ip=self.address, range=parent_range) }) if self._original_assigned_object_id and self._original_assigned_object_type_id: parent = getattr(self.assigned_object, 'parent_object', None) ct = ContentType.objects.get_for_id(self._original_assigned_object_type_id) original_assigned_object = ct.get_object_for_this_type(pk=self._original_assigned_object_id) original_parent = getattr(original_assigned_object, 'parent_object', None) # can't use is_primary_ip as self.assigned_object might be changed is_primary = False if self.family == 4 and hasattr(original_parent, 'primary_ip4'): if original_parent.primary_ip4_id == self.pk: is_primary = True if self.family == 6 and hasattr(original_parent, 'primary_ip6'): if original_parent.primary_ip6_id == self.pk: is_primary = True if is_primary and (parent != original_parent): raise ValidationError( _("Cannot reassign IP address while it is designated as the primary IP for the parent object") ) # can't use is_oob_ip as self.assigned_object might be changed if hasattr(original_parent, 'oob_ip') and original_parent.oob_ip_id == self.pk: if parent != original_parent: raise ValidationError( _("Cannot reassign IP address while it is designated as the OOB IP for the parent object") ) # Validate IP status selection if self.status == IPAddressStatusChoices.STATUS_SLAAC and self.family != 6: raise ValidationError({ 'status': _("Only IPv6 addresses can be assigned SLAAC status") }) def save(self, *args, **kwargs): # Force dns_name to lowercase self.dns_name = self.dns_name.lower() super().save(*args, **kwargs) def clone(self): attrs = super().clone() # Populate the address field with the next available IP (if any) if next_available_ip := self.get_next_available_ip(): attrs['address'] = f'{next_available_ip}/{self.address.prefixlen}' return attrs def to_objectchange(self, action): objectchange = super().to_objectchange(action) objectchange.related_object = self.assigned_object return objectchange @property def family(self): if not self.address: return None if isinstance(self.address, str): return netaddr.IPNetwork(self.address).version return self.address.version @property def is_oob_ip(self): if self.assigned_object: parent = getattr(self.assigned_object, 'parent_object', None) if hasattr(parent, 'oob_ip') and parent.oob_ip_id == self.pk: return True return False @property def is_primary_ip(self): if self.assigned_object: parent = getattr(self.assigned_object, 'parent_object', None) if self.family == 4 and hasattr(parent, 'primary_ip4') and parent.primary_ip4_id == self.pk: return True if self.family == 6 and hasattr(parent, 'primary_ip6') and parent.primary_ip6_id == self.pk: return True return False def _set_mask_length(self, value): """ Expose the IPNetwork object's prefixlen attribute on the parent model so that it can be manipulated directly, e.g. for bulk editing. """ if self.address is not None: self.address.prefixlen = value mask_length = property(fset=_set_mask_length) def get_status_color(self): return IPAddressStatusChoices.colors.get(self.status) def get_role_color(self): return IPAddressRoleChoices.colors.get(self.role)