当前位置：首页 > news >正文

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

news 2026/7/11 20:05:49

数据脱敏的智能化：用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_schema=schema_info) response = self.llm.chat( system="你是数据安全专家，请严格按照 JSON 格式返回。", user_message=prompt, temperature=0.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_message=verify_prompt, temperature=0.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" f"UPDATE `{table}` SET `{column}` = '***';" ) elif mask_type == 'HASH': # 哈希脱敏：SHA256 sql_statements.append( f"-- {table}.{column}: 哈希脱敏\n" f"UPDATE `{table}` SET `{column}` = " f"SHA2(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" f"UPDATE `{table}` SET `{column}` = " f"{mask_func.format(col=column)};" ) return "\n\n".join(sql_statements)