MemoryOS 聊天安全一张图:分层、时序与防护总表

说明:本文是 MemoryOS 聊天 / RAG 安全实现对照,与 《LLM 聊天与 RAG 安全怎么落地》 互补——那一篇讲方法论,本篇讲本系统在哪一层、用什么包、防什么、还缺什么。文中 DeSyntax 指 EntropyShield 包的打碎句法 mask,见第一节末「DeSyntax 是什么」。
回顾顺序:第一节总表与 DeSyntax 说明 → 分层图 → 时序图

一、防护总表(主索引)

下表为全文核心:防护模块名 = 业务称呼(第三方包 / 自研模块 / 规划组件)

流量分层防护模块(包 / 代码)精确部署位置执行时机实现方案抵御攻击设计理由优先级依赖成本短板配套互补风险提示
1. 前端 / BFF(可绕过)BFF Guardllm-prompt-guard · apps/web/lib/prompt-guard.tsNext BFF:app/api/chat/** 转发 API 前用户点发送,body 未到 FastAPIcreateGuard().sanitizetag 仅标记转发 / quarantine 高危英文 422直白英文 / 中文基础 Prompt 注入;简单角色劫持句式;少量带空格换行拆分的简易越狱指令;无编码、无字符变形的基础攻击载荷早反馈、减无效 API 请求;非权威P2npm 零后端依赖直连 API 完全绕过llm-injection-guard 中间件、prompt_security不可作唯一注入防护
2. API 网关层滑动窗口限流(规划 · Redis 自研 · rate_limit.py统一入口:/auth/login/chat/completions/demo-turnHTTP 进入、鉴权之前Redis 滑动窗口 → 429 code=42901HTTP 层 DoS;高频批量刷对话;恶意脚本短时间大量新建会话;批量长对话消耗 GPU / 算力;恶意批量请求拉高服务成本注入防住仍可能被高频烧算力P0Redis 计数慢速低频渗透注入Token 日配额、audit_log全路由强制,无豁免
2. API 网关层HTTP 注入中间件llm-injection-guard · middleware/injection_guard.py + llm_injection_guard_adapter.pyFastAPI InjectionGuardMiddleware;仅 POST /api/v1/chat/completionsBody 解析后、路由 handler 之前PromptScanner(包内规则引擎,非自研正则)中英文指令覆盖 / 角色劫持;Base64 / 十六进制 / URL 编码伪装注入;零宽空格、RTL 方向控制隐形字符绕过;伪造 <system> 分隔标签;长换行填充稀释系统规则;基础 RAG 埋毒短句载荷服务端第一层统一校验,弥补 BFF 可绕过P0纯内存 + 轻量 pip多语种语义诱导、字符变体prompt_securityllm-guard所有 chat 请求必经;与启发式对照实验
2. API 网关层JWT 会话鉴权PyJWT · core/deps.py get_current_user全部受保护路由 Depends(get_current_user)限流 / 中间件之后JWT 校验 + get_owned_conversation owner 绑定窃取 Token 越权访问他人对话;未登录调用受保护聊天接口;篡改会话归属读取他人知识库;盗用他人账号批量发起对话隔离用户数据边界P0内存解析;会话 PG不防注入、DoS限流、审计公开 demo 路由单独降级
3. 业务预处理层长度 + 启发式清洗(自研 · content_validator.py + prompt_security.py + injection_patterns.py;编排 user_input_guard.pyChatService.prepare_completion_turn进 LangGraph 之前regenerate 另校验 DB 末条 userCHAT_MAX_CONTENT_CHARS 截断拒绝;EN/ZH override 短语 → 422 prompt_injection_detected上万字符超长文本淹没顶层 System 规则;中英文标准忽略 / 无视类劫持句式;分段拆分的多层叠加注入;基础全角字母汉字变形攻击;直白诱导泄露配置、系统提示词的恶意提问权威业务层硬拦截P0纯字符串 / 正则变形字符、小语种语义诱导llm-guardllm-injection-guard所有 LLM 对话统一经 prepare_completion_turn
3. 业务预处理层LLM-Guard ML 语义检测llm-guard · llm_guard_adapter.pyuser_input_guard 链尾部(启发式之后)构造 Prompt 前PromptInjection + InvisibleText scanners日韩 / 泰 / 越等小语种注入;同义词改写、转述类诱导越狱;长篇故事叙事式隐性劫持;花体 / 下标 / 数学兼容字符变体绕过正则;无明显关键词纯语义诱导;混合多语言拼接注入载荷补正则固定句式短板P3可选 pip + HF/Torch,延迟↑极简单字符注入(上层已拦)prompt_security、EntropyShield性能敏感可关;高安全环境再开
4. RAG 全链路层RAG 双点清洗(自研 · rag_sanitizer.pyknowledge_ingest_service ETL 入库 ② graphs/nodes/retrieve.py 检索后入向量库前;拼进 <DOCS>NFKC、控制字符、劫持短语 neutralize、RAG_CHUNK_MAX_CHARS知识库文档预埋劫持指令间接注入;文档内嵌零宽隐形字符伪装攻击;超长分片稀释模型安全规则;全角 / Unicode 兼容字符埋毒;文档内伪造系统区块标签污染上下文源头净化 + 漏网二次清洗P1纯字符串完整祈使句法埋毒entropyshieldcontent_provenanceETL + retrieve 同一模块
5. Tools / 外源层来源信任 + EntropyShield(DeSyntax)(自研 content_provenance.py + entropyshield · entropyshield_adapter.pytavily_search.py snippet 出口;crawler-* collection retrieve外源数据进 RAG 链之前worldcup-*TRUSTED_ETL 跳过;WEB_SEARCH / CRAWLERshield_text_for_provenanceentropyshield.shield爬虫、联网搜索结果预埋埋毒指令;外网不可信来源祈使句劫持;多语种外网文档间接注入;完整动词-宾语命令句法链攻击;低信任数据源拼接绕过内层正则过滤外网不可信;打碎资料内可执行句法(见上文 DeSyntax 说明)P1可选 pip;纯算法纯语义类用户输入攻击rag_sanitizerllm-guard仅内网可信库可关 EntropyShield
6. LLM 模型层Prompt 分区隔离(自研模板 · graphs/prompts/rag_chat.pyunified_react.py · call_model.py每次 call_modelSystemMessage调 Ollama / OpenAI 兼容 API 前<POLICY> + <DOCS> + <TOOL_POLICY>;用户仅 HumanMessage用户 / 文档内容伪造 System 分区边界;把 RAG 检索资料、工具返回内容识别为顶层系统指令;多轮对话上下文拼接混淆规则优先级;区块标签篡改引发角色劫持多层漏检时模型侧软兜底P0模板拼接不防输出泄露 system输出扫描、PII 脱敏分区规则不在前端暴露
7. LLM 输出后置层输出扫描 / canary(规划 · llm-guard output scanners + 自研 canary)ChatService finalize 或流式拼接完成点模型生成结束、落库 / SSE 结束前Output scanners;敏感形态正则;XSS 转义(FE)system 泄露、密钥外带、PII、XSS只防输入不够P2可选 HF 模型不防输入侧注入Prompt 分区、audit_log流式宜分段校验
8. 计费管控层Token 日配额(规划 · token_usage 表 + token_quota_service.pychat 流式 finalize单次对话完成统计 usageuser_id + UTC 日聚合 → 429 code=42902恶意构造超长篇对话疯狂消耗 token;单人单日无限次长对话拉高云账单;批量生成高上下文会话耗尽算力额度;低频但单轮超大 token 滥用成本限流管次数,配额管 token 总量P0Redis / PG不防注入、越权网关限流、审计测试 / 生产阈值分离
9. 审计追溯层安全审计日志(规划 · audit_log 表 · rate-limit-audit.mdlogin 失败、demo-turn、敏感删改 handler敏感操作同步写入PG audit_log:user、action、摘要、风险标记事后追溯、合规取证防护失效时复盘攻击路径P0DB IO 轻微仅事后,不实时阻断全模块埋点合规硬性要求
10. CI 离线层Harness 契约测试(pytest · tests/harness/test_chat_security_contract.pyPR / pnpm test:api:harness合并前 CI正例足球问 + 合成注入 → 断言 422 / 200迭代修改清洗规则后防护逻辑退化;接口入参变更绕过注入拦截;业务代码重构导致安全校验分支丢失;基础注入用例回归失效、契约校验漂移比 Garak 轻、PR 必跑P1PG 测试库不覆盖全量载荷Garak、单元测试不能替代运行时多层防护
10. CI 离线层Garak 红队探针garak · scripts/security/garak_probe.sh · pnpm security:garaknightly / 发版前手动非运行时garak_memoryos.yaml;可选 garak_rest_chat.json 打 live API迭代新增功能产生新型注入漏洞;正则 / 语义检测器漏检未知对抗载荷;多层防护组合存在串联绕过漏洞;线上真实流量无法覆盖的边缘攻击样本;OWASP LLM Top10 全类风险自动化扫描线上流量覆盖不了的全量样本P2CI 算力;不占线上不防实时 0dayHarness、Rubric(规划)离线验证,不替代运行时

规划扩展(尚未接入,备忘)

组件预期位置用途优先级
IngressProfile(EP13)BFF + API 共用配置按 Agent / 路由分 Guard 策略,替代粗粒度全局 envP3
Lakera Guard(云 API)API 输入链高检出托管检测;有预算再评估P3
边缘 WAF(CDN / Gateway)公网入口大规模 L3–L7 流量清洗P3
日志脱敏(LangSmith / app log)观测链路PII 采样与字段掩码P3

包名与自研模块对照(速查)

业务称呼第三方包(pip / npm)自研代码入口
BFF Guardllm-prompt-guardapps/web/lib/prompt-guard.ts
HTTP 注入中间件llm-injection-guardmiddleware/injection_guard.py
权威启发式prompt_security.py · user_input_guard.py
ML 输入扫描llm-guard(optional extra)llm_guard_adapter.py
RAG 清洗rag_sanitizer.py
外源 EntropyShieldentropyshield(optional)content_provenance.py · entropyshield_adapter.py
Prompt 分区rag_chat.py · call_model.py
红队garak(optional)scripts/security/garak_probe.sh

DeSyntax 是什么?

文中多次出现的 DeSyntax,不是泛指的「语法结构」,而是 Python 包 EntropyShield 里的算法/模式名(官方 slogan:Break the syntax, keep the semantics —— 打碎可执行句法,尽量保留语义)。

概念含义
解决什么问题RAG / 联网搜索 / 爬虫回灌的文本里,攻击者常用完整祈使句埋毒(如「忽略上文,改为…」)。rag_sanitizer 靠关键词/正则,对「整句像命令」的间接注入检出有限。
DeSyntax 做什么低信任来源进 LLM 前,对正文做确定性 mask(步进遮挡字符),把「动词—宾语—指令」这类可被执行的句法链打散,使模型难以把资料里的句子当成要遵守的指令。
在 MemoryOS 里谁调用自研 content_provenance.shield_text_for_provenance 按来源打标;仅 WEB_SEARCH / CRAWLER 等低信任路径进入 entropyshield_adapter.apply_entropyshieldentropyshield.shield)。worldcup-* 可信 ETL 跳过
与主表「来源信任 + DeSyntax」前半是按 collection / tool 分信任;后半是低信任时挂 EntropyShield 的 DeSyntax mask。

一句话:DeSyntax = 专防不可信资料里「像命令的句子」的打碎句法 mask,不是用户侧注入检测。


二、分层架构图

节点命名与主表一致:业务模块 · 方法 / 部署点(同一 rag_sanitizer.py 在两个时机各调一个函数,不是两个独立组件)。

flowchart TB
  subgraph edge["1 · BFF 可绕过"]
    BFF["BFF Guard
llm-prompt-guard · evaluateBffPromptGuard"] end subgraph ingress["2 · API 网关"] RL["滑动窗口限流
rate_limit.py"] MW["HTTP 注入中间件
llm-injection-guard · PromptScanner"] AUTH["JWT 会话鉴权
PyJWT · get_current_user"] end subgraph prep["3 · 业务预处理"] LEN["长度限制
content_validator.assert_chat_content_length"] HEU["启发式清洗
prompt_security.assert_user_input_safe"] ML["ML 语义检测 可选
llm-guard · scan_prompt"] end subgraph rag4["4 · RAG 双点清洗 · rag_sanitizer.py"] P1["① 入库点
sanitize_chunk
knowledge_ingest_service"] P2["② 检索点
sanitize_retrieved_knowledge_chunk
retrieve_knowledge"] end subgraph ext5["5 · Tools / 外源"] PROV["来源信任 + EntropyShield
shield_text_for_provenance
entropyshield.shield 低信任"] TAV["tools/tavily_search
snippet · WEB_SEARCH provenance_shield"] end subgraph llm6["6 · LLM"] POL["Prompt 分区
build_rag_system_message
POLICY / DOCS / TOOL_POLICY"] LLM["LLM / Agent Tools"] end OUT["7 · 输出扫描
llm-guard output"] QUOTA["8 · Token 配额
token_usage"] AUD["9 · audit_log"] CI["10 · CI 离线
Harness · garak"] BFF --> RL RL --> MW --> AUTH --> prep prep --> P2 P1 -.->|离线 ETL 先入向量库| P2 P2 --> PROV PROV --> POL TAV -.->|Agent 联网回灌| POL POL --> LLM --> OUT LLM --> QUOTA AUTH -.-> AUD CI -.-> prep
flowchart TB
  subgraph edge["1 · BFF 可绕过"]
    BFF["BFF Guard
llm-prompt-guard · evaluateBffPromptGuard"] end subgraph ingress["2 · API 网关"] RL["滑动窗口限流
rate_limit.py"] MW["HTTP 注入中间件
llm-injection-guard · PromptScanner"] AUTH["JWT 会话鉴权
PyJWT · get_current_user"] end subgraph prep["3 · 业务预处理"] LEN["长度限制
content_validator.assert_chat_content_length"] HEU["启发式清洗
prompt_security.assert_user_input_safe"] ML["ML 语义检测 可选
llm-guard · scan_prompt"] end subgraph rag4["4 · RAG 双点清洗 · rag_sanitizer.py"] P1["① 入库点
sanitize_chunk
knowledge_ingest_service"] P2["② 检索点
sanitize_retrieved_knowledge_chunk
retrieve_knowledge"] end subgraph ext5["5 · Tools / 外源"] PROV["来源信任 + EntropyShield
shield_text_for_provenance
entropyshield.shield 低信任"] TAV["tools/tavily_search
snippet · WEB_SEARCH provenance_shield"] end subgraph llm6["6 · LLM"] POL["Prompt 分区
build_rag_system_message
POLICY / DOCS / TOOL_POLICY"] LLM["LLM / Agent Tools"] end OUT["7 · 输出扫描
llm-guard output"] QUOTA["8 · Token 配额
token_usage"] AUD["9 · audit_log"] CI["10 · CI 离线
Harness · garak"] BFF --> RL RL --> MW --> AUTH --> prep prep --> P2 P1 -.->|离线 ETL 先入向量库| P2 P2 --> PROV PROV --> POL TAV -.->|Agent 联网回灌| POL POL --> LLM --> OUT LLM --> QUOTA AUTH -.-> AUD CI -.-> prep
flowchart TB
  subgraph edge["1 · BFF 可绕过"]
    BFF["BFF Guard
llm-prompt-guard · evaluateBffPromptGuard"] end subgraph ingress["2 · API 网关"] RL["滑动窗口限流
rate_limit.py"] MW["HTTP 注入中间件
llm-injection-guard · PromptScanner"] AUTH["JWT 会话鉴权
PyJWT · get_current_user"] end subgraph prep["3 · 业务预处理"] LEN["长度限制
content_validator.assert_chat_content_length"] HEU["启发式清洗
prompt_security.assert_user_input_safe"] ML["ML 语义检测 可选
llm-guard · scan_prompt"] end subgraph rag4["4 · RAG 双点清洗 · rag_sanitizer.py"] P1["① 入库点
sanitize_chunk
knowledge_ingest_service"] P2["② 检索点
sanitize_retrieved_knowledge_chunk
retrieve_knowledge"] end subgraph ext5["5 · Tools / 外源"] PROV["来源信任 + EntropyShield
shield_text_for_provenance
entropyshield.shield 低信任"] TAV["tools/tavily_search
snippet · WEB_SEARCH provenance_shield"] end subgraph llm6["6 · LLM"] POL["Prompt 分区
build_rag_system_message
POLICY / DOCS / TOOL_POLICY"] LLM["LLM / Agent Tools"] end OUT["7 · 输出扫描
llm-guard output"] QUOTA["8 · Token 配额
token_usage"] AUD["9 · audit_log"] CI["10 · CI 离线
Harness · garak"] BFF --> RL RL --> MW --> AUTH --> prep prep --> P2 P1 -.->|离线 ETL 先入向量库| P2 P2 --> PROV PROV --> POL TAV -.->|Agent 联网回灌| POL POL --> LLM --> OUT LLM --> QUOTA AUTH -.-> AUD CI -.-> prep

读图提示

图上节点对应主表行说明
sanitize_chunkRAG 双点清洗ETL 入库;运行时 Chat 不经过此节点
sanitize_retrieved_knowledge_chunkRAG 双点清洗retrieve 后<DOCS> 前;内部会再调 shield_text_for_provenance
content_provenance + entropyshield来源信任 + EntropyShield(DeSyntax)② 与 tools 共用;worldcup-* 可信源跳过 mask
tools/tavily_search来源信任 + EntropyShield(DeSyntax)_format_tavily_response 对每个 snippet 调 shield_text_for_provenance(WEB_SEARCH)

三、时序图

3.1 主路径(Chat · 正常放行)

主路径前提:BFF 未开,或 BFF 开且 mode=tag(仅扫描标记后转发)。quarantine 命中会在 BFF 直接 422,不进 API,见 3.2。

sequenceDiagram
  autonumber
  actor U as 用户
  participant B as BFF prompt-guard.ts
  participant A as FastAPI
  participant M as llm-injection-guard
  participant P as prepare_completion_turn
  participant G as LangGraph
  participant R as retrieve_knowledge
  participant C as call_model
  participant L as LLM

  U->>B: app/api/chat/route.ts
  alt BFF_PROMPT_GUARD_ENABLED=false(默认)
    B->>B: evaluateBffPromptGuard → forward
  else mode=tag
    B->>B: createGuard().sanitize(mode: tag)
    Note over B: 仅标记可疑,原文转发 API
  end
  B->>A: POST /api/v1/chat/completions

  opt 限流(规划 rate_limit.py)
    A->>A: Redis 滑动窗口
  end

  A->>M: InjectionGuardMiddleware
  M->>M: PromptScanner.scan(content)
  M->>A: 放行

  A->>A: get_current_user (PyJWT)
  A->>A: get_owned_conversation

  A->>P: ChatService.prepare_completion_turn
  P->>P: assert_chat_content_length
  P->>P: run_user_input_guards
  Note over P: prompt_security.assert_user_input_safe
  opt LLM_GUARD_ENABLED
    P->>P: llm_guard_adapter.scan_prompt
  end
  P->>G: turn_lock · 落库 user message

  G->>R: retrieve_knowledge
  R->>R: KnowledgeSearchService.search
  R->>R: rag_sanitizer.sanitize_retrieved_knowledge_chunk
  Note over R: shield_text_for_provenance
低信任时 entropyshield.shield G->>C: call_model C->>C: build_rag_system_message
POLICY + DOCS + TOOL_POLICY C->>L: ChatOpenAI.stream L-->>G: token chunks G-->>B: SSE B-->>U: 展示回复 opt finalize(规划 token_usage) G->>G: UsageRecorder 日累计 end
sequenceDiagram
  autonumber
  actor U as 用户
  participant B as BFF prompt-guard.ts
  participant A as FastAPI
  participant M as llm-injection-guard
  participant P as prepare_completion_turn
  participant G as LangGraph
  participant R as retrieve_knowledge
  participant C as call_model
  participant L as LLM

  U->>B: app/api/chat/route.ts
  alt BFF_PROMPT_GUARD_ENABLED=false(默认)
    B->>B: evaluateBffPromptGuard → forward
  else mode=tag
    B->>B: createGuard().sanitize(mode: tag)
    Note over B: 仅标记可疑,原文转发 API
  end
  B->>A: POST /api/v1/chat/completions

  opt 限流(规划 rate_limit.py)
    A->>A: Redis 滑动窗口
  end

  A->>M: InjectionGuardMiddleware
  M->>M: PromptScanner.scan(content)
  M->>A: 放行

  A->>A: get_current_user (PyJWT)
  A->>A: get_owned_conversation

  A->>P: ChatService.prepare_completion_turn
  P->>P: assert_chat_content_length
  P->>P: run_user_input_guards
  Note over P: prompt_security.assert_user_input_safe
  opt LLM_GUARD_ENABLED
    P->>P: llm_guard_adapter.scan_prompt
  end
  P->>G: turn_lock · 落库 user message

  G->>R: retrieve_knowledge
  R->>R: KnowledgeSearchService.search
  R->>R: rag_sanitizer.sanitize_retrieved_knowledge_chunk
  Note over R: shield_text_for_provenance
低信任时 entropyshield.shield G->>C: call_model C->>C: build_rag_system_message
POLICY + DOCS + TOOL_POLICY C->>L: ChatOpenAI.stream L-->>G: token chunks G-->>B: SSE B-->>U: 展示回复 opt finalize(规划 token_usage) G->>G: UsageRecorder 日累计 end
sequenceDiagram
  autonumber
  actor U as 用户
  participant B as BFF prompt-guard.ts
  participant A as FastAPI
  participant M as llm-injection-guard
  participant P as prepare_completion_turn
  participant G as LangGraph
  participant R as retrieve_knowledge
  participant C as call_model
  participant L as LLM

  U->>B: app/api/chat/route.ts
  alt BFF_PROMPT_GUARD_ENABLED=false(默认)
    B->>B: evaluateBffPromptGuard → forward
  else mode=tag
    B->>B: createGuard().sanitize(mode: tag)
    Note over B: 仅标记可疑,原文转发 API
  end
  B->>A: POST /api/v1/chat/completions

  opt 限流(规划 rate_limit.py)
    A->>A: Redis 滑动窗口
  end

  A->>M: InjectionGuardMiddleware
  M->>M: PromptScanner.scan(content)
  M->>A: 放行

  A->>A: get_current_user (PyJWT)
  A->>A: get_owned_conversation

  A->>P: ChatService.prepare_completion_turn
  P->>P: assert_chat_content_length
  P->>P: run_user_input_guards
  Note over P: prompt_security.assert_user_input_safe
  opt LLM_GUARD_ENABLED
    P->>P: llm_guard_adapter.scan_prompt
  end
  P->>G: turn_lock · 落库 user message

  G->>R: retrieve_knowledge
  R->>R: KnowledgeSearchService.search
  R->>R: rag_sanitizer.sanitize_retrieved_knowledge_chunk
  Note over R: shield_text_for_provenance
低信任时 entropyshield.shield G->>C: call_model C->>C: build_rag_system_message
POLICY + DOCS + TOOL_POLICY C->>L: ChatOpenAI.stream L-->>G: token chunks G-->>B: SSE B-->>U: 展示回复 opt finalize(规划 token_usage) G->>G: UsageRecorder 日累计 end

3.2 拒绝路径(注入 · 任一层 422)

与 3.1 同一请求,但在 BFF / HTTP 中间件 / prepare 任一处被拦。三层互斥递进,不是并行扫描。

拦截层主表模块方法响应
BFF(仅 quarantine)BFF Guard · llm-prompt-guardevaluateBffPromptGuardsanitize(mode: block)422 prompt_injection_detected
API 入口HTTP 注入中间件 · llm-injection-guardassert_llm_injection_guard_user_inputPromptScanner.scan422 prompt_injection_detected
业务预处理长度 + 启发式 · 自研assert_chat_content_length / assert_user_input_safe422 content_too_longprompt_injection_detected
业务预处理(可选)LLM-Guard · llm-guardllm_guard_adapter.scan_prompt422 prompt_injection_detected
sequenceDiagram
  autonumber
  actor U as 用户
  participant B as BFF prompt-guard.ts
  participant A as FastAPI
  participant M as llm-injection-guard
  participant P as prepare_completion_turn

  U->>B: app/api/chat/route.ts · 注入载荷

  alt BFF 开 + mode=quarantine
    B->>B: evaluateBffPromptGuard
    B->>B: llm-prompt-guard.sanitize(mode: block)
    Note over B: wasBlocked → reject
    B-->>U: 42201 prompt_injection_detected
    Note over B,A: 不转发 API
  else BFF 关 或 mode=tag
    B->>B: forward(tag 仅标记,不拦截)
    B->>A: POST /api/v1/chat/completions

    A->>M: InjectionGuardMiddleware
    M->>M: llm_injection_guard_adapter
.assert_llm_injection_guard_user_input M->>M: PromptScanner.scan(content) alt PromptScanner 命中 M-->>U: 42201 prompt_injection_detected Note over M,P: 不进 prepare_completion_turn else 中间件放行 A->>A: get_current_user · get_owned_conversation A->>P: ChatService.prepare_completion_turn P->>P: content_validator.assert_chat_content_length alt 超长 P-->>U: 42201 content_too_long else 长度通过 P->>P: user_input_guard.run_user_input_guards P->>P: prompt_security.assert_user_input_safe opt LLM_GUARD_ENABLED P->>P: llm_guard_adapter.scan_prompt end alt 启发式 / ML 命中 P-->>U: 42201 prompt_injection_detected Note over P: 不进 LangGraph end end end end
sequenceDiagram
  autonumber
  actor U as 用户
  participant B as BFF prompt-guard.ts
  participant A as FastAPI
  participant M as llm-injection-guard
  participant P as prepare_completion_turn

  U->>B: app/api/chat/route.ts · 注入载荷

  alt BFF 开 + mode=quarantine
    B->>B: evaluateBffPromptGuard
    B->>B: llm-prompt-guard.sanitize(mode: block)
    Note over B: wasBlocked → reject
    B-->>U: 42201 prompt_injection_detected
    Note over B,A: 不转发 API
  else BFF 关 或 mode=tag
    B->>B: forward(tag 仅标记,不拦截)
    B->>A: POST /api/v1/chat/completions

    A->>M: InjectionGuardMiddleware
    M->>M: llm_injection_guard_adapter
.assert_llm_injection_guard_user_input M->>M: PromptScanner.scan(content) alt PromptScanner 命中 M-->>U: 42201 prompt_injection_detected Note over M,P: 不进 prepare_completion_turn else 中间件放行 A->>A: get_current_user · get_owned_conversation A->>P: ChatService.prepare_completion_turn P->>P: content_validator.assert_chat_content_length alt 超长 P-->>U: 42201 content_too_long else 长度通过 P->>P: user_input_guard.run_user_input_guards P->>P: prompt_security.assert_user_input_safe opt LLM_GUARD_ENABLED P->>P: llm_guard_adapter.scan_prompt end alt 启发式 / ML 命中 P-->>U: 42201 prompt_injection_detected Note over P: 不进 LangGraph end end end end
sequenceDiagram
  autonumber
  actor U as 用户
  participant B as BFF prompt-guard.ts
  participant A as FastAPI
  participant M as llm-injection-guard
  participant P as prepare_completion_turn

  U->>B: app/api/chat/route.ts · 注入载荷

  alt BFF 开 + mode=quarantine
    B->>B: evaluateBffPromptGuard
    B->>B: llm-prompt-guard.sanitize(mode: block)
    Note over B: wasBlocked → reject
    B-->>U: 42201 prompt_injection_detected
    Note over B,A: 不转发 API
  else BFF 关 或 mode=tag
    B->>B: forward(tag 仅标记,不拦截)
    B->>A: POST /api/v1/chat/completions

    A->>M: InjectionGuardMiddleware
    M->>M: llm_injection_guard_adapter
.assert_llm_injection_guard_user_input M->>M: PromptScanner.scan(content) alt PromptScanner 命中 M-->>U: 42201 prompt_injection_detected Note over M,P: 不进 prepare_completion_turn else 中间件放行 A->>A: get_current_user · get_owned_conversation A->>P: ChatService.prepare_completion_turn P->>P: content_validator.assert_chat_content_length alt 超长 P-->>U: 42201 content_too_long else 长度通过 P->>P: user_input_guard.run_user_input_guards P->>P: prompt_security.assert_user_input_safe opt LLM_GUARD_ENABLED P->>P: llm_guard_adapter.scan_prompt end alt 启发式 / ML 命中 P-->>U: 42201 prompt_injection_detected Note over P: 不进 LangGraph end end end end

读图提示mode=tag 时 BFF 不会 422,拒绝发生在 API 侧;直连 API(绕过 BFF)时从第 11 步 InjectionGuardMiddleware 起判。

3.3 离线路径(CI · 非运行时)

与 3.1 / 3.2 热路径无关;对应主表 Harness 契约测试Garak 红队探针 两行。

sequenceDiagram
  autonumber
  participant CI as CI
  participant H as Harness
  participant A as FastAPI test client
  participant G as Garak

  CI->>H: pnpm test:api:harness
  H->>H: test_chat_security_contract.py

  H->>A: test_chat_football_analysis_passes_injection_filter
  Note over A: 正例足球问句,期望 200 SSE
  A-->>H: assert event=start

  H->>A: test_chat_rejects_prompt_injection_before_stream
  Note over A: 反例注入句,期望 42201
  A-->>H: assert messages 未落库

  opt GARAK_PROBE_ENABLED
    CI->>G: pnpm security:garak
    G->>G: garak_probe.sh
    G->>G: garak_memoryos.yaml promptinject
    alt GARAK_TARGET_TYPE rest
      G->>A: garak_rest_chat.json live API
    else mock default
      G->>G: mock target smoke
    end
    Note over G,CI: 报告落盘,exit 0 非阻塞
  end
sequenceDiagram
  autonumber
  participant CI as CI
  participant H as Harness
  participant A as FastAPI test client
  participant G as Garak

  CI->>H: pnpm test:api:harness
  H->>H: test_chat_security_contract.py

  H->>A: test_chat_football_analysis_passes_injection_filter
  Note over A: 正例足球问句,期望 200 SSE
  A-->>H: assert event=start

  H->>A: test_chat_rejects_prompt_injection_before_stream
  Note over A: 反例注入句,期望 42201
  A-->>H: assert messages 未落库

  opt GARAK_PROBE_ENABLED
    CI->>G: pnpm security:garak
    G->>G: garak_probe.sh
    G->>G: garak_memoryos.yaml promptinject
    alt GARAK_TARGET_TYPE rest
      G->>A: garak_rest_chat.json live API
    else mock default
      G->>G: mock target smoke
    end
    Note over G,CI: 报告落盘,exit 0 非阻塞
  end
sequenceDiagram
  autonumber
  participant CI as CI
  participant H as Harness
  participant A as FastAPI test client
  participant G as Garak

  CI->>H: pnpm test:api:harness
  H->>H: test_chat_security_contract.py

  H->>A: test_chat_football_analysis_passes_injection_filter
  Note over A: 正例足球问句,期望 200 SSE
  A-->>H: assert event=start

  H->>A: test_chat_rejects_prompt_injection_before_stream
  Note over A: 反例注入句,期望 42201
  A-->>H: assert messages 未落库

  opt GARAK_PROBE_ENABLED
    CI->>G: pnpm security:garak
    G->>G: garak_probe.sh
    G->>G: garak_memoryos.yaml promptinject
    alt GARAK_TARGET_TYPE rest
      G->>A: garak_rest_chat.json live API
    else mock default
      G->>G: mock target smoke
    end
    Note over G,CI: 报告落盘,exit 0 非阻塞
  end
步骤主表模块包 / 脚本说明
Harness PR 门禁Harness 契约测试test_chat_security_contract.py正例 + 反例;合并前必跑
Garak nightlyGarak 红队探针garak · garak_probe.sh全量探针回归;默认关,失败不挡 PR

四、Chat vs Demo 路径

路径用户自由文本BFF llm-prompt-guardAPI 注入链
Chat可选llm-injection-guard + prompt_security
Demo 分析按钮否(服务端模板)仅参数 / 长度,不扫自由文本

五、落地顺序(从总表提炼)

  1. P0:限流 + Token 配额 + 审计(滥用三角,与注入正交)
  2. P1:RAG 双点清洗;外源链路 ENTROPYSHIELD_ENABLED;Harness PR 门禁
  3. P2:BFF 按需开;输出扫描;Garak live API 模板
  4. P3LLM_GUARD_ENABLED;IngressProfile / 云 Guard / WAF

参考