ixgadmin
cdefbd010c
Media was unreachable from either portal, which is why the certificate button on the buyer compliance page returned page-not-found and no document row had anything to click. The cause was structural: the portal and ops hosts proxy location / to the app with a path prefix, so a stored /media/... URL arrives as /portal/media/... and matches no route. Add a download view per portal behind the same role gate as the page that links to it — buyer-visible documents and published certificates for buyers, anything the ops list shows for operations — resolving the stored path safely under MEDIA_ROOT so a traversal or an absolute path is refused. Point the seven templates that list files at the new routes. Also add a portal-wide render sweep: nothing in the suite was fetching the ops or portal pages, so a view could lose its template_name and 500 in front of a user with every test still green.
209 lines
8.5 KiB
Python
209 lines
8.5 KiB
Python
"""Downloads must be gated the same way the page that links to them is.
|
|
|
|
Two classes of failure matter here and neither is caught by "the link renders":
|
|
a file that cannot be served at all (the state the portals were in), and a file
|
|
served to someone who should not have it. Both are covered below, plus the path
|
|
handling that turns a stored value into a real file.
|
|
"""
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import Client, TestCase, override_settings
|
|
|
|
from apps.batches.models import ProcessingBatch
|
|
|
|
from .downloads import resolve_media_path
|
|
from .models import CertificateOfAnalysis, Document
|
|
|
|
User = get_user_model()
|
|
|
|
PDF_BYTES = b"%PDF-1.4\n% synthetic document for tests\n%%EOF\n"
|
|
|
|
|
|
def make_user(email, role):
|
|
# Mirrors the Keycloak OIDC backend, which creates all users as is_staff=True.
|
|
return User.objects.create_user(
|
|
email=email, password="pass1234", role=role,
|
|
full_name=email.split("@")[0], is_staff=True,
|
|
)
|
|
|
|
|
|
class DownloadTestCase(TestCase):
|
|
"""Fixtures: a real temp media root, plus one user per role."""
|
|
|
|
def setUp(self):
|
|
self.media_root = Path(tempfile.mkdtemp(prefix="ixg-download-test-"))
|
|
self.addCleanup(shutil.rmtree, self.media_root, True)
|
|
self.settings_override = override_settings(MEDIA_ROOT=self.media_root)
|
|
self.settings_override.enable()
|
|
self.addCleanup(self.settings_override.disable)
|
|
|
|
self.buyer = make_user("buyer@ixg.local", "buyer")
|
|
self.ops = make_user("ops@ixg.local", "ops")
|
|
self.admin = make_user("admin@ixg.local", "admin")
|
|
|
|
self.batch = ProcessingBatch.objects.create(
|
|
batch_reference="IXG-BATCH-TEST-1",
|
|
grade="A",
|
|
weight_kg=Decimal("1000.00"),
|
|
origin_facility="ASI - Katsina State, Nigeria",
|
|
created_by=self.ops,
|
|
)
|
|
|
|
self.buyer_doc = self.make_document("packing-list.pdf", Document.Visibility.BUYER)
|
|
self.ops_doc = self.make_document("lab-report.pdf", Document.Visibility.OPS)
|
|
self.admin_doc = self.make_document("internal-note.pdf", Document.Visibility.ADMIN)
|
|
self.published_coa = self.make_coa("coa-published.pdf", CertificateOfAnalysis.Status.PUBLISHED)
|
|
self.pending_coa = self.make_coa("coa-pending.pdf", CertificateOfAnalysis.Status.PENDING_APPROVAL)
|
|
|
|
def write_file(self, name):
|
|
"""Put real bytes on disk under MEDIA_ROOT and return the stored file_url."""
|
|
rel = f"ixg-documents/{self.batch.batch_reference.lower()}/{name}"
|
|
target = self.media_root / rel
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(PDF_BYTES)
|
|
return f"/media/{rel}"
|
|
|
|
def make_document(self, name, visibility):
|
|
return Document.objects.create(
|
|
batch=self.batch,
|
|
doc_type=Document.DocType.OTHER,
|
|
file_url=self.write_file(name),
|
|
file_name=name,
|
|
visibility=visibility,
|
|
uploaded_by=self.ops,
|
|
)
|
|
|
|
def make_coa(self, name, status):
|
|
return CertificateOfAnalysis.objects.create(
|
|
batch=self.batch,
|
|
file_url=self.write_file(name),
|
|
file_name=name,
|
|
status=status,
|
|
uploaded_by=self.ops,
|
|
)
|
|
|
|
def get(self, user, url, host):
|
|
c = Client(SERVER_NAME=host)
|
|
c.force_login(user)
|
|
return c.get(url)
|
|
|
|
@staticmethod
|
|
def body(response):
|
|
if getattr(response, "streaming", False):
|
|
return b"".join(response.streaming_content)
|
|
return response.content
|
|
|
|
|
|
class BuyerDownloadTests(DownloadTestCase):
|
|
def test_buyer_downloads_a_buyer_visible_document(self):
|
|
r = self.get(self.buyer, f"/portal/documents/{self.buyer_doc.id}/download/", "portal.ixgltd.com")
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(self.body(r), PDF_BYTES)
|
|
self.assertIn("packing-list.pdf", r.headers.get("Content-Disposition", ""))
|
|
|
|
def test_buyer_cannot_download_an_operations_document(self):
|
|
r = self.get(self.buyer, f"/portal/documents/{self.ops_doc.id}/download/", "portal.ixgltd.com")
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
def test_buyer_cannot_download_an_admin_document(self):
|
|
r = self.get(self.buyer, f"/portal/documents/{self.admin_doc.id}/download/", "portal.ixgltd.com")
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
def test_buyer_downloads_a_published_certificate(self):
|
|
r = self.get(self.buyer, f"/portal/certificates/{self.published_coa.id}/download/", "portal.ixgltd.com")
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(self.body(r), PDF_BYTES)
|
|
|
|
def test_buyer_cannot_download_a_certificate_awaiting_release(self):
|
|
r = self.get(self.buyer, f"/portal/certificates/{self.pending_coa.id}/download/", "portal.ixgltd.com")
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
def test_anonymous_visitor_is_refused(self):
|
|
c = Client(SERVER_NAME="portal.ixgltd.com")
|
|
r = c.get(f"/portal/documents/{self.buyer_doc.id}/download/")
|
|
self.assertIn(r.status_code, (302, 403))
|
|
|
|
def test_ops_user_is_refused_on_the_buyer_route(self):
|
|
r = self.get(self.ops, f"/portal/documents/{self.buyer_doc.id}/download/", "portal.ixgltd.com")
|
|
self.assertEqual(r.status_code, 403)
|
|
|
|
|
|
class OpsDownloadTests(DownloadTestCase):
|
|
def test_ops_downloads_a_buyer_visible_document(self):
|
|
r = self.get(self.ops, f"/ops/documents/{self.buyer_doc.id}/download/", "ops.ixgltd.com")
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(self.body(r), PDF_BYTES)
|
|
|
|
def test_ops_downloads_an_operations_document(self):
|
|
r = self.get(self.ops, f"/ops/documents/{self.ops_doc.id}/download/", "ops.ixgltd.com")
|
|
self.assertEqual(r.status_code, 200)
|
|
|
|
def test_ops_can_open_a_certificate_awaiting_release(self):
|
|
"""A reviewer needs the pending certificate most of all."""
|
|
r = self.get(self.ops, f"/ops/certificates/{self.pending_coa.id}/download/", "ops.ixgltd.com")
|
|
self.assertEqual(r.status_code, 200)
|
|
self.assertEqual(self.body(r), PDF_BYTES)
|
|
|
|
def test_buyer_is_refused_on_the_ops_route(self):
|
|
r = self.get(self.buyer, f"/ops/documents/{self.ops_doc.id}/download/", "ops.ixgltd.com")
|
|
self.assertIn(r.status_code, (302, 403))
|
|
|
|
|
|
class MissingAndHostileFileTests(DownloadTestCase):
|
|
def test_missing_file_is_not_found_rather_than_a_server_error(self):
|
|
doc = Document.objects.create(
|
|
batch=self.batch,
|
|
doc_type=Document.DocType.OTHER,
|
|
file_url="/media/ixg-documents/nowhere/nothing-here.pdf",
|
|
file_name="nothing-here.pdf",
|
|
visibility=Document.Visibility.BUYER,
|
|
uploaded_by=self.ops,
|
|
)
|
|
r = self.get(self.buyer, f"/portal/documents/{doc.id}/download/", "portal.ixgltd.com")
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
def test_empty_file_url_is_not_found(self):
|
|
doc = Document.objects.create(
|
|
batch=self.batch,
|
|
doc_type=Document.DocType.OTHER,
|
|
file_url="",
|
|
file_name="empty.pdf",
|
|
visibility=Document.Visibility.BUYER,
|
|
uploaded_by=self.ops,
|
|
)
|
|
r = self.get(self.buyer, f"/portal/documents/{doc.id}/download/", "portal.ixgltd.com")
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
def test_a_traversal_out_of_media_root_is_refused(self):
|
|
doc = Document.objects.create(
|
|
batch=self.batch,
|
|
doc_type=Document.DocType.OTHER,
|
|
file_url="/media/../../config/settings.py",
|
|
file_name="escaped.pdf",
|
|
visibility=Document.Visibility.BUYER,
|
|
uploaded_by=self.ops,
|
|
)
|
|
r = self.get(self.buyer, f"/portal/documents/{doc.id}/download/", "portal.ixgltd.com")
|
|
self.assertEqual(r.status_code, 404)
|
|
|
|
|
|
class ResolveMediaPathTests(DownloadTestCase):
|
|
def test_resolves_a_normal_media_url(self):
|
|
path = resolve_media_path(self.buyer_doc.file_url)
|
|
self.assertTrue(path and path.is_file())
|
|
|
|
def test_refuses_paths_outside_media_url(self):
|
|
self.assertIsNone(resolve_media_path("/etc/passwd"))
|
|
self.assertIsNone(resolve_media_path("/app/config/settings.py"))
|
|
|
|
def test_refuses_traversal(self):
|
|
self.assertIsNone(resolve_media_path("/media/../config/settings.py"))
|
|
|
|
def test_refuses_empty_values(self):
|
|
self.assertIsNone(resolve_media_path(""))
|
|
self.assertIsNone(resolve_media_path(None))
|