6 changed files with 498 additions and 93 deletions
-
187apps/account/management/commands/audit_admin_panel_access.py
-
68apps/account/models/user.py
-
18apps/account/permissions.py
-
68apps/account/serializers/user.py
-
125apps/account/tests/test_admin_panel_access.py
-
25apps/account/views/user.py
@ -0,0 +1,187 @@ |
|||
from django.core.management.base import BaseCommand |
|||
from django.contrib.auth.models import Group |
|||
|
|||
from apps.account.models import User |
|||
|
|||
|
|||
class Command(BaseCommand): |
|||
help = ( |
|||
"Audit admin panel access consistency for users. " |
|||
"Shows users with panel access and highlights unsafe role/flag/group combinations. " |
|||
"Use --fix to automatically normalize unsafe records." |
|||
) |
|||
|
|||
PANEL_GROUPS = {"Professor Group", "Admin Group", "Super Admin Group", "Super admin Group"} |
|||
|
|||
def add_arguments(self, parser): |
|||
parser.add_argument( |
|||
"--fix", |
|||
action="store_true", |
|||
help="Automatically fix unsafe user records.", |
|||
) |
|||
parser.add_argument( |
|||
"--email", |
|||
action="append", |
|||
dest="emails", |
|||
help="Limit audit to one or more email addresses. Can be passed multiple times.", |
|||
) |
|||
|
|||
def handle(self, *args, **options): |
|||
should_fix = options["fix"] |
|||
emails = options.get("emails") or [] |
|||
|
|||
queryset = User.objects.filter(email__isnull=False).exclude(email="").order_by("id") |
|||
if emails: |
|||
queryset = queryset.filter(email__in=emails) |
|||
|
|||
panel_users = [] |
|||
anomalies = [] |
|||
fixed_count = 0 |
|||
|
|||
for user in queryset: |
|||
groups = list(user.groups.values_list("name", flat=True)) |
|||
can_panel = user.can_access_admin_panel() |
|||
|
|||
if can_panel or user.is_staff or user.is_superuser or any(g in self.PANEL_GROUPS for g in groups): |
|||
panel_users.append( |
|||
{ |
|||
"id": user.id, |
|||
"email": user.email, |
|||
"user_type": user.user_type, |
|||
"is_staff": user.is_staff, |
|||
"is_superuser": user.is_superuser, |
|||
"groups": groups, |
|||
"can_panel": can_panel, |
|||
} |
|||
) |
|||
|
|||
user_anomalies = self.get_anomalies(user, groups) |
|||
if user_anomalies: |
|||
anomalies.append( |
|||
{ |
|||
"user": user, |
|||
"groups": groups, |
|||
"anomalies": user_anomalies, |
|||
} |
|||
) |
|||
|
|||
self.stdout.write(self.style.MIGRATE_HEADING("Panel Access Users")) |
|||
if panel_users: |
|||
for item in panel_users: |
|||
self.stdout.write(str(item)) |
|||
else: |
|||
self.stdout.write(self.style.SUCCESS("No panel-related users found.")) |
|||
|
|||
self.stdout.write("") |
|||
self.stdout.write(self.style.MIGRATE_HEADING("Anomalies")) |
|||
if anomalies: |
|||
for item in anomalies: |
|||
user = item["user"] |
|||
self.stdout.write( |
|||
self.style.WARNING( |
|||
str( |
|||
{ |
|||
"id": user.id, |
|||
"email": user.email, |
|||
"user_type": user.user_type, |
|||
"is_staff": user.is_staff, |
|||
"is_superuser": user.is_superuser, |
|||
"groups": item["groups"], |
|||
"anomalies": item["anomalies"], |
|||
} |
|||
) |
|||
) |
|||
) |
|||
if should_fix: |
|||
if self.fix_user(user, item["anomalies"]): |
|||
fixed_count += 1 |
|||
else: |
|||
self.stdout.write(self.style.SUCCESS("No anomalies found.")) |
|||
|
|||
if should_fix: |
|||
self.stdout.write("") |
|||
self.stdout.write(self.style.SUCCESS(f"Fixed {fixed_count} user(s).")) |
|||
|
|||
def get_anomalies(self, user, groups): |
|||
anomalies = [] |
|||
|
|||
if user.user_type in [User.UserType.STUDENT, User.UserType.CLIENT, User.UserType.CONSULTANT]: |
|||
if user.is_staff or user.is_superuser: |
|||
anomalies.append("low_role_with_staff_flags") |
|||
if any(group in self.PANEL_GROUPS for group in groups): |
|||
anomalies.append("low_role_with_panel_group") |
|||
if user.can_access_admin_panel(): |
|||
anomalies.append("low_role_can_access_panel") |
|||
|
|||
if user.user_type == User.UserType.PROFESSOR and "Professor Group" not in groups: |
|||
anomalies.append("professor_missing_group") |
|||
|
|||
if user.user_type == User.UserType.ADMIN and "Admin Group" not in groups: |
|||
anomalies.append("admin_missing_group") |
|||
|
|||
if user.user_type == User.UserType.SUPER_ADMIN and not any( |
|||
group in groups for group in ["Super Admin Group", "Super admin Group"] |
|||
): |
|||
anomalies.append("super_admin_missing_group") |
|||
|
|||
if user.user_type == User.UserType.SUPER_ADMIN and not user.is_superuser: |
|||
anomalies.append("super_admin_missing_superuser_flag") |
|||
|
|||
return anomalies |
|||
|
|||
def fix_user(self, user, anomalies): |
|||
changed = False |
|||
|
|||
if user.user_type == User.UserType.SUPER_ADMIN: |
|||
if not user.is_staff: |
|||
user.is_staff = True |
|||
changed = True |
|||
if not user.is_superuser: |
|||
user.is_superuser = True |
|||
changed = True |
|||
changed = self.ensure_group(user, "Super Admin Group") or changed |
|||
changed = self.remove_groups(user, {"Professor Group", "Admin Group", "Super admin Group"}) or changed |
|||
|
|||
elif user.user_type == User.UserType.ADMIN: |
|||
if user.is_superuser: |
|||
user.is_superuser = False |
|||
changed = True |
|||
changed = self.ensure_group(user, "Admin Group") or changed |
|||
changed = self.remove_groups(user, {"Professor Group", "Super Admin Group", "Super admin Group"}) or changed |
|||
|
|||
elif user.user_type == User.UserType.PROFESSOR: |
|||
if user.is_superuser: |
|||
user.is_superuser = False |
|||
changed = True |
|||
changed = self.ensure_group(user, "Professor Group") or changed |
|||
changed = self.remove_groups(user, {"Admin Group", "Super Admin Group", "Super admin Group"}) or changed |
|||
|
|||
elif user.user_type in [User.UserType.STUDENT, User.UserType.CLIENT, User.UserType.CONSULTANT]: |
|||
if user.is_staff: |
|||
user.is_staff = False |
|||
changed = True |
|||
if user.is_superuser: |
|||
user.is_superuser = False |
|||
changed = True |
|||
changed = self.remove_groups(user, self.PANEL_GROUPS) or changed |
|||
|
|||
if changed: |
|||
user.save() |
|||
self.stdout.write(self.style.SUCCESS(f"Fixed user {user.id} <{user.email}>")) |
|||
|
|||
return changed |
|||
|
|||
def ensure_group(self, user, group_name): |
|||
group, _ = Group.objects.get_or_create(name=group_name) |
|||
if not user.groups.filter(id=group.id).exists(): |
|||
user.groups.add(group) |
|||
return True |
|||
return False |
|||
|
|||
def remove_groups(self, user, group_names): |
|||
groups = Group.objects.filter(name__in=group_names) |
|||
existing_ids = set(user.groups.filter(id__in=groups.values("id")).values_list("id", flat=True)) |
|||
if existing_ids: |
|||
user.groups.remove(*groups) |
|||
return True |
|||
return False |
|||
@ -0,0 +1,125 @@ |
|||
from django.contrib.auth.models import Group |
|||
from django.urls import reverse |
|||
from rest_framework import status |
|||
from rest_framework.authtoken.models import Token |
|||
from rest_framework.test import APITestCase |
|||
|
|||
from apps.account.models import User |
|||
|
|||
|
|||
class AdminPanelAccessTests(APITestCase): |
|||
def setUp(self): |
|||
self.student_group = Group.objects.create(name="Student Group") |
|||
self.professor_group = Group.objects.create(name="Professor Group") |
|||
self.admin_group = Group.objects.create(name="Admin Group") |
|||
self.super_admin_group = Group.objects.create(name="Super Admin Group") |
|||
|
|||
self.super_admin = User.objects.create_superuser( |
|||
email="superadmin@example.com", |
|||
password="SuperSecret123!", |
|||
fullname="Super Admin", |
|||
) |
|||
self.super_admin.language = None |
|||
self.super_admin.save() |
|||
self.super_admin_token = Token.objects.create(user=self.super_admin) |
|||
|
|||
def _create_user(self, **kwargs): |
|||
password = kwargs.pop("password") |
|||
user = User.objects.create_user(password=password, **kwargs) |
|||
user.language = None |
|||
user.save() |
|||
return user |
|||
|
|||
def test_student_with_staff_flag_cannot_login_to_admin_panel(self): |
|||
user = self._create_user( |
|||
email="student@example.com", |
|||
password="StudentPass123!", |
|||
fullname="Student User", |
|||
user_type=User.UserType.STUDENT, |
|||
is_staff=True, |
|||
) |
|||
|
|||
response = self.client.post( |
|||
reverse("admin-login"), |
|||
{"email": user.email, "password": "StudentPass123!"}, |
|||
format="json", |
|||
) |
|||
|
|||
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) |
|||
|
|||
def test_professor_can_login_to_admin_panel(self): |
|||
professor = self._create_user( |
|||
email="professor@example.com", |
|||
password="ProfessorPass123!", |
|||
fullname="Professor User", |
|||
user_type=User.UserType.PROFESSOR, |
|||
) |
|||
professor.add_role("professor") |
|||
|
|||
response = self.client.post( |
|||
reverse("admin-login"), |
|||
{"email": professor.email, "password": "ProfessorPass123!"}, |
|||
format="json", |
|||
) |
|||
|
|||
self.assertEqual(response.status_code, status.HTTP_201_CREATED) |
|||
self.assertEqual(response.data["user_type"], "professor") |
|||
|
|||
def test_admin_can_login_to_admin_panel(self): |
|||
admin = self._create_user( |
|||
email="admin@example.com", |
|||
password="AdminPass123!", |
|||
fullname="Admin User", |
|||
user_type=User.UserType.ADMIN, |
|||
) |
|||
admin.add_role("admin") |
|||
|
|||
response = self.client.post( |
|||
reverse("admin-login"), |
|||
{"email": admin.email, "password": "AdminPass123!"}, |
|||
format="json", |
|||
) |
|||
|
|||
self.assertEqual(response.status_code, status.HTTP_201_CREATED) |
|||
self.assertEqual(response.data["user_type"], "admin") |
|||
|
|||
def test_updating_user_to_student_revokes_panel_access(self): |
|||
target_user = self._create_user( |
|||
email="target@example.com", |
|||
password="TargetPass123!", |
|||
fullname="Target User", |
|||
user_type=User.UserType.PROFESSOR, |
|||
is_staff=True, |
|||
) |
|||
target_user.add_role("professor") |
|||
target_user.add_role("admin") |
|||
target_user.is_superuser = True |
|||
target_user.save(update_fields=["is_superuser", "is_staff"]) |
|||
|
|||
self.client.credentials(HTTP_AUTHORIZATION=f"Token {self.super_admin_token.key}") |
|||
update_response = self.client.patch( |
|||
reverse("admin-users-detail", args=[target_user.id]), |
|||
{ |
|||
"user_type": User.UserType.STUDENT, |
|||
"is_staff": True, |
|||
"is_superuser": True, |
|||
}, |
|||
format="json", |
|||
) |
|||
|
|||
self.assertEqual(update_response.status_code, status.HTTP_200_OK) |
|||
|
|||
target_user.refresh_from_db() |
|||
self.assertEqual(target_user.user_type, User.UserType.STUDENT) |
|||
self.assertFalse(target_user.is_staff) |
|||
self.assertFalse(target_user.is_superuser) |
|||
self.assertFalse(target_user.has_role("professor")) |
|||
self.assertFalse(target_user.has_role("admin")) |
|||
|
|||
self.client.credentials() |
|||
login_response = self.client.post( |
|||
reverse("admin-login"), |
|||
{"email": target_user.email, "password": "TargetPass123!"}, |
|||
format="json", |
|||
) |
|||
self.assertEqual(login_response.status_code, status.HTTP_401_UNAUTHORIZED) |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue