Dovodi_Backend/apps/account/management/commands/audit_admin_panel_access.py

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