import logging from typing import Any, Dict, List, Set, Tuple, Optional, Text from django.contrib.auth.backends import RemoteUserBackend from django.conf import settings from django.http import HttpResponse import django.contrib.auth from django.core.mail.backends.base import BaseEmailBackend from django.core.mail import EmailMultiAlternatives from django.template import loader from django_auth_ldap.backend import LDAPBackend, _LDAPUser from zerver.lib.actions import do_create_user from zerver.models import UserProfile, Realm, get_user_profile_by_id, \ get_user_profile_by_email, remote_user_to_email, email_to_username, \ get_realm from apiclient.sample_tools import client as googleapiclient from oauth2client.crypt import AppIdentityError from social_core.backends.github import GithubOAuth2, GithubOrganizationOAuth2, \ GithubTeamOAuth2 from social_core.exceptions import AuthFailed, SocialAuthBaseException from django.contrib.auth import authenticate from zerver.lib.users import check_full_name from zerver.lib.request import JsonableError from zerver.lib.utils import check_subdomain, get_subdomain from social_django.models import DjangoStorage from social_django.strategy import DjangoStrategy def pad_method_dict(method_dict): # type: (Dict[Text, bool]) -> Dict[Text, 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, realm): # type: (List[Text], Optional[Realm]) -> bool if realm is not None: enabled_method_dict = realm.authentication_methods_dict() pad_method_dict(enabled_method_dict) else: enabled_method_dict = dict((method, True) for method in Realm.AUTHENTICATION_FLAGS) pad_method_dict(enabled_method_dict) for supported_backend in django.contrib.auth.get_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=None): # type: (Optional[Realm]) -> bool return auth_enabled_helper([u'LDAP'], realm) def email_auth_enabled(realm=None): # type: (Optional[Realm]) -> bool return auth_enabled_helper([u'Email'], realm) def password_auth_enabled(realm=None): # type: (Optional[Realm]) -> bool return ldap_auth_enabled(realm) or email_auth_enabled(realm) def dev_auth_enabled(realm=None): # type: (Optional[Realm]) -> bool return auth_enabled_helper([u'Dev'], realm) def google_auth_enabled(realm=None): # type: (Optional[Realm]) -> bool return auth_enabled_helper([u'Google'], realm) def github_auth_enabled(realm=None): # type: (Optional[Realm]) -> bool return auth_enabled_helper([u'GitHub'], realm) def any_oauth_backend_enabled(realm=None): # type: (Optional[Realm]) -> bool """Used by the login page process to determine whether to show the 'OR' for login with Google""" return auth_enabled_helper([u'GitHub', u'Google'], realm) def require_email_format_usernames(realm=None): # type: (Optional[Realm]) -> bool if ldap_auth_enabled(realm): if settings.LDAP_EMAIL_ATTR or settings.LDAP_APPEND_DOMAIN: return False return True def common_get_active_user_by_email(email, return_data=None): # type: (Text, Optional[Dict[str, Any]]) -> Optional[UserProfile] try: user_profile = get_user_profile_by_email(email) except UserProfile.DoesNotExist: return None if not user_profile.is_active: if return_data is not None: return_data['inactive_user'] = True return None if user_profile.realm.deactivated: if return_data is not None: return_data['inactive_realm'] = True return None return user_profile class ZulipAuthMixin(object): def get_user(self, user_profile_id): # type: (int) -> Optional[UserProfile] """ Get a UserProfile object from the user_profile_id. """ try: return get_user_profile_by_id(user_profile_id) except UserProfile.DoesNotExist: return None class SocialAuthMixin(ZulipAuthMixin): auth_backend_name = None # type: Text def get_email_address(self, *args, **kwargs): # type: (*Any, **Any) -> Text raise NotImplementedError def get_full_name(self, *args, **kwargs): # type: (*Any, **Any) -> Text raise NotImplementedError def authenticate(self, realm_subdomain='', # type: Optional[Text] storage=None, # type: Optional[DjangoStorage] strategy=None, # type: Optional[DjangoStrategy] user=None, # type: Optional[Dict[str, Any]] return_data=None, # type: Optional[Dict[str, Any]] response=None, # type: Optional[Dict[str, Any]] backend=None # type: Optional[GithubOAuth2] ): # type: (...) -> Optional[UserProfile] """ Django decides which `authenticate` to call by inspecting the arguments. So it's better to create `authenticate` function with well defined arguments. Keeping this function separate so that it can easily be overridden. """ if user is None: user = {} assert return_data is not None assert response is not None return self._common_authenticate(self, realm_subdomain=realm_subdomain, storage=storage, strategy=strategy, user=user, return_data=return_data, response=response, backend=backend) def _common_authenticate(self, *args, **kwargs): # type: (*Any, **Any) -> Optional[UserProfile] return_data = kwargs.get('return_data', {}) email_address = self.get_email_address(*args, **kwargs) if not email_address: return_data['invalid_email'] = True return None try: user_profile = get_user_profile_by_email(email_address) except UserProfile.DoesNotExist: return_data["valid_attestation"] = True return None if not user_profile.is_active: return_data["inactive_user"] = True return None if user_profile.realm.deactivated: return_data["inactive_realm"] = True return None if not check_subdomain(kwargs.get("realm_subdomain"), user_profile.realm.subdomain): return_data["invalid_subdomain"] = True return None if not auth_enabled_helper([self.auth_backend_name], user_profile.realm): return_data["auth_backend_disabled"] = True return None return user_profile def process_do_auth(self, user_profile, *args, **kwargs): # type: (UserProfile, *Any, **Any) -> Optional[HttpResponse] # These functions need to be imported here to avoid cyclic # dependency. from zerver.views.auth import (login_or_register_remote_user, redirect_to_subdomain_login_url) from zerver.views.registration import redirect_and_log_into_subdomain return_data = kwargs.get('return_data', {}) inactive_user = return_data.get('inactive_user') inactive_realm = return_data.get('inactive_realm') invalid_subdomain = return_data.get('invalid_subdomain') invalid_email = return_data.get('invalid_email') if inactive_user or inactive_realm: # Redirect to login page. We can't send to registration # workflow with these errors. We will redirect to login page. return None if invalid_email: # In case of invalid email, we will end up on registration page. # This seems better than redirecting to login page. logging.warning( "{} got invalid email argument.".format(self.auth_backend_name) ) strategy = self.strategy # type: ignore # This comes from Python Social Auth. request = strategy.request email_address = self.get_email_address(*args, **kwargs) full_name = self.get_full_name(*args, **kwargs) is_signup = strategy.session_get('is_signup') == '1' subdomain = strategy.session_get('subdomain') mobile_flow_otp = strategy.session_get('mobile_flow_otp') if not subdomain or mobile_flow_otp is not None: return login_or_register_remote_user(request, email_address, user_profile, full_name, invalid_subdomain=bool(invalid_subdomain), mobile_flow_otp=mobile_flow_otp, is_signup=is_signup) try: realm = Realm.objects.get(string_id=subdomain) except Realm.DoesNotExist: return redirect_to_subdomain_login_url() return redirect_and_log_into_subdomain(realm, full_name, email_address, is_signup=is_signup) def auth_complete(self, *args, **kwargs): # type: (*Any, **Any) -> Optional[HttpResponse] """ Returning `None` from this function will redirect the browser to the login page. """ try: # Call the auth_complete method of social_core.backends.oauth.BaseOAuth2 return super(SocialAuthMixin, self).auth_complete(*args, **kwargs) # type: ignore # monkey-patching except AuthFailed: return None except SocialAuthBaseException as e: logging.warning(str(e)) return None class ZulipDummyBackend(ZulipAuthMixin): """ Used when we want to log you in but we don't know which backend to use. """ def authenticate(self, username=None, realm_subdomain=None, use_dummy_backend=False, return_data=None): # type: (Optional[Text], Optional[Text], bool, Optional[Dict[str, Any]]) -> Optional[UserProfile] assert username is not None if use_dummy_backend: user_profile = common_get_active_user_by_email(username) if user_profile is None: return None if not check_subdomain(realm_subdomain, user_profile.realm.subdomain): if return_data is not None: return_data["invalid_subdomain"] = True return None return user_profile return None class EmailAuthBackend(ZulipAuthMixin): """ Email Authentication Backend Allows a user to sign in using an email/password pair rather than a username/password pair. """ def authenticate(self, username=None, password=None, realm_subdomain=None, return_data=None): # type: (Optional[Text], Optional[str], Optional[Text], Optional[Dict[str, Any]]) -> Optional[UserProfile] """ Authenticate a user based on email address as the user name. """ if username is None or password is None: # Return immediately. Otherwise we will look for a SQL row with # NULL username. While that's probably harmless, it's needless # exposure. return None user_profile = common_get_active_user_by_email(username, return_data=return_data) if user_profile is None: return None if not password_auth_enabled(user_profile.realm): if return_data is not None: return_data['password_auth_disabled'] = True return None if not email_auth_enabled(user_profile.realm): if return_data is not None: return_data['email_auth_disabled'] = True return None if user_profile.check_password(password): if not check_subdomain(realm_subdomain, user_profile.realm.subdomain): if return_data is not None: return_data["invalid_subdomain"] = True return None return user_profile return None class GoogleMobileOauth2Backend(ZulipAuthMixin): """ Google Apps authentication for mobile devices Allows a user to sign in using a Google-issued OAuth2 token. Ref: https://developers.google.com/+/mobile/android/sign-in#server-side_access_for_your_app https://developers.google.com/accounts/docs/CrossClientAuth#offlineAccess """ def authenticate(self, google_oauth2_token=None, realm_subdomain=None, return_data=None): # type: (Optional[str], Optional[Text], Optional[Dict[str, Any]]) -> Optional[UserProfile] if return_data is None: return_data = {} try: token_payload = googleapiclient.verify_id_token(google_oauth2_token, settings.GOOGLE_CLIENT_ID) except AppIdentityError: return None if token_payload["email_verified"] in (True, "true"): try: user_profile = get_user_profile_by_email(token_payload["email"]) except UserProfile.DoesNotExist: return_data["valid_attestation"] = True return None if not user_profile.is_active: return_data["inactive_user"] = True return None if user_profile.realm.deactivated: return_data["inactive_realm"] = True return None if not check_subdomain(realm_subdomain, user_profile.realm.subdomain): return_data["invalid_subdomain"] = True return None if not google_auth_enabled(realm=user_profile.realm): return_data["google_auth_disabled"] = True return None return user_profile else: return_data["valid_attestation"] = False return None class ZulipRemoteUserBackend(RemoteUserBackend): create_unknown_user = False def authenticate(self, remote_user, realm_subdomain=None): # type: (Optional[str], Optional[Text]) -> Optional[UserProfile] if not remote_user: return None email = remote_user_to_email(remote_user) user_profile = common_get_active_user_by_email(email) if user_profile is None: return None if not check_subdomain(realm_subdomain, user_profile.realm.subdomain): return None if not auth_enabled_helper([u"RemoteUser"], user_profile.realm): return None return user_profile class ZulipLDAPException(_LDAPUser.AuthenticationFailed): pass class ZulipLDAPConfigurationError(Exception): pass class ZulipLDAPAuthBackendBase(ZulipAuthMixin, LDAPBackend): # Don't use Django LDAP's permissions functions def has_perm(self, user, perm, obj=None): # type: (Optional[UserProfile], Any, Any) -> bool # Using Any type is safe because we are not doing anything with # the arguments. return False def has_module_perms(self, user, app_label): # type: (Optional[UserProfile], Optional[str]) -> bool return False def get_all_permissions(self, user, obj=None): # type: (Optional[UserProfile], Any) -> Set # Using Any type is safe because we are not doing anything with # the arguments. return set() def get_group_permissions(self, user, obj=None): # type: (Optional[UserProfile], Any) -> Set # Using Any type is safe because we are not doing anything with # the arguments. return set() def django_to_ldap_username(self, username): # type: (Text) -> Text if settings.LDAP_APPEND_DOMAIN: if not username.endswith("@" + settings.LDAP_APPEND_DOMAIN): raise ZulipLDAPException("Username does not match LDAP domain.") return email_to_username(username) return username def ldap_to_django_username(self, username): # type: (str) -> str if settings.LDAP_APPEND_DOMAIN: return "@".join((username, settings.LDAP_APPEND_DOMAIN)) return username class ZulipLDAPAuthBackend(ZulipLDAPAuthBackendBase): REALM_IS_NONE_ERROR = 1 def authenticate(self, username, password, realm_subdomain=None, return_data=None): # type: (Text, str, Optional[Text], Optional[Dict[str, Any]]) -> Optional[UserProfile] try: self._realm = get_realm(realm_subdomain) username = self.django_to_ldap_username(username) user_profile = ZulipLDAPAuthBackendBase.authenticate(self, username=username, password=password) if user_profile is None: return None if not check_subdomain(realm_subdomain, user_profile.realm.subdomain): return None return user_profile except Realm.DoesNotExist: return None # nocoverage # TODO: this may no longer be possible except ZulipLDAPException: return None # nocoverage # TODO: this may no longer be possible def get_or_create_user(self, username, ldap_user): # type: (str, _LDAPUser) -> Tuple[UserProfile, bool] try: if settings.LDAP_EMAIL_ATTR is not None: # Get email from ldap attributes. if settings.LDAP_EMAIL_ATTR not in ldap_user.attrs: raise ZulipLDAPException("LDAP user doesn't have the needed %s attribute" % (settings.LDAP_EMAIL_ATTR,)) username = ldap_user.attrs[settings.LDAP_EMAIL_ATTR][0] user_profile = get_user_profile_by_email(username) if not user_profile.is_active or user_profile.realm.deactivated: raise ZulipLDAPException("Realm has been deactivated") if not ldap_auth_enabled(user_profile.realm): raise ZulipLDAPException("LDAP Authentication is not enabled") return user_profile, False except UserProfile.DoesNotExist: if self._realm is None: raise ZulipLDAPConfigurationError("Realm is None", self.REALM_IS_NONE_ERROR) # No need to check for an inactive user since they don't exist yet if self._realm.deactivated: raise ZulipLDAPException("Realm has been deactivated") full_name_attr = settings.AUTH_LDAP_USER_ATTR_MAP["full_name"] short_name = full_name = ldap_user.attrs[full_name_attr][0] try: full_name = check_full_name(full_name) except JsonableError as e: raise ZulipLDAPException(e.msg) if "short_name" in settings.AUTH_LDAP_USER_ATTR_MAP: short_name_attr = settings.AUTH_LDAP_USER_ATTR_MAP["short_name"] short_name = ldap_user.attrs[short_name_attr][0] user_profile = do_create_user(username, None, self._realm, full_name, short_name) return user_profile, True # Just like ZulipLDAPAuthBackend, but doesn't let you log in. class ZulipLDAPUserPopulator(ZulipLDAPAuthBackendBase): def authenticate(self, username, password, realm_subdomain=None): # type: (Text, str, Optional[Text]) -> None return None class DevAuthBackend(ZulipAuthMixin): # Allow logging in as any user without a password. # This is used for convenience when developing Zulip. def authenticate(self, username, realm_subdomain=None, return_data=None): # type: (Text, Optional[Text], Optional[Dict[str, Any]]) -> Optional[UserProfile] user_profile = common_get_active_user_by_email(username, return_data=return_data) if user_profile is None: return None if not dev_auth_enabled(user_profile.realm): return None return user_profile class GitHubAuthBackend(SocialAuthMixin, GithubOAuth2): auth_backend_name = u"GitHub" def get_email_address(self, *args, **kwargs): # type: (*Any, **Any) -> Optional[Text] try: return kwargs['response']['email'] except KeyError: # nocoverage # TODO: investigate return None def get_full_name(self, *args, **kwargs): # type: (*Any, **Any) -> Text # In case of any error return an empty string. Name is used by # the registration page to pre-populate the name field. However, # if it is not supplied, our registration process will make sure # that the user enters a valid name. try: name = kwargs['response']['name'] except KeyError: name = '' if name is None: return '' return name def do_auth(self, *args, **kwargs): # type: (*Any, **Any) -> Optional[HttpResponse] """ This function is called once the OAuth2 workflow is complete. We override this function to: 1. Inject `return_data` and `realm_admin` kwargs. These will be used by `authenticate()` function to make the decision. 2. Call the proper `do_auth` function depending on whether we are doing individual, team or organization based GitHub authentication. The actual decision on authentication is done in SocialAuthMixin._common_authenticate(). """ kwargs['return_data'] = {} request = self.strategy.request kwargs['realm_subdomain'] = get_subdomain(request) user_profile = None team_id = settings.SOCIAL_AUTH_GITHUB_TEAM_ID org_name = settings.SOCIAL_AUTH_GITHUB_ORG_NAME if (team_id is None and org_name is None): try: user_profile = GithubOAuth2.do_auth(self, *args, **kwargs) except AuthFailed: logging.info("User authentication failed.") user_profile = None elif (team_id): backend = GithubTeamOAuth2(self.strategy, self.redirect_uri) try: user_profile = backend.do_auth(*args, **kwargs) except AuthFailed: logging.info("User is not member of GitHub team.") user_profile = None elif (org_name): backend = GithubOrganizationOAuth2(self.strategy, self.redirect_uri) try: user_profile = backend.do_auth(*args, **kwargs) except AuthFailed: logging.info("User is not member of GitHub organization.") user_profile = None return self.process_do_auth(user_profile, *args, **kwargs) AUTH_BACKEND_NAME_MAP = { u'Dev': DevAuthBackend, u'Email': EmailAuthBackend, u'GitHub': GitHubAuthBackend, u'Google': GoogleMobileOauth2Backend, u'LDAP': ZulipLDAPAuthBackend, u'RemoteUser': ZulipRemoteUserBackend, } # type: Dict[Text, Any] class EmailLogBackEnd(BaseEmailBackend): def log_email(self, email): # type: (EmailMultiAlternatives) -> None """Used in development to record sent emails in a nice HTML log""" html_message = 'Missing HTML message' if len(email.alternatives) > 0: html_message = email.alternatives[0][0] context = { 'subject': email.subject, 'from_email': email.from_email, 'recipients': email.to, 'body': email.body, 'html_message': html_message } new_email = loader.render_to_string('zerver/email.html', context) # Read in the pre-existing log, so that we can add the new entry # at the top. try: with open(settings.EMAIL_CONTENT_LOG_PATH, "r") as f: previous_emails = f.read() except FileNotFoundError: previous_emails = "" with open(settings.EMAIL_CONTENT_LOG_PATH, "w+") as f: f.write(new_email + previous_emails) def send_messages(self, email_messages): # type: (List[EmailMultiAlternatives]) -> int for email in email_messages: self.log_email(email) email_log_url = settings.ROOT_DOMAIN_URI + "/emails" print("You can access all the emails sent in development environment by visiting ", email_log_url) return len(email_messages)