From 2629cde722de786b52781b798fac25e042a39d9e Mon Sep 17 00:00:00 2001 From: Oliver Date: Mon, 2 Nov 2026 20:41:47 +1100 Subject: [PATCH] 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 | 25 +++- src/backend/InvenTree/importer/serializers.py | 3 + src/backend/InvenTree/importer/tests.py | 138 ++++++++++++++++++ src/backend/InvenTree/importer/validators.py | 51 +++++++ 4 files changed, 216 insertions(+), 1 deletion(-) diff --git a/src/backend/InvenTree/importer/models.py b/src/backend/InvenTree/importer/models.py index 157fd72128..4cd73f9956 100644 --- a/src/backend/InvenTree/importer/models.py +++ b/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]) diff --git a/src/backend/InvenTree/importer/serializers.py b/src/backend/InvenTree/importer/serializers.py index f297e46cb9..6393541f0e 100644 --- a/src/backend/InvenTree/importer/serializers.py +++ b/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): diff --git a/src/backend/InvenTree/importer/tests.py b/src/backend/InvenTree/importer/tests.py index f9a4dc271b..d80067d9cc 100644 --- a/src/backend/InvenTree/importer/tests.py +++ b/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. diff --git a/src/backend/InvenTree/importer/validators.py b/src/backend/InvenTree/importer/validators.py index 5f81ff6dd5..db16e978af 100644 --- a/src/backend/InvenTree/importer/validators.py +++ b/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 + ) + )