MrBlake
acb4e449bc
Fehlende Prefixe können direkt aus der IPAM-Bereichsliste für alle sichtbaren IP-Adressen angelegt werden. Zusätzlich zeigt die Prefix-IP-Ansicht das zugewiesene Gerät oder die VM standardmäßig an.
123 lines
4.8 KiB
Python
123 lines
4.8 KiB
Python
from django.contrib.contenttypes.models import ContentType
|
|
from django.contrib import messages
|
|
from django.contrib.auth.decorators import permission_required
|
|
from django.contrib.auth.mixins import PermissionRequiredMixin
|
|
from django.core.paginator import Paginator
|
|
from django.db.models import Q
|
|
from django.shortcuts import redirect
|
|
from django.views.decorators.http import require_POST
|
|
from django.views.generic import TemplateView
|
|
from dcim.models import Location, Region, Site, SiteGroup
|
|
from ipam.models import IPAddress, Prefix
|
|
import netaddr
|
|
|
|
from .forms import NetworkOverviewFilterForm
|
|
from .services import ensure_prefix_for_ip
|
|
|
|
|
|
def _descendant_ids(obj):
|
|
"""Return an object's ID plus nested-tree descendants across NetBox versions."""
|
|
try:
|
|
return list(obj.get_descendants(include_self=True).values_list("pk", flat=True))
|
|
except (AttributeError, TypeError):
|
|
ids = list(obj.get_descendants().values_list("pk", flat=True))
|
|
return [obj.pk, *ids]
|
|
|
|
|
|
def _scope_query(model, ids):
|
|
content_type = ContentType.objects.get_for_model(model)
|
|
return Q(scope_type=content_type, scope_id__in=ids)
|
|
|
|
|
|
def apply_overview_filters(queryset, data):
|
|
if q := data.get("q"):
|
|
search = Q(description__icontains=q) | Q(comments__icontains=q)
|
|
try:
|
|
search |= Q(prefix__net_contains_or_equals=str(netaddr.IPNetwork(q)))
|
|
except (netaddr.AddrFormatError, ValueError):
|
|
pass
|
|
queryset = queryset.filter(search)
|
|
if tenant := data.get("tenant"):
|
|
queryset = queryset.filter(tenant=tenant)
|
|
if group := data.get("tenant_group"):
|
|
queryset = queryset.filter(tenant__group_id__in=_descendant_ids(group))
|
|
if location := data.get("location"):
|
|
queryset = queryset.filter(_scope_query(Location, _descendant_ids(location)))
|
|
if site := data.get("site"):
|
|
location_ids = Location.objects.filter(site=site).values_list("pk", flat=True)
|
|
queryset = queryset.filter(
|
|
_scope_query(Site, [site.pk]) | _scope_query(Location, location_ids)
|
|
)
|
|
if site_group := data.get("site_group"):
|
|
site_ids = Site.objects.filter(group_id__in=_descendant_ids(site_group)).values_list(
|
|
"pk", flat=True
|
|
)
|
|
location_ids = Location.objects.filter(site_id__in=site_ids).values_list("pk", flat=True)
|
|
queryset = queryset.filter(
|
|
_scope_query(SiteGroup, _descendant_ids(site_group))
|
|
| _scope_query(Site, site_ids)
|
|
| _scope_query(Location, location_ids)
|
|
)
|
|
if region := data.get("organization"):
|
|
region_ids = _descendant_ids(region)
|
|
site_ids = Site.objects.filter(region_id__in=region_ids).values_list("pk", flat=True)
|
|
location_ids = Location.objects.filter(site_id__in=site_ids).values_list("pk", flat=True)
|
|
queryset = queryset.filter(
|
|
_scope_query(Region, region_ids)
|
|
| _scope_query(Site, site_ids)
|
|
| _scope_query(Location, location_ids)
|
|
)
|
|
return queryset
|
|
|
|
|
|
class NetworkOverviewView(PermissionRequiredMixin, TemplateView):
|
|
template_name = "netbox_better_ips/network_overview.html"
|
|
permission_required = "ipam.view_prefix"
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
form = NetworkOverviewFilterForm(self.request.GET or None)
|
|
prefixes = Prefix.objects.restrict(self.request.user, "view").select_related(
|
|
"vrf", "tenant", "scope_type"
|
|
)
|
|
if form.is_valid():
|
|
prefixes = apply_overview_filters(prefixes, form.cleaned_data)
|
|
prefixes = prefixes.order_by("prefix", "vrf")
|
|
paginator = Paginator(prefixes, 50)
|
|
context.update(
|
|
{
|
|
"filter_form": form,
|
|
"page_obj": paginator.get_page(self.request.GET.get("page")),
|
|
"total_count": prefixes.count(),
|
|
}
|
|
)
|
|
return context
|
|
|
|
|
|
@require_POST
|
|
@permission_required(("ipam.view_ipaddress", "ipam.add_prefix"), raise_exception=True)
|
|
def reconcile_missing_prefixes(request):
|
|
created = skipped = failed = 0
|
|
addresses = IPAddress.objects.restrict(request.user, "view").select_related(
|
|
"tenant", "assigned_object_type"
|
|
)
|
|
for address in addresses.iterator():
|
|
try:
|
|
prefix, was_created = ensure_prefix_for_ip(address)
|
|
created += int(was_created)
|
|
skipped += int(prefix is None)
|
|
except Exception:
|
|
failed += 1
|
|
|
|
if failed:
|
|
messages.warning(
|
|
request,
|
|
f"{created} Bereiche wurden angelegt; {failed} IP-Adressen konnten nicht verarbeitet werden.",
|
|
)
|
|
else:
|
|
messages.success(
|
|
request,
|
|
f"{created} fehlende Bereiche wurden angelegt. {skipped} Host-Adressen wurden ignoriert.",
|
|
)
|
|
return redirect("ipam:prefix_list")
|