"""Cross-tenant authorization tests for the media download view. The bug these lock down: MEDIA_ROOT was served with no authentication and no upload_to path carried an organization prefix, so one tenant's HR contracts and bank statements were readable by anyone holding (or guessing) the filename. """ import pytest from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile from django.test import Client from django.urls import reverse from apps.core.uploads import _id_segment from apps.disseminations.models import MediaAsset from apps.companies.models import Organization, OrganizationMembership User = get_user_model() def client_for(user): """Django's Client, not DRF's APIClient. serve_media is a plain Django view, and APIClient.force_authenticate only applies to DRF views — request.user would stay anonymous and every request would 404 for the wrong reason. """ client = Client() client.force_login(user) return client @pytest.fixture def two_orgs(db): from apps.hr.signals import ensure_employee_stub org_a = Organization.objects.create(name="Media Org A") org_b = Organization.objects.create(name="Media Org B") user_a = User.objects.create_user(email="a@media.test", password="pw12345!") user_b = User.objects.create_user(email="b@media.test", password="pw12345!") OrganizationMembership.objects.create(organization=org_a, user=user_a) OrganizationMembership.objects.create(organization=org_b, user=user_b) ensure_employee_stub(user_a) ensure_employee_stub(user_b) return org_a, org_b, user_a, user_b @pytest.fixture def receipt_in_org_a(two_orgs): """A MediaAsset (direct organization FK) owned by org A.""" org_a = two_orgs[0] return MediaAsset.objects.create( organization=org_a, file=SimpleUploadedFile("secret-receipt.pdf", b"%PDF-1.4 org-a-only"), ) def media_url(name): return reverse("serve-media", kwargs={"path": name}) @pytest.mark.django_db class TestMediaTenantIsolation: def test_owner_can_read_own_file(self, two_orgs, receipt_in_org_a): user_a = two_orgs[2] res = client_for(user_a).get(media_url(receipt_in_org_a.file.name)) assert res.status_code == 200 assert b"org-a-only" in b"".join(res.streaming_content) def test_other_org_is_refused(self, two_orgs, receipt_in_org_a): """The whole point: org B holds a valid path and still cannot read it.""" user_b = two_orgs[3] res = client_for(user_b).get(media_url(receipt_in_org_a.file.name)) assert res.status_code == 403 def test_anonymous_is_404_not_401(self, receipt_in_org_a): """401 would confirm the path exists; the filename is the only secret.""" res = Client().get(media_url(receipt_in_org_a.file.name)) assert res.status_code == 404 def test_superuser_may_read_across_orgs(self, two_orgs, receipt_in_org_a): admin = User.objects.create_superuser(email="root@media.test", password="pw12345!") res = client_for(admin).get(media_url(receipt_in_org_a.file.name)) assert res.status_code == 200 def test_unowned_file_is_404(self, two_orgs): """Fails closed: a path with no owning row has nothing to authorize.""" res = client_for(two_orgs[2]).get(media_url("orphaned/nobody.pdf")) assert res.status_code == 404 @pytest.mark.parametrize("evil", [ "../../etc/passwd", "..%2f..%2fetc%2fpasswd", "subdir/../../../secrets.env", ]) def test_path_traversal_refused(self, two_orgs, evil): res = client_for(two_orgs[2]).get(f"/api/media/{evil}") assert res.status_code in (403, 404) def test_response_is_not_cacheable_by_shared_caches(self, two_orgs, receipt_in_org_a): """A shared cache holding the body would serve it to the next requester.""" res = client_for(two_orgs[2]).get(media_url(receipt_in_org_a.file.name)) assert "no-store" in res["Cache-Control"] assert "private" in res["Cache-Control"] def test_write_methods_rejected(self, two_orgs, receipt_in_org_a): res = client_for(two_orgs[2]).post(media_url(receipt_in_org_a.file.name)) assert res.status_code == 405 @pytest.mark.django_db class TestOrganizationResolution: def test_resolves_via_direct_fk(self, receipt_in_org_a, two_orgs): from apps.core.tenancy import organization_of assert organization_of(receipt_in_org_a).id == two_orgs[0].id def test_uploader_alone_does_not_grant_access(self, two_orgs): """An uploader FK must not be treated as ownership. Otherwise anyone who once uploaded a file keeps reading it after moving organizations. """ from apps.core.tenancy import organization_of from apps.core.models import Attachment org_a, _, user_a, _ = two_orgs att = Attachment.objects.create( organization=org_a, file=SimpleUploadedFile("a.txt", b"x"), uploaded_by=user_a, name="a.txt", size=1, mime_type="text/plain", ) # Resolves through the real organization FK, not through uploaded_by. assert organization_of(att).id == org_a.id def test_charfield_named_organization_is_not_an_org(self, two_orgs): """EventSpeaker.organization names the speaker's employer, as a string. Resolution must type-check rather than trust the attribute name, or the walk returns a str and every caller that touches `.id` breaks — while a model with no real tenant link silently looks owned. """ from apps.core.tenancy import organization_of from apps.events.models import Event, EventSpeaker from django.utils import timezone org_a = two_orgs[0] now = timezone.now() event = Event.objects.create( organization=org_a, name="Summit", start_date=now, end_date=now, organizer=two_orgs[2], ) speaker = EventSpeaker.objects.create( event=event, name="Dr Who", organization="Some Other Institute", ) assert organization_of(speaker).id == org_a.id def test_resolves_through_generic_foreign_key(self, two_orgs): """SubmissionVersion reaches its tenant only via a generic relation.""" from apps.core.tenancy import organization_of from apps.projects.models import Project, ProjectFund, FundReport, SubmissionVersion org_a = two_orgs[0] project = Project.objects.create(organization=org_a, name="Grant Project") fund = ProjectFund.objects.create( organization=org_a, project=project, source="Donor", amount=1000, ) report = FundReport.objects.create(fund=fund, title="Q1") version = SubmissionVersion.objects.create( content_object=report, file=SimpleUploadedFile("v1.pdf", b"%PDF-1.4 v1"), ) # Normalized before comparing: org_a holds the dashless hex string that # companies.uuid7() assigned in memory, while the value reached through # the relation is a real UUID. Same id, two renderings. assert _id_segment(organization_of(version).id) == _id_segment(org_a.id) assert version.file.name.startswith( f"org/{_id_segment(org_a.id)}/submission-versions/" ) def test_organization_owns_itself(self, two_orgs): from apps.core.tenancy import organization_of org_a = two_orgs[0] assert organization_of(org_a).id == org_a.id def test_prefix_is_identical_in_memory_and_from_the_database(self, two_orgs): """One tenant must have exactly one prefix. uuid7() returns a dashless str in most apps and a uuid.UUID in apps.core, so a raw str(pk) gives the dashless form for a row still in memory and the dashed form once it is read back. That would scatter a single org's files across two directories and defeat a prefix-scoped R2 credential. """ org_a = two_orgs[0] in_memory = MediaAsset.objects.create( organization=org_a, file=SimpleUploadedFile("m.pdf", b"a"), ) reloaded_org = Organization.objects.get(pk=org_a.pk) from_db = MediaAsset.objects.create( organization=reloaded_org, file=SimpleUploadedFile("d.pdf", b"b"), ) assert in_memory.file.name.rsplit("/", 1)[0] == from_db.file.name.rsplit("/", 1)[0] assert "-" not in in_memory.file.name.split("/")[1] @pytest.mark.django_db class TestTenantUploadPaths: """Every new upload lands under a prefix naming its tenant.""" def test_direct_fk_file_is_org_prefixed(self, receipt_in_org_a, two_orgs): org_a = two_orgs[0] assert receipt_in_org_a.file.name.startswith(f"org/{_id_segment(org_a.id)}/dissemination-media/") def test_original_filename_is_not_preserved(self, receipt_in_org_a): """The stem is the only secret in front of a presigned R2 URL.""" assert "secret-receipt" not in receipt_in_org_a.file.name assert receipt_in_org_a.file.name.endswith(".pdf") def test_two_orgs_do_not_share_a_directory(self, two_orgs): org_a, org_b, _, _ = two_orgs a = MediaAsset.objects.create( organization=org_a, file=SimpleUploadedFile("x.pdf", b"a"), ) b = MediaAsset.objects.create( organization=org_b, file=SimpleUploadedFile("x.pdf", b"b"), ) assert a.file.name.rsplit("/", 1)[0] != b.file.name.rsplit("/", 1)[0] def test_organization_branding_uses_its_own_id(self, db): """Organization has no `organization` field; its own pk is the prefix.""" org = Organization.objects.create( name="Branded", logo=SimpleUploadedFile("logo.png", b"\x89PNG"), ) assert org.logo.name.startswith(f"org/{_id_segment(org.id)}/branding/") def test_user_avatar_is_not_filed_under_an_org(self, two_orgs): """A user may belong to several orgs, so the avatar is not org-owned. Filing it under whichever membership came first would strand it there: serve_media would refuse the user their own avatar once they switched. """ user_a = two_orgs[2] user_a.avatar = SimpleUploadedFile("me.png", b"\x89PNG") user_a.save() assert user_a.avatar.name.startswith(f"users/{_id_segment(user_a.id)}/avatar/") assert not user_a.avatar.name.startswith("org/") def test_unresolvable_org_falls_back_rather_than_raising(self, db): """upload_to runs before save, so a parent FK may still be unset. Raising would turn an ordering detail into a user-facing 500; the file lands in a sweepable prefix that is no tenant's directory instead. """ from apps.core.uploads import UNASSIGNED_PREFIX, tenant_upload_path class Orphan: pk = "no-parent" def __init__(self): self._meta = type("M", (), {"get_fields": staticmethod(lambda: [])})() name = tenant_upload_path("strays")(Orphan(), "thing.pdf") assert name.startswith(f"org/{UNASSIGNED_PREFIX}/strays/")