Delete system users.HEADmain

1 files changed, 82 insertions, 0 deletions

diff --git a/gn_auth/auth/authorisation/users/views.py b/gn_auth/auth/authorisation/users/views.py
index 5140bcb..be4296b 100644
--- a/gn_auth/auth/authorisation/users/views.py
+++ b/gn_auth/auth/authorisation/users/views.py
@@ -28,6 +28,7 @@ from gn_auth.auth.requests import request_json
 from gn_auth.auth.db import sqlite3 as db
 from gn_auth.auth.db.sqlite3 import with_db_connection
 
+from gn_auth.auth.authorisation.resources.checks import authorised_for2
 from gn_auth.auth.authorisation.resources.models import (
     user_resources as _user_resources)
 from gn_auth.auth.authorisation.roles.models import (
@@ -39,6 +40,7 @@ from gn_auth.auth.errors import (
     NotFoundError,
     UsernameError,
     PasswordError,
+    AuthorisationError,
     UserRegistrationError)
 
 
@@ -548,3 +550,83 @@ def change_password(forgot_password_token):
         flash("Both the password and its confirmation MUST be provided!",
               "alert-danger")
         return change_password_page
+
+
+@users.route("/delete", methods=["POST"])
+@require_oauth("profile user role")
+def delete_users():
+    """Delete the specified user."""
+    with (require_oauth.acquire("profile") as _token,
+          db.connection(current_app.config["AUTH_DB"]) as conn,
+          db.cursor(conn) as cursor):
+        if not authorised_for2(conn,
+                               _token.user,
+                               system_resource(conn),
+                               ("system:user:delete-user",)):
+            raise AuthorisationError(
+                "You need the `system:user:delete-user` privilege to delete "
+                "users from the system.")
+
+        _form = request_json()
+        _user_ids = _form.get("user_ids", [])
+        _non_deletable = set((str(_token.user.user_id),))
+
+        cursor.execute("SELECT user_id FROM group_users")
+        _non_deletable.update(row["user_id"] for row in cursor.fetchall())
+
+        cursor.execute("SELECT user_id FROM oauth2_clients;")
+        _non_deletable.update(row["user_id"] for row in cursor.fetchall())
+
+        _important_roles = (
+            "group-leader",
+            "resource-owner",
+            "system-administrator",
+            "inbredset-group-owner")
+        _paramstr = ",".join(["?"] * len(_important_roles))
+        cursor.execute(
+            "SELECT DISTINCT user_roles.user_id FROM user_roles "
+            "INNER JOIN roles ON user_roles.role_id=roles.role_id "
+            f"WHERE roles.role_name IN ({_paramstr})",
+            _important_roles)
+        _non_deletable.update(row["user_id"] for row in cursor.fetchall())
+
+        _delete = tuple(uid for uid in _user_ids if uid not in _non_deletable)
+        _paramstr = ", ".join(["?"] * len(_delete))
+        if len(_delete) > 0:
+            _dependent_tables = (
+                ("authorisation_code", "user_id"),
+                ("forgot_password_tokens", "user_id"),
+                ("group_join_requests", "requester_id"),
+                ("jwt_refresh_tokens", "user_id"),
+                ("oauth2_tokens", "user_id"),
+                ("user_credentials", "user_id"),
+                ("user_roles", "user_id"),
+                ("user_verification_codes", "user_id"))
+            for _table, _col in _dependent_tables:
+                cursor.execute(
+                    f"DELETE FROM {_table} WHERE {_col} IN ({_paramstr})",
+                    _delete)
+
+            cursor.execute(
+                f"DELETE FROM users WHERE user_id IN ({_paramstr})",
+                _delete)
+            _deleted_rows = cursor.rowcount
+            _diff = len(_user_ids) - _deleted_rows
+            return jsonify({
+                "total-requested": len(_user_ids),
+                "total-deleted": _deleted_rows,
+                "not-deleted": _diff,
+                "message": (
+                    f"Successfully deleted {_deleted_rows} users." +
+                    (f" Some users could not be deleted." if _diff > 0 else ""))
+            })
+
+    return jsonify({
+        "total-requested": len(_user_ids),
+        "total-deleted": 0,
+        "not-deleted": len(_user_ids),
+        "error": "Zero users were deleted",
+        "error_description": (
+            "Either no users were selected or all the selected users are "
+            "system administrators, group members, or resource owners.")
+    }), 400