GUARDRAIL: Security Framework for Large Language Model Applications

GUARDRAIL：大模型应用安全框架

应用安全洋葱模型

概述

应用安全洋葱模型图展示了现代应用面临的分层安全问题，尤其关注较新的 LLM（大语言模型）和基于 Agent 的系统如何构建在传统安全基础之上。

了解各层

如同洋葱一样，应用安全包含多个层面，必须全部加以解决，才能实现全面防护：

第 1 层：传统 Web 安全（最外层）

所有应用安全的基础，涵盖身份验证、授权、输入验证和传输安全等基本问题。无论应用类型如何，这些成熟实践始终不可或缺。

第 2 层：数据与基础设施安全

在基本 Web 安全的基础上，这一层关注数据如何在基础设施组件中存储、处理和传输，包括数据库安全、容器防护、网络分段和依赖项管理。

第 3 层：LLM 应用安全

针对利用大语言模型的应用的特定安全问题。这个较新的领域包括防范提示词注入、越狱尝试、输出清理，以及防止通过模型交互泄露数据。

第 4 层：Agent 与 MCP 安全（核心）

中心层关注针对自主 Agent 和模型上下文协议（MCP）的新兴安全问题，包括消息分类、上下文验证、信任评分，以及 Agent 组件间的流量控制。

应用场景

应用安全洋葱模型可用于：

• 评估不同领域的安全覆盖情况
• 识别安全规划中的缺口
• 根据基础要求确定安全工作的优先级
• 帮助团队了解传统安全问题与新兴安全问题之间的关系
• 制定安全检查清单，确保涵盖所有层面

实施指南

处理应用安全时，请从外向内逐层推进：

1. 确保传统 Web 安全控制措施健全且实施到位
2. 解决数据和基础设施安全问题
3. 根据需要实施 LLM 专用防护措施
4. 在核心层应用 Agent 和 MCP 专用安全控制措施

请记住，内层依赖外层——如果身份验证等传统 Web 安全要素薄弱，LLM 应用安全措施也会受到影响。

贡献

随着 LLM 应用和 Agent 的安全实践日趋成熟，这一模型也在不断演进。欢迎为改进应用安全洋葱模型提出贡献和建议。

---

应用安全洋葱模型图采用 CC BY 4.0 许可。欢迎分享和改编，但须注明出处。

实用安全与框架复杂性：GUARDRAIL 现实检验

1. 隔离与框架复杂性

简单的容器化或 VM 隔离，往往比复杂框架更能立即提升安全性。GUARDRAIL 提供了复杂的协议级安全机制，而通过 Docker 或 VM 实现的基本隔离可以：

• 建立明确的安全边界
• 限制安全事件可能造成的损害
• 无需太多专业知识即可实施
• 以较低的实施风险提供经过验证的保护

现实检验：对许多团队而言，恰当的容器化，加上资源限制和最小化攻击面，比实施 GUARDRAIL 的多层架构更能立即提升安全性。

2. 凭证安全缺口

GUARDRAIL 缺乏明确的凭证安全模式，这是一个重大疏漏。基本的凭证安全实践包括：

• 密钥管理（使用 HashiCorp Vault 或云密钥管理等工具）
• 环境变量隔离
• 凭证轮换
• 最小权限访问模式

……但框架并未明确涵盖这些做法，尽管它们是应用安全的基础。

现实检验：安全地存储 API 密钥，比实现复杂的协议级安全注解更能立竿见影地提升安全性。

3. HTTP/SSE 实现不匹配

GUARDRAIL 与协议无关的方法看似颇具吸引力，但它与 HTTP 和 SSE 实现中的具体安全问题脱节，包括：

• CORS 策略
• 内容安全策略
• HTTP 标头安全
• 跨站脚本防护
• 特定于 HTTP 的身份验证模式

现实检验：如果框架抽象无法与你实际使用的传输机制（HTTP/SSE）衔接，就会造成实现缺口。

4. 身份验证盲点

GUARDRAIL 对身份验证的关注出奇地不足，它更侧重服务间的证明，而非用户身份验证。它缺少：

• 与身份提供商的集成模式
• 令牌验证与管理
• 会话安全
• 授权框架

现实检验：如果没有坚实的身份验证基础，其他安全层的效果会大打折扣。

5. 开发者知识问题

GUARDRAIL 无法解决“缺乏应用级安全背景的新手开发者”这一问题。复杂的框架实际上可能会加剧这个问题，因为它们可能：

• 营造虚假的安全感（“我们实现了 GUARDRAIL，所以我们是安全的”）
• 增加复杂性，掩盖基本安全原则
• 需要专业知识才能正确实现
• 可能因配置错误而引入新的安全缺口

现实检验：相比复杂框架，开发者安全教育以及简单、一致的安全模式往往能带来更好的结果。

实用建议

与其完整实现 GUARDRAIL，不如考虑：

1. 从基础做起：做好隔离、凭证管理、输入验证和输出净化

2. 实施 LLM 专用控制措施：增加针对提示词注入、越狱和数据泄露的专项防护

3. 采用成熟的身份验证模式：利用经过实战检验的身份验证库和框架，而非自行构建解决方案

4. 安全教育：投入资源，提升开发者对 LLM 应用安全的认知

5. 有选择地采用：如果 GUARDRAIL 的某些概念（例如 Information Gateway Layer）看起来有价值，就务实地加以实施，无需引入整个框架的复杂性

结论

GUARDRAIL 为 LLM 安全提供了一种在学术上颇有趣味的方法，但它可能会让通常相当直接的安全挑战变得过于复杂。有时，做好基本隔离、妥善处理凭证并遵循成熟的 HTTP 安全实践，比实施复杂的架构框架更能切实提升安全性。

安全理论与实践之间仍存在巨大差距。与其通过 GUARDRAIL 这类综合框架增加架构复杂性，不如弥补开发者在基础安全知识上的不足，这或许能带来更好的结果。

原始 README：

执行摘要

GUARDRAIL 是一个全面的安全框架，旨在保护大语言模型（LLM）应用生态系统，尤其是使用 Model Context Protocol（MCP）构建的应用。它针对 LLM 应用固有的关键安全漏洞，重点防止数据外泄、数据渗入、未授权访问和资源滥用。GUARDRAIL 提供模块化、分层且可扩展的架构，在不牺牲性能或可用性的前提下提供可靠保护。它优先考虑务实、渐进式的采用方式，让开发者能够逐步增强安全性。

项目状态

GUARDRAIL 目前正在积极开发中。此仓库包含架构设计、技术规格和实现文档。可用于生产环境的代码组件和参考实现将逐步发布。本 README 是了解该项目的中心入口。

目录

• GUARDRAIL：大语言模型应用安全框架
• 应用安全洋葱模型
  • 概述
  • 了解各层
    • 第 1 层：传统 Web 安全（外层）
    • 第 2 层：数据与基础设施安全
    • 第 3 层：大语言模型应用安全
    • 第 4 层：Agent 与 MCP 安全（核心）
  • 应用
  • 实施指南
  • 参与贡献
• 实用安全与框架复杂性：GUARDRAIL 现实检验
  • 1. 隔离与框架复杂性
  • 2. 凭据安全差距
  • 3. HTTP/SSE 实现不匹配
  • 4. 身份验证盲点
  • 5. 开发者知识问题
  • 实用建议
  • 结论
• 原始 README：
  • 执行摘要
  • 项目状态
  • 目录
  • 1. 简介
  • 2. 核心原则
  • 3. 架构概述
    • 3.1 安全层
    • 3.2 部署模式
  • 4. 主要优势
  • 5. 与 MCP 集成
  • 6. 实用安全创新
    • 6.1 可扩展安全中间件（ESM）
    • 6.2 动态安全上下文（DSC）
    • 6.3 协议级安全注解
    • 6.4 轻量级证明协议（LAP）
    • 6.5 自适应资源配额（ARQ）
    • 6.6 安全事件关联与报告（SECR）
  • 7. 可视化与图表
  • 8. 详细文档
  • 9. 应急响应框架
  • 10. 许可证
  • 11. 第 1 阶段实施：基础（0-3 个月）
    • 11.1 执行摘要
    • 11.2 目标
    • 11.3 协议级安全注解（实施细节）
      • 11.3.1 JSON Schema
      • 11.3.2 代码示例（TypeScript/Python）
      • 11.3.3 分类级别
      • 11.3.4 完整性验证
    • 11.4 动态安全上下文（DSC）——初始实施
      • 11.4.1 DSC 类定义（TypeScript/Python）
      • 11.4.2 信任评分算法
      • 11.4.3 安全初始化
      • 11.4.4 代码示例（TypeScript/Python）
    • 11.5 可扩展安全中间件（ESM）——核心模块
      • 11.5.1 ESM 模块接口
      • 11.5.2 核心模块实现
      • 11.5.3 ESM 配置
      • 11.5.4 与 MCP 客户端/服务集成
    • 11.6 时间表与交付成果
    • 11.7 测试策略
    • 11.8 资源需求
    • 11.9 风险缓解
  • 12. 未来增强与路线图
    • 12.1 短期（第 2 阶段 - 3-6 个月）
    • 12.2 中期（第 3 阶段 - 6-12 个月）
    • 12.3 长期（第 4 阶段 - 12 个月以上）
  • 13. 社区与治理
    • 13.1 指导原则
    • 13.2 治理结构
    • 13.3 贡献流程
    • 13.4 沟通渠道
    • 13.5 行为准则

1. 简介

大模型（LLM）正在迅速改变各行各业，但其强大能力与复杂性也带来了重大安全风险。模型上下文协议（MCP）旨在规范大模型应用与服务之间的通信。GUARDRAIL 通过提供专为 MCP 及其他大模型应用生态系统量身打造、全面且可渐进采用的安全框架，应对固有的安全挑战。该框架注重实用性，易于集成，可从简单的增强措施开始，逐步采用更高级的安全措施。

2. 核心原则

GUARDRAIL 基于以下核心原则构建：

1. 信息流控制： 严格的策略管控信息流动，防止未经授权的数据外泄和渗入。这包括数据分类以及对数据访问进行细粒度控制。
2. 上下文安全： 安全决策基于动态执行上下文，包括持续的信任评估。
3. 与传输方式无关的保护： 无论底层传输机制如何，都能提供安全保障。
4. 最小权限执行： 所有组件仅使用最低限度的必要权限，权限按需即时授予，并在不再需要时撤销。
5. 零信任： 不会无条件信任任何组件或用户。必须持续进行验证。

3. 架构概述

3.1 安全层

GUARDRAIL 采用多层架构，每一层都提供不同的安全控制。这些层可以逐步实现，优先从最关键的层开始。

flowchart TB
    subgraph "MCP Client Environment"
        MC[MCP Client]
    end

    subgraph "GUARDRAIL"
        direction TB
        IGL[Information Gateway Layer]
        CVL[Context Verification Layer]
        RCL[Request Control Layer]
        ECL[Execution Containment Layer]
        AML[Audit & Monitoring Layer]

        IGL --> CVL
        CVL --> RCL
        RCL --> ECL
        ECL --> AML
    end

    subgraph "MCP Server Environment"
        MS[MCP Server]
    end

    MC -- "MCP Protocol" --> IGL
    ECL -- "Contained Execution" --> MS

• 信息网关层（IGL）： 管理信息流、内容分类、流策略执行和传输安全。这是初期实现的主要关注点。
• 上下文验证层（CVL）： 通过证明和客户端/服务端验证建立信任。可在 IGL 建立后添加。
• 请求控制层（RCL）： 实现细粒度访问控制、请求过滤和操作限制。在上下文验证的基础上增加一层防御。
• 执行隔离层（ECL）： 确保代码在安全、隔离的边界内执行（例如，使用容器或沙箱）。适用于安全需求更高的高级部署。
• 审计与监控层（AML）： 提供系统操作和安全事件的可见性。对于监控和事件响应至关重要。

3.2 部署模式

GUARDRAIL 可以通过多种配置部署：

1. 嵌入式模式： GUARDRAIL 组件直接集成到宿主进程中（客户端和服务端均如此）。适用于独立应用，或开发/测试环境。

2. 网关模式： GUARDRAIL 作为独立的安全网关运行，对所有 MCP 流量进行中介。这非常适合安全要求严格的企业部署。

3. 服务网格模式： GUARDRAIL 组件作为边车部署在 Kubernetes 环境（或类似环境）中。最适合云原生和微服务架构。

flowchart TB
    subgraph "Deployment Models"
        direction LR

        subgraph "Embedded Model"
            direction TB
            HC[Host Process]

            subgraph "Client Side"
                MCC[MCP Client]
                GC[GUARDRAIL Client Module]

                MCC --> GC
            end

            subgraph "Server Side"
                MCS[MCP Server]
                GS[GUARDRAIL Server Module]

                GS --> MCS
            end

            GC <---> GS
        end

        subgraph "Gateway Model"
            direction TB
            CL[Client]
            GW[GUARDRAIL Gateway]
            SV[Server]

            CL <---> GW
            GW <---> SV
        end

        subgraph "Service Mesh Model"
            direction TB

            subgraph "Client Pod"
                CP[Client Container]
                CS[GUARDRAIL Sidecar]

                CP <---> CS
            end

            subgraph "Server Pod"
                SP[Server Container]
                SS[GUARDRAIL Sidecar]

                SS <---> SP
            end

            subgraph "Control Plane"
                PS[Policy Server]
                IS[Identity Service]
                AS[Audit Collector]
            end

            CS <---> SS
            CS <-..-> PS
            CS <-..-> IS
            CS <-..-> AS
            SS <-..-> PS
            SS <-..-> IS
            SS <-..-> AS
        end
    end

4. 主要优势

GUARDRAIL 提供：

1. 全面的信息流控制： 防止未经授权的数据外泄和渗入。
2. 情境感知安全： 安全措施会根据执行环境进行调整。
3. 兼容性： 可与现有 MCP 实现配合使用。
4. 攻击防护： 防范常见的大模型特有漏洞（例如提示词注入、资源滥用）。
5. 可审计性： 提供信息流的可见性，便于调查和合规。
6. 渐进式采用： 允许组织从基础安全措施着手，并根据需要逐步增加更多防护层。

5. 与 MCP 集成

GUARDRAIL 专为与 MCP 无缝集成而设计。它为 MCP 客户端和服务提供包装类，也可以使用 ESM 在更底层进行接入。

// Client Integration Example
import { Client } from "@modelcontextprotocol/sdk/client";
import { GuardrailClient } from "@guardrail/sdk/client";

// Initialize standard MCP client
const mcpClient = new Client({
  name: "example-client",
  version: "1.0.0"
});

// Wrap with Guardrail protection (Basic)
const client = new GuardrailClient(mcpClient, {
  security: {
    classification_level: "INTERNAL", // Start with basic classification
    // Add more options later as needed.
  }
});

// Normal MCP operations now protected by Guardrail
await client.connect(transport);

6. 实用安全创新

GUARDRAIL 融入了多项关键创新，旨在实现实用部署和渐进式采用。这些创新是构建安全 MCP 生态系统的基石。

6.1 可扩展安全中间件（ESM）

ESM 在 MCP 客户端和服务实现内部提供可插拔架构，支持对 MCP 消息进行定制化安全处理。它位于 MCP 协议层和传输层之间。

flowchart LR
    subgraph "MCP Client/Server"
        direction TB
        MP[MCP Protocol Layer]
        ESM[Extensible Security Middleware]
        TL[Transport Layer]

        MP -- "MCP Message" --> ESM
        ESM -- "Processed Message" --> TL
    end

    subgraph "ESM Internals"
        direction TB
        PM[Plugin Manager]
        subgraph "Security Plugins"
            SP1[Validation Plugin]
            SP2[Classification Plugin]
            SP3[Encryption Plugin]
            SP4[Custom Plugins...]
        end
        PM --> SP1
        PM --> SP2
        PM --> SP3
        PM --> SP4
    end
    MC[MCP Client/Server] -- "Config" --> PM

• 主要特性：

  • 可插拔模块： 安全功能以模块形式实现（例如，validate、transform、preSend、postReceive）。
  • 模块链： 可按可配置的顺序串联模块。
  • 异步运行： 避免阻塞 MCP 主线程。
  • 情境感知： 模块可访问 MCP 上下文和动态安全上下文（DSC）。
  • 策略驱动： 模块行为由声明式策略控制。

• 优势： 灵活性、可扩展性、性能和可测试性。

6.2 动态安全上下文（DSC）

DSC 是一个共享且可变的对象，用于维护与 MCP 连接相关的安全信息。它能够根据观察到的行为实现自适应安全。

sequenceDiagram
    participant Client as MCP Client
    participant Server as MCP Server
    participant DSC as Dynamic Security Context
    participant ESM as Extensible Security Middleware

    Client->>Server: MCP Request
    activate Server
    Server->>DSC: Get Initial DSC (Trust Score, etc.)
    activate DSC
    DSC-->>Server: Initial DSC Data
    deactivate DSC
    Server->>ESM: Process Request (Pre-Processing)
    activate ESM

    loop Security Checks
        ESM->>DSC: Update DSC (e.g., Trust Score)
        activate DSC
        DSC-->>ESM: Updated DSC Data
        deactivate DSC
        ESM->>ESM: Apply Policies based on DSC
    end

    ESM-->>Server: Modified MCP Request
    deactivate ESM

    Server->>Server: Process Request (Application Logic)

    Server->>ESM: Process Response (Post-Processing)
    activate ESM
      loop Security Checks
        ESM->>DSC: Update DSC (e.g., Audit Event)
        activate DSC
        DSC-->>ESM: Updated DSC Data
        deactivate DSC
        ESM->>ESM: Apply Policies based on DSC
    end
    ESM-->>Server: Modified MCP Response

    Server->>Client: MCP Response
    deactivate Server

• 主要特性：

  • 信任评分： 以数值表示可信度，并根据事件进行调整。
  • 威胁级别： 表示风险的类别指标（例如，“低”“中”“高”）。
  • 能力衰减： 可动态限制能力。
  • 会话数据： 安全存储特定于会话的信息。
  • 事件历史： 维护与安全相关的事件历史记录。

• 优势： 自适应安全、零信任基础、精细控制和增强的韧性。

6.3 协议级安全注释

此功能在 MCP 消息结构本身中引入可选安全元数据字段。

{
  "jsonrpc": "2.0",
  "method": "someMethod",
  "params": {
    "data": "..."
  },
  "security": { // OPTIONAL security metadata
    "classification": "INTERNAL", // e.g., PUBLIC, INTERNAL, SENSITIVE
    "integrity": "sha256:...",  // Hash of message content
    "source": "client:123",       // Identifier of sender
    "sequence": 42,              // Sequence number
    "transformations": [        // OPTIONAL array of transformations
        "redacted:pii"
    ]
  },
  "id": 1
}

flowchart LR
    subgraph "MCP Message"
        MSG[Message Content]

        subgraph "Security Field"
            CL[Classification]
            INT[Integrity Hash]
            SRC[Source Identifier]
            SEQ[Sequence Number]
            TR[Transformation Record]
        end
    end

    MSG --- CL
    MSG --- INT
    MSG --- SRC
    MSG --- SEQ
    MSG --- TR

    CL --> PL[Policy Enforcer]
    INT --> IV[Integrity Verifier]
    SRC --> SA[Source Authenticator]
    SEQ --> RA[Replay Attack Detector]
    TR --> TA[Transformation Auditor]

• 主要特性：

  • classification：消息内容的敏感级别。
  • integrity：用于验证消息完整性的加密哈希。
  • source：标识发送方。
  • sequence：防止重放攻击。
  • transformations（可选）：描述对消息应用的转换。

• 优势： 提高透明度、简化安全处理、改善互操作性并实现纵深防御。

6.4 轻量级证明协议（LAP）

LAP 为 MCP 客户端和服务端提供了一种机制，用于验证彼此的身份和环境完整性。它构建于 MCP 之上，并使用自定义消息类型。

sequenceDiagram
    participant Client as "MCP Client"
    participant Server as "MCP Server"

    Client->>Server: initialize Request
    Server-->>Client: initialize Response + attestation_challenge (nonce)

    Client->>Server: attest_client Request {<br>    os: "...",<br>    mcp_sdk_version: "...",<br>    esm_modules: [...],<br>    signature: "...", // Signature over the above data + nonce<br>    nonce: "..." // Server's nonce<br>}
    Server->>Server: Verify client attestation

    Server->>Client: attest_server Response {<br>    os: "...",<br>    mcp_sdk_version: "...",<br>    esm_modules: [...],<br>    signature: "...",<br>    client_nonce: "..." // Client's original nonce (if provided)<br>}

    Client->>Client: Verify server attestation

• 主要功能：

  • 双向证明： 客户端和服务端相互验证。
  • 质询-响应： 使用随机数防止重放攻击。
  • 环境信息： 交换操作系统、SDK 版本和已加载的 ESM 模块信息。
  • 加密签名： 确保证明数据的真实性。
  • 定期重新证明： 可定期执行。
  • 信任评分集成： 结果会影响 DSC 信任评分。

• 优势： 增强信任、缩小攻击面并改善安全态势。

6.5 自适应资源配额（ARQ）

ARQ 允许 MCP 服务根据 DSC，动态调整分配给每个客户端的资源配额（CPU、内存、网络、每秒请求数）。

flowchart LR
    subgraph "MCP Server"
        DSC[Dynamic Security Context]
        RQ[Resource Quota Manager]
        RM[Resource Monitor]
        MC[MCP Client]

        DSC -- "Trust Score & Threat Level" --> RQ
        MC -- "Resource Usage" --> RM
        RM -- "Current Usage" --> RQ
        RQ -- "Quota Limits" --> MC
        MC -- "MCP Requests" --> MS[MCP Server Logic]
        RQ --"Enforce Limits"--> MS
    end

• 主要功能：

  • 基准配额： 为每种资源设置默认配额。
  • 动态调整： 根据 DSC 实时调整配额。
  • 客户端专属配额： 分别跟踪每个客户端的配额。
  • 分级强制执行： 实施节流和速率限制。
  • 向客户端反馈： 告知客户端其配额。

• 优势： 保护资源、实现自适应安全、保障公平性并提升稳定性。

6.6 安全事件关联与报告（SECR）

SECR 基于 MCP 的通知系统构建，形成一个全面的安全事件报告和分析框架。

flowchart TB
    subgraph "MCP Client/Server"
        ESM[Extensible Security Middleware]
        DSC[Dynamic Security Context]
        ARQ[Adaptive Resource Quotas]
        LAP[Lightweight Attestation Protocol]

        ESM -- "Security Events" --> SECR[Security Event Correlation & Reporting]
        DSC -- "Security Events" --> SECR
        ARQ -- "Security Events" --> SECR
        LAP -- "Security Events" --> SECR
    end

    subgraph "External Systems"
        SIEM[SIEM/SOAR]
        DB[(Security Event Database)]
        AA[Alerting & Analytics]
    end

        SECR -- "Filtered Events" --> SIEM
    SECR -- "Aggregated Events" --> DB
    SECR -- "Correlated Events & Alerts" --> AA

• 主要功能：

  • 标准化事件格式： 为安全事件定义通用架构（例如，event.security.authentication.failed）。
  • 事件筛选与路由： 订阅特定事件类型。
  • 事件关联： 识别可疑活动模式。
  • 外部集成： 将事件导出到 SIEM/SOAR 系统。
  • 审计与报告： 集中查看安全事件。

• 优势： 提高可见性、加快事件响应、主动搜寻威胁并生成合规报告。

7. 可视化与图表

以下图表以可视化方式展示了 GUARDRAIL 的架构、部署模型和内部组件：

• 嵌入式部署模型：

• 网关部署模型：

• 服务网格部署模型：

• 网关 - 内部架构：

• 网关 - 19" 机架式设备

• 网关 - 数据流架构：

• 服务网格 - 容器化架构：

• 服务网格 Sidecar - 内部架构：

• 服务网格 - 控制平面架构：

8. 详细文档

虽然本 README 是了解 GUARDRAIL 框架的主要且权威的资料，但也有一些补充文档，记录了不同阶段的头脑风暴、设计迭代和更深入的技术探索。这些文档应视为辅助材料，其中可能包含仍在开发中的想法、已被取代的方案或替代方法。不过，它们有助于了解设计的演进过程，也便于深入探索特定方面。

重要提示： 以下引用的文档反映了 GUARDRAIL 设计过程中的不同阶段。本 README 展示的是当前优先采用且更实用的方法。如果 README 与这些补充文档之间存在差异，应以 README 为准。

以下是补充文档指南：

• 技术规范（2-technical-spec.md）：这是最全面的技术深入文档，展现了 GUARDRAIL 的早期宏大愿景。其中包含各层、组件、配置和算法的详细规范。核心原则仍然适用，但本 README 对实现细节和整体架构进行了简化（尤其是第 6 节）。可参考此文档了解：

  • 详细的组件要求： 各层的细化规范。
  • 配置架构： 所有可配置方面的 JSON 架构。
  • 算法说明： 核心算法的深入解释。
  • 威胁模型： 详细的威胁模型和应对措施。
  • 符合性级别： 不同符合性级别的定义。

• 架构图（6-diags-mermaid.md）：一组 Mermaid 图，用于展示 GUARDRAIL 的各个方面，其中也包括一些较早期、更复杂的设计构想。可参考此文档了解：

  • 替代工作流图： 从不同角度呈现核心工作流。
  • 详细的组件交互： 内部组件关系的细化视图。
  • 部署模型变体： 更详细地展示部署选项。
  • 紧急响应流程： 紧急响应流程的可视化图示。
  • 早期概念探索： 早期设计迭代的图示。

• GUARDRAIL 初始提案（1-GUARDRAIL-mcp-security-claude.md）：本文档是由 Claude 生成的 GUARDRAIL 框架初始提案。它很好地概述了最初的问题陈述和拟议的解决方案，但编写时间早于本 README 所述的完善内容和实用考量。它有助于了解 GUARDRAIL 的最初动机和概念基础。

• 紧急响应框架。本文档详细介绍了紧急响应计划，这是整体安全策略的关键组成部分。请注意，本文档仍使用 SHIELD 命名约定，这是历史遗留。

• 综合与创新文档（各类 *-synthesis-*.md 文件）： 这些文档（例如 7-synthesis-gemini.md、9-new-innovations-diags-claude.md、10-new-innovations-claude.md）记录了中间阶段的头脑风暴和综合工作。它们探索了 GUARDRAIL 的各种创新和改进，整合不同来源的想法并迭代设计。这些文档有助于了解本 README 第 6 节所述“实用安全创新”背后的思考过程。不过，这些创新最终完善后的版本均在本 README 中说明。

• **关键分析文档（15-CRITICAL-ANALYSIS-claude-and-gemini.md）：**本文档对 GUARDRAIL 框架进行了批判性评审，并探讨了保护 MCP 安全的替代方法。文档指出了 GUARDRAIL 的优势与不足，并讨论了不同设计选择所带来的权衡。这有助于理解转向更实用、渐进式方法背后的原因。

• **图表源文件（各种 SVG-*.svg.txt 文件）：**这些文件包含本 README 和 6-diags-mermaid.md 中所含图表的源代码（SVG 格式）。它们主要适用于希望修改或扩展图表的开发者。

• **Mermaid 图表源文件（各种 SHIELD-*.md 文件）：**这些文件是来自之前相关项目的 SHIELD 原始 Mermaid 图表源文件。这些文件早于 6-diags-mermaid.md 和 README.md 中提取并重构的图表，通常应视为历史资料。

这组文档全面呈现了 GUARDRAIL 项目，从最初构想到当前完善后的状态。通过了解设计的演变过程以及不同选择背后的原因，开发者和利益相关者可以更深入地认识该框架及其能力。请始终以本 README 中的信息为准。

9. 应急响应框架

GUARDRAIL 纳入了应急响应框架，提供全面的流程，用于检测、响应安全事件并从中恢复。详情请参阅 应急响应框架。

10. 许可证

本项目采用 MIT License 授权。详情请参阅 LICENSE。

11. 第一阶段实施：基础（0-3 个月）

11.1 执行摘要

GUARDRAIL 第一阶段的重点是在尽量减少对现有实现影响的前提下，在模型上下文协议（MCP）生态系统中建立基准安全水平。本阶段优先采用渐进式采纳，并提供即时安全收益。第一阶段实施的核心组件包括：

1. **协议级安全注释：**为 MCP 消息添加可选元数据，用于分类、完整性校验、来源标识和重放保护。
2. **动态安全上下文（DSC）：**创建共享且可变的上下文，用于在整个 MCP 连接期间跟踪与安全相关的信息。
3. **可扩展安全中间件（ESM）（部分）：**实现基础 ESM 框架，并提供一组有限的核心模块，用于验证、分类、完整性检查、序列检查和信任评分更新。

本阶段为后续所有安全层奠定基础，并让早期采用者能够开始保护其 MCP 实现。

11.2 目标

第一阶段的可衡量目标包括：

1. **实施强制安全注释：**确保所有 MCP 消息都包含 security 字段，以及 classification、integrity、source 和 sequence 属性。
2. **建立基准信任评分：**在 DSC 中实现基本的信任评分机制，至少设置五种会改变信任评分的不同事件。
3. **减少重放攻击：**将成功的重放攻击减少 99%（通过比较第一阶段实施前后检测到的重放尝试次数进行衡量）。
4. **验证消息完整性：**对包含 security 字段的消息，实现 100% 的消息完整性验证。
5. **尽量降低性能开销：**将安全处理导致的额外延迟限制在典型 MCP 消息的 5% 以内。

11.3 协议级安全注释（实现细节）

11.3.1 JSON Schema

以下 JSON schema 定义了 MCP 消息中 security 字段的结构：

{
  "type": "object",
  "properties": {
    "classification": {
      "type": "string",
      "enum": ["PUBLIC", "INTERNAL", "SENSITIVE", "RESTRICTED"]
    },
    "integrity": {
      "type": "string",
      "pattern": "^sha256:[a-f0-9]{64}$"
    },
    "source": {
      "type": "string"
    },
    "sequence": {
      "type": "integer",
      "minimum": 0
    },
    "transformations": {
      "type": "array",
      "items": {
        "type": "string"
      }
    }
  },
  "required": ["classification", "integrity", "source", "sequence"]
}

11.3.2 代码示例（TypeScript/Python）

TypeScript - 在请求中添加 security 字段：

import { createHmac } from 'crypto';

function addSecurityField(message: any, classification: string, source: string, sequence: number, sharedSecret: Buffer): any {
  const messageCopy = { ...message }; // Avoid modifying the original object
    delete messageCopy.security;

  const hmac = createHmac('sha256', sharedSecret);
    hmac.update(JSON.stringify(messageCopy));

    message.security = {
        classification: classification,
        integrity: 'sha256:' + hmac.digest('hex'),
        source: source,
        sequence: sequence,
        transformations: [] // Initialize transformations as empty
    };
    return message;
}

Python - 在请求中添加 security 字段：

import hashlib
import hmac
import json

def add_security_field(message, classification, source, sequence, shared_secret):
    message_copy = message.copy() # Avoid modifying the original
    if "security" in message_copy:
        del message_copy["security"]
    
    message_string = json.dumps(message_copy, sort_keys=True).encode('utf-8') # Consistent encoding
    integrity_hash = hmac.new(shared_secret, message_string, hashlib.sha256).hexdigest()
    message['security'] = {
        'classification': classification,
        'integrity': 'sha256:' + integrity_hash,
        'source': source,
        'sequence': sequence,
        'transformations': []  # Initialize transformations as empty
    }
    return message

TypeScript - 在响应中添加 security 字段：

（响应也可以使用上文中相同的 addSecurityField 函数。）

Python - 在响应中添加 security 字段：

（响应也可以使用上文中相同的 add_security_field 函数。）

TypeScript - 验证 security 字段：

function validateSecurityField(message: any, sharedSecret: Buffer): boolean {
  if (!message.security) {
    return false;
  }

  const { classification, integrity, source, sequence } = message.security;

  if (!classification || !['PUBLIC', 'INTERNAL', 'SENSITIVE', 'RESTRICTED'].includes(classification)) {
    return false;
  }
  if (!integrity || !integrity.startsWith('sha256:')) {
    return false;
  }
  if (!source) {
    return false;
  }
  if (typeof sequence !== 'number' || sequence < 0) {
    return false;
  }
    //Verify integrity
    const messageCopy = { ...message };
    delete messageCopy.security;

    const hmac = createHmac('sha256', sharedSecret);
    hmac.update(JSON.stringify(messageCopy));
    const calculatedIntegrity = 'sha256:' + hmac.digest('hex');

    return integrity === calculatedIntegrity;
}

Python - 验证 security 字段：

def validate_security_field(message, shared_secret):
    if not message.get('security'):
        return False

    security = message['security']
    classification = security.get('classification')
    integrity = security.get('integrity')
    source = security.get('source')
    sequence = security.get('sequence')

    if not classification or classification not in ['PUBLIC', 'INTERNAL', 'SENSITIVE', 'RESTRICTED']:
        return False
    if not integrity or not integrity.startswith('sha256:'):
        return False
    if not source:
        return False
    if not isinstance(sequence, int) or sequence < 0:
        return False
        
    #Verify integrity
    message_copy = message.copy()
    del message_copy["security"]
    message_string = json.dumps(message_copy, sort_keys=True).encode('utf-8')
    calculated_integrity = 'sha256:' + hmac.new(shared_secret, message_string, hashlib.sha256).hexdigest()

    return integrity == calculated_integrity

TypeScript - 处理验证错误：

// ... inside your message handling logic ...
if (!validateSecurityField(receivedMessage, sharedSecret)) {
  console.error("SEC001: Invalid security field."); // Standardized error code
  // Take appropriate action: reject the message, log the error, update DSC, etc.
}

Python - 处理验证错误：

# ... inside your message handling logic ...
if not validate_security_field(received_message, shared_secret):
    print("SEC001: Invalid security field.")  # Standardized error code
    # Take appropriate action: reject the message, log the error, update DSC, etc.

11.3.3 分类级别

第一阶段定义了以下分类级别：

• PUBLIC： 适合公开披露的信息。访问或分发不受限制。
• INTERNAL： 仅限组织及其授权合作伙伴访问的信息。
• SENSITIVE： 因隐私、法律或监管要求而需要额外保护的信息。
• RESTRICTED： 具有严格访问控制的高度敏感信息。披露可能造成重大损害。

11.3.4 完整性验证

integrity 字段使用基于 SHA-256 的 HMAC（基于哈希的消息认证码）。HMAC 使用在 initialize 握手期间生成的共享密钥计算（参见 DSC 安全初始化）。

公式：

integrity = "sha256:" + HMAC-SHA256(sharedSecret, message)

其中，message 是 MCP 消息的 JSON 字符串表示形式，不包含 security 字段本身。sort_keys=True 对于确保不同 JSON 实现中的结果一致至关重要。

11.4 动态安全上下文（DSC）——初始实现

11.4.1 DSC 类定义（TypeScript/Python）

TypeScript：

class DynamicSecurityContext {
  private trustScore: number;
  private threatLevel: 'low' | 'medium' | 'high' | 'critical';
  private capabilities: Set<string>;
  private sessionData: Map<string, any>;
  private eventHistory: any[]; //  Replace 'any' with a proper event type
  private sharedSecret: Buffer;

  constructor(initialTrustScore: number, initialSharedSecret: Buffer) {
    this.trustScore = initialTrustScore;
    this.threatLevel = 'low';
    this.capabilities = new Set();
    this.sessionData = new Map();
    this.eventHistory = [];
    this.sharedSecret = initialSharedSecret;
  }

  getTrustScore(): number {
    return this.trustScore;
  }

  updateTrustScore(adjustment: number, reason: string): void {
    this.trustScore += adjustment;
    this.trustScore = Math.max(0, Math.min(100, this.trustScore)); // Clamp to 0-100

        // Update threat level based on trust score
    if (this.trustScore <= 25) {
      this.threatLevel = 'critical';
    } else if (this.trustScore <= 50) {
      this.threatLevel = 'high';
    } else if (this.trustScore <= 75) {
      this.threatLevel = 'medium';
    } else {
        this.threatLevel = 'low'
    }

    this.logEvent({ type: 'trust_score_update', adjustment, reason, timestamp: Date.now() });
  }

  getThreatLevel(): 'low' | 'medium' | 'high' | 'critical' {
    return this.threatLevel;
  }
    
  getSharedSecret(): Buffer {
      return this.sharedSecret;
  }

  // ... other methods (addCapability, removeCapability, getSessionData, setSessionData, logEvent) ...

    logEvent(event: any): void {
      this.eventHistory.push(event)
    }
}

export { DynamicSecurityContext };

Python：

class DynamicSecurityContext:
    def __init__(self, initial_trust_score, initial_shared_secret):
        self.trust_score = initial_trust_score
        self.threat_level = 'low'
        self.capabilities = set()
        self.session_data = {}
        self.event_history = []
        self.shared_secret = initial_shared_secret

    def get_trust_score(self):
        return self.trust_score

    def update_trust_score(self, adjustment, reason):
        self.trust_score += adjustment
        self.trust_score = max(0, min(100, self.trust_score))  # Clamp to 0-100

        # Update threat level based on trust score
        if self.trust_score <= 25:
            self.threat_level = 'critical'
        elif self.trust_score <= 50:
            self.threat_level = 'high'
        elif self.trust_score <= 75:
            self.threat_level = 'medium'
        else:
            self.threat_level = 'low'

        self.log_event({
            'type': 'trust_score_update',
            'adjustment': adjustment,
            'reason': reason,
            'timestamp': datetime.now().timestamp()
        })

    def get_threat_level(self):
        return self.threat_level

    def get_shared_secret(self):
        return self.shared_secret
        
    # ... other methods (add_capability, remove_capability, get_session_data, set_session_data, log_event) ...
    def log_event(self, event):
        self.event_history.append(event)

11.4.2 信任分数算法

信任分数是一个介于 0 和 100 之间的数值，表示 MCP 客户端或服务的可信度。

算法：

trustScore = initialTrustValue;
trustScore += event.trustImpact * weight; // trustImpact is +ve or -ve
trustScore = Math.max(0, Math.min(100, trustScore)); // Clamp to 0-100

事件及影响示例：

事件	trustImpact	weight
身份验证成功	+5	1.0
签名无效	-10	1.0
超出资源配额	-2	0.5
收到有效消息	+1	0.1
收到无效消息	-5	1.0
序列号不匹配	-8	1.0

威胁级别 威胁级别将根据以下阈值变化：

• low：介于 76 和 100 之间。
• medium：介于 51 和 75 之间。
• high：介于 26 和 50 之间。
• critical：介于 0 和 25 之间。

11.4.3 安全初始化

DSC 的安全初始化（包括交换用于完整性验证的共享密钥）在 MCP initialize 握手期间执行。此示例使用简化的 Diffie-Hellman 密钥交换，仅用于演示。生产实现应使用更健壮且经过充分审查的加密库和协议。

sequenceDiagram
    participant Client
    participant Server

    Client->>Server: initialize Request (with client public key)
    activate Server
    Server->>Server: Generate key pair
    Server->>Server: Create initial DSC (with initial trust score)
    Server-->>Client: initialize Response (with server public key, initial trust score)
    deactivate Server
    activate Client
    Client->>Client: Generate key pair
    Client->>Client: Calculate shared secret
    Client->>Client: Create client-side DSC (with shared secret, initial trust score)
    deactivate Client

简化示例（概念性示例——不可用于生产环境）：

TypeScript：

// --- SERVER SIDE --- (Simplified, conceptual Diffie-Hellman)
import * as crypto from 'crypto';
const connectionContexts = new Map();

function handleInitializeRequest(request: any, transport: any) {
  // 1. Generate an ephemeral key pair.
  const serverKeyPair = crypto.generateKeyPairSync('ec', {
        namedCurve: 'secp256k1', // Use a strong curve
        publicKeyEncoding: { type: 'spki', format: 'pem' },
        privateKeyEncoding: { type: 'pkcs8', format: 'pem' }
    });

  // 2. Create the initial DSC.
  const initialTrustScore = 50;
    const serverSecret = crypto.randomBytes(32); // Generate random secret
  const dsc = new DynamicSecurityContext(initialTrustScore, serverSecret);
    connectionContexts.set(transport, dsc);

  // 3. Include the server's *public* key in the response.
    let initializeResponse: any = {
    jsonrpc: "2.0",
    id: request.id,
    result: {
      serverId: "server-abc",
      publicKey: serverKeyPair.publicKey, // Send the PUBLIC key
      initialTrustScore: initialTrustScore, // Report initial score
    }
  };
    
    // 4. Use the addSecurityField method created earlier
    initializeResponse = addSecurityField(initializeResponse, "INTERNAL", "server:abc", 1, dsc.getSharedSecret());
    
    return initializeResponse
}

// --- CLIENT SIDE --- (Simplified, conceptual Diffie-Hellman)
import * as crypto from 'crypto';
let clientDsc: DynamicSecurityContext;
let clientKeyPair: crypto.KeyPairKeyObjectResult;

function initializeClient() {
    clientKeyPair = crypto.generateKeyPairSync('ec', {
        namedCurve: 'secp256k1',
        publicKeyEncoding: { type: 'spki', format: 'pem' },
        privateKeyEncoding: { type: 'pkcs8', format: 'pem' }
    });
}

function getClientPublicKey() {
    return clientKeyPair.publicKey
}

function setClientDSC(dsc: DynamicSecurityContext) {
    clientDsc = dsc;
}

function getClientSecret() {
    return clientDsc.getSharedSecret();
}

async function handleInitializeResponse(response: any) {
  // 1. Validate the response.
  if (!response.result || !response.result.serverId || !response.result.publicKey) {
    throw new Error("Invalid initialize response");
  }
    
    // 2. Calculate the shared secret. *THIS IS A SIMPLIFIED EXAMPLE*.
  //    In a real implementation, use a proper Diffie-Hellman library.

  const serverPublicKey = crypto.createPublicKey(response.result.publicKey);
  const sharedSecret = crypto.diffieHellman({
    privateKey: clientKeyPair.privateKey,
    publicKey: serverPublicKey,
  });


    // 3.  Create *client-side* DSC.
  const dsc = new DynamicSecurityContext(response.result.initialTrustScore, sharedSecret);

    setClientDSC(dsc);
}

Python：

# --- SERVER SIDE --- (Simplified, conceptual Diffie-Hellman)
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.backends import default_backend
import os
connection_contexts = {}

def handle_initialize_request(request, transport):
    # 1. Generate an ephemeral key pair.
    server_private_key = ec.generate_private_key(
        ec.SECP256K1(), default_backend()
    )
    server_public_key = server_private_key.public_key()
    
    # Serialize the public key
    public_pem = server_public_key.public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )

    # 2. Create the initial DSC.
    initial_trust_score = 50
    server_secret = os.urandom(32) # Generate a random secret
    dsc = DynamicSecurityContext(initial_trust_score, server_secret)
    connection_contexts[transport] = dsc

    # 3. Include the server's *public* key in the response.
    initialize_response = {
        'jsonrpc': '2.0',
        'id': request['id'],
        'result': {
            'serverId': 'server-abc',
            'publicKey': public_pem.decode('utf-8'),  # Send the PUBLIC key
            'initialTrustScore': initial_trust_score,  # Report initial score
        }
    }
    
    initialize_response = add_security_field(initialize_response, "INTERNAL", "server:abc", 1, dsc.get_shared_secret())
    return initialize_response

# --- CLIENT SIDE --- (Simplified, conceptual Diffie-Hellman)
client_dsc = None
client_private_key = None

def initialize_client():
    global client_private_key
    client_private_key = ec.generate_private_key(
        ec.SECP256K1(), default_backend()
    )

def get_client_public_key():
    public_key = client_private_key.public_key()
    public_pem = public_key.public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )
    return public_pem.decode('utf-8')

def set_client_dsc(dsc):
    global client_dsc
    client_dsc = dsc

def get_client_secret():
    return client_dsc.get_shared_secret()

async def handle_initialize_response(response):
    # 1. Validate the response.
    if not response.get('result') or not response['result'].get('serverId') or not response['result'].get('publicKey'):
        raise ValueError("Invalid initialize response")

    # 2. Calculate the shared secret.  *THIS IS A SIMPLIFIED EXAMPLE*.
    server_public_key = serialization.load_pem_public_key(
        response['result']['publicKey'].encode('utf-8'),
        backend=default_backend()
    )

    shared_key = client_private_key.exchange(ec.ECDH(), server_public_key)

    # Derive a key from the shared secret using HKDF
    derived_key = HKDF(
        algorithm=hashes.SHA256(),
        length=32,
        salt=None,
        info=b'handshake data',
        backend=default_backend()
    ).derive(shared_key)

    # 3. Create *client-side* DSC.
    dsc = DynamicSecurityContext(response['result']['initialTrustScore'], derived_key)
    set_client_dsc(dsc)

11.4.4 代码示例（TypeScript/Python）

TypeScript - 访问和修改 DSC：

// Assuming you have a way to get the DSC for the current connection
// (e.g., from a context object or a global map).

function someFunction(context: any) {
  const dsc = context.dsc; // Get the DSC

  const currentTrustScore = dsc.getTrustScore();
  console.log("Current Trust Score:", currentTrustScore);

  dsc.updateTrustScore(-5, "Invalid parameter in request");

  const updatedTrustScore = dsc.getTrustScore();
  console.log("Updated Trust Score:", updatedTrustScore);
    console.log("Current Threat Level:", dsc.getThreatLevel())
}

Python - 访问和修改 DSC：

# Assuming you have a way to get the DSC for the current connection
# (e.g., from a context object or a global map).

def some_function(context):
    dsc = context.dsc  # Get the DSC

    current_trust_score = dsc.get_trust_score()
    print("Current Trust Score:", current_trust_score)

    dsc.update_trust_score(-5, "Invalid parameter in request")

    updated_trust_score = dsc.get_trust_score()
    print("Updated Trust Score:", updated_trust_score)
    print("Current Threat Level:", dsc.get_threat_level())

11.5 可扩展安全中间件（ESM）- 核心模块

11.5.1 ESM 模块接口

TypeScript：

interface ESMModule {
  name: string;
  process(message: any, context: DynamicSecurityContext): Promise<any>;
}

Python：

from abc import ABC, abstractmethod

class ESMModule(ABC):
    @property
    @abstractmethod
    def name(self):
        pass
    
    @abstractmethod
    async def process(self, message, context):
        pass

11.5.2 核心模块实现

TypeScript - 验证器模块：

import { ESMModule } from "./esm-module-interface"; // Adjust path as needed
import { DynamicSecurityContext } from "./dsc"; // Adjust path as needed

class ValidatorModule implements ESMModule {
  name = "Validator";

  async process(message: any, context: DynamicSecurityContext): Promise<any> {
    if (!message.security) {
      context.updateTrustScore(-10, "Missing security field");
      throw new Error("SEC001: Missing security field.");
    }

    // Basic schema validation (using the validateSecurityField function)
        if (!validateSecurityField(message, context.getSharedSecret())) {
            context.updateTrustScore(-10, "Invalid security field");
            throw new Error("SEC002: Invalid security field.");
        }
    return message;
  }
}

Python - 验证器模块：

from esm_module_interface import ESMModule  # Adjust path as needed
from dsc import DynamicSecurityContext  # Adjust path as needed

class ValidatorModule(ESMModule):
    @property
    def name(self):
        return "Validator"

    async def process(self, message, context):
        if not message.get('security'):
            context.update_trust_score(-10, "Missing security field")
            raise ValueError("SEC001: Missing security field.")

        # Basic schema validation (using the validate_security_field function)
        if not validate_security_field(message, context.get_shared_secret()):
            context.update_trust_score(-10, "Invalid security field")
            raise ValueError("SEC002: Invalid security field.")
        return message

TypeScript - 分类器模块（简化版）：

import { ESMModule } from "./esm-module-interface"; // Adjust path as needed
import { DynamicSecurityContext } from "./dsc"; // Adjust path as needed

class ClassifierModule implements ESMModule {
  name = "Classifier";

  async process(message: any, context: DynamicSecurityContext): Promise<any> {
    if (!message.security || !message.security.classification) {
      // Simple keyword-based classification (for demonstration)
      let classification = 'INTERNAL'; // Default
      if (message.params && typeof message.params === 'object') {
        const content = JSON.stringify(message.params).toLowerCase();
        if (content.includes('confidential')) {
          classification = 'RESTRICTED';
        } else if (content.includes('sensitive')) {
          classification = 'SENSITIVE';
        } else if (content.includes('public')) {
          classification = 'PUBLIC';
        }
      }
            // Add a security field, so next modules won't fail.
            message = addSecurityField(message, classification, message.security.source , message.security.sequence, context.getSharedSecret()) // Set a default classification.
    }
    return message;
  }
}

Python - 分类器模块（简化版）：

from esm_module_interface import ESMModule  # Adjust path as needed
from dsc import DynamicSecurityContext # Adjust path as needed
import json

class ClassifierModule(ESMModule):
    @property
    def name(self):
        return "Classifier"

    async def process(self, message, context):
        if not message.get('security') or not message['security'].get('classification'):
            # Simple keyword-based classification (for demonstration)
            classification = 'INTERNAL'  # Default
            if message.get('params') and isinstance(message['params'], dict):
                content = json.dumps(message['params']).lower()
                if 'confidential' in content:
                    classification = 'RESTRICTED'
                elif 'sensitive' in content:
                    classification = 'SENSITIVE'
                elif 'public' in content:
                    classification = 'PUBLIC'

            message = add_security_field(message, classification, message['security']['source'], message['security']['sequence'], context.get_shared_secret())
        return message

TypeScript - IntegrityChecker 模块：

import { ESMModule } from "./esm-module-interface"; // Adjust path as needed
import { DynamicSecurityContext } from "./dsc"; // Adjust path as needed
import { createHmac } from 'crypto';

class IntegrityCheckerModule implements ESMModule {
  name = "IntegrityChecker";

  async process(message: any, context: DynamicSecurityContext): Promise<any> {
        if (!validateSecurityField(message, context.getSharedSecret())) {
            context.updateTrustScore(-20, "Message integrity check failed");
            throw new Error("SEC003: Message integrity check failed.");
        }
    return message;
  }
}

Python - IntegrityChecker 模块：

from esm_module_interface import ESMModule  # Adjust path as needed
from dsc import DynamicSecurityContext  # Adjust path as needed
import hashlib
import hmac
import json

class IntegrityCheckerModule(ESMModule):
    @property
    def name(self):
        return "IntegrityChecker"

    async def process(self, message, context):
        if not validate_security_field(message, context.get_shared_secret()):
            context.update_trust_score(-20, "Message integrity check failed")
            raise ValueError("SEC003: Message integrity check failed.")
        return message

TypeScript - SequenceChecker 模块：

import { ESMModule } from "./esm-module-interface";
import { DynamicSecurityContext } from "./dsc";

class SequenceCheckerModule implements ESMModule {
  name = "SequenceChecker";
  private lastSequenceNumbers: Map<string, number> = new Map(); // Store per-source

  async process(message: any, context: DynamicSecurityContext): Promise<any> {
    const source = message.security.source;
    const sequence = message.security.sequence;
    const lastSequence = this.lastSequenceNumbers.get(source) || 0;

    if (sequence <= lastSequence) {
      context.updateTrustScore(-15, "Potential replay attack detected");
      throw new Error(`SEC004: Potential replay attack detected (source: ${source}, sequence: ${sequence}, last sequence: ${lastSequence}).`);
    }

    this.lastSequenceNumbers.set(source, sequence);
    return message;
  }
}

Python - SequenceChecker 模块：

from esm_module_interface import ESMModule
from dsc import DynamicSecurityContext

class SequenceCheckerModule(ESMModule):
    @property
    def name(self):
      return "SequenceChecker"
    
    def __init__(self):
        self.last_sequence_numbers = {}  # Store per-source

    async def process(self, message, context):
        source = message['security']['source']
        sequence = message['security']['sequence']
        last_sequence = self.last_sequence_numbers.get(source, 0)

        if sequence <= last_sequence:
            context.update_trust_score(-15, "Potential replay attack detected")
            raise ValueError(f"SEC004: Potential replay attack detected (source: {source}, sequence: {sequence}, last sequence: {last_sequence}).")

        self.last_sequence_numbers[source] = sequence
        return message

TypeScript - TrustUpdater 模块：

import { ESMModule } from "./esm-module-interface"; // Adjust path as needed
import { DynamicSecurityContext } from "./dsc"; // Adjust path as needed

class TrustUpdaterModule implements ESMModule {
  name = "TrustUpdater";

  async process(message: any, context: DynamicSecurityContext): Promise<any> {
    // In a real implementation, this module might perform more complex
    // trust calculations based on the results of other modules.
    // For Phase 1, we've already updated the trust score in other modules.
    return message;
  }
}

Python - TrustUpdater 模块：

from esm_module_interface import ESMModule  # Adjust path as needed
from dsc import DynamicSecurityContext  # Adjust path as needed

class TrustUpdaterModule(ESMModule):
    @property
    def name(self):
        return "TrustUpdater"

    async def process(self, message, context):
        # In a real implementation, this module might perform more complex
        # trust calculations based on the results of other modules.
        # For Phase 1, we've already updated the trust score in other modules.
        return message

11.5.3 ESM 配置

ESM 模块按定义的顺序执行。对于阶段 1，使用简单的 JSON 配置：

{
  "esm": {
    "modules": [
      { "name": "Validator", "enabled": true },
      { "name": "Classifier", "enabled": true },
      { "name": "IntegrityChecker", "enabled": true },
      { "name": "SequenceChecker", "enabled": true },
      { "name": "TrustUpdater", "enabled": true }
    ]
  }
}

11.5.4 与 MCP 客户端/服务端集成

（请参阅 README 中现有的集成示例，但现在使用具体的模块实现。）

TypeScript - 客户端集成（简化版）：

import { Client } from '@modelcontextprotocol/sdk/client';
import { ValidatorModule } from './validator-module';
import { ClassifierModule } from './classifier-module';
import { IntegrityCheckerModule } from './integrity-checker-module';
import { SequenceCheckerModule } from './sequence-checker-module';
import { TrustUpdaterModule } from './trust-updater-module';
import { DynamicSecurityContext } from './dsc';
import { addSecurityField, validateSecurityField } from './security-utils'; // Helper functions

// ... other imports ...
const clientInitialSecret = crypto.randomBytes(32);
const clientDsc = new DynamicSecurityContext(50, clientInitialSecret); // Example initial values

const mcpClient = new Client({ /* ... your MCP client options ... */ });

// ESM configuration (could be loaded from a file)
const esmConfig = {
  modules: [
    { name: "Validator", enabled: true },
    { name: "Classifier", enabled: true },
    { name: "IntegrityChecker", enabled: true },
    { name: "SequenceChecker", enabled: true },
    { name: "TrustUpdater", enabled: true }
  ]
};

// Create ESM module instances
const modules = esmConfig.modules
  .filter(m => m.enabled)
  .map(m => {
    switch (m.name) {
      case "Validator": return new ValidatorModule();
      case "Classifier": return new ClassifierModule();
      case "IntegrityChecker": return new IntegrityCheckerModule();
      case "SequenceChecker": return new SequenceCheckerModule();
      case "TrustUpdater": return new TrustUpdaterModule();
      default:
        console.warn(`Unknown ESM module: ${m.name}`);
        return null;
    }
  })
  .filter(m => m !== null);

// Intercept outbound messages
mcpClient.setOutboundInterceptor(async (message) => {
    let sequence = 1; //Initialize sequence
    if (message.security && typeof message.security.sequence === 'number') {
        sequence = message.security.sequence + 1;
    }
  // Add security annotations (using helper function)
  let secureMessage = addSecurityField(message, 'INTERNAL', 'client:123', sequence , clientDsc.getSharedSecret()); // Example

  // Process through ESM modules
  for (const module of modules) {
    try {
      secureMessage = await module.process(secureMessage, clientDsc);
    } catch (error) {
      console.error(`ESM module ${module.name} failed:`, error);
      // Handle the error (e.g., reject the message, log, etc.)
      throw error; // Re-throw to stop processing
    }
  }

  return secureMessage;
});

// Intercept inbound messages
mcpClient.setInboundInterceptor(async (message) => {
    // Process through ESM modules
    let secureMessage = message;
    for (const module of modules) {
        try {
            secureMessage = await module.process(secureMessage, clientDsc);
        } catch (error) {
            console.error(`ESM module ${module.name} failed:`, error.message);
            // Decide how to handle:  Reject?  Continue with reduced trust?
            throw error; //For now, stop processing.
        }
    }

    return secureMessage;
});

Python - 客户端集成（简化版）：

from mcp.client import Client
from validator_module import ValidatorModule
from classifier_module import ClassifierModule
from integrity_checker_module import IntegrityCheckerModule
from sequence_checker_module import SequenceCheckerModule
from trust_updater_module import TrustUpdaterModule
from dsc import DynamicSecurityContext
from security_utils import add_security_field, validate_security_field  # Helper functions
import os
import asyncio

# ... other imports ...

client_initial_secret = os.urandom(32)
client_dsc = DynamicSecurityContext(50, client_initial_secret)  # Example

mcp_client = Client( # your MCP client options
    name="example-client",
    version="1.0.0"
)

# ESM configuration (could be loaded from a file)
esm_config = {
    "modules": [
        {"name": "Validator", "enabled": True},
        {"name": "Classifier", "enabled": True},
        {"name": "IntegrityChecker", "enabled": True},
        {"name": "SequenceChecker", "enabled": True},
        {"name": "TrustUpdater", "enabled": True},
    ]
}

# Create ESM module instances
modules = []
for m in esm_config["modules"]:
    if m["enabled"]:
        if m["name"] == "Validator":
            modules.append(ValidatorModule())
        elif m["name"] == "Classifier":
            modules.append(ClassifierModule())
        elif m["name"] == "IntegrityChecker":
            modules.append(IntegrityCheckerModule())
        elif m["name"] == "SequenceChecker":
            modules.append(SequenceCheckerModule())
        elif m["name"] == "TrustUpdater":
            modules.append(TrustUpdaterModule())
        else:
            print(f"Unknown ESM module: {m['name']}")

# Intercept outbound messages
async def outbound_interceptor(message):
    sequence = 1 # Initialize sequence
    if message.get('security') and isinstance(message['security'].get('sequence'), int):
        sequence = message['security']['sequence'] + 1
    # Add security annotations
    secure_message = add_security_field(message, 'INTERNAL', 'client:123', sequence, client_dsc.get_shared_secret()) # Example

    # Process through ESM modules
    for module in modules:
        try:
            secure_message = await module.process(secure_message, client_dsc)
        except ValueError as e:
            print(f"ESM module {module.name} failed: {e}")
            # Handle error, potentially re-raise
            raise
    return secure_message

# Intercept inbound messages
async def inbound_interceptor(message):
    secure_message = message
    # Process through ESM modules
    for module in modules:
        try:
            secure_message = await module.process(secure_message, client_dsc)
        except ValueError as e:
            print(f"ESM module {module.name} failed: {e}")
            # Handle the error (e.g., reject, log, reduce trust)
            raise # For now, stop processing.
    return secure_message

mcp_client.set_outbound_interceptor(outbound_interceptor)
mcp_client.set_inbound_interceptor(inbound_interceptor)

服务端集成将遵循类似的模式。

11.6 时间线和交付成果

周	任务	交付成果
1-2	设计与原型开发	- DSC 和核心 ESM 模块的详细设计文档。
		- security 字段的 JSON 架构。
		- 初始信任评分算法。
		- 密钥交换原型代码（简化的 Diffie-Hellman）。
3-4	核心实现	- 以下内容的 TypeScript 和 Python 实现：
		- DynamicSecurityContext 类。
		- 核心 ESM 模块（Validator、Classifier、IntegrityChecker,SequenceChecker、TrustUpdater）。
		- 用于添加/验证安全注释的辅助函数。
		- 与 MCP 客户端/服务端的基本集成（拦截器示例）。
5-6	测试与优化	- 所有类和函数的单元测试。
		- ESM 与 MCP 的集成测试。
		- 安全测试（重放攻击模拟、注入无效签名）。
		- 性能基准测试。
		- 根据测试结果优化后的代码。
7-8	文档编写与阶段 2 准备	- 更新后的 README.md，包含阶段 1 的实现详情。
		- 使用阶段 1 组件的开发者文档。
		- 阶段 2 计划（范围、资源、时间线）。

11.7 测试策略

阶段 1 的测试将包括：

• 单元测试： 验证各个类和函数（DSC、ESM 模块、辅助函数）是否正常运行。
• 集成测试： 测试 ESM 与 MCP 客户端/服务端之间的交互，确保消息得到正确处理，且安全注释得到妥善处理。
• 安全测试：
  • 重放攻击模拟： 发送序列号相同的重复消息，以验证重放保护。
  • 注入无效签名： 发送内容已修改或完整性哈希不正确的消息，以验证完整性检查。
  • 缺少安全字段： 发送不含 security 字段的消息，以验证这些消息会被拒绝（或根据配置进行适当处理）。
  • 无效分类： 发送分类值无效的消息。
• 边界条件测试： 使用处于规格限制边界的参数进行测试（超长字符串、超大数字等）。
• 性能基准测试： 测量安全处理所增加的延迟。

11.8 资源需求

• 人员：
  • 1-2 名高级软件工程师（具备安全经验）。
  • 1 名 QA 工程师（负责测试）。
• 工具：
  • TypeScript/Python 开发环境（IDE、调试器）。
  • 测试框架（TypeScript 使用 Jest，Python 使用 pytest）。
  • 加密库（例如，Node.js 中的 crypto、Python 中的 cryptography）。
  • MCP 客户端/服务端 SDK。
  • 版本控制（Git）。
  • CI/CD 流水线（用于自动化测试和构建）。
• 基础设施：
  • 开发用计算机。
  • 用于运行 MCP 和 GUARDRAIL 组件的测试服务。

11.9 风险缓解

风险	缓解策略
性能开销	- 使用高效的算法和数据结构。 - 在 ESM 中实现异步处理。 - 定期进行基准测试，并优化关键代码路径。
安全实现不正确	- 进行全面的代码审查。 - 开展广泛测试（单元测试、集成测试、安全测试）。 - 遵循安全编码最佳实践。
与现有 MCP 的兼容性问题	- 优先采用非破坏性变更（可选的 security 字段）。 - 为现有 MCP 用户提供清晰的迁移指南。 - 使用功能标志启用/禁用 GUARDRAIL 功能。
密钥管理漏洞	- 使用安全的密钥交换协议（例如，稳健实现的 Diffie-Hellman 或类似协议）。 - 切勿将密钥硬编码。 - 在后续阶段集成安全的密钥管理系统（例如，Vault）。
实现复杂性	- 从简化实现开始（仅包含核心 ESM 模块）。 - 将实现拆分为更小且易于管理的任务。 - 使用清晰且一致的编码风格。 - 编写全面的文档。

12. 未来增强功能与路线图

12.1 短期（第 2 阶段 - 3-6 个月）

• 完整的 ESM 实现： 实现完整的 ESM 插件系统，让开发者能够创建和注册自定义安全模块。
• 轻量级证明协议（LAP）： 实现 LAP，以进行客户端/服务端双向身份验证和环境验证。现有 README 中 LAP 部分的链接。
• 基本安全事件关联与报告（SECR）： 实现基本事件日志记录和报告。现有 README 中 SECR 部分的链接。
• 与常见安全工具集成： 提供与常见日志记录和监控工具的初步集成。

12.2 中期（阶段 3 - 6-12 个月）

• 自适应资源配额（ARQ）： 实施 ARQ，根据 DSC 的信任分数动态调整资源限制。现有 README 中 ARQ 部分的链接。
• 全面的审计与合规报告： 增强 SECR，提供更详细的审计日志记录和报告功能，以满足合规要求。
• 企业集成连接器： 开发连接器，将 GUARDRAIL 与企业安全系统（SIEM、IAM）集成。
• 高级威胁检测能力： 探索更复杂的威胁检测技术（例如，使用机器学习进行异常检测）。

12.3 长期（阶段 4 - 12 个月以上）

• 隐私增强技术： 集成差分隐私和同态加密等隐私增强技术。

• 多租户安全隔离： 提供在共享 MCP 环境中隔离不同租户的机制。

• 联邦安全模型： 支持在不同组织或 MCP 部署之间进行安全通信和信任管理。

• 自动化安全测试框架： 开发工具，自动测试 GUARDRAIL 和 MCP 应用的安全性。

• AI 驱动的威胁检测： 集成机器学习，在 AML 中进行实时异常检测，增强主动威胁搜寻能力。

• 动态策略更新： 通过 SECR，支持从外部威胁情报源实时更新策略，提升响应能力。

• 去中心化身份： 将 W3C 去中心化标识符（DID）纳入 CVL，以实现保护隐私的身份验证。

• 抗量子增强： 升级密码学原语（例如，基于格的算法），以抵御未来的量子威胁。

• 12.4. 开源参考实现： 开发并维护 GUARDRAIL 的开源参考实现，以推动采用并鼓励社区贡献。

13. 社区与治理

13.1 指导原则

1. 安全至上： 安全是所有决策中的首要考量。
2. 务实采用： 注重循序渐进地采用，并确保易于使用。
3. 开放治理： 决策透明，并鼓励广泛参与。
4. 技术卓越： 保持高标准的代码质量和安全性。
5. 可持续性： 确保项目长期可行。
6. 互操作性： 保持与不断演进的 MCP 标准兼容。
7. 社区驱动： 积极响应社区需求和贡献。

13.2 治理结构

• 技术指导委员会（TSC）： 提供技术领导，确定架构方向，并批准重大设计决策。
• 工作组： 专注于特定领域（例如，安全层、集成、文档）。
• 维护者： 负责特定组件的日常技术领导工作。
• 贡献者： 任何为项目做出贡献的人。

13.3 贡献流程

1. 创建 GitHub issue，讨论拟议的更改。
2. 提交包含代码、测试和文档的拉取请求。
3. 由维护者审查并批准拉取请求。
4. 将更改合并到主分支。

13.4 沟通渠道

• GitHub： 代码、问题和拉取请求的主要存储库。
• 讨论论坛： 用于讨论和提案。
• 聊天平台：（例如，Discord、Slack）用于实时沟通。
• 邮件列表： 用于发布公告。

13.5 行为准则

所有参与者都应遵守行为准则，共同营造一个相互尊重、包容的社区。