import logging from django.contrib import messages from django.contrib.auth import get_user_model from django.contrib.auth.models import Group, Permission from django.http import HttpResponseForbidden from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.views.decorators.http import require_http_methods, require_POST from apps.core.authentication.invite_email import send_admin_created_user_invitation from apps.core.forms_user_access import UserCreateForm, UserManageForm from apps.core.permissions.access import ( can_manage_role_permissions, can_manage_users, iilmp_role_groups, permissions_grouped_by_app, sync_user_role_group, user_effective_perm_codes, ) from apps.core.permissions.roles import UserRole User = get_user_model() logger = logging.getLogger(__name__) def _forbid(): return HttpResponseForbidden() @require_http_methods(["GET"]) def user_list_view(request): """Redirect to the staff Manage users surface (apps.core.authentication). The richer page covers the same listing plus pending-review queue, bulk import, and inline activate/deactivate. Keeping the /access/users/ URL so existing links and bookmarks continue to work. """ if not can_manage_users(request.user): return _forbid() target = reverse("accounts:manage_users") qs = request.GET.urlencode() return redirect(f"{target}?{qs}" if qs else target) @require_http_methods(["GET", "POST"]) def user_create_view(request): if not can_manage_users(request.user): return _forbid() form = UserCreateForm(request.POST or None, actor=request.user) if request.method == "POST" and form.is_valid(): user = form.save() if form.cleaned_data["additional_roles"]: user.groups.add(*form.cleaned_data["additional_roles"]) messages.success(request, f"User {user.email} created.") if form.cleaned_data.get("send_invitation_email", True): try: send_admin_created_user_invitation(user) messages.info( request, "Invitation email sent with a secure link to set their password.", ) except Exception: logger.exception("admin invitation email failed for %s", user.email) messages.warning( request, "User was saved, but the invitation email could not be sent. " "Check email configuration, or ask them to use Forgot password on the login page.", ) return redirect("core:user_manage_edit", pk=user.pk) return render( request, "core/access/user_create.html", { "form": form, "show_role_permissions_link": can_manage_role_permissions(request.user), "breadcrumbs": [ ("Dashboard", reverse("core:dashboard")), ("User management", reverse("core:user_manage_list")), ("New user", ""), ], }, ) def _can_edit_target(actor, target: User) -> bool: if not can_manage_users(actor): return False if target.is_superuser and not actor.is_superuser: return False return True @require_http_methods(["GET", "POST"]) def user_edit_view(request, pk): target = get_object_or_404(User.objects.select_related("institution"), pk=pk) if not _can_edit_target(request.user, target): return _forbid() form = UserManageForm( request.POST or None, instance=target, actor=request.user, ) if request.method == "POST" and form.is_valid(): user = form.save() enum_group_ids = iilmp_role_groups().values_list("pk", flat=True) user.groups.remove(*user.groups.exclude(pk__in=enum_group_ids)) if form.cleaned_data["additional_roles"]: user.groups.add(*form.cleaned_data["additional_roles"]) messages.success(request, "User updated.") return redirect("core:user_manage_edit", pk=user.pk) eff = user_effective_perm_codes(target) role_group = None if target.role: role_group = Group.objects.filter(name=target.role).first() return render( request, "core/access/user_edit.html", { "form": form, "target_user": target, "effective_perm_codes": sorted(eff) if eff != {"*"} else ["* (superuser — all permissions)"], "role_group": role_group, "show_role_permissions_link": can_manage_role_permissions(request.user), "breadcrumbs": [ ("Dashboard", reverse("core:dashboard")), ("User management", reverse("core:user_manage_list")), (target.email, ""), ], }, ) @require_http_methods(["GET", "POST"]) def role_list_view(request): if not can_manage_role_permissions(request.user): return _forbid() # Show enum-mapped roles + any custom groups created via the UI. enum_groups = list(iilmp_role_groups()) enum_ids = {g.pk for g in enum_groups} custom_groups = list( Group.objects.exclude(pk__in=enum_ids).order_by("name") ) return render( request, "core/access/role_list.html", { "groups": enum_groups, "custom_groups": custom_groups, "breadcrumbs": [ ("Dashboard", reverse("core:dashboard")), ("Role permissions", ""), ], }, ) @require_http_methods(["GET", "POST"]) def role_create_view(request): """Create a new Django auth Group (custom role). Custom roles aren't selectable in the user.role enum field, but they can still hold permissions and be attached to users via Django admin or future group-membership flows. """ if not can_manage_role_permissions(request.user): return _forbid() name = (request.POST.get("name") or "").strip() error = None if request.method == "POST": if not name: error = "Role name is required." elif len(name) > 150: error = "Role name must be 150 characters or fewer." elif Group.objects.filter(name__iexact=name).exists(): error = f"A role named “{name}” already exists." else: group = Group.objects.create(name=name) messages.success(request, f"Role “{group.name}” created. Set its permissions below.") return redirect("core:role_permissions_edit", group_id=group.pk) return render( request, "core/access/role_create.html", { "name": name, "error": error, "breadcrumbs": [ ("Dashboard", reverse("core:dashboard")), ("Role permissions", reverse("core:role_permissions_list")), ("New role", ""), ], }, ) @require_http_methods(["GET", "POST"]) def permission_create_view(request): """Create a custom auth Permission row. Useful when an admin needs a named capability that isn't auto-generated by a model (e.g. cross-cutting "view_dashboard_x"). The permission gets a content_type so it can be assigned via the role permissions matrix. """ if not can_manage_role_permissions(request.user): return _forbid() from django.contrib.contenttypes.models import ContentType name = (request.POST.get("name") or "").strip() codename = (request.POST.get("codename") or "").strip() content_type_id = (request.POST.get("content_type") or "").strip() error = None if request.method == "POST": if not name: error = "Display name is required." elif len(name) > 255: error = "Display name must be 255 characters or fewer." elif not codename: error = "Codename is required." elif len(codename) > 100: error = "Codename must be 100 characters or fewer." elif not codename.replace("_", "").isalnum(): error = "Codename may only contain letters, digits, and underscores." elif not content_type_id.isdigit(): error = "Choose a content type." else: ct = ContentType.objects.filter(pk=int(content_type_id)).first() if ct is None: error = "Selected content type no longer exists." elif Permission.objects.filter(content_type=ct, codename=codename).exists(): error = f"A permission with codename “{codename}” already exists for that content type." else: perm = Permission.objects.create( name=name, codename=codename, content_type=ct ) messages.success( request, f"Permission “{perm.name}” ({ct.app_label}.{perm.codename}) created.", ) return redirect("core:role_permissions_list") content_types = ( ContentType.objects.order_by("app_label", "model").all() ) return render( request, "core/access/permission_create.html", { "name": name, "codename": codename, "selected_content_type_id": content_type_id, "content_types": content_types, "error": error, "breadcrumbs": [ ("Dashboard", reverse("core:dashboard")), ("Role permissions", reverse("core:role_permissions_list")), ("New permission", ""), ], }, ) @require_http_methods(["GET", "POST"]) def role_permissions_view(request, group_id): if not can_manage_role_permissions(request.user): return _forbid() group = get_object_or_404(Group, pk=group_id) grouped = permissions_grouped_by_app() selected_ids = set(group.permissions.values_list("pk", flat=True)) permission_groups = [ { "app_label": app_label, "perms": perms, "granted_count": sum(1 for p in perms if p.pk in selected_ids), "total_count": len(perms), } for app_label, perms in grouped.items() ] default_group = next((g for g in permission_groups if g["granted_count"]), None) default_app_label = (default_group or permission_groups[0])["app_label"] if permission_groups else "" if request.method == "POST": ids = request.POST.getlist("perm") try: id_ints = {int(x) for x in ids if x.isdigit()} except ValueError: id_ints = set() perms = Permission.objects.filter(pk__in=id_ints) group.permissions.set(perms) messages.success(request, f"Permissions updated for role “{group.name}”.") return redirect("core:role_permissions_edit", group_id=group.pk) return render( request, "core/access/role_permissions.html", { "group": group, "permission_groups": permission_groups, "default_app_label": default_app_label, "selected_ids": selected_ids, "breadcrumbs": [ ("Dashboard", reverse("core:dashboard")), ("Role permissions", reverse("core:role_permissions_list")), (group.name, ""), ], }, ) @require_POST def user_resync_groups_view(request, pk): if not can_manage_users(request.user): return _forbid() target = get_object_or_404(User, pk=pk) if not _can_edit_target(request.user, target): return _forbid() sync_user_role_group(target) messages.info(request, "Role group membership re-synced from the user’s role field.") return redirect("core:user_manage_edit", pk=pk)