diff --git a/dje/admin.py b/dje/admin.py index 10b884d6..7c1d7257 100644 --- a/dje/admin.py +++ b/dje/admin.py @@ -1624,9 +1624,34 @@ 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_change_permission(self, request, obj=None): + return request.user.is_superuser + + def has_delete_permission(self, request, obj=None): + return request.user.is_superuser + def get_queryset(self, request): """Scope the QuerySet to the current user Dataspace.""" qs = super().get_queryset(request) @@ -1656,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 36c04c39..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 @@ -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,89 @@ 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_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"]) + 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") + 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, change_url, delete_url]: + response = self.client.get(url) + self.assertEqual(403, response.status_code, msg=url) + + data = {"username": regular_user.username, "email": "attacker@mail.com"} + response = self.client.post(change_url, data) + self.assertEqual(403, response.status_code) + + response = self.client.post(delete_url, {"post": "yes"}) + self.assertEqual(403, response.status_code) + + 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) + + response = self.client.post(send_activation_email_url) + self.assertEqual(403, response.status_code) + self.assertEqual(0, len(mail.outbox)) + + 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_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"