mirror of https://github.com/zulip/zulip.git
668 lines
29 KiB
Python
668 lines
29 KiB
Python
import logging
|
|
from typing import Any, Dict, List, Set, Tuple, Optional, Sequence
|
|
|
|
from django_auth_ldap.backend import LDAPBackend, _LDAPUser
|
|
import django.contrib.auth
|
|
from django.contrib.auth.backends import RemoteUserBackend
|
|
from django.conf import settings
|
|
from django.core.exceptions import ValidationError
|
|
from django.core.validators import validate_email
|
|
from django.http import HttpResponse
|
|
from requests import HTTPError
|
|
from social_core.backends.github import GithubOAuth2, GithubOrganizationOAuth2, \
|
|
GithubTeamOAuth2
|
|
from social_core.backends.azuread import AzureADOAuth2
|
|
from social_core.backends.base import BaseAuth
|
|
from social_core.backends.oauth import BaseOAuth2
|
|
from social_core.utils import handle_http_errors
|
|
from social_core.exceptions import AuthFailed, SocialAuthBaseException
|
|
from social_django.models import DjangoStorage
|
|
from social_django.strategy import DjangoStrategy
|
|
|
|
from zerver.lib.actions import do_create_user, do_reactivate_user, do_deactivate_user
|
|
from zerver.lib.request import JsonableError
|
|
from zerver.lib.subdomains import user_matches_subdomain, get_subdomain
|
|
from zerver.lib.users import check_full_name
|
|
from zerver.models import UserProfile, Realm, get_user_profile_by_id, \
|
|
remote_user_to_email, email_to_username, get_realm, get_user_by_delivery_email
|
|
|
|
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]) -> 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: Optional[Realm]=None) -> bool:
|
|
return auth_enabled_helper(['LDAP'], realm)
|
|
|
|
def email_auth_enabled(realm: Optional[Realm]=None) -> bool:
|
|
return auth_enabled_helper(['Email'], realm)
|
|
|
|
def password_auth_enabled(realm: Optional[Realm]=None) -> bool:
|
|
return ldap_auth_enabled(realm) or email_auth_enabled(realm)
|
|
|
|
def dev_auth_enabled(realm: Optional[Realm]=None) -> bool:
|
|
return auth_enabled_helper(['Dev'], realm)
|
|
|
|
def google_auth_enabled(realm: Optional[Realm]=None) -> bool:
|
|
return auth_enabled_helper(['Google'], realm)
|
|
|
|
def github_auth_enabled(realm: Optional[Realm]=None) -> bool:
|
|
return auth_enabled_helper(['GitHub'], realm)
|
|
|
|
def any_oauth_backend_enabled(realm: Optional[Realm]=None) -> bool:
|
|
"""Used by the login page process to determine whether to show the
|
|
'OR' for login with Google"""
|
|
return auth_enabled_helper(OAUTH_BACKEND_NAMES, realm)
|
|
|
|
def require_email_format_usernames(realm: Optional[Realm]=None) -> 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(email: str, realm: Realm,
|
|
return_data: Optional[Dict[str, Any]]=None) -> Optional[UserProfile]:
|
|
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 None
|
|
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 None
|
|
if user_profile.realm.deactivated:
|
|
if return_data is not None:
|
|
return_data['inactive_realm'] = True
|
|
return None
|
|
return user_profile
|
|
|
|
class ZulipAuthMixin:
|
|
def get_user(self, user_profile_id: 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 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).
|
|
"""
|
|
|
|
def authenticate(self, username: Optional[str]=None, realm: Optional[Realm]=None,
|
|
use_dummy_backend: bool=False,
|
|
return_data: Optional[Dict[str, Any]]=None) -> Optional[UserProfile]:
|
|
if use_dummy_backend:
|
|
# These are kwargs only for readability; they should never be None
|
|
assert username is not None
|
|
assert realm is not None
|
|
return common_get_active_user(username, realm, return_data)
|
|
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: Optional[str]=None, password: Optional[str]=None,
|
|
realm: Optional[Realm]=None,
|
|
return_data: Optional[Dict[str, Any]]=None) -> Optional[UserProfile]:
|
|
""" Authenticate a user based on email address as the user name. """
|
|
if username is None or password is None:
|
|
# Because of how we structure our auth calls to always
|
|
# specify which backend to use when not using
|
|
# EmailAuthBackend, username and password should always be set.
|
|
raise AssertionError("Invalid call to authenticate for EmailAuthBackend")
|
|
if realm is None:
|
|
return None
|
|
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
|
|
|
|
user_profile = common_get_active_user(username, realm, return_data=return_data)
|
|
if user_profile is None:
|
|
return None
|
|
if user_profile.check_password(password):
|
|
return user_profile
|
|
return None
|
|
|
|
class GoogleMobileOauth2Backend(ZulipAuthMixin):
|
|
"""
|
|
Google Apps authentication for the legacy Android app.
|
|
DummyAuthBackend is what's actually used for our modern Google auth,
|
|
both for web and mobile (the latter via the mobile_flow_otp feature).
|
|
|
|
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: Optional[str]=None, realm: Optional[Realm]=None,
|
|
return_data: Optional[Dict[str, Any]]=None) -> Optional[UserProfile]:
|
|
# We lazily import apiclient as part of optimizing the base
|
|
# import time for a Zulip management command, since it's only
|
|
# used in this one code path and takes 30-50ms to import.
|
|
from apiclient.sample_tools import client as googleapiclient
|
|
from oauth2client.crypt import AppIdentityError
|
|
if realm is None:
|
|
return None
|
|
if return_data is None:
|
|
return_data = {}
|
|
|
|
if not google_auth_enabled(realm=realm):
|
|
return_data["google_auth_disabled"] = True
|
|
return None
|
|
|
|
try:
|
|
token_payload = googleapiclient.verify_id_token(google_oauth2_token, settings.GOOGLE_CLIENT_ID)
|
|
except AppIdentityError:
|
|
return None
|
|
|
|
if token_payload["email_verified"] not in (True, "true"):
|
|
return_data["valid_attestation"] = False
|
|
return None
|
|
|
|
return_data["valid_attestation"] = True
|
|
return common_get_active_user(token_payload["email"], realm, return_data)
|
|
|
|
class ZulipRemoteUserBackend(RemoteUserBackend):
|
|
create_unknown_user = False
|
|
|
|
def authenticate(self, remote_user: Optional[str], realm: Optional[Realm]=None,
|
|
return_data: Optional[Dict[str, Any]]=None) -> Optional[UserProfile]:
|
|
assert remote_user is not None
|
|
if realm is None:
|
|
return None
|
|
if not auth_enabled_helper(["RemoteUser"], realm):
|
|
return None
|
|
|
|
email = remote_user_to_email(remote_user)
|
|
return common_get_active_user(email, realm, return_data=return_data)
|
|
|
|
def email_belongs_to_ldap(realm: Realm, email: str) -> bool:
|
|
if not ldap_auth_enabled(realm):
|
|
return False
|
|
|
|
# If we don't have an LDAP domain, it's impossible to tell which
|
|
# accounts are LDAP accounts, so treat all of them as LDAP
|
|
# accounts
|
|
if not settings.LDAP_APPEND_DOMAIN:
|
|
return True
|
|
|
|
# Otherwise, check if the email ends with LDAP_APPEND_DOMAIN
|
|
return email.strip().lower().endswith("@" + settings.LDAP_APPEND_DOMAIN)
|
|
|
|
class ZulipLDAPException(_LDAPUser.AuthenticationFailed):
|
|
"""Since this inherits from _LDAPUser.AuthenticationFailed, these will
|
|
be caught and logged at debug level inside django-auth-ldap's authenticate()"""
|
|
pass
|
|
|
|
class ZulipLDAPExceptionOutsideDomain(ZulipLDAPException):
|
|
pass
|
|
|
|
class ZulipLDAPConfigurationError(Exception):
|
|
pass
|
|
|
|
LDAP_USER_ACCOUNT_CONTROL_DISABLED_MASK = 2
|
|
|
|
class ZulipLDAPAuthBackendBase(ZulipAuthMixin, LDAPBackend):
|
|
def __init__(self) -> None:
|
|
if settings.DEVELOPMENT and settings.FAKE_LDAP_MODE: # nocoverage
|
|
# We only use this in development. Importing mock inside
|
|
# this function is an import time optimization, which
|
|
# avoids the expensive import of the mock module (slow
|
|
# because its dependency pbr uses pkgresources, which is
|
|
# really slow to import.)
|
|
import mock
|
|
from fakeldap import MockLDAP
|
|
from zerver.lib.dev_ldap_directory import generate_dev_ldap_dir
|
|
|
|
ldap_patcher = mock.patch('django_auth_ldap.config.ldap.initialize')
|
|
self.mock_initialize = ldap_patcher.start()
|
|
self.mock_ldap = MockLDAP()
|
|
self.mock_initialize.return_value = self.mock_ldap
|
|
|
|
self.mock_ldap.directory = generate_dev_ldap_dir(settings.FAKE_LDAP_MODE,
|
|
settings.FAKE_LDAP_NUM_USERS)
|
|
|
|
# Don't use Django LDAP's permissions functions
|
|
def has_perm(self, user: Optional[UserProfile], perm: Any, obj: Any=None) -> bool:
|
|
# Using Any type is safe because we are not doing anything with
|
|
# the arguments.
|
|
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]:
|
|
# Using Any type is safe because we are not doing anything with
|
|
# the arguments and always return empty set.
|
|
return set()
|
|
|
|
def get_group_permissions(self, user: Optional[UserProfile], obj: Any=None) -> Set[Any]:
|
|
# Using Any type is safe because we are not doing anything with
|
|
# the arguments and always return empty set.
|
|
return set()
|
|
|
|
def django_to_ldap_username(self, username: str) -> str:
|
|
if settings.LDAP_APPEND_DOMAIN:
|
|
if not username.endswith("@" + settings.LDAP_APPEND_DOMAIN):
|
|
raise ZulipLDAPExceptionOutsideDomain("Email %s does not match LDAP domain %s." % (
|
|
username, settings.LDAP_APPEND_DOMAIN))
|
|
return email_to_username(username)
|
|
return username
|
|
|
|
def ldap_to_django_username(self, username: str) -> str:
|
|
if settings.LDAP_APPEND_DOMAIN:
|
|
return "@".join((username, settings.LDAP_APPEND_DOMAIN))
|
|
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 zerver.lib.upload import upload_avatar_image
|
|
from zerver.lib.actions import do_change_avatar_fields
|
|
from io import BytesIO
|
|
|
|
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
|
|
upload_avatar_image(BytesIO(ldap_user.attrs[avatar_attr_name][0]), user, user)
|
|
do_change_avatar_fields(user, UserProfile.AVATAR_FROM_USER)
|
|
|
|
def is_account_control_disabled_user(self, ldap_user: _LDAPUser) -> bool: # nocoverage
|
|
account_control_value = ldap_user.attrs[settings.AUTH_LDAP_USER_ATTR_MAP['userAccountControl']][0]
|
|
ldap_disabled = bool(int(account_control_value) & LDAP_USER_ACCOUNT_CONTROL_DISABLED_MASK)
|
|
return ldap_disabled
|
|
|
|
def get_or_build_user(self, username: str,
|
|
ldap_user: _LDAPUser) -> Tuple[UserProfile, bool]: # nocoverage
|
|
(user, built) = super().get_or_build_user(username, ldap_user)
|
|
self.sync_avatar_from_ldap(user, ldap_user)
|
|
if 'userAccountControl' in settings.AUTH_LDAP_USER_ATTR_MAP:
|
|
user_disabled_in_ldap = self.is_account_control_disabled_user(ldap_user)
|
|
if user_disabled_in_ldap and user.is_active:
|
|
logging.info("Deactivating user %s because they are disabled in LDAP." %
|
|
(user.email,))
|
|
do_deactivate_user(user)
|
|
return (user, built)
|
|
if not user_disabled_in_ldap and not user.is_active:
|
|
logging.info("Reactivating user %s because they are not disabled in LDAP." %
|
|
(user.email,))
|
|
do_reactivate_user(user)
|
|
return (user, built)
|
|
|
|
class ZulipLDAPAuthBackend(ZulipLDAPAuthBackendBase):
|
|
REALM_IS_NONE_ERROR = 1
|
|
|
|
def authenticate(self, username: str, password: str, realm: Optional[Realm]=None,
|
|
return_data: Optional[Dict[str, Any]]=None) -> Optional[UserProfile]:
|
|
if realm is None:
|
|
return None
|
|
self._realm = realm
|
|
if not ldap_auth_enabled(realm):
|
|
return None
|
|
|
|
try:
|
|
username = self.django_to_ldap_username(username)
|
|
except ZulipLDAPExceptionOutsideDomain:
|
|
if return_data is not None:
|
|
return_data['outside_ldap_domain'] = True
|
|
return None
|
|
|
|
return ZulipLDAPAuthBackendBase.authenticate(self,
|
|
request=None,
|
|
username=username,
|
|
password=password)
|
|
|
|
def get_or_build_user(self, username: str, ldap_user: _LDAPUser) -> Tuple[UserProfile, bool]:
|
|
return_data = {} # type: Dict[str, Any]
|
|
|
|
if settings.LDAP_EMAIL_ATTR is not None:
|
|
# Get email from ldap attributes.
|
|
if settings.LDAP_EMAIL_ATTR not in ldap_user.attrs:
|
|
return_data["ldap_missing_attribute"] = settings.LDAP_EMAIL_ATTR
|
|
raise ZulipLDAPException("LDAP user doesn't have the needed %s attribute" % (
|
|
settings.LDAP_EMAIL_ATTR,))
|
|
|
|
username = ldap_user.attrs[settings.LDAP_EMAIL_ATTR][0]
|
|
|
|
if 'userAccountControl' in settings.AUTH_LDAP_USER_ATTR_MAP: # nocoverage
|
|
ldap_disabled = self.is_account_control_disabled_user(ldap_user)
|
|
if ldap_disabled:
|
|
# Treat disabled users as deactivated in Zulip.
|
|
return_data["inactive_user"] = True
|
|
raise ZulipLDAPException("User has been deactivated")
|
|
|
|
user_profile = common_get_active_user(username, self._realm, return_data)
|
|
if user_profile is not None:
|
|
# An existing user, successfully authed; return it.
|
|
return user_profile, False
|
|
|
|
if return_data.get("inactive_realm"):
|
|
# This happens if there is a user account in a deactivated realm
|
|
raise ZulipLDAPException("Realm has been deactivated")
|
|
if return_data.get("inactive_user"):
|
|
raise ZulipLDAPException("User has been deactivated")
|
|
if return_data.get("invalid_subdomain"):
|
|
# TODO: Implement something in the caller for this to
|
|
# provide a nice user-facing error message for this
|
|
# situation (right now it just acts like any other auth
|
|
# failure).
|
|
raise ZulipLDAPException("Wrong subdomain")
|
|
if self._realm.deactivated:
|
|
# This happens if no account exists, but the realm is
|
|
# deactivated, so we shouldn't create a new user account
|
|
raise ZulipLDAPException("Realm has been deactivated")
|
|
|
|
# We have valid LDAP credentials; time to create an account.
|
|
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)
|
|
self.sync_avatar_from_ldap(user_profile, ldap_user)
|
|
|
|
return user_profile, True
|
|
|
|
# Just like ZulipLDAPAuthBackend, but doesn't let you log in.
|
|
class ZulipLDAPUserPopulator(ZulipLDAPAuthBackendBase):
|
|
def authenticate(self, username: str, password: str, realm: Optional[Realm]=None,
|
|
return_data: Optional[Dict[str, Any]]=None) -> 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, dev_auth_username: Optional[str]=None, realm: Optional[Realm]=None,
|
|
return_data: Optional[Dict[str, Any]]=None) -> Optional[UserProfile]:
|
|
assert dev_auth_username is not None
|
|
if realm is None:
|
|
return None
|
|
if not dev_auth_enabled(realm):
|
|
return None
|
|
return common_get_active_user(dev_auth_username, realm, return_data=return_data)
|
|
|
|
def social_associate_user_helper(backend: BaseAuth, return_data: Dict[str, Any],
|
|
*args: Any, **kwargs: Any) -> Optional[UserProfile]:
|
|
"""Responsible for doing the Zulip-account lookup and validation parts
|
|
of the Zulip Social auth pipeline (similar to the authenticate()
|
|
methods in most other auth backends in this file).
|
|
"""
|
|
subdomain = backend.strategy.session_get('subdomain')
|
|
realm = get_realm(subdomain)
|
|
if realm is None:
|
|
return_data["invalid_realm"] = True
|
|
return None
|
|
return_data["realm_id"] = realm.id
|
|
|
|
if not auth_enabled_helper([backend.auth_backend_name], realm):
|
|
return_data["auth_backend_disabled"] = True
|
|
return None
|
|
|
|
if 'auth_failed_reason' in kwargs.get('response', {}):
|
|
return_data["social_auth_failed_reason"] = kwargs['response']["auth_failed_reason"]
|
|
return None
|
|
elif hasattr(backend, 'get_verified_emails'):
|
|
# Some social backends, like GitHubAuthBackend, don't guarantee that
|
|
# the `details` data is validated.
|
|
verified_emails = backend.get_verified_emails(*args, **kwargs)
|
|
if len(verified_emails) == 0:
|
|
# TODO: Provide a nice error message screen to the user
|
|
# for this case, rather than just logging a warning.
|
|
logging.warning("Social auth (%s) failed because user has no verified emails" %
|
|
(backend.auth_backend_name,))
|
|
return_data["email_not_verified"] = True
|
|
return None
|
|
# TODO: ideally, we'd prompt the user for which email they
|
|
# want to use with another pipeline stage here.
|
|
validated_email = verified_emails[0]
|
|
else: # nocoverage
|
|
# This code path isn't used by GitHubAuthBackend
|
|
validated_email = kwargs["details"].get("email")
|
|
|
|
if not validated_email: # nocoverage
|
|
# This code path isn't used with GitHubAuthBackend, but may be relevant for other
|
|
# social auth backends.
|
|
return_data['invalid_email'] = True
|
|
return None
|
|
try:
|
|
validate_email(validated_email)
|
|
except ValidationError:
|
|
return_data['invalid_email'] = True
|
|
return None
|
|
|
|
return_data["valid_attestation"] = True
|
|
return_data['validated_email'] = validated_email
|
|
user_profile = common_get_active_user(validated_email, realm, return_data)
|
|
|
|
if 'fullname' in kwargs["details"]:
|
|
return_data["full_name"] = kwargs["details"]["fullname"]
|
|
else:
|
|
# If we add support for any of the social auth backends that
|
|
# don't provide this feature, we'll need to add code here.
|
|
raise AssertionError("Social auth backend doesn't provide fullname")
|
|
|
|
return user_profile
|
|
|
|
def social_auth_associate_user(
|
|
backend: BaseAuth,
|
|
*args: Any,
|
|
**kwargs: Any) -> Dict[str, Any]:
|
|
return_data = {} # type: Dict[str, Any]
|
|
user_profile = social_associate_user_helper(
|
|
backend, return_data, *args, **kwargs)
|
|
|
|
return {'user_profile': user_profile,
|
|
'return_data': return_data}
|
|
|
|
def social_auth_finish(backend: Any,
|
|
details: Dict[str, Any],
|
|
response: HttpResponse,
|
|
*args: Any,
|
|
**kwargs: Any) -> Optional[UserProfile]:
|
|
from zerver.views.auth import (login_or_register_remote_user,
|
|
redirect_and_log_into_subdomain)
|
|
|
|
user_profile = kwargs['user_profile']
|
|
return_data = kwargs['return_data']
|
|
|
|
no_verified_email = return_data.get("email_not_verified")
|
|
auth_backend_disabled = return_data.get('auth_backend_disabled')
|
|
inactive_user = return_data.get('inactive_user')
|
|
inactive_realm = return_data.get('inactive_realm')
|
|
invalid_realm = return_data.get('invalid_realm')
|
|
invalid_subdomain = return_data.get('invalid_subdomain')
|
|
invalid_email = return_data.get('invalid_email')
|
|
auth_failed_reason = return_data.get("social_auth_failed_reason")
|
|
|
|
if invalid_realm:
|
|
from zerver.views.auth import redirect_to_subdomain_login_url
|
|
return redirect_to_subdomain_login_url()
|
|
if auth_backend_disabled or inactive_user or inactive_realm or no_verified_email:
|
|
# 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(backend.auth_backend_name)
|
|
)
|
|
return None
|
|
|
|
if auth_failed_reason:
|
|
logging.info(auth_failed_reason)
|
|
return None
|
|
|
|
# Structurally, all the cases where we don't have an authenticated
|
|
# email for the user should be handled above; this assertion helps
|
|
# prevent any violations of that contract from resulting in a user
|
|
# being incorrectly authenticated.
|
|
assert return_data.get('valid_attestation') is True
|
|
|
|
strategy = backend.strategy # type: ignore # This comes from Python Social Auth.
|
|
email_address = return_data['validated_email']
|
|
full_name = return_data['full_name']
|
|
is_signup = strategy.session_get('is_signup') == '1'
|
|
redirect_to = strategy.session_get('next')
|
|
realm = Realm.objects.get(id=return_data["realm_id"])
|
|
|
|
mobile_flow_otp = strategy.session_get('mobile_flow_otp')
|
|
if mobile_flow_otp is not None:
|
|
return login_or_register_remote_user(strategy.request, email_address,
|
|
user_profile, full_name,
|
|
invalid_subdomain=bool(invalid_subdomain),
|
|
mobile_flow_otp=mobile_flow_otp,
|
|
is_signup=is_signup,
|
|
redirect_to=redirect_to)
|
|
return redirect_and_log_into_subdomain(realm, full_name, email_address,
|
|
is_signup=is_signup,
|
|
redirect_to=redirect_to)
|
|
|
|
class SocialAuthMixin(ZulipAuthMixin):
|
|
auth_backend_name = "undeclared"
|
|
|
|
def auth_complete(self, *args: Any, **kwargs: Any) -> Optional[HttpResponse]:
|
|
"""This is a small wrapper around the core `auth_complete` method of
|
|
python-social-auth, designed primarily to prevent 500s for
|
|
exceptions in the social auth code from situations that are
|
|
really user errors. 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().auth_complete(*args, **kwargs) # type: ignore # monkey-patching
|
|
except AuthFailed as e:
|
|
# When a user's social authentication fails (e.g. because
|
|
# they did something funny with reloading in the middle of
|
|
# the flow), don't throw a 500, just send them back to the
|
|
# login page and record the event at the info log level.
|
|
logging.info(str(e))
|
|
return None
|
|
except SocialAuthBaseException as e:
|
|
# Other python-social-auth exceptions are likely
|
|
# interesting enough that we should log a warning.
|
|
logging.warning(str(e))
|
|
return None
|
|
|
|
class GitHubAuthBackend(SocialAuthMixin, GithubOAuth2):
|
|
auth_backend_name = "GitHub"
|
|
|
|
def get_verified_emails(self, *args: Any, **kwargs: Any) -> List[str]:
|
|
access_token = kwargs["response"]["access_token"]
|
|
try:
|
|
emails = self._user_data(access_token, '/emails')
|
|
except (HTTPError, ValueError, TypeError): # nocoverage
|
|
# We don't really need an explicit test for this code
|
|
# path, since the outcome will be the same as any other
|
|
# case without any verified emails
|
|
emails = []
|
|
|
|
verified_emails = [] # type: List[str]
|
|
for email_obj in emails:
|
|
if not email_obj.get("verified"):
|
|
continue
|
|
# social_associate_user_helper assumes that the first email in
|
|
# verified_emails is primary.
|
|
if email_obj.get("primary"):
|
|
verified_emails.insert(0, email_obj["email"])
|
|
else:
|
|
verified_emails.append(email_obj["email"])
|
|
|
|
return verified_emails
|
|
|
|
def user_data(self, access_token: str, *args: Any, **kwargs: Any) -> Dict[str, str]:
|
|
"""This patched user_data function lets us combine together the 3
|
|
social auth backends into a single Zulip backend for GitHub Oauth2"""
|
|
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:
|
|
# I believe this can't raise AuthFailed, so we don't try to catch it here.
|
|
return super().user_data(
|
|
access_token, *args, **kwargs
|
|
)
|
|
elif team_id is not None:
|
|
backend = GithubTeamOAuth2(self.strategy, self.redirect_uri)
|
|
try:
|
|
return backend.user_data(access_token, *args, **kwargs)
|
|
except AuthFailed:
|
|
return dict(auth_failed_reason="GitHub user is not member of required team")
|
|
elif org_name is not None:
|
|
backend = GithubOrganizationOAuth2(self.strategy, self.redirect_uri)
|
|
try:
|
|
return backend.user_data(access_token, *args, **kwargs)
|
|
except AuthFailed:
|
|
return dict(auth_failed_reason="GitHub user is not member of required organization")
|
|
|
|
raise AssertionError("Invalid configuration")
|
|
|
|
class AzureADAuthBackend(SocialAuthMixin, AzureADOAuth2):
|
|
auth_backend_name = "AzureAD"
|
|
|
|
AUTH_BACKEND_NAME_MAP = {
|
|
'Dev': DevAuthBackend,
|
|
'Email': EmailAuthBackend,
|
|
'Google': GoogleMobileOauth2Backend,
|
|
'LDAP': ZulipLDAPAuthBackend,
|
|
'RemoteUser': ZulipRemoteUserBackend,
|
|
} # type: Dict[str, Any]
|
|
OAUTH_BACKEND_NAMES = ["Google"] # type: List[str]
|
|
|
|
# Authomatically add all of our social auth backends to relevant data structures.
|
|
for social_auth_subclass in SocialAuthMixin.__subclasses__():
|
|
AUTH_BACKEND_NAME_MAP[social_auth_subclass.auth_backend_name] = social_auth_subclass
|
|
if issubclass(social_auth_subclass, BaseOAuth2):
|
|
OAUTH_BACKEND_NAMES.append(social_auth_subclass.auth_backend_name)
|