Change ACL -> UserAccess (135f5369) · Commits · certic / Jama

src/nui/views.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -3,7 +3,7 @@ from django.shortcuts import render, get_object_or_404, redirect, reverse # typ
		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.acl import UserAccess
		from resources.models import (
		ProjectAccess,
		Collection,
		@@ -123,7 +123,7 @@ def resource(
		raise Http404()
		project_instance = resource_instance.ptr_project
		user_has_project_access(request.user, project_instance) # type: ignore
		user_has_update_permission = ACL(request.user, project_instance).can_update( # type: ignore
		user_has_update_permission = UserAccess(request.user, project_instance).can_update( # type: ignore
		resource_instance
		)
		if request.method == "POST" and user_has_update_permission:

src/resources/acl.py

+141 −31

Original line number	Diff line number	Diff line
		from django.contrib.auth.models import User
		from resources.models import (
		Project,
		Role,
		ProjectAccess,
		ObjectPermission,
		Resource,
		Collection,
		@@ -13,27 +15,30 @@ from resources.models import (
		from typing import Iterator, Union, List
		from rpc.errors import ServiceException
		from rpc.const import NO_ACCESS
		import logging


		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
		logger = logging.getLogger(__file__)
		CRUD_CREATE = "create"
		CRUD_READ = "read"
		CRUD_UPDATE = "update"
		CRUD_DELETE = "delete"
		CRUD_VERBS = [CRUD_CREATE, CRUD_READ, CRUD_UPDATE, CRUD_DELETE]
		REJECTION_LOGGING_MESSAGE = "ACL rejected {} access to user {} for item {}"


		def _climb_tree(
		object_instance: Union[Resource, Collection],
		) -> Iterator[Union[Resource, Collection]]:
		"""
		Traverse from the leaf Resource or Collection to the root 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 str, considered to be already a class name
		if type(object_instance) is str:
		return object_instance
		if isinstance(object_instance, Resource):
		@@ -50,7 +55,7 @@ def _class_name_from_instance(object_instance: Union[str, Resource, Collection])
		return "project_property"
		if isinstance(object_instance, Tag):
		return "tag"
		return ""
		raise ValueError(f"Unsupported object: {object_instance}")


		def _has_crud_access(
		@@ -59,21 +64,23 @@ def _has_crud_access(
		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}")
		if crud_access not in CRUD_VERBS:
		raise ValueError(
		f"not a crud access right: {crud_access}. Must be one of {', '.join(CRUD_VERBS)}."
		)
		# 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:
		if crud_access == CRUD_CREATE and permission.object_create:
		return True
		if crud_access == "read" and permission.object_read:
		if crud_access == CRUD_READ and permission.object_read:
		return True
		if crud_access == "update" and permission.object_update:
		if crud_access == CRUD_UPDATE and permission.object_update:
		return True
		if crud_access == "delete" and permission.object_delete:
		if crud_access == CRUD_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
		@@ -83,55 +90,158 @@ def _has_crud_access(
		permission.object_class == _class_name_from_instance(node)
		and permission.object_pk == node.pk
		):
		if crud_access == "create" and permission.object_create:
		if crud_access == CRUD_CREATE and permission.object_create:
		return True
		if crud_access == "read" and permission.object_read:
		if crud_access == CRUD_READ and permission.object_read:
		return True
		if crud_access == "update" and permission.object_update:
		if crud_access == CRUD_UPDATE and permission.object_update:
		return True
		if crud_access == "delete" and permission.object_delete:
		if crud_access == CRUD_DELETE and permission.object_delete:
		return True
		return False


		class ACL:
		def set_permission_on_role(
		role: Role,
		object_or_class,
		can_create: bool = False,
		can_read: bool = False,
		can_update: bool = False,
		can_delete: bool = False,
		):
		object_class_name = _class_name_from_instance(object_or_class)
		object_pk = None
		if isinstance(object_or_class, Resource) or isinstance(object_or_class, Collection):
		object_pk = object_or_class.pk
		ObjectPermission.objects.filter(
		role=role,
		object_class=object_class_name,
		object_pk=object_pk,
		).delete()
		ObjectPermission.objects.create(
		role=role,
		object_class=object_class_name,
		object_pk=object_pk,
		object_create=can_create,
		object_read=can_read,
		object_update=can_update,
		object_delete=can_delete,
		)


		def set_role_on_project(project: Project, role_label: str) -> Role:
		role, _ = Role.objects.get_or_create(project=project, label=role_label)
		return role


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

		@property
		def permissions(self) -> List[ObjectPermission]:
		"""
		Gets all rows of ObjectPermission tied to the user.

		(keeps a cached copy in the ACL instance)
		"""
		if self._permissions is None:
		self._permissions = _user_permissions(self.user, self.project)
		self._permissions = []
		for perm in ObjectPermission.objects.filter(
		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:
		return _has_crud_access(self.permissions(), object_or_class, crud_access)
		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")
		return _has_crud_access(self.permissions, object_or_class, CRUD_CREATE)

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

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

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

		def check_create(self, object_or_class):
		"""
		Check if user can create item and raise ServiceException if not
		"""
		if not self.can_create(object_or_class):
		raise ServiceException(NO_ACCESS.format("create", object_or_class))
		logger.info(
		REJECTION_LOGGING_MESSAGE.format(
		CRUD_CREATE, self.user.username, object_or_class
		)
		)
		raise ServiceException(NO_ACCESS.format(CRUD_CREATE, object_or_class))

		def check_read(self, object_or_class):
		"""
		Check if user can read item and raise ServiceException if not
		"""
		if not self.can_read(object_or_class):
		raise ServiceException(NO_ACCESS.format("read", object_or_class))
		logger.info(
		REJECTION_LOGGING_MESSAGE.format(
		CRUD_READ, self.user.username, object_or_class
		)
		)
		raise ServiceException(NO_ACCESS.format(CRUD_READ, object_or_class))

		def check_update(self, object_or_class):
		"""
		Check if user can update item and raise ServiceException if not
		"""
		if not self.can_update(object_or_class):
		raise ServiceException(NO_ACCESS.format("update", object_or_class))
		logger.info(
		REJECTION_LOGGING_MESSAGE.format(
		CRUD_UPDATE, self.user.username, object_or_class
		)
		)
		raise ServiceException(NO_ACCESS.format(CRUD_UPDATE, object_or_class))

		def check_delete(self, object_or_class):
		"""
		Check if user can delete item and raise ServiceException if not
		"""
		if not self.can_delete(object_or_class):
		raise ServiceException(NO_ACCESS.format("delete", object_or_class))
		logger.info(
		REJECTION_LOGGING_MESSAGE.format(
		CRUD_DELETE, self.user.username, object_or_class
		)
		)
		raise ServiceException(NO_ACCESS.format(CRUD_DELETE, object_or_class))

		def user_roles(self) -> Iterator[Role]:
		for access in ProjectAccess.objects.filter(
		project=self.project, user=self.user
		):
		yield access.role

		def project_roles(self) -> Iterator[Role]:
		for role in Role.objects.filter(project=self.project):
		yield role

		def add_role_to_user(self, role: Role) -> ProjectAccess:
		if role.project.pk != self.project.pk:
		raise ValueError("Role does not belong to Project")
		access, _ = ProjectAccess.objects.get_or_create(
		project=self.project, user=self.user, role=role
		)
		self._permissions = None
		return access

		def remove_role_from_user(self, role: Role):
		if role.project.pk != self.project.pk:
		raise ValueError("Role does not belong to Project")
		ProjectAccess.objects.filter(
		project=self.project, user=self.user, role=role
		).delete()
		self._permissions = None

src/resources/models.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -1091,7 +1091,7 @@ class Permission(models.Model):
		class Role(models.Model):
		label = models.TextField()
		project = models.ForeignKey(Project, on_delete=models.CASCADE)
		# old "global" permissions
		# old "global" permissions, deprecated, don't use
		permissions = models.ManyToManyField(Permission)

		def __str__(self):

src/rpc/methods.py

+55 −55

File changed.

Preview size limit exceeded, changes collapsed.

src/rpc/tests.py

+7 −3

Original line number	Diff line number	Diff line
		@@ -4,7 +4,7 @@ from resources import models
		from django.contrib.auth.models import User
		from resources.management.commands.loadfixtures import load_fixtures
		from rpc.methods import ServiceException
		from resources.acl import ACL
		from resources.acl import UserAccess
		from pathlib import Path
		import os

		@@ -105,10 +105,14 @@ class ServiceTestCase(TestCase):
		for object_class in object_classes:
		for c in crud:
		self.assertTrue(
		ACL(self.test_user, self.test_project).can_crud(object_class, c)
		UserAccess(self.test_user, self.test_project).can_crud(
		object_class, c
		)
		)
		with self.assertRaises(ServiceException):
		ACL(self.test_user, self.test_project).check_read("some_unknown_class")
		UserAccess(self.test_user, self.test_project).check_read(
		"some_unknown_class"
		)

		def test_add_collection(self):
		methods.add_collection(