From b98fc9c7a01c56e604227dbfd231c9d0b388507d Mon Sep 17 00:00:00 2001
From: Oliver <oliver.henry.walters@gmail.com>
Date: Tue, 24 Mar 2026 23:28:58 +1100
Subject: [PATCH] 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     |  2 +-
 src/backend/InvenTree/importer/api.py   | 16 +++++++++++++
 src/backend/InvenTree/importer/tests.py | 30 +++++++++++++++++++++++++
 3 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/src/backend/InvenTree/common/api.py b/src/backend/InvenTree/common/api.py
index 2fb2a6b09f..802081af3d 100644
--- a/src/backend/InvenTree/common/api.py
+++ b/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:
diff --git a/src/backend/InvenTree/importer/api.py b/src/backend/InvenTree/importer/api.py
index 5d3fd6a748..c64f71c654 100644
--- a/src/backend/InvenTree/importer/api.py
+++ b/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."""
diff --git a/src/backend/InvenTree/importer/tests.py b/src/backend/InvenTree/importer/tests.py
index 97619c70a3..b4b7b99a6f 100644
--- a/src/backend/InvenTree/importer/tests.py
+++ b/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."""