add missing ACL checks in RPC (608ece3f) · Commits · certic / Jama

src/resources/acl.py

+19 −0

Original line number	Diff line number	Diff line
		@@ -308,6 +308,25 @@ class UserAccess:
		def can_delete(self, object_or_class: AccessControlledObject) -> bool:
		return self._can_crud(object_or_class, CRUD_DELETE)

		def has_project_access(self) -> bool:
		return (
		self.user.is_active
		and ProjectAccess.objects.filter(
		project=self.project,
		user=self.user,
		role__project=self.project,
		).exists()
		)

		def check_project_access(self) -> None:
		if not self.has_project_access():
		logger.info(
		REJECTION_LOGGING_MESSAGE.format(
		"project", self.user.username, self.project
		)
		)
		raise ServiceException(NO_ACCESS.format("project", self.project))

		def check_create(self, object_or_class):
		"""
		Check if user can create item and raise ServiceException if not

src/rpc/methods.py

+24 −7

Original line number	Diff line number	Diff line
		@@ -2332,6 +2332,8 @@ def set_representative_resource(
		collection_instance = Collection.objects.get(
		pk=collection_id, deleted_at__isnull=True
		)
		acl = UserAccess(user, collection_instance.project)
		acl.check_update(collection_instance)
		if not resource_id:
		collection_instance.representative = None
		collection_instance.save()
		@@ -2339,9 +2341,7 @@ def set_representative_resource(
		resource_instance = Resource.objects.get(
		pk=resource_id, deleted_at__isnull=True
		)
		acl = UserAccess(user, resource_instance.ptr_project)
		acl.check_read(resource_instance)
		acl.check_update(collection_instance)
		if resource_instance.ptr_project_id == collection_instance.project_id:
		collection_instance.representative = resource_instance
		collection_instance.save()
		@@ -2723,6 +2723,7 @@ def project_stats(user: User, project_id: int) -> dict:
		try:
		project = Project.objects.get(pk=project_id)
		project_root = project.root_collection
		UserAccess(user, project).check_project_access()
		return {
		"project_root_collection_id": project_root.pk,
		"descendants_count": Collection.objects.filter(
		@@ -2784,16 +2785,23 @@ def recycle_bin(user: User, project_id: int) -> List[Dict]:
		- deleted_at
		"""

		date_limit = timezone.now() - timedelta(days=15)
		project = Project.objects.filter(pk=project_id).first()
		if not project:
		raise ServiceException(NO_SUCH_PROJECT)
		acl = UserAccess(user, project)
		acl.check_project_access()

		date_limit = timezone.now() - timedelta(days=15)
		results = []
		for collection_instance in (
		Collection.objects.filter(
		deleted_at__isnull=False, project_id=project_id, deleted_at__gt=date_limit
		deleted_at__isnull=False, project=project, deleted_at__gt=date_limit
		)
		.only("id", "title", "deleted_at")
		.only("id", "title", "deleted_at", "project_id", "parent_id")
		.order_by("-deleted_at")
		):
		if not acl.can_read(collection_instance):
		continue
		results.append(
		{
		"object_type": "collection",
		@@ -2805,12 +2813,14 @@ def recycle_bin(user: User, project_id: int) -> List[Dict]:
		for resource_instance in (
		Resource.objects.filter(
		deleted_at__isnull=False,
		ptr_project_id=project_id,
		ptr_project=project,
		deleted_at__gt=date_limit,
		)
		.only("id", "title", "deleted_at")
		.only("id", "title", "deleted_at", "ptr_project_id")
		.order_by("-deleted_at")
		):
		if not acl.can_read(resource_instance):
		continue
		results.append(
		{
		"object_type": "resource",
		@@ -3316,6 +3326,7 @@ def auto_find_rotate_angle(user: User, resource_id: int) -> float:
		).first()
		if not resource_instance:
		raise ServiceException(NO_SUCH_RESOURCE)
		UserAccess(user, resource_instance.ptr_project).check_read(resource_instance)
		if not resource_instance.file:
		raise ServiceException(NOT_A_FILE)
		if not resource_instance.file.should_have_iiif:
		@@ -3722,9 +3733,12 @@ def collection_access_for_role(user: User, collection_id: int, role_id: int) ->
		collection_instance = Collection.objects.filter(pk=collection_id).first()
		if not collection_instance:
		raise ServiceException(NO_SUCH_COLLECTION)
		UserAccess(user, collection_instance.project).check_project_access()
		role_instance = Role.objects.filter(pk=role_id).first()
		if not role_instance:
		raise ServiceException("no such role")
		if role_instance.project_id != collection_instance.project_id:
		raise ServiceException(PROJECT_MISMATCH)
		perms = list(ObjectPermission.objects.filter(role=role_instance))
		return {
		"object_type": "permission",
		@@ -3744,9 +3758,12 @@ def resource_access_for_role(user: User, resource_id: int, role_id: int) -> dict
		).first()
		if not resource_instance:
		raise ServiceException(NO_SUCH_RESOURCE)
		UserAccess(user, resource_instance.ptr_project).check_project_access()
		role_instance = Role.objects.filter(pk=role_id).first()
		if not role_instance:
		raise ServiceException("no such role")
		if role_instance.project_id != resource_instance.ptr_project_id:
		raise ServiceException(PROJECT_MISMATCH)
		perms = list(ObjectPermission.objects.filter(role=role_instance))
		return {
		"object_type": "permission",

src/rpc/tests.py

+186 −1

Original line number	Diff line number	Diff line
		@@ -1056,6 +1056,147 @@ class ServiceTestCase(TestCase):
		models.Project.objects.filter(label="inactive admin project").exists()
		)

		def test_access_for_role_requires_project_membership_and_matching_project(self):
		resource = models.Resource.objects.create(
		title="role access resource",
		ptr_project=self.test_project,
		)
		no_access_user = User.objects.create(username="no_role_access")

		with self.assertRaises(ServiceException):
		methods.collection_access_for_role(
		no_access_user,
		self.test_project_root_collection.pk,
		self.admin_role.pk,
		)
		with self.assertRaises(ServiceException):
		methods.resource_access_for_role(
		no_access_user,
		resource.pk,
		self.admin_role.pk,
		)

		project_member = User.objects.create(username="role_access_member")
		membership_role = models.Role.objects.create(
		label="role access membership",
		project=self.test_project,
		)
		models.ProjectAccess.objects.create(
		project=self.test_project,
		user=project_member,
		role=membership_role,
		)
		collection_access = methods.collection_access_for_role(
		project_member,
		self.test_project_root_collection.pk,
		self.admin_role.pk,
		)
		resource_access = methods.resource_access_for_role(
		project_member,
		resource.pk,
		self.admin_role.pk,
		)
		self.assertTrue(collection_access["object_read"])
		self.assertTrue(resource_access["object_read"])

		other_project = models.Project.objects.create(label="other role project")
		other_role = models.Role.objects.create(
		label="other project role",
		project=other_project,
		)
		with self.assertRaises(ServiceException) as context:
		methods.collection_access_for_role(
		project_member,
		self.test_project_root_collection.pk,
		other_role.pk,
		)
		self.assertEqual(context.exception.message, methods.PROJECT_MISMATCH)

		def test_project_stats_requires_project_membership(self):
		no_access_user = User.objects.create(username="no_project_stats_access")

		with self.assertRaises(ServiceException):
		methods.project_stats(no_access_user, self.test_project.pk)

		mismatched_user = User.objects.create(username="mismatched_project_member")
		other_project = models.Project.objects.create(label="membership role owner")
		mismatched_role = models.Role.objects.create(
		label="foreign membership role",
		project=other_project,
		)
		models.ProjectAccess.objects.create(
		project=self.test_project,
		user=mismatched_user,
		role=mismatched_role,
		)
		with self.assertRaises(ServiceException):
		methods.project_stats(mismatched_user, self.test_project.pk)

		project_member = User.objects.create(username="project_stats_member")
		membership_role = models.Role.objects.create(
		label="project stats membership",
		project=self.test_project,
		)
		models.ProjectAccess.objects.create(
		project=self.test_project,
		user=project_member,
		role=membership_role,
		)

		stats = methods.project_stats(project_member, self.test_project.pk)
		self.assertEqual(
		stats["project_root_collection_id"],
		self.test_project_root_collection.pk,
		)

		def test_auto_find_rotate_angle_requires_resource_read_access(self):
		file_instance = self._public_download_test_file()
		no_access_user = User.objects.create(username="no_deskew_access")

		with patch.object(methods, "_deskew") as deskew:
		with self.assertRaises(ServiceException):
		methods.auto_find_rotate_angle(no_access_user, file_instance.pk)

		deskew.assert_not_called()
		with patch.object(methods, "_deskew", return_value=1.5):
		self.assertEqual(
		methods.auto_find_rotate_angle(self.test_user, file_instance.pk),
		1.5,
		)

		def test_clear_representative_requires_collection_update_access(self):
		resource = models.Resource.objects.create(
		title="representative resource",
		ptr_project=self.test_project,
		)
		collection = models.Collection.objects.create(
		title="represented collection",
		parent=self.test_project_root_collection,
		project=self.test_project,
		representative=resource,
		)
		no_access_user = User.objects.create(username="no_representative_access")

		with self.assertRaises(ServiceException):
		methods.set_representative_resource(
		no_access_user,
		collection.pk,
		resource_id=None,
		)

		collection.refresh_from_db()
		self.assertEqual(collection.representative_id, resource.pk)

		self.assertTrue(
		methods.set_representative_resource(
		self.test_user,
		collection.pk,
		resource_id=None,
		)
		)
		collection.refresh_from_db()
		self.assertIsNone(collection.representative_id)

		def test_resources_bad_order_by(self):
		collection = methods.add_collection(
		self.test_user,
		@@ -1078,8 +1219,52 @@ class ServiceTestCase(TestCase):
		self.test_user, "delete me", self.test_project.root_collection.pk
		)
		methods.delete_collection(self.test_user, col["id"])
		hidden_collection = models.Collection.objects.create(
		title="hidden recycled collection",
		parent=self.test_project_root_collection,
		project=self.test_project,
		)
		hidden_collection.soft_delete()
		models.ObjectPermission.objects.create(
		object_class="collection",
		object_pk=hidden_collection.pk,
		role=self.admin_role,
		object_read=False,
		)

		recycled_items = methods.recycle_bin(self.test_user, self.test_project.pk)

		self.assertEqual(
		[item["id"] for item in recycled_items],
		[col["id"]],
		)

		def test_recycle_bin_requires_project_membership(self):
		collection = models.Collection.objects.create(
		title="private recycled collection",
		parent=self.test_project_root_collection,
		project=self.test_project,
		)
		collection.soft_delete()
		no_access_user = User.objects.create(username="no_recycle_bin_access")

		with self.assertRaises(ServiceException):
		methods.recycle_bin(no_access_user, self.test_project.pk)

		project_member = User.objects.create(username="recycle_bin_member")
		membership_role = models.Role.objects.create(
		label="recycle bin membership",
		project=self.test_project,
		)
		models.ProjectAccess.objects.create(
		project=self.test_project,
		user=project_member,
		role=membership_role,
		)

		self.assertEqual(
		len(methods.recycle_bin(self.test_user, self.test_project.pk)), 1
		methods.recycle_bin(project_member, self.test_project.pk),
		[],
		)

		def test_search(self):