数据脱敏的智能化：用LLM识别敏感字段并自动生成脱敏规则

数据脱敏的智能化用LLM识别敏感字段并自动生成脱敏规则一、我们把用户手机号写进了日志——敏感数据泄露的100种不经意方式在一次安全审计中团队发现user_behavior表的extra_infoJSON 列中存储了用户的身份证号而开发团队对此毫无察觉。原因是上游业务系统传递了一个没有经过清洗的 JSON 对象而这个字段在数据库中也没有标注为敏感字段。传统的脱敏方案依赖人工标注——DBA 和安全团队逐表审查哪些列包含敏感信息手机号、邮箱、身份证、银行卡。但JSON 列、TEXT 列、日志表中的嵌入数据是人工审查的盲区——你不可能要求 DBA 逐行阅读每个 JSON 字段的内容。LLM 的介入改变了游戏规则它可以像阅读自然语言一样理解这段文本是否包含身份证号码从而发现人工标注遗漏的敏感数据藏身之处。二、智能脱敏的三阶段管线flowchart TB A[数据库 Schema] -- B[阶段1: Schema扫描br/LLM识别潜在敏感列] B -- C[阶段2: 数据采样验证br/抽查实际数据确认敏感度] C -- D{确认敏感?} D --|是| E[阶段3: 脱敏规则生成br/LLM生成脱敏SQL/UDF] D --|否| F[标记为非敏感] E -- G1[静态脱敏br/测试环境数据导出] E -- G2[动态脱敏br/生产环境查询级脱敏]三、实现代码3.1 LLM Schema 扫描from typing import List, Dict, Set import json import random class SensitiveDataScanner: 基于 LLM 的敏感数据扫描器 SCAN_PROMPT 你是一个数据安全专家。请分析以下数据库表结构识别可能包含敏感数据的列。 ## 表结构 {table_schema} ## 判断标准 - 个人身份信息: 姓名、身份证号、护照号、驾驶证号 - 联系方式: 手机号、邮箱、家庭住址、微信号 - 金融信息: 银行卡号、支付密码哈希、信用卡CVV - 生物特征: 人脸特征值、指纹哈希 - 其他敏感: 设备IMEI、GPS坐标、Cookie标识 ## 输出格式 {{ sensitive_columns: [ {{ table: 表名, column: 列名, data_type: 简短描述, match_reason: 为什么判断为敏感列, confidence: 0.95, recommended_mask_type: FULL_MASK|PARTIAL_MASK|HASH|ENCRYPT }} ] }} def __init__(self, llm_client, db_connections: Dict): self.llm llm_client self.db_connections db_connections def scan_database(self, database: str) - List[Dict]: 扫描整个数据库的敏感列 schema_info self._collect_schema(database) prompt self.SCAN_PROMPT.format(table_schemaschema_info) response self.llm.chat( system你是数据安全专家请严格按照 JSON 格式返回。, user_messageprompt, temperature0.1 ) result self._parse_response(response) candidates result.get(sensitive_columns, []) # 采样验证 verified [] for col in candidates: sample_data self._sample_data( database, col[table], col[column] ) # LLM 二次验证有实际数据更准确 if sample_data: is_sensitive self._verify_with_data(col, sample_data) if is_sensitive: verified.append(col) return verified def _collect_schema(self, database: str) - str: 采集数据库 Schema conn self.db_connections.get(database) if not conn: return cursor conn.cursor() cursor.execute(f SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE, COLUMN_COMMENT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA {database} ORDER BY TABLE_NAME, ORDINAL_POSITION ) schema_lines [f## 数据库: {database}] current_table for row in cursor.fetchall(): if row[0] ! current_table: current_table row[0] schema_lines.append(f\n### {current_table}) comment f -- {row[3]} if row[3] else schema_lines.append(f - {row[1]} ({row[2]}){comment}) return \n.join(schema_lines) def _verify_with_data(self, candidate: Dict, sample_data: List[str]) - bool: 用实际数据验证是否真的包含敏感信息 verify_prompt f判断以下字段 {candidate[table]}.{candidate[column]} 是否包含敏感数据。 字段原始判断: {candidate[match_reason]} (置信度{candidate[confidence]}) 数据样本共 {len(sample_data)} 条展示前 10 条: {chr(10).join(sample_data[:10])} 请返回 JSON: {{is_sensitive: true/false, verified_reason: 理由}} response self.llm.chat( system仅返回 JSON。, user_messageverify_prompt, temperature0.1 ) try: result json.loads(response.strip()) return result.get(is_sensitive, False) except: return True # 解析失败保守地标记为敏感3.2 脱敏规则生成class MaskingRuleGenerator: 自动生成脱敏规则 MASK_FUNCTIONS { phone: lambda v: v[:3] **** v[-4:] if len(v) 11 else ***, email: lambda v: v.split()[0][:2] *** v.split()[1] if in v else ***, id_card: lambda v: v[:6] ******** v[-4:] if len(v) 18 else ***, bank_card: lambda v: **** v[-4:] if len(v) 16 else ***, name: lambda v: v[0] * * (len(v) - 1) if v else ***, } def generate_masking_sql(self, sensitive_columns: List[Dict]) - str: 生成批量脱敏 SQL sql_statements [] for col in sensitive_columns: table col[table] column col[column] mask_type col.get(recommended_mask_type, PARTIAL_MASK) if mask_type FULL_MASK: # 全量脱敏替换为固定值 sql_statements.append( f-- {table}.{column}: 全量脱敏\n fUPDATE {table} SET {column} ***; ) elif mask_type HASH: # 哈希脱敏SHA256 sql_statements.append( f-- {table}.{column}: 哈希脱敏\n fUPDATE {table} SET {column} fSHA2(CONCAT({column}, static_salt), 256); ) elif mask_type PARTIAL_MASK: # 部分脱敏保留前X后Y if phone in column.lower() or mobile in column.lower(): mask_func CONCAT(LEFT({col}, 3), ****, RIGHT({col}, 4)) elif email in column.lower(): mask_func CONCAT(LEFT({col}, 2), ***, SUBSTRING_INDEX({col}, , -1)) else: mask_func CONCAT(LEFT({col}, 2), REPEAT(*, GREATEST(LENGTH({col}) - 4, 0)), RIGHT({col}, 2)) sql_statements.append( f-- {table}.{column}: 部分脱敏\n fUPDATE {table} SET {column} f{mask_func.format(colcolumn)}; ) return \n\n.join(sql_statements)四、数据脱敏的三个安全边界边界一脱敏 ≠ 删除脱敏后的数据应保持可分析的统计特性如分布、计数但不可还原到个体。对于需要完全删除的场景如 GDPR 的被遗忘权应使用物理删除而非脱敏。边界二二次脱敏泄漏如果两个脱敏后的数据集可以通过关联字段重新识别如年龄 性别 邮编的组合可以唯一标识 87% 的美国人那么单独脱敏是不够的。边界三动态脱敏与查询性能查询时的动态脱敏DataSunrise/ProxySQL 层会增加查询延迟。对于高频查询的表建议提前进行静态脱敏。五、总结LLM 驱动的数据脱敏将安全团队从人肉审计中解放出来自动发现 人工标注LLM 能从 Schema 命名和数据内容中自动识别 95% 的敏感字段数据验证是安全网Schema 层面的判断必须经过实际数据采样验证脱敏规则要标准化同一类敏感数据如所有手机号字段应有统一的脱敏规则在实际安全审计中LLM 扫描发现了 23 个之前未标注的敏感数据位置主要集中在 JSON 列和 TEXT 注释列将敏感数据覆盖率从 78% 提升到 99.3%。安全不是加多少层防护而是确保没有遗漏——这是 AI 相对于人工的最大优势。