Data Import Fixes (#13162)

- Improved validation of data import session
- Handle FieldError if incorrect filter field is provided
- Prevent model traversal via import session filtering
- Added regression tests

src/backend/InvenTree/importer/models.py

@@ -6,7 +6,7 @@ from datetime import datetime
 from typing import Optional
 
 from django.contrib.auth.models import User
-from django.core.exceptions import FieldDoesNotExist
+from django.core.exceptions import FieldDoesNotExist, FieldError
 from django.core.exceptions import ValidationError as DjangoValidationError
 from django.core.serializers.json import DjangoJSONEncoder
 from django.core.validators import FileExtensionValidator
@@ -56,6 +56,15 @@ class DataImportSession(models.Model):
         """Return the API URL associated with the DataImportSession model."""
         return reverse('api-importer-session-list')
 
+    def clean(self):
+        """Validate the DataImportSession object."""
+        super().clean()
+
+        try:
+            importer.validators.validate_field_filters(self.field_filters)
+        except DjangoValidationError as exc:
+            raise DjangoValidationError({'field_filters': exc.messages})
+
     def save(self, *args, **kwargs):
         """Save the DataImportSession object."""
         initial = self.pk is None
@@ -890,12 +899,21 @@ class DataImportRow(models.Model):
             else {}
         )
 
+        # Re-validate filters before applying them to the queryset
+        # (e.g. sessions created before filter validation was enforced)
+        importer.validators.validate_field_filters({field_name: base_filters})
+
         if lookup_field and type(lookup_field) is str:
             # A specific lookup field has been chosen by the user — query only that field
             try:
                 queryset = model.objects.filter(**{lookup_field: value}, **base_filters)
             except ValueError:
                 return value
+            except FieldError:
+                # Field filters reference a field which does not exist on the related model
+                raise DjangoValidationError(
+                    _('Invalid filters for field: {field}').format(field=field_name)
+                )
 
             results = list(queryset[:2])
 
@@ -918,6 +936,11 @@ class DataImportRow(models.Model):
                 queryset = model.objects.filter(**{id_field: value}, **base_filters)
             except ValueError:
                 continue
+            except FieldError:
+                # Field filters reference a field which does not exist on the related model
+                raise DjangoValidationError(
+                    _('Invalid filters for field: {field}').format(field=field_name)
+                )
 
             # Evaluate at most two results to determine if there is exactly one match
             results = list(queryset[:2])

src/backend/InvenTree/importer/serializers.py

@@ -9,6 +9,7 @@ from rest_framework import serializers
 
 import importer.models
 import importer.registry
+import importer.validators
 from InvenTree.serializers import (
     InvenTreeAttachmentSerializerField,
     InvenTreeModelSerializer,
@@ -147,6 +148,8 @@ class DataImportSessionSerializer(InvenTreeModelSerializer):
             if type(filters) is not dict:
                 raise ValidationError(_('Invalid field filters'))
 
+        importer.validators.validate_field_filters(filters)
+
         return filters
 
     def create(self, validated_data):

src/backend/InvenTree/importer/tests.py

@@ -439,6 +439,97 @@ class ImporterTest(ImporterMixin, InvenTreeTestCase):
 
         self.assertEqual(row.errors, {'part': 'No related model found for field: part'})
 
+    def test_field_filters_validation(self):
+        """Test that field filters cannot be used to traverse related models.
+
+        Field filters are applied as ORM filters against the related model,
+        so lookups which traverse relations (or inject Q arguments) must be rejected.
+        """
+        from django.core.exceptions import ValidationError as DjangoValidationError
+
+        from importer.validators import validate_field_filters
+
+        # Valid filters
+        for filters in [
+            None,
+            {},
+            {'part': {}},
+            {'part': {'supplier': 1, 'active': True}},
+            {'part': {'name': 'abc', 'pk': 1.5, 'category': None}},
+        ]:
+            validate_field_filters(filters)
+
+        # Invalid filters
+        for filters in [
+            [],
+            {'user': []},
+            {'user': {'usersession__session_key__startswith': 'a'}},
+            {'user': {'password__startswith': 'pbkdf2'}},
+            {'part': {'name__icontains': 'abc'}},
+            {'part': {'_negated': True}},
+            {'part': {'_connector': 'OR'}},
+            {'part': {'pk': 1, 'active': True, '': 1}},
+            {'part': {'not a field': 1}},
+            {'part': {'active': [True]}},
+            {'part': {'active': {'a': 1}}},
+        ]:
+            with self.assertRaises(DjangoValidationError, msg=str(filters)):
+                validate_field_filters(filters)
+
+        # Invalid filters are rejected on save
+        session = DataImportSession(
+            data_file=self.helper_file('companies.csv'),
+            model_type='stockitem',
+            field_filters={'part': {'category__parent__name': 'abc'}},
+        )
+
+        with self.assertRaises(DjangoValidationError):
+            session.save()
+
+        # Filters stored before validation was enforced are rejected at lookup time
+        session.field_filters = {}
+        session.save()
+
+        DataImportSession.objects.filter(pk=session.pk).update(
+            field_filters={'part': {'category__parent__name': 'abc'}}
+        )
+        session.refresh_from_db()
+
+        row = DataImportRow(session=session, row_data={})
+        row.related_field_map = {}
+
+        with self.assertRaises(DjangoValidationError):
+            row.lookup_related_field('part', '1')
+
+    def test_field_filters_invalid_field(self):
+        """Test that field filters referencing a non-existent field do not crash the import.
+
+        Regression test: an unknown filter key raises a FieldError from the ORM,
+        which was not caught by lookup_related_field (only ValueError was handled).
+        """
+        from django.core.exceptions import ValidationError as DjangoValidationError
+
+        session = DataImportSession.objects.create(
+            data_file=self.helper_file('companies.csv'),
+            model_type='stockitem',
+            field_filters={'part': {'not_a_field': 1}},
+        )
+
+        row = DataImportRow(session=session, row_data={'Website': '1'})
+
+        # Auto-lookup: error is recorded against the row, rather than raising FieldError
+        row.extract_data(
+            field_mapping={'part': 'Website'},
+            available_fields={'part': {'type': 'related field'}},
+            commit=False,
+        )
+
+        self.assertEqual(row.errors, {'part': 'Invalid filters for field: part'})
+
+        # Explicit lookup field
+        with self.assertRaises(DjangoValidationError):
+            row.lookup_related_field('part', '1', lookup_field='name')
+
     def test_lookup_field_validation(self):
         """Test that DataImportColumnMap.clean() validates the lookup_field value."""
         from django.core.exceptions import ValidationError as DjangoValidationError
@@ -774,6 +865,53 @@ class ImportAPITest(ImporterMixin, InvenTreeAPITestCase):
         self.assignRole('purchase_order.change')
         self.post(url, expected_code=200)
 
+    def test_field_filters_api(self):
+        """Test that the API rejects field filters which traverse related models."""
+        import json
+
+        url = reverse('api-importer-session-list')
+
+        for filters in [
+            {'user': {'usersession__session_key__startswith': 'a'}},
+            {'user': {'_negated': True}},
+        ]:
+            response = self.post(
+                url,
+                {
+                    'model_type': 'stockitemtestresult',
+                    'data_file': self.helper_file('companies.csv'),
+                    'field_filters': json.dumps(filters),
+                },
+                format='multipart',
+                expected_code=400,
+            )
+            self.assertIn('field_filters', response.data)
+
+        # Simple equality filters are accepted
+        data = self.post(
+            url,
+            {
+                'model_type': 'purchaseorderlineitem',
+                'data_file': self.helper_file('companies.csv'),
+                'field_filters': json.dumps({'part': {'supplier': 1, 'active': True}}),
+            },
+            format='multipart',
+            expected_code=201,
+        ).data
+
+        self.assertEqual(
+            data['field_filters'], {'part': {'supplier': 1, 'active': True}}
+        )
+
+        # Filters cannot be modified to traverse related models after creation
+        self.assignRole('purchase_order.change')
+
+        self.patch(
+            reverse('api-import-session-detail', kwargs={'pk': data['pk']}),
+            {'field_filters': {'part': {'supplier__name': 'abc'}}},
+            expected_code=400,
+        )
+
     def test_model_type_immutable(self):
         """Test that a session's model_type cannot be changed after creation.
 

src/backend/InvenTree/importer/validators.py

@@ -54,3 +54,54 @@ def validate_field_defaults(value):
 
         if type(value) is not dict:
             raise ValidationError(_('Value must be a valid dictionary object'))
+
+
+def validate_field_filters(value):
+    """Validate the provided field filters.
+
+    Field filters are applied as ORM filters against the related model for each field,
+    so they are restricted to simple equality lookups against fields on that model:
+
+    - Must be a dict of {field_name: {filter_key: filter_value}}
+    - Filter keys cannot contain '__' (prevents traversal of related models and lookup expressions)
+    - Filter keys cannot start with '_' (prevents injection of Q object arguments e.g. '_negated')
+    - Filter values must be simple scalar values
+    """
+    if value is None:
+        return
+
+    if type(value) is not dict:
+        raise ValidationError(_('Value must be a valid dictionary object'))
+
+    for field_name, filters in value.items():
+        if type(filters) is not dict:
+            raise ValidationError(
+                _('Filters for field {field} must be a valid dictionary object').format(
+                    field=field_name
+                )
+            )
+
+        for key, filter_value in filters.items():
+            if (
+                type(key) is not str
+                or not key.isidentifier()
+                or key.startswith('_')
+                or '__' in key
+            ):
+                raise ValidationError(
+                    _('Invalid filter key for field {field}: {key}').format(
+                        field=field_name, key=key
+                    )
+                )
+
+            if filter_value is not None and type(filter_value) not in [
+                str,
+                int,
+                float,
+                bool,
+            ]:
+                raise ValidationError(
+                    _('Invalid filter value for field {field}: {key}').format(
+                        field=field_name, key=key
+                    )
+                )