django_airavata/apps/auth/backends.py

"""Django Airavata Auth Backends: KeycloakBackend."""
import logging
import os
import time

import requests
from django.conf import settings
from django.contrib.auth.models import User
from django.views.decorators.debug import sensitive_variables
from oauthlib.oauth2 import InvalidGrantError, LegacyApplicationClient
from requests_oauthlib import OAuth2Session

from django_airavata.apps.auth.utils import get_authz_token

from . import models, utils

logger = logging.getLogger(__name__)


class KeycloakBackend(object):
    """Django authentication backend for Keycloak."""

    # mask all local variables from error emails since they contain the user's
    # password and/or client_secret. Note, we could selectively just hide
    # variables that are sensitive, but this decorator doesn't apply explicitly
    # listed variable masking to library function calls
    @sensitive_variables()
    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

    def get_user(self, user_id):
        try:
            return User.objects.get(pk=user_id)
        except User.DoesNotExist:
            return None

    def _get_token_and_userinfo_password_flow(self, username, password):
        try:
            client_id = settings.KEYCLOAK_CLIENT_ID
            client_secret = settings.KEYCLOAK_CLIENT_SECRET
            token_url = settings.KEYCLOAK_TOKEN_URL
            userinfo_url = settings.KEYCLOAK_USERINFO_URL
            verify_ssl = settings.KEYCLOAK_VERIFY_SSL
            scope = ['openid', 'profile', 'email']
            oauth2_session = OAuth2Session(client=LegacyApplicationClient(client_id=client_id), scope=scope)
            verify = verify_ssl
            if verify_ssl and hasattr(settings, 'KEYCLOAK_CA_CERTFILE'):
                verify = settings.KEYCLOAK_CA_CERTFILE
            token = oauth2_session.fetch_token(token_url=token_url,
                                               username=username,
                                               password=password,
                                               client_id=client_id,
                                               client_secret=client_secret,
                                               verify=verify,
                                               scope=scope)
            userinfo = oauth2_session.get(userinfo_url).json()
            return token, userinfo
        except InvalidGrantError as e:
            # password wasn't valid, just log as a warning
            logger.warning(f"Failed to log in user {username} with "
                           f"password: {e}")
            return None, None

    def _get_token_and_userinfo_redirect_flow(self, request):
        authorization_code_url = request.build_absolute_uri()
        client_id = settings.KEYCLOAK_CLIENT_ID
        client_secret = settings.KEYCLOAK_CLIENT_SECRET
        token_url = settings.KEYCLOAK_TOKEN_URL
        userinfo_url = settings.KEYCLOAK_USERINFO_URL
        verify_ssl = settings.KEYCLOAK_VERIFY_SSL
        state = request.session['OAUTH2_STATE']
        redirect_uri = request.session['OAUTH2_REDIRECT_URI']
        logger.debug("state={}".format(state))
        oauth2_session = OAuth2Session(client_id,
                                       scope='openid profile email',
                                       redirect_uri=redirect_uri,
                                       state=state)
        verify = verify_ssl
        if verify_ssl and hasattr(settings, 'KEYCLOAK_CA_CERTFILE'):
            verify = settings.KEYCLOAK_CA_CERTFILE
        if not request.is_secure() and settings.DEBUG and not os.environ.get('OAUTHLIB_INSECURE_TRANSPORT'):
            # For local development (DEBUG=True), allow insecure OAuth redirect flow
            # if OAUTHLIB_INSECURE_TRANSPORT isn't already set
            os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = "1"
            logger.info("Adding env var OAUTHLIB_INSECURE_TRANSPORT=1 to allow "
                        "OAuth redirect flow even though request is not secure")
        token = oauth2_session.fetch_token(
            token_url, client_secret=client_secret,
            authorization_response=authorization_code_url, verify=verify)
        userinfo = oauth2_session.get(userinfo_url).json()
        return token, userinfo

    def _get_token_and_userinfo_from_refresh_token(self,
                                                   request,
                                                   refresh_token=None):
        client_id = settings.KEYCLOAK_CLIENT_ID
        client_secret = settings.KEYCLOAK_CLIENT_SECRET
        token_url = settings.KEYCLOAK_TOKEN_URL
        userinfo_url = settings.KEYCLOAK_USERINFO_URL
        verify_ssl = settings.KEYCLOAK_VERIFY_SSL
        oauth2_session = OAuth2Session(client_id, scope='openid profile email')
        verify = verify_ssl
        if verify_ssl and hasattr(settings, 'KEYCLOAK_CA_CERTFILE'):
            verify = settings.KEYCLOAK_CA_CERTFILE
        refresh_token_ = (refresh_token
                          if refresh_token is not None
                          else request.session['REFRESH_TOKEN'])
        # refresh_token doesn't take client_secret kwarg, so create auth
        # explicitly
        auth = requests.auth.HTTPBasicAuth(client_id, client_secret)
        try:
            token = oauth2_session.refresh_token(token_url=token_url,
                                                 refresh_token=refresh_token_,
                                                 auth=auth,
                                                 verify=verify)
            userinfo = oauth2_session.get(userinfo_url).json()
            return token, userinfo
        except InvalidGrantError as e:
            # probably session was terminated by admin or by user logging out in another client
            logger.warning(f"Failed to refresh token for user {request.user.username} "
                           f": {e}")
            return None, None

    def _get_userinfo_from_token(self, request, token):
        client_id = settings.KEYCLOAK_CLIENT_ID
        userinfo_url = settings.KEYCLOAK_USERINFO_URL
        verify_ssl = settings.KEYCLOAK_VERIFY_SSL
        oauth2_session = OAuth2Session(
            client_id, token={'access_token': token})
        verify = verify_ssl
        if verify_ssl and hasattr(settings, 'KEYCLOAK_CA_CERTFILE'):
            verify = settings.KEYCLOAK_CA_CERTFILE
        userinfo = oauth2_session.get(
            userinfo_url, verify=verify).json()
        if 'error' in userinfo:
            msg = userinfo.get('error_description')
            if msg is None:
                msg = f"Error fetching userinfo: {userinfo['error']}"
            raise Exception(msg)
        return userinfo

    def _process_token(self, request, token):
        # TODO validate the JWS signature
        logger.debug("token: {}".format(token))
        now = time.time()
        # Put access_token into session to be used for authenticating with API
        # server
        sess = request.session
        sess['ACCESS_TOKEN'] = token['access_token']
        sess['ACCESS_TOKEN_EXPIRES_AT'] = now + token['expires_in']
        sess['REFRESH_TOKEN'] = token['refresh_token']
        sess['REFRESH_TOKEN_EXPIRES_AT'] = now + token['refresh_expires_in']

    def _process_userinfo(self, request, userinfo):
        logger.debug("userinfo: {}".format(userinfo))
        sub = userinfo['sub']
        username = userinfo['preferred_username']
        email = userinfo.get('email', '')
        first_name = userinfo.get('given_name', None)
        last_name = userinfo.get('family_name', None)

        user = self._get_or_create_user(sub, username)
        user_profile = user.user_profile

        # Save the user info claims
        for (claim, value) in userinfo.items():
            if user_profile.userinfo_set.filter(claim=claim).exists():
                userinfo_claim = user_profile.userinfo_set.get(claim=claim)
                userinfo_claim.value = value
                userinfo_claim.save()
            else:
                user_profile.userinfo_set.create(claim=claim, value=value)

        # Update User model fields
        user = user_profile.user
        user.username = username
        user.email = email
        user.first_name = first_name
        user.last_name = last_name
        user.save()

        return user

    def _get_or_create_user(self, sub, username):

        try:
            user_profile = models.UserProfile.objects.get(
                userinfo__claim='sub', userinfo__value=sub)
            return user_profile.user
        except models.UserProfile.DoesNotExist:
            try:
                # For backwards compatibility, lookup by username
                user = User.objects.get(username=username)
                # Make sure there is a user_profile with the sub claim, which
                # will be used to do the lookup next time
                if not hasattr(user, 'user_profile'):
                    user_profile = models.UserProfile(user=user)
                    user_profile.save()
                    user_profile.userinfo_set.create(
                        claim='sub', value=sub)
                else:
                    userinfo = user.user_profile.userinfo_set.get(claim='sub')
                    logger.warning(
                        f"User {username} exists but sub claims don't match: "
                        f"old={userinfo.value}, new={sub}. Updating to new "
                        "sub claim.")
                    userinfo.value = sub
                    userinfo.save()
                return user
            except User.DoesNotExist:
                user = User(username=username)
                user.save()
                user_profile = models.UserProfile(user=user)
                user_profile.save()
                user_profile.userinfo_set.create(claim='sub', value=sub)
                return user

    def _store_idp_userinfo(self, user, token, idp_alias):
        try:
            idp_token_url = None
            userinfo_url = None
            for auth_option in settings.AUTHENTICATION_OPTIONS['external']:
                if auth_option['idp_alias'] == idp_alias:
                    idp_token_url = auth_option.get('idp_token_url')
                    userinfo_url = auth_option.get('userinfo_url')
                    break
            if idp_token_url is None or userinfo_url is None:
                logger.debug(f"idp_token_url and/or userinfo_url not set for {idp_alias} "
                             "in AUTHENTICATION_OPTIONS, skipping retrieval of external IDP userinfo")
                return
            access_token = token['access_token']
            logger.debug(f"access_token={access_token} for idp_alias={idp_alias}")
            # fetch the idp's token
            headers = {'Authorization': f'Bearer {access_token}'}
            # For the following to work, in Keycloak the IDP should have 'Store
            # Tokens' and 'Stored Tokens Readable' enabled and the user needs
            # the broker/read-token role
            r = requests.get(idp_token_url, headers=headers)
            idp_token = r.json()
            idp_headers = {'Authorization': f"Bearer {idp_token['access_token']}"}
            r = requests.get(userinfo_url, headers=idp_headers)
            userinfo = r.json()
            logger.debug(f"userinfo={userinfo}")

            # Save the idp user info claims
            user_profile = user.user_profile
            for (claim, value) in userinfo.items():
                if user_profile.idp_userinfo.filter(idp_alias=idp_alias, claim=claim).exists():
                    userinfo_claim = user_profile.idp_userinfo.get(idp_alias=idp_alias, claim=claim)
                    userinfo_claim.value = value
                    userinfo_claim.save()
                else:
                    user_profile.idp_userinfo.create(idp_alias=idp_alias, claim=claim, value=value)
        except Exception:
            logger.exception(f"Failed to store IDP userinfo for {user.username} from IDP {idp_alias}")

    def _check_username_initialization(self, request, user):
        # Check if the username assigned to the user was based on the user's
        # email address or if it was assigned some random string (Keycloak's
        # sub). If the latter, we'll want to alert the admins so that they can
        # assign a proper username for the user.
        user_profile = user.user_profile
        if (not user_profile.username_initialized and
            user_profile.userinfo_set.filter(claim='email').exists() and
            user_profile.userinfo_set.filter(claim='preferred_username').exists() and
                user_profile.userinfo_set.get(claim='email').value == user_profile.userinfo_set.get(claim='preferred_username').value):
            user_profile.username_initialized = True
            user_profile.save()

        # TODO: also check idp_userinfo.preferred_username if it exists

        if not user_profile.username_initialized and not user_profile.is_username_valid:
            try:
                utils.send_admin_alert_about_uninitialized_username(
                    request, user.username, user.email, user.first_name, user.last_name)
            except Exception:
                logger.exception(f"Failed to send alert about username being uninitialized: {user.username}", extra={'request': request})