feat: 添加脱敏Mixin
This commit is contained in:
@@ -1,11 +1,15 @@
|
||||
from ai.models import AIApiKey
|
||||
from utils.serializers import CustomModelSerializer
|
||||
from utils.serializers import CustomModelSerializer, DesensitizationMixin
|
||||
from utils.custom_model_viewSet import CustomModelViewSet
|
||||
|
||||
class AIApiKeySerializer(CustomModelSerializer):
|
||||
|
||||
class AIApiKeySerializer(DesensitizationMixin, CustomModelSerializer):
|
||||
"""
|
||||
AI API 密钥 序列化器
|
||||
"""
|
||||
# 指定需要脱敏的字段
|
||||
desensitize_fields = ['api_key']
|
||||
|
||||
class Meta:
|
||||
model = AIApiKey
|
||||
fields = '__all__'
|
||||
|
||||
@@ -19,7 +19,7 @@ class AuditUserFieldsMixin:
|
||||
creator_field_id = 'creator'
|
||||
|
||||
def set_audit_user_fields(self, validated_data, is_create=True):
|
||||
username = self.get_request_username() if hasattr(self, 'get_request_username') else None
|
||||
username = self.get_request_user_name() if hasattr(self, 'get_request_user_name') else None
|
||||
if getattr(self, 'request', None):
|
||||
if self.modifier_field_id in self.fields:
|
||||
validated_data[self.modifier_field_id] = username
|
||||
@@ -27,6 +27,128 @@ class AuditUserFieldsMixin:
|
||||
validated_data[self.creator_field_id] = username
|
||||
|
||||
|
||||
class DesensitizationMixin:
|
||||
"""
|
||||
用于敏感字段脱敏的通用 Mixin
|
||||
使用方法:
|
||||
1. 在序列化器中继承此 Mixin
|
||||
2. 设置 desensitize_fields 属性,指定需要脱敏的字段
|
||||
3. 可选:设置 desensitize_prefix_length 和 desensitize_suffix_length 来自定义脱敏格式
|
||||
"""
|
||||
|
||||
# 需要脱敏的字段列表,格式:['field_name', 'related_field.field_name']
|
||||
desensitize_fields = []
|
||||
|
||||
# 脱敏时保留的前缀长度,默认4
|
||||
desensitize_prefix_length = 4
|
||||
|
||||
# 脱敏时保留的后缀长度,默认4
|
||||
desensitize_suffix_length = 4
|
||||
|
||||
# 脱敏阈值,字段长度小于等于此值时全部用*替换,默认8
|
||||
desensitize_threshold = 8
|
||||
|
||||
# 脱敏字符,默认使用*
|
||||
desensitize_char = '*'
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
# 为每个脱敏字段创建脱敏方法
|
||||
for field_name in self.desensitize_fields:
|
||||
if '.' in field_name:
|
||||
# 处理关联字段,如 'key.api_key'
|
||||
method_name = f'get_{field_name.replace(".", "_")}_desensitized'
|
||||
setattr(self, method_name, self._create_desensitize_method(field_name))
|
||||
else:
|
||||
# 处理直接字段
|
||||
method_name = f'get_{field_name}_desensitized'
|
||||
setattr(self, method_name, self._create_desensitize_method(field_name))
|
||||
|
||||
def _create_desensitize_method(self, field_name):
|
||||
"""创建脱敏方法的闭包"""
|
||||
def desensitize_method(obj):
|
||||
return self._desensitize_field(obj, field_name)
|
||||
return desensitize_method
|
||||
|
||||
def _desensitize_field(self, obj, field_name):
|
||||
"""脱敏指定字段"""
|
||||
if '.' in field_name:
|
||||
# 处理关联字段,如 'key.api_key'
|
||||
parts = field_name.split('.')
|
||||
value = obj
|
||||
for part in parts:
|
||||
if hasattr(value, part):
|
||||
value = getattr(value, part)
|
||||
else:
|
||||
return None
|
||||
else:
|
||||
# 处理直接字段
|
||||
if not hasattr(obj, field_name):
|
||||
return None
|
||||
value = getattr(obj, field_name)
|
||||
|
||||
# 如果值为空,直接返回
|
||||
if not value:
|
||||
return value
|
||||
|
||||
# 检查用户权限
|
||||
if self._can_view_full_value():
|
||||
return value
|
||||
|
||||
# 执行脱敏
|
||||
return self._apply_desensitization(str(value))
|
||||
|
||||
def _can_view_full_value(self):
|
||||
"""检查当前用户是否可以查看完整值"""
|
||||
# request = self.context.get('request')
|
||||
# if not request or not request.user:
|
||||
# return False
|
||||
#
|
||||
# # 超级用户或管理员可以查看完整值
|
||||
# return request.user.is_superuser or request.user.is_staff
|
||||
return False # 默认不允许查看完整值,需根据实际权限逻辑调整
|
||||
|
||||
def _apply_desensitization(self, value):
|
||||
"""应用脱敏规则"""
|
||||
if len(value) <= self.desensitize_threshold:
|
||||
# 如果长度小于等于阈值,则全部用脱敏字符替换
|
||||
return self.desensitize_char * len(value)
|
||||
else:
|
||||
# 显示前缀和后缀,中间用脱敏字符替换
|
||||
prefix = value[:self.desensitize_prefix_length]
|
||||
suffix = value[-self.desensitize_suffix_length:]
|
||||
middle_length = len(value) - self.desensitize_prefix_length - self.desensitize_suffix_length
|
||||
middle = self.desensitize_char * middle_length
|
||||
return prefix + middle + suffix
|
||||
|
||||
def get_fields(self):
|
||||
"""重写 get_fields 方法,为脱敏字段添加脱敏版本"""
|
||||
fields = super().get_fields()
|
||||
|
||||
is_list = getattr(self.root, 'many', False)
|
||||
|
||||
for field_name in self.desensitize_fields:
|
||||
if '.' in field_name:
|
||||
# 处理关联字段,如 'key.api_key'
|
||||
method_name = f'get_{field_name.replace(".", "_")}_desensitized'
|
||||
field_key = f"{field_name.replace('.', '_')}_desensitized"
|
||||
else:
|
||||
# 处理直接字段
|
||||
method_name = f'get_{field_name}_desensitized'
|
||||
field_key = f"{field_name}_desensitized"
|
||||
|
||||
# 创建脱敏字段的 SerializerMethodField
|
||||
fields[field_key] = serializers.SerializerMethodField(method_name=method_name)
|
||||
|
||||
# 保持原字段不变,确保创建/更新功能正常
|
||||
# 原字段仍然可以接收输入数据
|
||||
if is_list:
|
||||
# 如果是列表序列化,移除原始字段
|
||||
fields.pop(field_name, None)
|
||||
|
||||
return fields
|
||||
|
||||
|
||||
class CustomModelSerializer(AuditUserFieldsMixin, ModelSerializer):
|
||||
"""
|
||||
增强DRF的ModelSerializer,可自动更新模型的审计字段记录
|
||||
|
||||
@@ -130,7 +130,7 @@ export function useColumns(
|
||||
formatter: dictFormatter('ai_platform'),
|
||||
},
|
||||
{
|
||||
field: 'api_key',
|
||||
field: 'api_key_desensitized',
|
||||
title: '密钥',
|
||||
},
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user