def authenticate()

in django_airavata/apps/auth/backends.py [0:0]

• 59 lines of code
• 15 McCabe index (conditional complexity)

    def authenticate(self,
                     request=None,
                     username=None,
                     password=None,
                     refresh_token=None,
                     idp_alias=None):
        try:
            user = None
            access_token = None
            if username and password:
                token, userinfo = self._get_token_and_userinfo_password_flow(
                    username, password)
                if token is None:  # login failed
                    return None
                self._process_token(request, token)
                user = self._process_userinfo(request, userinfo)
                access_token = token['access_token']
            elif 'HTTP_AUTHORIZATION' in request.META:
                bearer, token = request.META.get('HTTP_AUTHORIZATION').split()
                if bearer != "Bearer":
                    raise Exception("Unexpected Authorization header")
                # implicitly validate token by using it to get userinfo
                userinfo = self._get_userinfo_from_token(request, token)
                user = self._process_userinfo(request, userinfo)
                access_token = token
            # user is already logged in and can use refresh token
            elif request.user.is_authenticated and not utils.is_refresh_token_expired(request):
                logger.debug("Refreshing token...")
                token, userinfo = \
                    self._get_token_and_userinfo_from_refresh_token(request)
                if token is None:  # refresh failed
                    return None
                self._process_token(request, token)
                # user is already logged in
                user = request.user
                access_token = token['access_token']
            elif refresh_token:
                logger.debug("Refreshing supplied token...")
                token, userinfo = \
                    self._get_token_and_userinfo_from_refresh_token(
                        request, refresh_token=refresh_token)
                if token is None:  # refresh failed
                    return None
                self._process_token(request, token)
                user = self._process_userinfo(request, userinfo)
                access_token = token['access_token']
            else:
                token, userinfo = self._get_token_and_userinfo_redirect_flow(
                    request)
                self._process_token(request, token)
                user = self._process_userinfo(request, userinfo)
                if idp_alias is not None:
                    self._store_idp_userinfo(user, token, idp_alias)
                    self._check_username_initialization(request, user)
                access_token = token['access_token']
            # authz_token_middleware has already run, so must manually add
            # the `request.authz_token` attribute
            if user is not None:
                request.authz_token = get_authz_token(
                    request, user=user, access_token=access_token)
            return user
        except Exception as e:
            logger.warning("login failed", exc_info=e)
            raise