Coverage for netbox_data_import/ip_assignment.py: 98.28%
174 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 09:47 +0000
1# SPDX-License-Identifier: Apache-2.0
2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""Where an address off a source row lands on a device.
5The preview names the interface and the sync writes to it, so both read this module. Two copies
6would let a row promise one interface and the write pick another.
7"""
9from __future__ import annotations
11import ipaddress
12import re
13from copy import copy
14from dataclasses import dataclass
15from typing import Any
16from .public_refusal import PublicRefusal
18# `oob_ip` carries no family in NetBox, so it takes either.
19IP_FIELD_FAMILY: dict[str, int | None] = {"primary_ip4": 4, "primary_ip6": 6, "oob_ip": None}
21# Bounded at both ends: a word cannot leave a shorter valid address, and 45 covers the longest one.
22_IP_TOKEN = re.compile(r"(?<![0-9A-Za-z])[0-9A-Fa-f:.]{1,45}(?:/\d{1,3})?(?![0-9A-Za-z])")
25class IPAssignmentError(PublicRefusal):
26 """The address cannot be placed, with the repair the operator has to make."""
29@dataclass(frozen=True)
30class IPTarget:
31 """The interface an address would land on, and the row it would use."""
33 address: str
34 interface: Any
35 existing: Any | None
36 already_held: bool
38 @property
39 def interface_name(self) -> str:
40 """Return the interface name, or an empty string when the row is unassigned."""
41 return getattr(self.interface, "name", "") or ""
43 @property
44 def held(self):
45 """Return the stored row the device already carries, which only a held target has."""
46 if not self.already_held or self.existing is None:
47 raise IPAssignmentError(f"{self.address} is not already on this device, so it has no stored row.")
48 return self.existing
50 @property
51 def summary(self) -> str:
52 """Return what the sync reports once it has written."""
53 if not self.interface_name:
54 return self.address
55 return f"{self.address} on {self.interface_name}"
57 @property
58 def placement(self) -> str:
59 """Return what the preview shows beside the address it already prints."""
60 if not self.interface_name:
61 return ""
62 if self.already_held:
63 # The stored mask can differ from the one the row states, so it is worth printing.
64 return f"already on {self.interface_name} as {self.address}"
65 return f"would go to {self.interface_name}"
68@dataclass(frozen=True)
69class ProspectiveAddress:
70 """The final IPAddress row and fields whose values are generated during execution."""
72 instance: Any
73 generated_fields: frozenset[str] = frozenset()
76def _normalized_ip(token: str) -> str | None:
77 """Return *token* as 'address/prefix', or None when it is not one address."""
78 try:
79 if "/" in token:
80 return str(ipaddress.ip_interface(token))
81 addr = ipaddress.ip_address(token)
82 except ValueError:
83 return None
84 return f"{addr}/32" if addr.version == 4 else f"{addr}/128"
87def parse_address(raw_value) -> str | None:
88 """Return the one address a source value names, as 'address/prefix', or None.
90 Sources export an address inside a label or with a separator appended, so the whole value is
91 tried first and the addresses spelled inside it only after that.
92 """
93 raw = str(raw_value).strip()
94 if not raw:
95 return None
96 whole = _normalized_ip(raw)
97 if whole is not None:
98 return whole
99 for raw_token in _IP_TOKEN.findall(raw):
100 token = raw_token
101 while token:
102 found = _normalized_ip(token)
103 if found is not None:
104 return found
105 if token[-1] not in ".:":
106 break
107 token = token[:-1]
108 return None
111def normalized_address(field: str, value) -> str:
112 """Return the address the row names as 'host/prefix', in the family the field takes."""
113 address = parse_address(value)
114 if address is None:
115 raise IPAssignmentError(f"Cannot read an IP address from '{value}'.")
116 family = IP_FIELD_FAMILY.get(field)
117 version = ipaddress.ip_interface(address).version
118 if family is not None and version != family:
119 raise IPAssignmentError(f"'{value}' is an IPv{version} address; this field takes IPv{family}.")
120 return address
123def already_assigned(device, field, address) -> bool:
124 """Return whether the device already carries exactly this address on *field*.
126 The writer selects the address that this device holds. The primary field must point to that
127 same row, while duplicate rows held by other objects do not affect the settled state.
128 """
129 current = getattr(device, field, None)
130 if current is None:
131 return False
132 try:
133 if _host(current.address) != _host(address):
134 return False
135 held = held_by_device(device, address)
136 except ValueError:
137 return False
138 return held is not None and held.pk == current.pk
141def _host(address) -> str:
142 """Return the host part, which is what identifies an address inside one VRF."""
143 return str(ipaddress.ip_interface(str(address)).ip)
146def held_by_device(device, address):
147 """Return the address this device already carries, or None.
149 Every interface is searched, not only a management one, and the mask is ignored: a workbook
150 states a bare address where the device holds the same host inside its real subnet.
152 An address the device points at but no interface of its own holds is not a match. NetBox
153 requires an IP field to name an address on that device, so that state is repaired through the
154 normal path rather than copied onto a second field.
155 """
156 from ipam.models import IPAddress
158 wanted = _host(address)
159 candidates = IPAddress.objects.filter(interface__device=device).select_related("vrf")
160 matches = [candidate for candidate in candidates if _host(candidate.address) == wanted]
161 if not matches:
162 return None
163 # A management interface answers first when the device holds the address more than once.
164 matches.sort(key=lambda ip: not getattr(ip.assigned_object, "mgmt_only", False))
165 return matches[0]
168def interface_for(device):
169 """Return the interface an address off a source file belongs on.
171 A management interface answers first: that is what the device type marks it for, and an
172 address in a source workbook is a management address far more often than not.
173 """
174 from dcim.models import InterfaceTemplate
176 interfaces = sorted(device.interfaces.all(), key=lambda i: (not i.mgmt_only, i.name))
177 if interfaces:
178 return interfaces[0]
179 model = device.device_type.model
180 declared = list(InterfaceTemplate.objects.filter(device_type=device.device_type).order_by("name")[:5])
181 if not declared:
182 raise IPAssignmentError(
183 f"The device type '{model}' declares no interfaces, so there is nowhere to put this "
184 f"address. Add an interface to the device type, then sync again."
185 )
186 names = ", ".join(template.name for template in declared)
187 raise IPAssignmentError(
188 f"This device has none of the interfaces its type '{model}' declares ({names}). "
189 f"Add them to the device, then sync again."
190 )
193def resolve(device, field: str, value) -> IPTarget:
194 """Return where *value* would land on *device*, or raise with the repair to make."""
195 from ipam.models import IPAddress
197 address = normalized_address(field, value)
198 held = held_by_device(device, address)
199 if held is not None:
200 return IPTarget(address=str(held.address), interface=held.assigned_object, existing=held, already_held=True)
202 interface = interface_for(device)
203 # The interface's VRF scopes the address: the same host in another VRF is a different address.
204 existing = IPAddress.objects.filter(address__net_host=_host(address), vrf=interface.vrf).first()
205 if existing is not None and existing.assigned_object is not None:
206 owner = getattr(existing.assigned_object, "device", None) or existing.assigned_object
207 raise IPAssignmentError(f"Address {existing.address} is already assigned to '{owner}'.")
208 return IPTarget(address=address, interface=interface, existing=existing, already_held=False)
211def _prospective_new_device_target(device, field: str, value, interface) -> IPTarget:
212 """Return the target a new Device will get after its interface templates instantiate."""
213 from ipam.models import IPAddress
215 address = normalized_address(field, value)
216 existing = IPAddress.objects.filter(address__net_host=_host(address), vrf=interface.vrf).first()
217 if existing is not None and existing.assigned_object is not None:
218 owner = getattr(existing.assigned_object, "device", None) or existing.assigned_object
219 raise IPAssignmentError(f"Address {existing.address} is already assigned to '{owner}'.")
220 return IPTarget(address=address, interface=interface, existing=existing, already_held=False)
223def prospective_addresses(device, ip_fields) -> dict[str, ProspectiveAddress]:
224 """Return placeable final address rows without saving or mutating caller-owned rows."""
225 from dcim.models import InterfaceTemplate
226 from ipam.models import IPAddress
228 future_interface = None
229 if device.pk is None:
230 templates = list(InterfaceTemplate.objects.filter(device_type=device.device_type))
231 templates.sort(key=lambda template: (not template.mgmt_only, template.name))
232 if templates:
233 future_interface = templates[0].instantiate(device=device)
234 future_interface.pk = -1
235 while type(future_interface)._base_manager.filter(pk=future_interface.pk).exists():
236 future_interface.pk -= 1
237 else:
238 return {}
240 found = {}
241 shared: dict[tuple[str, int | None], ProspectiveAddress] = {}
242 for field, value in ip_fields.items():
243 try:
244 target = (
245 _prospective_new_device_target(device, field, value, future_interface)
246 if future_interface is not None
247 else resolve(device, field, value)
248 )
249 except IPAssignmentError:
250 continue
251 vrf_key = getattr(target.interface, "vrf_id", None)
252 key = (_host(target.address), vrf_key)
253 prospective = shared.get(key)
254 if prospective is None:
255 address = (
256 copy(target.existing)
257 if target.existing is not None
258 else IPAddress(
259 address=target.address,
260 vrf=target.interface.vrf,
261 )
262 )
263 generated_fields: frozenset[str] = frozenset()
264 if not target.already_held:
265 address.assigned_object = target.interface
266 if future_interface is not None:
267 generated_fields = frozenset({"assigned_object_id"})
268 prospective = ProspectiveAddress(address, generated_fields)
269 shared[key] = prospective
270 found[field] = prospective
271 return found
274def apply(target: IPTarget, user=None):
275 """Write the address *target* names onto its interface, and return the IPAddress.
277 The row sync and the import writer both land here, so an address is created, scoped and
278 checked the same way whichever one runs.
279 """
280 from ipam.models import IPAddress
282 from .object_permissions import enforce_saved_object_permission
284 address = target.existing
285 action = "change" if address is not None else "add"
286 if address is None:
287 address = IPAddress(address=target.address, vrf=target.interface.vrf)
288 else:
289 address.snapshot()
290 address.assigned_object = target.interface
291 address.full_clean()
292 address.save()
293 # An ObjectPermission's constraints are only evaluated against the saved row.
294 enforce_saved_object_permission(address, user, action)
295 return address