CustomUnit fixes (#13161)

* CustomUnit fixes

- Require staff access for data import
- Prevent overload of built-in units

* fix unit tests

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.
 

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
 

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:

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')

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'))

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

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.
 

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):

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):