一、为什么需要数据脱敏前三讲分别解决了「输入过滤」「输出验证」「权限控制」但还有一个关键缺口工具返回的数据本身可能包含敏感信息。考虑一个典型的电商场景Agent 调用 query_orders(user_idu_123) → 返回数据 { order_id: ord_001, user_name: 张三, phone: 138****5678, ← 部分脱敏 ✓ id_card: 110101199001011234, ← 完整明文 ✗ address: 北京市朝阳区xxx小区3号楼1单元201室, ← 完整地址 ✗ payment_info: { card_number: 6222021234567890, ← 银行卡号 ✗ cvv: 123 ← CVV 码 ✗ } }即使 Agent 有权限调用query_orders它也不应该看到用户的完整身份证号、银行卡号和家庭住址。数据脱敏是L4 层防御确保 Agent 返回给用户的数据已经过清洗。二、脱敏策略分类策略方法适用场景遮盖​保留首尾中间替换为*手机号、邮箱、身份证截断​只保留前 N 位信用卡号显示后4位替换​用静态值替换真实值密码、Token、密钥哈希​单向散列不可逆用户 ID、设备指纹泛化​缩小精度地址只显示城市、年龄显示年龄段空值​直接删除该字段CVV、安全问题的答案差分隐私​添加噪声统计数据、报表三、Go 实现声明式数据脱敏引擎下面实现一个基于标签声明的脱敏引擎可以无缝集成到 MCP Server 的响应管道中。package main import ( encoding/json fmt log reflect regexp strings time ) // ---- 脱敏标签定义 ---- type MaskType string const ( MaskPhone MaskType phone // 138****5678 MaskEmail MaskType email // a***example.com MaskIDCard MaskType id_card // 110101********34 MaskBankCard MaskType bank_card // 6222****7890 MaskPassword MaskType password // ****** MaskAddress MaskType address // 北京市*** MaskName MaskType name // 张* MaskCustom MaskType custom // 自定义规则 MaskFull MaskType full // 完全遮盖 MaskHash MaskType hash // SHA256 哈希 ) // 脱敏规则 type MaskRule struct { Type MaskType json:type Params map[string]string json:params,omitempty // 额外参数 Preserve int json:preserve // 保留位数开头 Trail int json:trail // 保留位数结尾 Replace string json:replace // 替换字符 } // 默认规则集 var defaultRules map[MaskType]*MaskRule{ MaskPhone: { Type: MaskPhone, Preserve: 3, Trail: 4, Replace: *, }, MaskEmail: { Type: MaskEmail, Preserve: 1, Trail: 0, Replace: *, }, MaskIDCard: { Type: MaskIDCard, Preserve: 6, Trail: 4, Replace: *, }, MaskBankCard: { Type: MaskBankCard, Preserve: 4, Trail: 4, Replace: *, }, MaskPassword: { Type: MaskPassword, Replace: *, }, MaskName: { Type: MaskName, Preserve: 1, Trail: 0, Replace: *, }, MaskFull: { Type: MaskFull, Replace: *, }, } // ---- 脱敏引擎 ---- type MaskEngine struct { rules map[MaskType]*MaskRule customMatchers map[string]*regexp.Regexp // 自定义正则匹配 } func NewMaskEngine() *MaskEngine { engine : MaskEngine{ rules: make(map[MaskType]*MaskRule), customMatchers: make(map[string]*regexp.Regexp), } // 加载默认规则 for k, v : range defaultRules { engine.rules[k] v } return engine } // 注册自定义规则 func (me *MaskEngine) RegisterRule(name MaskType, rule *MaskRule) { me.rules[name] rule } // 注册自定义正则匹配器 func (me *MaskEngine) RegisterMatcher(name string, pattern string) { me.customMatchers[name] regexp.MustCompile(pattern) } // 对单个值进行脱敏 func (me *MaskEngine) MaskValue(value string, rule *MaskRule) string { switch rule.Type { case MaskFull: return strings.Repeat(rule.Replace, len(value)) case MaskPhone: if len(value) ! 11 { return value // 非标准手机号不做处理 } return value[:rule.Preserve] strings.Repeat(rule.Replace, len(value)-rule.Preserve-rule.Trail) value[len(value)-rule.Trail:] case MaskEmail: parts : strings.SplitN(value, , 2) if len(parts) ! 2 { return value } local : parts[0] if len(local) rule.Preserve { return local[:1] strings.Repeat(rule.Replace, len(local)-1) parts[1] } return local[:rule.Preserve] strings.Repeat(rule.Replace, len(local)-rule.Preserve) parts[1] case MaskIDCard: if len(value) rule.Preserverule.Trail { return value } return value[:rule.Preserve] strings.Repeat(rule.Replace, len(value)-rule.Preserve-rule.Trail) value[len(value)-rule.Trail:] case MaskBankCard: if len(value) rule.Preserverule.Trail { return value } return value[:rule.Preserve] strings.Repeat(rule.Replace, len(value)-rule.Preserve-rule.Trail) value[len(value)-rule.Trail:] case MaskName: if len(value) 1 { return value } return value[:rule.Preserve] strings.Repeat(rule.Replace, len(value)-rule.Preserve) case MaskPassword: return strings.Repeat(rule.Replace, len(value)) case MaskHash: return fmt.Sprintf(%x, sha256Hash(value))[:16] // 取前16位 case MaskCustom: if matcher, ok : me.customMatchers[rule.Params[matcher]]; ok { return matcher.ReplaceAllString(value, rule.Replace) } return value default: return value } } // ---- 声明式脱敏注解 ---- // 通过结构体 tag 声明脱敏规则 // // type Order struct { // Phone string mask:phone // Email string mask:email // Name string mask:name // IDCard string mask:id_card // Amount float64 mask:- // - 表示不脱敏 // } // 递归脱敏任意数据结构 func (me *MaskEngine) MaskData(data interface{}, path string) interface{} { if data nil { return nil } val : reflect.ValueOf(data) switch val.Kind() { case reflect.Map: result : make(map[string]interface{}) for _, key : range val.MapKeys() { keyStr : fmt.Sprintf(%v, key.Interface()) newPath : path . keyStr result[keyStr] me.MaskData(val.MapIndex(key).Interface(), newPath) } return result case reflect.Slice, reflect.Array: result : make([]interface{}, val.Len()) for i : 0; i val.Len(); i { newPath : fmt.Sprintf(%s[%d], path, i) result[i] me.MaskData(val.Index(i).Interface(), newPath) } return result case reflect.Struct: result : make(map[string]interface{}) t : val.Type() for i : 0; i t.NumField(); i { field : t.Field(i) fieldVal : val.Field(i) tag : field.Tag.Get(mask) newPath : path . field.Name if tag - || tag { // 不脱敏但递归处理内部结构 result[field.Name] me.MaskData(fieldVal.Interface(), newPath) } else { // 按标签脱敏 maskType : MaskType(tag) if rule, ok : me.rules[maskType]; ok { strVal : fmt.Sprintf(%v, fieldVal.Interface()) result[field.Name] me.MaskValue(strVal, rule) } else { result[field.Name] me.MaskData(fieldVal.Interface(), newPath) } } } return result case reflect.String: // 对于裸字符串尝试自动识别敏感模式 return me.autoDetectAndMask(val.String()) default: return data } } // 自动识别并脱敏常见敏感数据 func (me *MaskEngine) autoDetectAndMask(s string) string { // 手机号11位数字 phoneRegex : regexp.MustCompile(1[3-9]\d{9}) if phoneRegex.MatchString(s) { rule : me.rules[MaskPhone] return phoneRegex.ReplaceAllStringFunc(s, func(match string) string { return me.MaskValue(match, rule) }) } // 身份证18位最后一位可能是X idCardRegex : regexp.MustCompile(\d{17}[\dXx]) if idCardRegex.MatchString(s) { rule : me.rules[MaskIDCard] return idCardRegex.ReplaceAllStringFunc(s, func(match string) string { return me.MaskValue(match, rule) }) } // 邮箱 emailRegex : regexp.MustCompile([a-zA-Z0-9._%-][a-zA-Z0-9.-]\.[a-zA-Z]{2,}) if emailRegex.MatchString(s) { rule : me.rules[MaskEmail] return emailRegex.ReplaceAllStringFunc(s, func(match string) string { return me.MaskValue(match, rule) }) } return s } // ---- 集成到 MCP Server 的脱敏中间件 ---- type MaskingMiddleware struct { engine *MaskEngine // 路径到脱敏规则的映射手工配置 pathRules map[string]*MaskRule } func NewMaskingMiddleware(engine *MaskEngine) *MaskingMiddleware { return MaskingMiddleware{ engine: engine, pathRules: make(map[string]*MaskRule), } } // 为特定路径配置脱敏规则 func (mm *MaskingMiddleware) AddPathRule(path string, rule *MaskRule) { mm.pathRules[path] rule } // 对 MCP 工具的返回数据进行脱敏 func (mm *MaskingMiddleware) ProcessResponse(toolName string, responseData interface{}) interface{} { // 基于工具名称的脱敏配置 switch toolName { case query_orders: return mm.maskOrderResponse(responseData) case get_user_profile: return mm.maskUserProfile(responseData) case search_logs: return mm.maskLogResponse(responseData) default: // 通用脱敏递归扫描所有字符串 return mm.engine.MaskData(responseData, toolName) } } // 订单脱敏 func (mm *MaskingMiddleware) maskOrderResponse(data interface{}) interface{} { masked : mm.engine.MaskData(data, query_orders) // 额外的业务规则金额超过 10000 的订单脱敏商品名称 if m, ok : masked.(map[string]interface{}); ok { if amount, exists : m[total_amount]; exists { if amt, ok : amount.(float64); ok amt 10000 { if items, exists : m[items]; exists { if itemsArr, ok : items.([]interface{}); ok { for _, item : range itemsArr { if itemMap, ok : item.(map[string]interface{}); ok { if _, hasName : itemMap[product_name]; hasName { itemMap[product_name] [高价值商品已脱敏] } } } } } } } } return masked } // 用户资料脱敏 func (mm *MaskingMiddleware) maskUserProfile(data interface{}) interface{} { masked : mm.engine.MaskData(data, get_user_profile) // 删除完全不能暴露的字段 if m, ok : masked.(map[string]interface{}); ok { delete(m, password_hash) delete(m, security_question) delete(m, security_answer) delete(m, internal_notes) } return masked } // 日志脱敏保留错误信息脱敏敏感数据 func (mm *MaskingMiddleware) maskLogResponse(data interface{}) interface{} { return mm.engine.MaskData(data, search_logs) } // ---- 工具函数 ---- func sha256Hash(s string) []byte { h : sha256.New() h.Write([]byte(s)) return h.Sum(nil) } // ---- 演示 ---- func main() { engine : NewMaskEngine() middleware : NewMaskingMiddleware(engine) // 模拟订单查询返回数据 orderResponse : map[string]interface{}{ order_id: ord_001, user: map[string]interface{}{ name: 张三丰, phone: 13812345678, email: zhangsanexample.com, id_card: 110101199001011234, address: 北京市海淀区中关村大街1号院3号楼501, }, payment: map[string]interface{}{ method: credit_card, card_number: 6222021234567890, cvv: 123, amount: 15999.00, }, items: []interface{}{ map[string]interface{}{ product_name: iPhone 16 Pro Max 1TB, quantity: 1, price: 12999.00, }, map[string]interface{}{ product_name: AirPods Pro 3, quantity: 2, price: 1500.00, }, }, internal_notes: VIP客户历史投诉3次谨慎处理, } fmt.Println( 原始数据 ) printJSON(orderResponse) fmt.Println(\n 脱敏后数据 ) masked : middleware.ProcessResponse(query_orders, orderResponse) printJSON(masked) // 测试日志脱敏 fmt.Println(\n 日志脱敏测试 ) logLine : ERROR: user 13812345678 login failed from IP 192.168.1.100 fmt.Printf(原始: %s\n, logLine) fmt.Printf(脱敏: %s\n, engine.autoDetectAndMask(logLine)) } func printJSON(data interface{}) { b, _ : json.MarshalIndent(data, , ) fmt.Println(string(b)) }四、脱敏策略的最佳实践4.1 分级脱敏策略根据不同角色的数据可见级别同一字段可以有不同脱敏程度// 客服手机号显示 138****5678 // 物流手机号显示 13812345678需要联系客户 // 财务手机号显示完整对账需要在 MCP Gateway 中实现func getMaskRuleForRole(role string, field string) *MaskRule { switch role { case cs_agent: return MaskRule{Type: MaskPhone, Preserve: 3, Trail: 4} case logistics: return nil // 不脱敏 case finance: return nil // 不脱敏 default: return MaskRule{Type: MaskFull} } }4.2 脱敏审计每次脱敏操作都应该记录审计日志type MaskAuditLog struct { Timestamp time.Time json:timestamp AgentID string json:agent_id ToolName string json:tool_name FieldPath string json:field_path MaskType MaskType json:mask_type OriginalLen int json:original_len // 原始长度不记录原始值 }4.3 不可逆 vs 可逆脱敏场景方案说明客服查看不可逆脱敏遮盖/替换无法还原数据分析可逆脱敏保留可还原的映射表测试环境合成数据用假数据替代真实数据监管审计不脱敏只有特定角色可看原始数据五、安全分层L4本讲在 L1-L3 基础上叠加数据脱敏层层级防御手段本讲新增L1​Prompt 注入检测输入过滤L2​输出验证与对齐参数校验、业务规则L3​工具权限与最小特权RBAC/ABAC 策略引擎L4a​字段级脱敏✅ 声明式标签脱敏L4b​自动敏感数据识别✅ 正则自动匹配L4c​角色感知脱敏✅ 不同角色不同脱敏等级L4d​脱敏审计日志✅ 记录脱敏操作六、延伸阅读GDPR 数据脱敏指南欧盟对个人数据的保护要求PCI DSS 支付卡行业数据安全标准信用卡号的脱敏规范Apache ShardingSphere 数据脱敏数据库层面的脱敏方案Google Differential Privacy Library差分隐私库用于统计数据的隐私保护七、下一讲预告第5讲行为审计与异常检测——谁在什么时候做了什么有了输入过滤、输出验证、权限控制和数据脱敏我们已经构筑了四道防线。但攻击往往是渐进的——一次正常的查询、一次合法的调用组合起来可能就是一次数据窃取。下一讲实现行为审计系统记录每一次工具调用并实时检测异常行为模式。开发之余的小工具推荐处理 Base64、JSON 格式化、JWT 解析、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top。所有计算在浏览器完成文件不上服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。