authentication_backends.py

from mozilla_django_oidc.auth import OIDCAuthenticationBackend
import logging
from lofar.sas.tmss.tmss.tmssapp.models import ProjectRole
from django.contrib.auth.models import Group
from rest_framework.authtoken.models import Token

logger = logging.getLogger(__name__)

class TMSSOIDCAuthenticationBackend(OIDCAuthenticationBackend):
    """
    A custom OIDCAuthenticationBackend, that allows us to perform extra actions when a user gets authenticated,
    most importantly we can assign the user's system and project roles according to the claims that we get from the
    identity provider.
    """

    def _set_user_project_roles_from_claims(self, user, claims):
        project_roles = []
        project_role_prefix = 'urn:mace:astron.nl:science:group:lofar:project:'
        for entitlement in claims.get('eduperson_entitlement', []):
            try:
                if entitlement.startswith(project_role_prefix):
                    project_entitlement = entitlement.replace(project_role_prefix, '')
                    if 'role' in project_entitlement:
                        project_name, role_name = project_entitlement.split(':role=')
                        if ProjectRole.objects.filter(value=role_name).count() > 0:
                            project_roles.append({'project': project_name, 'role': role_name})
                        else:
                            logger.error('could not handle entitlement=%s because no project role exists that matches the entitlement role=%s' % (entitlement, role_name))
                    elif ':' not in project_entitlement:
                        # TMSS only has project roles, we interpret 'general' membership in a project as a co_i role,
                        # but may make that explicit in Keycloak later on
                        project_roles.append({'project': project_entitlement, 'role': 'co_i'})
            except Exception as e:
                logger.error('could not handle entitlement=%s because of exception=%s' % (entitlement, e))
        user.project_roles = project_roles
        logger.info("### assigned project_roles=%s to user=%s" % (project_roles, user))  # todo: review GDPR
        user.save()

    def _set_user_system_roles_from_claims(self, user, claims):
        groups = []
        system_role_prefix = 'urn:mace:astron.nl:science:group:lofar:role='
        for entitlement in claims.get('eduperson_entitlement', []):
            try:
                if entitlement.startswith(system_role_prefix):
                    role_name = entitlement.replace(system_role_prefix, '')
                    role_name = role_name.replace('_', ' ')
                    if role_name in ['admin', 'developer']:
                        # instead of granting all kinds of explicit permissions to admin/developer roles, we mark them
                        # as superuser, which then essentially bypasses the enforcement of permissions during request
                        # handling.
                        # todo: confirm that this is ok for developer role!
                        user.is_superuser = True
                    if Group.objects.filter(name__iexact=role_name).count() > 0:
                        groups.append(Group.objects.filter(name__iexact=role_name).first())
                    else:
                        logger.error('could not handle entitlement=%s because no system role / group exists that matches the entitlement role=%s' % (entitlement, role_name))
            except Exception as e:
                logger.error('could not handle entitlement=%s because of exception=%s' % (entitlement, e))
        logger.info("### assigned groups=%s to user=%s" % (groups, user))  # todo: review GDPR
        user.groups.set(groups)
        user.save()

    def create_user(self, claims):
        user = super(TMSSOIDCAuthenticationBackend, self).create_user(claims)
        logger.info('create user=%s claims=%s' % (user, claims))
        # take some more user details from claims  # todo: check GDPR compliance!
        user.first_name = claims.get('given_name', user.first_name)
        user.last_name = claims.get('family_name', user.last_name)
        user.username = claims.get('preferred_username', user.username)
        user.save()
        Token.objects.create(user=user)  # required for websockets
        self._set_user_project_roles_from_claims(user, claims)
        self._set_user_system_roles_from_claims(user, claims)
        return user

    def update_user(self, user, claims):
        logger.info('update user=%s claims=%s' % (user, claims))
        if not Token.objects.filter(user=user).first():
            Token.objects.create(user=user)  # required for websockets
        self._set_user_project_roles_from_claims(user, claims)
        self._set_user_system_roles_from_claims(user, claims)
        return user