Rework org perms again

temba/channels/tests.py

@@ -972,8 +972,7 @@ def test_configuration(self):
         config_url = reverse("channels.channel_configuration", args=[self.ex_channel.uuid])
 
         # can't view configuration if not logged in
-        response = self.client.get(config_url)
-        self.assertLoginRedirect(response)
+        self.assertRequestDisallowed(config_url, [None, self.agent])
 
         self.login(self.admin)
 

temba/contacts/tests.py

@@ -468,8 +468,7 @@ def test_read(self, mr_mocks):
 
         read_url = reverse("contacts.contact_read", args=[joe.uuid])
 
-        response = self.client.get(read_url)
-        self.assertLoginRedirect(response)
+        self.assertRequestDisallowed(read_url, [None, self.agent])
 
         self.assertContentMenu(read_url, self.user, [])
         self.assertContentMenu(read_url, self.editor, ["Edit", "Start Flow", "Open Ticket"])

temba/orgs/views/base.py

@@ -35,9 +35,9 @@ class BaseReadView(OrgObjPermsMixin, SmartReadView):
     def derive_queryset(self, **kwargs):
         qs = super().derive_queryset(**kwargs)
 
-        # don't filter by org for staff users but let OrgObjPermsMixin provide a redirect
+        # filter by allowed org as we'll let OrgObjPermsMixin provide a redirect
         if not self.request.user.is_staff:
-            qs = qs.filter(org=self.request.org)
+            qs = qs.filter(org__in=self.request.user.orgs.all())
 
         if hasattr(self.model, "is_active"):
             qs = qs.filter(is_active=True)

temba/orgs/views/mixins.py

@@ -22,54 +22,45 @@ class OrgPermsMixin:
     def derive_org(self):
         return self.request.org
 
-    def _check_org_perm(self, permission: str) -> tuple[bool, bool]:
+    def has_org_perm(self, permission: str) -> bool:
         """
-        Checks if the current user has the given permission for the current org and whether they have it by servicing.
+        Figures out if the current user has the given permission.
         """
 
         org = self.derive_org()
         user = self.request.user
 
         # can't have an org perm without an org
         if not org:
-            return False, False
+            return False
 
         if user.is_anonymous:
-            return False, False
-
-        has_perm = user.has_org_perm(org, permission)
-
-        if not has_perm and user.is_staff:
-            return True, True
-
-        return has_perm, False
+            return False
 
-    def has_org_perm(self, permission: str) -> bool:
-        has_perm, by_servicing = self._check_org_perm(permission)
-        return has_perm
+        return user.is_staff or user.has_org_perm(org, permission)
 
     def has_permission(self, request, *args, **kwargs) -> bool:
         """
-        Figures out if the current user has permissions for this view.
+        Figures out if the current user has the required permission for this view.
         """
 
         self.kwargs = kwargs
         self.args = args
         self.request = request
 
-        has_perm, by_servicing = self._check_org_perm(self.permission)
-
-        # by default staff are only allowed to GET views when servicing
-        if self.readonly_servicing and by_servicing and request.method != "GET":
-            return False
-
-        return has_perm
+        return self.has_org_perm(self.permission)
 
     def dispatch(self, request, *args, **kwargs):
-        # non admin authenticated users without orgs get the org chooser
+        org = self.derive_org()
         user = self.request.user
-        if user.is_authenticated and not user.is_staff:
-            if not self.derive_org():
+
+        if org:
+            # when servicing, non-superuser staff can only GET
+            is_servicing = request.user.is_staff and not org.users.filter(id=request.user.id).exists()
+            if is_servicing and not request.user.is_superuser and request.method != "GET":
+                return HttpResponseForbidden()
+        else:
+            if user.is_authenticated and not user.is_staff:
                 return HttpResponseRedirect(reverse("orgs.org_choose"))
 
         return super().dispatch(request, *args, **kwargs)
@@ -86,20 +77,30 @@ def get_object_org(self):
     def has_permission(self, request, *args, **kwargs) -> bool:
         has_perm = super().has_permission(request, *args, **kwargs)
 
-        # allow staff to GET any object because they'll be redirected on org mismatch
-        if self.request.user.is_staff and self.request.method == "GET":
+        # even without a current org, staff can switch to the object's org
+        if request.user.is_staff:
             return True
 
-        return has_perm and self.request.org == self.get_object_org()
+        if not has_perm:
+            return False
+
+        obj_org = self.get_object_org()
+        is_this_org = self.request.org == obj_org
+
+        return has_perm and (is_this_org or obj_org.get_users().filter(id=request.user.id).exists())
 
     def pre_process(self, request, *args, **kwargs):
         org = self.get_object_org()
 
-        # staff users are redirected to service page if org doesn't match
-        if request.user.is_staff and self.request.org != org:
-            return HttpResponseRedirect(
-                f"{reverse('staff.org_service')}?next={quote_plus(request.path)}&other_org={org.id}"
-            )
+        if self.request.org != org:
+            if request.user.is_staff:
+                # staff users are redirected to service page if org doesn't match
+                return HttpResponseRedirect(
+                    f"{reverse('staff.org_service')}?next={quote_plus(request.path)}&other_org={org.id}"
+                )
+            else:
+                # TODO implement view to let regular users switch orgs as well
+                return HttpResponseRedirect(reverse("orgs.org_choose"))
 
         return super().pre_process(request, *args, **kwargs)
 

temba/orgs/views/tests.py

@@ -48,7 +48,7 @@ def test_has_permission(self):
         self.assertEqual(200, self.client.get(create_url).status_code)
 
         # staff still can't POST
-        self.assertLoginRedirect(self.client.post(create_url, {"name": "Sales"}))
+        self.assertEqual(403, self.client.post(create_url, {"name": "Sales"}).status_code)
 
         # but superuser can
         self.customer_support.is_superuser = True
@@ -64,9 +64,10 @@ def test_has_permission(self):
         self.assertEqual(200, self.client.get(create_url).status_code)
         self.assertRedirect(self.client.post(create_url, {"name": "Support"}), "hide")
 
-    def test_org_obj_perms_mixin(self):
+    def test_obj_perms_mixin(self):
         contact1 = self.create_contact("Bob", phone="+18001234567", org=self.org)
         contact2 = self.create_contact("Zob", phone="+18001234567", org=self.org2)
+        self.org2.add_user(self.admin, OrgRole.ADMINISTRATOR)
 
         contact1_url = reverse("contacts.contact_update", args=[contact1.id])
         contact2_url = reverse("contacts.contact_update", args=[contact2.id])
@@ -80,11 +81,16 @@ def test_org_obj_perms_mixin(self):
         self.assertLoginRedirect(self.client.get(contact1_url))
         self.assertLoginRedirect(self.client.get(contact2_url))
 
-        # editor role does have access tho.. when the URL is for a group in their org
+        # editor does have access tho.. when the URL is for a contact in their org
         self.login(self.editor)
         self.assertEqual(200, self.client.get(contact1_url).status_code)
         self.assertLoginRedirect(self.client.get(contact2_url))
 
+        # admin belongs to both orgs
+        self.login(self.admin, choose_org=self.org)
+        self.assertEqual(200, self.client.get(contact1_url).status_code)
+        self.assertRedirect(self.client.get(contact2_url), reverse("orgs.org_choose"))
+
         # staff can't access without org
         self.login(self.customer_support)
         self.assertRedirect(self.client.get(contact1_url), "/staff/org/service/")
@@ -94,7 +100,7 @@ def test_org_obj_perms_mixin(self):
         self.assertRedirect(self.client.get(contact2_url), "/staff/org/service/")  # wrong org
 
         # staff still can't POST
-        self.assertLoginRedirect(self.client.post(contact1_url, {"name": "Bob"}))
+        self.assertEqual(403, self.client.post(contact1_url, {"name": "Bob"}).status_code)
         self.assertRedirect(self.client.get(contact2_url), "/staff/org/service/")
 
 

temba/staff/tests.py

@@ -191,7 +191,7 @@ def test_service(self, mr_mocks):
         response = self.client.post(
             reverse("contacts.contact_create"), data={"name": "Ben Haggerty", "phone": "0788123123"}
         )
-        self.assertLoginRedirect(response)
+        self.assertEqual(403, response.status_code)
 
         # become super user
         self.customer_support.is_superuser = True