diff --git a/src/backend/InvenTree/InvenTree/conversion.py b/src/backend/InvenTree/InvenTree/conversion.py index 060bea5b71..d677ee06b4 100644 --- a/src/backend/InvenTree/InvenTree/conversion.py +++ b/src/backend/InvenTree/InvenTree/conversion.py @@ -18,6 +18,9 @@ _UNIT_REG_CACHE_KEY = 'unit_registry_hash' _unit_registry = None _unit_registry_hash: str = '' +# Registry of built-in units only (never modified after construction) +_base_unit_registry = None + # Relative tolerance used when comparing converted (floating point) numeric values NUMERIC_RELATIVE_TOLERANCE = 1e-9 @@ -119,6 +122,27 @@ def new_base_registry() -> pint.UnitRegistry: return reg +def is_base_unit_name(name: str) -> bool: + """Return True if the provided name resolves to a built-in (non-custom) unit. + + This includes prefixed and plural forms (e.g. 'kpound', 'meters'), + as pint would resolve these to the existing unit definition. + """ + global _base_unit_registry + + if not name: + return False + + if _base_unit_registry is None: + _base_unit_registry = new_base_registry() + + # Note: 'name in registry' cannot be used here, as it evaluates expressions (e.g. 'm/A') + try: + return len(_base_unit_registry.parse_unit_name(name)) > 0 + except Exception: + return False + + def reload_unit_registry(): """Reload the unit registry from the database. diff --git a/src/backend/InvenTree/InvenTree/permissions.py b/src/backend/InvenTree/InvenTree/permissions.py index 807117c21d..68196292a1 100644 --- a/src/backend/InvenTree/InvenTree/permissions.py +++ b/src/backend/InvenTree/InvenTree/permissions.py @@ -463,10 +463,7 @@ class DataImporterPermission(OASTokenMixin, permissions.BasePermission): session = getattr(obj, 'session', None) if session: - if model_class := session.model_class: - return users.permissions.check_user_permission( - request.user, model_class, 'change' - ) + return session.check_import_permission(request.user) return True diff --git a/src/backend/InvenTree/common/models.py b/src/backend/InvenTree/common/models.py index ebb0f0d206..a2f930f5ba 100644 --- a/src/backend/InvenTree/common/models.py +++ b/src/backend/InvenTree/common/models.py @@ -158,6 +158,9 @@ class ProjectCode( IMPORT_ID_FIELDS = ['code'] + # API write access is restricted to staff users + IMPORT_REQUIRES_STAFF = True + class Meta: """Class options for the ProjectCode model.""" @@ -1880,6 +1883,9 @@ class CustomUnit(InvenTree.models.InvenTreeAuditMixin, models.Model): https://pint.readthedocs.io/en/stable/advanced/defining.html """ + # API write access is restricted to staff users + IMPORT_REQUIRES_STAFF = True + class Meta: """Class meta options.""" @@ -1914,6 +1920,7 @@ class CustomUnit(InvenTree.models.InvenTreeAuditMixin, models.Model): from InvenTree.conversion import ( build_candidate_unit_registry, get_unit_registry, + is_base_unit_name, ) registry = get_unit_registry() @@ -1925,6 +1932,19 @@ class CustomUnit(InvenTree.models.InvenTreeAuditMixin, models.Model): if not self.name.isidentifier(): raise ValidationError({'name': _('Unit name must be a valid identifier')}) + # Custom units cannot override built-in unit definitions + if is_base_unit_name(self.name): + raise ValidationError({ + 'name': _('Unit name conflicts with an existing built-in unit') + }) + + self.symbol = self.symbol.strip() + + if is_base_unit_name(self.symbol): + raise ValidationError({ + 'symbol': _('Unit symbol conflicts with an existing built-in unit') + }) + self.definition = self.definition.strip() # Check that the 'definition' is valid, by itself @@ -2394,6 +2414,9 @@ class InvenTreeCustomUserStateModel(models.Model): """ + # API write access is restricted to staff users + IMPORT_REQUIRES_STAFF = True + class Meta: """Metaclass options for this mixin.""" @@ -2733,6 +2756,9 @@ class ParameterTemplate( enabled: Is this template enabled? """ + # API write access is restricted to staff users + IMPORT_REQUIRES_STAFF = True + IMPORT_ID_FIELDS = ['name'] class Meta: diff --git a/src/backend/InvenTree/common/test_audit.py b/src/backend/InvenTree/common/test_audit.py index 9c4b7ae014..923e007bac 100644 --- a/src/backend/InvenTree/common/test_audit.py +++ b/src/backend/InvenTree/common/test_audit.py @@ -652,14 +652,14 @@ class ConfigurationAuditTest(InvenTreeAPITestCase): CustomUnit, 'api-custom-unit-list', 'api-custom-unit-detail', - {'name': 'furlong', 'definition': '220 yard', 'symbol': 'fur'}, + {'name': 'widget_length', 'definition': '220 yard', 'symbol': 'wl'}, {'definition': '201.168 meter'}, ) self.assertEqual( update.changes_dict['definition'], ['220 yard', '201.168 meter'] ) - self.assertEqual(update.object_repr, 'furlong') + self.assertEqual(update.object_repr, 'widget_length') # Registry reload (post_save receiver) is unaffected by auditing CustomUnit.objects.create(name='smoot', definition='1.7018 meter') diff --git a/src/backend/InvenTree/common/tests.py b/src/backend/InvenTree/common/tests.py index 998e56a71a..e3a89d1999 100644 --- a/src/backend/InvenTree/common/tests.py +++ b/src/backend/InvenTree/common/tests.py @@ -2314,6 +2314,45 @@ class CustomUnitAPITest(InvenTreeAPITestCase): self.assertEqual(a.definition, 'meter') self.assertEqual(b.definition, '3 * circular_a') + def test_validation_builtin_clash(self): + """Test that custom units cannot override built-in unit definitions.""" + import InvenTree.conversion as conversion + + self.user.is_staff = True + self.user.save() + + url = reverse('api-custom-unit-list') + + # Names which resolve to a built-in unit (including prefixed / plural forms) + for name in ['pound', 'gram', 'kpound', 'meters', 'piece', 'R']: + response = self.post( + url, {'name': name, 'definition': '2 * kilogram'}, expected_code=400 + ) + self.assertIn('name', response.data) + + # Symbols which resolve to a built-in unit + for symbol in ['lb', 'kg', 'ea']: + response = self.post( + url, + {'name': 'custom_mass', 'definition': '2 * kilogram', 'symbol': symbol}, + expected_code=400, + ) + self.assertIn('symbol', response.data) + + self.assertFalse(CustomUnit.objects.filter(name='custom_mass').exists()) + + # Built-in conversions are unaffected + self.assertAlmostEqual( + float(conversion.convert_physical_value('1 lb', 'kg')), 0.45359237 + ) + + # A non-clashing unit is accepted + self.post( + url, + {'name': 'custom_mass', 'definition': '2 * kilogram', 'symbol': 'cm2kg'}, + expected_code=201, + ) + def test_api(self): """Test the CustomUnit API.""" response = self.get(reverse('api-custom-unit-all')) diff --git a/src/backend/InvenTree/importer/api.py b/src/backend/InvenTree/importer/api.py index 763ed99a50..642c22df6e 100644 --- a/src/backend/InvenTree/importer/api.py +++ b/src/backend/InvenTree/importer/api.py @@ -22,7 +22,6 @@ from InvenTree.mixins import ( RetrieveUpdateAPI, RetrieveUpdateDestroyAPI, ) -from users.permissions import check_user_permission class DataImporterPermissionMixin: @@ -120,9 +119,8 @@ class DataImportSessionAcceptFields(APIView): raise PermissionDenied() # Check that the user has permission to accept the field mapping - if model_class := session.model_class: - if not check_user_permission(request.user, model_class, 'change'): - raise PermissionDenied() + if not session.check_import_permission(request.user): + raise PermissionDenied() # Attempt to accept the mapping (may raise an exception if the mapping is invalid) session.accept_mapping() @@ -151,6 +149,8 @@ class DataImportSessionAcceptRows(DataImporterPermissionMixin, CreateAPI): user = self.request.user if not user.is_staff and session.user != user: raise PermissionDenied() + if not session.check_import_permission(user): + raise PermissionDenied() ctx['session'] = session ctx['request'] = self.request diff --git a/src/backend/InvenTree/importer/models.py b/src/backend/InvenTree/importer/models.py index 4cd73f9956..796b15ed8c 100644 --- a/src/backend/InvenTree/importer/models.py +++ b/src/backend/InvenTree/importer/models.py @@ -169,6 +169,27 @@ class DataImportSession(models.Model): return supported_models().get(self.model_type, None) + def check_import_permission(self, user) -> bool: + """Check if the provided user is allowed to import data into the target model. + + - Models marked with IMPORT_REQUIRES_STAFF require the user to be staff, + matching the API write permissions for that model. + - The user must also have 'change' permission for the target model. + """ + from users.permissions import check_user_permission + + model_class = self.model_class + + if not model_class: + return True + + if getattr(model_class, 'IMPORT_REQUIRES_STAFF', False) and not getattr( + user, 'is_staff', False + ): + return False + + return check_user_permission(user, model_class, 'change') + def get_lookup_fields_for_field(self, field_name: str) -> list: """Return the valid lookup fields for a given related (FK) field. diff --git a/src/backend/InvenTree/importer/serializers.py b/src/backend/InvenTree/importer/serializers.py index 6393541f0e..fcb8a5aeb1 100644 --- a/src/backend/InvenTree/importer/serializers.py +++ b/src/backend/InvenTree/importer/serializers.py @@ -100,6 +100,19 @@ class DataImportSessionSerializer(InvenTreeModelSerializer): ) ) + # Prevent creation of an import session for a staff-only model by a non-staff user + if self.instance is None and (request := self.context.get('request', None)): + model_class = importer.models.DataImportSession( + model_type=value + ).model_class + + if getattr(model_class, 'IMPORT_REQUIRES_STAFF', False) and not getattr( + request.user, 'is_staff', False + ): + raise ValidationError( + _('Only staff users can import data for this model') + ) + return value def validate_field_defaults(self, defaults): diff --git a/src/backend/InvenTree/importer/tests.py b/src/backend/InvenTree/importer/tests.py index d80067d9cc..13f1e428b3 100644 --- a/src/backend/InvenTree/importer/tests.py +++ b/src/backend/InvenTree/importer/tests.py @@ -865,6 +865,96 @@ class ImportAPITest(ImporterMixin, InvenTreeAPITestCase): self.assignRole('purchase_order.change') self.post(url, expected_code=200) + def test_staff_only_models(self): + """Test that models with IMPORT_REQUIRES_STAFF cannot be imported by non-staff users. + + The importer must enforce the same restrictions as the API endpoints for these models. + """ + import common.models + + self.user.is_staff = False + self.user.save() + + url = reverse('api-importer-session-list') + + for model_type in [ + 'customunit', + 'projectcode', + 'parametertemplate', + 'inventreecustomuserstatemodel', + ]: + # Non-staff user cannot create an import session for a staff-only model + response = self.post( + url, + { + 'model_type': model_type, + 'data_file': ContentFile('name\nx', 'x.csv'), + }, + format='multipart', + expected_code=400, + ) + self.assertIn('model_type', response.data) + + # A pre-existing session (e.g. created while the user was staff) cannot be accepted + self.user.is_staff = True + self.user.save() + + data = self.post( + url, + { + 'model_type': 'customunit', + 'data_file': ContentFile( + 'name,definition\nimported_unit,2 * kilogram', 'units.csv' + ), + }, + format='multipart', + expected_code=201, + ).data + + accept_url = reverse( + 'api-import-session-accept-fields', kwargs={'pk': data['pk']} + ) + accept_rows_url = reverse( + 'api-import-session-accept-rows', kwargs={'pk': data['pk']} + ) + + self.user.is_staff = False + self.user.save() + + self.post(accept_url, expected_code=403) + self.post(accept_rows_url, {'rows': []}, expected_code=403) + + # Staff user can accept the mapping + self.user.is_staff = True + self.user.save() + + self.post(accept_url, expected_code=200) + + row_ids = list( + DataImportRow.objects.filter(session__pk=data['pk']).values_list( + 'pk', flat=True + ) + ) + self.assertEqual(len(row_ids), 1) + + # Non-staff user cannot accept the rows + self.user.is_staff = False + self.user.save() + + self.post(accept_rows_url, {'rows': row_ids}, expected_code=403) + self.assertFalse( + common.models.CustomUnit.objects.filter(name='imported_unit').exists() + ) + + # Staff user can accept the rows + self.user.is_staff = True + self.user.save() + + self.post(accept_rows_url, {'rows': row_ids}, expected_code=201) + self.assertTrue( + common.models.CustomUnit.objects.filter(name='imported_unit').exists() + ) + def test_field_filters_api(self): """Test that the API rejects field filters which traverse related models.""" import json @@ -997,8 +1087,10 @@ class ImportAPITest(ImporterMixin, InvenTreeAPITestCase): self.post(url, {'rows': []}, expected_code=403) # Staff can reach the endpoint (rows list is empty so validation rejects with 400, not 403) + # Company is part of the purchase_order ruleset self.user.is_staff = True self.user.save() + self.assignRole('purchase_order.change') self.post(url, {'rows': []}, expected_code=400) def test_session_cleanup_on_complete(self):