netbox/netbox/ipam/models/ip.py

import netaddr
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
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', '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)')
    )
    l2vpn_terminations = GenericRelation(
        to='vpn.L2VPNTermination',
        content_type_field='assigned_object_type',
        object_id_field='assigned_object_id',
        related_query_name='ipaddress',
    )

    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)