-
-
Notifications
You must be signed in to change notification settings - Fork 7.5k
/
backends.py
3175 lines (2687 loc) · 130 KB
/
backends.py
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
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# Documentation for Zulip's authentication backends is split across a few places:
#
# * https://zulip.readthedocs.io/en/latest/production/authentication-methods.html and
# zproject/prod_settings_template.py have user-level configuration documentation.
# * https://zulip.readthedocs.io/en/latest/development/authentication.html
# has developer-level documentation, especially on testing authentication backends
# in the Zulip development environment.
#
# Django upstream's documentation for authentication backends is also
# helpful background. The most important detail to understand for
# reading this file is that the Django authenticate() function will
# call the authenticate methods of all backends registered in
# settings.AUTHENTICATION_BACKENDS that have a function signature
# matching the args/kwargs passed in the authenticate() call.
import binascii
import json
import logging
from abc import ABC, abstractmethod
from email.headerregistry import Address
from typing import (
Any,
Callable,
Dict,
List,
Optional,
Set,
Tuple,
Type,
TypedDict,
TypeVar,
Union,
cast,
)
from urllib.parse import urlencode
import magic
import orjson
from decorator import decorator
from django.conf import settings
from django.contrib.auth import authenticate, get_backends
from django.contrib.auth.backends import RemoteUserBackend
from django.contrib.staticfiles.storage import staticfiles_storage
from django.core.exceptions import ValidationError
from django.core.validators import validate_email
from django.dispatch import Signal, receiver
from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
from django.shortcuts import render
from django.urls import reverse
from django.utils.translation import gettext as _
from django_auth_ldap.backend import LDAPBackend, _LDAPUser, ldap_error
from lxml.etree import XMLSyntaxError
from onelogin.saml2 import compat as onelogin_saml2_compat
from onelogin.saml2.auth import OneLogin_Saml2_Auth
from onelogin.saml2.errors import OneLogin_Saml2_Error, OneLogin_Saml2_ValidationError
from onelogin.saml2.logout_request import OneLogin_Saml2_Logout_Request
from onelogin.saml2.logout_response import OneLogin_Saml2_Logout_Response
from onelogin.saml2.response import OneLogin_Saml2_Response
from onelogin.saml2.settings import OneLogin_Saml2_Settings
from onelogin.saml2.utils import OneLogin_Saml2_Utils
from onelogin.saml2.xml_utils import OneLogin_Saml2_XML
from requests import HTTPError
from social_core.backends.apple import AppleIdAuth
from social_core.backends.azuread import AzureADOAuth2
from social_core.backends.base import BaseAuth
from social_core.backends.github import GithubOAuth2, GithubOrganizationOAuth2, GithubTeamOAuth2
from social_core.backends.gitlab import GitLabOAuth2
from social_core.backends.google import GoogleOAuth2
from social_core.backends.open_id_connect import OpenIdConnectAuth
from social_core.backends.saml import SAMLAuth, SAMLIdentityProvider
from social_core.exceptions import (
AuthCanceled,
AuthFailed,
AuthMissingParameter,
AuthStateForbidden,
SocialAuthBaseException,
)
from social_core.pipeline.partial import partial
from social_django.utils import load_backend, load_strategy
from typing_extensions import override
from zxcvbn import zxcvbn
from zerver.actions.create_user import do_create_user, do_reactivate_user
from zerver.actions.custom_profile_fields import do_update_user_custom_profile_data_if_changed
from zerver.actions.user_groups import (
bulk_add_members_to_user_groups,
bulk_remove_members_from_user_groups,
)
from zerver.actions.user_settings import do_regenerate_api_key
from zerver.actions.users import do_deactivate_user
from zerver.lib.avatar import avatar_url, is_avatar_new
from zerver.lib.avatar_hash import user_avatar_content_hash
from zerver.lib.dev_ldap_directory import init_fakeldap
from zerver.lib.email_validation import email_allowed_for_realm, validate_email_not_already_in_realm
from zerver.lib.exceptions import JsonableError
from zerver.lib.mobile_auth_otp import is_valid_otp
from zerver.lib.rate_limiter import RateLimitedObject, client_is_exempt_from_rate_limiting
from zerver.lib.redis_utils import get_dict_from_redis, get_redis_client, put_dict_in_redis
from zerver.lib.request import RequestNotes
from zerver.lib.sessions import delete_user_sessions
from zerver.lib.subdomains import get_subdomain
from zerver.lib.types import ProfileDataElementUpdateDict
from zerver.lib.url_encoding import append_url_query_string
from zerver.lib.users import check_full_name, validate_user_custom_profile_field
from zerver.models import (
CustomProfileField,
PreregistrationRealm,
PreregistrationUser,
Realm,
UserGroup,
UserGroupMembership,
UserProfile,
)
from zerver.models.custom_profile_fields import custom_profile_fields_for_realm
from zerver.models.realms import (
DisposableEmailError,
DomainNotAllowedForRealmError,
EmailContainsPlusError,
get_realm,
supported_auth_backends,
)
from zerver.models.users import (
PasswordTooWeakError,
get_user_by_delivery_email,
get_user_profile_by_id,
remote_user_to_email,
)
from zproject.settings_types import OIDCIdPConfigDict
redis_client = get_redis_client()
def all_implemented_backend_names() -> List[str]:
return list(AUTH_BACKEND_NAME_MAP.keys())
# This first batch of methods is used by other code in Zulip to check
# whether a given authentication backend is enabled for a given realm.
# In each case, we both needs to check at the server level (via
# `settings.AUTHENTICATION_BACKENDS`, queried via
# `django.contrib.auth.get_backends`) and at the realm level (via the
# `RealmAuthenticationMethod` table).
def pad_method_dict(method_dict: Dict[str, bool]) -> Dict[str, bool]:
"""Pads an authentication methods dict to contain all auth backends
supported by the software, regardless of whether they are
configured on this server"""
for key in AUTH_BACKEND_NAME_MAP:
if key not in method_dict:
method_dict[key] = False
return method_dict
def auth_enabled_helper(
backends_to_check: List[str],
realm: Optional[Realm],
realm_authentication_methods: Optional[Dict[str, bool]] = None,
) -> bool:
"""
realm_authentication_methods can be passed if already fetched to avoid
a database query.
"""
if realm is not None:
if realm_authentication_methods is not None:
# Copy the dict to avoid mutating the original if it was passed in as argument.
enabled_method_dict = realm_authentication_methods.copy()
else:
enabled_method_dict = realm.authentication_methods_dict()
else:
enabled_method_dict = {method: True for method in AUTH_BACKEND_NAME_MAP}
pad_method_dict(enabled_method_dict)
for supported_backend in supported_auth_backends():
for backend_name in backends_to_check:
backend = AUTH_BACKEND_NAME_MAP[backend_name]
if enabled_method_dict[backend_name] and isinstance(supported_backend, backend):
return True
return False
def ldap_auth_enabled(
realm: Optional[Realm] = None, realm_authentication_methods: Optional[Dict[str, bool]] = None
) -> bool:
return auth_enabled_helper(["LDAP"], realm, realm_authentication_methods)
def email_auth_enabled(
realm: Optional[Realm] = None, realm_authentication_methods: Optional[Dict[str, bool]] = None
) -> bool:
return auth_enabled_helper(["Email"], realm, realm_authentication_methods)
def password_auth_enabled(
realm: Optional[Realm] = None, realm_authentication_methods: Optional[Dict[str, bool]] = None
) -> bool:
return ldap_auth_enabled(realm, realm_authentication_methods) or email_auth_enabled(
realm, realm_authentication_methods
)
def dev_auth_enabled(
realm: Optional[Realm] = None, realm_authentication_methods: Optional[Dict[str, bool]] = None
) -> bool:
return auth_enabled_helper(["Dev"], realm, realm_authentication_methods)
def google_auth_enabled(
realm: Optional[Realm] = None, realm_authentication_methods: Optional[Dict[str, bool]] = None
) -> bool:
return auth_enabled_helper(["Google"], realm, realm_authentication_methods)
def github_auth_enabled(
realm: Optional[Realm] = None, realm_authentication_methods: Optional[Dict[str, bool]] = None
) -> bool:
return auth_enabled_helper(["GitHub"], realm, realm_authentication_methods)
def gitlab_auth_enabled(
realm: Optional[Realm] = None, realm_authentication_methods: Optional[Dict[str, bool]] = None
) -> bool:
return auth_enabled_helper(["GitLab"], realm, realm_authentication_methods)
def apple_auth_enabled(
realm: Optional[Realm] = None, realm_authentication_methods: Optional[Dict[str, bool]] = None
) -> bool:
return auth_enabled_helper(["Apple"], realm, realm_authentication_methods)
def saml_auth_enabled(
realm: Optional[Realm] = None, realm_authentication_methods: Optional[Dict[str, bool]] = None
) -> bool:
return auth_enabled_helper(["SAML"], realm, realm_authentication_methods)
def require_email_format_usernames(realm: Optional[Realm] = None) -> bool:
if ldap_auth_enabled(realm) and (settings.LDAP_EMAIL_ATTR or settings.LDAP_APPEND_DOMAIN):
return False
return True
def is_user_active(user_profile: UserProfile, return_data: Optional[Dict[str, Any]] = None) -> bool:
if user_profile.realm.deactivated:
if return_data is not None:
return_data["inactive_realm"] = True
return False
if not user_profile.is_active:
if return_data is not None:
if user_profile.is_mirror_dummy:
# Record whether it's a mirror dummy account
return_data["is_mirror_dummy"] = True
return_data["inactive_user"] = True
return_data["inactive_user_id"] = user_profile.id
return False
return True
def common_get_active_user(
email: str, realm: Realm, return_data: Optional[Dict[str, Any]] = None
) -> Optional[UserProfile]:
"""This is the core common function used by essentially all
authentication backends to check if there's an active user account
with a given email address in the organization, handling both
user-level and realm-level deactivation correctly.
"""
try:
user_profile = get_user_by_delivery_email(email, realm)
except UserProfile.DoesNotExist:
# If the user doesn't have an account in the target realm, we
# check whether they might have an account in another realm,
# and if so, provide a helpful error message via
# `invalid_subdomain`.
if not UserProfile.objects.filter(delivery_email__iexact=email).exists():
return None
if return_data is not None:
return_data["invalid_subdomain"] = True
return_data["matching_user_ids_in_different_realms"] = list(
UserProfile.objects.filter(delivery_email__iexact=email).values("realm_id", "id")
)
return None
if not is_user_active(user_profile, return_data):
return None
return user_profile
def is_subdomain_in_allowed_subdomains_list(subdomain: str, allowed_subdomains: List[str]) -> bool:
if subdomain in allowed_subdomains:
return True
# The root subdomain is a special case, as sending an
# empty string in the list of values of the attribute may
# not be viable. So, any of the ROOT_SUBDOMAIN_ALIASES can
# be used to signify the user is authorized for the root
# subdomain.
if (
subdomain == Realm.SUBDOMAIN_FOR_ROOT_DOMAIN
and not settings.ROOT_DOMAIN_LANDING_PAGE
and any(alias in allowed_subdomains for alias in settings.ROOT_SUBDOMAIN_ALIASES)
):
return True
return False
AuthFuncT = TypeVar("AuthFuncT", bound=Callable[..., Optional[UserProfile]])
class RateLimitedAuthenticationByUsername(RateLimitedObject):
def __init__(self, username: str) -> None:
self.username = username
super().__init__()
@override
def key(self) -> str:
return f"{type(self).__name__}:{self.username}"
@override
def rules(self) -> List[Tuple[int, int]]:
return settings.RATE_LIMITING_RULES["authenticate_by_username"]
def rate_limit_authentication_by_username(request: HttpRequest, username: str) -> None:
RateLimitedAuthenticationByUsername(username).rate_limit_request(request)
def auth_rate_limiting_already_applied(request: HttpRequest) -> bool:
request_notes = RequestNotes.get_notes(request)
return any(
isinstance(r.entity, RateLimitedAuthenticationByUsername)
for r in request_notes.ratelimits_applied
)
# Django's authentication mechanism uses introspection on the various authenticate() functions
# defined by backends, so we need a decorator that doesn't break function signatures.
# @decorator does this for us.
# The usual @wraps from functools breaks signatures, so it can't be used here.
@decorator
def custom_auth_decorator(auth_func: AuthFuncT, *args: Any, **kwargs: Any) -> Optional[UserProfile]:
custom_auth_wrapper_func = settings.CUSTOM_AUTHENTICATION_WRAPPER_FUNCTION
if custom_auth_wrapper_func is None:
return auth_func(*args, **kwargs)
else:
return custom_auth_wrapper_func(auth_func, *args, **kwargs)
@decorator
def rate_limit_auth(auth_func: AuthFuncT, *args: Any, **kwargs: Any) -> Optional[UserProfile]:
if not settings.RATE_LIMITING_AUTHENTICATE:
return auth_func(*args, **kwargs)
request = args[1]
username = kwargs["username"]
if RequestNotes.get_notes(request).client is None or not client_is_exempt_from_rate_limiting(
request
):
# Django cycles through enabled authentication backends until one succeeds,
# or all of them fail. If multiple backends are tried like this, we only want
# to execute rate_limit_authentication_* once, on the first attempt:
if auth_rate_limiting_already_applied(request):
pass
else:
# Apply rate limiting. If this request is above the limit,
# RateLimitedError will be raised, interrupting the authentication process.
# From there, the code calling authenticate() can either catch the exception
# and handle it on its own, or it will be processed by RateLimitMiddleware.
rate_limit_authentication_by_username(request, username)
result = auth_func(*args, **kwargs)
if result is not None:
# Authentication succeeded, clear the rate-limiting record.
RateLimitedAuthenticationByUsername(username).clear_history()
return result
@decorator
def log_auth_attempts(auth_func: AuthFuncT, *args: Any, **kwargs: Any) -> Optional[UserProfile]:
result = auth_func(*args, **kwargs)
backend_instance = args[0]
request = args[1]
username = kwargs["username"]
realm = kwargs["realm"]
return_data = kwargs["return_data"]
log_auth_attempt(
backend_instance.logger,
request,
realm,
username,
succeeded=result is not None,
return_data=return_data,
)
return result
def log_auth_attempt(
logger: logging.Logger,
request: HttpRequest,
realm: Realm,
username: str,
succeeded: bool,
return_data: Dict[str, Any],
) -> None:
ip_addr = request.META.get("REMOTE_ADDR")
outcome = "success" if succeeded else "failed"
logger.info(
"Authentication attempt from %s: subdomain=%s;username=%s;outcome=%s;return_data=%s",
ip_addr,
realm.subdomain,
username,
outcome,
return_data,
)
class ZulipAuthMixin:
"""This common mixin is used to override Django's default behavior for
looking up a logged-in user by ID to use a version that fetches
from memcached before checking the database (avoiding a database
query in most cases).
"""
name = "undefined"
_logger: Optional[logging.Logger] = None
@property
def logger(self) -> logging.Logger:
if self._logger is None:
self._logger = logging.getLogger(f"zulip.auth.{self.name}")
return self._logger
def get_user(self, user_profile_id: int) -> Optional[UserProfile]:
"""Override the Django method for getting a UserProfile object from
the user_profile_id,."""
try:
return get_user_profile_by_id(user_profile_id)
except UserProfile.DoesNotExist:
return None
class ZulipDummyBackend(ZulipAuthMixin):
"""Used when we want to log you in without checking any
authentication (i.e. new user registration or when otherwise
authentication has already been checked earlier in the process).
We ensure that this backend only ever successfully authenticates
when explicitly requested by including the use_dummy_backend kwarg.
"""
name = "dummy"
@custom_auth_decorator
def authenticate(
self,
request: Optional[HttpRequest] = None,
*,
username: str,
realm: Realm,
use_dummy_backend: bool = False,
return_data: Optional[Dict[str, Any]] = None,
) -> Optional[UserProfile]:
if use_dummy_backend:
return common_get_active_user(username, realm, return_data)
return None
def check_password_strength(password: str) -> bool:
"""
Returns True if the password is strong enough,
False otherwise.
"""
if len(password) < settings.PASSWORD_MIN_LENGTH:
return False
if password == "":
# zxcvbn throws an exception when passed the empty string, so
# we need a special case for the empty string password here.
return False
if int(zxcvbn(password)["guesses"]) < settings.PASSWORD_MIN_GUESSES:
return False
return True
class EmailAuthBackend(ZulipAuthMixin):
"""
Email+Password authentication backend (the default).
Allows a user to sign in using an email/password pair.
"""
name = "email"
@rate_limit_auth
@log_auth_attempts
@custom_auth_decorator
def authenticate(
self,
request: HttpRequest,
*,
username: str,
password: str,
realm: Realm,
return_data: Optional[Dict[str, Any]] = None,
) -> Optional[UserProfile]:
"""Authenticate a user based on email address as the user name."""
if not password_auth_enabled(realm):
if return_data is not None:
return_data["password_auth_disabled"] = True
return None
if not email_auth_enabled(realm):
if return_data is not None:
return_data["email_auth_disabled"] = True
return None
if password == "":
# Never allow an empty password. This is defensive code;
# a user having password "" should only be possible
# through a bug somewhere else.
return None
user_profile = common_get_active_user(username, realm, return_data=return_data)
if user_profile is None:
return None
try:
is_password_correct = user_profile.check_password(password)
except PasswordTooWeakError:
# In some rare cases when password hasher is changed and the user has
# a weak password, PasswordTooWeakError will be raised.
self.logger.info(
"User %s password can't be rehashed due to being too weak.", user_profile.id
)
if return_data is not None:
return_data["password_reset_needed"] = True
return None
else:
# Since we can't communicate the situation via return_data,
# we have to raise an error - a silent failure would not be right
# because the password actually is correct, just can't be re-hashed.
raise JsonableError(_("You need to reset your password."))
if is_password_correct:
return user_profile
return None
def is_valid_email(email: str) -> bool:
try:
validate_email(email)
except ValidationError:
return False
return True
def check_ldap_config() -> None:
if not settings.LDAP_APPEND_DOMAIN:
# Email search needs to be configured in this case.
assert settings.AUTH_LDAP_USERNAME_ATTR and settings.AUTH_LDAP_REVERSE_EMAIL_SEARCH
# These two are alternatives approaches to deactivating users based on an ldap attribute
# and thus don't make sense to have enabled together.
assert not (
settings.AUTH_LDAP_USER_ATTR_MAP.get("userAccountControl")
and settings.AUTH_LDAP_USER_ATTR_MAP.get("deactivated")
)
def ldap_should_sync_active_status() -> bool:
if "userAccountControl" in settings.AUTH_LDAP_USER_ATTR_MAP:
return True
if "deactivated" in settings.AUTH_LDAP_USER_ATTR_MAP:
return True
return False
def find_ldap_users_by_email(email: str) -> List[_LDAPUser]:
"""
Returns list of _LDAPUsers matching the email search
"""
return LDAPReverseEmailSearch().search_for_users(email)
def email_belongs_to_ldap(realm: Realm, email: str) -> bool:
"""Used to make determinations on whether a user's email address is
managed by LDAP. For environments using both LDAP and
Email+Password authentication, we do not allow EmailAuthBackend
authentication for email addresses managed by LDAP (to avoid a
security issue where one create separate credentials for an LDAP
user), and this function is used to enforce that rule.
"""
if not ldap_auth_enabled(realm):
return False
check_ldap_config()
if settings.LDAP_APPEND_DOMAIN:
# Check if the email ends with LDAP_APPEND_DOMAIN
return Address(addr_spec=email).domain.lower() == settings.LDAP_APPEND_DOMAIN
# If we don't have an LDAP domain, we have to do a lookup for the email.
if find_ldap_users_by_email(email):
return True
else:
return False
ldap_logger = logging.getLogger("zulip.ldap")
class LDAPReverseEmailSearch(_LDAPUser):
"""
This class is a workaround - we want to use
django-auth-ldap to query the ldap directory for
users with the specified email address, but it doesn't
provide an API for that or an isolated class for handling
the connection. Because connection-handling is tightly integrated
into the _LDAPUser class, we have to make this strange inheritance here,
in order to be able to comfortably have an ldap connection and make search
queries.
We may be able to get rid of this in the future if we can get
https://github.com/django-auth-ldap/django-auth-ldap/pull/150 merged upstream.
"""
def __init__(self) -> None:
# Superclass __init__ requires a username argument - it doesn't actually
# impact anything for us in this class, given its very limited use
# for only making a search query, so we pass an empty string.
super().__init__(LDAPBackend(), username="")
def search_for_users(self, email: str) -> List[_LDAPUser]:
search = settings.AUTH_LDAP_REVERSE_EMAIL_SEARCH
USERNAME_ATTR = settings.AUTH_LDAP_USERNAME_ATTR
assert search is not None
results = search.execute(self.connection, {"email": email})
ldap_users = []
for result in results:
user_dn, user_attrs = result
username = user_attrs[USERNAME_ATTR][0]
ldap_user = _LDAPUser(self.backend, username=username)
ldap_user._user_dn = user_dn
ldap_user._user_attrs = user_attrs
ldap_users.append(ldap_user)
return ldap_users
class ZulipLDAPError(_LDAPUser.AuthenticationFailed):
"""Since this inherits from _LDAPUser.AuthenticationFailed, these will
be caught and logged at debug level inside django-auth-ldap's authenticate()"""
class NoMatchingLDAPUserError(ZulipLDAPError):
pass
class OutsideLDAPDomainError(NoMatchingLDAPUserError):
pass
class ZulipLDAPConfigurationError(Exception):
pass
LDAP_USER_ACCOUNT_CONTROL_DISABLED_MASK = 2
class ZulipLDAPAuthBackendBase(ZulipAuthMixin, LDAPBackend):
"""Common code between LDAP authentication (ZulipLDAPAuthBackend) and
using LDAP just to sync user data (ZulipLDAPUserPopulator).
To fully understand our LDAP backend, you may want to skim
django_auth_ldap/backend.py from the upstream django-auth-ldap
library. It's not a lot of code, and searching around in that
file makes the flow for LDAP authentication clear.
"""
name = "ldap"
def __init__(self) -> None:
# Used to initialize a fake LDAP directly for both manual
# and automated testing in a development environment where
# there is no actual LDAP server.
if settings.DEVELOPMENT and settings.FAKE_LDAP_MODE: # nocoverage
init_fakeldap()
check_ldap_config()
# Disable django-auth-ldap's permissions functions -- we don't use
# the standard Django user/group permissions system because they
# are prone to performance issues.
def has_perm(self, user: Optional[UserProfile], perm: Any, obj: Any = None) -> bool:
return False
def has_module_perms(self, user: Optional[UserProfile], app_label: Optional[str]) -> bool:
return False
def get_all_permissions(self, user: Optional[UserProfile], obj: Any = None) -> Set[Any]:
return set()
def get_group_permissions(self, user: Optional[UserProfile], obj: Any = None) -> Set[Any]:
return set()
def django_to_ldap_username(self, username: str) -> str:
"""
Translates django username (user_profile.delivery_email or whatever the user typed in the login
field when authenticating via the LDAP backend) into LDAP username.
Guarantees that the username it returns actually has an entry in the LDAP directory.
Raises NoMatchingLDAPUserError if that's not possible.
"""
result = username
if settings.LDAP_APPEND_DOMAIN:
if is_valid_email(username):
address = Address(addr_spec=username)
if address.domain != settings.LDAP_APPEND_DOMAIN:
raise OutsideLDAPDomainError(
f"Email {username} does not match LDAP domain {settings.LDAP_APPEND_DOMAIN}."
)
result = address.username
else:
# We can use find_ldap_users_by_email
if is_valid_email(username):
email_search_result = find_ldap_users_by_email(username)
if not email_search_result:
result = username
elif len(email_search_result) == 1:
return email_search_result[0]._username
elif len(email_search_result) > 1:
# This is possible, but strange, so worth logging a warning about.
# We can't translate the email to a unique username,
# so we don't do anything else here.
logging.warning("Multiple users with email %s found in LDAP.", username)
result = username
if _LDAPUser(self, result).attrs is None:
# Check that there actually is an LDAP entry matching the result username
# we want to return. Otherwise, raise an exception.
error_message = (
"No LDAP user matching django_to_ldap_username result: {}. Input username: {}"
)
raise NoMatchingLDAPUserError(
error_message.format(result, username),
)
return result
def user_email_from_ldapuser(self, username: str, ldap_user: _LDAPUser) -> str:
if hasattr(ldap_user, "_username"):
# In tests, we sometimes pass a simplified _LDAPUser without _username attr,
# and with the intended username in the username argument.
username = ldap_user._username
if settings.LDAP_APPEND_DOMAIN:
return Address(username=username, domain=settings.LDAP_APPEND_DOMAIN).addr_spec
if settings.LDAP_EMAIL_ATTR is not None:
# Get email from LDAP attributes.
if settings.LDAP_EMAIL_ATTR not in ldap_user.attrs:
raise ZulipLDAPError(
f"LDAP user doesn't have the needed {settings.LDAP_EMAIL_ATTR} attribute"
)
else:
return ldap_user.attrs[settings.LDAP_EMAIL_ATTR][0]
return username
def ldap_to_django_username(self, username: str) -> str:
"""
This is called inside django_auth_ldap with only one role:
to convert _LDAPUser._username to django username (so in Zulip, the email)
and pass that as "username" argument to get_or_build_user(username, ldapuser).
In many cases, the email is stored in the _LDAPUser's attributes, so it can't be
constructed just from the username. We choose to do nothing in this function,
and our overrides of get_or_build_user() obtain that username from the _LDAPUser
object on their own, through our user_email_from_ldapuser function.
"""
return username
def sync_avatar_from_ldap(self, user: UserProfile, ldap_user: _LDAPUser) -> None:
if "avatar" in settings.AUTH_LDAP_USER_ATTR_MAP:
# We do local imports here to avoid import loops
from io import BytesIO
from zerver.actions.user_settings import do_change_avatar_fields
from zerver.lib.upload import upload_avatar_image
avatar_attr_name = settings.AUTH_LDAP_USER_ATTR_MAP["avatar"]
if avatar_attr_name not in ldap_user.attrs: # nocoverage
# If this specific user doesn't have e.g. a
# thumbnailPhoto set in LDAP, just skip that user.
return
ldap_avatar = ldap_user.attrs[avatar_attr_name][0]
avatar_changed = is_avatar_new(ldap_avatar, user)
if not avatar_changed:
# Don't do work to replace the avatar with itself.
return
# Structurally, to make the S3 backend happy, we need to
# provide a Content-Type; since that isn't specified in
# any metadata, we auto-detect it.
content_type = magic.from_buffer(ldap_avatar[:1024], mime=True)
if content_type.startswith("image/"):
upload_avatar_image(BytesIO(ldap_avatar), user, user, content_type=content_type)
do_change_avatar_fields(user, UserProfile.AVATAR_FROM_USER, acting_user=None)
# Update avatar hash.
user.avatar_hash = user_avatar_content_hash(ldap_avatar)
user.save(update_fields=["avatar_hash"])
else:
logging.warning("Could not parse %s field for user %s", avatar_attr_name, user.id)
def is_user_disabled_in_ldap(self, ldap_user: _LDAPUser) -> bool:
"""Implements checks for whether a user has been
disabled in the LDAP server being integrated with
Zulip."""
if "userAccountControl" in settings.AUTH_LDAP_USER_ATTR_MAP:
account_control_value = ldap_user.attrs[
settings.AUTH_LDAP_USER_ATTR_MAP["userAccountControl"]
][0]
return bool(int(account_control_value) & LDAP_USER_ACCOUNT_CONTROL_DISABLED_MASK)
assert "deactivated" in settings.AUTH_LDAP_USER_ATTR_MAP
attr_value = ldap_user.attrs[settings.AUTH_LDAP_USER_ATTR_MAP["deactivated"]][0]
# In the LDAP specification, a Boolean attribute should be
# *exactly* either "TRUE" or "FALSE". However,
# https://www.freeipa.org/page/V4/User_Life-Cycle_Management suggests
# that FreeIPA at least documents using Yes/No for booleans.
true_values = ["TRUE", "YES"]
false_values = ["FALSE", "NO"]
attr_value_upper = attr_value.upper()
assert (
attr_value_upper in true_values or attr_value_upper in false_values
), f"Invalid value '{attr_value}' in the LDAP attribute mapped to deactivated"
return attr_value_upper in true_values
def is_account_realm_access_forbidden(self, ldap_user: _LDAPUser, realm: Realm) -> bool:
realm_access_control = settings.AUTH_LDAP_ADVANCED_REALM_ACCESS_CONTROL
# org_membership takes priority over AUTH_LDAP_ADVANCED_REALM_ACCESS_CONTROL.
if "org_membership" in settings.AUTH_LDAP_USER_ATTR_MAP:
org_membership_attr = settings.AUTH_LDAP_USER_ATTR_MAP["org_membership"]
allowed_orgs: List[str] = ldap_user.attrs.get(org_membership_attr, [])
if is_subdomain_in_allowed_subdomains_list(realm.subdomain, allowed_orgs):
return False
# If Advanced is not configured, forbid access
if realm_access_control is None:
return True
# If neither setting is configured, allow access.
if realm_access_control is None:
return False
if realm.subdomain not in realm_access_control:
# If a realm is not configured in this setting, it shouldn't
# be affected by it - therefore, allow access.
return False
# With settings.AUTH_LDAP_ADVANCED_REALM_ACCESS_CONTROL, we
# allow access if and only if one of the entries for the
# target subdomain matches the user's LDAP attributes.
# Make sure the format of the setting makes sense.
assert isinstance(realm_access_control[realm.subdomain], list)
assert len(realm_access_control[realm.subdomain]) > 0
# Go through every "or" check
for attribute_group in realm_access_control[realm.subdomain]:
access = True
for attribute in attribute_group:
if not (
attribute in ldap_user.attrs
and attribute_group[attribute] in ldap_user.attrs[attribute]
):
access = False
if access:
return False
return True
@classmethod
def get_mapped_name(cls, ldap_user: _LDAPUser) -> str:
"""Constructs the user's Zulip full_name from the LDAP data"""
if "full_name" in settings.AUTH_LDAP_USER_ATTR_MAP:
full_name_attr = settings.AUTH_LDAP_USER_ATTR_MAP["full_name"]
full_name = ldap_user.attrs[full_name_attr][0]
elif all(key in settings.AUTH_LDAP_USER_ATTR_MAP for key in ["first_name", "last_name"]):
first_name_attr = settings.AUTH_LDAP_USER_ATTR_MAP["first_name"]
last_name_attr = settings.AUTH_LDAP_USER_ATTR_MAP["last_name"]
first_name = ldap_user.attrs[first_name_attr][0]
last_name = ldap_user.attrs[last_name_attr][0]
full_name = f"{first_name} {last_name}"
else:
raise ZulipLDAPError("Missing required mapping for user's full name")
return full_name
def sync_full_name_from_ldap(self, user_profile: UserProfile, ldap_user: _LDAPUser) -> None:
from zerver.actions.user_settings import do_change_full_name
full_name = self.get_mapped_name(ldap_user)
if full_name != user_profile.full_name:
try:
full_name = check_full_name(full_name)
except JsonableError as e:
raise ZulipLDAPError(e.msg)
do_change_full_name(user_profile, full_name, None)
def sync_custom_profile_fields_from_ldap(
self, user_profile: UserProfile, ldap_user: _LDAPUser
) -> None:
values_by_var_name: Dict[str, Union[int, str, List[int]]] = {}
for attr, ldap_attr in settings.AUTH_LDAP_USER_ATTR_MAP.items():
if not attr.startswith("custom_profile_field__"):
continue
var_name = attr.split("custom_profile_field__")[1]
try:
value = ldap_user.attrs[ldap_attr][0]
except KeyError:
# If this user doesn't have this field set then ignore this
# field and continue syncing other fields. `django-auth-ldap`
# automatically logs error about missing field.
continue
values_by_var_name[var_name] = value
try:
sync_user_profile_custom_fields(user_profile, values_by_var_name)
except SyncUserError as e:
raise ZulipLDAPError(str(e)) from e
def sync_groups_from_ldap(self, user_profile: UserProfile, ldap_user: _LDAPUser) -> None:
"""
For the groups set up for syncing for the realm in LDAP_SYNCHRONIZED_GROUPS_BY_REALM:
(1) Makes sure the user has membership in the Zulip UserGroups corresponding
to the LDAP groups ldap_user belongs to.
(2) Makes sure the user doesn't have membership in the Zulip UserGroups corresponding
to the LDAP groups ldap_user doesn't belong to.
"""
if user_profile.realm.string_id not in settings.LDAP_SYNCHRONIZED_GROUPS_BY_REALM:
# no groups to sync for this realm
return
configured_ldap_group_names_for_sync = set(
settings.LDAP_SYNCHRONIZED_GROUPS_BY_REALM[user_profile.realm.string_id]
)
try:
ldap_logger.debug("Syncing groups for user: %s", user_profile.id)
intended_group_name_set_for_user = set(ldap_user.group_names).intersection(
configured_ldap_group_names_for_sync
)
existing_group_name_set_for_user = set(
UserGroupMembership.objects.filter(
user_group__realm=user_profile.realm,
user_group__name__in=set(
settings.LDAP_SYNCHRONIZED_GROUPS_BY_REALM[user_profile.realm.string_id]
),
user_profile=user_profile,
).values_list("user_group__name", flat=True)
)
ldap_logger.debug(
"intended groups: %s; zulip groups: %s",
repr(intended_group_name_set_for_user),
repr(existing_group_name_set_for_user),
)
new_groups = UserGroup.objects.filter(
name__in=intended_group_name_set_for_user.difference(
existing_group_name_set_for_user
),
realm=user_profile.realm,
)
if new_groups:
ldap_logger.debug(
"add %s to %s", user_profile.id, [group.name for group in new_groups]
)
bulk_add_members_to_user_groups(new_groups, [user_profile.id], acting_user=None)
group_names_for_membership_deletion = existing_group_name_set_for_user.difference(
intended_group_name_set_for_user
)
groups_for_membership_deletion = UserGroup.objects.filter(
name__in=group_names_for_membership_deletion, realm=user_profile.realm
)
if group_names_for_membership_deletion:
ldap_logger.debug(