本文目录导读：
我来详细介绍如何实现脱敏规则配置脚本,包括几种常见的实现方式。
基础架构设计
from enum import Enum
from typing import Any, Dict, List, Optional
import re
import json
import yaml
class MaskingType(Enum):
"""脱敏类型枚举"""
PHONE = "phone"
EMAIL = "email"
ID_CARD = "id_card"
NAME = "name"
BANK_CARD = "bank_card"
ADDRESS = "address"
CUSTOM = "custom"
class MaskingRule:
"""脱敏规则基类"""
def __init__(self, rule_config: Dict):
self.name = rule_config.get('name', '')
self.type = MaskingType(rule_config.get('type', 'custom'))
self.pattern = rule_config.get('pattern', '')
self.replacement = rule_config.get('replacement', '')
self.mask_char = rule_config.get('mask_char', '*')
self.enabled = rule_config.get('enabled', True)
YAML配置脚本实现
配置文件示例 (masking_rules.yaml)
# 脱敏规则配置
version: '1.0'
rules:
# 手机号脱敏
- name: "phone_masking"
type: "phone"
pattern: "(\\d{3})\\d{4}(\\d{4})"
replacement: "\\1****\\2"
mask_char: "*"
enabled: true
description: "手机号中间4位脱敏"
# 邮箱脱敏
- name: "email_masking"
type: "email"
pattern: "(\\w{3})\\w+(@\\w+\\.\\w+)"
replacement: "\\1***\\2"
enabled: true
description: "邮箱用户名前3位保留"
# 身份证号脱敏
- name: "id_card_masking"
type: "id_card"
pattern: "(\\d{6})\\d{8}(\\d{4})"
replacement: "\\1********\\2"
enabled: true
description: "身份证号中间8位脱敏"
配置加载器
class MaskingConfigLoader:
"""配置加载器"""
@staticmethod
def load_yaml(config_path: str) -> Dict:
"""加载YAML配置"""
with open(config_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
@staticmethod
def load_json(config_path: str) -> Dict:
"""加载JSON配置"""
with open(config_path, 'r', encoding='utf-8') as f:
return json.load(f)
@staticmethod
def validate_config(config: Dict) -> bool:
"""验证配置有效性"""
if 'rules' not in config:
return False
for rule in config['rules']:
required_fields = ['name', 'type', 'pattern', 'replacement']
if not all(field in rule for field in required_fields):
return False
return True
规则引擎实现
class MaskingEngine:
"""脱敏引擎"""
def __init__(self, config_path: str = None):
self.rules: List[MaskingRule] = []
self.loader = MaskingConfigLoader()
if config_path:
self.load_rules(config_path)
def load_rules(self, config_path: str):
"""加载脱敏规则"""
config = self.loader.load_yaml(config_path)
if not self.loader.validate_config(config):
raise ValueError("Invalid masking configuration")
for rule_config in config['rules']:
if rule_config.get('enabled', True):
self.rules.append(MaskingRule(rule_config))
def add_rule(self, rule: MaskingRule):
"""动态添加规则"""
self.rules.append(rule)
def remove_rule(self, rule_name: str):
"""删除规则"""
self.rules = [r for r in self.rules if r.name != rule_name]
def mask_data(self, data: str, rule_type: Optional[MaskingType] = None) -> str:
"""执行脱敏"""
if not data:
return data
result = data
for rule in self.rules:
if rule_type and rule.type != rule_type:
continue
if rule.enabled:
result = re.sub(rule.pattern, rule.replacement, result)
return result
高级配置脚本
支持条件判断的配置
# advanced_masking_rules.yaml
version: '2.0'
global_settings:
default_mask_char: '*'
log_masking: true
rules:
- name: "dynamic_phone_masking"
type: "phone"
conditions:
- field: "country_code"
operator: "equals"
value: "86"
action:
pattern: "(\\d{3})\\d{4}(\\d{4})"
replacement: "\\1****\\2"
- field: "country_code"
operator: "equals"
value: "1"
action:
pattern: "(\\d{3})\\d{3}(\\d{4})"
replacement: "\\1-***-\\2"
- name: "sensitive_field_masking"
type: "custom"
fields: ["password", "token", "secret"]
pattern: ".*"
replacement: "********"
conditions:
- field: "security_level"
operator: "gte"
value: 3
动态规则生成器
class RuleGenerator:
"""动态规则生成器"""
@staticmethod
def generate_phone_rule(country_code: str = "86") -> Dict:
"""生成手机号脱敏规则"""
if country_code == "86":
return {
"name": "phone_mask_cn",
"type": "phone",
"pattern": r"(\d{3})\d{4}(\d{4})",
"replacement": r"\1****\2",
"enabled": True
}
elif country_code == "1":
return {
"name": "phone_mask_us",
"type": "phone",
"pattern": r"(\d{3})(\d{3})(\d{4})",
"replacement": r"\1-***-\3",
"enabled": True
}
@staticmethod
def generate_custom_rule(field_name: str,
start_keep: int = 0,
end_keep: int = 0) -> Dict:
"""生成自定义脱敏规则"""
if start_keep == 0 and end_keep == 0:
return {
"name": f"full_mask_{field_name}",
"type": "custom",
"pattern": r".*",
"replacement": "******",
"enabled": True
}
pattern = f"(.{{{start_keep}}}).*(.{{{end_keep}}})"
replacement = f"\\1{'*' * 6}\\2"
return {
"name": f"partial_mask_{field_name}",
"type": "custom",
"pattern": pattern,
"replacement": replacement,
"enabled": True
}
完整使用示例
def main():
# 初始化脱敏引擎
engine = MaskingEngine("masking_rules.yaml")
# 测试数据
test_data = {
"phone": "13812345678",
"email": "testuser@example.com",
"id_card": "110101199001011234",
"name": "张三"
}
# 执行脱敏
for field, value in test_data.items():
masked_value = engine.mask_data(value)
print(f"{field}: {value} -> {masked_value}")
# 动态添加规则
custom_rule = MaskingRule({
"name": "name_masking",
"type": "name",
"pattern": r"(.)(.*)",
"replacement": r"\1**",
"enabled": True
})
engine.add_rule(custom_rule)
# 批量脱敏
sensitive_data = {
"user1": {"phone": "13912345678", "email": "user1@test.com"},
"user2": {"phone": "15887654321", "email": "user2@test.com"}
}
for user, data in sensitive_data.items():
for field, value in data.items():
sensitive_data[user][field] = engine.mask_data(value)
return sensitive_data
if __name__ == "__main__":
result = main()
print(json.dumps(result, ensure_ascii=False, indent=2))
Web API集成
from flask import Flask, request, jsonify
from flask_restful import Api, Resource
app = Flask(__name__)
api = Api(app)
engine = MaskingEngine("masking_rules.yaml")
class MaskingAPI(Resource):
def post(self):
"""脱敏API"""
data = request.get_json()
text = data.get('text', '')
rule_type = data.get('rule_type', None)
if rule_type:
rule_type = MaskingType(rule_type)
masked_text = engine.mask_data(text, rule_type)
return jsonify({
"original": text,
"masked": masked_text,
"rule_type": rule_type.value if rule_type else "all"
})
def put(self):
"""更新规则"""
rule_config = request.get_json()
new_rule = MaskingRule(rule_config)
engine.add_rule(new_rule)
return jsonify({"message": "Rule added successfully"})
api.add_resource(MaskingAPI, '/api/mask')
if __name__ == '__main__':
app.run(debug=True)
最佳实践建议
- 规则分级：将规则分为基础规则和业务规则
- 性能优化：使用规则缓存和预编译正则
- 审计日志：记录脱敏操作日志
- 动态更新：支持热加载规则配置
- 灰度发布：支持规则的分批次生效
这个实现方案灵活可扩展,可以根据实际需求调整配置格式和规则引擎的实现。
