forked from openedx/openedx-authz
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathusers.py
More file actions
456 lines (370 loc) · 16.5 KB
/
users.py
File metadata and controls
456 lines (370 loc) · 16.5 KB
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
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
"""User-related API methods for role assignments and retrievals.
This module provides user-related API methods for assigning roles to users,
unassigning roles from users, and retrieving roles assigned to users within
the Open edX AuthZ framework.
These methods internally namespace user identifiers to ensure consistency
with the role management system, which uses namespaced subjects
(e.g., 'user^john_doe').
"""
from django.contrib.auth import get_user_model
from django.db.models import Q
from openedx_authz.api.data import (
ActionData,
PermissionData,
RoleAssignmentData,
RoleData,
ScopeData,
SuperAdminAssignmentData,
UserAssignments,
UserAssignmentsFilter,
UserData,
)
from openedx_authz.api.permissions import is_subject_allowed
from openedx_authz.api.roles import (
assign_role_to_subject_in_scope,
batch_assign_role_to_subjects_in_scope,
batch_unassign_role_from_subjects_in_scope,
get_all_subject_role_assignments_in_scope,
get_role_assignments,
get_scopes_for_subject_and_permission,
get_subject_role_assignments,
get_subject_role_assignments_for_role_in_scope,
get_subject_role_assignments_in_scope,
get_subjects_for_role_in_scope,
unassign_role_from_subject_in_scope,
unassign_subject_from_all_roles,
)
from openedx_authz.api.utils import filter_user_assignments, get_user_assignment_map
from openedx_authz.utils import get_user_by_username_or_email
User = get_user_model()
__all__ = [
"assign_role_to_user_in_scope",
"batch_assign_role_to_users_in_scope",
"unassign_role_from_user",
"batch_unassign_role_from_users",
"get_user_role_assignments",
"get_user_role_assignments_in_scope",
"get_user_role_assignments_for_role_in_scope",
"get_user_role_assignments_filtered",
"get_all_user_role_assignments_in_scope",
"get_visible_role_assignments_for_user",
"get_visible_user_role_assignments_filtered_by_current_user",
"is_user_allowed",
"get_scopes_for_user_and_permission",
"get_users_for_role_in_scope",
"unassign_all_roles_from_user",
"validate_users",
"get_superadmin_assignments",
]
def assign_role_to_user_in_scope(user_external_key: str, role_external_key: str, scope_external_key: str) -> bool:
"""Assign a role to a user in a specific scope.
Args:
user (str): ID of the user (e.g., 'john_doe').
role_external_key (str): Name of the role to assign.
scope (str): Scope in which to assign the role.
Returns:
bool: True if the role was assigned successfully, False otherwise.
"""
return assign_role_to_subject_in_scope(
UserData(external_key=user_external_key),
RoleData(external_key=role_external_key),
ScopeData(external_key=scope_external_key),
)
def batch_assign_role_to_users_in_scope(users: list[str], role_external_key: str, scope_external_key: str):
"""Assign a role to multiple users in a specific scope.
Args:
users (list of str): List of user IDs (e.g., ['john_doe', 'jane_smith']).
role_external_key (str): Name of the role to assign.
scope (str): Scope in which to assign the role.
"""
namespaced_users = [UserData(external_key=username) for username in users]
batch_assign_role_to_subjects_in_scope(
namespaced_users,
RoleData(external_key=role_external_key),
ScopeData(external_key=scope_external_key),
)
def unassign_role_from_user(user_external_key: str, role_external_key: str, scope_external_key: str):
"""Unassign a role from a user in a specific scope.
Args:
user_external_key (str): ID of the user (e.g., 'john_doe').
role_external_key (str): Name of the role to unassign.
scope_external_key (str): Scope in which to unassign the role.
Returns:
bool: True if the role was unassigned successfully, False otherwise.
"""
return unassign_role_from_subject_in_scope(
UserData(external_key=user_external_key),
RoleData(external_key=role_external_key),
ScopeData(external_key=scope_external_key),
)
def batch_unassign_role_from_users(users: list[str], role_external_key: str, scope_external_key: str):
"""Unassign a role from multiple users in a specific scope.
Args:
users (list of str): List of user IDs (e.g., ['john_doe', 'jane_smith']).
role_external_key (str): Name of the role to unassign.
scope (str): Scope in which to unassign the role.
"""
namespaced_users = [UserData(external_key=user) for user in users]
batch_unassign_role_from_subjects_in_scope(
namespaced_users,
RoleData(external_key=role_external_key),
ScopeData(external_key=scope_external_key),
)
def get_user_role_assignments(user_external_key: str) -> list[RoleAssignmentData]:
"""Get all roles for a user across all scopes.
Args:
user_external_key (str): ID of the user (e.g., 'john_doe').
Returns:
list[RoleAssignmentData]: A list of role assignments and all their metadata assigned to the user.
"""
return get_subject_role_assignments(UserData(external_key=user_external_key))
def get_user_role_assignments_in_scope(user_external_key: str, scope_external_key: str) -> list[RoleAssignmentData]:
"""Get the roles assigned to a user in a specific scope.
Args:
user (str): ID of the user (e.g., 'john_doe').
scope (str): Scope in which to retrieve the roles.
Returns:
list[RoleAssignmentData]: A list of role assignments assigned to the user in the specified scope.
"""
return get_subject_role_assignments_in_scope(
UserData(external_key=user_external_key),
ScopeData(external_key=scope_external_key),
)
def get_user_role_assignments_for_role_in_scope(
role_external_key: str, scope_external_key: str
) -> list[RoleAssignmentData]:
"""Get all users assigned to a specific role across all scopes.
Args:
role_external_key (str): Name of the role (e.g., 'instructor').
scope (str): Scope in which to retrieve the role assignments.
Returns:
list[RoleAssignmentData]: List of users assigned to the specified role in the given scope.
"""
return get_subject_role_assignments_for_role_in_scope(
RoleData(external_key=role_external_key),
ScopeData(external_key=scope_external_key),
)
def get_visible_user_role_assignments_filtered_by_current_user(
user_external_key: str,
orgs: list[str] = None,
roles: list[str] = None,
allowed_for_user_external_key: str = None,
) -> list[RoleAssignmentData]:
"""
Get role assignments for a specific user, filtered by orgs and/or roles,
and only include assignments that the specified user has permission to view.
Args:
user_external_key: The user to get assignments for (e.g., 'john_doe').
orgs: Optional list of orgs to filter by (e.g., ['edX', 'MITx']).
roles: Optional list of roles to filter by (e.g., ['library_admin']).
allowed_for_user_external_key: The username to check permissions against (e.g., 'john_doe').
Returns:
list[RoleAssignmentData]: A list of role assignments for the user, filtered by orgs/roles and permissions.
"""
user_role_assignments = get_user_role_assignments(user_external_key=user_external_key)
# Filter assignments based on the user's permissions
user_role_assignments = _filter_allowed_assignments(
user_external_key=allowed_for_user_external_key,
assignments=user_role_assignments,
)
# Only include assignments whose subject corresponds to an active user,
# consistent with get_superadmin_assignments which filters by is_active=True.
active_usernames = set(
User.objects.filter(
username__in=[a.subject.username for a in user_role_assignments],
is_active=True,
).values_list("username", flat=True)
)
user_role_assignments = [a for a in user_role_assignments if a.subject.username in active_usernames]
if orgs:
# Filter by orgs
user_role_assignments = [a for a in user_role_assignments if getattr(a.scope, "org", None) in orgs]
if roles:
# Filter by roles
user_role_assignments = [
a for a in user_role_assignments if any(role.external_key in roles for role in a.roles)
]
return user_role_assignments
def get_user_role_assignments_filtered(
*,
user_external_key: str | None = None,
role_external_key: str | None = None,
scope_external_key: str | None = None,
) -> list[RoleAssignmentData]:
"""Get role assignments filtered by user, role, and/or scope.
This function provides flexible filtering of role assignments by any combination
of user, role, and scope. At least one filter parameter should be provided for
meaningful results.
Args:
user_external_key: Optional user ID to filter by (e.g., 'john_doe').
role_external_key: Optional role name to filter by (e.g., 'library_admin').
scope_external_key: Optional scope to filter by (e.g., 'lib:DemoX:CSPROB').
Returns:
list[RoleAssignmentData]: Filtered role assignments.
"""
return get_role_assignments(
subject=UserData(external_key=user_external_key) if user_external_key else None,
role=RoleData(external_key=role_external_key) if role_external_key else None,
scope=ScopeData(external_key=scope_external_key) if scope_external_key else None,
)
def get_all_user_role_assignments_in_scope(
scope_external_key: str,
) -> list[RoleAssignmentData]:
"""Get all user role assignments in a specific scope.
Args:
scope (str): Scope in which to retrieve the user role assignments.
Returns:
list[RoleAssignmentData]: A list of user role assignments and all their metadata in the specified scope.
"""
return get_all_subject_role_assignments_in_scope(ScopeData(external_key=scope_external_key))
def _filter_allowed_assignments(
assignments: list[RoleAssignmentData], user_external_key: str = None
) -> list[RoleAssignmentData]:
"""
Filter the given role assignments to only include those that the user has permission to view.
"""
if not user_external_key:
# If no user is specified, return all assignments
return assignments
allowed_assignments: list[RoleAssignmentData] = []
for assignment in assignments:
permission = None
# Get the permission needed to view the specific scope in the admin console
permission = assignment.scope.get_admin_view_permission().identifier
if permission and is_user_allowed(
user_external_key=user_external_key,
action_external_key=permission,
scope_external_key=assignment.scope.external_key,
):
allowed_assignments.append(assignment)
return allowed_assignments
def get_visible_role_assignments_for_user(
orgs: list[str] = None,
scopes: list[str] = None,
allowed_for_user_external_key: str = None,
) -> list[UserAssignments]:
"""
Get all user role assignments filtered by orgs and/or scopes, and only include
assignments that the specified user has permission to view.
Args:
orgs: Optional list of orgs to filter by (e.g., ['edX', 'MITx']).
scopes: Optional list of scopes to filter by (e.g., ['lib:DemoX:CSPROB']).
allowed_for_user_external_key: The username to check permissions against (e.g., 'john_doe').
Returns:
list[UserAssignments]: A list of users with their role assignments, filtered by orgs/scopes and permissions.
"""
user_role_assignments = get_user_role_assignments_filtered()
# Filter assignments based on the user's permissions
user_role_assignments = _filter_allowed_assignments(
user_external_key=allowed_for_user_external_key,
assignments=user_role_assignments,
)
# Group assignments by user
users_with_assignments = get_user_assignment_map(user_role_assignments)
users_with_assignments = filter_user_assignments(
users_with_assignments=users_with_assignments,
by=UserAssignmentsFilter.SCOPES,
values=scopes,
)
users_with_assignments = filter_user_assignments(
users_with_assignments=users_with_assignments,
by=UserAssignmentsFilter.ORGS,
values=orgs,
)
return users_with_assignments
def is_user_allowed(
user_external_key: str,
action_external_key: str,
scope_external_key: str,
) -> bool:
"""Check if a user has a specific permission in a given scope.
Args:
user_external_key (str): ID of the user (e.g., 'john_doe').
action_external_key (str): The action to check (e.g., 'view_course').
scope_external_key (str): The scope in which to check the permission (e.g., 'course-v1:edX+DemoX+2021_T1').
Returns:
bool: True if the user has the specified permission in the scope, False otherwise.
"""
return is_subject_allowed(
UserData(external_key=user_external_key),
ActionData(external_key=action_external_key),
ScopeData(external_key=scope_external_key),
)
def get_users_for_role_in_scope(role_external_key: str, scope_external_key: str) -> list[UserData]:
"""Get all the users assigned to a specific role in a specific scope.
Args:
role_external_key (str): The role to filter users (e.g., 'library_admin').
scope_external_key (str): The scope to filter users (e.g., 'lib:DemoX:CSPROB').
Returns:
list[UserData]: A list of users assigned to the specified role in the specified scope.
"""
users = get_subjects_for_role_in_scope(
RoleData(external_key=role_external_key),
ScopeData(external_key=scope_external_key),
)
return [UserData(namespaced_key=user.namespaced_key) for user in users]
def get_scopes_for_user_and_permission(
user_external_key: str,
action_external_key: str,
) -> list[ScopeData]:
"""Get all scopes where a specific user is assigned a specific permission.
Args:
user_external_key (str): ID of the user (e.g., 'john_doe').
action_external_key (str): The action to filter scopes (e.g., 'view', 'edit').
Returns:
list[ScopeData]: A list of scopes where the user is assigned the specified permission.
"""
return get_scopes_for_subject_and_permission(
UserData(external_key=user_external_key),
PermissionData(action=ActionData(external_key=action_external_key)),
)
def unassign_all_roles_from_user(user_external_key: str) -> bool:
"""Unassign all roles from a user across all scopes.
Args:
user_external_key (str): ID of the user (e.g., 'john_doe').
Returns:
bool: True if any roles were removed, False otherwise.
"""
return unassign_subject_from_all_roles(UserData(external_key=user_external_key))
def validate_users(user_identifiers: list[str]) -> tuple[list[str], list[str]]:
"""Validate a list of user identifiers.
Args:
user_identifiers (list[str]): List of usernames or emails to validate
Returns:
tuple: (valid_users, invalid_users) lists
"""
valid_users = []
invalid_users = []
for user_identifier in user_identifiers:
try:
user = get_user_by_username_or_email(user_identifier)
if user.is_active:
valid_users.append(user_identifier)
else:
invalid_users.append(user_identifier)
except User.DoesNotExist:
invalid_users.append(user_identifier)
return valid_users, invalid_users
def get_superadmin_assignments(user_external_keys: list[str] | None = None) -> list[SuperAdminAssignmentData]:
"""Returns all superadmins as SuperAdminAssignmentData.
A superadmin is a User with a Django staff or superuser role.
Superadmins automatically are allowed to do any action.
Args:
user_external_keys (list[str] or None): To filter by usernames
Returns:
list[SuperAdminAssignmentData]: The superadmin data
"""
superadmin_filter = Q(is_active=True) & (Q(is_staff=True) | Q(is_superuser=True))
if user_external_keys is not None:
superadmin_filter &= Q(username__in=user_external_keys)
requested_users = User.objects.filter(superadmin_filter)
superadmin_assignments: list[SuperAdminAssignmentData] = []
for requested_user in requested_users:
superadmin_assignments.append(
SuperAdminAssignmentData(
user=requested_user,
is_staff=requested_user.is_staff,
is_superuser=requested_user.is_superuser,
)
)
return superadmin_assignments