"""数据脱敏引擎 — 对查询结果进行动态脱敏处理。 支持脱敏类型: - full: 完全遮挡 (****) - partial: 部分遮挡 (138****8888, 李**) - hash: SHA256 哈希前16位 - truncate: 截断显示 (abc...) - regex: 正则表达式替换 使用方式: rules = await load_masking_rules(instance_id, database) masker = DataMasker(rules) masked_rows = [masker.mask_row(table_name, row) for row in rows] """ import hashlib import json import re from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, List, Optional, Callable class MaskType(str, Enum): """脱敏类型枚举。""" FULL = "full" PARTIAL = "partial" HASH = "hash" TRUNCATE = "truncate" REGEX = "regex" @dataclass class MaskingRule: """脱敏规则数据类，用于运行时脱敏引擎。 Attributes: instance_id: 数据源实例ID database: 数据库名称 table_name: 表名（支持通配符 * 表示所有表） column_name: 列名（支持通配符 * 表示所有列） mask_type: 脱敏类型 mask_config: 类型特定的配置字典 is_enabled: 是否启用 """ instance_id: int database: str table_name: str column_name: str mask_type: MaskType mask_config: Dict[str, Any] = field(default_factory=dict) is_enabled: bool = True @classmethod def from_db_model(cls, db_rule: Any) -> "MaskingRule": """从数据库模型创建 MaskingRule 实例。 Args: db_rule: SQLAlchemy MaskingRule 模型实例 Returns: MaskingRule 数据类实例 """ config: Dict[str, Any] = {} if db_rule.mask_config: try: config = json.loads(db_rule.mask_config) except json.JSONDecodeError: config = {} mask_type = MaskType(db_rule.mask_type) if db_rule.mask_type in { m.value for m in MaskType } else MaskType.FULL return cls( instance_id=db_rule.instance_id, database=db_rule.database, table_name=db_rule.table_name, column_name=db_rule.column_name, mask_type=mask_type, mask_config=config, is_enabled=db_rule.is_enabled, ) class DataMasker: """数据脱敏引擎，对查询结果进行动态脱敏。 根据配置的脱敏规则，对行数据进行后处理脱敏。 脱敏不改变原始查询语句，仅对返回结果进行处理。 Attributes: _rules: 脱敏规则列表 _rule_map: 规则查找映射 {(table, column): MaskingRule} _table_wildcards: 表级通配规则 {table: [rules]} _global_columns: 全局列规则 {column: rule} """ def __init__(self, rules: List[MaskingRule]) -> None: """初始化脱敏引擎。 Args: rules: 脱敏规则列表 """ self._rules: List[MaskingRule] = [r for r in rules if r.is_enabled] self._rule_map: Dict[tuple, MaskingRule] = {} self._table_wildcards: Dict[str, List[MaskingRule]] = {} self._global_columns: Dict[str, MaskingRule] = {} for rule in self._rules: if rule.table_name == "*" and rule.column_name == "*": # 全局通配规则 — 暂不支持 continue elif rule.table_name == "*": # 全局列规则：适用于所有表的指定列 self._global_columns[rule.column_name.lower()] = rule elif rule.column_name == "*": # 表级通配规则：适用于指定表的所有列 self._table_wildcards.setdefault(rule.table_name.lower(), []).append(rule) else: # 精确匹配规则 key = (rule.table_name.lower(), rule.column_name.lower()) self._rule_map[key] = rule # ===== 公共接口 ===== def mask(self, table: str, column: str, value: Any) -> Any: """对单个值进行脱敏。 Args: table: 表名 column: 列名 value: 原始值 Returns: 脱敏后的值。如果值为 None 或规则不存在，返回原值。 """ if value is None: return None rule = self._find_rule(table, column) if not rule: return value return self._apply_mask(rule, value) def mask_row(self, table: str, row: Dict[str, Any]) -> Dict[str, Any]: """对整行数据进行脱敏。 Args: table: 表名 row: 行数据字典 {column: value} Returns: 脱敏后的行数据字典 """ if not row: return row result: Dict[str, Any] = {} for col, val in row.items(): result[col] = self.mask(table, col, val) return result def mask_rows(self, table: str, rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """对多行数据进行脱敏。 Args: table: 表名 rows: 行数据列表 Returns: 脱敏后的行数据列表 """ if not rows: return rows return [self.mask_row(table, row) for row in rows] def has_rules_for_table(self, table: str) -> bool: """检查是否有规则适用于指定表。 Args: table: 表名 Returns: 如果存在适用的脱敏规则返回 True """ table_lower = table.lower() # 检查精确匹配 if any(k[0] == table_lower for k in self._rule_map): return True # 检查表级通配 if table_lower in self._table_wildcards: return True # 检查全局列规则 if self._global_columns: return True return False # ===== 规则查找 ===== def _find_rule(self, table: str, column: str) -> Optional[MaskingRule]: """查找适用于指定表和列的脱敏规则。 优先级: 精确匹配 > 表级通配 > 全局列规则 Args: table: 表名 column: 列名 Returns: 匹配的脱敏规则，未找到返回 None """ table_lower = table.lower() col_lower = column.lower() # 1. 精确匹配 key = (table_lower, col_lower) if key in self._rule_map: return self._rule_map[key] # 2. 表级通配 (table.*) if table_lower in self._table_wildcards: for rule in self._table_wildcards[table_lower]: if rule.column_name == "*": return rule # 3. 全局列规则 (*.column) if col_lower in self._global_columns: return self._global_columns[col_lower] return None # ===== 脱敏实现 ===== def _apply_mask(self, rule: MaskingRule, value: Any) -> Any: """根据规则类型应用对应的脱敏算法。 Args: rule: 脱敏规则 value: 原始值 Returns: 脱敏后的值 Raises: ValueError: 不支持的脱敏类型 """ # 非字符串值转换为字符串处理 str_value = str(value) if value is not None else "" mask_methods: Dict[MaskType, Callable[[str, Dict[str, Any]], str]] = { MaskType.FULL: self._mask_full, MaskType.PARTIAL: self._mask_partial, MaskType.HASH: self._mask_hash, MaskType.TRUNCATE: self._mask_truncate, MaskType.REGEX: self._mask_regex, } method = mask_methods.get(rule.mask_type) if not method: raise ValueError(f"不支持的脱敏类型: {rule.mask_type}") return method(str_value, rule.mask_config) @staticmethod def _mask_full(value: str, config: Dict[str, Any]) -> str: """完全遮挡 — 将值替换为固定数量的星号。 Args: value: 原始值 config: 配置项 {length: 遮挡长度(默认4), char: 遮挡字符(默认*)} Returns: 完全遮挡后的字符串 """ length = config.get("length", 4) char = config.get("char", "*") return char * length @staticmethod def _mask_partial(value: str, config: Dict[str, Any]) -> str: """部分遮挡 — 保留前后若干字符，中间用星号替换。 Args: value: 原始值 config: 配置项 { keep_prefix: 保留前缀字符数(默认2), keep_suffix: 保留后缀字符数(默认2), mask_char: 遮挡字符(默认*) } Returns: 部分遮挡后的字符串 Examples: >>> DataMasker._mask_partial("13812345678", {"keep_prefix": 3, "keep_suffix": 4}) '138****5678' >>> DataMasker._mask_partial("张三丰", {"keep_prefix": 1, "keep_suffix": 0}) '张**' """ keep_prefix = config.get("keep_prefix", 2) keep_suffix = config.get("keep_suffix", 2) mask_char = config.get("mask_char", "*") if len(value) <= keep_prefix + keep_suffix: # 值太短，全部遮挡 return mask_char * len(value) prefix = value[:keep_prefix] suffix = value[-keep_suffix:] if keep_suffix > 0 else "" mask_len = len(value) - keep_prefix - keep_suffix return prefix + mask_char * mask_len + suffix @staticmethod def _mask_hash(value: str, config: Dict[str, Any]) -> str: """SHA256 哈希 — 取哈希值的前N位。 Args: value: 原始值 config: 配置项 {length: 截取长度(默认16)} Returns: 哈希前缀字符串 """ length = config.get("length", 16) hash_val = hashlib.sha256(value.encode("utf-8")).hexdigest() return hash_val[:length] @staticmethod def _mask_truncate(value: str, config: Dict[str, Any]) -> str: """截断显示 — 只保留前N个字符，后面加省略号。 Args: value: 原始值 config: 配置项 {length: 保留长度(默认3), ellipsis: 省略号(默认...) Returns: 截断后的字符串 """ length = config.get("length", 3) ellipsis = config.get("ellipsis", "...") if len(value) <= length: return value return value[:length] + ellipsis @staticmethod def _mask_regex(value: str, config: Dict[str, Any]) -> str: """正则表达式替换 — 使用正则匹配并替换。 Args: value: 原始值 config: 配置项 { pattern: 正则表达式, replacement: 替换字符串(默认 ***) } Returns: 替换后的字符串 Raises: ValueError: 正则表达式无效 """ pattern = config.get("pattern", ".*") replacement = config.get("replacement", "***") try: return re.sub(pattern, replacement, value) except re.error as e: raise ValueError(f"无效的正则表达式: {pattern}, 错误: {e}") # ===== 便捷函数 ===== async def load_masking_rules( db: Any, instance_id: int, database: str ) -> List[MaskingRule]: """从数据库加载指定实例和数据库的脱敏规则。 Args: db: 异步数据库会话 (AsyncSession) instance_id: 数据源实例ID database: 数据库名称 Returns: MaskingRule 列表 """ from sqlalchemy import select from app.models.masking_rule import MaskingRule as MaskingRuleModel stmt = ( select(MaskingRuleModel) .where( MaskingRuleModel.instance_id == instance_id, MaskingRuleModel.database == database, MaskingRuleModel.is_enabled == True, ) ) result = await db.execute(stmt) db_rules = result.scalars().all() return [MaskingRule.from_db_model(r) for r in db_rules]