ismail
e5705a1b1c
All checks were successful
Build and Push Docker Image / build (push) Successful in 2m8s
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.
331 lines
15 KiB
TypeScript
331 lines
15 KiB
TypeScript
/* 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');
|
|
});
|