From 9506a82be972cc579611e3b4f3c6c983467a6271 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Thu, 25 Jun 2026 16:46:08 +0800 Subject: [PATCH] =?UTF-8?q?feat(harness):=20=E8=BE=93=E5=87=BA=E8=84=B1?= =?UTF-8?q?=E6=95=8F=E5=A2=9E=E5=BC=BA=20=E2=80=94=E2=80=94=20=E8=B7=A8?= =?UTF-8?q?=E5=88=86=E7=89=87=20StreamRedactor=20+=20PII=EF=BC=8C=E6=9D=9C?= =?UTF-8?q?=E7=BB=9D=E5=AF=86=E9=92=A5=E7=A2=8E=E7=89=87=E6=B3=84=E6=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原逐片脱敏有两个漏：①密钥被切成两片（"sk-912cf85b"|"16d0..."）逐片都不命中正则而漏检； ②贪婪正则在缓冲末尾凑够最短长度就把半截密钥提前脱敏发走、剩余字符随后明文流出（碎片泄漏）。 有状态 StreamRedactor 跨分片缓冲，切点在「原文」上定且绝不切断任何完整匹配： - opener 暂留末尾仍在增长的疑似密钥（sk/AKIA/JWT/Bearer/手机/邮箱/长数字） - 始终留 16B 尾窗兜底 opener 未覆盖的短模式；勿切断完整匹配（循环至稳定） - rune 边界安全：cut 退到最近 rune 起点，中文不被切成半个发出乱码 - 暂留封顶 256B，防对抗性长串无限暂留 / O(n²) - 新增 PII：手机号 / 邮箱 / 身份证（18 位） 3 个流式点（graph/react_agent/compose_graph）统一接入，逐片 Push + 收尾 Flush。 7 单测（跨片/逐字符 JWT/碎片回归/尾窗内匹配/干净重建/PII/无误伤），-race 干净。 live 实测：26 位密钥（曾泄漏 ijkl90mnop 碎片）与邮箱整条 [已脱敏]。 Co-Authored-By: Claude Opus 4.8 (1M context) --- project_analysis.md | 4 +- .../internal/eino/compose_graph.go | 15 +- sundynix-dispatcher/internal/eino/graph.go | 17 +-- .../internal/eino/react_agent.go | 15 +- .../internal/harness/output.go | 134 ++++++++++++++++-- .../internal/harness/output_test.go | 120 ++++++++++++++++ 6 files changed, 278 insertions(+), 27 deletions(-) diff --git a/project_analysis.md b/project_analysis.md index ee194e7..6db2a52 100644 --- a/project_analysis.md +++ b/project_analysis.md @@ -116,7 +116,9 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为 `GET /tasks/:id/eval` 可查。dispatcher 评→广播→网关落库。剩：桌面端质量面板、低分自动重试(P3)。 - [x] **P1 RAG 忠实度评测** ✅：检索原文喂给 judge，一次评质量+忠实度，未被来源支持的说法进 Flags。 综合分(有来源)=0.3规则+0.35质量+0.35忠实。runGraph 透传 refs → evaluate。live 实测忠实 1.00/来源 1，单测覆盖。 -- [ ] **P2 输出脱敏增强**：滑动窗口跨片检测（现流式逐片会漏跨 token 的密钥）+ PII 模式（手机号/邮箱/身份证）。 +- [x] **P2 输出脱敏增强** ✅：有状态 `StreamRedactor` 跨分片缓冲，切点在原文上定且**绝不切断完整匹配**， + 杜绝密钥被切成两片漏检 / 提前脱敏半截致碎片泄漏（逐字符 JWT 流亦完整捕获）；rune 边界安全（中文不乱码）； + 新增 PII（手机号/邮箱/身份证）；opener+尾窗双兜底、暂留封顶防 DoS。3 流式点接入，7 单测，live 实测密钥/邮箱整条脱敏。 - [ ] **P2 输入护栏升级**：纯正则易被改写/编码绕过；加轻量 jailbreak 分类器或 LLM 兜底；`bannedTerms` 落地。 - [x] **P3 坏输出自动纠偏** ✅：poor(<0.5) 触发评语驱动的重生成，重评后**仅采纳更优者(不退步)**， 采纳的修订版落会话历史 + 评测终值带 `corrected` 标记落库。`maxRefineRounds=1`、`canRefine`(模型就绪且熔断未开)门控； diff --git a/sundynix-dispatcher/internal/eino/compose_graph.go b/sundynix-dispatcher/internal/eino/compose_graph.go index e49d318..864c3e5 100644 --- a/sundynix-dispatcher/internal/eino/compose_graph.go +++ b/sundynix-dispatcher/internal/eino/compose_graph.go @@ -75,6 +75,15 @@ func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string chunks := 0 var produced strings.Builder // 本节点产出（供下游 agent 接力） + red := harness.NewStreamRedactor() // 输出护栏：跨分片脱敏，杜绝密钥被切断而漏检 + emit := func(safe string) { + if safe == "" { + return + } + _ = o.sink.PublishToken(taskID, []byte(safe)) + produced.WriteString(safe) + chunks++ + } for { chunk, rerr := sr.Recv() if rerr == io.EOF { @@ -87,11 +96,9 @@ func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string if chunk.Content == "" { continue } - safe, _ := harness.RedactSecrets(chunk.Content) // 输出护栏：逐片脱敏 - _ = o.sink.PublishToken(taskID, []byte(safe)) - produced.WriteString(safe) - chunks++ + emit(red.Push(chunk.Content)) } + emit(red.Flush()) // 吐出暂留尾部 o.recordAgentOutput(b, produced.String()) tr.info(node, "system", "compose 图", fmt.Sprintf("%d 段输出 / %d 字（Eino compose 运行时）", chunks, len([]rune(produced.String())))) } diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go index 6dc7a1c..1d6019d 100644 --- a/sundynix-dispatcher/internal/eino/graph.go +++ b/sundynix-dispatcher/internal/eino/graph.go @@ -275,19 +275,19 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy msgs, _ := buildMessages(ctx, rc) tr.emit(node, "model", "start", "模型流式推理", "", 0) t0 := time.Now() - n, redacted := 0, 0 + n := 0 var produced strings.Builder // 本节点自身产出（用于沿图向下游传递） - send := func(s string) { - if s == "" { + // 输出护栏：有状态流式脱敏，跨分片缓冲，杜绝密钥被切断而漏检（逐片正则会漏跨片密钥）。 + red := harness.NewStreamRedactor() + emit := func(safe string) { + if safe == "" { return } - // 输出护栏：发射前逐片脱敏疑似密钥/令牌（流式无法回收已发，故逐片处理）。 - safe, hit := harness.RedactSecrets(s) - redacted += hit _ = o.sink.PublishToken(taskID, []byte(safe)) produced.WriteString(safe) n++ } + send := func(s string) { emit(red.Push(s)) } var err error if o.pool.Ready() { err = o.pool.ChatStream(ctx, toChatMessages(msgs), send) @@ -303,8 +303,9 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy } return } - if redacted > 0 { - tr.info(node, "system", "输出护栏", fmt.Sprintf("已脱敏 %d 处疑似密钥/令牌", redacted)) + emit(red.Flush()) // 吐出暂留的尾部（最后一段疑似密钥的判定） + if red.Hits() > 0 { + tr.info(node, "system", "输出护栏", fmt.Sprintf("已脱敏 %d 处疑似密钥/PII", red.Hits())) } o.recordAgentOutput(b, produced.String()) // 产出入黑板：成当前成稿 + 供下游接力 tr.emit(node, "model", "end", "模型流式推理", diff --git a/sundynix-dispatcher/internal/eino/react_agent.go b/sundynix-dispatcher/internal/eino/react_agent.go index 52a87eb..d9faa8a 100644 --- a/sundynix-dispatcher/internal/eino/react_agent.go +++ b/sundynix-dispatcher/internal/eino/react_agent.go @@ -214,6 +214,15 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar chunks := 0 var produced strings.Builder // 本节点产出（供下游 agent 接力） + red := harness.NewStreamRedactor() // 输出护栏：跨分片脱敏，杜绝密钥被切断而漏检 + emit := func(safe string) { + if safe == "" { + return + } + _ = o.sink.PublishToken(taskID, []byte(safe)) + produced.WriteString(safe) + chunks++ + } for { chunk, rerr := sr.Recv() if rerr == io.EOF { @@ -226,11 +235,9 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar if chunk.Content == "" { continue // 工具调用片段无正文，跳过；正文只来自模型答复 } - safe, _ := harness.RedactSecrets(chunk.Content) - _ = o.sink.PublishToken(taskID, []byte(safe)) - produced.WriteString(safe) - chunks++ + emit(red.Push(chunk.Content)) } + emit(red.Flush()) // 吐出暂留尾部 o.recordAgentOutput(b, produced.String()) tr.emit(node, "model", "end", "ReAct 智能体", fmt.Sprintf("%d 段输出 / %d 字", chunks, len([]rune(produced.String()))), time.Since(t0).Milliseconds()) diff --git a/sundynix-dispatcher/internal/harness/output.go b/sundynix-dispatcher/internal/harness/output.go index 439af41..85b5b32 100644 --- a/sundynix-dispatcher/internal/harness/output.go +++ b/sundynix-dispatcher/internal/harness/output.go @@ -1,26 +1,140 @@ package harness -import "regexp" +import ( + "regexp" + "strings" + "unicode/utf8" +) // secretPatterns 是输出里疑似密钥/令牌的特征（命中即脱敏，防模型把密钥回吐给用户）。 var secretPatterns = []*regexp.Regexp{ - regexp.MustCompile(`sk-[A-Za-z0-9_-]{16,}`), // OpenAI/DeepSeek 风格 key - regexp.MustCompile(`AKIA[0-9A-Z]{16}`), // AWS Access Key ID + regexp.MustCompile(`sk-[A-Za-z0-9_-]{16,}`), // OpenAI/DeepSeek 风格 key + regexp.MustCompile(`AKIA[0-9A-Z]{16}`), // AWS Access Key ID regexp.MustCompile(`eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{4,}`), // JWT - regexp.MustCompile(`(?i)bearer\s+[A-Za-z0-9._-]{16,}`), // Bearer 令牌 + regexp.MustCompile(`(?i)bearer\s+[A-Za-z0-9._-]{16,}`), // Bearer 令牌 } +// piiPatterns 是个人隐私信息（PII）特征：手机号 / 邮箱 / 身份证。用 \b 边界避免吞进更长数字串。 +var piiPatterns = []*regexp.Regexp{ + regexp.MustCompile(`\b1[3-9]\d{9}\b`), // 中国大陆手机号（11 位） + regexp.MustCompile(`[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}`), // 邮箱 + regexp.MustCompile(`\b\d{17}[\dXx]\b`), // 身份证（18 位，末位可 X） +} + +// openerPatterns 匹配「位于字符串末尾、可能仍在增长」的敏感串前缀（锚定 $）。 +// 流式逐片脱敏会漏掉被分片切断的密钥（如 "sk-912cf85b"|"16d0..."），逐片都不单独命中。 +// StreamRedactor 据此把"半截疑似密钥"的起点之后整段暂留，与下一分片拼接后再判，消除跨片漏检。 +var openerPatterns = []*regexp.Regexp{ + regexp.MustCompile(`sk-[A-Za-z0-9_-]*$`), + regexp.MustCompile(`AKIA[0-9A-Z]*$`), + regexp.MustCompile(`eyJ[A-Za-z0-9_-]*(\.[A-Za-z0-9_-]*){0,2}$`), + regexp.MustCompile(`(?i)bearer\s*[A-Za-z0-9._-]*$`), + regexp.MustCompile(`1[3-9]\d*$`), // 手机号（增长中） + regexp.MustCompile(`[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]*$`), // 邮箱 @ 之后增长中 + regexp.MustCompile(`\d{6,}$`), // 长数字串（身份证等）增长中 +} + +// allPatterns 是全部完整匹配特征（密钥 + PII），供流式切点判定"勿切断完整匹配"。 +var allPatterns = append(append([]*regexp.Regexp{}, secretPatterns...), piiPatterns...) + const redactMark = "[已脱敏]" -// RedactSecrets 把文本里疑似密钥/令牌替换为脱敏标记，返回脱敏后文本与命中次数。 -// 输出护栏：在 token 发射前对每个分片调用（流式无法回收已发，故逐片脱敏 + 最终标记）。 +// RedactSecrets 把文本里疑似密钥/令牌/PII 替换为脱敏标记，返回脱敏后文本与命中次数。 +// 整段脱敏（非流式场景、StreamRedactor 内部、最终兜底都用它）。 func RedactSecrets(s string) (string, int) { n := 0 + repl := func(string) string { n++; return redactMark } for _, re := range secretPatterns { - s = re.ReplaceAllStringFunc(s, func(string) string { - n++ - return redactMark - }) + s = re.ReplaceAllStringFunc(s, repl) + } + for _, re := range piiPatterns { + s = re.ReplaceAllStringFunc(s, repl) } return s, n } + +const ( + // redactMinTail 始终暂留的尾窗：兜住 opener 未识别、却被跨片切断的短模式。 + redactMinTail = 16 + // redactMaxHold 暂留上限：超过即强制发射（脱敏后），防对抗性长串导致无限暂留 / O(n²)。 + redactMaxHold = 256 +) + +// StreamRedactor 是有状态的流式脱敏器：跨分片缓冲，解决密钥被切断而漏检。 +// 用法：每分片 Push() 得可安全发射的文本，流结束 Flush() 吐出暂留尾部。非并发安全（单流单实例）。 +type StreamRedactor struct { + carry strings.Builder + hits int +} + +// NewStreamRedactor 建一个流式脱敏器（每个 token 流一个实例）。 +func NewStreamRedactor() *StreamRedactor { return &StreamRedactor{} } + +// Push 吃一个分片，返回当前可安全发射的脱敏文本（可能为空——尾部疑似半截敏感串被暂留待下片）。 +// 关键：切点在「原文」缓冲上定，只对已结算的头部脱敏、尾部按原文暂留 —— 否则贪婪正则会在 +// 缓冲末尾凑够最短长度就把半截密钥提前脱敏发走，剩余字符随后明文泄漏。 +func (r *StreamRedactor) Push(chunk string) string { + if chunk == "" { + return "" + } + r.carry.WriteString(chunk) + buf := r.carry.String() + cut := safeCut(buf) + head, n := RedactSecrets(buf[:cut]) + r.hits += n + r.carry.Reset() + r.carry.WriteString(buf[cut:]) // 原文暂留（未脱敏，供下片继续判定） + return head +} + +// Flush 流结束时吐出暂留尾部（做最终脱敏）。返回脱敏后的剩余文本。 +func (r *StreamRedactor) Flush() string { + s, n := RedactSecrets(r.carry.String()) + r.hits += n + r.carry.Reset() + return s +} + +// Hits 返回累计脱敏命中次数。 +func (r *StreamRedactor) Hits() int { return r.hits } + +// safeCut 在「原文」缓冲上计算可安全发射的截断点：暂留 [cut:]，发射并脱敏 [:cut]。 +// 三层保障：① opener 暂留末尾仍在增长的疑似密钥；② 始终留一段尾窗兜底未被 opener 覆盖的短模式； +// ③ 切点绝不切断任何「完整匹配」（否则半截匹配的头部会明文流出）。最后以暂留上限封顶防 DoS。 +func safeCut(s string) int { + cut := len(s) + for _, re := range openerPatterns { + if loc := re.FindStringIndex(s); loc != nil && loc[0] < cut { + cut = loc[0] + } + } + if minCut := len(s) - redactMinTail; minCut < cut { + cut = minCut // 始终暂留尾窗 + } + // 勿切断完整匹配：把任何跨越 cut 的完整匹配整体并入暂留尾（循环至稳定，匹配数有限）。 + for { + moved := false + for _, re := range allPatterns { + for _, loc := range re.FindAllStringIndex(s, -1) { + if loc[0] < cut && loc[1] > cut { + cut = loc[0] + moved = true + } + } + } + if !moved { + break + } + } + if maxCut := len(s) - redactMaxHold; maxCut > cut { + cut = maxCut // 暂留封顶（最终硬限，防对抗性长串无限暂留 / O(n²)） + } + if cut < 0 { + cut = 0 + } + // 退到最近的 rune 边界：cut 来自字节索引，直接切多字节字符（中文）会发出半个 rune 乱码。 + for cut > 0 && cut < len(s) && !utf8.RuneStart(s[cut]) { + cut-- + } + return cut +} diff --git a/sundynix-dispatcher/internal/harness/output_test.go b/sundynix-dispatcher/internal/harness/output_test.go index 981eb48..eacfe67 100644 --- a/sundynix-dispatcher/internal/harness/output_test.go +++ b/sundynix-dispatcher/internal/harness/output_test.go @@ -37,3 +37,123 @@ func TestRedactSecrets_Clean(t *testing.T) { t.Errorf("正常文本不应被改动: n=%d out=%q", n, out) } } + +// PII：手机号 / 邮箱 / 身份证应被脱敏。 +func TestRedactSecrets_PII(t *testing.T) { + cases := []struct { + in string + mustGone string + }{ + {"联系我 13812345678 谢谢", "13812345678"}, + {"邮箱 zhang.san@example.com 收", "zhang.san@example.com"}, + {"身份证 11010519900307123X 备案", "11010519900307123X"}, + } + for _, c := range cases { + out, n := RedactSecrets(c.in) + if n != 1 { + t.Errorf("RedactSecrets(%q) 命中=%d want 1", c.in, n) + } + if strings.Contains(out, c.mustGone) { + t.Errorf("脱敏后仍含 PII: %q", out) + } + } +} + +// 不应误伤：更长数字串里的合法 11 位子串不该被吞，普通词不该当邮箱。 +func TestRedactSecrets_NoFalsePositive(t *testing.T) { + for _, in := range []string{"订单号 138123456789012", "用户名 a@b（非邮箱）", "数量 12345"} { + if out, n := RedactSecrets(in); n != 0 { + t.Errorf("不应脱敏 %q，却命中 %d → %q", in, n, out) + } + } +} + +// streamAll 把分片喂给 StreamRedactor，拼回消费端看到的完整文本与命中数。 +func streamAll(chunks ...string) (string, int) { + r := NewStreamRedactor() + var sb strings.Builder + for _, c := range chunks { + sb.WriteString(r.Push(c)) + } + sb.WriteString(r.Flush()) + return sb.String(), r.Hits() +} + +// 跨分片：密钥被切成两片，逐片都不单独命中正则，流式脱敏须靠缓冲拼接捕获。 +func TestStreamRedactor_CrossChunkSecret(t *testing.T) { + // "sk-912cf85b" + "16d04b22bcb95f4576423bfb"：第一片仅 8 位 body（<16 不命中），第二片无前缀。 + out, n := streamAll("我的 key 是 sk-912cf85b", "16d04b22bcb95f4576423bfb 别外传") + if strings.Contains(out, "sk-912cf85b16d04b22bcb95f4576423bfb") { + t.Errorf("跨分片密钥漏检: %q", out) + } + if n != 1 { + t.Errorf("应脱敏 1 处，got %d (%q)", n, out) + } + if !strings.Contains(out, redactMark) { + t.Errorf("应含脱敏标记: %q", out) + } +} + +// 跨分片 + 逐字符流（最严苛）：JWT 一个字符一片，仍须完整捕获。 +func TestStreamRedactor_CharByChar(t *testing.T) { + secret := "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.abcd1234" + full := "令牌：" + secret + "，妥善保管" + chunks := make([]string, 0, len(full)) + for _, r := range full { + chunks = append(chunks, string(r)) + } + out, n := streamAll(chunks...) + if strings.Contains(out, secret) { + t.Errorf("逐字符流 JWT 漏检: %q", out) + } + if n != 1 { + t.Errorf("应脱敏 1 处，got %d (%q)", n, out) + } +} + +// 回归：密钥 body 跨片，且第一片恰好凑够最短长度(16)。贪婪正则会提前脱敏半截、剩余明文泄漏。 +// 切点须在原文上定、勿切断完整匹配 → 整条密钥完整脱敏，无碎片。 +func TestStreamRedactor_NoFragmentLeak(t *testing.T) { + out, n := streamAll("凭据 sk-ABCD1234efgh5678", "ijkl90mnop 完毕") + if strings.Contains(out, "ijkl90mnop") { + t.Errorf("密钥碎片泄漏: %q", out) + } + if strings.Contains(out, "sk-ABCD1234efgh5678") { + t.Errorf("密钥头部泄漏: %q", out) + } + if n != 1 { + t.Errorf("应脱敏 1 处，got %d (%q)", n, out) + } +} + +// 完整匹配落在尾窗内（身份证后仅跟少量文字）：切点须整体并入暂留，不可从中切断。 +func TestStreamRedactor_MatchInTailWindow(t *testing.T) { + out, n := streamAll("身份证 11010519900307123X 完") + if strings.Contains(out, "11010519900307123X") { + t.Errorf("尾窗内身份证漏检: %q", out) + } + if n != 1 { + t.Errorf("应脱敏 1 处，got %d (%q)", n, out) + } +} + +// 干净文本逐片流入，消费端拼回必须与原文逐字节一致（无乱码、无丢字、无暂留残留）。 +func TestStreamRedactor_CleanReconstruct(t *testing.T) { + full := "杭州西湖十景：苏堤春晓、曲院风荷、平湖秋月、断桥残雪，四季皆宜。" + chunks := make([]string, 0) + rs := []rune(full) + for i := 0; i < len(rs); i += 2 { // 每 2 个中文字一片，制造多字节边界压力 + end := i + 2 + if end > len(rs) { + end = len(rs) + } + chunks = append(chunks, string(rs[i:end])) + } + out, n := streamAll(chunks...) + if out != full { + t.Errorf("干净文本未原样重建:\n got %q\nwant %q", out, full) + } + if n != 0 { + t.Errorf("干净文本不应脱敏，got %d", n) + } +}