diff options
Diffstat (limited to 'gn_auth/auth/system/admin/users.py')
| -rw-r--r-- | gn_auth/auth/system/admin/users.py | 232 |
1 files changed, 232 insertions, 0 deletions
diff --git a/gn_auth/auth/system/admin/users.py b/gn_auth/auth/system/admin/users.py new file mode 100644 index 0000000..7dfc7ca --- /dev/null +++ b/gn_auth/auth/system/admin/users.py @@ -0,0 +1,232 @@ +"""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) |