Pin the API permission boundary for every viewset

A buyer account is a legitimate authenticated user, so IsAuthenticated alone
lets it through. Sweep every ModelViewSet endpoint and assert a buyer gets 403
on write, and on read where the API is ops-only; assert ops is not forbidden;
and keep an explicit allowlist for the chat, user, notification and subscription
reads a buyer may still make. Also assert the escalation is closed: a buyer
cannot PATCH their own role to admin, sees only themselves in the user API, and
cannot reach another user's notification or subscription.

apps/core/tests.py

@@ -23,6 +23,7 @@ from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.core.management import call_command
 from django.test import Client, SimpleTestCase, TestCase, override_settings
+from rest_framework.test import APIClient
 
 from apps.batches.models import ProcessingBatch
 from apps.capa.models import CAPARecord
@@ -33,7 +34,9 @@ from apps.compliance.models import (
     IntakeBatchLineage,
     ProductSpecification,
 )
+from apps.notifications.models import Notification
 from apps.qc.models import QCResult
+from apps.subscriptions.models import Subscription
 
 from .tiles import browser_tile_url, upstream_url
 
@@ -118,6 +121,126 @@ class PortalPagesRenderTests(TestCase):
         self.sweep(self.make_user("buyer-sweep@ixg.local", "buyer"), PORTAL_ROUTES, "portal.ixgltd.com")
 
 
+class ApiPermissionBoundaryTests(TestCase):
+    """The API permission boundary, pinned for every viewset.
+
+    Batch writes were closed first; this sweeps the rest. A buyer account is the
+    adversary: it is a legitimate, authenticated user, so ``IsAuthenticated``
+    alone lets it through. Anything that changes a record — a document, a QC
+    result, a certificate, a shipment status — must be ops/admin only. The buyer
+    portal is server-rendered through the ORM, so it needs no API access; the one
+    exception is the AI chat, which the buyer portal calls from the browser.
+    """
+
+    # Every ModelViewSet that changes a record. A buyer must get 403 on write.
+    OPS_WRITE_ENDPOINTS = [
+        "/api/batches/",
+        "/api/qc-results/",
+        "/api/shipments/",
+        "/api/documents/",
+        "/api/certificates-of-analysis/",
+        "/api/regulatory-certs/",
+        "/api/capa-records/",
+        "/api/esg-reports/",
+        "/api/agents/",
+        "/api/product-specs/",
+        "/api/facilities/",
+        "/api/origin-zones/",
+        "/api/collector-groups/",
+        "/api/harvest-intakes/",
+        "/api/intake-batch-lineage/",
+        "/api/eudr-cases/",
+        "/api/eudr-evidence/",
+        "/api/risk-assessments/",
+        "/api/mitigation-actions/",
+        "/api/dds-references/",
+    ]
+
+    # Reads a buyer may still make: the chat it calls from the browser, and the
+    # user/notification/subscription rows scoped to the buyer itself.
+    BUYER_READABLE_ENDPOINTS = [
+        "/api/agents/",
+        "/api/ai-interactions/",
+        "/api/users/",
+        "/api/notifications/",
+        "/api/subscriptions/",
+    ]
+
+    def make_user(self, email, role):
+        return User.objects.create_user(
+            email=email, password="pass1234", role=role,
+            full_name=email.split("@")[0], is_staff=True,
+        )
+
+    def setUp(self):
+        self.buyer = self.make_user("boundary-buyer@ixg.local", "buyer")
+        self.other_buyer = self.make_user("boundary-other@ixg.local", "buyer")
+        self.ops = self.make_user("boundary-ops@ixg.local", "ops")
+        self.client = APIClient()
+
+    def test_a_buyer_cannot_write_to_any_ops_api(self):
+        self.client.force_authenticate(user=self.buyer)
+        for endpoint in self.OPS_WRITE_ENDPOINTS:
+            with self.subTest(endpoint=endpoint):
+                self.assertEqual(
+                    self.client.post(endpoint, {}, format="json").status_code, 403
+                )
+
+    def test_a_buyer_cannot_read_the_ops_apis(self):
+        self.client.force_authenticate(user=self.buyer)
+        for endpoint in self.OPS_WRITE_ENDPOINTS:
+            if endpoint in self.BUYER_READABLE_ENDPOINTS:
+                continue
+            with self.subTest(endpoint=endpoint):
+                self.assertEqual(self.client.get(endpoint).status_code, 403)
+
+    def test_ops_is_not_forbidden_on_the_ops_apis(self):
+        self.client.force_authenticate(user=self.ops)
+        for endpoint in self.OPS_WRITE_ENDPOINTS:
+            with self.subTest(endpoint=endpoint):
+                self.assertNotEqual(self.client.get(endpoint).status_code, 403)
+
+    def test_a_buyer_can_still_reach_the_chat_endpoints(self):
+        self.client.force_authenticate(user=self.buyer)
+        for endpoint in self.BUYER_READABLE_ENDPOINTS:
+            with self.subTest(endpoint=endpoint):
+                self.assertNotEqual(self.client.get(endpoint).status_code, 403)
+        response = self.client.post("/api/ai-interactions/chat/", {}, format="json")
+        self.assertNotEqual(response.status_code, 403)
+
+    def test_a_buyer_cannot_promote_themselves_to_admin(self):
+        self.client.force_authenticate(user=self.buyer)
+        response = self.client.patch(
+            f"/api/users/{self.buyer.id}/", {"role": "admin"}, format="json"
+        )
+        self.assertEqual(response.status_code, 403)
+        self.buyer.refresh_from_db()
+        self.assertEqual(self.buyer.role, "buyer")
+
+    def test_the_user_api_shows_a_buyer_only_themselves(self):
+        self.client.force_authenticate(user=self.buyer)
+        response = self.client.get("/api/users/")
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        rows = body["results"] if isinstance(body, dict) else body
+        self.assertEqual([row["email"] for row in rows], [self.buyer.email])
+
+    def test_a_user_cannot_reach_another_users_rows(self):
+        note = Notification.objects.create(
+            user=self.other_buyer, title="Other", message="Other"
+        )
+        sub = Subscription.objects.create(
+            user=self.other_buyer, plan=Subscription.Plan.STARTER
+        )
+        self.client.force_authenticate(user=self.buyer)
+        self.assertEqual(
+            self.client.get(f"/api/notifications/{note.id}/").status_code, 404
+        )
+        self.assertEqual(
+            self.client.get(f"/api/subscriptions/{sub.id}/").status_code, 404
+        )
+
+
 class TileProxyTests(TestCase):
     """Tiles come from our origin, so the map does not depend on the viewer's
     network being able to reach a third-party tile server."""