"""Administrative endpoints for user management.""" import sqlite3 from functools import reduce from typing import Sequence from flask import jsonify, Response, Blueprint, make_response, current_app as app from gn_libs import sqlite3 as db from email_validator import validate_email, EmailNotValidError from gn_libs.privileges.system import can_create_or_delete_user from gn_auth.auth.errors import ( PasswordError, UsernameError, ForbiddenAccess, UserRegistrationError) from gn_auth.auth.requests import request_json from gn_auth.auth.authentication.oauth2.resource_server import require_oauth from gn_auth.auth.authorisation.resources.system.models import user_roles_on_system from gn_auth.auth.authorisation.users.admin.models import create_verified_user from gn_auth.auth.authorisation.users.views import ( validate_password, validate_username) users = Blueprint("users", __name__) @users.route("/create", methods=["POST"]) def create_user() -> Response: """Create a new user. Requires system:user:create-user privilege.""" with (require_oauth.acquire("profile") as token, db.connection(app.config["AUTH_DB"]) as conn): u_roles = user_roles_on_system(conn, token.user) if not can_create_or_delete_user(tuple( priv.privilege_id for role in u_roles for priv in role.privileges)): raise ForbiddenAccess( "You need the `system:user:create-user` privilege.") form = request_json() errors = {} try: email = validate_email( form.get("email", "").strip(), check_deliverability=False) except EmailNotValidError as enve: errors["email"] = ( f"E-Mail error: {'==>'.join(str(arg) for arg in enve.args)}") try: username = validate_username(form.get("name", "").strip()) except UsernameError as uerr: errors["name"] = str(uerr.args[0]) try: passwd = validate_password( form.get("password", "").strip(), form.get("password", "").strip()) except PasswordError as perr: errors["password"] = str(perr.args[0]) if len(tuple(errors.keys())) > 0: raise UserRegistrationError(tuple(errors.values())) user = create_verified_user(conn, email.normalized, username, passwd) return make_response(jsonify({ "user_id": str(user.user_id), "email": user.email, "name": user.name }), 201) def __delete_users_individually__(cursor, user_ids, tables): """Recovery function with dismal performance.""" _errors = tuple() for _user_id in user_ids: for _table, _col in tables: try: cursor.execute( f"DELETE FROM {_table} WHERE {_col}=?", (str(_user_id),)) except sqlite3.IntegrityError: _errors = _errors + ( (("user_id", _user_id), ("reason", f"User has data in table {_table}")),) return _errors def __fetch_non_deletable_users__(cursor, ids_and_reasons): """Fetch detail for non-deletable users.""" def __merge__(acc, curr): _curr = dict(curr) _this_dict = acc.get( curr["user_id"], {"reasons": tuple()}) _this_dict["reasons"] = _this_dict["reasons"] + (_curr["reason"],) return {**acc, curr["user_id"]: _this_dict} _reasons_by_id = reduce(__merge__, (dict(row) for row in ids_and_reasons), {}) _user_ids = tuple(_reasons_by_id.keys()) _paramstr = ", ".join(["?"] * len(_user_ids)) cursor.execute(f"SELECT * FROM users WHERE user_id IN ({_paramstr})", _user_ids) return tuple({ "user": dict(row), "reasons": _reasons_by_id[row["user_id"]]["reasons"] } for row in cursor.fetchall()) def __non_deletable_with_reason__( user_ids: tuple[str, ...], dbrows: Sequence[sqlite3.Row], reason: str ) -> tuple[tuple[tuple[str, str], tuple[str, str]], ...]: """Build a list of 'non-deletable' user objects.""" return tuple((("user_id", _uid), ("reason", reason)) for _uid in user_ids if _uid in tuple(row["user_id"] for row in dbrows)) @users.route("/delete", methods=["POST"]) def delete_users() -> Response: """Delete the specified users. Requires system:user:delete-user privilege.""" with (require_oauth.acquire("profile user role") as _token, db.connection(app.config["AUTH_DB"]) as conn, db.cursor(conn) as cursor): u_roles = user_roles_on_system(conn, _token.user) if not can_create_or_delete_user(tuple( priv.privilege_id for role in u_roles for priv in role.privileges)): raise ForbiddenAccess( "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() if str(_token.user.user_id) in _user_ids: _non_deletable.add( (("user_id", str(_token.user.user_id),), ("reason", "You are not allowed to delete yourself."))) cursor.execute("SELECT user_id FROM group_users") _group_members = tuple(row["user_id"] for row in cursor.fetchall()) _non_deletable.update(__non_deletable_with_reason__( _user_ids, cursor.fetchall(), "User is member of a user group.")) cursor.execute("SELECT user_id FROM oauth2_clients;") _non_deletable.update(__non_deletable_with_reason__( _user_ids, cursor.fetchall(), "User is registered owner of an OAuth client.")) _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(__non_deletable_with_reason__( _user_ids, cursor.fetchall(), f"User holds on of the following roles: {_important_roles}")) _delete = tuple(uid for uid in _user_ids if uid not in (dict(row)["user_id"] for row 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")) try: for _table, _col in _dependent_tables: cursor.execute( f"DELETE FROM {_table} WHERE {_col} IN ({_paramstr})", _delete) except sqlite3.IntegrityError: _non_deletable.update(__delete_users_individually__( cursor, _delete, _dependent_tables)) _not_deleted = __fetch_non_deletable_users__( cursor, _non_deletable) _delete = tuple(# rebuild with those that failed. _user_id for _user_id in _delete if _user_id not in tuple(row["user"]["user_id"] for row in _not_deleted)) _paramstr = ", ".join(["?"] * len(_delete)) cursor.execute( f"DELETE FROM users WHERE user_id IN ({_paramstr})", _delete) _deleted_rows = cursor.rowcount return jsonify({ "total-requested": len(_user_ids), "total-deleted": _deleted_rows, "not-deleted": _not_deleted, "deleted": _deleted_rows, "message": ( f"Successfully deleted {_deleted_rows} users." + (" Some users could not be deleted." if len(_user_ids) - _deleted_rows > 0 else "")) }) _not_deleted = __fetch_non_deletable_users__(cursor, _non_deletable) return make_response(jsonify({ "total-requested": len(_user_ids), "total-deleted": 0, "not-deleted": _not_deleted, "deleted": 0, "error": "Zero users were deleted", "error_description": ( "No users were selected for deletion." if len(_user_ids) == 0 else ("The selected users are system administrators, group " "members, or resource owners.")) }), 400)