authentik.core.migrations.0056_user_roles

  1# Generated by Django 5.1.12 on 2025-09-30 12:29
  2
  3from django.db import migrations, models
  4
  5from django.apps.registry import Apps
  6from django.db.backends.base.schema import BaseDatabaseSchemaEditor
  7
  8
  9def migrate_object_permissions(apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
 10    db_alias = schema_editor.connection.alias
 11
 12    User = apps.get_model("authentik_core", "User")
 13    Group = apps.get_model("auth", "Group")
 14    Role = apps.get_model("authentik_rbac", "Role")
 15    UserObjectPermission = apps.get_model("guardian", "UserObjectPermission")
 16    GroupObjectPermission = apps.get_model("guardian", "GroupObjectPermission")
 17    RoleObjectPermission = apps.get_model("guardian", "RoleObjectPermission")
 18    RoleModelPermission = apps.get_model("guardian", "RoleModelPermission")
 19
 20    def get_role_for_user_id(user_id: int) -> Role:
 21        name = f"ak-migrated-role--user-{user_id}"
 22        role, created = Role.objects.using(db_alias).get_or_create(
 23            name=name,
 24        )
 25        if created:
 26            role.users.add(user_id)
 27        return role
 28
 29    def get_role_for_group_id(group_id: int) -> Role:
 30        role = Role.objects.using(db_alias).filter(group_id=group_id).first()
 31        if not role:
 32            # Every django group should already have a role, so this should never happen.
 33            # But let's be nice.
 34            name = f"ak-migrated-role--group-{group_id}"
 35            role, created = Role.objects.using(db_alias).get_or_create(
 36                group_id=group_id,
 37                name=name,
 38            )
 39            if created:
 40                role.group_id = group_id
 41                role.save()
 42        return role
 43
 44    # Below are 4 very similar pieces of code, for (user, group) x (model, object).
 45    # Since this is a one-off migration, I won't attempt DRYing them.
 46
 47    # User model permissions
 48    user_ids_with_model_permissions = (
 49        User.user_permissions.through.objects.using(db_alias)
 50        .values_list("user", flat=True)
 51        .distinct()
 52    )
 53    for user_id in user_ids_with_model_permissions:
 54        role = get_role_for_user_id(user_id)
 55        user_model_permissions = User.user_permissions.through.objects.using(db_alias).filter(
 56            user_id=user_id
 57        )
 58
 59        role_model_permissions = []
 60        for user_model_permission in user_model_permissions:
 61            role_model_permissions.append(
 62                RoleModelPermission(
 63                    permission=user_model_permission.permission,
 64                    content_type=user_model_permission.permission.content_type,
 65                    role=role,
 66                )
 67            )
 68
 69        RoleModelPermission.objects.using(db_alias).bulk_create(role_model_permissions)
 70
 71    # Group model permissions
 72    group_ids_with_model_permissions = (
 73        Group.permissions.through.objects.using(db_alias).values_list("group", flat=True).distinct()
 74    )
 75    for group_id in group_ids_with_model_permissions:
 76        role = get_role_for_group_id(group_id)
 77        group_model_permissions = Group.permissions.through.objects.using(db_alias).filter(
 78            group_id=group_id
 79        )
 80
 81        role_model_permissions = []
 82        for group_model_permission in group_model_permissions:
 83            role_model_permissions.append(
 84                RoleModelPermission(
 85                    permission=group_model_permission.permission,
 86                    content_type=group_model_permission.permission.content_type,
 87                    role=role,
 88                )
 89            )
 90
 91        RoleModelPermission.objects.using(db_alias).bulk_create(role_model_permissions)
 92
 93    # User object permissions
 94    user_ids_with_object_permissions = (
 95        UserObjectPermission.objects.using(db_alias).values_list("user", flat=True).distinct()
 96    )
 97    for user_id in user_ids_with_object_permissions:
 98        role = get_role_for_user_id(user_id)
 99        user_object_permissions = UserObjectPermission.objects.using(db_alias).filter(user=user_id)
100
101        role_object_permissions = []
102        for user_object_permission in user_object_permissions:
103            role_object_permissions.append(
104                RoleObjectPermission(
105                    permission=user_object_permission.permission,
106                    content_type=user_object_permission.content_type,
107                    object_pk=user_object_permission.object_pk,
108                    role=role,
109                )
110            )
111
112        RoleObjectPermission.objects.using(db_alias).bulk_create(role_object_permissions)
113
114    # Group object permissions
115    group_ids_with_object_permissions = (
116        GroupObjectPermission.objects.using(db_alias).values_list("group", flat=True).distinct()
117    )
118    for group_id in group_ids_with_object_permissions:
119        role = get_role_for_group_id(group_id)
120        group_object_permissions = GroupObjectPermission.objects.using(db_alias).filter(
121            group=group_id
122        )
123
124        role_object_permissions = []
125        for group_object_permission in group_object_permissions:
126            role_object_permissions.append(
127                RoleObjectPermission(
128                    permission=group_object_permission.permission,
129                    content_type=group_object_permission.content_type,
130                    object_pk=group_object_permission.object_pk,
131                    role=role,
132                )
133            )
134
135        RoleObjectPermission.objects.using(db_alias).bulk_create(role_object_permissions)
136
137
138class Migration(migrations.Migration):
139
140    dependencies = [
141        ("guardian", "0004_role_permissions"),
142        ("authentik_core", "0055_groupancestor_groupparentagenode_group_parents"),
143        ("authentik_rbac", "0008_alter_role_group"),
144    ]
145
146    operations = [
147        migrations.AddField(
148            model_name="user",
149            name="roles",
150            field=models.ManyToManyField(
151                blank=True, related_name="users", to="authentik_rbac.role"
152            ),
153        ),
154        migrations.RunPython(migrate_object_permissions),
155        migrations.AlterUniqueTogether(
156            name="group",
157            unique_together=set(),
158        ),
159        migrations.AlterField(
160            model_name="group",
161            name="parents",
162            field=models.ManyToManyField(
163                blank=True,
164                related_name="children",
165                through="authentik_core.GroupParentageNode",
166                to="authentik_core.group",
167            ),
168        ),
169        migrations.RemoveField(
170            model_name="group",
171            name="parent",
172        ),
173        migrations.AlterField(
174            model_name="group",
175            name="name",
176            field=models.TextField(unique=True, verbose_name="name"),
177        ),
178    ]