django-for-runners_ynh/conf/ynh_authenticate.py

"""
    * remote user authentication backend
    * remote user middleware

    Note: SSOwat/nginx add authentication headers:

        'HTTP_AUTHORIZATION': 'Basic XXXXXXXXXXXXXXXX='
        'HTTP_AUTH_USER': 'username'
        'HTTP_REMOTE_USER': 'username'

    Basic auth contains "{username}:{plaintext-password}"

    and we get SSOwat cookies like:

         'HTTP_COOKIE': 'SSOwAuthUser=username; '
                        'SSOwAuthHash=593876aa66...99e69f88af1e; '
                        'SSOwAuthExpire=1609227697.998; '

    * Login a user via HTTP_REMOTE_USER header, but check also username in:
        * SSOwAuthUser
        * HTTP_AUTH_USER
        * HTTP_AUTHORIZATION (Basic auth)
    * Create new users
    * Update Email, First / Last name for existing users
"""
import base64
import logging

from axes.exceptions import AxesBackendPermissionDenied
from django.contrib.auth.backends import RemoteUserBackend as OriginRemoteUserBackend
from django.contrib.auth.middleware import RemoteUserMiddleware as OriginRemoteUserMiddleware
from django.core.exceptions import ValidationError
from inventory.permissions import get_or_create_normal_user_group

logger = logging.getLogger(__name__)


def update_user_profile(request):
    """
    Update existing user information:
     * Email
     * First / Last name
    """
    user = request.user
    assert user.is_authenticated

    update_fields = []

    if not user.password:
        # Empty password is not valid, so we can't save the model, because of full_clean() call
        logger.info('Set unusable password for user: %s', user)
        user.set_unusable_password()
        update_fields.append('password')

    email = request.META.get('HTTP_EMAIL')
    if email and user.email != email:
        logger.info('Update email: %r -> %r', user.email, email)
        user.email = email
        update_fields.append('email')

    raw_username = request.META.get('HTTP_NAME')
    if raw_username:
        if ' ' in raw_username:
            first_name, last_name = raw_username.split(' ', 1)
        else:
            first_name = ''
            last_name = raw_username

        if user.first_name != first_name:
            logger.info('Update first name: %r -> %r', user.first_name, first_name)
            user.first_name = first_name
            update_fields.append('first_name')

        if user.last_name != last_name:
            logger.info('Update last name: %r -> %r', user.last_name, last_name)
            user.last_name = last_name
            update_fields.append('last_name')

    if update_fields:
        try:
            user.full_clean()
        except ValidationError:
            logger.exception('Can not update user: %s', user)
        else:
            user.save(update_fields=update_fields)


class RemoteUserMiddleware(OriginRemoteUserMiddleware):
    """
    Middleware to login a user HTTP_REMOTE_USER header.
    Use Django Axes if something is wrong.
    Update exising user informations.
    """
    header = 'HTTP_REMOTE_USER'
    force_logout_if_no_header = True

    def process_request(self, request):
        # Keep the information if the user is already logged in
        was_authenticated = request.user.is_authenticated

        super().process_request(request)  # login remote user

        if not request.user.is_authenticated:
            # Not logged in -> nothing to verify here
            return

        # Check SSOwat cookie informations:
        try:
            username = request.COOKIES['SSOwAuthUser']
        except KeyError:
            logger.error('SSOwAuthUser cookie missing!')

            # emits a signal indicating user login failed, which is processed by
            # axes.signals.log_user_login_failed which logs and flags the failed request.
            raise AxesBackendPermissionDenied('Cookie missing')

        logger.info('SSOwat username from cookies: %r', username)
        if username != request.user.username:
            raise AxesBackendPermissionDenied('Wrong username')

        # Compare with HTTP_AUTH_USER
        try:
            username = request.META['HTTP_AUTH_USER']
        except KeyError:
            logger.error('HTTP_AUTH_USER missing!')
            raise AxesBackendPermissionDenied('No HTTP_AUTH_USER')

        if username != request.user.username:
            raise AxesBackendPermissionDenied('Wrong HTTP_AUTH_USER username')

        # Also check 'HTTP_AUTHORIZATION', but only the username ;)
        try:
            auth = request.META['HTTP_AUTHORIZATION']
        except KeyError:
            logger.error('HTTP_AUTHORIZATION missing!')
            raise AxesBackendPermissionDenied('No HTTP_AUTHORIZATION')

        scheme, creds = auth.split(' ', 1)
        if scheme.lower() != 'basic':
            logger.error('HTTP_AUTHORIZATION with %r not supported', scheme)
            raise AxesBackendPermissionDenied('HTTP_AUTHORIZATION scheme not supported')

        creds = str(base64.b64decode(creds), encoding='utf-8')
        username = creds.split(':', 1)[0]
        if username != request.user.username:
            raise AxesBackendPermissionDenied('Wrong HTTP_AUTHORIZATION username')

        if not was_authenticated:
            # First request, after login -> update user informations
            logger.info('Remote used was logged in')
            update_user_profile(request)


class RemoteUserBackend(OriginRemoteUserBackend):
    """
    Authentication backend via SSO/nginx header
    """
    create_unknown_user = True

    def authenticate(self, request, remote_user):
        logger.info('Remote user authenticate: %r', remote_user)
        return super().authenticate(request, remote_user)

    def configure_user(self, request, user):
        """
        Configure a user after creation and return the updated user.
        Setup a normal, non-superuser
        """
        logger.warning('Configure user %s', user)

        user.set_unusable_password()  # Always login via SSO
        user.is_staff = True
        user.is_superuser = False
        user.save()

        pyinventory_user_group = get_or_create_normal_user_group()[0]
        user.groups.set([pyinventory_user_group])

        update_user_profile(request)

        return user

    def user_can_authenticate(self, user):
        logger.warning('Remote user login: %s', user)
        return True