Coverage for netbox_data_import/ip_assignment.py: 100%
125 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-09 20:50 +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 dataclasses import dataclass
14from typing import Any
16# `oob_ip` carries no family in NetBox, so it takes either.
17IP_FIELD_FAMILY: dict[str, int | None] = {"primary_ip4": 4, "primary_ip6": 6, "oob_ip": None}
19# Bounded at both ends: a word cannot leave a shorter valid address, and 45 covers the longest one.
20_IP_TOKEN = re.compile(r"(?<![0-9A-Za-z])[0-9A-Fa-f:.]{1,45}(?:/\d{1,3})?(?![0-9A-Za-z])")
23class IPAssignmentError(Exception):
24 """The address cannot be placed, with the repair the operator has to make."""
27@dataclass(frozen=True)
28class IPTarget:
29 """The interface an address would land on, and the row it would use."""
31 address: str
32 interface: Any
33 existing: Any | None
34 already_held: bool
36 @property
37 def interface_name(self) -> str:
38 """Return the interface name, or an empty string when the row is unassigned."""
39 return getattr(self.interface, "name", "") or ""
41 @property
42 def held(self):
43 """Return the stored row the device already carries, which only a held target has."""
44 if not self.already_held or self.existing is None:
45 raise IPAssignmentError(f"{self.address} is not already on this device, so it has no stored row.")
46 return self.existing
48 @property
49 def summary(self) -> str:
50 """Return what the sync reports once it has written."""
51 if not self.interface_name:
52 return self.address
53 return f"{self.address} on {self.interface_name}"
55 @property
56 def placement(self) -> str:
57 """Return what the preview shows beside the address it already prints."""
58 if not self.interface_name:
59 return ""
60 if self.already_held:
61 # The stored mask can differ from the one the row states, so it is worth printing.
62 return f"already on {self.interface_name} as {self.address}"
63 return f"would go to {self.interface_name}"
66def _normalized_ip(token: str) -> str | None:
67 """Return *token* as 'address/prefix', or None when it is not one address."""
68 try:
69 if "/" in token:
70 return str(ipaddress.ip_interface(token))
71 addr = ipaddress.ip_address(token)
72 except ValueError:
73 return None
74 return f"{addr}/32" if addr.version == 4 else f"{addr}/128"
77def parse_address(raw_value) -> str | None:
78 """Return the one address a source value names, as 'address/prefix', or None.
80 Sources export an address inside a label or with a separator appended, so the whole value is
81 tried first and the addresses spelled inside it only after that.
82 """
83 raw = str(raw_value).strip()
84 if not raw:
85 return None
86 whole = _normalized_ip(raw)
87 if whole is not None:
88 return whole
89 for raw_token in _IP_TOKEN.findall(raw):
90 token = raw_token
91 while token:
92 found = _normalized_ip(token)
93 if found is not None:
94 return found
95 if token[-1] not in ".:":
96 break
97 token = token[:-1]
98 return None
101def normalized_address(field: str, value) -> str:
102 """Return the address the row names as 'host/prefix', in the family the field takes."""
103 address = parse_address(value)
104 if address is None:
105 raise IPAssignmentError(f"Cannot read an IP address from '{value}'.")
106 family = IP_FIELD_FAMILY.get(field)
107 version = ipaddress.ip_interface(address).version
108 if family is not None and version != family:
109 raise IPAssignmentError(f"'{value}' is an IPv{version} address; this field takes IPv{family}.")
110 return address
113def already_assigned(device, field, address) -> bool:
114 """Return whether the device already carries exactly this address on *field*.
116 The writer selects the address that this device holds. The primary field must point to that
117 same row, while duplicate rows held by other objects do not affect the settled state.
118 """
119 current = getattr(device, field, None)
120 if current is None:
121 return False
122 try:
123 if _host(current.address) != _host(address):
124 return False
125 held = held_by_device(device, address)
126 except ValueError:
127 return False
128 return held is not None and held.pk == current.pk
131def _host(address) -> str:
132 """Return the host part, which is what identifies an address inside one VRF."""
133 return str(ipaddress.ip_interface(str(address)).ip)
136def held_by_device(device, address):
137 """Return the address this device already carries, or None.
139 Every interface is searched, not only a management one, and the mask is ignored: a workbook
140 states a bare address where the device holds the same host inside its real subnet.
142 An address the device points at but no interface of its own holds is not a match. NetBox
143 requires an IP field to name an address on that device, so that state is repaired through the
144 normal path rather than copied onto a second field.
145 """
146 from ipam.models import IPAddress
148 wanted = _host(address)
149 candidates = IPAddress.objects.filter(interface__device=device).select_related("vrf")
150 matches = [candidate for candidate in candidates if _host(candidate.address) == wanted]
151 if not matches:
152 return None
153 # A management interface answers first when the device holds the address more than once.
154 matches.sort(key=lambda ip: not getattr(ip.assigned_object, "mgmt_only", False))
155 return matches[0]
158def interface_for(device):
159 """Return the interface an address off a source file belongs on.
161 A management interface answers first: that is what the device type marks it for, and an
162 address in a source workbook is a management address far more often than not.
163 """
164 from dcim.models import InterfaceTemplate
166 interfaces = sorted(device.interfaces.all(), key=lambda i: (not i.mgmt_only, i.name))
167 if interfaces:
168 return interfaces[0]
169 model = device.device_type.model
170 declared = list(InterfaceTemplate.objects.filter(device_type=device.device_type).order_by("name")[:5])
171 if not declared:
172 raise IPAssignmentError(
173 f"The device type '{model}' declares no interfaces, so there is nowhere to put this "
174 f"address. Add an interface to the device type, then sync again."
175 )
176 names = ", ".join(template.name for template in declared)
177 raise IPAssignmentError(
178 f"This device has none of the interfaces its type '{model}' declares ({names}). "
179 f"Add them to the device, then sync again."
180 )
183def resolve(device, field: str, value) -> IPTarget:
184 """Return where *value* would land on *device*, or raise with the repair to make."""
185 from ipam.models import IPAddress
187 address = normalized_address(field, value)
188 held = held_by_device(device, address)
189 if held is not None:
190 return IPTarget(address=str(held.address), interface=held.assigned_object, existing=held, already_held=True)
192 interface = interface_for(device)
193 # The interface's VRF scopes the address: the same host in another VRF is a different address.
194 existing = IPAddress.objects.filter(address__net_host=_host(address), vrf=interface.vrf).first()
195 if existing is not None and existing.assigned_object is not None:
196 owner = getattr(existing.assigned_object, "device", None) or existing.assigned_object
197 raise IPAssignmentError(f"Address {existing.address} is already assigned to '{owner}'.")
198 return IPTarget(address=address, interface=interface, existing=existing, already_held=False)
201def apply(target: IPTarget, user=None):
202 """Write the address *target* names onto its interface, and return the IPAddress.
204 The row sync and the import writer both land here, so an address is created, scoped and
205 checked the same way whichever one runs.
206 """
207 from ipam.models import IPAddress
209 from .object_permissions import enforce_saved_object_permission
211 address = target.existing
212 action = "change" if address is not None else "add"
213 if address is None:
214 address = IPAddress(address=target.address, vrf=target.interface.vrf)
215 address.assigned_object = target.interface
216 address.full_clean()
217 address.save()
218 # An ObjectPermission's constraints are only evaluated against the saved row.
219 enforce_saved_object_permission(address, user, action)
220 return address