Prevent workspace admins to modify the system user on a workspace

temba/api/v2/tests.py

@@ -5300,7 +5300,7 @@ def test_users(self):
                 },
             ],
             # one query per user for their settings
-            num_queries=NUM_BASE_SESSION_QUERIES + 3,
+            num_queries=NUM_BASE_SESSION_QUERIES + 2,
         )
 
         # filter by roles

temba/api/v2/views.py

@@ -3543,7 +3543,7 @@ def derive_queryset(self):
         else:
             roles = None
 
-        return org.get_users(roles=roles).prefetch_related("settings")
+        return org.get_users(roles=roles)
 
     def get_serializer_context(self):
         context = super().get_serializer_context()

temba/orgs/models.py

@@ -1043,7 +1043,7 @@ def get_users(self, *, roles: list = None, with_perm: str = None):
         """
         Gets users in this org, filtered by role or permission.
         """
-        qs = self.users.filter(is_active=True)
+        qs = self.users.filter(is_active=True).select_related("settings")
 
         if with_perm:
             app_label, codename = with_perm.split(".")

temba/orgs/views/tests.py

@@ -1728,6 +1728,9 @@ class UserCRUDLTest(TembaTest, CRUDLTestMixin):
     def test_list(self):
         list_url = reverse("orgs.user_list")
 
+        # add system user to workspace
+        self.org.add_user(self.system_user, OrgRole.ADMINISTRATOR)
+
         # nobody can access if users feature not enabled
         response = self.requestView(list_url, self.admin)
         self.assertRedirect(response, reverse("orgs.org_workspace"))
@@ -1825,14 +1828,28 @@ def test_update(self):
         self.assertEqual({self.user, self.editor}, set(self.org.get_users(roles=[OrgRole.EDITOR])))
         self.assertEqual({self.admin}, set(self.org.get_users(roles=[OrgRole.ADMINISTRATOR])))
 
+        # even if we add system user to workspace
+        self.org.add_user(self.system_user, OrgRole.ADMINISTRATOR)
+        response = self.assertUpdateSubmit(update_url, self.admin, {"role": "E"}, object_unchanged=self.admin)
+        self.assertRedirect(response, reverse("orgs.user_list"))
+        self.assertEqual({self.user, self.editor}, set(self.org.get_users(roles=[OrgRole.EDITOR])))
+        self.assertEqual({self.admin, self.system_user}, set(self.org.get_users(roles=[OrgRole.ADMINISTRATOR])))
+
         # add another admin to workspace and try again
         self.org.add_user(self.admin2, OrgRole.ADMINISTRATOR)
 
         response = self.assertUpdateSubmit(update_url, self.admin, {"role": "E"}, object_unchanged=self.admin)
         self.assertRedirect(response, reverse("orgs.org_start"))  # no longer have access to user list page
 
         self.assertEqual({self.user, self.editor, self.admin}, set(self.org.get_users(roles=[OrgRole.EDITOR])))
-        self.assertEqual({self.admin2}, set(self.org.get_users(roles=[OrgRole.ADMINISTRATOR])))
+        self.assertEqual({self.admin2, self.system_user}, set(self.org.get_users(roles=[OrgRole.ADMINISTRATOR])))
+
+        # cannot update system user on a workspace
+        update_url = reverse("orgs.user_update", args=[self.system_user.id])
+        response = self.assertUpdateSubmit(update_url, self.admin2, {"role": "E"}, object_unchanged=self.system_user)
+        self.assertRedirect(response, reverse("orgs.user_list"))
+        self.assertEqual({self.user, self.editor, self.admin}, set(self.org.get_users(roles=[OrgRole.EDITOR])))
+        self.assertEqual({self.admin2, self.system_user}, set(self.org.get_users(roles=[OrgRole.ADMINISTRATOR])))
 
     def test_delete(self):
         delete_url = reverse("orgs.user_delete", args=[self.agent.id])
@@ -1868,6 +1885,19 @@ def test_delete(self):
         self.assertRedirect(response, reverse("orgs.user_list"))
         self.assertEqual({self.user, self.editor, self.admin}, set(self.org.get_users()))
 
+        # cannot still even when the other admin is a system user
+        self.org.add_user(self.system_user, OrgRole.ADMINISTRATOR)
+        response = self.assertDeleteSubmit(delete_url, self.admin, object_unchanged=self.admin)
+        self.assertRedirect(response, reverse("orgs.user_list"))
+        self.assertEqual({self.user, self.editor, self.admin, self.system_user}, set(self.org.get_users()))
+
+        # cannot remove system user too
+        response = self.assertDeleteSubmit(
+            reverse("orgs.user_delete", args=[self.system_user.id]), self.admin, object_unchanged=self.system_user
+        )
+        self.assertRedirect(response, reverse("orgs.user_list"))
+        self.assertEqual({self.user, self.editor, self.admin, self.system_user}, set(self.org.get_users()))
+
         # add another admin to workspace and try again
         self.org.add_user(self.admin2, OrgRole.ADMINISTRATOR)
 
@@ -1876,7 +1906,7 @@ def test_delete(self):
         # this time we could remove ourselves
         response = self.assertDeleteSubmit(delete_url, self.admin, object_unchanged=self.admin)
         self.assertRedirect(response, reverse("orgs.org_choose"))
-        self.assertEqual({self.user, self.editor, self.admin2}, set(self.org.get_users()))
+        self.assertEqual({self.user, self.editor, self.admin2, self.system_user}, set(self.org.get_users()))
 
     def test_account(self):
         self.login(self.agent)

temba/orgs/views/views.py

@@ -408,14 +408,19 @@ class List(RequireFeatureMixin, SpaMixin, BaseListView):
         search_fields = ("email__icontains", "first_name__icontains", "last_name__icontains")
 
         def derive_queryset(self, **kwargs):
-            return (
+            qs = (
                 super(BaseListView, self)
                 .derive_queryset(**kwargs)
                 .filter(id__in=self.request.org.get_users().values_list("id", flat=True))
                 .order_by(Lower("email"))
                 .select_related("settings")
             )
 
+            if not self.request.user.is_staff:
+                qs = qs.exclude(settings__is_system=True)
+
+            return qs
+
         def get_context_data(self, **kwargs):
             context = super().get_context_data(**kwargs)
 
@@ -527,8 +532,17 @@ def save(self, obj):
             team = self.form.cleaned_data.get("team")
             team = (team or self.request.org.default_ticket_team) if role == OrgRole.AGENT else None
 
+            # do not update a system user
+            if obj.settings.is_system:
+                return obj
+
             # don't update if user is the last administrator and role is being changed to something else
-            has_other_admins = self.request.org.get_users(roles=[OrgRole.ADMINISTRATOR]).exclude(id=obj.id).exists()
+            has_other_admins = (
+                self.request.org.get_users(roles=[OrgRole.ADMINISTRATOR])
+                .exclude(id=obj.id)
+                .exclude(settings__is_system=True)
+                .exists()
+            )
             if role != OrgRole.ADMINISTRATOR and not has_other_admins:
                 return obj
 
@@ -560,8 +574,14 @@ def get_context_data(self, **kwargs):
         def post(self, request, *args, **kwargs):
             user = self.get_object()
 
-            # only actually remove user if they're not the last administator
-            if self.request.org.get_users(roles=[OrgRole.ADMINISTRATOR]).exclude(id=user.id).exists():
+            # only actually remove user if they're not the last administator or a system user
+            if (
+                self.request.org.get_users(roles=[OrgRole.ADMINISTRATOR])
+                .exclude(id=user.id)
+                .exclude(settings__is_system=True)
+                .exists()
+                and not user.settings.is_system
+            ):
                 self.request.org.remove_user(user)
 
             return HttpResponseRedirect(self.get_redirect_url())

temba/staff/tests.py

@@ -223,7 +223,7 @@ def test_list(self):
         self.assertStaffOnly(list_url)
 
         response = self.requestView(list_url, self.customer_support)
-        self.assertEqual(8, len(response.context["object_list"]))
+        self.assertEqual(9, len(response.context["object_list"]))
         self.assertEqual("/staff/users/all", response.headers[TEMBA_MENU_SELECTION])
 
         response = self.requestView(list_url + "?filter=beta", self.customer_support)

temba/tests/base.py

@@ -72,6 +72,10 @@ def setUp(self):
         self.agent = self.create_user("[email protected]", first_name="Agnes")
         self.customer_support = self.create_user("[email protected]", is_staff=True)
 
+        self.system_user = self.create_user("[email protected]")
+        self.system_user.settings.is_system = True
+        self.system_user.settings.save(update_fields=("is_system",))
+
         self.org = Org.objects.create(
             name="Nyaruka",
             timezone=ZoneInfo("Africa/Kigali"),