import ipaddress import logging from django.contrib.contenttypes.models import ContentType from django.db import transaction from ipam.models import Prefix from netbox.plugins import get_plugin_config logger = logging.getLogger(__name__) def network_for_address(address): """Return the canonical network encoded by an IPAddress.address value.""" return ipaddress.ip_interface(str(address)).network def _scope_from_ip(ip): """Derive the most specific supported Prefix scope from an IP assignment.""" assigned = getattr(ip, "assigned_object", None) if assigned is None: return None parent = ( getattr(assigned, "parent_object", None) or getattr(assigned, "device", None) or getattr(assigned, "virtual_machine", None) ) if parent is None: return None location = getattr(parent, "location", None) if location is not None: return location site = getattr(parent, "site", None) if site is not None: return site cluster = getattr(parent, "cluster", None) return getattr(cluster, "scope", None) if cluster is not None else None def _tenant_from_ip(ip): tenant = getattr(ip, "tenant", None) if tenant is not None: return tenant assigned = getattr(ip, "assigned_object", None) parent = ( getattr(assigned, "parent_object", None) or getattr(assigned, "device", None) or getattr(assigned, "virtual_machine", None) or assigned ) return getattr(parent, "tenant", None) @transaction.atomic def ensure_prefix_for_ip(ip, *, config=None): """Get or create the exact Prefix represented by an IP address and its mask. NetBox derives hierarchy from address/prefix + VRF. There is deliberately no parent FK to update on IPAddress; creating the missing Prefix is the assignment. """ settings = config or { key: get_plugin_config("netbox_better_ips", key) for key in ( "prefix_status", "inherit_scope", "inherit_tenant", "ignore_host_prefixes" ) } network = network_for_address(ip.address) if settings.get("ignore_host_prefixes", True) and network.prefixlen == network.max_prefixlen: return None, False lookup = {"prefix": str(network), "vrf_id": ip.vrf_id} existing = Prefix.objects.filter(**lookup).first() if existing: return existing, False defaults = {"status": settings.get("prefix_status", "active")} if settings.get("inherit_tenant", True): defaults["tenant"] = _tenant_from_ip(ip) if settings.get("inherit_scope", True): scope = _scope_from_ip(ip) if scope is not None: defaults["scope_type"] = ContentType.objects.get_for_model(scope) defaults["scope_id"] = scope.pk prefix, created = Prefix.objects.get_or_create(defaults=defaults, **lookup) if created: logger.info("Created missing prefix %s for IP address %s", prefix, ip) return prefix, created