pulseapi/users/adapter.py

from allauth.account.adapter import DefaultAccountAdapter
from allauth.socialaccount.adapter import DefaultSocialAccountAdapter
from allauth.account.utils import user_email
from allauth.socialaccount import providers
from allauth.socialaccount.models import SocialAccount
from allauth.socialaccount.providers.base import AuthProcess
from allauth.socialaccount.providers.google.provider import GoogleProvider
from allauth.exceptions import ImmediateHttpResponse
from allauth.utils import (
    email_address_exists,
    get_user_model,
)
from django.conf import settings
from django.http import HttpResponseRedirect, QueryDict
from django.urls import reverse
from urllib.parse import urlparse, urlunparse, parse_qsl, urlencode

from pulseapi.utility.userpermissions import is_staff_address
from pulseapi.profiles.models import UserProfile


google_provider_id = GoogleProvider.id


class PulseAccountAdapter(DefaultAccountAdapter):
    def is_open_for_signup(self, request):
        return settings.ALLOW_SIGNUP

    def is_safe_url(self, url):
        """
        We override this because the default implementation only
        allows redirects to same origin urls.
        """
        from django.utils.http import is_safe_url
        return is_safe_url(url, allowed_hosts=settings.LOGIN_ALLOWED_REDIRECT_DOMAINS)

    def get_email_confirmation_redirect_url(self, request):
        """
        Override this so that we can redirect to the `?next=` url
        provided since allauth does not do this out of the box
        """
        if request.user.is_authenticated:
            next_url = request.GET.get('next')
            if next_url and self.is_safe_url(next_url):
                return next_url

        return super().get_email_confirmation_redirect_url(request)

    def get_email_confirmation_url(self, request, emailconfirmation):
        """
        Override this so that we can set the `?next=` url in the email
        confirmation url so that the user is redirected correctly after
        confirming their email.

        FIXME: This currently does not work because no `?next=` parameter
        is passed to the email confirmation view. Our current workaround
        for this is to set the
        `ACCOUNT_EMAIL_CONFIRMATION_AUTHENTICATED_REDIRECT_URL` to be
        the pulse front-end url via an environment veriable on production.
        """
        url = super().get_email_confirmation_url(request, emailconfirmation)
        next_url = request.GET.get('next')

        if not (next_url and self.is_safe_url(next_url)):
            return url

        # Parse the url and add the `next` url to it as a query string
        url_parts = list(urlparse(url))
        qs = dict(parse_qsl(url_parts[4]))
        qs.update({'next': next_url})
        url_parts[4] = urlencode(qs)

        return urlunparse(url_parts)


class PulseSocialAccountAdapter(DefaultSocialAccountAdapter):
    """
    Override the default implementation for DefaultSocialAccountAdapter
    for two reasons:
        1. Auto populate our custom EmailUser model with information from
        the social provider
        2. Handle the transition from the old auth system to "upgrade" existing
        accounts to the new allauth system.
    """
    def is_open_for_signup(self, request, socialaccount):
        return settings.ALLOW_SIGNUP

    def populate_user(self, request, sociallogin, data):
        user = super().populate_user(request, sociallogin, data)

        name = data.get('name')
        if not name:
            first_name = data.get('first_name') or ''
            last_name = data.get('last_name') or ''
            name = f"{first_name} {last_name}".strip()
        user.name = name if name else 'Unnamed Pulse user'

        if sociallogin.account.provider == google_provider_id and is_staff_address(user_email(user)):
            user.is_staff = True

        return user

    def save_user(self, request, sociallogin, form=None):
        """
        For some reason the signal for creating a user is not triggered
        by allauth, so we manually create a profile when a user first
        logs in.
        """
        user = super().save_user(request, sociallogin, form)

        try:
            UserProfile.objects.get(related_user=user)
        except UserProfile.DoesNotExist:
            # Is_active is False by default, so we can hide this
            # users profile and entries, until set to active by a moderator.
            profile = UserProfile.objects.create(is_active=False)
            user.profile = profile
            user.save()

        return user

    def pre_social_login(self, request, sociallogin):
        email = user_email(sociallogin.user)
        UserModel = get_user_model()
        login_provider_id = sociallogin.account.provider

        if (
                not email or
                not email_address_exists(email) or
                sociallogin.state.get('process') != AuthProcess.LOGIN
        ):
            # This is a new email address, or we're connecting social accounts
            # so we don't need to do anything
            return

        try:
            user = UserModel.objects.get(email=email)
        except UserModel.DoesNotExist:
            # This case shouldn't really happen, but even if it does, we
            # don't do anything and let the default behavior kick in.
            return

        social_accounts = list(SocialAccount.objects.filter(
            user=user
        ).values_list('provider', flat=True))

        if len(social_accounts) == 0:
            # This is a hack to associate existing accounts on pulse
            # that were added via Google Auth the old way to the new allauth
            # system. We only do this for new logins into this system.
            request.migrate_user = user
        elif login_provider_id in social_accounts:
            # In this case, the existing user already has a social account
            # and is logging into it using the same provider.
            return
        else:
            # Here the user already has a Pulse social account (e.g. Google)
            # but is logging in through a different social network
            # (e.g. Github) that uses the same email for the first time.
            # We redirect them to the login view where they have to login
            # through their existing social account on Pulse before going to
            # the Social Account Connections view to connect their secondary
            # social account.
            url = reverse('account_login')
            qs = QueryDict(mutable=True)
            next_url = sociallogin.get_redirect_url(request)
            if next_url:
                # We encode the final destination url in the connection
                # view url so that users are correctly rerouted after
                # connecting their accounts.
                qs['next'] = next_url
            next_url = '{url}?{qs}'.format(
                url=reverse('socialaccount_connections'),
                qs=qs.urlencode()
            )
            qs = QueryDict(mutable=True)
            qs['next'] = next_url
            qs['promptconnection'] = True
            qs['provider'] = providers.registry.by_id(login_provider_id).name

            raise ImmediateHttpResponse(
                response=HttpResponseRedirect(f'{url}?{qs.urlencode()}')
            )

    def is_auto_signup_allowed(self, request, sociallogin):
        if hasattr(request, 'migrate_user'):
            # Associate the existing user with the SocialLogin object
            # being created
            sociallogin.user = request.migrate_user
            return True

        return super().is_auto_signup_allowed(request, sociallogin)