Coverage for ipam/utils.py: 16%
198 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from dataclasses import dataclass
3import netaddr
4from django.apps import apps
5from django.core.exceptions import ValidationError
6from django.db.models import BooleanField, F, Func, Q
7from django.utils.translation import gettext_lazy as _
9from .constants import *
11__all__ = (
12 'PORT_MAPPING_LOOKUPS',
13 'AvailableIPSpace',
14 'PortMappingMatch',
15 'add_available_vlans',
16 'add_requested_prefixes',
17 'annotate_ip_space',
18 'expand_port_mapping',
19 'get_next_available_prefix',
20 'group_port_mapping_rows',
21 'group_port_mappings',
22 'legacy_protocol_and_ports',
23 'normalize_port_mapping',
24 'port_mapping_q',
25 'rebuild_prefixes',
26 'sorted_int_ports',
27 'split_port_mapping',
28)
31@dataclass
32class AvailableIPSpace:
33 """
34 A representation of available IP space between two IP addresses/ranges.
35 """
36 size: int
37 first_ip: str
39 @property
40 def title(self):
41 if self.size == 1:
42 return _('1 IP available')
43 if self.size <= 65536:
44 return _('{count} IPs available').format(count=self.size)
45 return _('Many IPs available')
48def add_requested_prefixes(parent, prefix_list, show_available=True, show_assigned=True):
49 """
50 Return a list of requested prefixes using show_available, show_assigned filters. If available prefixes are
51 requested, create fake Prefix objects for all unallocated space within a prefix.
53 :param parent: Parent Prefix instance
54 :param prefix_list: Child prefixes list (or queryset)
55 :param show_available: Include available prefixes.
56 :param show_assigned: Show assigned prefixes.
57 """
58 child_prefixes = []
60 # Add available prefixes to the table if requested
61 if prefix_list and show_available:
62 Prefix = apps.get_model('ipam', 'Prefix')
64 # Find all unallocated space, add fake Prefix objects to child_prefixes.
65 # IMPORTANT: These are unsaved Prefix instances (pk=None). If this is ever changed to use
66 # saved Prefix instances with real pks, bulk delete will fail for mixed-type selections
67 # due to single-model form validation. See: https://github.com/netbox-community/netbox/issues/21176
68 available_prefixes = netaddr.IPSet(parent) ^ netaddr.IPSet([p.prefix for p in prefix_list])
69 available_prefixes = [Prefix(prefix=p, status=None) for p in available_prefixes.iter_cidrs()]
70 child_prefixes = child_prefixes + available_prefixes
72 # Add assigned prefixes to the table if requested
73 if prefix_list and show_assigned:
74 child_prefixes = child_prefixes + list(prefix_list)
76 # Sort child prefixes after additions
77 child_prefixes.sort(key=lambda p: p.prefix)
79 return child_prefixes
82def annotate_ip_space(prefix, *, ip_addresses=None, ip_ranges=None):
83 """
84 Return a prefix's child ranges and IPs interleaved with available space records.
86 :param prefix: Parent Prefix instance
87 :param ip_addresses: Child IP addresses queryset (defaults to all child IPs)
88 :param ip_ranges: Child IP ranges queryset (defaults to all populated child ranges)
89 """
90 if ip_addresses is None:
91 ip_addresses = prefix.get_child_ips()
92 if ip_ranges is None:
93 ip_ranges = prefix.get_child_ranges(mark_populated=True)
95 # Compile child objects
96 records = []
97 records.extend([
98 (iprange.start_address.ip, iprange) for iprange in ip_ranges
99 ])
100 records.extend([
101 (ip.address.ip, ip) for ip in ip_addresses
102 ])
103 records = sorted(records, key=lambda x: x[0])
105 # Determine the first & last valid IP addresses in the prefix
106 first_ip_in_prefix, last_ip_in_prefix = prefix.usable_ip_bounds
108 if not records:
109 return [
110 AvailableIPSpace(
111 size=int(last_ip_in_prefix - first_ip_in_prefix + 1),
112 first_ip=f'{first_ip_in_prefix}/{prefix.mask_length}'
113 )
114 ]
116 output = []
117 prev_ip = None
119 # Account for any available IPs before the first real IP
120 if records[0][0] > first_ip_in_prefix:
121 output.append(AvailableIPSpace(
122 size=int(records[0][0] - first_ip_in_prefix),
123 first_ip=f'{first_ip_in_prefix}/{prefix.mask_length}'
124 ))
126 # Add IP ranges & addresses, annotating available space in between records
127 for record in records:
128 if prev_ip:
129 # Annotate available space
130 if (diff := int(record[0]) - int(prev_ip)) > 1:
131 first_skipped = f'{prev_ip + 1}/{prefix.mask_length}'
132 output.append(AvailableIPSpace(
133 size=diff - 1,
134 first_ip=first_skipped
135 ))
137 output.append(record[1])
139 # Update the previous IP address
140 if hasattr(record[1], 'end_address'):
141 prev_ip = record[1].end_address.ip
142 else:
143 prev_ip = record[0]
145 # Include any remaining available IPs
146 if prev_ip < last_ip_in_prefix:
147 output.append(AvailableIPSpace(
148 size=int(last_ip_in_prefix - prev_ip),
149 first_ip=f'{prev_ip + 1}/{prefix.mask_length}'
150 ))
152 return output
155def available_vlans_from_range(vlans, vlan_group, vid_range):
156 """
157 Create fake records for all gaps between used VLANs
158 """
159 min_vid = int(vid_range.lower) if vid_range else VLAN_VID_MIN
160 max_vid = int(vid_range.upper) if vid_range else VLAN_VID_MAX
162 if not vlans:
163 return [{
164 'vid': min_vid,
165 'vlan_group': vlan_group,
166 'available': max_vid - min_vid
167 }]
169 prev_vid = min_vid - 1
170 new_vlans = []
171 for vlan in vlans:
173 # Ignore VIDs outside the range
174 if not min_vid <= vlan.vid < max_vid:
175 continue
177 # Annotate any available VIDs between the previous (or minimum) VID
178 # and the current VID
179 if vlan.vid - prev_vid > 1:
180 new_vlans.append({
181 'vid': prev_vid + 1,
182 'vlan_group': vlan_group,
183 'available': vlan.vid - prev_vid - 1,
184 })
186 prev_vid = vlan.vid
188 # Annotate any remaining available VLANs
189 if prev_vid < max_vid - 1:
190 new_vlans.append({
191 'vid': prev_vid + 1,
192 'vlan_group': vlan_group,
193 'available': max_vid - prev_vid - 1,
194 })
196 return new_vlans
199def add_available_vlans(vlans, vlan_group):
200 """
201 Create fake records for all gaps between used VLANs
202 """
203 new_vlans = []
204 for vid_range in vlan_group.vid_ranges:
205 new_vlans.extend(available_vlans_from_range(vlans, vlan_group, vid_range))
207 vlans = list(vlans) + new_vlans
208 vlans.sort(key=lambda v: v['vid'] if isinstance(v, dict) else v.vid)
210 return vlans
213def rebuild_prefixes(vrf):
214 """
215 Rebuild the prefix hierarchy for all prefixes in the specified VRF (or global table).
216 """
217 Prefix = apps.get_model('ipam', 'Prefix')
218 prefix_queryset = Prefix.objects.filter(vrf=vrf)
220 def contains(parent, child):
221 return child in parent and child != parent
223 def push_to_stack(prefix):
224 # Increment child count on parent nodes
225 for n in stack:
226 n['children'] += 1
227 stack.append({
228 'pk': [prefix['pk']],
229 'prefix': prefix['prefix'],
230 'children': 0,
231 })
233 stack = []
234 update_queue = []
235 prefixes = prefix_queryset.order_by('prefix', 'pk').values('pk', 'prefix')
237 # Iterate through all Prefixes in the table, growing and shrinking the stack as we go
238 for p in prefixes:
240 # Grow the stack if this is a child of the most recent prefix
241 if not stack or contains(stack[-1]['prefix'], p['prefix']):
242 push_to_stack(p)
244 # Handle duplicate prefixes
245 elif stack[-1]['prefix'] == p['prefix']:
246 stack[-1]['pk'].append(p['pk'])
248 # If this is a sibling or parent of the most recent prefix, pop nodes from the
249 # stack until we reach a parent prefix (or the root)
250 else:
251 while stack and not contains(stack[-1]['prefix'], p['prefix']):
252 node = stack.pop()
253 for pk in node['pk']:
254 update_queue.append(
255 Prefix(pk=pk, _depth=len(stack), _children=node['children'])
256 )
257 push_to_stack(p)
259 # Flush the update queue once it reaches 100 Prefixes
260 if len(update_queue) >= 100:
261 Prefix.objects.bulk_update(update_queue, ['_depth', '_children'])
262 update_queue = []
264 # Clear out any prefixes remaining in the stack
265 while stack:
266 node = stack.pop()
267 for pk in node['pk']:
268 update_queue.append(
269 Prefix(pk=pk, _depth=len(stack), _children=node['children'])
270 )
272 # Final flush of any remaining Prefixes
273 Prefix.objects.bulk_update(update_queue, ['_depth', '_children'])
276def get_next_available_prefix(ipset, prefix_size):
277 """
278 Given a prefix length, allocate the next available prefix from an IPSet.
279 """
280 for available_prefix in ipset.iter_cidrs():
281 if prefix_size >= available_prefix.prefixlen:
282 allocated_prefix = f"{available_prefix.network}/{prefix_size}"
283 ipset.remove(allocated_prefix)
284 return allocated_prefix
285 return None
288#
289# Service port mappings
290#
292def split_port_mapping(mapping):
293 """
294 Split a ``protocol/port`` string (e.g. ``'tcp/80'``) into its ``(protocol, port)`` parts. A missing
295 separator or port yields an empty string for that part, leaving validation to report the problem.
296 """
297 protocol, _sep, port = mapping.partition('/')
298 return protocol, port
301def normalize_port_mapping(mapping):
302 """
303 Canonicalize a single ``protocol/port`` string as far as possible *without raising*: the protocol is
304 lowercased and a numeric port loses any leading zeros, so ``'TCP/080'`` becomes ``'tcp/80'``. Anything
305 unrecognized is returned unchanged, in which case it simply won't match a stored (always-canonical)
306 mapping.
308 This is the lookup-side counterpart to ``validate_port_mappings()``, which enforces the same
309 canonical form on write but rejects bad input. Filtering must not 400 on an unknown protocol or a
310 malformed pair — an empty result set is the right answer there — hence the separate, lenient variant.
311 """
312 # Imported lazily to avoid a circular import during settings load (ipam.choices reads
313 # settings.FIELD_CHOICES), matching validate_port_mappings().
314 from ipam.choices import ServiceProtocolChoices
316 protocol, port = split_port_mapping(mapping)
317 if not port or not port.isdigit():
318 return mapping
319 protocol = protocol.lower()
320 if protocol not in ServiceProtocolChoices.values():
321 return mapping
322 return f'{protocol}/{int(port)}'
325def group_port_mappings(mappings):
326 """
327 Group a flat ``['tcp/80', 'tcp/443', 'udp/53']`` list into an ordered ``{protocol: [ports]}`` dict,
328 preserving first-seen protocol order. Shared by the display property and the form widget so the
329 ``protocol/port`` string is parsed in exactly one place.
330 """
331 grouped = {}
332 for mapping in mappings:
333 protocol, port = split_port_mapping(mapping)
334 grouped.setdefault(protocol, []).append(port)
335 return grouped
338def group_port_mapping_rows(mappings):
339 """
340 Group a flat ``['tcp/80', 'tcp/443', 'udp/53']`` list into per-protocol rows
341 ``[{'protocol': 'tcp', 'ports': '80,443'}, {'protocol': 'udp', 'ports': '53'}]`` — the shape the
342 port-mapping form widget renders, one row per protocol.
343 """
344 return [
345 {'protocol': protocol, 'ports': ','.join(ports)}
346 for protocol, ports in group_port_mappings(mappings).items()
347 ]
350def sorted_int_ports(ports):
351 """
352 Sort a protocol's port strings numerically and return them as integers. Any entry that bypassed
353 validation (a raw SQL write, a plugin, or an unmigrated row) and isn't a plain integer is skipped
354 rather than raising, so a single malformed mapping degrades gracefully on API reads instead of
355 raising a 500 — mirroring the tolerance of ``ServiceBase.port_mappings_list``.
356 """
357 return sorted(int(port) for port in ports if str(port).isdigit())
360def legacy_protocol_and_ports(mappings):
361 """
362 Collapse port mappings into the deprecated single-protocol ``(protocol, ports)`` representation.
363 Single source of truth for the backward-compatibility contract shared by the REST serializers and
364 the GraphQL types:
366 * single protocol -> ``(protocol, [sorted int ports])``
367 * no mappings -> ``(None, [])`` (representable as an empty legacy ports list)
368 * multiple protocols -> ``(None, None)`` (not representable; ``ports=None`` signals "read
369 port_mappings instead")
370 * single protocol, but a port fails integer coercion (malformed raw/plugin data) -> ``(None, None)``
371 (a subset would be plausible-but-wrong, so signal "not representable" rather than silently
372 dropping the bad mapping)
373 """
374 grouped = group_port_mappings(mappings)
375 if len(grouped) == 1:
376 protocol, ports = next(iter(grouped.items()))
377 int_ports = sorted_int_ports(ports)
378 # If any port was dropped by coercion, the legacy single-protocol view can't faithfully
379 # represent this service; signal "not representable" instead of returning a partial list.
380 if len(int_ports) != len(ports):
381 return None, None
382 return protocol, int_ports
383 return (None, []) if not grouped else (None, None)
386# Whitelisted SQL comparison operators for the port half of a mapping, keyed by the django-filter
387# lookup name. Only these five names are ever interpolated into SQL by PortMappingMatch, so the
388# operator can never originate from user input.
389PORT_MAPPING_LOOKUPS = {
390 'exact': '=',
391 'gt': '>',
392 'gte': '>=',
393 'lt': '<',
394 'lte': '<=',
395}
397# The port half of an unnested mapping, as an integer. Guarded by a numeric test so a malformed mapping
398# written outside the ORM (raw SQL, a plugin) evaluates to NULL — which no comparison matches — instead
399# of aborting the whole query with an invalid-input-syntax error. Mirrors the tolerance that
400# sorted_int_ports() and ServiceBase.port_mappings_list already apply on reads.
401_PORT_MAPPING_PORT_SQL = (
402 "CASE WHEN split_part(port_mapping, '/', 2) ~ '^[0-9]+$' "
403 "THEN split_part(port_mapping, '/', 2)::integer END"
404)
407class PortMappingMatch(Func):
408 """
409 A boolean expression which is true for services having at least one port mapping that satisfies the
410 given protocol and port tests:
412 EXISTS (
413 SELECT 1 FROM unnest(port_mappings) AS port_mapping
414 WHERE split_part(port_mapping, '/', 1) = ANY(<protocols>)
415 AND <port> >= <value> AND <port> <= <value> ...
416 )
418 Testing every condition against the *same* unnested mapping is what keeps protocol and port
419 correlated: a service exposing tcp/80 and udp/9999 must not match ``protocol=tcp&port__gt=1000``,
420 and one exposing tcp/500 and tcp/5000 must not match ``port__gte=1000&port__lte=2000``.
422 This is deliberately a sequential scan. GIN's ``array_ops`` opclass supports only ``=``, ``&&``,
423 ``@>`` and ``<@``, so no array index can serve a range comparison, and the alternatives (a
424 trigger-maintained denormalized column, or a related table) either cannot express the correlation or
425 cost far more than the scan — measured at ~200 ms over 400k services and ~1 s over 2M.
426 ``port_mapping_q()`` therefore reserves this for the cases an array overlap cannot express and uses
427 the GIN-indexable overlap for exact protocol+port lookups.
428 """
429 output_field = BooleanField()
431 def __init__(self, protocols=(), port_tests=()):
432 """
433 Args:
434 protocols: protocol values to match, OR'd together.
435 port_tests: ``(lookup, values)`` pairs, where ``lookup`` is a key of
436 ``PORT_MAPPING_LOOKUPS``. Pairs are AND'd (and so must hold for one single mapping);
437 the values within a pair are OR'd, matching how django-filter's multi-value filters
438 combine ``?port=80&port=443``.
439 """
440 self.protocols = list(protocols or ())
441 self.port_tests = [
442 (lookup, list(values)) for lookup, values in (port_tests or ()) if values
443 ]
444 for lookup, _values in self.port_tests:
445 if lookup not in PORT_MAPPING_LOOKUPS:
446 raise ValueError(f"Unsupported port mapping lookup: {lookup}")
447 super().__init__(F('port_mappings'))
449 def as_sql(self, compiler, connection, **extra_context):
450 mappings_sql, mappings_params = compiler.compile(self.source_expressions[0])
451 conditions = []
452 params = list(mappings_params)
454 if self.protocols:
455 conditions.append("split_part(port_mapping, '/', 1) = ANY(%s)")
456 params.append(self.protocols)
457 for lookup, values in self.port_tests:
458 operator = PORT_MAPPING_LOOKUPS[lookup]
459 conditions.append('({})'.format(
460 ' OR '.join(f'{_PORT_MAPPING_PORT_SQL} {operator} %s' for _value in values)
461 ))
462 params.extend(values)
464 if not conditions:
465 # port_mapping_q() never builds an unconstrained match, but be explicit rather than emit an
466 # EXISTS with an empty WHERE clause.
467 return 'TRUE', []
469 sql = (
470 f"EXISTS (SELECT 1 FROM unnest({mappings_sql}) AS port_mapping "
471 f"WHERE {' AND '.join(conditions)})"
472 )
473 return sql, params
476def port_mapping_q(protocols=(), port_tests=()):
477 """
478 Build a ``Q`` filtering services by protocol and/or port, correlated so that a combined query must
479 be satisfied by a *single* mapping. See ``PortMappingMatch`` for the argument shapes.
481 A lone exact port test reduces to a GIN-indexable array overlap on ``port_mappings``
482 (``port_mappings && ['tcp/80', ...]`` — each element is one whole mapping, so an overlap means
483 "shares any mapping"); for a port-only query each port is paired with every valid protocol to keep
484 it a single overlap. Everything else — a protocol-only query, whose ports are unbounded and cannot
485 be enumerated, and any range lookup, which no array index can serve — falls back to
486 ``PortMappingMatch``. Shared by the FilterSet and the GraphQL filters.
487 """
488 # Imported lazily to avoid a circular import during settings load (ipam.choices reads
489 # settings.FIELD_CHOICES), matching ipam.validators.
490 from ipam.choices import ServiceProtocolChoices
492 protocols = list(protocols or ())
493 port_tests = [(lookup, list(values)) for lookup, values in (port_tests or ()) if values]
495 if not protocols and not port_tests: 495 ↛ 496line 495 didn't jump to line 496 because the condition on line 495 was never true
496 return Q()
498 if len(port_tests) == 1 and port_tests[0][0] == 'exact': 498 ↛ 506line 498 didn't jump to line 506 because the condition on line 498 was always true
499 # Every stored mapping's protocol is validated against ServiceProtocolChoices, so enumerating
500 # the (small, fixed) protocol set covers all valid data for a port-only query.
501 ports = port_tests[0][1]
502 mapping_protocols = protocols or ServiceProtocolChoices.values()
503 combos = [f'{protocol}/{port}' for protocol in mapping_protocols for port in ports]
504 return Q(port_mappings__overlap=combos)
506 return Q(PortMappingMatch(protocols=protocols, port_tests=port_tests))
509def expand_port_mapping(protocol, ports):
510 """
511 Expand a single protocol plus its ports into the model's flat ``['tcp/80', 'tcp/443', ...]`` tokens.
512 ``ports`` may be a comma/range string (the form widget's format, e.g. ``('tcp', '80,443,8000-8010')``)
513 or an already-expanded list of ports (e.g. set programmatically).
515 An empty ``ports`` yields a single bare ``'protocol/'`` token so ``validate_port_mappings`` reports a
516 clear "expected protocol/port" error (rather than ``parse_numeric_range`` raising a confusing
517 'Range "" is invalid'). An empty ``protocol`` raises a clear error rather than producing a ``'/80'``
518 token that surfaces as "Invalid protocol:" with a blank value. Shared by the model form field so the
519 protocol/port pairing is built in one place, and so every entry path gets the blank-protocol check.
520 """
521 # Imported lazily to avoid pulling the forms layer in at module load.
522 from utilities.forms.utils import parse_numeric_range
524 # No case-folding here: validate_port_mappings (which every token below flows through) matches the
525 # protocol case-insensitively and stores the canonical value.
526 protocol = (protocol or '').strip()
527 if not protocol:
528 # Ports given with no protocol would otherwise expand to '/80' and surface as a confusing
529 # "Invalid protocol:" with a blank value. Report the real problem instead, in wording that fits
530 # all entry paths that route through here (the form widget and CSV import).
531 raise ValidationError(_("Each port mapping must specify a protocol."))
533 if isinstance(ports, (list, tuple)):
534 # Already-expanded ports are paired as-is; validate_port_mappings() checks each value's range.
535 if not ports:
536 return [f'{protocol}/']
537 return [f'{protocol}/{port}' for port in ports]
539 ports_str = (ports or '').strip()
540 if not ports_str:
541 return [f'{protocol}/']
542 # parse_numeric_range validates each range against the port bounds (rejecting reversed and
543 # out-of-range values before expansion), so a non-empty string always yields >=1 port.
544 return [
545 f'{protocol}/{port}'
546 for port in parse_numeric_range(ports_str, min_value=SERVICE_PORT_MIN, max_value=SERVICE_PORT_MAX)
547 ]