fix: 3 cross-hospital data isolation violations (0 violations after fix)

Isolation test: HH-N admin vs E2E-HOSP data across 5 scenarios.
Found + fixed 3 real isolation gaps:

1. observation_list LEAKED unassigned observations across hospitals —
   the filter Q(assigned_department__hospital=X) | Q(assigned_department__isnull=True)
   showed ALL unassigned observations globally. Fixed: add hospital= filter
   to the isnull branch.

2. complaint_add_note allowed cross-hospital note creation —
   ComplaintService.add_note had no hospital check. Any authenticated user
   from any hospital could add notes to any complaint. Fixed: added hospital
   isolation check (same hospital or px_admin).

3. observation_detail accessible cross-hospital when assigned_department is null —
   the RBAC check only ran if observation.assigned_department was set.
   Fixed: added fallback hospital check for observations with no department.

Result: 16 PASS, 0 FAIL, 0 isolation violations.
Tested: list isolation (8 modules), detail isolation (3), write isolation (2),
px_admin hospital switching, observation no-dept edge case.

apps/complaints/services/complaint_service.py

@@ -481,6 +481,11 @@ class ComplaintService:
         if not message:
             raise ComplaintServiceError("Please enter a note.")
 
+        # Cross-hospital isolation: only users from the same hospital can add notes
+        if not created_by.is_px_admin() and complaint.hospital_id and created_by.hospital_id and \
+                complaint.hospital_id != created_by.hospital_id:
+            raise ComplaintServiceError("You don't have permission to add notes to this complaint.")
+
         update = ComplaintUpdate.objects.create(
             complaint=complaint,
             update_type="note",

apps/core/management/commands/create_hhn_test_users.py

@@ -0,0 +1,89 @@
+"""
+Create test users in HH-N (Al Nuzha) for cross-hospital isolation testing.
+Creates a hospital_admin + px_employee + dept_manager in HH-N with Dev@123456.
+
+Usage:
+  manage.py create_hhn_test_users
+  manage.py create_hhn_test_users --delete-existing
+"""
+
+from django.contrib.auth.models import Group
+from django.core.management.base import BaseCommand
+
+from apps.accounts.models import User
+from apps.organizations.models import Department, Hospital, Staff
+
+
+USERS_CONFIG = [
+    {"email": "e2e-hhn-admin@px360.test", "role": "Hospital Admin", "first": "E2E", "last": "HHN Admin"},
+    {"email": "e2e-hhn-employee@px360.test", "role": "PX Employee", "first": "E2E", "last": "HHN Employee"},
+    {"email": "e2e-hhn-manager@px360.test", "role": "Department Manager", "first": "E2E", "last": "HHN Manager"},
+]
+
+
+class Command(BaseCommand):
+    help = "Create HH-N test users for cross-hospital isolation testing."
+
+    def add_arguments(self, parser):
+        parser.add_argument("--password", default="Dev@123456")
+        parser.add_argument("--delete-existing", action="store_true")
+
+    def handle(self, *args, **options):
+        password = options["password"]
+        hhn = Hospital.objects.get(code="HH-N")
+
+        if options["delete_existing"]:
+            deleted = User.objects.filter(email__endswith="@hhn.px360.test").delete()
+            self.stdout.write(f"Deleted {deleted[0]} old HHN test users.")
+
+        dept = Department.objects.filter(hospital=hhn).first()
+        created = 0
+        for cfg in USERS_CONFIG:
+            group = Group.objects.filter(name=cfg["role"]).first()
+            if not group:
+                self.stdout.write(self.style.WARNING(f"SKIP {cfg['email']}: group '{cfg['role']}' missing"))
+                continue
+
+            user, was_created = User.objects.get_or_create(
+                email=cfg["email"],
+                defaults={
+                    "first_name": cfg["first"],
+                    "last_name": cfg["last"],
+                    "hospital": hhn,
+                    "is_active": True,
+                    "is_staff": False,
+                },
+            )
+            if was_created:
+                user.set_password(password)
+                user.save()
+                created += 1
+                self.stdout.write(self.style.SUCCESS(f"CREATED: {cfg['email']} ({cfg['role']})"))
+            else:
+                user.groups.clear()
+                user.hospital = hhn
+                user.set_password(password)
+                user.save()
+                self.stdout.write(f"RESET: {cfg['email']} ({cfg['role']})")
+
+            user.groups.add(group)
+
+            # Give dept_manager a department + staff profile
+            if cfg["role"] == "Department Manager" and dept:
+                user.department = dept
+                user.save(update_fields=["department"])
+                Staff.objects.get_or_create(
+                    user=user,
+                    defaults={
+                        "first_name": user.first_name,
+                        "last_name": user.last_name,
+                        "hospital": hhn,
+                        "department": dept,
+                        "status": "active",
+                        "staff_type": "admin",
+                        "job_title": "HHN Test Manager",
+                        "employee_id": "HHN-TEST-MGR",
+                    },
+                )
+
+        self.stdout.write(self.style.SUCCESS(f"\nDone. {created} new HH-N test users. Password: {password}"))

apps/observations/views.py

@@ -422,10 +422,14 @@ def observation_list(request):
         # PX Admins see all, but filter by selected hospital if set
         if selected_hospital:
             queryset = queryset.filter(
-                Q(assigned_department__hospital=selected_hospital) | Q(assigned_department__isnull=True)
+                Q(assigned_department__hospital=selected_hospital, hospital=selected_hospital) |
+                Q(assigned_department__isnull=True, hospital=selected_hospital)
             )
     elif user.is_hospital_admin() and user.hospital:
-        queryset = queryset.filter(Q(assigned_department__hospital=user.hospital) | Q(assigned_department__isnull=True))
+        queryset = queryset.filter(
+            Q(assigned_department__hospital=user.hospital, hospital=user.hospital) |
+            Q(assigned_department__isnull=True, hospital=user.hospital)
+        )
     elif user.is_department_manager() and user.department:
         queryset = queryset.filter(assigned_department=user.department)
     elif user.hospital:
@@ -551,6 +555,11 @@ def observation_detail(request, pk):
             if observation.assigned_department != user.department:
                 messages.error(request, "You don't have permission to view this observation.")
                 return redirect("observations:observation_list")
+        # Fallback: if no assigned_department, check by hospital
+        if not observation.assigned_department and observation.hospital_id:
+            if not user.is_px_admin() and user.hospital_id and observation.hospital_id != user.hospital_id:
+                messages.error(request, "You don't have permission to view this observation.")
+                return redirect("observations:observation_list")
 
     # Get timeline (combine status logs and notes)
     status_logs = list(observation.status_logs.all())

e2e/tests/workflows/cross-hospital-isolation.spec.ts

@@ -0,0 +1,330 @@
+/* eslint-disable */
+/**
+ * CROSS-HOSPITAL DATA ISOLATION TEST
+ *
+ * Verifies that users from HH-N cannot see or modify E2E-HOSP data, and vice versa.
+ * 5 scenarios: list isolation, detail isolation, write isolation, px_admin switching,
+ * public track cross-hospital.
+ *
+ * Run headed:
+ *   E2E_MAXIMIZED=1 npx playwright test --headed --project chromium cross-hospital-isolation --workers=1
+ */
+import { test } from '@playwright/test';
+import { execSync } from 'child_process';
+import * as path from 'path';
+import * as fs from 'fs';
+import { attachObservers, observe, loginAndScope, OBS, BASE_URL } from '../../helpers/audit';
+import { RoleName } from '../../helpers/helpers';
+
+const PROJECT_ROOT = path.resolve(__dirname, '..', '..', '..');
+const M = 'CrossHospital';
+
+type Page = import('@playwright/test').Page;
+
+function uv(args: string): string { return execSync(args, { cwd: PROJECT_ROOT }).toString(); }
+function shellPy(code: string): string {
+  const tmpFile = `/tmp/e2e_iso_${Date.now()}_${Math.random().toString(36).slice(2,6)}.py`;
+  fs.writeFileSync(tmpFile, code);
+  const out = uv(`uv run manage.py shell < ${tmpFile}`);
+  try { fs.unlinkSync(tmpFile); } catch {}
+  return out;
+}
+
+// Get E2E-HOSP record IDs (the "other" hospital's data that HH-N users should NOT see)
+function getE2EIds(): { cid: string; iid: string; oid: string; ref: string } {
+  const out = shellPy(`
+import django, os
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.dev')
+django.setup()
+from apps.complaints.models import Complaint, Inquiry
+from apps.observations.models import Observation
+from apps.organizations.models import Hospital
+e = Hospital.objects.get(code='E2E-HOSP')
+c = Complaint.objects.filter(hospital=e).first()
+i = Inquiry.objects.filter(hospital=e).first()
+o = Observation.objects.filter(hospital=e).first()
+print(f'CID={c.id} IID={i.id} OID={o.id} REF={c.reference_number}')
+`);
+  const m = out.match(/CID=(\S+)\s+IID=(\S+)\s+OID=(\S+)\s+REF=(\S+)/);
+  if (!m) throw new Error('Failed to get E2E IDs');
+  return { cid: m[1], iid: m[2], oid: m[3], ref: m[4] };
+}
+
+async function loginHHN(page: Page, email: string) {
+  await page.context().clearCookies();
+  await page.goto(`${BASE_URL}/accounts/login/`);
+  await page.waitForSelector('#email');
+  await page.fill('#email', email);
+  await page.fill('#password', 'Dev@123456');
+  await page.click('button[type="submit"]');
+  await page.waitForURL(/\/(?!accounts\/login)/, { timeout: 15000 }).catch(() => {});
+  await page.waitForLoadState('domcontentloaded');
+}
+
+async function csrfOf(page: Page): Promise<string> {
+  return page.context().cookies().then((c) => c.find((x) => x.name === 'csrftoken')?.value || '');
+}
+
+test.describe('Cross-Hospital Isolation', () => {
+  test.describe.configure({ mode: 'serial' });
+  const e2e = getE2EIds();
+
+  // ── 1. LIST ISOLATION ───────────────────────────────────────────────────
+  test('HH-N admin sees NO E2E-HOSP data in lists', async ({ page }) => {
+    attachObservers(page, M, 'hhn_admin');
+    await loginHHN(page, 'e2e-hhn-admin@px360.test');
+
+    const pages = [
+      { url: '/complaints/', label: 'complaint-list' },
+      { url: '/inquiries/', label: 'inquiry-list' },
+      { url: '/observations/', label: 'observation-list' },
+      { url: '/actions/', label: 'action-list' },
+      { url: '/rca/', label: 'rca-list' },
+      { url: '/projects/', label: 'project-list' },
+      { url: '/suggestions/', label: 'suggestion-list' },
+      { url: '/appreciation/', label: 'appreciation-list' },
+    ];
+
+    for (const { url, label } of pages) {
+      try {
+        await page.goto(`${BASE_URL}${url}`);
+        await page.waitForLoadState('domcontentloaded');
+        await page.waitForTimeout(800);
+        const body = (await page.textContent('body')) || '';
+        // Check that the E2E-HOSP reference/code does NOT appear
+        const hasE2EData = body.includes('E2EHOSP') || body.includes('E2E-HOSP') ||
+                            body.includes(e2e.ref) || body.includes('E2E Test Hospital');
+        observe(M, label, !hasE2EData ? 'PASS' : 'FAIL',
+          `HH-N admin ${url}: ${hasE2EData ? 'E2E-HOSP DATA LEAKED!' : 'clean (no E2E data)'}`,
+          { url: page.url() });
+      } catch (e) {
+        observe(M, label, 'FAIL', `exception: ${(e as Error).message}`, { url });
+      }
+    }
+  });
+
+  // ── 2. DETAIL ISOLATION ─────────────────────────────────────────────────
+  test('HH-N admin CANNOT access E2E-HOSP detail pages by UUID', async ({ page }) => {
+    attachObservers(page, M, 'hhn_admin');
+    await loginHHN(page, 'e2e-hhn-admin@px360.test');
+
+    const detailUrls = [
+      { url: `/complaints/${e2e.cid}/`, label: 'complaint-detail' },
+      { url: `/inquiries/${e2e.iid}/`, label: 'inquiry-detail' },
+      { url: `/observations/${e2e.oid}/`, label: 'observation-detail' },
+    ];
+
+    for (const { url, label } of detailUrls) {
+      try {
+        const r = await page.context().request.get(`${BASE_URL}${url}`, { maxRedirects: 0 });
+        const blocked = r.status() === 404 || r.status() === 403 ||
+                        (r.status() === 302 && !url.includes('login'));
+        // For 302, follow and check if it redirected away from the detail
+        let actuallyBlocked = blocked;
+        if (r.status() === 200) {
+          // 200 might mean the detail rendered — check body for E2E reference
+          const body = await r.text();
+          actuallyBlocked = !body.includes(e2e.ref) && !body.includes('E2EHOSP');
+        }
+        observe(M, label, actuallyBlocked ? 'PASS' : 'FAIL',
+          `${url}: HTTP ${r.status()} → ${actuallyBlocked ? 'BLOCKED' : 'E2E DATA ACCESSIBLE!'}`,
+          { http: r.status() });
+      } catch (e) {
+        observe(M, label, 'FAIL', `exception: ${(e as Error).message}`, { url });
+      }
+    }
+  });
+
+  // ── 3. WRITE ISOLATION ──────────────────────────────────────────────────
+  test('HH-N admin CANNOT modify E2E-HOSP records', async ({ page }) => {
+    attachObservers(page, M, 'hhn_admin');
+    await loginHHN(page, 'e2e-hhn-admin@px360.test');
+
+    // Get original state
+    const beforeOut = shellPy(`
+import django, os
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.dev')
+django.setup()
+from apps.complaints.models import Complaint
+c = Complaint.objects.get(id='${e2e.cid}')
+print(f'STATUS={c.status}')
+from apps.complaints.models import ComplaintUpdate
+print(f'NOTES={ComplaintUpdate.objects.filter(complaint=c).count()}')
+`);
+    const beforeStatus = beforeOut.match(/STATUS=(\S+)/)?.[1] || '';
+    const beforeNotes = parseInt(beforeOut.match(/NOTES=(\d+)/)?.[1] || '0');
+
+    const csrf = await csrfOf(page);
+
+    // Attempt change-status
+    try {
+      const r = await page.context().request.post(`${BASE_URL}/complaints/${e2e.cid}/change-status/`, {
+        maxRedirects: 0, headers: { 'X-CSRFToken': csrf },
+        form: { csrfmiddlewaretoken: csrf, status: 'closed', note: 'HHN cross-hospital test' },
+      });
+      // Check if state actually changed
+      const afterOut = shellPy(`
+import django, os
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.dev')
+django.setup()
+from apps.complaints.models import Complaint
+c = Complaint.objects.get(id='${e2e.cid}')
+print(f'STATUS={c.status}')
+`);
+      const afterStatus = afterOut.match(/STATUS=(\S+)/)?.[1] || '';
+      const blocked = afterStatus === beforeStatus;
+      observe(M, 'write-change-status', blocked ? 'PASS' : 'FAIL',
+        `HHN admin change-status on E2E complaint: before=${beforeStatus} after=${afterStatus} → ${blocked ? 'BLOCKED' : 'MODIFIED!'}`,
+        { http: r.status() });
+    } catch (e) {
+      observe(M, 'write-change-status', 'FAIL', `exception: ${(e as Error).message}`, {});
+    }
+
+    // Attempt add-note
+    try {
+      const r = await page.context().request.post(`${BASE_URL}/complaints/${e2e.cid}/add-note/`, {
+        maxRedirects: 0, headers: { 'X-CSRFToken': csrf },
+        form: { csrfmiddlewaretoken: csrf, note: 'HHN cross-hospital note test' },
+      });
+      const afterOut = shellPy(`
+import django, os
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.dev')
+django.setup()
+from apps.complaints.models import Complaint, ComplaintUpdate
+c = Complaint.objects.get(id='${e2e.cid}')
+print(f'NOTES={ComplaintUpdate.objects.filter(complaint=c).count()}')
+`);
+      const afterNotes = parseInt(afterOut.match(/NOTES=(\d+)/)?.[1] || '0');
+      const blocked = afterNotes === beforeNotes;
+      observe(M, 'write-add-note', blocked ? 'PASS' : 'FAIL',
+        `HHN admin add-note on E2E complaint: before=${beforeNotes} after=${afterNotes} → ${blocked ? 'BLOCKED' : 'MODIFIED!'}`,
+        { http: r.status() });
+    } catch (e) {
+      observe(M, 'write-add-note', 'FAIL', `exception: ${(e as Error).message}`, {});
+    }
+  });
+
+  // ── 4. PX ADMIN HOSPITAL SWITCHING ──────────────────────────────────────
+  test('PX admin: switching hospital changes visible data', async ({ page }) => {
+    attachObservers(page, M, 'px_admin');
+    await page.context().clearCookies();
+    // Login as px_admin (no hospital scope yet)
+    await page.goto(`${BASE_URL}/accounts/login/`);
+    await page.waitForSelector('#email');
+    await page.fill('#email', 'e2e-px-admin@px360.test');
+    await page.fill('#password', 'Dev@123456');
+    await page.click('button[type="submit"]');
+    await page.waitForURL(/select-hospital|dashboard|\//, { timeout: 15000 }).catch(() => {});
+    await page.waitForLoadState('domcontentloaded');
+    await page.waitForTimeout(1000);
+
+    // Step 1: Select E2E-HOSP and verify data is visible
+    await page.goto(`${BASE_URL}/core/select-hospital/`);
+    await page.waitForLoadState('domcontentloaded');
+    // Click the E2E hospital radio + submit the form
+    const e2eRadio = page.locator('input[name="hospital_id"][value="' + e2e.cid + '"]').first();
+    // E2E hospital ID might not match complaint ID — get the hospital ID
+    const e2eHospOut = shellPy(`
+import django, os
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.dev')
+django.setup()
+from apps.organizations.models import Hospital
+e = Hospital.objects.get(code='E2E-HOSP')
+h = Hospital.objects.get(code='HH-N')
+print(f'E2E={e.id} HHN={h.id}')
+`);
+    const e2eHospId = e2eHospOut.match(/E2E=(\S+)/)?.[1] || '';
+    const hhnId = e2eHospOut.match(/HHN=(\S+)/)?.[1] || '';
+
+    // Select E2E
+    await page.locator(`input[name="hospital_id"][value="${e2eHospId}"]`).first().check({ force: true }).catch(() => {});
+    await page.locator('button[type="submit"]').first().click().catch(() => {});
+    await page.waitForLoadState('domcontentloaded').catch(() => {});
+    await page.waitForTimeout(1000);
+
+    // Check complaints — should have E2E data
+    await page.goto(`${BASE_URL}/complaints/`);
+    await page.waitForLoadState('domcontentloaded');
+    await page.waitForTimeout(800);
+    const body1 = (await page.textContent('body')) || '';
+    const hasE2ERefs = /CMP-\d{6}-E2EHOSP-\d{4}/.test(body1);
+    observe(M, 'pxadmin-e2e-visible', hasE2ERefs ? 'PASS' : 'WARN',
+      `px_admin with E2E selected: ${hasE2ERefs ? 'E2E complaints visible' : 'no E2E refs found'}`, {});
+
+    // Step 2: Switch to HH-N
+    await page.goto(`${BASE_URL}/core/select-hospital/`);
+    await page.waitForLoadState('domcontentloaded');
+    await page.locator(`input[name="hospital_id"][value="${hhnId}"]`).first().check({ force: true }).catch(() => {});
+    await page.locator('button[type="submit"]').first().click().catch(() => {});
+    await page.waitForLoadState('domcontentloaded').catch(() => {});
+    await page.waitForTimeout(1000);
+
+    // Check complaints — should NOT have E2E refs
+    await page.goto(`${BASE_URL}/complaints/`);
+    await page.waitForLoadState('domcontentloaded');
+    await page.waitForTimeout(800);
+    const body2 = (await page.textContent('body')) || '';
+    const hasE2EAfterSwitch = /CMP-\d{6}-E2EHOSP-\d{4}/.test(body2);
+    observe(M, 'pxadmin-switch-isolation', !hasE2EAfterSwitch ? 'PASS' : 'FAIL',
+      `px_admin after switching to HH-N: ${hasE2EAfterSwitch ? 'E2E COMPLAINTS STILL VISIBLE!' : 'E2E refs hidden'}`, {});
+  });
+
+  // ── 5. OBSERVATION DETAIL GAP CHECK (no assigned_department) ─────────────
+  test('Observation with no assigned_department: cross-hospital detail access', async ({ page }) => {
+    attachObservers(page, M, 'hhn_admin');
+    await loginHHN(page, 'e2e-hhn-admin@px360.test');
+
+    // Find an E2E observation with no assigned_department (or create one)
+    const out = shellPy(`
+import django, os
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.dev')
+django.setup()
+from apps.observations.models import Observation
+from apps.organizations.models import Hospital
+e = Hospital.objects.get(code='E2E-HOSP')
+# Find one without assigned_department
+o = Observation.objects.filter(hospital=e, assigned_department__isnull=True).first()
+if o is None:
+    o = Observation.objects.filter(hospital=e).first()
+    if o:
+        o.assigned_department = None
+        o.save(update_fields=['assigned_department'])
+print(f'OID={o.id if o else "NONE"} TRACKING={o.tracking_code if o else "NONE"}')
+`);
+    const m = out.match(/OID=(\S+)\s+TRACKING=(\S+)/);
+    if (!m || m[1] === 'NONE') {
+      observe(M, 'obs-no-dept-gap', 'WARN', 'no E2E observation to test', {});
+      return;
+    }
+    const oid = m[1];
+
+    try {
+      const r = await page.context().request.get(`${BASE_URL}/observations/${oid}/`, { maxRedirects: 0 });
+      let blocked = r.status() === 404 || r.status() === 403;
+      if (r.status() === 200) {
+        const body = await r.text();
+        blocked = !body.includes('E2EHOSP') && !body.includes('E2E');
+      } else if (r.status() === 302) {
+        // Redirect = likely blocked
+        blocked = true;
+      }
+      observe(M, 'obs-no-dept-gap', blocked ? 'PASS' : 'FAIL',
+        `HHN admin accessing E2E observation (no dept): HTTP ${r.status()} → ${blocked ? 'BLOCKED' : 'ACCESSIBLE!'}`,
+        { http: r.status() });
+    } catch (e) {
+      observe(M, 'obs-no-dept-gap', 'FAIL', `exception: ${(e as Error).message}`, {});
+    }
+  });
+});
+
+test.afterAll(async () => {
+  const counts = OBS.reduce<Record<string, number>>((a, o) => ((a[o.status] = (a[o.status] || 0) + 1), a), {});
+  const fails = OBS.filter((o) => o.status === 'FAIL');
+  console.log('\n=========== CROSS-HOSPITAL ISOLATION SUMMARY ===========');
+  console.log('Total observations:', OBS.length, JSON.stringify(counts));
+  console.log('ISOLATION VIOLATIONS:', fails.length);
+  for (const f of fails) {
+    console.log(`  ❌ ${f.module}/${f.step}: ${f.detail}`);
+  }
+  console.log('========================================================\n');
+});