OrgMiddleware should prevent cross-org POSTs

temba/middleware.py

@@ -6,6 +6,7 @@
 
 from django.conf import settings
 from django.contrib import messages
+from django.http import HttpResponseForbidden
 from django.utils import timezone, translation
 
 from temba.orgs.models import Org, User
@@ -42,6 +43,13 @@ def __call__(self, request):
         assert hasattr(request, "user"), "must be called after django.contrib.auth.middleware.AuthenticationMiddleware"
 
         request.org = self.determine_org(request)
+
+        # if request is a POST with an org header, ensure it matches the current org
+        if request.method == "POST":
+            posted_org_id = request.headers.get("X-Temba-Org")
+            if posted_org_id and request.org and request.org.id != int(posted_org_id):
+                return HttpResponseForbidden()
+
         if request.org:
             # set our current role for this org
             request.role = request.org.get_user_role(request.user)

temba/utils/tests.py

@@ -14,7 +14,7 @@
 from django.urls import reverse
 from django.utils import timezone
 
-from temba.orgs.models import Org
+from temba.orgs.models import OrgRole
 from temba.tests import TembaTest, matchers, override_brand
 from temba.utils import json, uuid
 from temba.utils.compose import compose_serialize
@@ -254,42 +254,59 @@ def test_task3(foo, bar):
 
 class MiddlewareTest(TembaTest):
     def test_org(self):
+        index_url = reverse("public.public_index")
 
-        self.other_org = Org.objects.create(
-            name="Other Org",
-            timezone=ZoneInfo("Africa/Kigali"),
-            flow_languages=["eng", "kin"],
-            created_by=self.admin,
-            modified_by=self.admin,
-        )
-        self.other_org.initialize()
+        response = self.client.get(index_url)
+        self.assertFalse(response.has_header("X-Temba-Org"))
+
+        # if a user has a single org, that becomes the current org
+        self.login(self.admin)
+
+        response = self.client.get(index_url)
+        self.assertEqual(str(self.org.id), response["X-Temba-Org"])
+
+        # if not, org isn't set
+        self.org2.add_user(self.admin, OrgRole.ADMINISTRATOR)
 
-        response = self.client.get(reverse("public.public_index"))
+        response = self.client.get(index_url)
         self.assertFalse(response.has_header("X-Temba-Org"))
 
+        # org will be read from session if set
+        s = self.client.session
+        s.update({"org_id": self.org.id})
+        s.save()
+
+        response = self.client.get(index_url)
+        self.assertEqual(str(self.org.id), response["X-Temba-Org"])
+
+        # org can be sent as a header too and we check it matches
+        response = self.client.post(reverse("flows.flow_create"), {}, headers={"X-Temba-Org": str(self.org.id)})
+        self.assertEqual(200, response.status_code)
+
+        response = self.client.post(reverse("flows.flow_create"), {}, headers={"X-Temba-Org": str(self.org2.id)})
+        self.assertEqual(403, response.status_code)
+
         self.login(self.customer_support)
 
         # our staff user doesn't have a default org
-        response = self.client.get(reverse("public.public_index"))
+        response = self.client.get(index_url)
         self.assertFalse(response.has_header("X-Temba-Org"))
 
         # but they can specify an org to service as a header
-        response = self.client.get(reverse("public.public_index"), headers={"X-Temba-Service-Org": str(self.org.id)})
+        response = self.client.get(index_url, headers={"X-Temba-Service-Org": str(self.org.id)})
         self.assertEqual(response["X-Temba-Org"], str(self.org.id))
 
-        response = self.client.get(reverse("public.public_index"))
+        response = self.client.get(index_url)
         self.assertFalse(response.has_header("X-Temba-Org"))
 
-        self.login(self.admin)
+        self.login(self.editor)
 
-        response = self.client.get(reverse("public.public_index"))
+        response = self.client.get(index_url)
         self.assertEqual(response["X-Temba-Org"], str(self.org.id))
 
         # non-staff can't specify a different org from there own
-        response = self.client.get(
-            reverse("public.public_index"), headers={"X-Temba-Service-Org": str(self.other_org.id)}
-        )
-        self.assertNotEqual(response["X-Temba-Org"], str(self.other_org.id))
+        response = self.client.get(index_url, headers={"X-Temba-Service-Org": str(self.org2.id)})
+        self.assertNotEqual(response["X-Temba-Org"], str(self.org2.id))
 
     def test_redirect(self):
         self.assertNotRedirect(self.client.get(reverse("public.public_index")), None)