Loading src/rpc/methods.py +9 −2 Original line number Diff line number Diff line Loading @@ -91,7 +91,7 @@ from rpc.const import ( WRONG_ARGUMENT, ) from django.db.models.functions import Cast from django.db.models import BinaryField, Count, Q from django.db.models import BinaryField, Count, F, Q logger = logging.getLogger(__name__) Loading Loading @@ -2543,7 +2543,10 @@ def projects_user_permissions( """ access_list = [] serializer_cache = SerializerCache() for project_access in ProjectAccess.objects.filter(user=user): for project_access in ProjectAccess.objects.filter( user=user, role__project_id=F("project_id"), ): access_list.append( serializers.project_access( project_access, Loading Loading @@ -2675,6 +2678,8 @@ def add_project_access(user: User, user_id: int, project_id: int, role_id: int) role_instance = Role.objects.filter(pk=role_id).first() if not role_instance: raise ServiceException("no such role") if role_instance.project_id != project_instance.pk: raise ServiceException(PROJECT_MISMATCH) ProjectAccess.objects.get_or_create( role=role_instance, project=project_instance, user=user_instance ) Loading @@ -2699,6 +2704,8 @@ def remove_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 != project_instance.pk: raise ServiceException(PROJECT_MISMATCH) if role_instance: ProjectAccess.objects.filter( Loading src/rpc/tests.py +67 −0 Original line number Diff line number Diff line Loading @@ -1112,6 +1112,73 @@ class ServiceTestCase(TestCase): ) self.assertEqual(context.exception.message, methods.PROJECT_MISMATCH) def test_project_access_rpc_enforces_role_project_invariant(self): target_user = User.objects.create( username="project_access_target", is_active=False, ) other_project = models.Project.objects.create(label="access role owner") other_role = models.Role.objects.create( label="foreign access role", project=other_project, ) with self.assertRaises(ServiceException) as context: methods.add_project_access( self.admin_user, target_user.pk, self.test_project.pk, other_role.pk, ) self.assertEqual(context.exception.message, methods.PROJECT_MISMATCH) target_user.refresh_from_db() self.assertFalse(target_user.is_active) self.assertFalse( models.ProjectAccess.objects.filter( project=self.test_project, user=target_user, ).exists() ) self.assertTrue( methods.add_project_access( self.admin_user, target_user.pk, self.test_project.pk, self.admin_role.pk, ) ) with self.assertRaises(ServiceException) as context: methods.remove_project_access( self.admin_user, target_user.pk, self.test_project.pk, other_role.pk, ) self.assertEqual(context.exception.message, methods.PROJECT_MISMATCH) self.assertTrue( models.ProjectAccess.objects.filter( project=self.test_project, user=target_user, role=self.admin_role, ).exists() ) self.assertTrue( methods.remove_project_access( self.admin_user, target_user.pk, self.test_project.pk, self.admin_role.pk, ) ) models.ProjectAccess.objects.create( project=self.test_project, user=target_user, role=other_role, ) self.assertEqual(methods.projects_user_permissions(target_user), []) def test_project_stats_requires_project_membership(self): no_access_user = User.objects.create(username="no_project_stats_access") Loading Loading
src/rpc/methods.py +9 −2 Original line number Diff line number Diff line Loading @@ -91,7 +91,7 @@ from rpc.const import ( WRONG_ARGUMENT, ) from django.db.models.functions import Cast from django.db.models import BinaryField, Count, Q from django.db.models import BinaryField, Count, F, Q logger = logging.getLogger(__name__) Loading Loading @@ -2543,7 +2543,10 @@ def projects_user_permissions( """ access_list = [] serializer_cache = SerializerCache() for project_access in ProjectAccess.objects.filter(user=user): for project_access in ProjectAccess.objects.filter( user=user, role__project_id=F("project_id"), ): access_list.append( serializers.project_access( project_access, Loading Loading @@ -2675,6 +2678,8 @@ def add_project_access(user: User, user_id: int, project_id: int, role_id: int) role_instance = Role.objects.filter(pk=role_id).first() if not role_instance: raise ServiceException("no such role") if role_instance.project_id != project_instance.pk: raise ServiceException(PROJECT_MISMATCH) ProjectAccess.objects.get_or_create( role=role_instance, project=project_instance, user=user_instance ) Loading @@ -2699,6 +2704,8 @@ def remove_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 != project_instance.pk: raise ServiceException(PROJECT_MISMATCH) if role_instance: ProjectAccess.objects.filter( Loading
src/rpc/tests.py +67 −0 Original line number Diff line number Diff line Loading @@ -1112,6 +1112,73 @@ class ServiceTestCase(TestCase): ) self.assertEqual(context.exception.message, methods.PROJECT_MISMATCH) def test_project_access_rpc_enforces_role_project_invariant(self): target_user = User.objects.create( username="project_access_target", is_active=False, ) other_project = models.Project.objects.create(label="access role owner") other_role = models.Role.objects.create( label="foreign access role", project=other_project, ) with self.assertRaises(ServiceException) as context: methods.add_project_access( self.admin_user, target_user.pk, self.test_project.pk, other_role.pk, ) self.assertEqual(context.exception.message, methods.PROJECT_MISMATCH) target_user.refresh_from_db() self.assertFalse(target_user.is_active) self.assertFalse( models.ProjectAccess.objects.filter( project=self.test_project, user=target_user, ).exists() ) self.assertTrue( methods.add_project_access( self.admin_user, target_user.pk, self.test_project.pk, self.admin_role.pk, ) ) with self.assertRaises(ServiceException) as context: methods.remove_project_access( self.admin_user, target_user.pk, self.test_project.pk, other_role.pk, ) self.assertEqual(context.exception.message, methods.PROJECT_MISMATCH) self.assertTrue( models.ProjectAccess.objects.filter( project=self.test_project, user=target_user, role=self.admin_role, ).exists() ) self.assertTrue( methods.remove_project_access( self.admin_user, target_user.pk, self.test_project.pk, self.admin_role.pk, ) ) models.ProjectAccess.objects.create( project=self.test_project, user=target_user, role=other_role, ) self.assertEqual(methods.projects_user_permissions(target_user), []) def test_project_stats_requires_project_membership(self): no_access_user = User.objects.create(username="no_project_stats_access") Loading