diff --git a/netbox/ipam/forms/bulk_create.py b/netbox/ipam/forms/bulk_create.py index 856476786..763dbac9c 100644 --- a/netbox/ipam/forms/bulk_create.py +++ b/netbox/ipam/forms/bulk_create.py @@ -1,14 +1,17 @@ from django import forms from django.utils.translation import gettext_lazy as _ -from utilities.forms.fields import ExpandableIPAddressField +from utilities.forms.fields import ExpandableIPNetworkField __all__ = ( - 'IPAddressBulkCreateForm', + 'IPNetworkBulkCreateForm', ) -class IPAddressBulkCreateForm(forms.Form): - pattern = ExpandableIPAddressField( - label=_('Address pattern') +class IPNetworkBulkCreateForm(forms.Form): + """ + Pattern form for bulk-creating IP-based objects (addresses, prefixes). + """ + pattern = ExpandableIPNetworkField( + label=_('Pattern') ) diff --git a/netbox/ipam/forms/model_forms.py b/netbox/ipam/forms/model_forms.py index 3748f2901..d3159320d 100644 --- a/netbox/ipam/forms/model_forms.py +++ b/netbox/ipam/forms/model_forms.py @@ -37,6 +37,7 @@ __all__ = ( 'IPAddressBulkAddForm', 'IPAddressForm', 'IPRangeForm', + 'PrefixBulkAddForm', 'PrefixForm', 'RIRForm', 'RoleForm', @@ -249,6 +250,32 @@ class PrefixForm(TenancyForm, ScopedForm, PrimaryModelForm): self.fields['vlan'].widget.attrs.pop('data-dynamic-params', None) +class PrefixBulkAddForm(TenancyForm, NetBoxModelForm): + vrf = DynamicModelChoiceField( + queryset=VRF.objects.all(), + required=False, + label=_('VRF') + ) + role = DynamicModelChoiceField( + label=_('Role'), + queryset=Role.objects.all(), + required=False, + quick_add=True + ) + + fieldsets = ( + FieldSet('status', 'role', 'vrf', 'is_pool', 'mark_utilized', 'description', 'tags', name=_('Prefix')), + FieldSet('tenant_group', 'tenant', name=_('Tenancy')), + ) + + class Meta: + model = Prefix + fields = [ + 'prefix', 'vrf', 'status', 'role', 'is_pool', 'mark_utilized', 'description', 'tenant_group', 'tenant', + 'tags', + ] + + class IPRangeForm(TenancyForm, PrimaryModelForm): vrf = DynamicModelChoiceField( queryset=VRF.objects.all(), @@ -472,6 +499,11 @@ class IPAddressBulkAddForm(TenancyForm, NetBoxModelForm): label=_('VRF') ) + fieldsets = ( + FieldSet('status', 'role', 'vrf', 'dns_name', 'description', 'tags', name=_('IP Address')), + FieldSet('tenant_group', 'tenant', name=_('Tenancy')), + ) + class Meta: model = IPAddress fields = [ diff --git a/netbox/ipam/tests/test_views.py b/netbox/ipam/tests/test_views.py index 5154397bb..04a88ace4 100644 --- a/netbox/ipam/tests/test_views.py +++ b/netbox/ipam/tests/test_views.py @@ -467,6 +467,74 @@ class PrefixTestCase(ViewTestCases.PrimaryObjectViewTestCase): 'description': 'New description', } + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_bulk_add_ipv4_prefixes(self): + """Test bulk creating IPv4 prefixes using a pattern.""" + obj_perm = ObjectPermission(name='Test permission', actions=['add']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix)) + + initial_count = Prefix.objects.count() + url = reverse('ipam:prefix_bulk_add') + data = { + 'pattern': '10.0.[0-2].0/24', + 'status': PrefixStatusChoices.STATUS_ACTIVE, + } + response = self.client.post(url, data) + self.assertHttpStatus(response, 302) + self.assertEqual(Prefix.objects.count(), initial_count + 3) + + for i in range(3): + self.assertTrue(Prefix.objects.filter(prefix=IPNetwork(f'10.0.{i}.0/24')).exists()) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_bulk_add_ipv6_prefixes(self): + """Test bulk creating IPv6 prefixes using a pattern.""" + obj_perm = ObjectPermission(name='Test permission', actions=['add']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix)) + + initial_count = Prefix.objects.count() + url = reverse('ipam:prefix_bulk_add') + data = { + 'pattern': 'fd00:db8:[0-3]::/48', + 'status': PrefixStatusChoices.STATUS_ACTIVE, + } + response = self.client.post(url, data) + self.assertHttpStatus(response, 302) + self.assertEqual(Prefix.objects.count(), initial_count + 4) + + for i in range(4): + self.assertTrue(Prefix.objects.filter(prefix=IPNetwork(f'fd00:db8:{i}::/48')).exists()) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_bulk_add_ipv6_prefixes_uppercase_hex(self): + """Test bulk creating IPv6 prefixes using uppercase hex in the pattern.""" + obj_perm = ObjectPermission(name='Test permission', actions=['add']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ObjectType.objects.get_for_model(Prefix)) + + initial_count = Prefix.objects.count() + url = reverse('ipam:prefix_bulk_add') + data = { + 'pattern': 'fd00:0:0:[48-4F]00::/56', + 'status': PrefixStatusChoices.STATUS_ACTIVE, + } + response = self.client.post(url, data) + self.assertHttpStatus(response, 302) + self.assertEqual(Prefix.objects.count(), initial_count + 8) + + expected_hex = ['48', '49', '4a', '4b', '4c', '4d', '4e', '4f'] + for h in expected_hex: + prefix_str = f'fd00:0:0:{h}00::/56' + self.assertTrue( + Prefix.objects.filter(prefix=IPNetwork(prefix_str)).exists(), + f'Expected prefix {prefix_str} was not created' + ) + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_prefix_prefixes(self): prefixes = ( diff --git a/netbox/ipam/views.py b/netbox/ipam/views.py index 6ceb0c53a..77e1daa8c 100644 --- a/netbox/ipam/views.py +++ b/netbox/ipam/views.py @@ -714,6 +714,7 @@ class PrefixIPAddressesView(generic.ObjectChildrenView): class PrefixEditView(generic.ObjectEditView): queryset = Prefix.objects.all() form = forms.PrefixForm + template_name = 'ipam/prefix_edit.html' @register_model_view(Prefix, 'delete') @@ -721,6 +722,15 @@ class PrefixDeleteView(generic.ObjectDeleteView): queryset = Prefix.objects.all() +@register_model_view(Prefix, 'bulk_add', path='bulk-add', detail=False) +class PrefixBulkCreateView(generic.BulkCreateView): + queryset = Prefix.objects.all() + form = forms.IPNetworkBulkCreateForm + model_form = forms.PrefixBulkAddForm + pattern_target = 'prefix' + template_name = 'ipam/prefix_bulk_add.html' + + @register_model_view(Prefix, 'bulk_import', path='import', detail=False) class PrefixBulkImportView(generic.BulkImportView): queryset = Prefix.objects.all() @@ -979,7 +989,7 @@ class IPAddressDeleteView(generic.ObjectDeleteView): @register_model_view(IPAddress, 'bulk_add', path='bulk-add', detail=False) class IPAddressBulkCreateView(generic.BulkCreateView): queryset = IPAddress.objects.all() - form = forms.IPAddressBulkCreateForm + form = forms.IPNetworkBulkCreateForm model_form = forms.IPAddressBulkAddForm pattern_target = 'address' template_name = 'ipam/ipaddress_bulk_add.html' diff --git a/netbox/netbox/views/generic/bulk_views.py b/netbox/netbox/views/generic/bulk_views.py index 2522a079a..ec18412a4 100644 --- a/netbox/netbox/views/generic/bulk_views.py +++ b/netbox/netbox/views/generic/bulk_views.py @@ -254,6 +254,18 @@ class BulkCreateView(GetReturnURLMixin, BaseMultiObjectView): return new_objects + def _get_context(self, request, form, model_form): + model = self.queryset.model + return { + 'obj_type': model._meta.verbose_name, + 'obj_type_plural': model._meta.verbose_name_plural, + 'form': form, + 'model_form': model_form, + 'return_url': self.get_return_url(request), + 'add_url': get_action_url(model, 'add'), + **self.get_extra_context(request), + } + # # Request handlers # @@ -268,13 +280,7 @@ class BulkCreateView(GetReturnURLMixin, BaseMultiObjectView): form = self.form() model_form = self.model_form(initial=initial) - return render(request, self.template_name, { - 'obj_type': self.model_form._meta.model._meta.verbose_name, - 'form': form, - 'model_form': model_form, - 'return_url': self.get_return_url(request), - **self.get_extra_context(request), - }) + return render(request, self.template_name, self._get_context(request, form, model_form)) def post(self, request): logger = logging.getLogger('netbox.views.BulkCreateView') @@ -313,13 +319,7 @@ class BulkCreateView(GetReturnURLMixin, BaseMultiObjectView): else: logger.debug("Form validation failed") - return render(request, self.template_name, { - 'form': form, - 'model_form': model_form, - 'obj_type': model._meta.verbose_name, - 'return_url': self.get_return_url(request), - **self.get_extra_context(request), - }) + return render(request, self.template_name, self._get_context(request, form, model_form)) class BulkImportView(GetReturnURLMixin, BaseMultiObjectView): diff --git a/netbox/templates/generic/bulk_add.html b/netbox/templates/generic/bulk_add.html new file mode 100644 index 000000000..ebc7bc0be --- /dev/null +++ b/netbox/templates/generic/bulk_add.html @@ -0,0 +1,49 @@ +{% extends 'generic/object_edit.html' %} +{% load helpers %} +{% load form_helpers %} +{% load i18n %} + +{% block title %}{% blocktrans trimmed with object_type_plural=obj_type_plural %}Bulk Add {{ object_type_plural }}{% endblocktrans %}{% endblock %} + +{% block tabs %} +
+{% endblock %} + +{% block form %} +192.0.2.[1,5,100-254]/24')
+ self.help_text = _(
+ 'Use bracket notation to specify numeric ranges for bulk creation (CIDR required).192.0.2.[1-10]/32, 10.[0-3,10-13].0.0/16, '
+ '2001:db8:[a-f]::/64'
+ )
def to_python(self, value):
- # Hackish address family detection but it's all we have to work with
- if '.' in value and re.search(IP4_EXPANSION_PATTERN, value):
- return list(expand_ipaddress_pattern(value, 4))
- if ':' in value and re.search(IP6_EXPANSION_PATTERN, value):
- return list(expand_ipaddress_pattern(value, 6))
+ if not value:
+ return [value]
+
+ # Replace expansion brackets with a neutral value to get a parseable IP/CIDR
+ stripped = re.sub(r'\[[^\]]+\]', '0', value)
+ try:
+ family = netaddr.IPNetwork(stripped).version
+ except (netaddr.AddrFormatError, ValueError):
+ return [value]
+
+ if family == 4 and re.search(IP4_EXPANSION_PATTERN, value):
+ return list(expand_ipnetwork_pattern(value, 4))
+ if family == 6 and re.search(IP6_EXPANSION_PATTERN, value):
+ return list(expand_ipnetwork_pattern(value.lower(), 6))
return [value]
diff --git a/netbox/utilities/forms/utils.py b/netbox/utilities/forms/utils.py
index 1ed13cb9c..5e222c2bb 100644
--- a/netbox/utilities/forms/utils.py
+++ b/netbox/utilities/forms/utils.py
@@ -12,7 +12,7 @@ from .constants import *
__all__ = (
'add_blank_choice',
'expand_alphanumeric_pattern',
- 'expand_ipaddress_pattern',
+ 'expand_ipnetwork_pattern',
'form_from_model',
'get_field_value',
'get_selected_values',
@@ -106,9 +106,9 @@ def expand_alphanumeric_pattern(string):
yield "{}{}{}".format(lead, i, remnant)
-def expand_ipaddress_pattern(string, family):
+def expand_ipnetwork_pattern(string, family):
"""
- Expand an IP address pattern into a list of strings. Examples:
+ Expand an IP network pattern into a list of strings. Examples:
'192.0.2.[1,2,100-250]/24' => ['192.0.2.1/24', '192.0.2.2/24', '192.0.2.100/24' ... '192.0.2.250/24']
'2001:db8:0:[0,fd-ff]::/64' => ['2001:db8:0:0::/64', '2001:db8:0:fd::/64', ... '2001:db8:0:ff::/64']
"""
@@ -124,7 +124,7 @@ def expand_ipaddress_pattern(string, family):
parsed_range = parse_numeric_range(pattern, base)
for i in parsed_range:
if re.search(regex, remnant):
- for string in expand_ipaddress_pattern(remnant, family):
+ for string in expand_ipnetwork_pattern(remnant, family):
yield ''.join([lead, format(i, 'x' if family == 6 else 'd'), string])
else:
yield ''.join([lead, format(i, 'x' if family == 6 else 'd'), remnant])
diff --git a/netbox/utilities/tests/test_forms.py b/netbox/utilities/tests/test_forms.py
index 2224cc195..c8a0c46fa 100644
--- a/netbox/utilities/tests/test_forms.py
+++ b/netbox/utilities/tests/test_forms.py
@@ -6,13 +6,13 @@ from netbox.choices import ImportFormatChoices
from utilities.forms.bulk_import import BulkImportForm
from utilities.forms.fields.csv import CSVSelectWidget
from utilities.forms.forms import BulkRenameForm
-from utilities.forms.utils import expand_alphanumeric_pattern, expand_ipaddress_pattern, get_field_value
+from utilities.forms.utils import expand_alphanumeric_pattern, expand_ipnetwork_pattern, get_field_value
from utilities.forms.widgets.select import AvailableOptions, SelectedOptions
-class ExpandIPAddress(TestCase):
+class ExpandIPNetwork(TestCase):
"""
- Validate the operation of expand_ipaddress_pattern().
+ Validate the operation of expand_ipnetwork_pattern().
"""
def test_ipv4_range(self):
input = '1.2.3.[9-10]/32'
@@ -21,7 +21,7 @@ class ExpandIPAddress(TestCase):
'1.2.3.10/32',
])
- self.assertEqual(sorted(expand_ipaddress_pattern(input, 4)), output)
+ self.assertEqual(sorted(expand_ipnetwork_pattern(input, 4)), output)
def test_ipv4_set(self):
input = '1.2.3.[4,44]/32'
@@ -30,7 +30,7 @@ class ExpandIPAddress(TestCase):
'1.2.3.44/32',
])
- self.assertEqual(sorted(expand_ipaddress_pattern(input, 4)), output)
+ self.assertEqual(sorted(expand_ipnetwork_pattern(input, 4)), output)
def test_ipv4_multiple_ranges(self):
input = '1.[9-10].3.[9-11]/32'
@@ -43,7 +43,7 @@ class ExpandIPAddress(TestCase):
'1.10.3.11/32',
])
- self.assertEqual(sorted(expand_ipaddress_pattern(input, 4)), output)
+ self.assertEqual(sorted(expand_ipnetwork_pattern(input, 4)), output)
def test_ipv4_multiple_sets(self):
input = '1.[2,22].3.[4,44]/32'
@@ -54,7 +54,7 @@ class ExpandIPAddress(TestCase):
'1.22.3.44/32',
])
- self.assertEqual(sorted(expand_ipaddress_pattern(input, 4)), output)
+ self.assertEqual(sorted(expand_ipnetwork_pattern(input, 4)), output)
def test_ipv4_set_and_range(self):
input = '1.[2,22].3.[9-11]/32'
@@ -67,7 +67,7 @@ class ExpandIPAddress(TestCase):
'1.22.3.11/32',
])
- self.assertEqual(sorted(expand_ipaddress_pattern(input, 4)), output)
+ self.assertEqual(sorted(expand_ipnetwork_pattern(input, 4)), output)
def test_ipv6_range(self):
input = 'fec::abcd:[9-b]/64'
@@ -77,7 +77,7 @@ class ExpandIPAddress(TestCase):
'fec::abcd:b/64',
])
- self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
+ self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_ipv6_range_multichar_field(self):
input = 'fec::abcd:[f-11]/64'
@@ -87,7 +87,7 @@ class ExpandIPAddress(TestCase):
'fec::abcd:11/64',
])
- self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
+ self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_ipv6_set(self):
input = 'fec::abcd:[9,ab]/64'
@@ -96,7 +96,7 @@ class ExpandIPAddress(TestCase):
'fec::abcd:ab/64',
])
- self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
+ self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_ipv6_multiple_ranges(self):
input = 'fec::[1-2]bcd:[9-b]/64'
@@ -109,7 +109,7 @@ class ExpandIPAddress(TestCase):
'fec::2bcd:b/64',
])
- self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
+ self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_ipv6_multiple_sets(self):
input = 'fec::[a,f]bcd:[9,ab]/64'
@@ -120,7 +120,7 @@ class ExpandIPAddress(TestCase):
'fec::fbcd:ab/64',
])
- self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
+ self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_ipv6_set_and_range(self):
input = 'fec::[dead,beaf]:[9-b]/64'
@@ -133,41 +133,41 @@ class ExpandIPAddress(TestCase):
'fec::beaf:b/64',
])
- self.assertEqual(sorted(expand_ipaddress_pattern(input, 6)), output)
+ self.assertEqual(sorted(expand_ipnetwork_pattern(input, 6)), output)
def test_invalid_address_family(self):
with self.assertRaisesRegex(Exception, 'Invalid IP address family: 5'):
- sorted(expand_ipaddress_pattern(None, 5))
+ sorted(expand_ipnetwork_pattern(None, 5))
def test_invalid_non_pattern(self):
with self.assertRaises(ValueError):
- sorted(expand_ipaddress_pattern('1.2.3.4/32', 4))
+ sorted(expand_ipnetwork_pattern('1.2.3.4/32', 4))
def test_invalid_range(self):
with self.assertRaises(ValueError):
- sorted(expand_ipaddress_pattern('1.2.3.[4-]/32', 4))
+ sorted(expand_ipnetwork_pattern('1.2.3.[4-]/32', 4))
with self.assertRaises(ValueError):
- sorted(expand_ipaddress_pattern('1.2.3.[-4]/32', 4))
+ sorted(expand_ipnetwork_pattern('1.2.3.[-4]/32', 4))
with self.assertRaises(ValueError):
- sorted(expand_ipaddress_pattern('1.2.3.[4--5]/32', 4))
+ sorted(expand_ipnetwork_pattern('1.2.3.[4--5]/32', 4))
def test_invalid_range_bounds(self):
- self.assertEqual(sorted(expand_ipaddress_pattern('1.2.3.[4-3]/32', 6)), [])
+ self.assertEqual(sorted(expand_ipnetwork_pattern('1.2.3.[4-3]/32', 6)), [])
def test_invalid_set(self):
with self.assertRaises(ValueError):
- sorted(expand_ipaddress_pattern('1.2.3.[4]/32', 4))
+ sorted(expand_ipnetwork_pattern('1.2.3.[4]/32', 4))
with self.assertRaises(ValueError):
- sorted(expand_ipaddress_pattern('1.2.3.[4,]/32', 4))
+ sorted(expand_ipnetwork_pattern('1.2.3.[4,]/32', 4))
with self.assertRaises(ValueError):
- sorted(expand_ipaddress_pattern('1.2.3.[,4]/32', 4))
+ sorted(expand_ipnetwork_pattern('1.2.3.[,4]/32', 4))
with self.assertRaises(ValueError):
- sorted(expand_ipaddress_pattern('1.2.3.[4,,5]/32', 4))
+ sorted(expand_ipnetwork_pattern('1.2.3.[4,,5]/32', 4))
class ExpandAlphanumeric(TestCase):