티스토리 수익 글 보기
{{ message }}
-
-
Notifications
You must be signed in to change notification settings - Fork 34.2k
Expand file tree
/
Copy pathbackends.py
More file actions
347 lines (285 loc) · 12.6 KB
/
Copy pathbackends.py
File metadata and controls
347 lines (285 loc) · 12.6 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
from asgiref.sync import sync_to_async
from django.contrib.auth import (
acheck_password_with_timing_attack_mitigation,
check_password_with_timing_attack_mitigation,
get_user_model,
)
from django.contrib.auth.models import Permission
from django.db.models import Exists, OuterRef, Q
from django.views.decorators.debug import sensitive_variables
UserModel = get_user_model()
class BaseBackend:
def authenticate(self, request, **kwargs):
return None
async def aauthenticate(self, request, **kwargs):
return await sync_to_async(self.authenticate)(request, **kwargs)
def get_user(self, user_id):
return None
async def aget_user(self, user_id):
return await sync_to_async(self.get_user)(user_id)
def get_user_permissions(self, user_obj, obj=None):
return set()
async def aget_user_permissions(self, user_obj, obj=None):
return await sync_to_async(self.get_user_permissions)(user_obj, obj)
def get_group_permissions(self, user_obj, obj=None):
return set()
async def aget_group_permissions(self, user_obj, obj=None):
return await sync_to_async(self.get_group_permissions)(user_obj, obj)
def get_all_permissions(self, user_obj, obj=None):
return {
*self.get_user_permissions(user_obj, obj=obj),
*self.get_group_permissions(user_obj, obj=obj),
}
async def aget_all_permissions(self, user_obj, obj=None):
return {
*await self.aget_user_permissions(user_obj, obj=obj),
*await self.aget_group_permissions(user_obj, obj=obj),
}
def has_perm(self, user_obj, perm, obj=None):
return perm in self.get_all_permissions(user_obj, obj=obj)
async def ahas_perm(self, user_obj, perm, obj=None):
return perm in await self.aget_all_permissions(user_obj, obj)
class ModelBackend(BaseBackend):
"""
Authenticates against settings.AUTH_USER_MODEL.
"""
@sensitive_variables("password")
def authenticate(self, request, username=None, password=None, **kwargs):
if username is None:
username = kwargs.get(UserModel.USERNAME_FIELD)
if username is None or password is None:
return
try:
user = UserModel._default_manager.get_by_natural_key(username)
except UserModel.DoesNotExist:
user = None
if check_password_with_timing_attack_mitigation(
user, password
) and self.user_can_authenticate(user):
return user
@sensitive_variables("password")
async def aauthenticate(self, request, username=None, password=None, **kwargs):
if username is None:
username = kwargs.get(UserModel.USERNAME_FIELD)
if username is None or password is None:
return
try:
user = await UserModel._default_manager.aget_by_natural_key(username)
except UserModel.DoesNotExist:
user = None
if await acheck_password_with_timing_attack_mitigation(
user, password
) and self.user_can_authenticate(user):
return user
def user_can_authenticate(self, user):
"""
Reject users with is_active=False. Custom user models that don't have
that attribute are allowed.
"""
return getattr(user, "is_active", True)
def _get_user_permissions(self, user_obj):
return user_obj.user_permissions.all()
def _get_group_permissions(self, user_obj):
return Permission.objects.filter(group__in=user_obj.groups.all())
def _get_permissions(self, user_obj, obj, from_name):
"""
Return the permissions of `user_obj` from `from_name`. `from_name` can
be either "group" or "user" to return permissions from
`_get_group_permissions` or `_get_user_permissions` respectively.
"""
if not user_obj.is_active or user_obj.is_anonymous or obj is not None:
return set()
perm_cache_name = "_%s_perm_cache" % from_name
if not hasattr(user_obj, perm_cache_name):
if user_obj.is_superuser:
perms = Permission.objects.all()
else:
perms = getattr(self, "_get_%s_permissions" % from_name)(user_obj)
perms = perms.values_list("content_type__app_label", "codename").order_by()
setattr(
user_obj, perm_cache_name, {"%s.%s" % (ct, name) for ct, name in perms}
)
return getattr(user_obj, perm_cache_name)
async def _aget_permissions(self, user_obj, obj, from_name):
"""See _get_permissions()."""
if not user_obj.is_active or user_obj.is_anonymous or obj is not None:
return set()
perm_cache_name = "_%s_perm_cache" % from_name
if not hasattr(user_obj, perm_cache_name):
if user_obj.is_superuser:
perms = Permission.objects.all()
else:
perms = getattr(self, "_get_%s_permissions" % from_name)(user_obj)
perms = perms.values_list("content_type__app_label", "codename").order_by()
setattr(
user_obj,
perm_cache_name,
{"%s.%s" % (ct, name) async for ct, name in perms},
)
return getattr(user_obj, perm_cache_name)
def get_user_permissions(self, user_obj, obj=None):
"""
Return a set of permission strings the user `user_obj` has from their
`user_permissions`.
"""
return self._get_permissions(user_obj, obj, "user")
async def aget_user_permissions(self, user_obj, obj=None):
"""See get_user_permissions()."""
return await self._aget_permissions(user_obj, obj, "user")
def get_group_permissions(self, user_obj, obj=None):
"""
Return a set of permission strings the user `user_obj` has from the
groups they belong.
"""
return self._get_permissions(user_obj, obj, "group")
async def aget_group_permissions(self, user_obj, obj=None):
"""See get_group_permissions()."""
return await self._aget_permissions(user_obj, obj, "group")
def get_all_permissions(self, user_obj, obj=None):
if not user_obj.is_active or user_obj.is_anonymous or obj is not None:
return set()
if not hasattr(user_obj, "_perm_cache"):
user_obj._perm_cache = super().get_all_permissions(user_obj)
return user_obj._perm_cache
def has_perm(self, user_obj, perm, obj=None):
return user_obj.is_active and super().has_perm(user_obj, perm, obj=obj)
async def ahas_perm(self, user_obj, perm, obj=None):
return user_obj.is_active and await super().ahas_perm(user_obj, perm, obj=obj)
def has_module_perms(self, user_obj, app_label):
"""
Return True if user_obj has any permissions in the given app_label.
"""
return user_obj.is_active and any(
perm[: perm.index(".")] == app_label
for perm in self.get_all_permissions(user_obj)
)
async def ahas_module_perms(self, user_obj, app_label):
"""See has_module_perms()"""
return user_obj.is_active and any(
perm[: perm.index(".")] == app_label
for perm in await self.aget_all_permissions(user_obj)
)
def with_perm(self, perm, is_active=True, include_superusers=True, obj=None):
"""
Return users that have permission "perm". By default, filter out
inactive users and include superusers.
"""
if isinstance(perm, str):
try:
app_label, codename = perm.split(".")
except ValueError:
raise ValueError(
"Permission name should be in the form "
"app_label.permission_codename."
)
elif not isinstance(perm, Permission):
raise TypeError(
"The `perm` argument must be a string or a permission instance."
)
if obj is not None:
return UserModel._default_manager.none()
permission_q = Q(group__user=OuterRef("pk")) | Q(user=OuterRef("pk"))
if isinstance(perm, Permission):
permission_q &= Q(pk=perm.pk)
else:
permission_q &= Q(codename=codename, content_type__app_label=app_label)
user_q = Exists(Permission.objects.filter(permission_q))
if include_superusers:
user_q |= Q(is_superuser=True)
if is_active is not None:
user_q &= Q(is_active=is_active)
return UserModel._default_manager.filter(user_q)
def get_user(self, user_id):
try:
user = UserModel._default_manager.get(pk=user_id)
except UserModel.DoesNotExist:
return None
return user if self.user_can_authenticate(user) else None
async def aget_user(self, user_id):
try:
user = await UserModel._default_manager.aget(pk=user_id)
except UserModel.DoesNotExist:
return None
return user if self.user_can_authenticate(user) else None
class AllowAllUsersModelBackend(ModelBackend):
def user_can_authenticate(self, user):
return True
class RemoteUserBackend(ModelBackend):
"""
This backend is to be used in conjunction with the “RemoteUserMiddleware“
found in the middleware module of this package, and is used when the server
is handling authentication outside of Django.
By default, the “authenticate“ method creates “User“ objects for
usernames that don't already exist in the database. Subclasses can disable
this behavior by setting the “create_unknown_user“ attribute to
“False“.
"""
# Create a User object if not already in the database?
create_unknown_user = True
def authenticate(self, request, remote_user):
"""
The username passed as “remote_user“ is considered trusted. Return
the “User“ object with the given username. Create a new “User“
object if “create_unknown_user“ is “True“.
Return None if “create_unknown_user“ is “False“ and a “User“
object with the given username is not found in the database.
"""
if not remote_user:
return
created = False
user = None
username = self.clean_username(remote_user)
# Note that this could be accomplished in one try-except clause, but
# instead we use get_or_create when creating unknown users since it has
# built-in safeguards for multiple threads.
if self.create_unknown_user:
user, created = UserModel._default_manager.get_or_create(
**{UserModel.USERNAME_FIELD: username}
)
else:
try:
user = UserModel._default_manager.get_by_natural_key(username)
except UserModel.DoesNotExist:
pass
user = self.configure_user(request, user, created=created)
return user if self.user_can_authenticate(user) else None
async def aauthenticate(self, request, remote_user):
"""See authenticate()."""
if not remote_user:
return
created = False
user = None
username = self.clean_username(remote_user)
# Note that this could be accomplished in one try-except clause, but
# instead we use get_or_create when creating unknown users since it has
# built-in safeguards for multiple threads.
if self.create_unknown_user:
user, created = await UserModel._default_manager.aget_or_create(
**{UserModel.USERNAME_FIELD: username}
)
else:
try:
user = await UserModel._default_manager.aget_by_natural_key(username)
except UserModel.DoesNotExist:
pass
user = await self.aconfigure_user(request, user, created=created)
return user if self.user_can_authenticate(user) else None
def clean_username(self, username):
"""
Perform any cleaning on the "username" prior to using it to get or
create the user object. Return the cleaned username.
By default, return the username unchanged.
"""
return username
def configure_user(self, request, user, created=True):
"""
Configure a user and return the updated user.
By default, return the user unmodified.
"""
return user
async def aconfigure_user(self, request, user, created=True):
"""See configure_user()"""
return await sync_to_async(self.configure_user)(request, user, created)
class AllowAllUsersRemoteUserBackend(RemoteUserBackend):
def user_can_authenticate(self, user):
return True
You can’t perform that action at this time.