Function Calling 工具参数的敏感信息脱敏：手机号、身份证与密钥的网关拦截
在大模型深度接入企业内部业务系统（如客服查询、财务报表导出、自动化运维）的场景中，敏感数据泄露（PII & Secret Data Leakage）是企业面临的最严峻合规与安全红线：
- 用户在与大模型对话时，无意中输入了自己的真实手机号、身份证号或银行卡号；
- 大模型在调用第三方查询工具时，将这些明文个人隐私（PII）完整打包塞进了 HTTP 请求体发送给了外部公开的供应商；
- 更有甚者，大模型在生成调用配置工具时，可能将系统内部的
AWS_SECRET_KEY或数据库连接密码作为参数打印在了全局审计日志里！
如果仅靠在 System Prompt 里提醒模型：“请不要泄露敏感数据”，在面对复杂的提示词注入（Prompt Injection）和多轮对话泛化时，防御成功率几乎为零。
必须在应用层与底层工具/模型之间，构筑一套硬编码、基于正则表达式与高性能 Aho-Corasick 多模式匹配的“双向脱敏与安全网关（DLP Gatekeeper）”。
今天我们拆解这套在请求入站、工具出站与日志落盘全链路的敏感数据拦截与脱敏实践。
一、敏感数据双向脱敏架构全景图
flowchart TD UserIn[用户输入明文: 包含手机号/身份证/密钥] --> InGate[入站脱敏网关 Inbound DLP] subgraph Sanitization_Pipeline [双向脱敏流水线] InGate --> ReplaceFake[1. 正则+实体识别: 提取真实敏感值 -> 替换为匿名占位符 [PHONE_01]] ReplaceFake --> SaveVault[2. 临时存入安全密钥保险箱 (Memory Vault, TTL 5分钟)] SaveFakePrompt --> LLM[3. 向大模型传递脱敏后的安全 Prompt] LLM --> ToolCall[4. 大模型发出 Tool Call: query_user([PHONE_01])] ToolCall --> OutGate[出站还原网关 Outbound Vault Resolver] OutGate --> Restore[5. 仅在调用受信任的内部工具时, 还原真实手机号发起请求] OutGate --> MaskLog[6. 写入审计日志时, 强制马赛克脱敏 (138****1234)] end Restore --> InternalService[调用内部受信任微服务]二、生产级 Python 双向脱敏与保险箱（Vault）代码实现
import re import uuid from typing import Dict, Tuple, Any class SensitiveDataSanitizer: def __init__(self): # 1. 核心敏感数据正则表达式规则集 self.patterns = { "PHONE": (re.compile(r'(?<!\d)(?:(?:\+|00)86)?1[3-9]\d{9}(?!\d)'), "【手机号掩码_{id}】"), "ID_CARD": (re.compile(r'(?<!\d)[1-9]\d{5}(?:18|19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx](?!\d)'), "【身份证掩码_{id}】"), "API_KEY": (re.compile(r'(?:sk-[a-zA-Z0-9]{32,48}|ghp_[a-zA-Z0-9]{36})'), "【API密钥掩码_{id}】"), "BANK_CARD": (re.compile(r'(?<!\d)(?:62|4|5)[0-9]{15,18}(?!\d)'), "【银行卡掩码_{id}】") } def sanitize_for_llm(self, text: str) -> Tuple[str, Dict[str, str]]: """ 向大模型发送前：将明文替换为匿名占位符，并将映射存入 Vault """ vault_mapping = {} # placeholder -> real_value sanitized_text = text for p_type, (regex, placeholder_tpl) in self.patterns.items(): matches = list(regex.finditer(sanitized_text)) # 从后向前替换，防止字符串下标偏移 for match in reversed(matches): real_val = match.group(0) placeholder_id = uuid.uuid4().hex[:6] placeholder = placeholder_tpl.format(id=placeholder_id) vault_mapping[placeholder] = real_val start, end = match.span() sanitized_text = sanitized_text[:start] + placeholder + sanitized_text[end:] return sanitized_text, vault_mapping def resolve_for_internal_tool(self, tool_args: dict, vault_mapping: Dict[str, str]) -> dict: """ 工具真正发起前：将占位符还原为真实敏感数据 """ def _recursive_restore(val): if isinstance(val, str): for placeholder, real in vault_mapping.items(): if placeholder in val: val = val.replace(placeholder, real) return val elif isinstance(val, dict): return {k: _recursive_restore(v) for k, v in val.items()} elif isinstance(val, list): return [_recursive_restore(x) for x in val] return val return _recursive_restore(tool_args) def mask_for_logging(self, text: str) -> str: """ 写入审计日志时：强制打上马赛克 (138****1234) """ # 手机号脱敏 text = re.sub( r'(1[3-9]\d)(\d{4})(\d{4})', r'\1****\3', text ) # 身份证脱敏 text = re.sub( r'([1-9]\d{5})\d{8}(\d{3}[\dXx])', r'\1********\2', text ) # 密钥脱敏 text = re.sub( r'(sk-[a-zA-Z0-9]{4})[a-zA-Z0-9]+([a-zA-Z0-9]{4})', r'\1********\2', text ) return text三、端到端集成实战演示
def test_dlp_full_pipeline(): sanitizer = SensitiveDataSanitizer() # 1. 模拟用户输入了带敏感信息的提示词 raw_user_query = "请帮我给用户 13812345678 (身份证: 110101199003072391) 发送重置通知，密钥是 sk-abcdef1234567890abcdef1234567890" print(f"[*] 原始用户输入: {raw_user_query}") # 2. 入站脱敏：大模型看到的只有安全的占位符！ safe_prompt, vault = sanitizer.sanitize_for_llm(raw_user_query) print(f"[✓] 喂给大模型的安全 Prompt: {safe_prompt}") # 输出类似于: 请帮我给用户 【手机号掩码_a1b2c3】 (身份证: 【身份证掩码_d4e5f6】) 发送重置通知... # 3. 模拟大模型生成的工具参数 llm_tool_call_args = { "target_phone": "【手机号掩码_a1b2c3】", "action": "RESET_PASSWORD" } # 4. 出站网关还原：向受信任内部服务发送前还原真实数据 real_tool_args = sanitizer.resolve_for_internal_tool(llm_tool_call_args, vault) print(f"[✓] 还原后发送给内部微服务的真实参数: {real_tool_args}") # target_phone 已经安全还原为 13812345678！ # 5. 落盘审计日志：强制打上马赛克 log_line = sanitizer.mask_for_logging(f"执行重置: {real_tool_args}") print(f"[✓] 写入磁盘的合规脱敏日志: {log_line}") # 输出: 执行重置: {'target_phone': '138****5678', 'action': 'RESET_PASSWORD'}四、生产治理准则
- Vault 映射在内存中设置严格 TTL（$\le 300\text{s}$）：会话结束后立即自动从内存中销毁映射表，杜绝长期驻留；
- 严禁将未脱敏的 Raw Prompt 写入任何云端监控（如 Sentry、ELK）；
- 结合高性能 Rust 扩展：在每秒处理上万条文本的高并发网关中，使用基于 Rust 构建的
flashtext或ahocorasick-rs，将脱敏耗时控制在 0.5ms 以内。
把敏感数据双向脱敏做成基础设施的第一道城墙，大模型应用才能真正做到合规无忧、行稳致远。
