1
0
mirror of https://github.com/netbox-community/netbox.git synced 2024-05-10 07:54:54 +00:00
2016-05-13 16:36:58 -04:00

484 lines
16 KiB
Python

from netaddr import IPSet
from django_tables2 import RequestConfig
from django.conf import settings
from django.contrib import messages
from django.contrib.auth.mixins import PermissionRequiredMixin
from django.shortcuts import get_object_or_404, render
from dcim.models import Device
from utilities.paginator import EnhancedPaginator
from utilities.views import BulkImportView, BulkEditView, BulkDeleteView, ObjectListView, ObjectEditView,\
ObjectDeleteView
from .filters import AggregateFilter, PrefixFilter, IPAddressFilter, VLANFilter, VRFFilter
from .forms import AggregateForm, AggregateImportForm, AggregateBulkEditForm, AggregateBulkDeleteForm,\
AggregateFilterForm, PrefixForm, PrefixImportForm, PrefixBulkEditForm, PrefixBulkDeleteForm, PrefixFilterForm,\
IPAddressForm, IPAddressImportForm, IPAddressBulkEditForm, IPAddressBulkDeleteForm, IPAddressFilterForm, VLANForm,\
VLANImportForm, VLANBulkEditForm, VLANBulkDeleteForm, VRFForm, VRFImportForm, VRFBulkEditForm, VRFBulkDeleteForm,\
VLANFilterForm
from .models import VRF, Aggregate, Prefix, IPAddress, VLAN
from .tables import AggregateTable, PrefixTable, PrefixBriefTable, IPAddressBriefTable, IPAddressTable, VLANTable,\
VRFTable
def add_available_prefixes(parent, prefix_list):
"""
Create fake Prefix objects for all unallocated space within a prefix.
"""
# Find all unallocated space
available_prefixes = IPSet(parent) ^ IPSet([p.prefix for p in prefix_list])
available_prefixes = [Prefix(prefix=p) for p in available_prefixes.iter_cidrs()]
# Concatenate and sort complete list of children
prefix_list = list(prefix_list) + available_prefixes
prefix_list.sort(key=lambda p: p.prefix)
return prefix_list
#
# VRFs
#
class VRFListView(ObjectListView):
queryset = VRF.objects.all()
filter = VRFFilter
table = VRFTable
edit_permissions = ['ipam.change_vrf', 'ipam.delete_vrf']
template_name = 'ipam/vrf_list.html'
def vrf(request, pk):
vrf = get_object_or_404(VRF.objects.all(), pk=pk)
prefixes = Prefix.objects.filter(vrf=vrf)
return render(request, 'ipam/vrf.html', {
'vrf': vrf,
'prefixes': prefixes,
})
class VRFEditView(PermissionRequiredMixin, ObjectEditView):
permission_required = 'ipam.change_vrf'
model = VRF
form_class = VRFForm
cancel_url = 'ipam:vrf_list'
class VRFDeleteView(PermissionRequiredMixin, ObjectDeleteView):
permission_required = 'ipam.delete_vrf'
model = VRF
redirect_url = 'ipam:vrf_list'
class VRFBulkImportView(PermissionRequiredMixin, BulkImportView):
permission_required = 'ipam.add_vrf'
form = VRFImportForm
table = VRFTable
template_name = 'ipam/vrf_import.html'
obj_list_url = 'ipam:vrf_list'
class VRFBulkEditView(PermissionRequiredMixin, BulkEditView):
permission_required = 'ipam.change_vrf'
cls = VRF
form = VRFBulkEditForm
template_name = 'ipam/vrf_bulk_edit.html'
default_redirect_url = 'ipam:vrf_list'
def update_objects(self, pk_list, form):
fields_to_update = {}
for field in ['description']:
if form.cleaned_data[field]:
fields_to_update[field] = form.cleaned_data[field]
updated_count = self.cls.objects.filter(pk__in=pk_list).update(**fields_to_update)
messages.success(self.request, "Updated {} VRFs".format(updated_count))
class VRFBulkDeleteView(PermissionRequiredMixin, BulkDeleteView):
permission_required = 'ipam.delete_vrf'
cls = VRF
form = VRFBulkDeleteForm
default_redirect_url = 'ipam:vrf_list'
#
# Aggregates
#
class AggregateListView(ObjectListView):
queryset = Aggregate.objects.select_related('rir').extra(select={
'child_count': 'SELECT COUNT(*) FROM ipam_prefix WHERE ipam_prefix.prefix <<= ipam_aggregate.prefix',
})
filter = AggregateFilter
filter_form = AggregateFilterForm
table = AggregateTable
edit_permissions = ['ipam.change_aggregate', 'ipam.delete_aggregate']
template_name = 'ipam/aggregate_list.html'
def aggregate(request, pk):
aggregate = get_object_or_404(Aggregate, pk=pk)
# Find all child prefixes contained by this aggregate
child_prefixes = Prefix.objects.filter(prefix__net_contained_or_equal=str(aggregate.prefix))\
.select_related('site', 'status', 'role').annotate_depth(limit=0)
child_prefixes = add_available_prefixes(aggregate.prefix, child_prefixes)
prefix_table = PrefixTable(child_prefixes)
if request.user.has_perm('ipam.change_prefix') or request.user.has_perm('ipam.delete_prefix'):
prefix_table.base_columns['pk'].visible = True
RequestConfig(request, paginate={'per_page': settings.PAGINATE_COUNT, 'klass': EnhancedPaginator})\
.configure(prefix_table)
return render(request, 'ipam/aggregate.html', {
'aggregate': aggregate,
'prefix_table': prefix_table,
})
class AggregateEditView(PermissionRequiredMixin, ObjectEditView):
permission_required = 'ipam.change_aggregate'
model = Aggregate
form_class = AggregateForm
cancel_url = 'ipam:aggregate_list'
class AggregateDeleteView(PermissionRequiredMixin, ObjectDeleteView):
permission_required = 'ipam.delete_aggregate'
model = Aggregate
redirect_url = 'ipam:aggregate_list'
class AggregateBulkImportView(PermissionRequiredMixin, BulkImportView):
permission_required = 'ipam.add_aggregate'
form = AggregateImportForm
table = AggregateTable
template_name = 'ipam/aggregate_import.html'
obj_list_url = 'ipam:aggregate_list'
class AggregateBulkEditView(PermissionRequiredMixin, BulkEditView):
permission_required = 'ipam.change_aggregate'
cls = Aggregate
form = AggregateBulkEditForm
template_name = 'ipam/aggregate_bulk_edit.html'
default_redirect_url = 'ipam:aggregate_list'
def update_objects(self, pk_list, form):
fields_to_update = {}
for field in ['rir', 'date_added', 'description']:
if form.cleaned_data[field]:
fields_to_update[field] = form.cleaned_data[field]
updated_count = self.cls.objects.filter(pk__in=pk_list).update(**fields_to_update)
messages.success(self.request, "Updated {} aggregates".format(updated_count))
class AggregateBulkDeleteView(PermissionRequiredMixin, BulkDeleteView):
permission_required = 'ipam.delete_aggregate'
cls = Aggregate
form = AggregateBulkDeleteForm
default_redirect_url = 'ipam:aggregate_list'
#
# Prefixes
#
class PrefixListView(ObjectListView):
queryset = Prefix.objects.select_related('site', 'status', 'role')
filter = PrefixFilter
filter_form = PrefixFilterForm
table = PrefixTable
edit_permissions = ['ipam.change_prefix', 'ipam.delete_prefix']
template_name = 'ipam/prefix_list.html'
def alter_queryset(self, request):
# Show only top-level prefixes by default (unless searching)
limit = None if request.GET.get('expand') or request.GET.get('q') else 0
return self.queryset.annotate_depth(limit=limit)
def prefix(request, pk):
prefix = get_object_or_404(Prefix.objects.select_related('site', 'vlan', 'status', 'role'), pk=pk)
try:
aggregate = Aggregate.objects.get(prefix__net_contains_or_equals=str(prefix.prefix))
except Aggregate.DoesNotExist:
aggregate = None
# Count child IP addresses
ipaddress_count = IPAddress.objects.filter(address__net_contained_or_equal=str(prefix.prefix)).count()
# Parent prefixes table
parent_prefixes = Prefix.objects.filter(vrf=prefix.vrf, prefix__net_contains=str(prefix.prefix))\
.select_related('site', 'status', 'role').annotate_depth()
parent_prefix_table = PrefixBriefTable(parent_prefixes)
# Duplicate prefixes table
duplicate_prefixes = Prefix.objects.filter(vrf=prefix.vrf, prefix=str(prefix.prefix)).exclude(pk=prefix.pk)\
.select_related('site', 'status', 'role')
duplicate_prefix_table = PrefixBriefTable(duplicate_prefixes)
# Child prefixes table
child_prefixes = Prefix.objects.filter(vrf=prefix.vrf, prefix__net_contained=str(prefix.prefix))\
.select_related('site', 'status', 'role').annotate_depth(limit=0)
if child_prefixes:
child_prefixes = add_available_prefixes(prefix.prefix, child_prefixes)
child_prefix_table = PrefixTable(child_prefixes)
if request.user.has_perm('ipam.change_prefix') or request.user.has_perm('ipam.delete_prefix'):
child_prefix_table.base_columns['pk'].visible = True
RequestConfig(request, paginate={'per_page': settings.PAGINATE_COUNT, 'klass': EnhancedPaginator})\
.configure(child_prefix_table)
return render(request, 'ipam/prefix.html', {
'prefix': prefix,
'aggregate': aggregate,
'ipaddress_count': ipaddress_count,
'parent_prefix_table': parent_prefix_table,
'child_prefix_table': child_prefix_table,
'duplicate_prefix_table': duplicate_prefix_table,
})
class PrefixEditView(PermissionRequiredMixin, ObjectEditView):
permission_required = 'ipam.change_prefix'
model = Prefix
form_class = PrefixForm
fields_initial = ['site', 'vrf', 'prefix']
cancel_url = 'ipam:prefix_list'
class PrefixDeleteView(PermissionRequiredMixin, ObjectDeleteView):
permission_required = 'ipam.delete_prefix'
model = Prefix
redirect_url = 'ipam:prefix_list'
class PrefixBulkImportView(PermissionRequiredMixin, BulkImportView):
permission_required = 'ipam.add_prefix'
form = PrefixImportForm
table = PrefixTable
template_name = 'ipam/prefix_import.html'
obj_list_url = 'ipam:prefix_list'
class PrefixBulkEditView(PermissionRequiredMixin, BulkEditView):
permission_required = 'ipam.change_prefix'
cls = Prefix
form = PrefixBulkEditForm
template_name = 'ipam/prefix_bulk_edit.html'
default_redirect_url = 'ipam:prefix_list'
def update_objects(self, pk_list, form):
fields_to_update = {}
if form.cleaned_data['vrf']:
fields_to_update['vrf'] = form.cleaned_data['vrf']
elif form.cleaned_data['vrf_global']:
fields_to_update['vrf'] = None
for field in ['site', 'status', 'role', 'description']:
if form.cleaned_data[field]:
fields_to_update[field] = form.cleaned_data[field]
updated_count = self.cls.objects.filter(pk__in=pk_list).update(**fields_to_update)
messages.success(self.request, "Updated {} prefixes".format(updated_count))
class PrefixBulkDeleteView(PermissionRequiredMixin, BulkDeleteView):
permission_required = 'ipam.delete_prefix'
cls = Prefix
form = PrefixBulkDeleteForm
default_redirect_url = 'ipam:prefix_list'
def prefix_ipaddresses(request, pk):
prefix = get_object_or_404(Prefix.objects.all(), pk=pk)
# Find all IPAddresses belonging to this Prefix
ipaddresses = IPAddress.objects.filter(address__net_contained_or_equal=str(prefix.prefix))\
.select_related('vrf', 'interface__device', 'primary_for')
ip_table = IPAddressTable(ipaddresses)
if request.user.has_perm('ipam.change_ipaddress') or request.user.has_perm('ipam.delete_ipaddress'):
ip_table.base_columns['pk'].visible = True
RequestConfig(request, paginate={'per_page': settings.PAGINATE_COUNT, 'klass': EnhancedPaginator})\
.configure(ip_table)
return render(request, 'ipam/prefix_ipaddresses.html', {
'prefix': prefix,
'ip_table': ip_table,
})
#
# IP addresses
#
class IPAddressListView(ObjectListView):
queryset = IPAddress.objects.select_related('vrf', 'interface__device', 'primary_for')
filter = IPAddressFilter
filter_form = IPAddressFilterForm
table = IPAddressTable
edit_permissions = ['ipam.change_ipaddress', 'ipam.delete_ipaddress']
template_name = 'ipam/ipaddress_list.html'
def ipaddress(request, pk):
ipaddress = get_object_or_404(IPAddress.objects.select_related('interface__device'), pk=pk)
parent_prefixes = Prefix.objects.filter(vrf=ipaddress.vrf, prefix__net_contains=str(ipaddress.address.ip))
related_ips = IPAddress.objects.select_related('interface__device').exclude(pk=ipaddress.pk).filter(vrf=ipaddress.vrf, address__net_contained_or_equal=str(ipaddress.address))
related_ips_table = IPAddressBriefTable(related_ips)
RequestConfig(request, paginate={'per_page': settings.PAGINATE_COUNT, 'klass': EnhancedPaginator}).configure(related_ips_table)
return render(request, 'ipam/ipaddress.html', {
'ipaddress': ipaddress,
'parent_prefixes': parent_prefixes,
'related_ips_table': related_ips_table,
})
class IPAddressEditView(PermissionRequiredMixin, ObjectEditView):
permission_required = 'ipam.change_ipaddress'
model = IPAddress
form_class = IPAddressForm
fields_initial = ['ipaddress']
template_name = 'ipam/ipaddress_edit.html'
cancel_url = 'ipam:ipaddress_list'
class IPAddressDeleteView(PermissionRequiredMixin, ObjectDeleteView):
permission_required = 'ipam.delete_ipaddress'
model = IPAddress
redirect_url = 'ipam:ipaddress_list'
class IPAddressBulkImportView(PermissionRequiredMixin, BulkImportView):
permission_required = 'ipam.add_ipaddress'
form = IPAddressImportForm
table = IPAddressTable
template_name = 'ipam/ipaddress_import.html'
obj_list_url = 'ipam:ipaddress_list'
def save_obj(self, obj):
obj.save()
# Update primary IP for device if needed
try:
device = obj.primary_for
device.primary_ip = obj
device.save()
except Device.DoesNotExist:
pass
class IPAddressBulkEditView(PermissionRequiredMixin, BulkEditView):
permission_required = 'ipam.change_ipaddress'
cls = IPAddress
form = IPAddressBulkEditForm
template_name = 'ipam/ipaddress_bulk_edit.html'
default_redirect_url = 'ipam:ipaddress_list'
def update_objects(self, pk_list, form):
fields_to_update = {}
if form.cleaned_data['vrf']:
fields_to_update['vrf'] = form.cleaned_data['vrf']
elif form.cleaned_data['vrf_global']:
fields_to_update['vrf'] = None
for field in ['description']:
if form.cleaned_data[field]:
fields_to_update[field] = form.cleaned_data[field]
updated_count = self.cls.objects.filter(pk__in=pk_list).update(**fields_to_update)
messages.success(self.request, "Updated {} IP addresses".format(updated_count))
class IPAddressBulkDeleteView(PermissionRequiredMixin, BulkDeleteView):
permission_required = 'ipam.delete_ipaddress'
cls = IPAddress
form = IPAddressBulkDeleteForm
default_redirect_url = 'ipam:ipaddress_list'
#
# VLANs
#
class VLANListView(ObjectListView):
queryset = VLAN.objects.select_related('site', 'status', 'role')
filter = VLANFilter
filter_form = VLANFilterForm
table = VLANTable
edit_permissions = ['ipam.change_vlan', 'ipam.delete_vlan']
template_name = 'ipam/vlan_list.html'
def vlan(request, pk):
vlan = get_object_or_404(VLAN.objects.select_related('site', 'status', 'role'), pk=pk)
prefixes = Prefix.objects.filter(vlan=vlan)
return render(request, 'ipam/vlan.html', {
'vlan': vlan,
'prefixes': prefixes,
})
class VLANEditView(PermissionRequiredMixin, ObjectEditView):
permission_required = 'ipam.change_vlan'
model = VLAN
form_class = VLANForm
cancel_url = 'ipam:vlan_list'
class VLANDeleteView(PermissionRequiredMixin, ObjectDeleteView):
permission_required = 'ipam.delete_vlan'
model = VLAN
redirect_url = 'ipam:vlan_list'
class VLANBulkImportView(PermissionRequiredMixin, BulkImportView):
permission_required = 'ipam.add_vlan'
form = VLANImportForm
table = VLANTable
template_name = 'ipam/vlan_import.html'
obj_list_url = 'ipam:vlan_list'
class VLANBulkEditView(PermissionRequiredMixin, BulkEditView):
permission_required = 'ipam.change_vlan'
cls = VLAN
form = VLANBulkEditForm
template_name = 'ipam/vlan_bulk_edit.html'
default_redirect_url = 'ipam:vlan_list'
def update_objects(self, pk_list, form):
fields_to_update = {}
for field in ['site', 'status', 'role']:
if form.cleaned_data[field]:
fields_to_update[field] = form.cleaned_data[field]
updated_count = self.cls.objects.filter(pk__in=pk_list).update(**fields_to_update)
messages.success(self.request, "Updated {} VLANs".format(updated_count))
class VLANBulkDeleteView(PermissionRequiredMixin, BulkDeleteView):
permission_required = 'ipam.delete_vlan'
cls = VLAN
form = VLANBulkDeleteForm
default_redirect_url = 'ipam:vlan_list'