Add bulk banning functionality to the admin

src/olympia/core/__init__.py

@@ -53,6 +53,14 @@ def select_request_metadata(headers):
     }
 
 
+@contextlib.contextmanager
+def override_user(user):
+    original_user = get_user()
+    set_user(user)
+    yield
+    set_user(original_user)
+
+
 @contextlib.contextmanager
 def override_remote_addr_or_metadata(*, ip_address=None, metadata=None):
     """Override value returned by get_remote_addr() for a specific context."""

src/olympia/core/tests/test_init.py

@@ -14,6 +14,25 @@ def test_override_remote_addr_or_metadata():
     assert core.get_remote_addr() == original
 
 
+def test_override_user():
+    user = UserProfile()
+    core.set_user(AnonymousUser())
+    with core.override_user(user):
+        assert core.get_user() == user
+    assert core.get_user() is None
+
+    core.set_user(user)
+    with core.override_user(None):
+        assert core.get_user() is None
+    assert core.get_user() == user
+
+    with core.override_user(AnonymousUser()):
+        assert core.get_user() is None
+    assert core.get_user() == user
+
+    core.set_user(None)
+
+
 def test_set_get_user_anonymous():
     core.set_user(AnonymousUser())
     assert core.get_user() is None

src/olympia/lib/settings_base.py

@@ -769,6 +769,7 @@ def get_language_url_map():
     'olympia.activity.tasks.create_ratinglog': {'queue': 'adhoc'},
     'olympia.files.tasks.extract_host_permissions': {'queue': 'adhoc'},
     'olympia.lib.crypto.tasks.bump_and_resign_addons': {'queue': 'adhoc'},
+    'olympia.users.tasks.bulk_ban': {'queue': 'adhoc'},
     'olympia.users.tasks.restrict_banned_users': {'queue': 'adhoc'},
     # Misc AMO tasks.
     'olympia.blocklist.tasks.monitor_remote_settings': {'queue': 'amo'},

src/olympia/users/admin.py

@@ -6,7 +6,7 @@
 from django.db import models
 from django.db.models import Count, Q
 from django.db.utils import IntegrityError
-from django.forms import TextInput
+from django.forms import HiddenInput, TextInput
 from django.http import (
     Http404,
     HttpResponseForbidden,
@@ -25,7 +25,12 @@
 from olympia.activity.models import ActivityLog, RequestFingerprintLog
 from olympia.addons.models import Addon, AddonUser
 from olympia.amo.admin import AMOModelAdmin
-from olympia.amo.utils import backup_storage_enabled, create_signed_url_for_file_backup
+from olympia.amo.templatetags.jinja_helpers import vite_asset
+from olympia.amo.utils import (
+    backup_storage_enabled,
+    chunked,
+    create_signed_url_for_file_backup,
+)
 from olympia.api.models import APIKey, APIKeyConfirmation
 from olympia.bandwagon.models import Collection
 from olympia.constants.activity import LOG_STORE_IPS
@@ -45,6 +50,7 @@
     UserProfile,
     UserRestrictionHistory,
 )
+from .tasks import bulk_ban
 
 
 class GroupUserInline(admin.TabularInline):
@@ -107,6 +113,9 @@ def picture_link(self, obj):
 
 @admin.register(UserProfile)
 class UserAdmin(AMOModelAdmin):
+    class Media:
+        css = {'all': (vite_asset('css/admin-user.less'),)}
+
     list_filter = ('banned',)
     list_display = (
         '__str__',
@@ -242,6 +251,11 @@ def wrapper(*args, **kwargs):
 
         urlpatterns = super().get_urls()
         custom_urlpatterns = [
+            re_path(
+                r'^bulk_ban/$',
+                wrap(self.bulk_ban_view),
+                name='users_userprofile_bulk_ban',
+            ),
             re_path(
                 r'^(?P<object_id>.+)/ban/$',
                 wrap(self.ban_view),
@@ -284,6 +298,19 @@ def get_actions(self, request):
 
         return actions
 
+    def changelist_view(self, request, extra_context=None):
+        extra_context = extra_context or {}
+        extra_context['has_users_edit_permission'] = acl.action_allowed_for(
+            request.user, amo.permissions.USERS_EDIT
+        )
+        extra_context['has_users_ban_permission'] = acl.action_allowed_for(
+            request.user, amo.permissions.USERS_BAN
+        )
+        return super().changelist_view(
+            request,
+            extra_context=extra_context,
+        )
+
     def change_view(self, request, object_id, form_url='', extra_context=None):
         extra_context = extra_context or {}
         extra_context['has_users_edit_permission'] = acl.action_allowed_for(
@@ -425,10 +452,47 @@ def delete_picture_view(self, request, object_id, extra_context=None):
             reverse('admin:users_userprofile_change', args=(obj.pk,))
         )
 
+    def bulk_ban_view(self, request, extra_context=None):
+        if not acl.action_allowed_for(request.user, amo.permissions.USERS_BAN):
+            return HttpResponseForbidden()
+
+        user_ids_to_ban = None
+
+        if request.method == 'POST':
+            form = forms.BulkBanForm(request.POST)
+            if form.is_valid():
+                user_ids_to_ban = form.cleaned_data['user_ids']
+                form.fields['user_ids'].widget = HiddenInput()
+                if request.POST.get('post'):
+                    self._trigger_bulk_ban_task(request, user_ids_to_ban)
+                    return HttpResponseRedirect(
+                        reverse('admin:users_userprofile_changelist')
+                    )
+        else:
+            form = forms.BulkBanForm()
+        context = {
+            'title': 'Bulk ban users',
+            'form': form,
+            'user_ids_to_ban': user_ids_to_ban,
+            **self.admin_site.each_context(request),
+            'opts': self.opts,
+            'media': self.media + form.media,
+            **(extra_context or {}),
+        }
+        return TemplateResponse(
+            request, 'admin/users/userprofile/bulk_ban.html', context
+        )
+
+    def _trigger_bulk_ban_task(self, request, user_ids):
+        for chunk in chunked(user_ids, 50):
+            bulk_ban.delay(list(chunk), user_responsible_id=request.user.pk)
+        self.message_user(
+            request,
+            f'Bulk-ban for {len(user_ids)} user id(s) will be processed shortly.',
+        )
+
     def ban_action(self, request, qs):
-        qs.ban_and_disable_related_content(hard_block_addons=True)
-        kw = {'users': ', '.join(str(user) for user in qs)}
-        self.message_user(request, 'The users "%(users)s" have been banned.' % kw)
+        self._trigger_bulk_ban_task(request, qs.values_list('id', flat=True))
 
     ban_action.short_description = 'Ban selected users'
 

src/olympia/users/forms.py

@@ -61,3 +61,18 @@ def clean(self):
             data['network'] = f'{ip_address}/32'
 
         return data
+
+
+class BulkBanForm(forms.Form):
+    user_ids = forms.CharField(
+        label='', required=True, widget=forms.Textarea(attrs={'rows': 30, 'cols': 80})
+    )
+
+    def clean_user_ids(self):
+        data = self.cleaned_data.get('user_ids', '')
+        user_ids = {
+            id_.strip() for id_ in data.strip().splitlines() if id_.strip().isdigit()
+        }
+        if not user_ids:
+            raise forms.ValidationError('This field must contain a least one user id')
+        return sorted(user_ids)

src/olympia/users/models.py

@@ -173,6 +173,7 @@ def ban_and_disable_related_content(
                     RESTRICTION_TYPES.ADDON_SUBMISSION,
                     RESTRICTION_TYPES.RATING,
                 ]
+                if user.email
             ],
             ignore_conflicts=True,
         )
@@ -318,9 +319,10 @@ def unban_and_reenable_related_content(self, *, skip_activity_log=False):
             user.deleted = False
             user.banned = None
             user.save()
-            EmailUserRestriction.objects.filter(
-                email_pattern=EmailUserRestriction.normalize_email(user.email)
-            ).delete()
+            if user.email:
+                EmailUserRestriction.objects.filter(
+                    email_pattern=EmailUserRestriction.normalize_email(user.email)
+                ).delete()
         if blocklist_submissions_pks:
             user_responsible = core.get_user() or get_task_user()
             revert_published_blocklist_submissions.delay(

src/olympia/users/tasks.py

@@ -13,6 +13,7 @@
 from requests.exceptions import HTTPError, Timeout
 
 import olympia.core.logger
+from olympia import core
 from olympia.amo.celery import task
 from olympia.amo.decorators import set_modified_on, use_primary_db
 from olympia.amo.templatetags.jinja_helpers import absolutify
@@ -278,3 +279,20 @@ def bulk_add_disposable_email_domains(entries: list[tuple[str, str]], batch_size
         f'Processed {len(processed_domains)} domains: '
         f'{[obj.domain for obj in processed_domains]}'
     )
+
+
+@task
+@use_primary_db
+def bulk_ban(ids, *, user_responsible_id):
+    task_log.info(
+        '[1@None] Bulk-banning users %s-%s [%d] with %s.',
+        ids[0],
+        ids[-1],
+        len(ids),
+        user_responsible_id,
+    )
+    user_responsible = UserProfile.objects.get(pk=user_responsible_id)
+    with core.override_user(user_responsible):
+        UserProfile.objects.filter(pk__in=ids).order_by(
+            'pk'
+        ).ban_and_disable_related_content()

src/olympia/users/templates/admin/users/userprofile/bulk_ban.html

@@ -0,0 +1,36 @@
+{% extends "admin/base_site.html" %}
+{% load i18n admin_urls static %}
+
+{% block extrahead %}
+  {{ block.super }}
+  {{ media }}
+  <script src="{% static 'admin/js/cancel.js' %}" async></script>
+{% endblock %}
+
+{% block bodyclass %}{{ block.super }} app-{{ opts.app_label }} model-{{ opts.model_name }} bulk-ban-confirmation{% endblock %}
+
+{% block breadcrumbs %}
+<div class="breadcrumbs">
+<a href="{% url 'admin:index' %}">Home</a>
+&rsaquo; <a href="{% url 'admin:app_list' app_label=opts.app_label %}">{{ opts.app_config.verbose_name }}</a>
+&rsaquo; <a href="{% url opts|admin_urlname:'changelist' %}">{{ opts.verbose_name_plural|capfirst }}</a>
+&rsaquo; Bulk ban
+</div>
+{% endblock %}
+
+{% block content %}
+<form method="post">
+  {% csrf_token %}
+  {% if user_ids_to_ban %}
+    <p>You're about to bulk-ban {{ user_ids_to_ban|length }} user(s). Are you sure ?</p>
+    {{ form.as_p }}
+    <input type="hidden" name="post" value="yes">
+    <input type="submit" value="Yes, I'm sure">
+    <a href="#" class="button cancel-link">No, take me back</a>
+  {% else %}
+    <p>Paste user ids to ban below, separated by newlines (only lines containing digits will be considered). A confirmation will be displayed on the next page.</p>
+    {{ form.as_p }}
+    <input type="submit" value="Submit">
+  {% endif %}
+</form>
+{% endblock content %}

src/olympia/users/templates/admin/users/userprofile/change_list_object_tools.html

@@ -0,0 +1,12 @@
+{% extends "admin/change_list_object_tools.html" %}
+{% load i18n admin_urls %}
+
+{% block object-tools-items %}
+  {{ block.super }}
+  {% if has_users_ban_permission %}
+  <li>
+    {% url 'admin:users_userprofile_bulk_ban' as ban_url %}
+    <a href="{{ ban_url }}">Bulk ban</a>
+  </li>
+  {% endif %}
+{% endblock %}

src/olympia/users/tests/test_admin.py

@@ -65,6 +65,19 @@ def test_list(self):
         assert format_datetime(banned_user.banned) in doc('#result_list').text()
         assert str(user) in doc('#result_list').text()
         assert str(user.email) in doc('#result_list').text()
+        # No extra permissions: no object tools
+        assert doc('.object-tools a') == []
+
+    def test_list_has_bulk_ban_link(self):
+        user = user_factory(email='someone@mozilla.com')
+        self.grant_permission(user, 'Users:Edit')
+        self.grant_permission(user, 'Users:Ban')
+        self.client.force_login(user)
+        response = self.client.get(self.list_url)
+        doc = pq(response.content)
+        assert doc('.object-tools a').attr('href') == reverse(
+            'admin:users_userprofile_bulk_ban'
+        )
 
     def test_search_by_email_simple(self):
         user = user_factory(email='someone@mozilla.com')
@@ -787,6 +800,71 @@ def test_ban(self):
         assert alog.action == amo.LOG.ADMIN_USER_BANNED.id
         assert alog.arguments == [self.user]
 
+    def test_bulk_ban_no_permission(self):
+        user = user_factory(email='someone@mozilla.com')
+        self.client.force_login(user)
+        url = reverse('admin:users_userprofile_bulk_ban')
+        response = self.client.get(url)
+        assert response.status_code == 403
+
+    def test_bulk_ban(self):
+        target1 = user_factory()
+        target2 = user_factory()
+        innocent = user_factory()
+        user = user_factory(email='someone@mozilla.com')
+        self.grant_permission(user, 'Users:Ban')
+        self.client.force_login(user)
+        url = reverse('admin:users_userprofile_bulk_ban')
+        response = self.client.get(url)
+        assert response.status_code == 200
+        doc = pq(response.content)
+        assert doc('textarea#id_user_ids')
+
+        data = {'user_ids': f'{target2.pk} \n{target1.pk}\n '}
+        response = self.client.post(url, data)
+        assert response.status_code == 200
+        doc = pq(response.content)
+        assert 'about to bulk-ban 2 user(s). Are you sure ?' in response.content.decode(
+            'utf-8'
+        )
+        assert doc('input#id_user_ids')
+        assert doc('input#id_user_ids')[0].attrib['type'] == 'hidden'
+        assert doc('input#id_user_ids')[0].value == data['user_ids']
+        assert doc('input[name=post]')
+        assert doc('input[name=post]')[0].attrib['type'] == 'hidden'
+        assert doc('input[name=post]')[0].value == 'yes'
+
+        data['post'] = 'yes'
+        response = self.client.post(url, data)
+        assert response.status_code == 302
+        self.assertCloseToNow(target1.reload().banned)
+        self.assertCloseToNow(target2.reload().banned)
+        assert not innocent.reload().banned
+        assert not user.reload().banned
+
+        activities = ActivityLog.objects.filter(
+            action=amo.LOG.ADMIN_USER_BANNED.id
+        ).order_by('pk')
+        assert len(activities) == 2
+        assert activities[0].user == user
+        assert activities[0].arguments == [target1]
+        assert activities[1].user == user
+        assert activities[1].arguments == [target2]
+
+    def test_bulk_ban_invalid(self):
+        user = user_factory(email='someone@mozilla.com')
+        self.grant_permission(user, 'Users:Ban')
+        self.client.force_login(user)
+        url = reverse('admin:users_userprofile_bulk_ban')
+        data = {'user_ids': 'invalid'}
+        response = self.client.post(url, data)
+        assert response.status_code == 200
+        doc = pq(response.content)
+        assert (
+            doc('form .errorlist')[0].text_content()
+            == 'This field must contain a least one user id'
+        )
+
     def test_unban(self):
         unban_url = reverse('admin:users_userprofile_unban', args=(self.user.pk,))
         wrong_unban_url = reverse(