tighter ACL checks (38812c44) · Commits · certic / Jama

src/resources/acl.py

+61 −31

Original line number	Diff line number	Diff line
		@@ -41,6 +41,16 @@ CRUD_READ_VERBS = [CRUD_READ]
		CRUD_WRITE_VERBS = [CRUD_CREATE, CRUD_UPDATE, CRUD_DELETE]
		CRUD_VERBS = [CRUD_CREATE, CRUD_READ, CRUD_UPDATE, CRUD_DELETE]
		REJECTION_LOGGING_MESSAGE = "ACL rejected {} access to user {} for item {}"
		AccessControlledObject = Union[
		str,
		Resource,
		Collection,
		File,
		Metadata,
		MetadataSet,
		Tag,
		ProjectProperty,
		]


		def _climb_tree(
		@@ -54,7 +64,7 @@ def _climb_tree(
		yield collec


		def _class_name_from_instance(object_instance: Union[str, Resource, Collection]) -> str:
		def _class_name_from_instance(object_instance: AccessControlledObject) -> str:
		# if str, considered to be already a class name
		if type(object_instance) is str:
		return object_instance
		@@ -77,11 +87,35 @@ def _class_name_from_instance(object_instance: Union[str, Resource, Collection])
		raise ValueError(f"Unsupported object: {object_instance}")


		def _crud_val(permission: ObjectPermission, crud_verb: str) -> bool:
		if crud_verb not in CRUD_VERBS:
		def _project_ids_from_instance(
		object_instance: AccessControlledObject,
		) -> tuple[int \| None, ...]:
		"""Return every project reference that constrains an ACL object."""
		if type(object_instance) is str:
		return ()
		if isinstance(object_instance, File):
		return object_instance.project_id, object_instance.ptr_project_id
		if isinstance(object_instance, Resource):
		return (object_instance.ptr_project_id,)
		if isinstance(
		object_instance,
		(Collection, Metadata, MetadataSet, ProjectProperty, Tag),
		):
		return (object_instance.project_id,)
		raise ValueError(f"Unsupported object: {object_instance}")


		def _normalize_crud_verb(crud_access: str) -> str:
		crud_access = str(crud_access).lower().strip()
		if crud_access not in CRUD_VERBS:
		raise ValueError(
		f"not a crud access right: {crud_verb}. Must be one of {', '.join(CRUD_VERBS)}."
		f"not a crud access right: {crud_access}. Must be one of {', '.join(CRUD_VERBS)}."
		)
		return crud_access


		def _crud_val(permission: ObjectPermission, crud_verb: str) -> bool:
		crud_verb = _normalize_crud_verb(crud_verb)
		if crud_verb == CRUD_CREATE:
		return permission.object_create
		if crud_verb == CRUD_READ:
		@@ -95,14 +129,10 @@ def _crud_val(permission: ObjectPermission, crud_verb: str) -> bool:

		def has_crud_access(
		user_permissions: List[ObjectPermission],
		object_instance: Union[str, Resource, Collection],
		object_instance: AccessControlledObject,
		crud_access: str,
		) -> bool:
		crud_access = str(crud_access).lower().strip()
		if crud_access not in CRUD_VERBS:
		raise ValueError(
		f"not a crud access right: {crud_access}. Must be one of {', '.join(CRUD_VERBS)}."
		)
		crud_access = _normalize_crud_verb(crud_access)
		# Only Resource and Collection are tested for instance-level permissions
		if len(
		[x.object_pk for x in user_permissions if x.object_pk is not None]
		@@ -242,43 +272,43 @@ class UserAccess:
		if self._permissions is None:
		self._permissions = []
		for perm in ObjectPermission.objects.filter(
		role__project=self.project,
		role__projectaccess__user=self.user,
		role__projectaccess__project=self.project,
		):
		self._permissions.append(perm)
		return self._permissions

		def can_crud(self, object_or_class, crud_access) -> bool:
		def _can_crud(
		self, object_or_class: AccessControlledObject, crud_access: str
		) -> bool:
		crud_access = _normalize_crud_verb(crud_access)
		project_ids = _project_ids_from_instance(object_or_class)
		if any(project_id != self.project.pk for project_id in project_ids):
		return False
		return (
		has_crud_access(self.permissions, object_or_class, crud_access)
		and self.user.is_active
		)

		def can_create(self, object_or_class) -> bool:
		return (
		has_crud_access(self.permissions, object_or_class, CRUD_CREATE)
		and self.user.is_active
		)
		def can_crud(
		self, object_or_class: AccessControlledObject, crud_access: str
		) -> bool:
		return self._can_crud(object_or_class, crud_access)

		def can_create(self, object_or_class: AccessControlledObject) -> bool:
		return self._can_crud(object_or_class, CRUD_CREATE)

		def can_read(self, object_or_class) -> bool:
		def can_read(self, object_or_class: AccessControlledObject) -> bool:
		# if self.user.is_superuser and self.user.is_active: # superuser gets a pass
		# return True
		return (
		has_crud_access(self.permissions, object_or_class, CRUD_READ)
		and self.user.is_active
		)
		return self._can_crud(object_or_class, CRUD_READ)

		def can_update(self, object_or_class) -> bool:
		return (
		has_crud_access(self.permissions, object_or_class, CRUD_UPDATE)
		and self.user.is_active
		)
		def can_update(self, object_or_class: AccessControlledObject) -> bool:
		return self._can_crud(object_or_class, CRUD_UPDATE)

		def can_delete(self, object_or_class) -> bool:
		return (
		has_crud_access(self.permissions, object_or_class, CRUD_DELETE)
		and self.user.is_active
		)
		def can_delete(self, object_or_class: AccessControlledObject) -> bool:
		return self._can_crud(object_or_class, CRUD_DELETE)

		def check_create(self, object_or_class):
		"""

src/resources/tests.py

+78 −0

Original line number	Diff line number	Diff line
		@@ -79,6 +79,84 @@ class ServiceTestCase(TestCase):
		self.assertFalse(access.can_update("resource"))
		self.assertFalse(access.can_delete("resource"))

		def test_user_access_rejects_objects_from_another_project(self):
		user = User.objects.create(username="project_bound_user")
		project = Project.objects.create(label="allowed project", description="")
		other_project = Project.objects.create(label="other project", description="")
		role = acl.make_or_create_global_readwrite_role_for_project(project)
		ProjectAccess.objects.create(project=project, user=user, role=role)
		access = UserAccess(user, project)

		file_type = FileType.objects.create(
		title="project boundary file",
		mime="application/x-project-boundary",
		)
		metadata_set = models.MetadataSet.objects.create(
		title="allowed metadata", project=project
		)
		other_metadata_set = models.MetadataSet.objects.create(
		title="other metadata", project=other_project
		)
		allowed_objects = [
		models.Collection.objects.create(title="allowed", project=project),
		models.Resource.objects.create(title="allowed", ptr_project=project),
		File.objects.create(
		title="allowed file",
		original_name="allowed.bin",
		project=project,
		hash="a" * 64,
		file_type=file_type,
		size=0,
		),
		metadata_set,
		models.Metadata.objects.create(
		title="allowed", set=metadata_set, project=project
		),
		models.Tag.objects.create(uid="allowed", project=project),
		models.ProjectProperty.objects.create(
		project=project, key="allowed", value=True
		),
		]
		other_objects = [
		models.Collection.objects.create(title="other", project=other_project),
		models.Resource.objects.create(title="other", ptr_project=other_project),
		File.objects.create(
		title="other file",
		original_name="other.bin",
		project=other_project,
		hash="b" * 64,
		file_type=file_type,
		size=0,
		),
		other_metadata_set,
		models.Metadata.objects.create(
		title="other", set=other_metadata_set, project=other_project
		),
		models.Tag.objects.create(uid="other", project=other_project),
		models.ProjectProperty.objects.create(
		project=other_project, key="other", value=True
		),
		]

		for crud_verb in acl.CRUD_VERBS:
		for allowed_object, other_object in zip(
		allowed_objects, other_objects, strict=True
		):
		self.assertTrue(access.can_crud(allowed_object, crud_verb))
		self.assertFalse(access.can_crud(other_object, crud_verb))

		with self.assertRaises(rpc_methods.ServiceException):
		access.check_read(other_objects[0])

		ProjectAccess.objects.create(
		project=other_project,
		user=user,
		role=role,
		)
		other_access = UserAccess(user, other_project)
		self.assertFalse(other_access.can_read("resource"))
		self.assertFalse(other_access.can_read(other_objects[1]))

		def test_media_command_uses_configured_timeout(self):
		with patch.object(
		helpers.settings, "JAMA_MEDIA_SUBPROCESS_TIMEOUT_SECONDS", 42

src/rpc/methods.py

+23 −23

Original line number	Diff line number	Diff line
		@@ -10,6 +10,7 @@ import numpy
		import jama.native_libs # noqa: F401 - configures libvips lookup on macOS
		import pyvips
		from django.contrib.auth.models import User
		from django.db import transaction
		from django.db.models import QuerySet
		from django.db.utils import IntegrityError
		from django.utils import timezone
		@@ -656,9 +657,11 @@ def add_collection(user: User, title: str, parent_id: int) -> Dict:
		title = title.strip()
		if not title:
		raise ServiceException("missing collection title")
		# fetch parent, check parent's project add access

		with transaction.atomic():
		parent = (
		Collection.objects.filter(
		Collection.objects.select_for_update()
		.filter(
		pk=parent_id,
		deleted_at__isnull=True,
		)
		@@ -667,21 +670,18 @@ def add_collection(user: User, title: str, parent_id: int) -> Dict:
		)
		if not parent:
		raise ServiceException(NO_SUCH_COLLECTION)

		acl = UserAccess(user, parent.project)
		collection_instance = Collection.objects.filter(
		title=title, parent=parent, project_id=parent.project.pk
		).first()
		if not collection_instance:
		acl.check_update(parent)
		collection_instance, created = Collection.objects.get_or_create(

		collection_instance, _ = Collection.objects.get_or_create(
		title=title, parent=parent, project_id=parent.project.pk
		)
		# collection was previously soft-deleted, reactivate it.
		if collection_instance.deleted_at:
		acl.check_update(parent)
		collection_instance.deleted_at = None
		collection_instance.save()
		# acl.check_read(collection_instance) # possible security problem here, leaking data to unauthorized user

		acl.check_read(collection_instance)
		return serializers.collection(collection_instance, cache=SerializerCache())

src/rpc/tests.py

+50 −0

Original line number	Diff line number	Diff line
		@@ -787,6 +787,56 @@ class ServiceTestCase(TestCase):
		)
		self.assertEqual(len(collections), 1)

		def test_add_collection_does_not_expose_existing_collection_without_access(self):
		models.Collection.objects.create(
		title="private collection",
		parent=self.test_project_root_collection,
		project=self.test_project,
		)
		no_access_user = User.objects.create(username="no_collection_access")

		errors = []
		for title in ("private collection", "missing collection"):
		with self.assertRaises(ServiceException) as context:
		methods.add_collection(
		no_access_user,
		title,
		parent_id=self.test_project_root_collection.pk,
		)
		errors.append(context.exception.message)

		self.assertEqual(errors[0], errors[1])
		self.assertFalse(
		models.Collection.objects.filter(title="missing collection").exists()
		)

		def test_add_collection_rolls_back_restore_when_read_is_denied(self):
		collection = models.Collection.objects.create(
		title="hidden deleted collection",
		parent=self.test_project_root_collection,
		project=self.test_project,
		)
		collection.soft_delete()
		models.ObjectPermission.objects.create(
		object_class="collection",
		object_pk=collection.pk,
		role=self.admin_role,
		object_create=True,
		object_read=False,
		object_update=True,
		object_delete=True,
		)

		with self.assertRaises(ServiceException):
		methods.add_collection(
		self.test_user,
		collection.title,
		parent_id=self.test_project_root_collection.pk,
		)

		collection.refresh_from_db()
		self.assertIsNotNone(collection.deleted_at)

		def test_add_tag(self):
		methods.set_tag(self.test_user, "tag bidon", self.test_project.pk)
		tags = methods.tags(self.test_user, self.test_project.pk)