Enforce read permissions throughout XLSX exports (218b70a7) · Commits · certic / Jama

src/resources/models.py

+47 −8

Changes for src/resources/models.py: 47 added lines, 8 removed lines.
Original line number	Diff line number	Diff line
		@@ -6,7 +6,7 @@ import os
		from jama import settings
		import hashlib
		import hmac
		from typing import Iterator, Union, List, Dict
		from typing import TYPE_CHECKING, Iterator, Union, List, Dict
		from django.db.models.signals import post_delete, post_save
		from django.dispatch import receiver
		import subprocess
		@@ -19,6 +19,10 @@ from functools import cache, cached_property
		from django.db.models.query import QuerySet
		from warnings import deprecated

		if TYPE_CHECKING:
		# ACL imports these models; keep this type-hint import out of runtime to avoid a cycle.
		from resources.acl import UserAccess

		Image.MAX_IMAGE_PIXELS = settings.JAMA_MAX_IMAGE_PIXELS

		XLSX_MULTIPLE_VALUES_SEPARATOR = "\n--\n"
		@@ -1087,7 +1091,7 @@ class Collection(models.Model):
		collections = {
		collection.pk: collection
		for collection in Collection.objects.filter(pk__in=collection_ids).only(
		"id", "parent_id", "title", "ark"
		"id", "parent_id", "project_id", "title", "ark"
		)
		}
		children_by_parent_id = defaultdict(list)
		@@ -1114,13 +1118,18 @@ class Collection(models.Model):
		self,
		metadata_ids: Union[List[int], None] = None,
		metadataset_ids: Union[List[int], None] = None,
		*,
		acl: "UserAccess \| None" = None,
		) -> Iterator[List[Union[int, str, None]]]:
		"""
		Stream XLSX rows for this collection subtree.

		Metadata values and resource memberships are fetched in batches and grouped
		by object id so row generation does not query once per collection/resource.
		When an ACL is provided, only readable collection and resource rows are exported.
		"""
		if acl is not None:
		acl.check_read(self)
		known_metadatas = list(
		self.project.metadatas(
		not settings.JAMA_XLSX_EXPORT_AUTOMATIC_METADATAS,
		@@ -1135,11 +1144,26 @@ class Collection(models.Model):
		}
		collection_rows = self._ordered_subtree_collections_for_export()
		collection_ids = [collection.pk for collection in collection_rows]
		readable_collection_rows = collection_rows
		if acl is not None:
		# Reuse the subtree and the export root's ancestors for every collection check.
		collections_by_id = {
		collection.pk: collection
		for collection in [collection_rows, self.ancestors()]
		}
		for collection in collection_rows:
		collection.parent = collections_by_id.get(collection.parent_id)
		readable_collection_rows = [
		collection for collection in collection_rows if acl.can_read(collection)
		]
		readable_collection_ids = [
		collection.pk for collection in readable_collection_rows
		]
		collection_values_by_collection_id = defaultdict(list)
		if collection_ids and metadata_column_indexes:
		if readable_collection_ids and metadata_column_indexes:
		for metadata_value in (
		MetadataCollectionValue.objects.filter(
		collection_id__in=collection_ids,
		collection_id__in=readable_collection_ids,
		metadata_id__in=metadata_column_indexes.keys(),
		)
		.exclude(metadata__title__in=XLSX_EXPORT_EXCLUDED_METADATA_TITLES)
		@@ -1152,7 +1176,7 @@ class Collection(models.Model):
		)

		memberships_by_collection_id = defaultdict(list)
		resource_ids = set()
		resources_by_id: dict[int, Resource] = {}
		if collection_ids:
		for membership in (
		CollectionMembership.objects.filter(
		@@ -1166,6 +1190,7 @@ class Collection(models.Model):
		"resource__id",
		"resource__title",
		"resource__ark",
		"resource__ptr_project_id",
		)
		.order_by("collection_id", "rank", "resource__title")
		.iterator()
		@@ -1173,7 +1198,17 @@ class Collection(models.Model):
		memberships_by_collection_id[membership.collection_id].append(
		membership
		)
		resource_ids.add(membership.resource_id)
		resources_by_id.setdefault(membership.resource_id, membership.resource)

		if acl is not None:
		acl.prefetch_resource_ancestors(list(resources_by_id.values()))
		resource_ids = {
		resource.pk
		for resource in resources_by_id.values()
		if acl.can_read(resource)
		}
		else:
		resource_ids = set(resources_by_id)

		resource_values_by_resource_id = defaultdict(list)
		if resource_ids and metadata_column_indexes:
		@@ -1195,7 +1230,7 @@ class Collection(models.Model):
		*XLSX_EXPORT_BASE_HEADERS,
		*[str(metadata) for metadata in known_metadatas],
		]
		for collection in collection_rows:
		for collection in readable_collection_rows:
		yield collection.metadatas_for_xlsx(
		collection_values_by_collection_id[collection.pk],
		metadata_column_indexes,
		@@ -1203,6 +1238,8 @@ class Collection(models.Model):
		)
		for collection in collection_rows:
		for membership in memberships_by_collection_id[collection.pk]:
		if membership.resource_id not in resource_ids:
		continue
		yield _flatten_resource(
		membership.resource,
		resource_values_by_resource_id[membership.resource_id],
		@@ -1215,6 +1252,8 @@ class Collection(models.Model):
		xlsx_path: str,
		metadata_ids: Union[List[int], None] = None,
		metadataset_ids: Union[List[int], None] = None,
		*,
		acl: "UserAccess \| None" = None,
		):
		from openpyxl import Workbook
		from openpyxl.cell import WriteOnlyCell
		@@ -1236,7 +1275,7 @@ class Collection(models.Model):
		wb = Workbook(write_only=True)
		ws = wb.create_sheet()
		for row in self.yield_resource_data_for_export(
		metadata_ids=metadata_ids, metadataset_ids=metadataset_ids
		metadata_ids=metadata_ids, metadataset_ids=metadataset_ids, acl=acl
		):
		append_row(row)
		wb.save(xlsx_path)

src/rpc/tests.py

+256 −1

Changes for src/rpc/tests.py: 256 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		from django.test import RequestFactory, TestCase
		from django.db import connection
		from django.test.utils import CaptureQueriesContext
		from django.contrib.auth.models import AbstractUser, AnonymousUser, User
		from rpc import methods
		from rpc import serializers
		@@ -6,7 +8,7 @@ from rpc import views
		from resources import models
		from resources.management.commands.loadfixtures import load_fixtures
		from rpc.methods import ServiceException
		from resources.acl import UserAccess
		from resources.acl import UserAccess, set_permission_on_role
		from pathlib import Path
		import os
		import base64
		@@ -35,6 +37,259 @@ object_classes = [
		crud = ["create", "read", "update", "delete"]


		class XlsxExportAccessTestCase(TestCase):
		@classmethod
		def setUpTestData(cls):
		cls.user = User.objects.create(username="xlsx_reader")
		cls.project = models.Project.objects.create(label="xlsx access", description="")
		cls.root = models.Collection.objects.create(project=cls.project, title="Root")
		cls.ancestor = models.Collection.objects.create(
		project=cls.project, parent=cls.root, title="Ancestor"
		)
		cls.collection = models.Collection.objects.create(
		project=cls.project, parent=cls.ancestor, title="Exported"
		)
		cls.role = models.Role.objects.create(project=cls.project, label="reader")
		models.ProjectAccess.objects.create(
		project=cls.project, user=cls.user, role=cls.role
		)
		set_permission_on_role(cls.role, "collection", can_read=True)
		set_permission_on_role(cls.role, "resource", can_read=True)
		cls.metadata_set = models.MetadataSet.objects.create(
		project=cls.project, title="Manual"
		)
		cls.metadata = models.Metadata.objects.create(
		project=cls.project, set=cls.metadata_set, title="description"
		)
		cls.file_type = models.FileType.objects.create(title="Text", mime="text/plain")

		def setUp(self):
		self.client.force_login(self.user)

		def _collection(self, title: str, parent: models.Collection) -> models.Collection:
		collection = models.Collection.objects.create(
		project=self.project, parent=parent, title=title
		)
		models.MetadataCollectionValue.objects.create(
		metadata=self.metadata, collection=collection, value=f"metadata for {title}"
		)
		return collection

		def _resource(
		self, title: str, collection: models.Collection, *, file_hash: str \| None = None
		) -> models.Resource:
		if file_hash:
		resource = models.File.objects.create(
		project=self.project,
		title=title,
		original_name=f"{title}.txt",
		hash=file_hash,
		file_type=self.file_type,
		size=1,
		)
		else:
		resource = models.Resource.objects.create(
		ptr_project=self.project, title=title
		)
		collection.resources.add(resource)
		models.MetadataResourceValue.objects.create(
		metadata=self.metadata, resource=resource, value=f"metadata for {title}"
		)
		return resource

		def _download_rows(self) -> list[tuple]:
		response = self.client.get(
		reverse("metas_download", kwargs={"collection_id": self.collection.pk})
		)
		self.assertEqual(response.status_code, 200)
		try:
		workbook = load_workbook(BytesIO(b"".join(response.streaming_content)))
		try:
		return list(workbook.active.values)
		finally:
		workbook.close()
		finally:
		response.close()

		def test_excludes_denied_collections_resources_and_metadata(self):
		allowed_collection = self._collection("Allowed collection", self.collection)
		allowed_resource = self._resource("Allowed resource", allowed_collection)
		denied_collection = self._collection("Secret collection", self.collection)
		set_permission_on_role(self.role, denied_collection, can_read=False)
		denied_child = self._collection("Secret descendant", denied_collection)
		denied_descendant_resource = self._resource(
		"Secret descendant resource", denied_child
		)
		denied_resource = self._resource("Secret resource", allowed_collection)
		set_permission_on_role(self.role, denied_resource, can_read=False)
		denied_file = self._resource("Secret file", self.collection, file_hash="a" * 64)
		set_permission_on_role(self.role, denied_file, can_read=False)

		rows = self._download_rows()

		self.assertEqual(
		{row[1] for row in rows[1:] if row[1]},
		{self.collection.pk, allowed_collection.pk},
		)
		self.assertEqual({row[0] for row in rows[1:] if row[0]}, {allowed_resource.pk})
		metadata_index = rows[0].index(str(self.metadata))
		self.assertIn(
		f"metadata for {allowed_resource.title}",
		[row[metadata_index] for row in rows[1:]],
		)
		for denied in (
		denied_collection,
		denied_child,
		denied_descendant_resource,
		denied_resource,
		denied_file,
		):
		self.assertFalse(
		any(
		denied.title in str(value) for row in rows for value in row if value
		)
		)

		def test_explicit_grants_inside_denied_collection_remain_exportable(self):
		denied_collection = self._collection("Denied collection", self.collection)
		set_permission_on_role(self.role, denied_collection, can_read=False)
		allowed_child = self._collection("Allowed child", denied_collection)
		set_permission_on_role(self.role, allowed_child, can_read=True)
		child_resource = self._resource("Child resource", allowed_child)
		allowed_resource = self._resource(
		"Explicitly allowed resource", denied_collection
		)
		set_permission_on_role(self.role, allowed_resource, can_read=True)

		rows = self._download_rows()

		self.assertEqual(
		{row[1] for row in rows[1:] if row[1]},
		{self.collection.pk, allowed_child.pk},
		)
		self.assertEqual(
		{row[0] for row in rows[1:] if row[0]},
		{child_resource.pk, allowed_resource.pk},
		)

		def test_inherited_grants_from_ancestors_outside_export_subtree(self):
		set_permission_on_role(self.role, "resource", can_read=False)
		set_permission_on_role(self.role, self.ancestor, can_read=True)
		child = self._collection("Child", self.collection)
		resource = self._resource("Inherited access", child)

		rows = self._download_rows()

		self.assertEqual({row[0] for row in rows[1:] if row[0]}, {resource.pk})
		self.assertEqual(
		{row[1] for row in rows[1:] if row[1]}, {self.collection.pk, child.pk}
		)

		def test_respects_permissions_from_memberships_outside_export_subtree(self):
		outside = self._collection("A denied outside collection", self.root)
		set_permission_on_role(self.role, outside, can_read=False)
		resource = self._resource("Shared but denied", self.collection)
		outside.resources.add(resource)
		self.assertFalse(UserAccess(self.user, self.project).can_read(resource))

		rows = self._download_rows()

		self.assertFalse(any(row[0] == resource.pk for row in rows[1:]))

		def test_excludes_resources_from_another_project(self):
		other_project = models.Project.objects.create(
		label="other xlsx", description=""
		)
		resource = models.Resource.objects.create(
		ptr_project=other_project, title="Foreign resource"
		)
		self.collection.resources.add(resource)

		rows = self._download_rows()

		self.assertFalse(any(row[0] == resource.pk for row in rows[1:]))

		def test_preserves_order_multiple_values_and_repeated_memberships(self):
		first_child = self._collection("A child", self.collection)
		second_child = self._collection("B child", self.collection)
		first = self._resource("Z first", self.collection, file_hash="a" * 64)
		second = self._resource("A second", self.collection)
		models.CollectionMembership.objects.filter(resource=first).update(rank=1)
		models.CollectionMembership.objects.filter(resource=second).update(rank=2)
		shared = self._resource("Shared", first_child)
		second_child.resources.add(shared)
		models.MetadataResourceValue.objects.create(
		metadata=self.metadata, resource=shared, value="another value"
		)

		rows = self._download_rows()

		self.assertEqual(
		[row[1] for row in rows[1:] if row[1]],
		[self.collection.pk, first_child.pk, second_child.pk],
		)
		resource_rows = [row for row in rows[1:] if row[0]]
		self.assertEqual(
		[row[0] for row in resource_rows],
		[first.pk, second.pk, shared.pk, shared.pk],
		)
		metadata_index = rows[0].index(str(self.metadata))
		self.assertEqual(
		resource_rows[-1][metadata_index],
		models.XLSX_MULTIPLE_VALUES_SEPARATOR.join(
		[f"metadata for {shared.title}", "another value"]
		),
		)

		def test_denied_export_root_returns_forbidden(self):
		set_permission_on_role(self.role, self.collection, can_read=False)

		response = self.client.get(
		reverse("metas_download", kwargs={"collection_id": self.collection.pk})
		)

		self.assertEqual(response.status_code, 403)
		with self.assertRaises(ServiceException):
		list(
		self.collection.yield_resource_data_for_export(
		acl=UserAccess(self.user, self.project)
		)
		)

		def test_trusted_export_without_acl_retains_all_rows(self):
		denied = self._resource("Denied resource", self.collection)
		set_permission_on_role(self.role, denied, can_read=False)

		rows = list(self.collection.yield_resource_data_for_export())

		self.assertIn(denied.pk, [row[0] for row in rows[1:]])

		def test_query_count_does_not_grow_with_collections_and_resources(self):
		for scoped in (False, True):
		with self.subTest(scoped=scoped):
		self.collection.collection_set.all().delete()
		self.collection.resources.clear()
		models.ObjectPermission.objects.filter(
		role=self.role, object_pk__isnull=False
		).delete()
		set_permission_on_role(self.role, "resource", can_read=not scoped)
		if scoped:
		set_permission_on_role(self.role, self.ancestor, can_read=True)
		child = self._collection(f"First {scoped}", self.collection)
		self._resource(f"First resource {scoped}", child)
		with CaptureQueriesContext(connection) as small_queries:
		small_rows = self._download_rows()
		for index in range(20):
		child = self._collection(f"Extra {scoped} {index}", self.collection)
		self._resource(f"Extra resource {scoped} {index}", child)
		with CaptureQueriesContext(connection) as large_queries:
		large_rows = self._download_rows()

		self.assertEqual(len(small_rows), 4)
		self.assertEqual(len(large_rows), 44)
		self.assertEqual(len(small_queries), len(large_queries))


		class ServiceTestCase(TestCase):
		@classmethod
		def setUpTestData(cls):

src/rpc/views.py

+6 −1

Changes for src/rpc/views.py: 6 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -942,7 +942,11 @@ def download_metas_xls(request: HttpRequest, collection_id):
		user = _get_user_from_request(request)
		if not user:
		return HttpResponse("Forbidden", status=403)
		UserAccess(user, collection_instance.project).check_read(collection_instance)
		acl = UserAccess(user, collection_instance.project)
		try:
		acl.check_read(collection_instance)
		except ServiceException:
		return HttpResponse("Forbidden", status=403)
		try:
		metadata_ids = _parse_int_list_query_param(request, "metadata_ids")
		metadataset_ids = _parse_int_list_query_param(request, "metadataset_ids")
		@@ -954,6 +958,7 @@ def download_metas_xls(request: HttpRequest, collection_id):
		tmp.name,
		metadata_ids=metadata_ids,
		metadataset_ids=metadataset_ids,
		acl=acl,
		)
		return RangedFileResponse(
		request,