jumpserver/apps/users/serializers/user.py

# -*- coding: utf-8 -*-
#
from django.core.cache import cache
from django.conf import settings
from django.utils.translation import ugettext_lazy as _
from rest_framework import serializers

from common.utils import validate_ssh_public_key
from common.mixins import CommonBulkSerializerMixin
from common.permissions import CanUpdateDeleteUser
from orgs.models import ROLE as ORG_ROLE
from ..models import User


__all__ = [
    'UserSerializer', 'UserPKUpdateSerializer',
    'ChangeUserPasswordSerializer', 'ResetOTPSerializer',
    'UserProfileSerializer', 'UserOrgSerializer',
    'UserUpdatePasswordSerializer', 'UserUpdatePublicKeySerializer',
    'UserRetrieveSerializer', 'MiniUserSerializer', 'InviteSerializer'
]


class UserOrgSerializer(serializers.Serializer):
    id = serializers.CharField()
    name = serializers.CharField()


class UserOrgLabelSerializer(serializers.Serializer):
    value = serializers.CharField(source='id')
    label = serializers.CharField(source='name')


class UserSerializer(CommonBulkSerializerMixin, serializers.ModelSerializer):
    EMAIL_SET_PASSWORD = _('Reset link will be generated and sent to the user')
    CUSTOM_PASSWORD = _('Set password')
    PASSWORD_STRATEGY_CHOICES = (
        (0, EMAIL_SET_PASSWORD),
        (1, CUSTOM_PASSWORD)
    )
    password_strategy = serializers.ChoiceField(
        choices=PASSWORD_STRATEGY_CHOICES, required=False, initial=0,
        label=_('Password strategy'), write_only=True
    )
    mfa_level_display = serializers.ReadOnlyField(source='get_mfa_level_display', label=_('MFA level for display'))
    login_blocked = serializers.SerializerMethodField(label=_('Login blocked'))
    can_update = serializers.SerializerMethodField(label=_('Can update'))
    can_delete = serializers.SerializerMethodField(label=_('Can delete'))
    org_roles = serializers.ListField(label=_('Organization role name'), allow_null=True, required=False,
                                      child=serializers.ChoiceField(choices=ORG_ROLE.choices))
    key_prefix_block = "_LOGIN_BLOCK_{}"

    class Meta:
        model = User
        # mini 是指能识别对象的最小单元
        fields_mini = ['id', 'name', 'username']
        # small 指的是 不需要计算的直接能从一张表中获取到的数据
        fields_small = fields_mini + [
            'password', 'email', 'public_key', 'wechat', 'phone', 'mfa_level', 'mfa_enabled',
            'mfa_level_display', 'mfa_force_enabled', 'role_display', 'org_role_display',
            'total_role_display', 'comment', 'source', 'is_valid', 'is_expired',
            'is_active', 'created_by', 'is_first_login',
            'password_strategy', 'date_password_last_updated', 'date_expired',
            'avatar_url', 'source_display', 'date_joined', 'last_login'
        ]
        fields = fields_small + [
            'groups', 'role', 'groups_display', 'role_display',
            'can_update', 'can_delete', 'login_blocked', 'org_roles'
        ]

        read_only_fields = [
            'date_joined', 'last_login', 'created_by', 'is_first_login', 'source'
        ]

        extra_kwargs = {
            'password': {'write_only': True, 'required': False, 'allow_null': True, 'allow_blank': True},
            'public_key': {'write_only': True},
            'is_first_login': {'label': _('Is first login'), 'read_only': True},
            'is_valid': {'label': _('Is valid')},
            'is_expired': {'label': _('Is expired')},
            'avatar_url': {'label': _('Avatar url')},
            'created_by': {'read_only': True, 'allow_blank': True},
            'groups_display': {'label': _('Groups name')},
            'source_display': {'label': _('Source name')},
            'org_role_display': {'label': _('Organization role name')},
            'role_display': {'label': _('Super role name')},
            'total_role_display': {'label': _('Total role name')},
            'mfa_enabled': {'label': _('MFA enabled')},
            'mfa_force_enabled': {'label': _('MFA force enabled')},
        }

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.set_role_choices()

    def set_role_choices(self):
        role = self.fields.get('role')
        if not role:
            return
        choices = role._choices
        choices.pop(User.ROLE.APP, None)
        request = self.context.get('request')
        if request and hasattr(request, 'user') and not request.user.is_superuser:
            choices.pop(User.ROLE.ADMIN, None)
            choices.pop(User.ROLE.AUDITOR, None)
        role._choices = choices

    def validate_role(self, value):
        request = self.context.get('request')
        if not request.user.is_superuser and value != User.ROLE.USER:
            role_display = User.ROLE.USER.label
            msg = _("Role limit to {}".format(role_display))
            raise serializers.ValidationError(msg)
        return value

    def validate_password(self, password):
        from ..utils import check_password_rules
        password_strategy = self.initial_data.get('password_strategy')
        if password_strategy == '0':
            return
        if password_strategy is None and not password:
            return
        if not check_password_rules(password):
            msg = _('Password does not match security rules')
            raise serializers.ValidationError(msg)
        return password

    def validate_groups(self, groups):
        """
        审计员不能加入到组中
        """
        role = self.initial_data.get('role')
        if self.instance:
            role = role or self.instance.role
        if role == User.ROLE.AUDITOR:
            return []
        return groups

    @staticmethod
    def change_password_to_raw(attrs):
        password = attrs.pop('password', None)
        if password:
            attrs['password_raw'] = password
        return attrs

    @staticmethod
    def clean_auth_fields(attrs):
        for field in ('password', 'public_key'):
            value = attrs.get(field)
            if not value:
                attrs.pop(field, None)
        return attrs

    def validate(self, attrs):
        attrs = self.change_password_to_raw(attrs)
        attrs = self.clean_auth_fields(attrs)
        attrs.pop('password_strategy', None)
        return attrs

    def get_can_update(self, obj):
        return CanUpdateDeleteUser.has_update_object_permission(
            self.context['request'], self.context['view'], obj
        )

    def get_can_delete(self, obj):
        return CanUpdateDeleteUser.has_delete_object_permission(
            self.context['request'], self.context['view'], obj
        )

    def get_login_blocked(self, obj):
        key_block = self.key_prefix_block.format(obj.username)
        blocked = bool(cache.get(key_block))
        return blocked


class UserRetrieveSerializer(UserSerializer):
    login_confirm_settings = serializers.PrimaryKeyRelatedField(read_only=True,
                                                                source='login_confirm_setting.reviewers', many=True)

    class Meta(UserSerializer.Meta):
        fields = UserSerializer.Meta.fields + ['login_confirm_settings']


class UserPKUpdateSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'public_key']

    @staticmethod
    def validate_public_key(value):
        if not validate_ssh_public_key(value):
            raise serializers.ValidationError(_('Not a valid ssh public key'))
        return value


class ChangeUserPasswordSerializer(serializers.ModelSerializer):

    class Meta:
        model = User
        fields = ['password']


class ResetOTPSerializer(serializers.Serializer):
    msg = serializers.CharField(read_only=True)

    def create(self, validated_data):
        pass

    def update(self, instance, validated_data):
        pass


class UserRoleSerializer(serializers.Serializer):
    name = serializers.CharField(max_length=24)
    display = serializers.CharField(max_length=64)


class UserProfileSerializer(UserSerializer):
    admin_or_audit_orgs = UserOrgSerializer(many=True, read_only=True)
    user_all_orgs = UserOrgLabelSerializer(many=True, read_only=True)
    current_org_roles = serializers.ListField(read_only=True)
    public_key_comment = serializers.CharField(
        source='get_public_key_comment', required=False, read_only=True, max_length=128
    )
    public_key_hash_md5 = serializers.CharField(
        source='get_public_key_hash_md5', required=False, read_only=True, max_length=128
    )
    MFA_LEVEL_CHOICES = (
        (0, _('Disable')),
        (1, _('Enable')),
    )
    mfa_level = serializers.ChoiceField(choices=MFA_LEVEL_CHOICES, label=_('MFA'), required=False)
    guide_url = serializers.SerializerMethodField()

    class Meta(UserSerializer.Meta):
        fields = UserSerializer.Meta.fields + [
            'public_key_comment', 'public_key_hash_md5', 'admin_or_audit_orgs', 'current_org_roles',
            'guide_url', 'user_all_orgs'
        ]
        read_only_fields = [
            'date_joined', 'last_login', 'created_by', 'source'
        ]
        extra_kwargs = dict(UserSerializer.Meta.extra_kwargs)
        extra_kwargs.update({
            'name': {'read_only': True, 'max_length': 128},
            'username': {'read_only': True, 'max_length': 128},
            'email': {'read_only': True},
            'is_first_login': {'label': _('Is first login'), 'read_only': False},
            'source': {'read_only': True},
            'is_valid': {'read_only': True},
            'is_active': {'read_only': True},
            'groups': {'read_only': True},
            'roles': {'read_only': True},
            'password_strategy': {'read_only': True},
            'date_expired': {'read_only': True},
            'date_joined': {'read_only': True},
            'last_login': {'read_only': True},
            'role': {'read_only': True},
        })

        if 'password' in fields:
            fields.remove('password')
            extra_kwargs.pop('password', None)

    @staticmethod
    def get_guide_url(obj):
        return settings.USER_GUIDE_URL

    def validate_mfa_level(self, mfa_level):
        if self.instance and self.instance.mfa_force_enabled:
            return 2
        return mfa_level

    def validate_public_key(self, public_key):
        if self.instance and self.instance.can_update_ssh_key():
            if not validate_ssh_public_key(public_key):
                raise serializers.ValidationError(_('Not a valid ssh public key'))
            return public_key
        return None


class UserUpdatePasswordSerializer(serializers.ModelSerializer):
    old_password = serializers.CharField(required=True, max_length=128, write_only=True)
    new_password = serializers.CharField(required=True, max_length=128, write_only=True)
    new_password_again = serializers.CharField(required=True, max_length=128, write_only=True)

    class Meta:
        model = User
        fields = ['old_password', 'new_password', 'new_password_again']

    def validate_old_password(self, value):
        if not self.instance.check_password(value):
            msg = _('The old password is incorrect')
            raise serializers.ValidationError(msg)
        return value

    @staticmethod
    def validate_new_password(value):
        from ..utils import check_password_rules
        if not check_password_rules(value):
            msg = _('Password does not match security rules')
            raise serializers.ValidationError(msg)
        return value

    def validate_new_password_again(self, value):
        if value != self.initial_data.get('new_password', ''):
            msg = _('The newly set password is inconsistent')
            raise serializers.ValidationError(msg)
        return value

    def update(self, instance, validated_data):
        new_password = self.validated_data.get('new_password')
        instance.reset_password(new_password)
        return instance


class UserUpdatePublicKeySerializer(serializers.ModelSerializer):
    public_key_comment = serializers.CharField(
        source='get_public_key_comment', required=False, read_only=True, max_length=128
    )
    public_key_hash_md5 = serializers.CharField(
        source='get_public_key_hash_md5', required=False, read_only=True, max_length=128
    )

    class Meta:
        model = User
        fields = ['public_key_comment', 'public_key_hash_md5', 'public_key']
        extra_kwargs = {
            'public_key': {'required': True, 'write_only': True, 'max_length': 2048}
        }

    @staticmethod
    def validate_public_key(value):
        if not validate_ssh_public_key(value):
            raise serializers.ValidationError(_('Not a valid ssh public key'))
        return value

    def update(self, instance, validated_data):
        new_public_key = self.validated_data.get('public_key')
        instance.set_public_key(new_public_key)
        return instance


class MiniUserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'name', 'username']


class InviteSerializer(serializers.Serializer):
    user = serializers.PrimaryKeyRelatedField(
        queryset=User.objects.exclude(role=User.ROLE.APP)
    )
    role = serializers.ChoiceField(choices=ORG_ROLE.choices)