rbac: add InitialPermissions (#13795)

* add `InitialPermissions` model to RBAC

This is a powerful construct between Permission and Role to set initial
permissions for newly created objects.

* use safer `request.user`

* fixup! use safer `request.user`

* force all self-defined serializers to descend from our custom one

See #10139

* reorganize initial permission assignment

* fixup! reorganize initial permission assignment

Author: gergosimonyi

authentik/blueprints/api.py

@@ -7,15 +7,15 @@
 from rest_framework.fields import CharField, DateTimeField
 from rest_framework.request import Request
 from rest_framework.response import Response
-from rest_framework.serializers import ListSerializer, ModelSerializer
+from rest_framework.serializers import ListSerializer
 from rest_framework.viewsets import ModelViewSet
 
 from authentik.blueprints.models import BlueprintInstance
 from authentik.blueprints.v1.importer import Importer
 from authentik.blueprints.v1.oci import OCI_PREFIX
 from authentik.blueprints.v1.tasks import apply_blueprint, blueprints_find_dict
 from authentik.core.api.used_by import UsedByMixin
-from authentik.core.api.utils import JSONDictField, PassiveSerializer
+from authentik.core.api.utils import JSONDictField, ModelSerializer, PassiveSerializer
 from authentik.rbac.decorators import permission_required
 
 

authentik/core/api/utils.py

@@ -20,6 +20,8 @@
     raise_errors_on_nested_writes,
 )
 
+from authentik.rbac.permissions import assign_initial_permissions
+
 
 def is_dict(value: Any):
     """Ensure a value is a dictionary, useful for JSONFields"""
@@ -29,6 +31,14 @@ def is_dict(value: Any):
 
 
 class ModelSerializer(BaseModelSerializer):
+    def create(self, validated_data):
+        instance = super().create(validated_data)
+
+        request = self.context.get("request")
+        if request and hasattr(request, "user") and not request.user.is_anonymous:
+            assign_initial_permissions(request.user, instance)
+
+        return instance
 
     def update(self, instance: Model, validated_data):
         raise_errors_on_nested_writes("update", self, validated_data)

authentik/core/tests/test_api_utils.py

@@ -1,9 +1,17 @@
 """Test API Utils"""
 
 from rest_framework.exceptions import ValidationError
+from rest_framework.serializers import (
+    HyperlinkedModelSerializer,
+)
+from rest_framework.serializers import (
+    ModelSerializer as BaseModelSerializer,
+)
 from rest_framework.test import APITestCase
 
+from authentik.core.api.utils import ModelSerializer as CustomModelSerializer
 from authentik.core.api.utils import is_dict
+from authentik.lib.utils.reflection import all_subclasses
 
 
 class TestAPIUtils(APITestCase):
@@ -14,3 +22,14 @@ def test_is_dict(self):
         self.assertIsNone(is_dict({}))
         with self.assertRaises(ValidationError):
             is_dict("foo")
+
+    def test_all_serializers_descend_from_custom(self):
+        """Test that every serializer we define descends from our own ModelSerializer"""
+        # Weirdly, there's only one serializer in `rest_framework` which descends from
+        # ModelSerializer: HyperlinkedModelSerializer
+        expected = {CustomModelSerializer, HyperlinkedModelSerializer}
+        actual = set(all_subclasses(BaseModelSerializer)) - set(
+            all_subclasses(CustomModelSerializer)
+        )
+
+        self.assertEqual(expected, actual)

authentik/enterprise/providers/ssf/views/stream.py

@@ -4,10 +4,9 @@
 from rest_framework.fields import CharField, ChoiceField, ListField, SerializerMethodField
 from rest_framework.request import Request
 from rest_framework.response import Response
-from rest_framework.serializers import ModelSerializer
 from structlog.stdlib import get_logger
 
-from authentik.core.api.utils import PassiveSerializer
+from authentik.core.api.utils import ModelSerializer, PassiveSerializer
 from authentik.enterprise.providers.ssf.models import (
     DeliveryMethods,
     EventTypes,

authentik/enterprise/stages/authenticator_endpoint_gdtc/api.py

@@ -2,11 +2,11 @@
 
 from rest_framework import mixins
 from rest_framework.permissions import IsAdminUser
-from rest_framework.serializers import ModelSerializer
 from rest_framework.viewsets import GenericViewSet, ModelViewSet
 from structlog.stdlib import get_logger
 
 from authentik.core.api.used_by import UsedByMixin
+from authentik.core.api.utils import ModelSerializer
 from authentik.enterprise.api import EnterpriseRequiredMixin
 from authentik.enterprise.stages.authenticator_endpoint_gdtc.models import (
     AuthenticatorEndpointGDTCStage,

authentik/rbac/api/initial_permissions.py

@@ -0,0 +1,41 @@
+"""RBAC Initial Permissions"""
+
+from rest_framework.serializers import ListSerializer
+from rest_framework.viewsets import ModelViewSet
+
+from authentik.core.api.used_by import UsedByMixin
+from authentik.core.api.utils import ModelSerializer
+from authentik.rbac.api.rbac import PermissionSerializer
+from authentik.rbac.models import InitialPermissions
+
+
+class InitialPermissionsSerializer(ModelSerializer):
+    """InitialPermissions serializer"""
+
+    permissions_obj = ListSerializer(
+        child=PermissionSerializer(),
+        read_only=True,
+        source="permissions",
+        required=False,
+    )
+
+    class Meta:
+        model = InitialPermissions
+        fields = [
+            "pk",
+            "name",
+            "mode",
+            "role",
+            "permissions",
+            "permissions_obj",
+        ]
+
+
+class InitialPermissionsViewSet(UsedByMixin, ModelViewSet):
+    """InitialPermissions viewset"""
+
+    queryset = InitialPermissions.objects.all()
+    serializer_class = InitialPermissionsSerializer
+    search_fields = ["name"]
+    ordering = ["name"]
+    filterset_fields = ["name"]

authentik/rbac/migrations/0005_initialpermissions.py

@@ -0,0 +1,39 @@
+# Generated by Django 5.0.13 on 2025-04-07 13:05
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("auth", "0012_alter_user_first_name_max_length"),
+        ("authentik_rbac", "0004_alter_systempermission_options"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="InitialPermissions",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True, primary_key=True, serialize=False, verbose_name="ID"
+                    ),
+                ),
+                ("name", models.TextField(max_length=150, unique=True)),
+                ("mode", models.CharField(choices=[("user", "User"), ("role", "Role")])),
+                ("permissions", models.ManyToManyField(blank=True, to="auth.permission")),
+                (
+                    "role",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE, to="authentik_rbac.role"
+                    ),
+                ),
+            ],
+            options={
+                "verbose_name": "Initial Permissions",
+                "verbose_name_plural": "Initial Permissions",
+            },
+        ),
+    ]

authentik/rbac/models.py

@@ -3,6 +3,7 @@
 from uuid import uuid4
 
 from django.contrib.auth.management import _get_all_permissions
+from django.contrib.auth.models import Permission
 from django.db import models
 from django.db.transaction import atomic
 from django.utils.translation import gettext_lazy as _
@@ -75,6 +76,35 @@ class Meta:
         ]
 
 
+class InitialPermissionsMode(models.TextChoices):
+    """Determines which entity the initial permissions are assigned to."""
+
+    USER = "user", _("User")
+    ROLE = "role", _("Role")
+
+
+class InitialPermissions(SerializerModel):
+    """Assigns permissions for newly created objects."""
+
+    name = models.TextField(max_length=150, unique=True)
+    mode = models.CharField(choices=InitialPermissionsMode.choices)
+    role = models.ForeignKey(Role, on_delete=models.CASCADE)
+    permissions = models.ManyToManyField(Permission, blank=True)
+
+    @property
+    def serializer(self) -> type[BaseSerializer]:
+        from authentik.rbac.api.initial_permissions import InitialPermissionsSerializer
+
+        return InitialPermissionsSerializer
+
+    def __str__(self) -> str:
+        return f"Initial Permissions for Role #{self.role_id}, applying to #{self.mode}"
+
+    class Meta:
+        verbose_name = _("Initial Permissions")
+        verbose_name_plural = _("Initial Permissions")
+
+
 class SystemPermission(models.Model):
     """System-wide permissions that are not related to any direct
     database model"""

authentik/rbac/permissions.py

@@ -1,9 +1,13 @@
 """RBAC Permissions"""
 
+from django.contrib.contenttypes.models import ContentType
 from django.db.models import Model
+from guardian.shortcuts import assign_perm
 from rest_framework.permissions import BasePermission, DjangoObjectPermissions
 from rest_framework.request import Request
 
+from authentik.rbac.models import InitialPermissions, InitialPermissionsMode
+
 
 class ObjectPermissions(DjangoObjectPermissions):
     """RBAC Permissions"""
@@ -51,3 +55,20 @@ def has_permission(self, request: Request, view):
             return bool(request.user and request.user.has_perms(perm))
 
     return checker
+
+
+# TODO: add `user: User` type annotation without circular dependencies.
+# The author of this function isn't proficient/patient enough to do it.
+def assign_initial_permissions(user, instance: Model):
+    # Performance here should not be an issue, but if needed, there are many optimization routes
+    initial_permissions_list = InitialPermissions.objects.filter(role__group__in=user.groups.all())
+    for initial_permissions in initial_permissions_list:
+        for permission in initial_permissions.permissions.all():
+            if permission.content_type != ContentType.objects.get_for_model(instance):
+                continue
+            assign_to = (
+                user
+                if initial_permissions.mode == InitialPermissionsMode.USER
+                else initial_permissions.role.group
+            )
+            assign_perm(permission, assign_to, instance)

authentik/rbac/tests/test_initial_permissions.py

@@ -0,0 +1,116 @@
+"""Test InitialPermissions"""
+
+from django.contrib.auth.models import Permission
+from guardian.shortcuts import assign_perm
+from rest_framework.reverse import reverse
+from rest_framework.test import APITestCase
+
+from authentik.core.models import Group
+from authentik.core.tests.utils import create_test_user
+from authentik.lib.generators import generate_id
+from authentik.rbac.models import InitialPermissions, InitialPermissionsMode, Role
+from authentik.stages.dummy.models import DummyStage
+
+
+class TestInitialPermissions(APITestCase):
+    """Test InitialPermissions"""
+
+    def setUp(self) -> None:
+        self.user = create_test_user()
+        self.same_role_user = create_test_user()
+        self.different_role_user = create_test_user()
+
+        self.role = Role.objects.create(name=generate_id())
+        self.different_role = Role.objects.create(name=generate_id())
+
+        self.group = Group.objects.create(name=generate_id())
+        self.different_group = Group.objects.create(name=generate_id())
+
+        self.group.roles.add(self.role)
+        self.group.users.add(self.user, self.same_role_user)
+        self.different_group.roles.add(self.different_role)
+        self.different_group.users.add(self.different_role_user)
+
+        self.ip = InitialPermissions.objects.create(
+            name=generate_id(), mode=InitialPermissionsMode.USER, role=self.role
+        )
+        self.view_role = Permission.objects.filter(codename="view_role").first()
+        self.ip.permissions.add(self.view_role)
+
+        assign_perm("authentik_rbac.add_role", self.user)
+        self.client.force_login(self.user)
+
+    def test_different_role(self):
+        """InitialPermissions for different role does nothing"""
+        self.ip.role = self.different_role
+        self.ip.save()
+
+        self.client.post(reverse("authentik_api:roles-list"), {"name": "test-role"})
+
+        role = Role.objects.filter(name="test-role").first()
+        self.assertFalse(self.user.has_perm("authentik_rbac.view_role", role))
+
+    def test_different_model(self):
+        """InitialPermissions for different model does nothing"""
+        assign_perm("authentik_stages_dummy.add_dummystage", self.user)
+
+        self.client.post(
+            reverse("authentik_api:stages-dummy-list"), {"name": "test-stage", "throw-error": False}
+        )
+
+        role = Role.objects.filter(name="test-role").first()
+        self.assertFalse(self.user.has_perm("authentik_rbac.view_role", role))
+        stage = DummyStage.objects.filter(name="test-stage").first()
+        self.assertFalse(self.user.has_perm("authentik_stages_dummy.view_dummystage", stage))
+
+    def test_mode_user(self):
+        """InitialPermissions adds user permission in user mode"""
+        self.client.post(reverse("authentik_api:roles-list"), {"name": "test-role"})
+
+        role = Role.objects.filter(name="test-role").first()
+        self.assertTrue(self.user.has_perm("authentik_rbac.view_role", role))
+        self.assertFalse(self.same_role_user.has_perm("authentik_rbac.view_role", role))
+
+    def test_mode_role(self):
+        """InitialPermissions adds role permission in role mode"""
+        self.ip.mode = InitialPermissionsMode.ROLE
+        self.ip.save()
+
+        self.client.post(reverse("authentik_api:roles-list"), {"name": "test-role"})
+
+        role = Role.objects.filter(name="test-role").first()
+        self.assertTrue(self.user.has_perm("authentik_rbac.view_role", role))
+        self.assertTrue(self.same_role_user.has_perm("authentik_rbac.view_role", role))
+
+    def test_many_permissions(self):
+        """InitialPermissions can add multiple permissions"""
+        change_role = Permission.objects.filter(codename="change_role").first()
+        self.ip.permissions.add(change_role)
+
+        self.client.post(reverse("authentik_api:roles-list"), {"name": "test-role"})
+
+        role = Role.objects.filter(name="test-role").first()
+        self.assertTrue(self.user.has_perm("authentik_rbac.view_role", role))
+        self.assertTrue(self.user.has_perm("authentik_rbac.change_role", role))
+
+    def test_permissions_separated_by_role(self):
+        """When the triggering user is part of two different roles with InitialPermissions in role
+        mode, it only adds permissions to the relevant role."""
+        self.ip.mode = InitialPermissionsMode.ROLE
+        self.ip.save()
+        different_ip = InitialPermissions.objects.create(
+            name=generate_id(), mode=InitialPermissionsMode.ROLE, role=self.different_role
+        )
+        change_role = Permission.objects.filter(codename="change_role").first()
+        different_ip.permissions.add(change_role)
+        self.different_group.users.add(self.user)
+
+        self.client.post(reverse("authentik_api:roles-list"), {"name": "test-role"})
+
+        role = Role.objects.filter(name="test-role").first()
+        self.assertTrue(self.user.has_perm("authentik_rbac.view_role", role))
+        self.assertTrue(self.same_role_user.has_perm("authentik_rbac.view_role", role))
+        self.assertFalse(self.different_role_user.has_perm("authentik_rbac.view_role", role))
+        self.assertTrue(self.user.has_perm("authentik_rbac.change_role", role))
+        self.assertFalse(self.same_role_user.has_perm("authentik_rbac.change_role", role))
+        self.assertTrue(self.different_role_user.has_perm("authentik_rbac.change_role", role))