1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
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)
|
