From 57fa19541b39fd12ab51cfa6f9aa972bc39427ff Mon Sep 17 00:00:00 2001 From: tdruez Date: Thu, 8 Oct 2026 13:41:44 +0400 Subject: [PATCH 1/4] fix: prevent privilege escalation by non-superusers in the user admin Signed-off-by: tdruez --- dje/admin.py | 45 +++++++++++++- dje/tests/test_user.py | 132 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 175 insertions(+), 2 deletions(-) diff --git a/dje/admin.py b/dje/admin.py index 10b884d6..60d83231 100644 --- a/dje/admin.py +++ b/dje/admin.py @@ -1624,9 +1624,42 @@ def get_form(self, request, obj=None, **kwargs): '{} (permission details)' ) groups_field.label = format_html(label_template, groups_field.label, permission_details_url) + if not request.user.is_superuser: + groups_field.queryset = self.get_assignable_groups(request.user, obj) return form + @staticmethod + def get_assignable_groups(user, edited_user): + """ + Return the groups a non-superuser `user` can assign: their own groups, plus the + current groups of the `edited_user` so they are kept on save. + """ + groups_filter = models.Q(user=user) + if edited_user: + groups_filter |= models.Q(user=edited_user) + return Group.objects.filter(groups_filter).distinct() + + @staticmethod + def is_protected_superuser(request, user): + """Return True if `user` is a superuser and the request user is not.""" + return user.is_superuser and not request.user.is_superuser + + def has_view_permission(self, request, obj=None): + if obj and self.is_protected_superuser(request, obj): + return False + return super().has_view_permission(request, obj) + + def has_change_permission(self, request, obj=None): + if obj and self.is_protected_superuser(request, obj): + return False + return super().has_change_permission(request, obj) + + def has_delete_permission(self, request, obj=None): + if obj and self.is_protected_superuser(request, obj): + return False + return super().has_delete_permission(request, obj) + def get_queryset(self, request): """Scope the QuerySet to the current user Dataspace.""" qs = super().get_queryset(request) @@ -1656,6 +1689,9 @@ def get_readonly_fields(self, request, obj=None): if obj and obj.dataspace != request.user.dataspace: readonly_fields += ("homepage_layout",) + if not request.user.is_superuser: + readonly_fields += ("is_superuser",) + return readonly_fields def user_change_password(self, request, id, form_url=""): @@ -1760,6 +1796,11 @@ def get_actions(self, request): @admin.display(description=_("Set selected users as inactive")) def set_inactive(self, request, queryset): + if not request.user.is_superuser and queryset.filter(is_superuser=True).exists(): + message = "Only a superuser can set a superuser as inactive." + self.message_user(request, message, messages.ERROR) + return + # Execute before the `update()` or the QuerySet will be empty for obj in queryset: History.log_change(request.user, obj, "Set as inactive.") @@ -1815,10 +1856,10 @@ def send_activation_email(self, request, object_id): Set the User inactive a send an activation link to the email address. See also registration.admin.RegistrationAdmin.resend_activation_email """ - if not self.has_change_permission(request): + user = get_object_or_404(self.get_queryset(request), pk=unquote(object_id)) + if not self.has_change_permission(request, user): raise PermissionDenied - user = get_object_or_404(self.get_queryset(request), pk=unquote(object_id)) # User needs to be de-activated for the activation_key to work. # Also needs a unusable_password for the proper redirection. # See ActivationView.get_user() for implementation details. diff --git a/dje/tests/test_user.py b/dje/tests/test_user.py index 36c04c39..fac2f27f 100644 --- a/dje/tests/test_user.py +++ b/dje/tests/test_user.py @@ -14,6 +14,7 @@ from django.contrib.auth import password_validation from django.contrib.auth.hashers import get_hasher from django.contrib.auth.management.commands import changepassword +from django.contrib.auth.models import Group from django.core import mail from django.core import signing from django.core.exceptions import NON_FIELD_ERRORS @@ -30,7 +31,9 @@ from dje.filters import DataspaceFilter from dje.models import Dataspace from dje.registration import DejaCodeActivationView +from dje.tests import add_perms from dje.tests import create +from dje.tests import create_admin from dje.tests import create_superuser from dje.tests import create_user from dje.tests import get_activation_key_from_email @@ -470,6 +473,135 @@ def test_user_admin_form_scope_homepage_layout_choices(self): self.assertContains(response, card_layout_nexb.name) self.assertNotContains(response, card_layout_other.name) + def make_user_admin(self): + user_admin = create_admin("user_admin", self.nexb_dataspace) + codenames = ["add_dejacodeuser", "change_dejacodeuser", "delete_dejacodeuser"] + return add_perms(user_admin, codenames) + + def get_user_change_data(self, user, **extra_data): + return { + "username": user.username, + "email": user.email, + "dataspace": user.dataspace_id, + "is_active": "on", + "is_staff": "on", + **extra_data, + } + + def test_user_admin_non_superuser_cannot_grant_superuser_status(self): + user_admin = self.make_user_admin() + self.client.login(username=user_admin.username, password="secret") + + add_url = reverse("admin:dje_dejacodeuser_add") + response = self.client.get(add_url) + self.assertNotContains(response, 'name="is_superuser"') + + data = { + "username": "new_user", + "email": "user@mail.com", + "dataspace": self.nexb_dataspace.id, + "is_superuser": "on", + } + response = self.client.post(add_url, data) + self.assertEqual(302, response.status_code) + self.assertFalse(get_user_model().objects.get(username="new_user").is_superuser) + + change_url = reverse("admin:dje_dejacodeuser_change", args=[user_admin.pk]) + data = self.get_user_change_data(user_admin, is_superuser="on") + response = self.client.post(change_url, data) + self.assertEqual(302, response.status_code) + user_admin.refresh_from_db() + self.assertFalse(user_admin.is_superuser) + + def test_user_admin_non_superuser_cannot_change_or_disable_superuser(self): + user_admin = self.make_user_admin() + self.client.login(username=user_admin.username, password="secret") + + change_url = reverse("admin:dje_dejacodeuser_change", args=[self.nexb_user.pk]) + response = self.client.get(change_url) + self.assertEqual(403, response.status_code) + + data = self.get_user_change_data(self.nexb_user, is_active="") + response = self.client.post(change_url, data) + self.assertEqual(403, response.status_code) + + delete_url = reverse("admin:dje_dejacodeuser_delete", args=[self.nexb_user.pk]) + response = self.client.post(delete_url, {"post": "yes"}) + self.assertEqual(403, response.status_code) + + send_activation_email_url = reverse( + "admin:dje_dejacodeuser_send_activation_email", args=[self.nexb_user.pk] + ) + response = self.client.post(send_activation_email_url) + self.assertEqual(403, response.status_code) + self.assertEqual(0, len(mail.outbox)) + + self.nexb_user.refresh_from_db() + self.assertTrue(self.nexb_user.is_active) + self.assertTrue(self.nexb_user.is_superuser) + + def test_user_admin_non_superuser_cannot_set_superuser_inactive(self): + user_admin = self.make_user_admin() + regular_user = create_user("regular_user", self.nexb_dataspace) + self.client.login(username=user_admin.username, password="secret") + + changelist_url = reverse("admin:dje_dejacodeuser_changelist") + data = { + "_selected_action": [self.nexb_user.pk, regular_user.pk], + "selected_across": 0, + "action": "set_inactive", + } + response = self.client.post(changelist_url, data, follow=True) + self.assertContains(response, "Only a superuser can set a superuser as inactive.") + self.nexb_user.refresh_from_db() + regular_user.refresh_from_db() + self.assertTrue(self.nexb_user.is_active) + self.assertTrue(regular_user.is_active) + + data["_selected_action"] = [regular_user.pk] + response = self.client.post(changelist_url, data, follow=True) + self.assertContains(response, "1 users set as inactive.") + regular_user.refresh_from_db() + self.assertFalse(regular_user.is_active) + + def test_user_admin_non_superuser_can_only_assign_own_groups(self): + legal_group = Group.objects.create(name="Legal") + engineering_group = Group.objects.create(name="Engineering") + data_admin_group = Group.objects.create(name="Data Administration") + user_admin = self.make_user_admin() + user_admin.groups.add(legal_group) + engineer = create_user("engineer", self.nexb_dataspace) + engineer.groups.add(engineering_group) + + change_url = reverse("admin:dje_dejacodeuser_change", args=[engineer.pk]) + self.client.login(username=self.nexb_user.username, password="secret") + response = self.client.get(change_url) + groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset + expected = [legal_group, engineering_group, data_admin_group] + self.assertQuerySetEqual(expected, groups_queryset, ordered=False) + + self.client.login(username=user_admin.username, password="secret") + response = self.client.get(change_url) + groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset + self.assertQuerySetEqual([legal_group, engineering_group], groups_queryset, ordered=False) + + data = self.get_user_change_data(engineer, groups=[data_admin_group.pk]) + response = self.client.post(change_url, data) + self.assertEqual(200, response.status_code) + self.assertIn("groups", response.context_data["adminform"].form.errors) + self.assertQuerySetEqual([engineering_group], engineer.groups.all()) + + data = self.get_user_change_data(engineer, groups=[legal_group.pk, engineering_group.pk]) + response = self.client.post(change_url, data) + self.assertEqual(302, response.status_code) + expected = [legal_group, engineering_group] + self.assertQuerySetEqual(expected, engineer.groups.all(), ordered=False) + + self.client.login(username=user_admin.username, password="secret") + response = self.client.get(reverse("admin:dje_dejacodeuser_add")) + groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset + self.assertQuerySetEqual([legal_group], groups_queryset) + def test_user_model_send_internal_notification(self): notification = self.nexb_user.send_internal_notification( verb="Updated", description="details" From 160f00b25b3c37504704c5a471bb42c556587e98 Mon Sep 17 00:00:00 2001 From: tdruez Date: Fri, 9 Oct 2026 08:54:45 +0400 Subject: [PATCH 2/4] fix: prevent non-superusers from managing users with privileges beyond their own Signed-off-by: tdruez --- dje/admin.py | 32 +++++------ dje/models.py | 9 ++++ dje/tests/test_user.py | 119 ++++++++++++++++++++++++----------------- 3 files changed, 92 insertions(+), 68 deletions(-) diff --git a/dje/admin.py b/dje/admin.py index 60d83231..fb5570a3 100644 --- a/dje/admin.py +++ b/dje/admin.py @@ -1625,38 +1625,30 @@ def get_form(self, request, obj=None, **kwargs): ) groups_field.label = format_html(label_template, groups_field.label, permission_details_url) if not request.user.is_superuser: - groups_field.queryset = self.get_assignable_groups(request.user, obj) + groups_field.queryset = request.user.groups.all() return form @staticmethod - def get_assignable_groups(user, edited_user): - """ - Return the groups a non-superuser `user` can assign: their own groups, plus the - current groups of the `edited_user` so they are kept on save. - """ - groups_filter = models.Q(user=user) - if edited_user: - groups_filter |= models.Q(user=edited_user) - return Group.objects.filter(groups_filter).distinct() - - @staticmethod - def is_protected_superuser(request, user): - """Return True if `user` is a superuser and the request user is not.""" - return user.is_superuser and not request.user.is_superuser + def is_protected_user(request, user): + """Return True if `user` has privileges beyond the non-superuser request user.""" + if request.user.is_superuser: + return False + protected_users = DejacodeUser.objects.with_privileges_beyond(request.user) + return protected_users.filter(pk=user.pk).exists() def has_view_permission(self, request, obj=None): - if obj and self.is_protected_superuser(request, obj): + if obj and self.is_protected_user(request, obj): return False return super().has_view_permission(request, obj) def has_change_permission(self, request, obj=None): - if obj and self.is_protected_superuser(request, obj): + if obj and self.is_protected_user(request, obj): return False return super().has_change_permission(request, obj) def has_delete_permission(self, request, obj=None): - if obj and self.is_protected_superuser(request, obj): + if obj and self.is_protected_user(request, obj): return False return super().has_delete_permission(request, obj) @@ -1796,8 +1788,8 @@ def get_actions(self, request): @admin.display(description=_("Set selected users as inactive")) def set_inactive(self, request, queryset): - if not request.user.is_superuser and queryset.filter(is_superuser=True).exists(): - message = "Only a superuser can set a superuser as inactive." + if not request.user.is_superuser and queryset.with_privileges_beyond(request.user).exists(): + message = "Only a superuser can set as inactive a user with privileges beyond yours." self.message_user(request, message, messages.ERROR) return diff --git a/dje/models.py b/dje/models.py index 7c19488a..086cd88b 100644 --- a/dje/models.py +++ b/dje/models.py @@ -1607,6 +1607,15 @@ def standards(self): def admins(self): return self.filter(is_staff=True) + def with_privileges_beyond(self, user): + """ + Return the users with privileges the provided `user` does not have: + superusers, and users in a group the `user` is not part of. + """ + other_groups = Group.objects.exclude(user=user) + privileged_filter = models.Q(is_superuser=True) | models.Q(groups__in=other_groups) + return self.filter(privileged_filter).distinct() + class DejacodeUserManager(BaseUserManager, DataspacedManager.from_queryset(DejacodeUserQuerySet)): def create_user(self, username, email, password, dataspace, **extra_fields): diff --git a/dje/tests/test_user.py b/dje/tests/test_user.py index fac2f27f..da3a2294 100644 --- a/dje/tests/test_user.py +++ b/dje/tests/test_user.py @@ -513,50 +513,57 @@ def test_user_admin_non_superuser_cannot_grant_superuser_status(self): user_admin.refresh_from_db() self.assertFalse(user_admin.is_superuser) - def test_user_admin_non_superuser_cannot_change_or_disable_superuser(self): + def test_user_admin_non_superuser_cannot_change_or_disable_privileged_user(self): user_admin = self.make_user_admin() + engineer = create_user("engineer", self.nexb_dataspace) + engineer.groups.add(Group.objects.create(name="Engineering")) self.client.login(username=user_admin.username, password="secret") - change_url = reverse("admin:dje_dejacodeuser_change", args=[self.nexb_user.pk]) - response = self.client.get(change_url) - self.assertEqual(403, response.status_code) + for privileged_user in [self.nexb_user, engineer]: + change_url = reverse("admin:dje_dejacodeuser_change", args=[privileged_user.pk]) + response = self.client.get(change_url) + self.assertEqual(403, response.status_code) - data = self.get_user_change_data(self.nexb_user, is_active="") - response = self.client.post(change_url, data) - self.assertEqual(403, response.status_code) + data = self.get_user_change_data(privileged_user, email="attacker@mail.com") + response = self.client.post(change_url, data) + self.assertEqual(403, response.status_code) - delete_url = reverse("admin:dje_dejacodeuser_delete", args=[self.nexb_user.pk]) - response = self.client.post(delete_url, {"post": "yes"}) - self.assertEqual(403, response.status_code) + delete_url = reverse("admin:dje_dejacodeuser_delete", args=[privileged_user.pk]) + response = self.client.post(delete_url, {"post": "yes"}) + self.assertEqual(403, response.status_code) - send_activation_email_url = reverse( - "admin:dje_dejacodeuser_send_activation_email", args=[self.nexb_user.pk] - ) - response = self.client.post(send_activation_email_url) - self.assertEqual(403, response.status_code) - self.assertEqual(0, len(mail.outbox)) + send_activation_email_url = reverse( + "admin:dje_dejacodeuser_send_activation_email", args=[privileged_user.pk] + ) + response = self.client.post(send_activation_email_url) + self.assertEqual(403, response.status_code) + self.assertEqual(0, len(mail.outbox)) - self.nexb_user.refresh_from_db() - self.assertTrue(self.nexb_user.is_active) - self.assertTrue(self.nexb_user.is_superuser) + privileged_user.refresh_from_db() + self.assertTrue(privileged_user.is_active) + self.assertEqual("user@email.com", privileged_user.email) - def test_user_admin_non_superuser_cannot_set_superuser_inactive(self): + def test_user_admin_non_superuser_cannot_set_privileged_user_inactive(self): user_admin = self.make_user_admin() + engineer = create_user("engineer", self.nexb_dataspace) + engineer.groups.add(Group.objects.create(name="Engineering")) regular_user = create_user("regular_user", self.nexb_dataspace) self.client.login(username=user_admin.username, password="secret") changelist_url = reverse("admin:dje_dejacodeuser_changelist") - data = { - "_selected_action": [self.nexb_user.pk, regular_user.pk], - "selected_across": 0, - "action": "set_inactive", - } - response = self.client.post(changelist_url, data, follow=True) - self.assertContains(response, "Only a superuser can set a superuser as inactive.") - self.nexb_user.refresh_from_db() - regular_user.refresh_from_db() - self.assertTrue(self.nexb_user.is_active) - self.assertTrue(regular_user.is_active) + expected = "Only a superuser can set as inactive a user with privileges beyond yours." + for privileged_user in [self.nexb_user, engineer]: + data = { + "_selected_action": [privileged_user.pk, regular_user.pk], + "selected_across": 0, + "action": "set_inactive", + } + response = self.client.post(changelist_url, data, follow=True) + self.assertContains(response, expected) + privileged_user.refresh_from_db() + regular_user.refresh_from_db() + self.assertTrue(privileged_user.is_active) + self.assertTrue(regular_user.is_active) data["_selected_action"] = [regular_user.pk] response = self.client.post(changelist_url, data, follow=True) @@ -566,41 +573,37 @@ def test_user_admin_non_superuser_cannot_set_superuser_inactive(self): def test_user_admin_non_superuser_can_only_assign_own_groups(self): legal_group = Group.objects.create(name="Legal") - engineering_group = Group.objects.create(name="Engineering") data_admin_group = Group.objects.create(name="Data Administration") user_admin = self.make_user_admin() user_admin.groups.add(legal_group) - engineer = create_user("engineer", self.nexb_dataspace) - engineer.groups.add(engineering_group) + legal_user = create_user("legal_user", self.nexb_dataspace) + legal_user.groups.add(legal_group) - change_url = reverse("admin:dje_dejacodeuser_change", args=[engineer.pk]) + change_url = reverse("admin:dje_dejacodeuser_change", args=[legal_user.pk]) self.client.login(username=self.nexb_user.username, password="secret") response = self.client.get(change_url) groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset - expected = [legal_group, engineering_group, data_admin_group] - self.assertQuerySetEqual(expected, groups_queryset, ordered=False) + self.assertQuerySetEqual([legal_group, data_admin_group], groups_queryset, ordered=False) self.client.login(username=user_admin.username, password="secret") + response = self.client.get(reverse("admin:dje_dejacodeuser_add")) + groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset + self.assertQuerySetEqual([legal_group], groups_queryset) + response = self.client.get(change_url) groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset - self.assertQuerySetEqual([legal_group, engineering_group], groups_queryset, ordered=False) + self.assertQuerySetEqual([legal_group], groups_queryset) - data = self.get_user_change_data(engineer, groups=[data_admin_group.pk]) + data = self.get_user_change_data(legal_user, groups=[data_admin_group.pk]) response = self.client.post(change_url, data) self.assertEqual(200, response.status_code) self.assertIn("groups", response.context_data["adminform"].form.errors) - self.assertQuerySetEqual([engineering_group], engineer.groups.all()) + self.assertQuerySetEqual([legal_group], legal_user.groups.all()) - data = self.get_user_change_data(engineer, groups=[legal_group.pk, engineering_group.pk]) + data = self.get_user_change_data(legal_user, groups=[]) response = self.client.post(change_url, data) self.assertEqual(302, response.status_code) - expected = [legal_group, engineering_group] - self.assertQuerySetEqual(expected, engineer.groups.all(), ordered=False) - - self.client.login(username=user_admin.username, password="secret") - response = self.client.get(reverse("admin:dje_dejacodeuser_add")) - groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset - self.assertQuerySetEqual([legal_group], groups_queryset) + self.assertQuerySetEqual([], legal_user.groups.all()) def test_user_model_send_internal_notification(self): notification = self.nexb_user.send_internal_notification( @@ -838,6 +841,26 @@ def test_user_model_queryset_manager(self): self.assertIn(superuser, admins_actives_qs) self.assertNotIn(inactive, admins_actives_qs) + def test_user_model_queryset_with_privileges_beyond(self): + legal_group = Group.objects.create(name="Legal") + engineering_group = Group.objects.create(name="Engineering") + user_admin = create_admin("user_admin", self.dataspace) + user_admin.groups.add(legal_group) + superuser = create_superuser("superuser", self.dataspace) + legal_user = create_user("legal_user", self.dataspace) + legal_user.groups.add(legal_group) + engineer = create_user("engineer", self.dataspace) + engineer.groups.add(engineering_group) + legal_engineer = create_user("legal_engineer", self.dataspace) + legal_engineer.groups.add(legal_group, engineering_group) + inactive_engineer = create_user("inactive_engineer", self.dataspace, is_active=False) + inactive_engineer.groups.add(engineering_group) + create_user("no_group_user", self.dataspace) + + privileged_users = get_user_model().objects.with_privileges_beyond(user_admin) + expected = [superuser, engineer, legal_engineer, inactive_engineer] + self.assertQuerySetEqual(expected, privileged_users, ordered=False) + def test_user_model_create_api_token(self): user = create_user("active", self.dataspace) self.assertFalse(hasattr(user, "api_token")) From 49d99adad4875e63d04f8cf3edf8867c98aee524 Mon Sep 17 00:00:00 2001 From: tdruez Date: Fri, 9 Oct 2026 09:06:14 +0400 Subject: [PATCH 3/4] fix: restrict the user admin to superusers to prevent privilege escalation Signed-off-by: tdruez --- dje/admin.py | 38 +++------ dje/models.py | 9 --- dje/tests/test_user.py | 176 +++++++++-------------------------------- 3 files changed, 45 insertions(+), 178 deletions(-) diff --git a/dje/admin.py b/dje/admin.py index fb5570a3..d81f7039 100644 --- a/dje/admin.py +++ b/dje/admin.py @@ -1624,33 +1624,21 @@ def get_form(self, request, obj=None, **kwargs): '{} (permission details)' ) groups_field.label = format_html(label_template, groups_field.label, permission_details_url) - if not request.user.is_superuser: - groups_field.queryset = request.user.groups.all() return form - @staticmethod - def is_protected_user(request, user): - """Return True if `user` has privileges beyond the non-superuser request user.""" - if request.user.is_superuser: - return False - protected_users = DejacodeUser.objects.with_privileges_beyond(request.user) - return protected_users.filter(pk=user.pk).exists() - def has_view_permission(self, request, obj=None): - if obj and self.is_protected_user(request, obj): - return False - return super().has_view_permission(request, obj) + """User management is restricted to superusers, to prevent privilege escalation.""" + return request.user.is_superuser + + def has_add_permission(self, request): + return request.user.is_superuser def has_change_permission(self, request, obj=None): - if obj and self.is_protected_user(request, obj): - return False - return super().has_change_permission(request, obj) + return request.user.is_superuser def has_delete_permission(self, request, obj=None): - if obj and self.is_protected_user(request, obj): - return False - return super().has_delete_permission(request, obj) + return request.user.is_superuser def get_queryset(self, request): """Scope the QuerySet to the current user Dataspace.""" @@ -1681,9 +1669,6 @@ def get_readonly_fields(self, request, obj=None): if obj and obj.dataspace != request.user.dataspace: readonly_fields += ("homepage_layout",) - if not request.user.is_superuser: - readonly_fields += ("is_superuser",) - return readonly_fields def user_change_password(self, request, id, form_url=""): @@ -1788,11 +1773,6 @@ def get_actions(self, request): @admin.display(description=_("Set selected users as inactive")) def set_inactive(self, request, queryset): - if not request.user.is_superuser and queryset.with_privileges_beyond(request.user).exists(): - message = "Only a superuser can set as inactive a user with privileges beyond yours." - self.message_user(request, message, messages.ERROR) - return - # Execute before the `update()` or the QuerySet will be empty for obj in queryset: History.log_change(request.user, obj, "Set as inactive.") @@ -1848,10 +1828,10 @@ def send_activation_email(self, request, object_id): Set the User inactive a send an activation link to the email address. See also registration.admin.RegistrationAdmin.resend_activation_email """ - user = get_object_or_404(self.get_queryset(request), pk=unquote(object_id)) - if not self.has_change_permission(request, user): + if not self.has_change_permission(request): raise PermissionDenied + user = get_object_or_404(self.get_queryset(request), pk=unquote(object_id)) # User needs to be de-activated for the activation_key to work. # Also needs a unusable_password for the proper redirection. # See ActivationView.get_user() for implementation details. diff --git a/dje/models.py b/dje/models.py index 086cd88b..7c19488a 100644 --- a/dje/models.py +++ b/dje/models.py @@ -1607,15 +1607,6 @@ def standards(self): def admins(self): return self.filter(is_staff=True) - def with_privileges_beyond(self, user): - """ - Return the users with privileges the provided `user` does not have: - superusers, and users in a group the `user` is not part of. - """ - other_groups = Group.objects.exclude(user=user) - privileged_filter = models.Q(is_superuser=True) | models.Q(groups__in=other_groups) - return self.filter(privileged_filter).distinct() - class DejacodeUserManager(BaseUserManager, DataspacedManager.from_queryset(DejacodeUserQuerySet)): def create_user(self, username, email, password, dataspace, **extra_fields): diff --git a/dje/tests/test_user.py b/dje/tests/test_user.py index da3a2294..eb9284c8 100644 --- a/dje/tests/test_user.py +++ b/dje/tests/test_user.py @@ -14,7 +14,6 @@ from django.contrib.auth import password_validation from django.contrib.auth.hashers import get_hasher from django.contrib.auth.management.commands import changepassword -from django.contrib.auth.models import Group from django.core import mail from django.core import signing from django.core.exceptions import NON_FIELD_ERRORS @@ -473,137 +472,54 @@ def test_user_admin_form_scope_homepage_layout_choices(self): self.assertContains(response, card_layout_nexb.name) self.assertNotContains(response, card_layout_other.name) - def make_user_admin(self): + def test_user_admin_restricted_to_superusers(self): user_admin = create_admin("user_admin", self.nexb_dataspace) - codenames = ["add_dejacodeuser", "change_dejacodeuser", "delete_dejacodeuser"] - return add_perms(user_admin, codenames) - - def get_user_change_data(self, user, **extra_data): - return { - "username": user.username, - "email": user.email, - "dataspace": user.dataspace_id, - "is_active": "on", - "is_staff": "on", - **extra_data, - } - - def test_user_admin_non_superuser_cannot_grant_superuser_status(self): - user_admin = self.make_user_admin() + codenames = ["view_dejacodeuser", "add_dejacodeuser", "change_dejacodeuser"] + user_admin = add_perms(user_admin, codenames + ["delete_dejacodeuser"]) + regular_user = create_user("regular_user", self.nexb_dataspace) self.client.login(username=user_admin.username, password="secret") + changelist_url = reverse("admin:dje_dejacodeuser_changelist") add_url = reverse("admin:dje_dejacodeuser_add") - response = self.client.get(add_url) - self.assertNotContains(response, 'name="is_superuser"') + change_url = reverse("admin:dje_dejacodeuser_change", args=[regular_user.pk]) + delete_url = reverse("admin:dje_dejacodeuser_delete", args=[regular_user.pk]) + send_activation_email_url = reverse( + "admin:dje_dejacodeuser_send_activation_email", args=[regular_user.pk] + ) - data = { - "username": "new_user", - "email": "user@mail.com", - "dataspace": self.nexb_dataspace.id, - "is_superuser": "on", - } + for url in [changelist_url, add_url, change_url, delete_url]: + response = self.client.get(url) + self.assertEqual(403, response.status_code, msg=url) + + data = {"username": "new_user", "email": "user@mail.com", "is_superuser": "on"} response = self.client.post(add_url, data) - self.assertEqual(302, response.status_code) - self.assertFalse(get_user_model().objects.get(username="new_user").is_superuser) + self.assertEqual(403, response.status_code) + self.assertFalse(get_user_model().objects.filter(username="new_user").exists()) - change_url = reverse("admin:dje_dejacodeuser_change", args=[user_admin.pk]) - data = self.get_user_change_data(user_admin, is_superuser="on") + data = {"username": regular_user.username, "email": "attacker@mail.com"} response = self.client.post(change_url, data) - self.assertEqual(302, response.status_code) - user_admin.refresh_from_db() - self.assertFalse(user_admin.is_superuser) - - def test_user_admin_non_superuser_cannot_change_or_disable_privileged_user(self): - user_admin = self.make_user_admin() - engineer = create_user("engineer", self.nexb_dataspace) - engineer.groups.add(Group.objects.create(name="Engineering")) - self.client.login(username=user_admin.username, password="secret") - - for privileged_user in [self.nexb_user, engineer]: - change_url = reverse("admin:dje_dejacodeuser_change", args=[privileged_user.pk]) - response = self.client.get(change_url) - self.assertEqual(403, response.status_code) - - data = self.get_user_change_data(privileged_user, email="attacker@mail.com") - response = self.client.post(change_url, data) - self.assertEqual(403, response.status_code) - - delete_url = reverse("admin:dje_dejacodeuser_delete", args=[privileged_user.pk]) - response = self.client.post(delete_url, {"post": "yes"}) - self.assertEqual(403, response.status_code) - - send_activation_email_url = reverse( - "admin:dje_dejacodeuser_send_activation_email", args=[privileged_user.pk] - ) - response = self.client.post(send_activation_email_url) - self.assertEqual(403, response.status_code) - self.assertEqual(0, len(mail.outbox)) - - privileged_user.refresh_from_db() - self.assertTrue(privileged_user.is_active) - self.assertEqual("user@email.com", privileged_user.email) - - def test_user_admin_non_superuser_cannot_set_privileged_user_inactive(self): - user_admin = self.make_user_admin() - engineer = create_user("engineer", self.nexb_dataspace) - engineer.groups.add(Group.objects.create(name="Engineering")) - regular_user = create_user("regular_user", self.nexb_dataspace) - self.client.login(username=user_admin.username, password="secret") - - changelist_url = reverse("admin:dje_dejacodeuser_changelist") - expected = "Only a superuser can set as inactive a user with privileges beyond yours." - for privileged_user in [self.nexb_user, engineer]: - data = { - "_selected_action": [privileged_user.pk, regular_user.pk], - "selected_across": 0, - "action": "set_inactive", - } - response = self.client.post(changelist_url, data, follow=True) - self.assertContains(response, expected) - privileged_user.refresh_from_db() - regular_user.refresh_from_db() - self.assertTrue(privileged_user.is_active) - self.assertTrue(regular_user.is_active) - - data["_selected_action"] = [regular_user.pk] - response = self.client.post(changelist_url, data, follow=True) - self.assertContains(response, "1 users set as inactive.") - regular_user.refresh_from_db() - self.assertFalse(regular_user.is_active) - - def test_user_admin_non_superuser_can_only_assign_own_groups(self): - legal_group = Group.objects.create(name="Legal") - data_admin_group = Group.objects.create(name="Data Administration") - user_admin = self.make_user_admin() - user_admin.groups.add(legal_group) - legal_user = create_user("legal_user", self.nexb_dataspace) - legal_user.groups.add(legal_group) - - change_url = reverse("admin:dje_dejacodeuser_change", args=[legal_user.pk]) - self.client.login(username=self.nexb_user.username, password="secret") - response = self.client.get(change_url) - groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset - self.assertQuerySetEqual([legal_group, data_admin_group], groups_queryset, ordered=False) + self.assertEqual(403, response.status_code) - self.client.login(username=user_admin.username, password="secret") - response = self.client.get(reverse("admin:dje_dejacodeuser_add")) - groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset - self.assertQuerySetEqual([legal_group], groups_queryset) + response = self.client.post(delete_url, {"post": "yes"}) + self.assertEqual(403, response.status_code) - response = self.client.get(change_url) - groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset - self.assertQuerySetEqual([legal_group], groups_queryset) + data = { + "_selected_action": [self.nexb_user.pk], + "selected_across": 0, + "action": "set_inactive", + } + response = self.client.post(changelist_url, data) + self.assertEqual(403, response.status_code) - data = self.get_user_change_data(legal_user, groups=[data_admin_group.pk]) - response = self.client.post(change_url, data) - self.assertEqual(200, response.status_code) - self.assertIn("groups", response.context_data["adminform"].form.errors) - self.assertQuerySetEqual([legal_group], legal_user.groups.all()) + response = self.client.post(send_activation_email_url) + self.assertEqual(403, response.status_code) + self.assertEqual(0, len(mail.outbox)) - data = self.get_user_change_data(legal_user, groups=[]) - response = self.client.post(change_url, data) - self.assertEqual(302, response.status_code) - self.assertQuerySetEqual([], legal_user.groups.all()) + regular_user.refresh_from_db() + self.nexb_user.refresh_from_db() + self.assertTrue(regular_user.is_active) + self.assertEqual("user@email.com", regular_user.email) + self.assertTrue(self.nexb_user.is_active) def test_user_model_send_internal_notification(self): notification = self.nexb_user.send_internal_notification( @@ -841,26 +757,6 @@ def test_user_model_queryset_manager(self): self.assertIn(superuser, admins_actives_qs) self.assertNotIn(inactive, admins_actives_qs) - def test_user_model_queryset_with_privileges_beyond(self): - legal_group = Group.objects.create(name="Legal") - engineering_group = Group.objects.create(name="Engineering") - user_admin = create_admin("user_admin", self.dataspace) - user_admin.groups.add(legal_group) - superuser = create_superuser("superuser", self.dataspace) - legal_user = create_user("legal_user", self.dataspace) - legal_user.groups.add(legal_group) - engineer = create_user("engineer", self.dataspace) - engineer.groups.add(engineering_group) - legal_engineer = create_user("legal_engineer", self.dataspace) - legal_engineer.groups.add(legal_group, engineering_group) - inactive_engineer = create_user("inactive_engineer", self.dataspace, is_active=False) - inactive_engineer.groups.add(engineering_group) - create_user("no_group_user", self.dataspace) - - privileged_users = get_user_model().objects.with_privileges_beyond(user_admin) - expected = [superuser, engineer, legal_engineer, inactive_engineer] - self.assertQuerySetEqual(expected, privileged_users, ordered=False) - def test_user_model_create_api_token(self): user = create_user("active", self.dataspace) self.assertFalse(hasattr(user, "api_token")) From ca23ee8bdf5ccfe7d0211cc2a0884c077e800f9b Mon Sep 17 00:00:00 2001 From: tdruez Date: Fri, 9 Oct 2026 11:17:05 +0400 Subject: [PATCH 4/4] fix: allow non-superusers to add users with their own groups only Signed-off-by: tdruez --- dje/admin.py | 21 ++++++++++++++--- dje/tests/test_user.py | 51 +++++++++++++++++++++++++++++++++++------- 2 files changed, 61 insertions(+), 11 deletions(-) diff --git a/dje/admin.py b/dje/admin.py index d81f7039..7c1d7257 100644 --- a/dje/admin.py +++ b/dje/admin.py @@ -1624,16 +1624,28 @@ def get_form(self, request, obj=None, **kwargs): '{} (permission details)' ) groups_field.label = format_html(label_template, groups_field.label, permission_details_url) + if not request.user.is_superuser: + groups_field.queryset = request.user.groups.all() return form + def add_view(self, request, form_url="", extra_context=None): + """ + Skip the UserAdmin check that requires the change permission to add users: + a non-superuser cannot set the superuser status nor assign groups beyond their own. + """ + return admin.ModelAdmin.add_view(self, request, form_url, extra_context) + + def response_add(self, request, obj, post_url_continue=None): + """Skip the UserAdmin redirect to the change form, not available to non-superusers.""" + if not request.user.is_superuser: + return admin.ModelAdmin.response_add(self, request, obj, post_url_continue) + return super().response_add(request, obj, post_url_continue) + def has_view_permission(self, request, obj=None): """User management is restricted to superusers, to prevent privilege escalation.""" return request.user.is_superuser - def has_add_permission(self, request): - return request.user.is_superuser - def has_change_permission(self, request, obj=None): return request.user.is_superuser @@ -1669,6 +1681,9 @@ def get_readonly_fields(self, request, obj=None): if obj and obj.dataspace != request.user.dataspace: readonly_fields += ("homepage_layout",) + if not request.user.is_superuser: + readonly_fields += ("is_superuser",) + return readonly_fields def user_change_password(self, request, id, form_url=""): diff --git a/dje/tests/test_user.py b/dje/tests/test_user.py index eb9284c8..bcecfe9f 100644 --- a/dje/tests/test_user.py +++ b/dje/tests/test_user.py @@ -14,6 +14,7 @@ from django.contrib.auth import password_validation from django.contrib.auth.hashers import get_hasher from django.contrib.auth.management.commands import changepassword +from django.contrib.auth.models import Group from django.core import mail from django.core import signing from django.core.exceptions import NON_FIELD_ERRORS @@ -472,7 +473,7 @@ def test_user_admin_form_scope_homepage_layout_choices(self): self.assertContains(response, card_layout_nexb.name) self.assertNotContains(response, card_layout_other.name) - def test_user_admin_restricted_to_superusers(self): + def test_user_admin_management_restricted_to_superusers(self): user_admin = create_admin("user_admin", self.nexb_dataspace) codenames = ["view_dejacodeuser", "add_dejacodeuser", "change_dejacodeuser"] user_admin = add_perms(user_admin, codenames + ["delete_dejacodeuser"]) @@ -480,22 +481,16 @@ def test_user_admin_restricted_to_superusers(self): self.client.login(username=user_admin.username, password="secret") changelist_url = reverse("admin:dje_dejacodeuser_changelist") - add_url = reverse("admin:dje_dejacodeuser_add") change_url = reverse("admin:dje_dejacodeuser_change", args=[regular_user.pk]) delete_url = reverse("admin:dje_dejacodeuser_delete", args=[regular_user.pk]) send_activation_email_url = reverse( "admin:dje_dejacodeuser_send_activation_email", args=[regular_user.pk] ) - for url in [changelist_url, add_url, change_url, delete_url]: + for url in [changelist_url, change_url, delete_url]: response = self.client.get(url) self.assertEqual(403, response.status_code, msg=url) - data = {"username": "new_user", "email": "user@mail.com", "is_superuser": "on"} - response = self.client.post(add_url, data) - self.assertEqual(403, response.status_code) - self.assertFalse(get_user_model().objects.filter(username="new_user").exists()) - data = {"username": regular_user.username, "email": "attacker@mail.com"} response = self.client.post(change_url, data) self.assertEqual(403, response.status_code) @@ -521,6 +516,46 @@ def test_user_admin_restricted_to_superusers(self): self.assertEqual("user@email.com", regular_user.email) self.assertTrue(self.nexb_user.is_active) + def test_user_admin_non_superuser_add_user(self): + legal_group = Group.objects.create(name="Legal") + data_admin_group = Group.objects.create(name="Data Administration") + user_admin = create_admin("user_admin", self.nexb_dataspace) + user_admin = add_perms(user_admin, ["add_dejacodeuser"]) + user_admin.groups.add(data_admin_group) + add_url = reverse("admin:dje_dejacodeuser_add") + + self.client.login(username=self.nexb_user.username, password="secret") + response = self.client.get(add_url) + groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset + self.assertQuerySetEqual([legal_group, data_admin_group], groups_queryset, ordered=False) + + self.client.login(username=user_admin.username, password="secret") + response = self.client.get(add_url) + self.assertEqual(200, response.status_code) + self.assertNotContains(response, 'name="is_superuser"') + groups_queryset = response.context_data["adminform"].form.fields["groups"].queryset + self.assertQuerySetEqual([data_admin_group], groups_queryset) + + data = { + "username": "new_user", + "email": "user@mail.com", + "dataspace": self.nexb_dataspace.id, + "groups": [legal_group.pk], + } + response = self.client.post(add_url, data) + self.assertEqual(200, response.status_code) + self.assertIn("groups", response.context_data["adminform"].form.errors) + self.assertFalse(get_user_model().objects.filter(username="new_user").exists()) + + data["groups"] = [data_admin_group.pk] + data["is_superuser"] = "on" + response = self.client.post(add_url, data) + self.assertRedirects(response, reverse("admin:index")) + new_user = get_user_model().objects.get(username="new_user") + self.assertFalse(new_user.is_superuser) + self.assertQuerySetEqual([data_admin_group], new_user.groups.all()) + self.assertEqual(1, len(mail.outbox)) + def test_user_model_send_internal_notification(self): notification = self.nexb_user.send_internal_notification( verb="Updated", description="details"