from django.contrib import messages from django.contrib.auth import get_user_model, update_session_auth_hash from django.db import transaction from django.http import JsonResponse from django.shortcuts import redirect from django.utils.translation import gettext_lazy as _ from ....acl.useracl import get_user_acl from ....admin.auth import authorize_admin from ....admin.views import generic from ....categories.models import Category from ....core.mail import mail_users from ....core.pgutils import chunk_queryset from ....threads.models import Thread from ...avatars.dynamic import set_avatar as set_dynamic_avatar from ...datadownloads import request_user_data_download, user_has_data_download_request from ...models import Ban from ...profilefields import profilefields from ...setupnewuser import setup_new_user from ...signatures import set_user_signature from ..forms import ( BanUsersForm, EditUserForm, EditUserFormFactory, NewUserForm, create_filter_users_form, ) from ..tasks import delete_user_with_content User = get_user_model() class UserAdmin(generic.AdminBaseMixin): root_link = "misago:admin:users:index" templates_dir = "misago/admin/users" model = User def get_form_class(self, request, target): add_is_active_fields = False add_admin_fields = False if not target.is_deleting_account: if not target.is_staff: add_is_active_fields = True elif request.user.is_superuser: add_is_active_fields = request.user.pk != target.pk if request.user.is_superuser: add_admin_fields = request.user.pk != target.pk return EditUserFormFactory( self.form_class, target, add_is_active_fields=add_is_active_fields, add_admin_fields=add_admin_fields, ) class UsersList(UserAdmin, generic.ListView): items_per_page = 24 ordering = [ ("-id", _("From newest")), ("id", _("From oldest")), ("slug", _("A to z")), ("-slug", _("Z to a")), ("-posts", _("Biggest posters")), ("posts", _("Smallest posters")), ] selection_label = _("With users: 0") empty_selection_label = _("Select users") mass_actions = [ {"action": "activate", "name": _("Activate accounts")}, {"action": "ban", "name": _("Ban users"), "icon": "fa fa-lock"}, {"action": "request_data_download", "name": _("Request data download")}, { "action": "delete_accounts", "name": _("Delete accounts"), "confirmation": _("Are you sure you want to delete selected users?"), }, { "action": "delete_all", "name": _("Delete with content"), "confirmation": _( "Are you sure you want to delete selected users? " "This will also delete all content associated with their accounts." ), "is_atomic": False, }, ] def get_queryset(self): qs = super().get_queryset() return qs.select_related("rank") def get_filter_form(self, request): return create_filter_users_form() def action_activate(self, request, users): inactive_users = [] for user in users: if user.requires_activation: inactive_users.append(user) if not inactive_users: message = _("You have to select inactive users.") raise generic.MassActionError(message) else: activated_users_pks = [u.pk for u in inactive_users] queryset = User.objects.filter(pk__in=activated_users_pks) queryset.update(requires_activation=User.ACTIVATION_NONE) subject = _("Your account on %(forum_name)s forums has been activated") mail_subject = subject % {"forum_name": request.settings.forum_name} mail_users( inactive_users, mail_subject, "misago/emails/activation/by_admin", context={"settings": request.settings}, ) messages.success(request, _("Selected users accounts have been activated.")) def action_ban( self, request, users ): # pylint: disable=too-many-locals, too-many-nested-blocks, too-many-branches users = users.order_by("slug") for user in users: if user.is_superuser: message = _("%(user)s is super admin and can't be banned.") mesage = message % {"user": user.username} raise generic.MassActionError(mesage) form = BanUsersForm(users=users) if "finalize" in request.POST: form = BanUsersForm(request.POST, users=users) if form.is_valid(): cleaned_data = form.cleaned_data banned_values = [] ban_kwargs = { "user_message": cleaned_data.get("user_message"), "staff_message": cleaned_data.get("staff_message"), "expires_on": cleaned_data.get("expires_on"), } for user in users: for ban in cleaned_data["ban_type"]: banned_value = None if ban == "usernames": check_type = Ban.USERNAME banned_value = user.username.lower() if ban == "emails": check_type = Ban.EMAIL banned_value = user.email.lower() if ban == "domains": check_type = Ban.EMAIL banned_value = user.email.lower() at_pos = banned_value.find("@") banned_value = "*%s" % banned_value[at_pos:] if ban == "ip" and user.joined_from_ip: check_type = Ban.IP banned_value = user.joined_from_ip if ban in ("ip_first", "ip_two") and user.joined_from_ip: check_type = Ban.IP if ":" in user.joined_from_ip: ip_separator = ":" if "." in user.joined_from_ip: ip_separator = "." bits = user.joined_from_ip.split(ip_separator) if ban == "ip_first": formats = (bits[0], ip_separator) if ban == "ip_two": formats = (bits[0], ip_separator, bits[1], ip_separator) banned_value = "%s*" % ("".join(formats)) if banned_value and banned_value not in banned_values: ban_kwargs.update( {"check_type": check_type, "banned_value": banned_value} ) Ban.objects.create(**ban_kwargs) banned_values.append(banned_value) Ban.objects.invalidate_cache() messages.success(request, _("Selected users have been banned.")) return None return self.render( request, {"users": users, "form": form}, template_name="misago/admin/users/ban.html", ) def action_request_data_download(self, request, users): for user in users: if not user_has_data_download_request(user): request_user_data_download(user, requester=request.user) messages.success( request, _("Data download requests have been placed for selected users.") ) def action_delete_accounts(self, request, users): for user in users: if user == request.user: raise generic.MassActionError(_("You can't delete yourself.")) if user.is_staff or user.is_superuser: message = _("%(user)s is admin and can't be deleted.") % { "user": user.username } raise generic.MassActionError(message) for user in users: user.delete() messages.success(request, _("Selected users have been deleted.")) def action_delete_all(self, request, users): for user in users: if user == request.user: raise generic.MassActionError(_("You can't delete yourself.")) if user.is_staff or user.is_superuser: message = _("%(user)s is admin and can't be deleted.") % { "user": user.username } raise generic.MassActionError(message) for user in users: user.is_active = False user.save() delete_user_with_content.delay(user.pk) messages.success( request, _( "Selected users have disabled and been queued for deletion " "together with their content." ), ) class NewUser(UserAdmin, generic.ModelFormView): form_class = NewUserForm template_name = "new.html" message_submit = _('New user "%(user)s" has been registered.') def get_form(self, form_class, request, target): if request.method == "POST": return form_class( request.POST, request.FILES, instance=target, request=request ) return form_class(instance=target, request=request) def handle_form(self, form, request, target): new_user = User.objects.create_user( form.cleaned_data["username"], form.cleaned_data["email"], form.cleaned_data["new_password"], title=form.cleaned_data["title"], rank=form.cleaned_data.get("rank"), joined_from_ip=request.user_ip, ) if form.cleaned_data.get("roles"): new_user.roles.add(*form.cleaned_data["roles"]) new_user.update_acl_key() setup_new_user(request.settings, new_user) messages.success(request, self.message_submit % {"user": target.username}) return redirect("misago:admin:users:edit", pk=new_user.pk) class EditUser(UserAdmin, generic.ModelFormView): form_class = EditUserForm template_name = "edit.html" message_submit = _('User "%(user)s" has been edited.') def real_dispatch(self, request, target): target.old_username = target.username target.old_is_avatar_locked = target.is_avatar_locked return super().real_dispatch(request, target) def get_form(self, form_class, request, target): if request.method == "POST": return form_class( request.POST, request.FILES, instance=target, request=request ) return form_class(instance=target, request=request) def handle_form(self, form, request, target): target.username = target.old_username if target.username != form.cleaned_data.get("username"): target.set_username( form.cleaned_data.get("username"), changed_by=request.user ) if form.cleaned_data.get("new_password"): target.set_password(form.cleaned_data["new_password"]) if form.cleaned_data.get("email"): target.set_email(form.cleaned_data["email"]) if form.cleaned_data.get("is_avatar_locked"): if not target.old_is_avatar_locked: set_dynamic_avatar(target) if "is_staff" in form.fields and "is_superuser" in form.fields: target.is_staff = form.cleaned_data.get("is_staff") target.is_superuser = form.cleaned_data.get("is_superuser") if "is_active" in form.fields and "is_active_staff_message" in form.fields: target.is_active = form.cleaned_data.get("is_active") target.is_active_staff_message = form.cleaned_data.get( "is_active_staff_message" ) target.rank = form.cleaned_data.get("rank") target.roles.clear() target.roles.add(*form.cleaned_data["roles"]) target_acl = get_user_acl(target, request.cache_versions) set_user_signature( request, target, target_acl, form.cleaned_data.get("signature") ) profilefields.update_user_profile_fields(request, target, form) target.update_acl_key() target.save() if target.pk == request.user.pk: authorize_admin(request) update_session_auth_hash(request, target) messages.success(request, self.message_submit % {"user": target.username})