jumpserver/apps/settings/api/settings.py

# -*- coding: utf-8 -*-
#
import re

from django.conf import settings
from django.core.cache import cache
from django.http import HttpResponse
from django.views.static import serve
from rest_framework import generics
from rest_framework import status
from rest_framework.permissions import AllowAny
from rest_framework.views import APIView

from common.utils import get_logger
from jumpserver.conf import Config
from rbac.permissions import RBACPermission
from .. import serializers
from ..models import Setting
from ..signals import category_setting_updated
from ..utils import get_interface_setting_or_default

logger = get_logger(__file__)


class SettingsApi(generics.RetrieveUpdateAPIView):
    permission_classes = (RBACPermission,)

    serializer_class_mapper = {
        'all': serializers.SettingsSerializer,
        'basic': serializers.BasicSettingSerializer,
        'terminal': serializers.TerminalSettingSerializer,
        'security': serializers.SecuritySettingSerializer,
        'security_auth': serializers.SecurityAuthSerializer,
        'security_basic': serializers.SecurityBasicSerializer,
        'security_session': serializers.SecuritySessionSerializer,
        'security_password': serializers.SecurityPasswordRuleSerializer,
        'security_login_limit': serializers.SecurityLoginLimitSerializer,
        'ldap': serializers.LDAPSettingSerializer,
        'ldap_ha': serializers.LDAPHASettingSerializer,
        'email': serializers.EmailSettingSerializer,
        'email_content': serializers.EmailContentSettingSerializer,
        'wecom': serializers.WeComSettingSerializer,
        'dingtalk': serializers.DingTalkSettingSerializer,
        'feishu': serializers.FeiShuSettingSerializer,
        'lark': serializers.LarkSettingSerializer,
        'slack': serializers.SlackSettingSerializer,
        'auth': serializers.AuthSettingSerializer,
        'oidc': serializers.OIDCSettingSerializer,
        'keycloak': serializers.KeycloakSettingSerializer,
        'radius': serializers.RadiusSettingSerializer,
        'cas': serializers.CASSettingSerializer,
        'saml2': serializers.SAML2SettingSerializer,
        'oauth2': serializers.OAuth2SettingSerializer,
        'passkey': serializers.PasskeySettingSerializer,
        'clean': serializers.CleaningSerializer,
        'other': serializers.OtherSettingSerializer,
        'sms': serializers.SMSSettingSerializer,
        'alibaba': serializers.AlibabaSMSSettingSerializer,
        'tencent': serializers.TencentSMSSettingSerializer,
        'huawei': serializers.HuaweiSMSSettingSerializer,
        'cmpp2': serializers.CMPP2SMSSettingSerializer,
        'custom': serializers.CustomSMSSettingSerializer,
        'vault': serializers.VaultSettingSerializer,
        'azure_kv': serializers.AzureKVSerializer,
        'hcp': serializers.HashicorpKVSerializer,
        'chat': serializers.ChatAISettingSerializer,
        'announcement': serializers.AnnouncementSettingSerializer,
        'ticket': serializers.TicketSettingSerializer,
        'ops': serializers.OpsSettingSerializer,
        'virtualapp': serializers.VirtualAppSerializer,
        'tool': serializers.ToolSerializer,
    }

    rbac_category_permissions = {
        'basic': 'settings.view_setting',
        'tool': 'settings.view_setting',
        'terminal': 'settings.change_terminal',
        'ops': 'settings.change_ops',
        'ticket': 'settings.change_ticket',
        'virtualapp': 'settings.change_virtualapp',
        'announcement': 'settings.change_announcement',
        'security': 'settings.change_security',
        'security_basic': 'settings.change_security',
        'security_auth': 'settings.change_security',
        'security_session': 'settings.change_security',
        'security_password': 'settings.change_security',
        'security_login_limit': 'settings.change_security',
        'ldap': 'settings.change_auth',
        'cas': 'settings.change_auth',
        'oidc': 'settings.change_auth',
        'saml2': 'settings.change_auth',
        'oauth2': 'settings.change_auth',
        'wecom': 'settings.change_auth',
        'dingtalk': 'settings.change_auth',
        'feishu': 'settings.change_auth',
        'lark': 'settings.change_auth',
        'slack': 'settings.change_auth',
        'auth': 'settings.change_auth',
        'passkey': 'settings.change_auth',
        'keycloak': 'settings.change_auth',
        'radius': 'settings.change_auth',
        'sso': 'settings.change_auth',
        'clean': 'settings.change_clean',
        'other': 'settings.change_other',
        'chat': 'settings.change_chatai',
        'email': 'settings.change_email',
        'email_content': 'settings.change_email',
        'sms': 'settings.change_sms',
        'alibaba': 'settings.change_sms',
        'tencent': 'settings.change_sms',
        'huawei': 'settings.change_sms',
        'cmpp2': 'settings.change_sms',
        'vault': 'settings.change_vault',
    }

    def get_queryset(self):
        return Setting.objects.all()

    def check_permissions(self, request):
        category = request.query_params.get('category', 'basic')
        perm_required = self.rbac_category_permissions.get(category)
        has = self.request.user.has_perm(perm_required)

        if not has:
            self.permission_denied(request)

    def get_serializer_class(self):
        category = self.request.query_params.get('category', 'basic')
        default = serializers.BasicSettingSerializer
        cls = self.serializer_class_mapper.get(category, default)
        return cls

    def get_fields(self):
        serializer = self.get_serializer_class()()
        fields = serializer.get_fields()
        return fields

    def get_object(self):
        items = self.get_fields().keys()
        obj = {}
        for item in items:
            if hasattr(settings, item):
                obj[item] = getattr(settings, item)
            else:
                obj[item] = Config.defaults[item]
        return obj

    def parse_serializer_data(self, serializer):
        data = []
        fields = self.get_fields()
        encrypted_items = [name for name, field in fields.items() if field.write_only]
        category = self.request.query_params.get('category', '')
        for name, value in serializer.validated_data.items():
            encrypted = name in encrypted_items
            if encrypted and value in ['', None]:
                continue
            data.append({
                'name': name, 'value': value,
                'encrypted': encrypted, 'category': category
            })
        return data

    def send_signal(self, serializer):
        category = self.request.query_params.get('category', '')
        category_setting_updated.send(sender=self.__class__, category=category, serializer=serializer)

    def perform_update(self, serializer):
        post_data_names = list(self.request.data.keys())
        settings_items = self.parse_serializer_data(serializer)
        serializer_data = getattr(serializer, 'data', {})

        for item in settings_items:
            if item['name'] not in post_data_names:
                continue
            changed, setting = Setting.update_or_create(**item)
            if not changed:
                continue
            serializer_data[setting.name] = setting.cleaned_value

        setattr(serializer, '_data', serializer_data)
        if hasattr(serializer, 'post_save'):
            serializer.post_save()
        self.send_signal(serializer)
        if self.request.query_params.get('category') == 'ldap':
            self.clean_ldap_user_dn_cache()

    @staticmethod
    def clean_ldap_user_dn_cache():
        del_count = cache.delete_pattern('django_auth_ldap.user_dn.*')
        logger.debug(f'clear LDAP user_dn_cache count={del_count}')


class SettingsLogoApi(APIView):
    permission_classes = (AllowAny,)

    def get(self, request, *args, **kwargs):
        size = request.GET.get('size', 'small')
        interface_data = get_interface_setting_or_default()
        if size == 'small':
            logo_path = interface_data['logo_logout']
        else:
            logo_path = interface_data['logo_index']

        if logo_path.startswith('/media/'):
            logo_path = logo_path.replace('/media/', '')
            document_root = settings.MEDIA_ROOT
        elif logo_path.startswith('/static/'):
            logo_path = logo_path.replace('/static/', '/')
            document_root = settings.STATIC_ROOT
        else:
            return HttpResponse(status=status.HTTP_404_NOT_FOUND)
        return serve(request, logo_path, document_root=document_root)