cr-connect-workflow/crc/services/user_service.py

from flask import g

from crc import session
from crc.api.common import ApiError
from crc.models.user import UserModel, AdminSessionModel


class UserService(object):
    """Provides common tools for working with users"""

    # Returns true if the current user is logged in.
    @staticmethod
    def has_user():
        return 'token' in g and \
               bool(g.token) and \
               'user' in g and \
               bool(g.user)

    # Returns true if the current user is an admin.
    @staticmethod
    def user_is_admin():
        return UserService.has_user() and g.user.is_admin()

    # Returns true if the current admin user is impersonating another user.
    @staticmethod
    def admin_is_impersonating():
        if UserService.user_is_admin():
            adminSession: AdminSessionModel = UserService.get_admin_session()
            return adminSession is not None

        else:
            raise ApiError("unauthorized", "You do not have permissions to do this.", status_code=403)

    # Returns true if the given user uid is different from the current user's uid.
    @staticmethod
    def is_different_user(uid):
        return UserService.has_user() and uid is not None and uid is not g.user.uid

    @staticmethod
    def current_user(allow_admin_impersonate=False) -> UserModel:
        if not UserService.has_user():
            raise ApiError("logged_out", "You are no longer logged in.", status_code=401)

        # Admins can pretend to be different users and act on a user's behalf in
        # some circumstances.
        if UserService.user_is_admin() and allow_admin_impersonate and UserService.admin_is_impersonating():
            return UserService.get_admin_session_user()
        else:
            return g.user

    # Admins can pretend to be different users and act on a user's behalf in some circumstances.
    # This method allows an admin user to start impersonating another user with the given uid.
    # Stops impersonating if the uid is None or invalid.
    @staticmethod
    def start_impersonating(uid=None):
        if not UserService.has_user():
            raise ApiError("logged_out", "You are no longer logged in.", status_code=401)

        if not UserService.user_is_admin():
            raise ApiError("unauthorized", "You do not have permissions to do this.", status_code=403)

        if uid is None:
            raise ApiError("invalid_uid", "Please provide a valid user uid.")

        if not UserService.admin_is_impersonating() and UserService.is_different_user(uid):
            # Impersonate the user if the given uid is valid.
            impersonate_user = session.query(UserModel).filter(UserModel.uid == uid).first()

            if impersonate_user is not None:
                g.impersonate_user = impersonate_user

                # Store the uid and user session token.
                session.add(AdminSessionModel(token=g.token, admin_impersonate_uid=uid))
                session.commit()
            else:
                raise ApiError("invalid_uid", "The uid provided is not valid.")

    @staticmethod
    def stop_impersonating():
        if not UserService.has_user():
            raise ApiError("logged_out", "You are no longer logged in.", status_code=401)

        # Clear out the current impersonating user.
        if 'impersonate_user' in g:
            del g.impersonate_user

        admin_session: AdminSessionModel = UserService.get_admin_session()
        if admin_session:
            session.delete(admin_session)
            session.commit()

    @staticmethod
    def in_list(uids, allow_admin_impersonate=False):
        """Returns true if the current user's id is in the given list of ids.  False if there
        is no user, or the user is not in the list."""
        if UserService.has_user():  # If someone is logged in, lock tasks that don't belong to them.
            user = UserService.current_user(allow_admin_impersonate)
            if user.uid in uids:
                return True
        return False

    @staticmethod
    def get_admin_session() -> AdminSessionModel:
        if UserService.user_is_admin():
            return session.query(AdminSessionModel).filter(AdminSessionModel.token == g.token).first()
        else:
            raise ApiError("unauthorized", "You do not have permissions to do this.", status_code=403)

    @staticmethod
    def get_admin_session_user() -> UserModel:
        if UserService.user_is_admin():
            admin_session = UserService.get_admin_session()

            if admin_session is not None:
                return session.query(UserModel).filter(UserModel.uid == admin_session.admin_impersonate_uid).first()
        else:
            raise ApiError("unauthorized", "You do not have permissions to do this.", status_code=403)