seahub/thirdpart/shibboleth/middleware.py

from collections import OrderedDict
from fnmatch import fnmatch
import logging
import os
import sys

from django.conf import settings
from django.contrib.auth.middleware import RemoteUserMiddleware
from django.core.exceptions import ImproperlyConfigured
from django.urls import reverse
from django.http import HttpResponseRedirect
from seaserv import seafile_api, ccnet_api

from shibboleth.app_settings import SHIB_ATTRIBUTE_MAP, LOGOUT_SESSION_KEY, \
        SHIB_USER_HEADER, SHIB_USER_HEADER_SECOND_UID

from seahub import auth
from seahub.base.accounts import User
from seahub.profile.models import Profile
from seahub.utils.file_size import get_quota_from_string
from seahub.role_permissions.utils import get_enabled_role_permissions_by_role
from seahub.utils.ccnet_db import CcnetDB

# Get an instance of a logger
logger = logging.getLogger(__name__)

try:
    conf_dir = os.environ['SEAFILE_CENTRAL_CONF_DIR']
    sys.path.append(conf_dir)
    try:
        from seahub_custom_functions import custom_shibboleth_get_user_role
        CUSTOM_SHIBBOLETH_GET_USER_ROLE = True
    except ImportError:
        CUSTOM_SHIBBOLETH_GET_USER_ROLE = False
except KeyError:
    CUSTOM_SHIBBOLETH_GET_USER_ROLE = False

SHIBBOLETH_PROVIDER_IDENTIFIER = getattr(settings, 'SHIBBOLETH_PROVIDER_IDENTIFIER', 'shibboleth')


class ShibbolethRemoteUserMiddleware(RemoteUserMiddleware):
    """
    Authentication Middleware for use with Shibboleth.
    """
    def process_request(self, request):
        if request.path.rstrip('/') != settings.SITE_ROOT + 'sso':
            return

        # AuthenticationMiddleware is required so that request.user exists.
        if not hasattr(request, 'user'):
            raise ImproperlyConfigured(
                "The Django remote user auth middleware requires the"
                " authentication middleware to be installed.  Edit your"
                " MIDDLEWARE setting to insert"
                " 'django.contrib.auth.middleware.AuthenticationMiddleware'"
                " before the RemoteUserMiddleware class.")

        # To support logout.  If this variable is True, do not
        # authenticate user and return now.
        if request.session.get(LOGOUT_SESSION_KEY) is True:
            return
        else:
            # Delete the shib reauth session key if present.
            request.session.pop(LOGOUT_SESSION_KEY, None)

        # Locate the remote user header.
        # import pprint; pprint.pprint(request.META)
        try:
            remote_user = request.META[SHIB_USER_HEADER]
        except KeyError:
            # If specified header doesn't exist then return (leaving
            # request.user set to AnonymousUser by the
            # AuthenticationMiddleware).
            return

        second_uid = request.META.get(SHIB_USER_HEADER_SECOND_UID, '')

        # If the user is already authenticated and that user is the user we are
        # getting passed in the headers, then the correct user is already
        # persisted in the session and we don't need to continue.
        if request.user.is_authenticated:
            return

        # Make sure we have all required Shiboleth elements before proceeding.
        shib_meta, error = self.parse_attributes(request)
        # Add parsed attributes to the session.
        request.session['shib'] = shib_meta
        if error:
            raise ShibbolethValidationError("All required Shibboleth elements"
                                            " not found.  %s" % shib_meta)

        # We are seeing this user for the first time in this session, attempt
        # to authenticate the user.
        user = auth.authenticate(remote_user=remote_user,
                                 shib_meta=shib_meta,
                                 second_uid=second_uid)
        if user:
            if not user.is_active:
                return HttpResponseRedirect(reverse('shib_complete'))

            # User is valid.  Set request.user and persist user in the session
            # by logging the user in.
            request.user = user
            auth.login(request, user)
            user.set_unusable_password()
            user.save()
            # call make profile.
            self.make_profile(user, shib_meta)

            db_api = CcnetDB()
            db_user_role =  db_api.get_user_role_from_db(user.email)
            if db_user_role.is_manual_set:
                user_role = db_user_role.role
            
            else:
                if CUSTOM_SHIBBOLETH_GET_USER_ROLE:
                    user_role = custom_shibboleth_get_user_role(shib_meta)
                    if user_role:
                        ccnet_api.update_role_emailuser(user.email, user_role, False)
                    else:
                        user_role = self.update_user_role(user, shib_meta, False)
                else:
                    user_role = self.update_user_role(user, shib_meta, False)

            if user_role:
                self.update_user_quota(user, user_role)

            # setup session.
            self.setup_session(request)
            request.shib_login = True

    def process_response(self, request, response):
        if getattr(request, 'shib_login', False):
            print('%s: set shibboleth cookie!' % id(self))
            self._set_auth_cookie(request, response)
        return response

    def _set_auth_cookie(self, request, response):
        from seahub.api2.utils import get_token_v1, get_token_v2
        # generate tokenv2 using information in request params
        keys = (
            'platform',
            'device_id',
            'device_name',
            'client_version',
            'platform_version',
        )
        if all(['shib_' + key in request.GET for key in keys]):
            platform = request.GET['shib_platform']
            device_id = request.GET['shib_device_id']
            device_name = request.GET['shib_device_name']
            client_version = request.GET['shib_client_version']
            platform_version = request.GET['shib_platform_version']
            token = get_token_v2(
                request, request.user.username, platform, device_id,
                device_name, client_version, platform_version)
        elif all(['shib_' + key not in request.GET for key in keys]):
            token = get_token_v1(request.user.username)
        else:
            return
        response.set_cookie('seahub_auth', request.user.username + '@' + token.key)

    def make_profile(self, user, shib_meta):
        """
        Extrat nickname(givenname surname), contact_email, institution from
        Shib attributs, and add those to user profile.
        """
        # use `display_name` as nickname in shib_meta first
        nickname = shib_meta.get('display_name', None)
        if nickname is None:
            # otherwise, fallback to givenname plus surname in shib_meta
            givenname = shib_meta.get('givenname', '')
            surname = shib_meta.get('surname', '')
            nickname = "%s %s" % (givenname, surname)

        institution = shib_meta.get('institution', None)
        contact_email = shib_meta.get('contact_email', None)

        p = Profile.objects.get_profile_by_user(user.username)
        if not p:
            p = Profile(user=user.username)

        if nickname.strip():  # set nickname when it's not empty
            p.nickname = nickname.encode("iso-8859-1").decode('utf8')

        if institution:
            p.institution = institution.encode("iso-8859-1").decode('utf8')

        if contact_email:
            p.contact_email = contact_email

        p.save()

    def _get_role_by_affiliation(self, affiliation):
        try:
            role_map = settings.SHIBBOLETH_AFFILIATION_ROLE_MAP
        except AttributeError:
            return

        role = role_map.get(affiliation)
        if role:
            return role

        if role_map.get('patterns') is not None:
            joker_map = role_map.get('patterns')
            try:
                od = OrderedDict(joker_map)
            except Exception as e:
                logger.error(e)
                return

            for k in od:
                if fnmatch(affiliation, k):
                    return od[k]

        return None

    def update_user_role(self, user, shib_meta, is_manual_set):
        affiliation = shib_meta.get('affiliation', '')
        if not affiliation:
            return

        for e in affiliation.split(';'):
            role = self._get_role_by_affiliation(e)
            if role:
                User.objects.update_role(user.email, role, is_manual_set)
                return role

    def update_user_quota(self, user, user_role):
        role_quota = get_enabled_role_permissions_by_role(user_role)['role_quota']
        if role_quota:
            quota = get_quota_from_string(role_quota)
            logger.info('Set quota[%d] for user: %s, role[%s]' % (quota, user.username, user_role))
            seafile_api.set_role_quota(user_role, quota)
        else:
            return

    def setup_session(self, request):
        """
        If you want to add custom code to setup user sessions, you
        can extend this.
        """
        return

    def parse_attributes(self, request):
        """
        Parse the incoming Shibboleth attributes.
        From: https://github.com/russell/django-shibboleth/blob/master/django_shibboleth/utils.py
        Pull the mapped attributes from the apache headers.
        """
        shib_attrs = {}
        error = False
        meta = request.META
        for header, attr in list(SHIB_ATTRIBUTE_MAP.items()):
            required, name = attr
            value = meta.get(header, None)
            shib_attrs[name] = value
            if not value or value == '':
                if required:
                    error = True
        return shib_attrs, error


class ShibbolethValidationError(Exception):
    pass