Authorization V2: Groups of users (#4503)

* proof of concept

* use Dojo_User instead of User

* merge db migrations

* rename db migration after rebase

* unit test for authorization of groups

dojo/api_v2/serializers.py

@@ -7,7 +7,8 @@
     Notes, DojoMeta, FindingImage, Note_Type, App_Analysis, Endpoint_Status, \
     Sonarqube_Issue, Sonarqube_Issue_Transition, Sonarqube_Product, Regulation, \
     System_Settings, FileUpload, SEVERITY_CHOICES, Test_Import, \
-    Test_Import_Finding_Action, Product_Type_Member, Product_Member
+    Test_Import_Finding_Action, Product_Type_Member, Product_Member, \
+    Product_Group, Product_Type_Group, Dojo_Group
 
 from dojo.forms import ImportScanForm
 from dojo.tools.factory import requires_file
@@ -319,6 +320,12 @@ class Meta:
         fields = ('id', 'username', 'first_name', 'last_name')
 
 
+class DojoGroupSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = Dojo_Group
+        fields = '__all__'
+
+
 class NoteHistorySerializer(serializers.ModelSerializer):
     current_editor = UserStubSerializer(read_only=True)
 
@@ -392,6 +399,17 @@ def validate(self, data):
         return data
 
 
+class ProductGroupSerializer(serializers.ModelSerializer):
+    role_name = serializers.SerializerMethodField()
+
+    def get_role_name(self, obj):
+        return Roles(obj.role).name
+
+    class Meta:
+        model = Product_Group
+        fields = '__all__'
+
+
 class ProductSerializer(TaggitSerializer, serializers.ModelSerializer):
     findings_count = serializers.SerializerMethodField()
     findings_list = serializers.SerializerMethodField()
@@ -444,6 +462,17 @@ def validate(self, data):
         return data
 
 
+class ProductTypeGroupSerializer(serializers.ModelSerializer):
+    role_name = serializers.SerializerMethodField()
+
+    def get_role_name(self, obj):
+        return Roles(obj.role).name
+
+    class Meta:
+        model = Product_Type_Group
+        fields = '__all__'
+
+
 class ProductTypeSerializer(serializers.ModelSerializer):
 
     class Meta:

dojo/authorization/authorization.py

@@ -3,7 +3,7 @@
 from dojo.request_cache import cache_for_request
 from dojo.authorization.roles_permissions import Permissions, Roles, get_roles_with_permissions
 from dojo.models import Product_Type, Product_Type_Member, Product, Product_Member, Engagement, \
-    Test, Finding, Endpoint, Finding_Group
+    Test, Finding, Endpoint, Finding_Group, Product_Group, Product_Type_Group
 
 
 def user_has_permission(user, obj, permission):
@@ -15,21 +15,30 @@ def user_has_permission(user, obj, permission):
         return True
 
     if isinstance(obj, Product_Type):
+        # Check if the user has a role for the product type with the requested permissions
         member = get_product_type_member(user, obj)
-        if member is None:
-            return False
-        return role_has_permission(member.role, permission)
+        if member is not None and role_has_permission(member.role, permission):
+            return True
+        # Check if the user is in a group with a role for the product type with the requested permissions
+        for product_type_group in get_product_type_groups(user, obj):
+            if role_has_permission(product_type_group.role, permission):
+                return True
+        return False
     elif (isinstance(obj, Product) and
             permission.value >= Permissions.Product_View.value):
         # Products inherit permissions of their product type
         if user_has_permission(user, obj.prod_type, permission):
             return True
 
-        # Maybe the user has a role for the product with the requested permissions
+        # Check if the user has a role for the product with the requested permissions
         member = get_product_member(user, obj)
-        if member is None:
-            return False
-        return role_has_permission(member.role, permission)
+        if member is not None and role_has_permission(member.role, permission):
+            return True
+        # Check if the user is in a group with a role for the product with the requested permissions
+        for product_group in get_product_groups(user, obj):
+            if role_has_permission(product_group.role, permission):
+                return True
+        return False
     elif isinstance(obj, Engagement) and permission in Permissions.get_engagement_permissions():
         return user_has_permission(user, obj.product, permission)
     elif isinstance(obj, Test) and permission in Permissions.get_test_permissions():
@@ -111,3 +120,13 @@ def get_product_type_member(user, product_type):
         return Product_Type_Member.objects.get(user=user, product_type=product_type)
     except Product_Type_Member.DoesNotExist:
         return None
+
+
+@cache_for_request
+def get_product_groups(user, product):
+    return Product_Group.objects.filter(product=product, group__users=user)
+
+
+@cache_for_request
+def get_product_type_groups(user, product_type):
+    return Product_Type_Group.objects.filter(product_type=product_type, group__users=user)

dojo/db_migrations/0102_dojo_group.py

@@ -0,0 +1,51 @@
+# Generated by Django 3.1.8 on 2021-05-13 05:09
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dojo', '0101_enable_features'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Dojo_Group',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('name', models.CharField(max_length=255, unique=True)),
+                ('description', models.CharField(max_length=4000, null=True)),
+                ('users', models.ManyToManyField(blank=True, to='dojo.Dojo_User')),
+            ],
+        ),
+        migrations.CreateModel(
+            name='Product_Type_Group',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('role', models.IntegerField(default=0)),
+                ('group', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='dojo.dojo_group')),
+                ('product_type', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='dojo.product_type')),
+            ],
+        ),
+        migrations.CreateModel(
+            name='Product_Group',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('role', models.IntegerField(default=0)),
+                ('group', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='dojo.dojo_group')),
+                ('product', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='dojo.product')),
+            ],
+        ),
+        migrations.AddField(
+            model_name='product',
+            name='authorization_groups',
+            field=models.ManyToManyField(blank=True, related_name='product_groups', through='dojo.Product_Group', to='dojo.Dojo_Group'),
+        ),
+        migrations.AddField(
+            model_name='product_type',
+            name='authorization_groups',
+            field=models.ManyToManyField(blank=True, related_name='product_type_groups', through='dojo.Product_Type_Group', to='dojo.Dojo_Group'),
+        ),
+    ]

dojo/models.py

@@ -381,6 +381,12 @@ class UserContactInfo(models.Model):
     block_execution = models.BooleanField(default=False, help_text="Instead of async deduping a finding the findings will be deduped synchronously and will 'block' the user until completion.")
 
 
+class Dojo_Group(models.Model):
+    name = models.CharField(max_length=255, unique=True)
+    description = models.CharField(max_length=4000, null=True)
+    users = models.ManyToManyField(Dojo_User, blank=True)
+
+
 class Contact(models.Model):
     name = models.CharField(max_length=100)
     email = models.EmailField()
@@ -452,6 +458,7 @@ class Product_Type(models.Model):
     created = models.DateTimeField(auto_now_add=True, null=True)
     authorized_users = models.ManyToManyField(User, blank=True)
     members = models.ManyToManyField(Dojo_User, through='Product_Type_Member', related_name='prod_type_members', blank=True)
+    authorization_groups = models.ManyToManyField(Dojo_Group, through='Product_Type_Group', related_name='product_type_groups', blank=True)
 
     @cached_property
     def critical_present(self):
@@ -653,6 +660,7 @@ class Product(models.Model):
     tid = models.IntegerField(default=0, editable=False)
     authorized_users = models.ManyToManyField(User, blank=True)
     members = models.ManyToManyField(Dojo_User, through='Product_Member', related_name='product_members', blank=True)
+    authorization_groups = models.ManyToManyField(Dojo_Group, through='Product_Group', related_name='product_groups', blank=True)
     prod_numeric_grade = models.IntegerField(null=True, blank=True)
 
     # Metadata
@@ -803,12 +811,24 @@ class Product_Member(models.Model):
     role = models.IntegerField(default=0)
 
 
+class Product_Group(models.Model):
+    product = models.ForeignKey(Product, on_delete=models.CASCADE)
+    group = models.ForeignKey(Dojo_Group, on_delete=models.CASCADE)
+    role = models.IntegerField(default=0)
+
+
 class Product_Type_Member(models.Model):
     product_type = models.ForeignKey(Product_Type, on_delete=models.CASCADE)
     user = models.ForeignKey(Dojo_User, on_delete=models.CASCADE)
     role = models.IntegerField(default=0)
 
 
+class Product_Type_Group(models.Model):
+    product_type = models.ForeignKey(Product_Type, on_delete=models.CASCADE)
+    group = models.ForeignKey(Dojo_Group, on_delete=models.CASCADE)
+    role = models.IntegerField(default=0)
+
+
 class Tool_Type(models.Model):
     name = models.CharField(max_length=200)
     description = models.CharField(max_length=2000, null=True)

dojo/product/queries.py

@@ -1,7 +1,8 @@
 from crum import get_current_user
 from django.conf import settings
 from django.db.models import Exists, OuterRef, Q
-from dojo.models import Product, Product_Member, Product_Type_Member, App_Analysis, DojoMeta
+from dojo.models import Product, Product_Member, Product_Type_Member, App_Analysis, \
+    DojoMeta, Product_Group, Product_Type_Group
 from dojo.authorization.authorization import get_roles_for_permission, user_has_permission
 
 
@@ -29,12 +30,22 @@ def get_authorized_products(permission, user=None):
             product=OuterRef('pk'),
             user=user,
             role__in=roles)
+        authorized_product_type_groups = Product_Type_Group.objects.filter(
+            product_type=OuterRef('prod_type_id'),
+            group__users=user,
+            role__in=roles)
+        authorized_product_groups = Product_Group.objects.filter(
+            product=OuterRef('pk'),
+            group__users=user,
+            role__in=roles)
         products = Product.objects.annotate(
             prod_type__member=Exists(authorized_product_type_roles),
-            member=Exists(authorized_product_roles)).order_by('name')
+            member=Exists(authorized_product_roles),
+            prod_type__authorized_group=Exists(authorized_product_type_groups),
+            authorized_group=Exists(authorized_product_groups)).order_by('name')
         products = products.filter(
-            Q(prod_type__member=True) |
-            Q(member=True))
+            Q(prod_type__member=True) | Q(member=True) |
+            Q(prod_type__authorized_group=True) | Q(authorized_group=True))
     else:
         if user.is_staff:
             products = Product.objects.all().order_by('name')

dojo/product_type/queries.py

@@ -1,7 +1,7 @@
 from crum import get_current_user
-from django.db.models import Exists, OuterRef
+from django.db.models import Exists, OuterRef, Q
 from django.conf import settings
-from dojo.models import Product_Type, Product_Type_Member
+from dojo.models import Product_Type, Product_Type_Member, Product_Type_Group
 from dojo.authorization.authorization import get_roles_for_permission, user_has_permission
 
 
@@ -22,8 +22,14 @@ def get_authorized_product_types(permission):
         authorized_roles = Product_Type_Member.objects.filter(product_type=OuterRef('pk'),
             user=user,
             role__in=roles)
-        product_types = Product_Type.objects.annotate(member=Exists(authorized_roles)).order_by('name')
-        product_types = product_types.filter(member=True)
+        authorized_groups = Product_Type_Group.objects.filter(
+            product_type=OuterRef('pk'),
+            group__users=user,
+            role__in=roles)
+        product_types = Product_Type.objects.annotate(
+            member=Exists(authorized_roles),
+            authorized_group=Exists(authorized_groups)).order_by('name')
+        product_types = product_types.filter(Q(member=True) | Q(authorized_group=True))
     else:
         if user.is_staff:
             product_types = Product_Type.objects.all().order_by('name')

dojo/unittests/authorization/test_authorization.py

@@ -2,8 +2,8 @@
 from django.core.exceptions import PermissionDenied
 from django.test import TestCase, override_settings
 from unittest.mock import patch
-from dojo.models import Product_Type, Product_Type_Member, Product, Product_Member, Engagement, \
-    Test, Finding, Endpoint
+from dojo.models import Dojo_User, Product_Type, Product_Type_Member, Product, Product_Member, Engagement, \
+    Test, Finding, Endpoint, Dojo_Group, Product_Group, Product_Type_Group
 import dojo.authorization.authorization
 from dojo.authorization.authorization import role_has_permission, get_roles_for_permission, \
     user_has_permission_or_403, user_has_permission, \
@@ -15,7 +15,7 @@ class TestAuthorization(TestCase):
 
     @classmethod
     def setUpTestData(cls):
-        cls.user = User()
+        cls.user = Dojo_User()
         cls.user.id = 1
         cls.product_type = Product_Type()
         cls.product_type.id = 1
@@ -55,6 +55,33 @@ def setUpTestData(cls):
         cls.product_member_owner.product = cls.product
         cls.product_member_owner.role = Roles.Owner
 
+        cls.group = Dojo_Group()
+        cls.group.id = 1
+
+        cls.product_group_reader = Product_Group()
+        cls.product_group_reader.id = 1
+        cls.product_group_reader.product = cls.product
+        cls.product_group_reader.group = cls.group
+        cls.product_group_reader.role = Roles.Reader
+
+        cls.product_group_owner = Product_Group()
+        cls.product_group_owner.id = 2
+        cls.product_group_owner.product = cls.product
+        cls.product_group_owner.group = cls.group
+        cls.product_group_owner.role = Roles.Owner
+
+        cls.product_type_group_reader = Product_Type_Group()
+        cls.product_type_group_reader.id = 1
+        cls.product_type_group_reader.product_type = cls.product_type
+        cls.product_type_group_reader.group = cls.group
+        cls.product_type_group_reader.role = Roles.Reader
+
+        cls.product_type_group_owner = Product_Type_Group()
+        cls.product_type_group_owner.id = 2
+        cls.product_type_group_owner.product_type = cls.product_type
+        cls.product_type_group_owner.group = cls.group
+        cls.product_type_group_owner.role = Roles.Owner
+
     def test_role_has_permission_exception(self):
         with self.assertRaisesMessage(RoleDoesNotExistError,
                 'Role 9999 does not exist'):
@@ -328,3 +355,43 @@ def test_user_has_permission_product_member_success(self, mock_get):
         self.assertTrue(result)
         self.assertEqual(mock_get.call_args[1]['user'], other_user)
         self.assertEqual(mock_get.call_args[1]['product'], self.product)
+
+    @patch('dojo.models.Product_Group.objects.filter')
+    def test_user_has_group_product_no_permissions(self, mock_get):
+        mock_get.return_value = {self.product_group_reader}
+
+        result = user_has_permission(self.user, self.product, Permissions.Product_Delete)
+
+        self.assertFalse(result)
+        self.assertEqual(mock_get.call_args[1]['group__users'], self.user)
+        self.assertEqual(mock_get.call_args[1]['product'], self.product)
+
+    @patch('dojo.models.Product_Group.objects.filter')
+    def test_user_has_group_product_success(self, mock_get):
+        mock_get.return_value = {self.product_group_owner}
+
+        result = user_has_permission(self.user, self.product, Permissions.Product_Delete)
+
+        self.assertTrue(result)
+        self.assertEqual(mock_get.call_args[1]['group__users'], self.user)
+        self.assertEqual(mock_get.call_args[1]['product'], self.product)
+
+    @patch('dojo.models.Product_Type_Group.objects.filter')
+    def test_user_has_group_product_type_no_permissions(self, mock_get):
+        mock_get.return_value = {self.product_type_group_reader}
+
+        result = user_has_permission(self.user, self.product_type, Permissions.Product_Type_Delete)
+
+        self.assertFalse(result)
+        self.assertEqual(mock_get.call_args[1]['group__users'], self.user)
+        self.assertEqual(mock_get.call_args[1]['product_type'], self.product_type)
+
+    @patch('dojo.models.Product_Type_Group.objects.filter')
+    def test_user_has_group_product_type_success(self, mock_get):
+        mock_get.return_value = {self.product_type_group_owner}
+
+        result = user_has_permission(self.user, self.product_type, Permissions.Product_Type_Delete)
+
+        self.assertTrue(result)
+        self.assertEqual(mock_get.call_args[1]['group__users'], self.user)
+        self.assertEqual(mock_get.call_args[1]['product_type'], self.product_type)