Bulk delete users (#11425)

Author: di

tests/unit/admin/test_routes.py

@@ -59,6 +59,11 @@ def test_includeme():
             "/admin/users/{user_id}/reset_password/",
             domain=warehouse,
         ),
+        pretend.call(
+            "admin.prohibited_user_names.bulk_add",
+            "/admin/prohibited_user_names/bulk/",
+            domain=warehouse,
+        ),
         pretend.call("admin.project.list", "/admin/projects/", domain=warehouse),
         pretend.call(
             "admin.project.detail",

tests/unit/admin/views/test_users.py

@@ -20,7 +20,7 @@
 from webob.multidict import MultiDict, NoVars
 
 from warehouse.accounts.interfaces import IUserService
-from warehouse.accounts.models import DisableReason
+from warehouse.accounts.models import DisableReason, ProhibitedUserName
 from warehouse.admin.views import users as views
 from warehouse.packaging.models import JournalEntry, Project
 
@@ -395,3 +395,60 @@ def test_resets_password_bad_confirm(self, db_request, monkeypatch):
         ]
         assert result.status_code == 303
         assert result.location == "/foobar"
+
+
+class TestBulkAddProhibitedUserName:
+    def test_get(self):
+        request = pretend.stub(method="GET")
+
+        assert views.bulk_add_prohibited_user_names(request) == {}
+
+    def test_bulk_add(self, db_request):
+        db_request.user = UserFactory.create()
+        db_request.method = "POST"
+
+        already_existing_prohibition = ProhibitedUserName(
+            name="prohibition-already-exists",
+            prohibited_by=db_request.user,
+            comment="comment",
+        )
+        db_request.db.add(already_existing_prohibition)
+
+        already_existing_user = UserFactory.create(username="user-already-exists")
+        UserFactory.create(username="deleted-user")
+
+        user_names = [
+            already_existing_prohibition.name,
+            already_existing_user.username,
+            "doesnt-already-exist",
+        ]
+
+        db_request.POST["users"] = "\n".join(user_names)
+
+        db_request.session = pretend.stub(
+            flash=pretend.call_recorder(lambda *a, **kw: None)
+        )
+        db_request.route_path = lambda a: "/admin/prohibited_user_names/bulk"
+
+        result = views.bulk_add_prohibited_user_names(db_request)
+
+        assert db_request.session.flash.calls == [
+            pretend.call(
+                f"Prohibited {len(user_names)!r} users",
+                queue="success",
+            )
+        ]
+        assert result.status_code == 303
+        assert result.headers["Location"] == "/admin/prohibited_user_names/bulk"
+
+        for user_name in user_names:
+            prohibition = (
+                db_request.db.query(ProhibitedUserName)
+                .filter(ProhibitedUserName.name == user_name)
+                .one()
+            )
+
+            assert prohibition.name == user_name
+            assert prohibition.prohibited_by == db_request.user
+
+            assert db_request.db.query(User).filter(User.name == user_name).count() == 0

warehouse/admin/routes.py

@@ -54,6 +54,11 @@ def includeme(config):
         "/admin/users/{user_id}/reset_password/",
         domain=warehouse,
     )
+    config.add_route(
+        "admin.prohibited_user_names.bulk_add",
+        "/admin/prohibited_user_names/bulk/",
+        domain=warehouse,
+    )
 
     # Project related Admin pages
     config.add_route("admin.project.list", "/admin/projects/", domain=warehouse)

warehouse/admin/templates/admin/prohibited_user_names/bulk.html

@@ -0,0 +1,45 @@
+{#
+ # Licensed under the Apache License, Version 2.0 (the "License");
+ # you may not use this file except in compliance with the License.
+ # You may obtain a copy of the License at
+ #
+ # http://www.apache.org/licenses/LICENSE-2.0
+ #
+ # Unless required by applicable law or agreed to in writing, software
+ # distributed under the License is distributed on an "AS IS" BASIS,
+ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ # See the License for the specific language governing permissions and
+ # limitations under the License.
+-#}
+{% extends "admin/base.html" %}
+
+{% block title %}Bulk User Name Prohibition{% endblock %}
+
+{% block content %}
+<div class="box box-primary">
+  <div class="box-header with-border">
+    <h3 class="box-title">Prohibit user name</h3>
+  </div>
+  <div class="box-body">
+    <p>
+    User names separated by whitespace. <b>Note: There is no confirmation step!</b>
+    </p>
+  </div>
+
+  <form method="POST" action="{{ request.route_path('admin.prohibited_user_names.bulk_add') }}">
+    <input name="csrf_token" type="hidden" value="{{ request.session.get_csrf_token() }}">
+    <div class="box-body">
+      <div class="form-group">
+        <label for="prohibitedUserName">User name(s)</label>
+        <textarea name="users" class="form-control" id="prohibitedUserName" rows="20" placeholder="Enter user name(s) to prohibit " {{ "disabled" if not request.has_permission('admin') }} autocomplete="off" autocorrect="off" autocapitalize="off"></textarea>
+      </div>
+    </div>
+
+    <div class="box-footer">
+      <div class="pull-right">
+        <button type="submit" class="btn btn-primary" title="{{ "Submitting requires superuser privileges" if not request.has_permission('admin') }}" {{ "disabled" if not request.has_permission('admin') }}>Submit</button>
+      </div>
+    </div>
+  </form>
+</div>
+{% endblock %}

warehouse/admin/views/users.py

@@ -19,7 +19,7 @@
 from paginate_sqlalchemy import SqlalchemyOrmPage as SQLAlchemyORMPage
 from pyramid.httpexceptions import HTTPBadRequest, HTTPNotFound, HTTPSeeOther
 from pyramid.view import view_config
-from sqlalchemy import or_
+from sqlalchemy import literal, or_
 from sqlalchemy.orm import joinedload
 from sqlalchemy.orm.exc import NoResultFound
 
@@ -180,20 +180,7 @@ def user_add_email(request):
     return HTTPSeeOther(request.route_path("admin.user.detail", user_id=user.id))
 
 
-@view_config(
-    route_name="admin.user.delete",
-    require_methods=["POST"],
-    permission="admin",
-    uses_session=True,
-    require_csrf=True,
-)
-def user_delete(request):
-    user = request.db.query(User).get(request.matchdict["user_id"])
-
-    if user.username != request.params.get("username"):
-        request.session.flash("Wrong confirmation input", queue="error")
-        return HTTPSeeOther(request.route_path("admin.user.detail", user_id=user.id))
-
+def _nuke_user(user, request):
     # Delete all the user's projects
     projects = request.db.query(Project).filter(
         Project.name.in_(
@@ -244,6 +231,24 @@ def user_delete(request):
             submitted_from=request.remote_addr,
         )
     )
+
+
+@view_config(
+    route_name="admin.user.delete",
+    require_methods=["POST"],
+    permission="admin",
+    uses_session=True,
+    require_csrf=True,
+)
+def user_delete(request):
+    user = request.db.query(User).get(request.matchdict["user_id"])
+
+    if user.username != request.params.get("username"):
+        request.session.flash("Wrong confirmation input", queue="error")
+        return HTTPSeeOther(request.route_path("admin.user.detail", user_id=user.id))
+
+    _nuke_user(user, request)
+
     request.session.flash(f"Nuked user {user.username!r}", queue="success")
     return HTTPSeeOther(request.route_path("admin.user.list"))
 
@@ -269,3 +274,48 @@ def user_reset_password(request):
 
     request.session.flash(f"Reset password for {user.username!r}", queue="success")
     return HTTPSeeOther(request.route_path("admin.user.detail", user_id=user.id))
+
+
+@view_config(
+    route_name="admin.prohibited_user_names.bulk_add",
+    renderer="admin/prohibited_user_names/bulk.html",
+    permission="admin",
+    uses_session=True,
+    require_methods=False,
+)
+def bulk_add_prohibited_user_names(request):
+    if request.method == "POST":
+        user_names = request.POST.get("users", "").split()
+
+        for user_name in user_names:
+
+            # Check to make sure the prohibition doesn't already exist.
+            if (
+                request.db.query(literal(True))
+                .filter(
+                    request.db.query(ProhibitedUserName)
+                    .filter(ProhibitedUserName.name == user_name.lower())
+                    .exists()
+                )
+                .scalar()
+            ):
+                continue
+
+            # Go through and delete the usernames
+
+            user = request.db.query(User).filter(User.username == user_name).first()
+            if user is not None:
+                _nuke_user(user, request)
+            else:
+                request.db.add(
+                    ProhibitedUserName(
+                        name=user_name.lower(),
+                        comment="nuked",
+                        prohibited_by=request.user,
+                    )
+                )
+
+        request.session.flash(f"Prohibited {len(user_names)!r} users", queue="success")
+
+        return HTTPSeeOther(request.route_path("admin.prohibited_user_names.bulk_add"))
+    return {}