Fixes #11617: Check for invalid CSV headers during bulk import

netbox/netbox/tests/test_import.py

@@ -17,6 +17,36 @@ def setUpTestData(cls):
     def _get_csv_data(self, csv_data):
         return '\n'.join(csv_data)
 
+    def test_invalid_headers(self):
+        """
+        Test that import form validation fails when an unknown CSV header is present.
+        """
+        self.add_permissions('dcim.add_region')
+
+        csv_data = [
+            'name,slug,INVALIDHEADER',
+            'Region 1,region-1,abc',
+            'Region 2,region-2,def',
+            'Region 3,region-3,ghi',
+        ]
+        data = {
+            'format': ImportFormatChoices.CSV,
+            'data': self._get_csv_data(csv_data),
+            'csv_delimiter': CSVDelimiterChoices.AUTO,
+        }
+
+        # Form validation should fail with invalid header present
+        self.assertHttpStatus(self.client.post(self._get_url('import'), data), 200)
+        self.assertEqual(Region.objects.count(), 0)
+
+        # Correct the CSV header name
+        csv_data[0] = 'name,slug,description'
+        data['data'] = self._get_csv_data(csv_data)
+
+        # Validation should succeed
+        self.assertHttpStatus(self.client.post(self._get_url('import'), data), 302)
+        self.assertEqual(Region.objects.count(), 3)
+
     @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
     def test_valid_tags(self):
         csv_data = (

netbox/utilities/forms/bulk_import.py

@@ -129,6 +129,7 @@ def _clean_csv(self, data, delimiter=CSVDelimiterChoices.AUTO):
         headers, records = parse_csv(reader)
 
         # Set CSV headers for reference by the model form
+        headers.pop('id', None)
         self._csv_headers = headers
 
         return records

netbox/utilities/forms/forms.py

@@ -70,22 +70,24 @@ class CSVModelForm(forms.ModelForm):
     """
     ModelForm used for the import of objects in CSV format.
     """
-    def __init__(self, *args, headers=None, fields=None, **kwargs):
-        headers = headers or {}
-        fields = fields or []
+    def __init__(self, *args, headers=None, **kwargs):
+        self.headers = headers or {}
         super().__init__(*args, **kwargs)
 
         # Modify the model form to accommodate any customized to_field_name properties
-        for field, to_field in headers.items():
+        for field, to_field in self.headers.items():
             if to_field is not None:
                 self.fields[field].to_field_name = to_field
 
-        # Omit any fields not specified (e.g. because the form is being used to
-        # updated rather than create objects)
-        if fields:
-            for field in list(self.fields.keys()):
-                if field not in fields:
-                    del self.fields[field]
+    def clean(self):
+        # Flag any invalid CSV headers
+        for header in self.headers:
+            if header not in self.fields:
+                raise forms.ValidationError(
+                    _("Unrecognized header: {name}").format(name=header)
+                )
+
+        return super().clean()
 
 
 class FilterForm(BootstrapMixin, forms.Form):