Restrict queryset for DataImportSession (#11602)

* Restrict queryset for DataImportSession

- Only allow non-staff users to see their own sessions

* Add unit test

* raise PermissionDenied if no user info available

src/backend/InvenTree/common/api.py

@@ -1185,7 +1185,7 @@ class DataOutputEndpointMixin:
         try:
             user = self.request.user
         except AttributeError:
-            return common.models.DataOutput.objects.none()
+            raise PermissionDenied('User information is not available')
 
         # Allow staff users access to all DataOutput objects
         if user.is_staff:

src/backend/InvenTree/importer/api.py

@@ -73,6 +73,22 @@ class DataImportSessionMixin:
     serializer_class = importer.serializers.DataImportSessionSerializer
     permission_classes = [InvenTree.permissions.DataImporterPermission]
 
+    def get_queryset(self):
+        """Return the set of DataImportSession objects that the user has permission to view."""
+        queryset = super().get_queryset()
+
+        try:
+            user = self.request.user
+        except AttributeError:
+            raise PermissionDenied('User information is not available')
+
+        # Allow staff users access to all DataImportSession objects
+        if user.is_staff:
+            return queryset
+
+        # For non-staff users, only allow access to sessions that they have created
+        return queryset.filter(user=user)
+
 
 class DataImportSessionList(BulkDeleteMixin, DataImportSessionMixin, ListCreateAPI):
     """API endpoint for accessing a list of DataImportSession objects."""

src/backend/InvenTree/importer/tests.py

@@ -174,6 +174,36 @@ class ImportAPITest(ImporterMixin, InvenTreeAPITestCase):
         # Check that there are new database records
         self.assertEqual(PartCategory.objects.count(), N + 4)
 
+    def test_session_list(self):
+        """Test API endpoint which details the list of import sessions."""
+        url = reverse('api-importer-session-list')
+
+        # Construct a dummy file
+        f = self.helper_file('companies.csv')
+
+        for ii in range(5):
+            DataImportSession.objects.create(
+                data_file=f,
+                model_type='company',
+                user=self.user if ii % 2 == 0 else None,
+            )
+
+        # Staff user should see all sessions
+        self.user.is_staff = True
+        self.user.save()
+
+        response = self.get(url)
+        self.assertEqual(len(response.data), 5)
+
+        # Non-staff user should only see sessions which they own
+        self.user.is_staff = False
+        self.user.save()
+
+        response = self.get(url)
+        self.assertEqual(len(response.data), 3)
+        for session in response.data:
+            self.assertEqual(session['user'], self.user.pk)
+
 
 class AdminTest(ImporterMixin, AdminTestCase):
     """Tests for the admin interface integration."""