Coverage for ipam/models/ip.py: 47%
500 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import netaddr
2from django.contrib.contenttypes.fields import GenericForeignKey
3from django.contrib.contenttypes.models import ContentType
4from django.contrib.postgres.indexes import GistIndex
5from django.core.exceptions import ValidationError
6from django.db import models
7from django.db.models import F
8from django.db.models.functions import Cast
9from django.utils.functional import cached_property
10from django.utils.translation import gettext_lazy as _
12from dcim.models.mixins import CachedScopeMixin
13from ipam.choices import *
14from ipam.constants import *
15from ipam.fields import IPAddressField, IPNetworkField
16from ipam.lookups import Host
17from ipam.managers import IPAddressManager
18from ipam.querysets import IPRangeQuerySet, PrefixQuerySet
19from ipam.validators import DNSValidator
20from netbox.config import get_config
21from netbox.models import OrganizationalModel, PrimaryModel
22from netbox.models.features import ContactsMixin
24__all__ = (
25 'RIR',
26 'Aggregate',
27 'IPAddress',
28 'IPRange',
29 'Prefix',
30 'Role',
31)
34class GetAvailablePrefixesMixin:
36 def get_available_prefixes(self):
37 """
38 Return all available prefixes within this Aggregate or Prefix as an IPSet.
39 """
40 params = {
41 'prefix__net_contained': str(self.prefix)
42 }
43 if hasattr(self, 'vrf'): 43 ↛ 46line 43 didn't jump to line 46 because the condition on line 43 was always true
44 params['vrf'] = self.vrf
46 child_prefixes = Prefix.objects.filter(**params).values_list('prefix', flat=True)
47 return netaddr.IPSet(self.prefix) - netaddr.IPSet(child_prefixes)
49 def get_first_available_prefix(self):
50 """
51 Return the first available child prefix within the prefix (or None).
52 """
53 available_prefixes = self.get_available_prefixes()
54 if not available_prefixes:
55 return None
56 return available_prefixes.iter_cidrs()[0]
59class RIR(OrganizationalModel):
60 """
61 A Regional Internet Registry (RIR) is responsible for the allocation of a large portion of the global IP address
62 space. This can be an organization like ARIN or RIPE, or a governing standard such as RFC 1918.
63 """
64 is_private = models.BooleanField(
65 default=False,
66 verbose_name=_('private'),
67 help_text=_('IP space managed by this RIR is considered private')
68 )
70 class Meta:
71 ordering = ('name',)
72 verbose_name = _('RIR')
73 verbose_name_plural = _('RIRs')
76class Aggregate(ContactsMixin, GetAvailablePrefixesMixin, PrimaryModel):
77 """
78 An aggregate exists at the root level of the IP address space hierarchy in NetBox. Aggregates are used to organize
79 the hierarchy and track the overall utilization of available address space. Each Aggregate is assigned to a RIR.
80 """
81 prefix = IPNetworkField(
82 help_text=_("IPv4 or IPv6 network")
83 )
84 rir = models.ForeignKey(
85 to='ipam.RIR',
86 on_delete=models.PROTECT,
87 related_name='aggregates',
88 verbose_name=_('RIR'),
89 help_text=_("Regional Internet Registry responsible for this IP space")
90 )
91 tenant = models.ForeignKey(
92 to='tenancy.Tenant',
93 on_delete=models.PROTECT,
94 related_name='aggregates',
95 blank=True,
96 null=True
97 )
98 date_added = models.DateField(
99 verbose_name=_('date added'),
100 blank=True,
101 null=True
102 )
104 clone_fields = (
105 'rir', 'tenant', 'date_added', 'description',
106 )
107 prerequisite_models = (
108 'ipam.RIR',
109 )
111 class Meta:
112 ordering = ('prefix', 'pk') # prefix may be non-unique
113 indexes = (
114 models.Index(fields=('prefix', 'id')), # Default ordering
115 )
116 verbose_name = _('aggregate')
117 verbose_name_plural = _('aggregates')
119 def __str__(self):
120 return str(self.prefix)
122 def clean(self):
123 super().clean()
125 if self.prefix:
127 # /0 masks are not acceptable
128 if self.prefix.prefixlen == 0:
129 raise ValidationError({
130 'prefix': _("Cannot create aggregate with /0 mask.")
131 })
133 # Ensure that the aggregate being added is not covered by an existing aggregate
134 covering_aggregates = Aggregate.objects.filter(
135 prefix__net_contains_or_equals=str(self.prefix)
136 )
137 if self.pk:
138 covering_aggregates = covering_aggregates.exclude(pk=self.pk)
139 if covering_aggregates:
140 raise ValidationError({
141 'prefix': _(
142 "Aggregates cannot overlap. {prefix} is already covered by an existing aggregate ({aggregate})."
143 ).format(
144 prefix=self.prefix,
145 aggregate=covering_aggregates[0]
146 )
147 })
149 # Ensure that the aggregate being added does not cover an existing aggregate
150 covered_aggregates = Aggregate.objects.filter(prefix__net_contained=str(self.prefix))
151 if self.pk:
152 covered_aggregates = covered_aggregates.exclude(pk=self.pk)
153 if covered_aggregates:
154 raise ValidationError({
155 'prefix': _(
156 "Prefixes cannot overlap aggregates. {prefix} covers an existing aggregate ({aggregate})."
157 ).format(
158 prefix=self.prefix,
159 aggregate=covered_aggregates[0]
160 )
161 })
163 @property
164 def family(self):
165 if not self.prefix:
166 return None
167 if isinstance(self.prefix, str):
168 return netaddr.IPNetwork(self.prefix).version
169 return self.prefix.version
171 @property
172 def ipv6_full(self):
173 if self.prefix and self.prefix.version == 6:
174 return netaddr.IPAddress(self.prefix).format(netaddr.ipv6_full)
175 return None
177 def get_child_prefixes(self):
178 """
179 Return all Prefixes within this Aggregate
180 """
181 return Prefix.objects.filter(prefix__net_contained=str(self.prefix))
183 def get_utilization(self):
184 """
185 Determine the prefix utilization of the aggregate and return it as a percentage.
186 """
187 queryset = Prefix.objects.filter(prefix__net_contained_or_equal=str(self.prefix))
188 child_prefixes = netaddr.IPSet([p.prefix for p in queryset])
189 utilization = float(child_prefixes.size) / self.prefix.size * 100
191 return min(utilization, 100)
194class Role(OrganizationalModel):
195 """
196 A Role represents the functional role of a Prefix or VLAN; for example, "Customer," "Infrastructure," or
197 "Management."
198 """
199 weight = models.PositiveSmallIntegerField(
200 verbose_name=_('weight'),
201 default=1000
202 )
204 class Meta:
205 ordering = ('weight', 'name')
206 indexes = (
207 models.Index(fields=('weight', 'name')), # Default ordering
208 )
209 verbose_name = _('role')
210 verbose_name_plural = _('roles')
212 def __str__(self):
213 return self.name
216class Prefix(ContactsMixin, GetAvailablePrefixesMixin, CachedScopeMixin, PrimaryModel):
217 """
218 A Prefix represents an IPv4 or IPv6 network, including mask length. Prefixes can optionally be scoped to certain
219 areas and/or assigned to VRFs. A Prefix must be assigned a status and may optionally be assigned a used-define Role.
220 A Prefix can also be assigned to a VLAN where appropriate.
221 """
222 prefix = IPNetworkField(
223 verbose_name=_('prefix'),
224 help_text=_('IPv4 or IPv6 network with mask')
225 )
226 vrf = models.ForeignKey(
227 to='ipam.VRF',
228 on_delete=models.PROTECT,
229 related_name='prefixes',
230 blank=True,
231 null=True,
232 verbose_name=_('VRF')
233 )
234 tenant = models.ForeignKey(
235 to='tenancy.Tenant',
236 on_delete=models.PROTECT,
237 related_name='prefixes',
238 blank=True,
239 null=True
240 )
241 vlan = models.ForeignKey(
242 to='ipam.VLAN',
243 on_delete=models.PROTECT,
244 related_name='prefixes',
245 blank=True,
246 null=True
247 )
248 status = models.CharField(
249 max_length=50,
250 choices=PrefixStatusChoices,
251 default=PrefixStatusChoices.STATUS_ACTIVE,
252 verbose_name=_('status'),
253 help_text=_('Operational status of this prefix')
254 )
255 role = models.ForeignKey(
256 to='ipam.Role',
257 on_delete=models.SET_NULL,
258 related_name='prefixes',
259 blank=True,
260 null=True,
261 help_text=_('The primary function of this prefix')
262 )
263 is_pool = models.BooleanField(
264 verbose_name=_('is a pool'),
265 default=False,
266 help_text=_('All IP addresses within this prefix are considered usable')
267 )
268 mark_utilized = models.BooleanField(
269 verbose_name=_('mark utilized'),
270 default=False,
271 help_text=_("Treat as fully utilized")
272 )
274 # Cached depth & child counts
275 _depth = models.PositiveSmallIntegerField(
276 default=0,
277 editable=False
278 )
279 _children = models.PositiveBigIntegerField(
280 default=0,
281 editable=False
282 )
284 objects = PrefixQuerySet.as_manager()
286 clone_fields = (
287 'scope', 'vrf', 'tenant', 'vlan', 'status', 'role', 'is_pool', 'mark_utilized', 'description',
288 )
290 class Meta:
291 ordering = (F('vrf').asc(nulls_first=True), 'prefix', 'pk') # (vrf, prefix) may be non-unique
292 verbose_name = _('prefix')
293 verbose_name_plural = _('prefixes')
294 indexes = (
295 models.Index(fields=('scope_type', 'scope_id')),
296 GistIndex(fields=['prefix'], name='ipam_prefix_gist_idx', opclasses=['inet_ops']),
297 )
299 def __init__(self, *args, **kwargs):
300 super().__init__(*args, **kwargs)
302 # Cache the original prefix and VRF so we can check if they have changed on post_save
303 self._prefix = self.__dict__.get('prefix')
304 self._vrf_id = self.__dict__.get('vrf_id')
306 def __str__(self):
307 return str(self.prefix)
309 def clean(self):
310 super().clean()
312 if self.prefix: 312 ↛ exitline 312 didn't return from function 'clean' because the condition on line 312 was always true
314 # /0 masks are not acceptable
315 if self.prefix.prefixlen == 0: 315 ↛ 316line 315 didn't jump to line 316 because the condition on line 315 was never true
316 raise ValidationError({
317 'prefix': _("Cannot create prefix with /0 mask.")
318 })
320 # Enforce unique IP space (if applicable)
321 if (self.vrf is None and get_config().ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique): 321 ↛ exitline 321 didn't return from function 'clean' because the condition on line 321 was always true
322 duplicate_prefixes = self.get_duplicates()
323 if duplicate_prefixes: 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true
324 table = _("VRF {vrf}").format(vrf=self.vrf) if self.vrf else _("global table")
325 raise ValidationError({
326 'prefix': _("Duplicate prefix found in {table}: {prefix}").format(
327 table=table,
328 prefix=duplicate_prefixes.first(),
329 )
330 })
332 def save(self, *args, **kwargs):
334 if isinstance(self.prefix, netaddr.IPNetwork):
336 # Clear host bits from prefix
337 self.prefix = self.prefix.cidr
339 # Cache objects associated with the terminating object (for filtering)
340 self.cache_related_objects()
342 super().save(*args, **kwargs)
344 @property
345 def family(self):
346 if not self.prefix: 346 ↛ 347line 346 didn't jump to line 347 because the condition on line 346 was never true
347 return None
348 if isinstance(self.prefix, str): 348 ↛ 349line 348 didn't jump to line 349 because the condition on line 348 was never true
349 return netaddr.IPNetwork(self.prefix).version
350 return self.prefix.version
352 @property
353 def mask_length(self):
354 if not self.prefix: 354 ↛ 355line 354 didn't jump to line 355 because the condition on line 354 was never true
355 return None
356 if isinstance(self.prefix, str): 356 ↛ 357line 356 didn't jump to line 357 because the condition on line 356 was never true
357 return netaddr.IPNetwork(self.prefix).prefixlen
358 return self.prefix.prefixlen
360 @property
361 def ipv6_full(self):
362 if self.prefix and self.prefix.version == 6:
363 return netaddr.IPAddress(self.prefix).format(netaddr.ipv6_full)
364 return None
366 @property
367 def depth(self):
368 return self._depth
370 @property
371 def children(self):
372 return self._children
374 def _set_prefix_length(self, value):
375 """
376 Expose the IPNetwork object's prefixlen attribute on the parent model so that it can be manipulated directly,
377 e.g. for bulk editing.
378 """
379 if self.prefix is not None:
380 self.prefix.prefixlen = value
381 prefix_length = property(fset=_set_prefix_length)
383 def get_status_color(self):
384 return PrefixStatusChoices.colors.get(self.status)
386 @cached_property
387 def aggregate(self):
388 """
389 Return the containing Aggregate for this Prefix, if any.
390 """
391 try:
392 return Aggregate.objects.get(prefix__net_contains_or_equals=str(self.prefix))
393 except Aggregate.DoesNotExist:
394 return None
396 def get_parents(self, include_self=False):
397 """
398 Return all containing Prefixes in the hierarchy.
399 """
400 lookup = 'net_contains_or_equals' if include_self else 'net_contains'
401 return Prefix.objects.filter(**{
402 'vrf_id': self.vrf_id,
403 f'prefix__{lookup}': self.prefix
404 })
406 def get_children(self, include_self=False):
407 """
408 Return all covered Prefixes in the hierarchy.
409 """
410 lookup = 'net_contained_or_equal' if include_self else 'net_contained'
411 return Prefix.objects.filter(**{
412 'vrf_id': self.vrf_id,
413 f'prefix__{lookup}': self.prefix
414 })
416 def get_duplicates(self):
417 return Prefix.objects.filter(vrf=self.vrf, prefix=str(self.prefix)).exclude(pk=self.pk)
419 def get_child_prefixes(self):
420 """
421 Return all Prefixes within this Prefix and VRF. If this Prefix is a container in the global table, return child
422 Prefixes belonging to any VRF.
423 """
424 if self.vrf is None and self.status == PrefixStatusChoices.STATUS_CONTAINER:
425 return Prefix.objects.filter(prefix__net_contained=str(self.prefix))
426 return Prefix.objects.filter(prefix__net_contained=str(self.prefix), vrf=self.vrf)
428 @property
429 def usable_ip_bounds(self):
430 """
431 Return the first and last IPs considered usable for available-IP calculations.
433 Pools and IPv4 /31-/32 / IPv6 /127-/128 are fully usable; otherwise IPv4 excludes
434 network and broadcast, IPv6 excludes the subnet-router anycast address.
435 """
436 network = netaddr.IPNetwork(self.prefix)
437 family = network.version
438 first = network.first
439 last = network.last
440 mask_length = network.prefixlen
442 if ( 442 ↛ 447line 442 didn't jump to line 447 because the condition on line 442 was never true
443 self.is_pool
444 or (family == 4 and mask_length >= 31)
445 or (family == 6 and mask_length >= 127)
446 ):
447 return (
448 netaddr.IPAddress(first, version=family),
449 netaddr.IPAddress(last, version=family),
450 )
452 if family == 4: 452 ↛ 458line 452 didn't jump to line 458 because the condition on line 452 was always true
453 return (
454 netaddr.IPAddress(first + 1, version=family),
455 netaddr.IPAddress(last - 1, version=family),
456 )
458 return (
459 netaddr.IPAddress(first + 1, version=family),
460 netaddr.IPAddress(last, version=family),
461 )
463 @property
464 def usable_size(self):
465 """
466 The number of usable host addresses within the prefix (excludes reserved addresses).
467 """
468 first_ip, last_ip = self.usable_ip_bounds
469 return int(last_ip) - int(first_ip) + 1
471 def get_child_ranges(self, **kwargs):
472 """
473 Return all IPRanges within this Prefix and VRF.
474 """
475 # A host BETWEEN over the prefix span uses the ipam_iprange_*_host btree indexes.
476 prefix = netaddr.IPNetwork(self.prefix)
477 bounds = (
478 netaddr.IPAddress(prefix.first, version=prefix.version),
479 netaddr.IPAddress(prefix.last, version=prefix.version),
480 )
481 return IPRange.objects.filter(
482 vrf=self.vrf,
483 start_address__host_between=bounds,
484 end_address__host_between=bounds,
485 **kwargs
486 )
488 def get_child_ips(self):
489 """
490 Return all IPAddresses within this Prefix and VRF. If this Prefix is a container in the global table, return
491 child IPAddresses belonging to any VRF.
492 """
493 # A host BETWEEN over the prefix span is index-sargable without the <<= containment recheck.
494 prefix = netaddr.IPNetwork(self.prefix)
495 bounds = (
496 netaddr.IPAddress(prefix.first, version=prefix.version),
497 netaddr.IPAddress(prefix.last, version=prefix.version),
498 )
499 if self.vrf is None and self.status == PrefixStatusChoices.STATUS_CONTAINER: 499 ↛ 500line 499 didn't jump to line 500 because the condition on line 499 was never true
500 return IPAddress.objects.filter(address__host_between=bounds)
501 return IPAddress.objects.filter(address__host_between=bounds, vrf=self.vrf)
503 def get_available_ips(self):
504 """
505 Return all available IPs within this prefix as an IPSet.
506 """
507 return netaddr.IPSet(
508 cidr
509 for start, end in self._available_intervals()
510 for cidr in netaddr.iprange_to_cidrs(start, end)
511 )
513 def iter_available_ips(self):
514 """
515 Yield the available IPs within this prefix as netaddr.IPAddress objects, in
516 ascending order. Unlike get_available_ips(), consumption is lazy: stopping
517 early stops reading from the database.
518 """
519 for start, end in self._available_intervals():
520 yield from netaddr.iter_iprange(start, end)
522 def get_available_ip_count(self):
523 """
524 Return the number of available IPs within the prefix.
525 """
526 first_ip, last_ip = self.usable_ip_bounds
527 usable_size = int(last_ip) - int(first_ip) + 1
529 populated_intervals = self.get_child_ranges(mark_populated=True).get_intervals(first_ip, last_ip)
530 populated_count = sum(int(end) - int(start) + 1 for start, end in populated_intervals)
532 # Populated ranges already cover the usable span; skip the child-IP count entirely.
533 if populated_count >= usable_size:
534 return 0
536 child_ip_count = (
537 self.get_child_ips()
538 .filter(address__host_between=(first_ip, last_ip))
539 .count_distinct_hosts(exclude_intervals=populated_intervals)
540 )
542 return max(usable_size - populated_count - child_ip_count, 0)
544 def get_ip_usage_summary(self):
545 """
546 Return the available IP count and utilization together as a dict, sharing a
547 single distinct-host scan. Intended for detail views rendering both values;
548 list views should call get_utilization() alone, which is cheaper per row.
549 """
550 # Marked-utilized and container utilization need no host scan; delegate.
551 if self.mark_utilized or self.status == PrefixStatusChoices.STATUS_CONTAINER:
552 return {
553 'available_ip_count': self.get_available_ip_count(),
554 'utilization': self.get_utilization(),
555 }
557 first_ip, last_ip = self.usable_ip_bounds
558 usable_size = int(last_ip) - int(first_ip) + 1
560 populated_intervals = self.get_child_ranges(mark_populated=True).get_intervals(first_ip, last_ip)
561 utilized_intervals = self.get_child_ranges(mark_utilized=True).get_intervals()
563 counts = self.get_child_ips().count_distinct_hosts_pair(
564 bounds=(first_ip, last_ip),
565 bounded_exclude=populated_intervals,
566 total_exclude=utilized_intervals,
567 )
569 populated_count = sum(int(end) - int(start) + 1 for start, end in populated_intervals)
570 utilized_range_count = sum(int(end) - int(start) + 1 for start, end in utilized_intervals)
572 prefix_size = self._get_utilization_denominator()
574 return {
575 'available_ip_count': max(usable_size - populated_count - counts['bounded'], 0),
576 'utilization': min(float(utilized_range_count + counts['total']) / prefix_size * 100, 100),
577 }
579 def get_first_available_ip(self):
580 """
581 Return the first available IP within the prefix (or None).
582 """
583 first_ip, last_ip = self.usable_ip_bounds
584 populated_intervals = self.get_child_ranges(mark_populated=True).get_intervals(first_ip, last_ip)
586 first_available_ip = self.get_child_ips().first_available_host(
587 first_ip, last_ip, exclude_intervals=populated_intervals,
588 )
590 if first_available_ip is None:
591 return None
592 return f'{first_available_ip}/{self.prefix.prefixlen}'
594 def get_utilization(self):
595 """
596 Determine the utilization of the prefix and return it as a percentage. For Prefixes with a status of
597 "container", calculate utilization based on child prefixes. For all others, count child IP addresses.
598 """
599 if self.mark_utilized:
600 return 100
602 if self.status == PrefixStatusChoices.STATUS_CONTAINER:
603 queryset = Prefix.objects.filter(
604 prefix__net_contained=str(self.prefix),
605 vrf=self.vrf
606 )
607 child_prefixes = netaddr.IPSet([p.prefix for p in queryset])
608 utilization = float(child_prefixes.size) / self.prefix.size * 100
609 else:
610 prefix_size = self._get_utilization_denominator()
611 utilized_intervals = self.get_child_ranges(mark_utilized=True).get_intervals()
612 utilized_range_count = sum(int(end) - int(start) + 1 for start, end in utilized_intervals)
614 # Utilized ranges already saturate the prefix; skip the child-IP count.
615 if utilized_range_count >= prefix_size:
616 return 100
618 child_ip_count = self.get_child_ips().count_distinct_hosts(
619 exclude_intervals=utilized_intervals,
620 )
622 utilization = float(utilized_range_count + child_ip_count) / prefix_size * 100
624 return min(utilization, 100)
626 def _available_intervals(self):
627 """
628 Yield the available (start, end) host intervals within the prefix.
629 """
630 first_ip, last_ip = self.usable_ip_bounds
631 populated_intervals = self.get_child_ranges(mark_populated=True).get_intervals(first_ip, last_ip)
633 return self.get_child_ips().available_intervals(
634 first_ip, last_ip, exclude_intervals=populated_intervals,
635 )
637 def _get_utilization_denominator(self):
638 """
639 The address count utilization is measured against (IPv4 non-pool prefixes
640 exclude the network and broadcast addresses; IPv6 uses the full prefix size).
641 """
642 prefix_size = self.prefix.size
643 if self.prefix.version == 4 and self.prefix.prefixlen < 31 and not self.is_pool:
644 return prefix_size - 2
645 return prefix_size
648class IPRange(ContactsMixin, PrimaryModel):
649 """
650 A range of IP addresses, defined by start and end addresses.
651 """
652 start_address = IPAddressField(
653 verbose_name=_('start address'),
654 help_text=_('IPv4 or IPv6 address (with mask)')
655 )
656 end_address = IPAddressField(
657 verbose_name=_('end address'),
658 help_text=_('IPv4 or IPv6 address (with mask)')
659 )
660 size = models.PositiveIntegerField(
661 verbose_name=_('size'),
662 editable=False
663 )
664 vrf = models.ForeignKey(
665 to='ipam.VRF',
666 on_delete=models.PROTECT,
667 related_name='ip_ranges',
668 blank=True,
669 null=True,
670 verbose_name=_('VRF')
671 )
672 tenant = models.ForeignKey(
673 to='tenancy.Tenant',
674 on_delete=models.PROTECT,
675 related_name='ip_ranges',
676 blank=True,
677 null=True
678 )
679 status = models.CharField(
680 verbose_name=_('status'),
681 max_length=50,
682 choices=IPRangeStatusChoices,
683 default=IPRangeStatusChoices.STATUS_ACTIVE,
684 help_text=_('Operational status of this range')
685 )
686 role = models.ForeignKey(
687 to='ipam.Role',
688 on_delete=models.SET_NULL,
689 related_name='ip_ranges',
690 blank=True,
691 null=True,
692 help_text=_('The primary function of this range')
693 )
694 mark_populated = models.BooleanField(
695 verbose_name=_('mark populated'),
696 default=False,
697 help_text=_("Prevent the creation of IP addresses within this range")
698 )
699 mark_utilized = models.BooleanField(
700 verbose_name=_('mark utilized'),
701 default=False,
702 help_text=_("Report space as fully utilized")
703 )
705 objects = IPRangeQuerySet.as_manager()
707 clone_fields = (
708 'vrf', 'tenant', 'status', 'role', 'description', 'mark_populated', 'mark_utilized',
709 )
711 class Meta:
712 ordering = (F('vrf').asc(nulls_first=True), 'start_address', 'pk') # (vrf, start_address) may be non-unique
713 indexes = (
714 models.Index(
715 Cast(Host('start_address'), output_field=IPAddressField()),
716 name='ipam_iprange_start_host',
717 ),
718 models.Index(
719 Cast(Host('end_address'), output_field=IPAddressField()),
720 name='ipam_iprange_end_host',
721 ),
722 )
723 verbose_name = _('IP range')
724 verbose_name_plural = _('IP ranges')
726 def __str__(self):
727 return self.name
729 def clean(self):
730 super().clean()
732 if self.start_address and self.end_address:
734 # Check that start & end IP versions match
735 if self.start_address.version != self.end_address.version:
736 raise ValidationError({
737 'end_address': _("Starting and ending IP address versions must match")
738 })
740 # Check that the start & end IP prefix lengths match
741 if self.start_address.prefixlen != self.end_address.prefixlen:
742 raise ValidationError({
743 'end_address': _("Starting and ending IP address masks must match")
744 })
746 # Equal start/end addresses are permitted for single-address ranges.
747 # Use .ip (host portion) rather than the IPNetwork object to avoid comparing prefix lengths again.
748 if self.end_address.ip < self.start_address.ip:
749 raise ValidationError({
750 'end_address': _(
751 "Ending address must be greater than or equal to the starting address ({start_address})"
752 ).format(start_address=self.start_address)
753 })
755 # Check for overlapping ranges
756 overlapping_ranges = (
757 IPRange.objects.exclude(pk=self.pk)
758 .filter(vrf=self.vrf)
759 .filter(
760 # Starts inside
761 Q(
762 start_address__host__inet__gte=self.start_address.ip,
763 start_address__host__inet__lte=self.end_address.ip,
764 ) |
765 # Ends inside
766 Q(
767 end_address__host__inet__gte=self.start_address.ip,
768 end_address__host__inet__lte=self.end_address.ip,
769 ) |
770 # Starts & ends outside
771 Q(
772 start_address__host__inet__lte=self.start_address.ip,
773 end_address__host__inet__gte=self.end_address.ip,
774 )
775 )
776 )
777 if overlapping_ranges.exists():
778 raise ValidationError(
779 _("Defined addresses overlap with range {overlapping_range} in VRF {vrf}").format(
780 overlapping_range=overlapping_ranges.first(),
781 vrf=self.vrf
782 ))
784 # Validate maximum size
785 MAX_SIZE = 2 ** 32 - 1
786 if int(self.end_address.ip - self.start_address.ip) + 1 > MAX_SIZE:
787 raise ValidationError(
788 _("Defined range exceeds maximum supported size ({max_size})").format(max_size=MAX_SIZE)
789 )
791 def save(self, *args, **kwargs):
793 # Record the range's size (number of IP addresses)
794 self.size = int(self.end_address.ip - self.start_address.ip) + 1
796 super().save(*args, **kwargs)
798 @property
799 def family(self):
800 if not self.start_address:
801 return None
802 if isinstance(self.start_address, str):
803 return netaddr.IPAddress(self.start_address.split('/')[0]).version
804 return self.start_address.version
806 @property
807 def range(self):
808 return netaddr.IPRange(self.start_address.ip, self.end_address.ip)
810 @property
811 def mask_length(self):
812 return self.start_address.prefixlen if self.start_address else None
814 @cached_property
815 def name(self):
816 """
817 Return an efficient string representation of the IP range.
818 """
819 # Single-address ranges render both endpoints to stay distinct from IPAddress rows.
820 if self.start_address.ip == self.end_address.ip:
821 return f'{self.start_address.ip}-{self.end_address.ip}/{self.start_address.prefixlen}'
823 separator = ':' if self.family == 6 else '.'
824 start_chunks = str(self.start_address.ip).split(separator)
825 end_chunks = str(self.end_address.ip).split(separator)
827 base_chunks = []
828 for a, b in zip(start_chunks, end_chunks):
829 if a == b:
830 base_chunks.append(a)
832 base_str = separator.join(base_chunks)
833 start_str = separator.join(start_chunks[len(base_chunks):])
834 end_str = separator.join(end_chunks[len(base_chunks):])
836 return f'{base_str}{separator}{start_str}-{end_str}/{self.start_address.prefixlen}'
838 def _set_prefix_length(self, value):
839 """
840 Expose the IPRange object's prefixlen attribute on the parent model so that it can be manipulated directly,
841 e.g. for bulk editing.
842 """
843 self.start_address.prefixlen = value
844 self.end_address.prefixlen = value
845 prefix_length = property(fset=_set_prefix_length)
847 def get_status_color(self):
848 return IPRangeStatusChoices.colors.get(self.status)
850 @cached_property
851 def first_available_ip(self):
852 """
853 Return the first available IP within the range (or None).
854 """
855 return self.get_first_available_ip()
857 @property
858 def utilization(self):
859 """
860 Determine the utilization of the range and return it as a percentage.
861 """
862 if self.mark_utilized:
863 return 100
865 return min(float(self._occupied_host_count) / self.size * 100, 100)
867 def get_child_ips(self):
868 """
869 Return all IPAddresses within this IPRange and VRF.
870 """
871 return IPAddress.objects.filter(
872 vrf=self.vrf,
873 address__host_between=(self.start_address.ip, self.end_address.ip),
874 )
876 def get_available_ips(self):
877 """
878 Return all available IPs within this range as an IPSet.
879 """
880 return netaddr.IPSet(
881 cidr
882 for start, end in self._available_intervals()
883 for cidr in netaddr.iprange_to_cidrs(start, end)
884 )
886 def iter_available_ips(self):
887 """
888 Yield the available IPs within this range as netaddr.IPAddress objects, in
889 ascending order. Unlike get_available_ips(), consumption is lazy: stopping
890 early stops reading from the database.
891 """
892 for start, end in self._available_intervals():
893 yield from netaddr.iter_iprange(start, end)
895 def get_available_ip_count(self):
896 """
897 Return the number of available IPs within the range.
898 """
899 if self.mark_populated:
900 return 0
902 return max(self.size - self._occupied_host_count, 0)
904 def get_first_available_ip(self):
905 """
906 Return the first available IP within the range (or None).
907 """
908 if self.mark_populated:
909 return None
911 first_available_ip = self.get_child_ips().first_available_host(
912 self.start_address.ip, self.end_address.ip,
913 )
915 if first_available_ip is None:
916 return None
918 return f'{first_available_ip}/{self.start_address.prefixlen}'
920 def _available_intervals(self):
921 """
922 Yield the available (start, end) host intervals within the range.
923 """
924 if self.mark_populated:
925 return iter(())
927 return self.get_child_ips().available_intervals(
928 self.start_address.ip, self.end_address.ip,
929 )
931 @cached_property
932 def _occupied_host_count(self):
933 """
934 The number of distinct occupied hosts within the range, cached for the
935 lifetime of the instance.
936 """
937 return self.get_child_ips().count_distinct_hosts()
940class IPAddress(ContactsMixin, PrimaryModel):
941 """
942 An IPAddress represents an individual IPv4 or IPv6 address and its mask. The mask length should match what is
943 configured in the real world. (Typically, only loopback interfaces are configured with /32 or /128 masks.) Like
944 Prefixes, IPAddresses can optionally be assigned to a VRF. An IPAddress can optionally be assigned to an Interface.
945 Interfaces can have zero or more IPAddresses assigned to them.
947 An IPAddress can also optionally point to a NAT inside IP, designating itself as a NAT outside IP. This is useful,
948 for example, when mapping public addresses to private addresses. When an Interface has been assigned an IPAddress
949 which has a NAT outside IP, that Interface's Device can use either the inside or outside IP as its primary IP.
950 """
951 address = IPAddressField(
952 verbose_name=_('address'),
953 help_text=_('IPv4 or IPv6 address (with mask)')
954 )
955 vrf = models.ForeignKey(
956 to='ipam.VRF',
957 on_delete=models.PROTECT,
958 related_name='ip_addresses',
959 blank=True,
960 null=True,
961 verbose_name=_('VRF')
962 )
963 tenant = models.ForeignKey(
964 to='tenancy.Tenant',
965 on_delete=models.PROTECT,
966 related_name='ip_addresses',
967 blank=True,
968 null=True
969 )
970 status = models.CharField(
971 verbose_name=_('status'),
972 max_length=50,
973 choices=IPAddressStatusChoices,
974 default=IPAddressStatusChoices.STATUS_ACTIVE,
975 help_text=_('The operational status of this IP')
976 )
977 role = models.CharField(
978 verbose_name=_('role'),
979 max_length=50,
980 choices=IPAddressRoleChoices,
981 blank=True,
982 null=True,
983 help_text=_('The functional role of this IP')
984 )
985 assigned_object_type = models.ForeignKey(
986 to='contenttypes.ContentType',
987 on_delete=models.PROTECT,
988 related_name='+',
989 blank=True,
990 null=True
991 )
992 assigned_object_id = models.PositiveBigIntegerField(
993 blank=True,
994 null=True
995 )
996 assigned_object = GenericForeignKey(
997 ct_field='assigned_object_type',
998 fk_field='assigned_object_id'
999 )
1000 nat_inside = models.ForeignKey(
1001 to='self',
1002 on_delete=models.SET_NULL,
1003 related_name='nat_outside',
1004 blank=True,
1005 null=True,
1006 verbose_name=_('NAT (inside)'),
1007 help_text=_('The IP for which this address is the "outside" IP')
1008 )
1009 dns_name = models.CharField(
1010 max_length=255,
1011 blank=True,
1012 validators=[DNSValidator],
1013 verbose_name=_('DNS name'),
1014 help_text=_('Hostname or FQDN (not case-sensitive)')
1015 )
1017 objects = IPAddressManager()
1019 clone_fields = (
1020 'vrf', 'tenant', 'status', 'role', 'dns_name', 'description',
1021 )
1023 class Meta:
1024 ordering = ('address', 'pk') # address may be non-unique
1025 indexes = (
1026 models.Index(fields=('address', 'id')),
1027 # Default ordering (see IPAddressManager). The primary key must be included so that the index can
1028 # satisfy the ordering outright; without it PostgreSQL falls back to an incremental sort, which
1029 # measurably slows deep pagination.
1030 models.Index(
1031 Cast(Host('address'), output_field=IPAddressField()), F('id'), name='ipam_ipaddress_host'
1032 ),
1033 models.Index(fields=('assigned_object_type', 'assigned_object_id')),
1034 )
1035 verbose_name = _('IP address')
1036 verbose_name_plural = _('IP addresses')
1038 def __str__(self):
1039 return str(self.address)
1041 def __init__(self, *args, **kwargs):
1042 super().__init__(*args, **kwargs)
1044 # Denote the original assigned object (if any) for validation in clean()
1045 self._original_assigned_object_id = self.__dict__.get('assigned_object_id')
1046 self._original_assigned_object_type_id = self.__dict__.get('assigned_object_type_id')
1048 @property
1049 def ipv6_full(self):
1050 if self.address and self.address.version == 6:
1051 return netaddr.IPAddress(self.address).format(netaddr.ipv6_full)
1052 return None
1054 def get_duplicates(self):
1055 return IPAddress.objects.filter(
1056 vrf=self.vrf,
1057 address__net_host=str(self.address.ip)
1058 ).exclude(pk=self.pk)
1060 def get_next_available_ip(self):
1061 """
1062 Return the next available IP address within this IP's network (if any)
1063 """
1064 if self.address and self.address.broadcast:
1065 start_ip = self.address.ip + 1
1066 end_ip = self.address.broadcast - 1
1067 if start_ip <= end_ip:
1068 available_ips = netaddr.IPSet(netaddr.IPRange(start_ip, end_ip))
1069 available_ips -= netaddr.IPSet([
1070 address.ip for address in IPAddress.objects.filter(
1071 vrf=self.vrf,
1072 address__gt=self.address,
1073 address__net_contained_or_equal=self.address.cidr
1074 ).values_list('address', flat=True)
1075 ])
1076 if available_ips:
1077 return next(iter(available_ips))
1078 return None
1080 def get_related_ips(self):
1081 """
1082 Return all IPAddresses belonging to the same VRF.
1083 """
1084 return IPAddress.objects.exclude(address=str(self.address)).filter(
1085 vrf=self.vrf, address__net_contained_or_equal=str(self.address)
1086 )
1088 def clean(self):
1089 super().clean()
1091 if self.address: 1091 ↛ 1147line 1091 didn't jump to line 1147 because the condition on line 1091 was always true
1093 # /0 masks are not acceptable
1094 if self.address.prefixlen == 0: 1094 ↛ 1095line 1094 didn't jump to line 1095 because the condition on line 1094 was never true
1095 raise ValidationError({
1096 'address': _("Cannot create IP address with /0 mask.")
1097 })
1099 # Do not allow assigning a network ID or broadcast address to an interface.
1100 if self.assigned_object: 1100 ↛ 1119line 1100 didn't jump to line 1119 because the condition on line 1100 was always true
1101 if self.address.ip == self.address.network: 1101 ↛ 1102line 1101 didn't jump to line 1102 because the condition on line 1101 was never true
1102 msg = _("{ip} is a network ID, which may not be assigned to an interface.").format(
1103 ip=self.address.ip
1104 )
1105 if self.address.version == 4 and self.address.prefixlen not in (31, 32):
1106 raise ValidationError(msg)
1107 if self.address.version == 6 and self.address.prefixlen not in (127, 128):
1108 raise ValidationError(msg)
1109 if ( 1109 ↛ 1113line 1109 didn't jump to line 1113 because the condition on line 1109 was never true
1110 self.address.version == 4 and self.address.ip == self.address.broadcast and
1111 self.address.prefixlen not in (31, 32)
1112 ):
1113 msg = _("{ip} is a broadcast address, which may not be assigned to an interface.").format(
1114 ip=self.address.ip
1115 )
1116 raise ValidationError(msg)
1118 # Enforce unique IP space (if applicable)
1119 if (self.vrf is None and get_config().ENFORCE_GLOBAL_UNIQUE) or (self.vrf and self.vrf.enforce_unique): 1119 ↛ 1134line 1119 didn't jump to line 1134 because the condition on line 1119 was always true
1120 duplicate_ips = self.get_duplicates()
1121 if duplicate_ips and ( 1121 ↛ 1125line 1121 didn't jump to line 1125 because the condition on line 1121 was never true
1122 self.role not in IPADDRESS_ROLES_NONUNIQUE or
1123 any(dip.role not in IPADDRESS_ROLES_NONUNIQUE for dip in duplicate_ips)
1124 ):
1125 table = _("VRF {vrf}").format(vrf=self.vrf) if self.vrf else _("global table")
1126 raise ValidationError({
1127 'address': _("Duplicate IP address found in {table}: {ipaddress}").format(
1128 table=table,
1129 ipaddress=duplicate_ips.first(),
1130 )
1131 })
1133 # Disallow the creation of IPAddresses within an IPRange with mark_populated=True
1134 parent_range_qs = IPRange.objects.filter(
1135 start_address__host__inet__lte=self.address.ip,
1136 end_address__host__inet__gte=self.address.ip,
1137 vrf=self.vrf,
1138 mark_populated=True,
1139 )
1140 if not self.pk and (parent_range := parent_range_qs.first()): 1140 ↛ 1141line 1140 didn't jump to line 1141 because the condition on line 1140 was never true
1141 raise ValidationError({
1142 'address': _(
1143 "Cannot create IP address {ip} inside range {range}."
1144 ).format(ip=self.address, range=parent_range)
1145 })
1147 if self._original_assigned_object_id and self._original_assigned_object_type_id: 1147 ↛ 1175line 1147 didn't jump to line 1175 because the condition on line 1147 was always true
1148 parent = getattr(self.assigned_object, 'parent_object', None)
1149 ct = ContentType.objects.get_for_id(self._original_assigned_object_type_id)
1150 original_assigned_object = ct.get_object_for_this_type(pk=self._original_assigned_object_id)
1151 original_parent = getattr(original_assigned_object, 'parent_object', None)
1153 # can't use is_primary_ip as self.assigned_object might be changed
1154 is_primary = False
1155 if self.family == 4 and hasattr(original_parent, 'primary_ip4'): 1155 ↛ 1158line 1155 didn't jump to line 1158 because the condition on line 1155 was always true
1156 if original_parent.primary_ip4_id == self.pk: 1156 ↛ 1157line 1156 didn't jump to line 1157 because the condition on line 1156 was never true
1157 is_primary = True
1158 if self.family == 6 and hasattr(original_parent, 'primary_ip6'): 1158 ↛ 1159line 1158 didn't jump to line 1159 because the condition on line 1158 was never true
1159 if original_parent.primary_ip6_id == self.pk:
1160 is_primary = True
1162 if is_primary and (parent != original_parent): 1162 ↛ 1163line 1162 didn't jump to line 1163 because the condition on line 1162 was never true
1163 raise ValidationError(
1164 _("Cannot reassign IP address while it is designated as the primary IP for the parent object")
1165 )
1167 # can't use is_oob_ip as self.assigned_object might be changed
1168 if hasattr(original_parent, 'oob_ip') and original_parent.oob_ip_id == self.pk: 1168 ↛ 1169line 1168 didn't jump to line 1169 because the condition on line 1168 was never true
1169 if parent != original_parent:
1170 raise ValidationError(
1171 _("Cannot reassign IP address while it is designated as the OOB IP for the parent object")
1172 )
1174 # Validate IP status selection
1175 if self.status == IPAddressStatusChoices.STATUS_SLAAC and self.family != 6: 1175 ↛ 1176line 1175 didn't jump to line 1176 because the condition on line 1175 was never true
1176 raise ValidationError({
1177 'status': _("Only IPv6 addresses can be assigned SLAAC status")
1178 })
1180 def save(self, *args, **kwargs):
1182 # Force dns_name to lowercase
1183 self.dns_name = self.dns_name.lower()
1185 super().save(*args, **kwargs)
1187 def clone(self):
1188 attrs = super().clone()
1190 # Populate the address field with the next available IP (if any)
1191 if next_available_ip := self.get_next_available_ip():
1192 attrs['address'] = f'{next_available_ip}/{self.address.prefixlen}'
1194 return attrs
1196 def to_objectchange(self, action):
1197 objectchange = super().to_objectchange(action)
1198 objectchange.related_object = self.assigned_object
1199 return objectchange
1201 @property
1202 def family(self):
1203 if not self.address: 1203 ↛ 1204line 1203 didn't jump to line 1204 because the condition on line 1203 was never true
1204 return None
1205 if isinstance(self.address, str): 1205 ↛ 1206line 1205 didn't jump to line 1206 because the condition on line 1205 was never true
1206 return netaddr.IPNetwork(self.address).version
1207 return self.address.version
1209 @property
1210 def is_oob_ip(self):
1211 if self.assigned_object:
1212 parent = getattr(self.assigned_object, 'parent_object', None)
1213 if hasattr(parent, 'oob_ip') and parent.oob_ip_id == self.pk:
1214 return True
1215 return False
1217 @property
1218 def is_primary_ip(self):
1219 if self.assigned_object:
1220 parent = getattr(self.assigned_object, 'parent_object', None)
1221 if self.family == 4 and hasattr(parent, 'primary_ip4') and parent.primary_ip4_id == self.pk:
1222 return True
1223 if self.family == 6 and hasattr(parent, 'primary_ip6') and parent.primary_ip6_id == self.pk:
1224 return True
1225 return False
1227 def _set_mask_length(self, value):
1228 """
1229 Expose the IPNetwork object's prefixlen attribute on the parent model so that it can be manipulated directly,
1230 e.g. for bulk editing.
1231 """
1232 if self.address is not None:
1233 self.address.prefixlen = value
1234 mask_length = property(fset=_set_mask_length)
1236 def get_status_color(self):
1237 return IPAddressStatusChoices.colors.get(self.status)
1239 def get_role_color(self):
1240 return IPAddressRoleChoices.colors.get(self.role)