jumpserver/apps/common/permissions.py

# -*- coding: utf-8 -*-
#
import time
from rest_framework import permissions
from django.conf import settings
from common.exceptions import MFAVerifyRequired

from orgs.utils import current_org
from common.utils import is_uuid


class IsValidUser(permissions.IsAuthenticated, permissions.BasePermission):
    """Allows access to valid user, is active and not expired"""

    def has_permission(self, request, view):
        return super(IsValidUser, self).has_permission(request, view) \
            and request.user.is_valid


class IsAppUser(IsValidUser):
    """Allows access only to app user """

    def has_permission(self, request, view):
        return super(IsAppUser, self).has_permission(request, view) \
            and request.user.is_app


class IsSuperUser(IsValidUser):
    def has_permission(self, request, view):
        return super(IsSuperUser, self).has_permission(request, view) \
               and request.user.is_superuser


class IsSuperUserOrAppUser(IsSuperUser):
    def has_permission(self, request, view):
        return super(IsSuperUserOrAppUser, self).has_permission(request, view) \
            or request.user.is_app


class IsSuperAuditor(IsValidUser):
    def has_permission(self, request, view):
        return super(IsSuperAuditor, self).has_permission(request, view) \
               and request.user.is_super_auditor


class IsOrgAuditor(IsValidUser):
    def has_permission(self, request, view):
        if not current_org:
            return False
        return super(IsOrgAuditor, self).has_permission(request, view) \
               and current_org.can_audit_by(request.user)


class IsOrgAdmin(IsValidUser):
    """Allows access only to superuser"""

    def has_permission(self, request, view):
        if not current_org:
            return False
        return super(IsOrgAdmin, self).has_permission(request, view) \
            and current_org.can_admin_by(request.user)


class IsOrgAdminOrAppUser(IsValidUser):
    """Allows access between superuser and app user"""

    def has_permission(self, request, view):
        if not current_org:
            return False
        return super(IsOrgAdminOrAppUser, self).has_permission(request, view) \
            and (current_org.can_admin_by(request.user) or request.user.is_app)


class IsOrgAdminOrAppUserOrUserReadonly(IsOrgAdminOrAppUser):
    def has_permission(self, request, view):
        if IsValidUser.has_permission(self, request, view) \
                and request.method in permissions.SAFE_METHODS:
            return True
        else:
            return IsOrgAdminOrAppUser.has_permission(self, request, view)


class IsCurrentUserOrReadOnly(permissions.BasePermission):
    def has_object_permission(self, request, view, obj):
        if request.method in permissions.SAFE_METHODS:
            return True
        return obj == request.user


class WithBootstrapToken(permissions.BasePermission):
    def has_permission(self, request, view):
        authorization = request.META.get('HTTP_AUTHORIZATION', '')
        if not authorization:
            return False
        request_bootstrap_token = authorization.split()[-1]
        return settings.BOOTSTRAP_TOKEN == request_bootstrap_token


class UserCanAnyPermCurrentOrg(permissions.BasePermission):
    def has_permission(self, request, view):
        return current_org.can_any_by(request.user)


class UserCanUpdatePassword(permissions.BasePermission):
    def has_permission(self, request, view):
        return request.user.can_update_password()


class UserCanUpdateSSHKey(permissions.BasePermission):
    def has_permission(self, request, view):
        return request.user.can_update_ssh_key()


class NeedMFAVerify(permissions.BasePermission):
    def has_permission(self, request, view):
        if not settings.SECURITY_VIEW_AUTH_NEED_MFA:
            return True

        mfa_verify_time = request.session.get('MFA_VERIFY_TIME', 0)
        if time.time() - mfa_verify_time < settings.SECURITY_MFA_VERIFY_TTL:
            return True
        raise MFAVerifyRequired()


class CanUpdateDeleteUser(permissions.BasePermission):

    @staticmethod
    def has_delete_object_permission(request, view, obj):
        if request.user.is_anonymous:
            return False
        if not request.user.can_admin_current_org:
            return False
        # 超级管理员 / 组织管理员
        if str(request.user.id) == str(obj.id):
            return False
        # 超级管理员
        if request.user.is_superuser:
            if obj.is_superuser and obj.username in ['admin']:
                return False
            return True
        # 组织管理员
        if obj.is_superuser:
            return False
        if obj.is_super_auditor:
            return False
        if obj.can_admin_current_org:
            return False
        return True

    @staticmethod
    def has_update_object_permission(request, view, obj):
        if request.user.is_anonymous:
            return False
        if not request.user.can_admin_current_org:
            return False
        # 超级管理员 / 组织管理员
        if str(request.user.id) == str(obj.id):
            return True
        # 超级管理员
        if request.user.is_superuser:
            return True
        # 组织管理员
        if obj.is_superuser:
            return False
        if obj.is_super_auditor:
            return False
        return True

    def has_object_permission(self, request, view, obj):
        if request.user.is_anonymous:
            return False
        if not request.user.can_admin_current_org:
            return False
        if request.method in ['DELETE']:
            return self.has_delete_object_permission(request, view, obj)
        if request.method in ['PUT', 'PATCH']:
            return self.has_update_object_permission(request, view, obj)
        return True


class IsObjectOwner(IsValidUser):
    def has_object_permission(self, request, view, obj):
        return (super().has_object_permission(request, view, obj) and
                request.user == getattr(obj, 'user', None))


class HasQueryParamsUserAndIsCurrentOrgMember(permissions.BasePermission):
    def has_permission(self, request, view):
        query_user_id = request.query_params.get('user')
        if not query_user_id or not is_uuid(query_user_id):
            return False
        query_user = current_org.get_members().filter(id=query_user_id).first()
        return bool(query_user)