new acl system, first pass (27f6aa9d) · Commits · certic / Jama

src/annotations/tests.py

+33 −4

Original line number	Diff line number	Diff line
		@@ -8,6 +8,19 @@ from rpc.methods import ServiceException

		test_json = {"pouet": "tagada"}

		object_classes = [
		"collection",
		"file",
		"metadata",
		"metadataset",
		"permission",
		"project",
		"project_property",
		"resource",
		"role",
		"tag",
		]


		class AnnotationTestCase(TestCase):
		def setUp(self):
		@@ -37,8 +50,16 @@ class AnnotationTestCase(TestCase):
		label="admin", project=self.test_project
		)
		# give all permissions to admin role
		for perm in models.Permission.objects.all():
		self.admin_role.permissions.add(perm)
		for object_class in object_classes:
		models.ObjectPermission.objects.get_or_create(
		object_class=object_class,
		object_pk=None,
		role=self.admin_role,
		object_create=True,
		object_read=True,
		object_update=True,
		object_delete=True,
		)

		# give admin role to the user on the project
		models.ProjectAccess.objects.get_or_create(
		@@ -52,8 +73,16 @@ class AnnotationTestCase(TestCase):
		self.public_role = models.Role.objects.create(
		label="public", project=self.test_project
		)
		for perm in models.Permission.objects.all():
		self.public_role.permissions.add(perm)
		for object_class in object_classes:
		models.ObjectPermission.objects.get_or_create(
		object_class=object_class,
		object_pk=None,
		role=self.public_role,
		object_create=True,
		object_read=True,
		object_update=True,
		object_delete=True,
		)
		models.ProjectAccess.objects.get_or_create(
		project=self.test_project, user=self.public_user, role=self.public_role
		)

src/jama/management/commands/repl.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -16,6 +16,7 @@ from resources.models import (
		APIKey,
		ProjectAccess,
		Permission,
		ObjectPermission,
		File,
		ProjectProperty,
		UserTask,
		@@ -264,6 +265,7 @@ def bpython(self, options):
		"Collection": Collection,
		"File": File,
		"Permission": Permission,
		"ObjectPermission": ObjectPermission,
		"Project": Project,
		"ProjectAccess": ProjectAccess,
		"Resource": Resource,

src/jama/settings.py

+1 −0

Original line number	Diff line number	Diff line
		@@ -42,6 +42,7 @@ if os.getenv("GEOS_LIBRARY_PATH"):


		MODSHIB_CREATE_ACCOUNT = True if os.getenv("MODSHIB_CREATE_ACCOUNT") == "1" else False
		JAMA_SHOW_RPC_DOCS = False if os.getenv("JAMA_SHOW_RPC_DOCS") != "1" else True
		JAMA_USE_POSTGIS = True if os.getenv("JAMA_USE_POSTGIS") == "1" else False
		JAMA_ROOT_URL_REDIRECT = os.getenv("JAMA_ROOT_URL_REDIRECT")
		JAMA_CACHE_BACKEND = os.getenv("JAMA_CACHE_BACKEND") or "file"

src/nui/views.py

+3 −4

Original line number	Diff line number	Diff line
		@@ -3,6 +3,7 @@ from django.shortcuts import render, get_object_or_404, redirect, reverse
		from django.core.exceptions import PermissionDenied
		from django.contrib.auth.decorators import login_required, user_passes_test
		from django.http import HttpRequest, HttpResponse, Http404
		from resources.acl import ACL
		from resources.models import (
		ProjectAccess,
		Collection,
		@@ -15,9 +16,7 @@ from resources.models import (
		)
		from django.core.paginator import Paginator, EmptyPage
		from django.views.decorators.cache import never_cache
		from rpc.methods import _user_has_permission as user_has_permission
		from django.contrib.auth.models import User, AbstractUser
		from rpc.const import PERM_RESOURCE_UPDATE
		from functools import cache
		from django.conf import settings

		@@ -122,8 +121,8 @@ def resource(
		raise Http404()
		project_instance = resource_instance.ptr_project
		user_has_project_access(request.user, project_instance)
		user_has_update_permission = user_has_permission(
		request.user, project_instance, PERM_RESOURCE_UPDATE
		user_has_update_permission = ACL(request.user, project_instance).can_update(
		resource_instance
		)
		if request.method == "POST" and user_has_update_permission:
		for key in request.POST.keys():

src/resources/acl.py

0 → 100644

+137 −0

Original line number	Diff line number	Diff line
		from django.contrib.auth.models import User
		from resources.models import (
		Project,
		ObjectPermission,
		Resource,
		Collection,
		File,
		Metadata,
		MetadataSet,
		Tag,
		ProjectProperty,
		)
		from typing import Iterator, Union, List
		from rpc.errors import ServiceException
		from rpc.const import NO_ACCESS


		def _user_permissions(user: User, project: Project) -> List[ObjectPermission]:
		perms = []
		for perm in ObjectPermission.objects.filter(
		role__projectaccess__user=user, role__projectaccess__project=project
		):
		perms.append(perm)
		return perms


		def _climb_tree(
		object_instance: Union[Resource, Collection],
		) -> Iterator[Union[Resource, Collection]]:
		yield object_instance
		for collec in object_instance.ancestors(reverse=False):
		yield collec


		def _class_name_from_instance(object_instance: Union[str, Resource, Collection]) -> str:
		# already a class name
		if type(object_instance) is str:
		return object_instance
		if isinstance(object_instance, Resource):
		return "resource"
		if isinstance(object_instance, Collection):
		return "collection"
		if isinstance(object_instance, File):
		return "file"
		if isinstance(object_instance, Metadata):
		return "metadata"
		if isinstance(object_instance, MetadataSet):
		return "metadataset"
		if isinstance(object_instance, ProjectProperty):
		return "project_property"
		if isinstance(object_instance, Tag):
		return "tag"
		return ""


		def _has_crud_access(
		user_permissions: List[ObjectPermission],
		object_instance: Union[str, Resource, Collection],
		crud_access: str,
		) -> bool:
		crud_access = str(crud_access).lower().strip()
		if crud_access not in ["create", "read", "update", "delete"]:
		raise ValueError(f"not a crud access right: {crud_access}")
		# class-level permissions take precedence over instance-level
		for permission in user_permissions:
		if (
		permission.object_class == _class_name_from_instance(object_instance)
		and permission.object_pk is None
		):
		if crud_access == "create" and permission.object_create:
		return True
		if crud_access == "read" and permission.object_read:
		return True
		if crud_access == "update" and permission.object_update:
		return True
		if crud_access == "delete" and permission.object_delete:
		return True
		if isinstance(object_instance, Resource) or isinstance(object_instance, Collection):
		# node is either a Resource or a Collection instance
		for node in _climb_tree(object_instance):
		for permission in user_permissions:
		if (
		permission.object_class == _class_name_from_instance(node)
		and permission.object_pk == node.pk
		):
		if crud_access == "create" and permission.object_create:
		return True
		if crud_access == "read" and permission.object_read:
		return True
		if crud_access == "update" and permission.object_update:
		return True
		if crud_access == "delete" and permission.object_delete:
		return True
		return False


		class ACL:
		def __init__(self, user: User, project: Project):
		self.user = user
		self.project = project
		self._permissions = None

		def permissions(self) -> List[ObjectPermission]:
		if self._permissions is None:
		self._permissions = _user_permissions(self.user, self.project)
		return self._permissions

		def can_crud(self, object_or_class, crud_access) -> bool:
		return _has_crud_access(self.permissions(), object_or_class, crud_access)

		def can_create(self, object_or_class) -> bool:
		return _has_crud_access(self.permissions(), object_or_class, "create")

		def can_read(self, object_or_class) -> bool:
		return _has_crud_access(self.permissions(), object_or_class, "read")

		def can_update(self, object_or_class) -> bool:
		return _has_crud_access(self.permissions(), object_or_class, "update")

		def can_delete(self, object_or_class) -> bool:
		return _has_crud_access(self.permissions(), object_or_class, "delete")

		def check_create(self, object_or_class):
		if not self.can_create(object_or_class):
		raise ServiceException(NO_ACCESS.format("create", object_or_class))

		def check_read(self, object_or_class):
		if not self.can_read(object_or_class):
		raise ServiceException(NO_ACCESS.format("read", object_or_class))

		def check_update(self, object_or_class):
		if not self.can_update(object_or_class):
		raise ServiceException(NO_ACCESS.format("update", object_or_class))

		def check_delete(self, object_or_class):
		if not self.can_delete(object_or_class):
		raise ServiceException(NO_ACCESS.format("delete", object_or_class))