GUARDRAIL: Security Framework for Large Language Model Applications について
GUARDRAIL は、MCP やエージェントシステムを重視した LLM アプリケーション向けセキュリティフレームワークの設計です。段階的に導入できる多層構成を提案し、現在のリポジトリには設計、仕様、実装ドキュメントが含まれています。
主な機能
- 01多層型アプリケーションセキュリティモデル
- 02情報フロー制御とデータ分類
- 03動的コンテキストと信頼評価
- 04最小権限とゼロトラストの原則
- 05トランスポート方式を問わない保護設計
活用シーン
- 01LLM アプリケーションのセキュリティ範囲の評価
- 02MCP とエージェントシステムの保護要件の整理
- 03セキュリティ対策の段階的な計画
MCP 対応
MCP および他の LLM アプリケーションを対象とし、MCP クライアントやサーバーへのセキュリティ機構の統合を設計しています。現在のリポジトリは設計とドキュメントが中心です。
設定
ローカルサーバーの設定はどのクライアントでも共通です。コピーしてクライアントの MCP 設定ファイルに貼り付けてください。
このサーバーのパッケージやリモートエンドポイントはまだ取得できていません。プロジェクトの README の手順に従ってください。
ツール
ローカルサーバーのツールはマシン上で起動して初めて取得できます。メンテナーはまだツール一覧を提供していません。
詳細
GUARDRAIL:大規模言語モデルアプリケーション向けセキュリティフレームワーク
アプリケーションセキュリティオニオン

概要
アプリケーションセキュリティオニオンの図は、現代のアプリケーションが直面する多層的なセキュリティ上の課題を視覚化したものです。特に、新しいLLM(大規模言語モデル)ベースおよびエージェントベースのシステムが、従来のセキュリティ基盤の上にどのように構築されるかに焦点を当てています。
各レイヤーについて
タマネギのように、包括的な保護を実現するには、アプリケーションセキュリティの複数のレイヤーすべてに対処する必要があります。
レイヤー1:従来のウェブセキュリティ(外側のレイヤー)
すべてのアプリケーションセキュリティの基盤であり、認証、認可、入力検証、トランスポートセキュリティといった基本的な課題を扱います。これらは確立されたプラクティスであり、アプリケーションの種類にかかわらず、今も不可欠です。
レイヤー2:データとインフラストラクチャのセキュリティ
基本的なウェブセキュリティを土台として、このレイヤーでは、インフラストラクチャの各コンポーネント間でデータがどのように保存、処理、転送されるかに対処します。これには、データベースセキュリティ、コンテナ保護、ネットワークセグメンテーション、依存関係管理が含まれます。
レイヤー3:LLMアプリケーションセキュリティ
大規模言語モデルを活用するアプリケーション特有のセキュリティ上の課題です。この新しい分野には、プロンプトインジェクションやジェイルブレイクの試みへの対策、出力のサニタイズ、モデルとのやり取りを通じたデータ漏えいの防止が含まれます。
レイヤー4:エージェントとMCPのセキュリティ(コア)
中心には、自律型エージェントとModel Context Protocol(MCP)に特有の、新たなセキュリティ上の課題があります。これには、メッセージの分類、コンテキストの検証、信頼度のスコアリング、エージェントコンポーネント間のフロー制御が含まれます。
活用方法
アプリケーションセキュリティオニオンは、次の目的に活用できます。
- さまざまな領域におけるセキュリティ対策の網羅状況を評価する
- セキュリティ計画の不足部分を特定する
- 基盤となる要件に基づいてセキュリティ施策の優先順位を決める
- 従来のセキュリティ上の課題と新たな課題の関係について、チームに周知する
- すべてのレイヤーへの対応を確実にするセキュリティチェックリストを作成する
実装ガイダンス
アプリケーションセキュリティに取り組む際は、外側から内側へ進めます。
- 従来のウェブセキュリティ対策が堅牢で、適切に実装されていることを確認します
- データとインフラストラクチャのセキュリティ上の課題に対処します
- 必要に応じて、LLM固有の保護策を実装します
- 中核部分に、エージェントとMCPに固有のセキュリティ対策を適用します
内側のレイヤーは外側のレイヤーに依存することを忘れないでください。認証などの従来のウェブセキュリティ要素が弱ければ、LLMアプリケーションのセキュリティ対策も損なわれます。
コントリビューション
LLMアプリケーションとエージェントのセキュリティプラクティスの成熟に伴い、このモデルも進化しています。アプリケーションセキュリティオニオンの改善に向けたコントリビューションや提案を歓迎します。
アプリケーションセキュリティオニオンの図は、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 では認証が意外なほど十分に扱われておらず、ユーザー認証よりもサービス間のアテステーションに重点が置かれています。次のものが不足しています。
- ID プロバイダーとの統合パターン
- トークンの検証と管理
- セッションセキュリティ
- 認可フレームワーク
現実を見ましょう:認証の基礎がしっかりしていなければ、ほかのセキュリティレイヤーの効果は大幅に低下します。
5. 開発者の知識に関する問題
GUARDRAIL は、「アプリケーションレベルのセキュリティに関する知識がない若手開発者」の問題を解決しません。複雑なフレームワークは、実際には次のような形でこの問題を悪化させる可能性があります。
- 誤った安心感を生み出す(「GUARDRAIL を導入したから安全だ」)
- 基本的なセキュリティ原則を見えにくくする複雑さを加える
- 正しく実装するために専門知識を必要とする
- 設定ミスによって新たなセキュリティ上のギャップを生み出す可能性がある
現実を見ましょう:開発者向けのセキュリティ教育や、シンプルで一貫性のあるセキュリティパターンは、複雑なフレームワークよりも良い成果につながることがよくあります。
実践的な推奨事項
GUARDRAIL を全面的に実装する代わりに、次の点を検討してください。
-
基本から始める:適切な分離、認証情報の管理、入力の検証、出力のサニタイズを行います
-
LLM 固有の制御を実装する:プロンプトインジェクション、ジェイルブレイク、データ漏えいに対する、対象を絞った保護策を追加します
-
実績のある認証パターンを使用する:独自のソリューションを構築するのではなく、十分に検証された認証ライブラリやフレームワークを活用します
-
セキュリティ教育:LLM アプリケーションに特化した開発者向けのセキュリティ意識向上に投資します
-
選択的な導入:Information Gateway Layer など、GUARDRAIL の概念の一部に価値があると思われる場合は、フレームワーク全体の複雑さを持ち込まず、実用的な方法で導入します
結論
GUARDRAIL は LLM セキュリティに対する学術的に興味深いアプローチを提示していますが、多くの場合は単純なセキュリティ上の課題を過度に複雑化する可能性があります。複雑なアーキテクチャフレームワークを実装するよりも、基本的な分離、適切な認証情報の取り扱い、確立された HTTP セキュリティの実践に従うほうが、実際のセキュリティ向上につながる場合があります。
セキュリティ理論と実践の隔たりは依然として大きく、GUARDRAIL のような包括的なフレームワークによってアーキテクチャを複雑化するよりも、開発者の基本的なセキュリティ知識を向上させるほうが、より良い成果につながる可能性が高いでしょう。
元の README:
エグゼクティブサマリー
GUARDRAIL は、特に Model Context Protocol(MCP)を使用して構築されたアプリケーションを中心に、LLM アプリケーションエコシステムを保護するために設計された包括的なセキュリティフレームワークです。LLM アプリケーションに内在する重大なセキュリティ脆弱性に対処し、データの流出、データの侵入、不正アクセス、リソースの悪用を防ぐことに重点を置いています。GUARDRAIL は、パフォーマンスや使いやすさを損なうことなく堅牢な保護を提供する、モジュール式で階層化された拡張可能なアーキテクチャを備えています。実用的かつ段階的な導入を優先しており、開発者はセキュリティを段階的に強化できます。
プロジェクトの状況
GUARDRAIL は現在、活発に開発が進められています。このリポジトリには、アーキテクチャ設計、技術仕様、実装ドキュメントが含まれています。本番環境に対応したコードコンポーネントとリファレンス実装は、段階的に公開される予定です。この README は、プロジェクトを理解するための中心的な情報源です。
目次
- GUARDRAIL:大規模言語モデルアプリケーション向けセキュリティフレームワーク
- アプリケーションセキュリティオニオン
- 実践的なセキュリティとフレームワークの複雑さ:GUARDRAIL の現実確認
- 元の README:
- エグゼクティブサマリー
- プロジェクトの状況
- 目次
- 1. はじめに
- 2. 基本原則
- 3. アーキテクチャの概要
- 4. 主なメリット
- 5. MCP との統合
- 6. 実践的なセキュリティの革新
- 7. 視覚化と図
- 8. 詳細ドキュメント
- 9. 緊急対応フレームワーク
- 10. ライセンス
- 11. フェーズ 1 の実装:基盤(0~3 か月)
- 12. 今後の機能強化とロードマップ
- 13. コミュニティとガバナンス
1. はじめに
大規模言語モデル(LLM)はさまざまな業界を急速に変革していますが、その強力さと複雑さによって重大なセキュリティリスクが生じます。Model Context Protocol(MCP)は、LLM アプリケーションとサービス間の通信を標準化することを目指しています。GUARDRAIL は、MCP およびその他の LLM アプリケーションエコシステムに特化した、包括的で段階的に導入できるセキュリティフレームワークを提供することで、固有のセキュリティ課題に対処します。実用的で統合しやすいように設計されており、シンプルな機能強化から始め、より高度なセキュリティ対策へと進められます。
2. 基本原則
GUARDRAIL は、次の基本原則に基づいて構築されています。
- 情報フロー制御: 厳格なポリシーによって情報の移動を管理し、許可されていないデータの流出や流入を防ぎます。これには、データの分類とデータアクセスのきめ細かな制御が含まれます。
- コンテキストに応じたセキュリティ: 継続的な信頼評価を含め、動的な実行コンテキストに基づいてセキュリティ上の判断を行います。
- トランスポートに依存しない保護: 基盤となるトランスポート機構にかかわらず、セキュリティを保証します。
- 最小権限での実行: すべてのコンポーネントは、必要最小限の権限で動作します。権限は必要な時点で付与され、不要になった時点で取り消されます。
- ゼロトラスト: どのコンポーネントやユーザーも、暗黙に信頼されることはありません。継続的な検証が必要です。
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 は、いくつかの構成でデプロイできます。
-
組み込みモデル: GUARDRAIL のコンポーネントをホストプロセス(クライアントとサーバーの両方)に直接統合します。スタンドアロンアプリケーションや、開発・テスト環境に適しています。
-
ゲートウェイモデル: GUARDRAIL を独立したセキュリティゲートウェイとして動作させ、すべての MCP トラフィックを仲介します。厳格なセキュリティ要件があるエンタープライズ環境に最適です。
-
サービスメッシュモデル: 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 は次の機能を提供します。
- 包括的な情報フロー制御: 不正なデータの流出と侵入を防止します。
- コンテキストに応じたセキュリティ: セキュリティが実行環境に応じて適応します。
- 互換性: 既存の MCP 実装で動作します。
- 攻撃の防止: LLM 固有の一般的な脆弱性(プロンプトインジェクション、リソースの不正使用など)から保護します。
- 監査可能性: 調査やコンプライアンス対応のために、情報フローを可視化します。
- 段階的な導入: 組織は基本的なセキュリティから始め、必要に応じて保護レイヤーを追加できます。
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
-
主な機能:
- 相互アテステーション: クライアントとサーバーの双方が互いを検証します。
- チャレンジレスポンス: リプレイ攻撃を防ぐためにノンスを使用します。
- 環境情報: OS、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" ラックアプライアンス

-
ゲートウェイ - データフローアーキテクチャ:

-
サービスメッシュ - コンテナ化アーキテクチャ:

-
サービスメッシュサイドカー - 内部アーキテクチャ:

-
サービスメッシュ - コントロールプレーンアーキテクチャ:

8. 詳細ドキュメント
この README は GUARDRAIL フレームワークを理解するための主要かつ決定的な情報源ですが、ブレインストーミング、設計の反復、より詳細な技術的検討のさまざまな段階を記録した補足資料もいくつかあります。これらの文書は参考資料として扱ってください。開発中のアイデア、すでに置き換えられたアイデア、または別のアプローチが記載されている場合があります。ただし、設計の変遷を理解し、特定の側面をより深く検討するうえで役立つことがあります。
重要な注意: 以下に挙げる文書は、GUARDRAIL の設計プロセスにおける異なる時点のものです。この README には、現時点で優先される、より実用的なアプローチが反映されています。README とこれらの補足資料に食い違いがある場合は、README を正式な情報源として扱ってください。
補足資料の概要は次のとおりです。
-
技術仕様(2-technical-spec.md):GUARDRAIL の初期の包括的な構想を示す、最も網羅的な技術解説です。すべてのレイヤー、コンポーネント、設定、アルゴリズムについて、詳細な仕様が記載されています。基本原則は現在も有効ですが、README では実装の詳細と全体的なアーキテクチャが簡素化されています(特にセクション 6)。次の内容を確認する際にご利用ください。
- コンポーネントの詳細要件: 各レイヤーの詳細な仕様。
- 設定スキーマ: 設定可能なすべての項目の JSON スキーマ。
- アルゴリズムの説明: 中核となるアルゴリズムの詳細な解説。
- 脅威モデル: 詳細な脅威モデルと対策。
- 適合レベル: 各種適合レベルの定義。
-
アーキテクチャ図(6-diags-mermaid.md):GUARDRAIL のさまざまな側面を視覚化した Mermaid 図を集めたもので、初期のより複雑な設計案も含まれています。次の内容を確認する際にご利用ください。
- 別のワークフロー図: 中核となるワークフローを異なる視点から示した図。
- コンポーネント間の詳細なやり取り: 内部コンポーネント間の関係を詳細に示した図。
- デプロイモデルのバリエーション: デプロイの選択肢をより詳しく示した図。
- 緊急対応フロー: 緊急対応手順を視覚化した図。
- 初期コンセプトの検討: 初期の設計案を示した図。
-
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 には、セキュリティインシデントの検出、対応、復旧のための包括的な手順を定めた緊急対応フレームワークが組み込まれています。詳細は、Emergency Response Framework を参照してください。
10. ライセンス
このプロジェクトは MIT License のもとでライセンスされています。詳細は LICENSE を参照してください。
11. フェーズ 1 の実装:基盤(0-3 か月)
11.1 エグゼクティブサマリー
GUARDRAIL のフェーズ 1 では、既存の実装への影響を最小限に抑えながら、Model Context Protocol(MCP)エコシステムにおける基本的なセキュリティレベルの確立に取り組みます。このフェーズでは、段階的な導入を優先し、即時のセキュリティ上のメリットをもたらします。フェーズ 1 で実装する主要コンポーネントは次のとおりです。
- プロトコルレベルのセキュリティ注釈: 分類、整合性、送信元の識別、リプレイ攻撃対策のための任意のメタデータを MCP メッセージに追加します。
- 動的セキュリティコンテキスト(DSC): MCP 接続全体を通じてセキュリティ関連情報を追跡するための、共有可能で変更可能なコンテキストを作成します。
- 拡張可能なセキュリティミドルウェア(ESM)(一部): 検証、分類、整合性チェック、シーケンスチェック、信頼スコアの更新を行う中核モジュールを限定的に備えた、基本的な ESM フレームワークを実装します。
このフェーズは、後続のすべてのセキュリティレイヤーの基盤となり、早期導入者が MCP の実装のセキュリティ保護を開始できるようにします。
11.2 目標
フェーズ 1 の測定可能な目標は次のとおりです。
- 必須セキュリティ注釈の実装: すべての MCP メッセージに
classification、integrity、source、sequence属性を含むsecurityフィールドがあることを保証します。 - 基本的な信頼スコアの確立: DSC 内に基本的な信頼スコアの仕組みを実装し、信頼スコアを変更するイベントを少なくとも 5 種類設けます。
- リプレイ攻撃の削減: フェーズ 1 の実装前後で検出されたリプレイ試行の数を比較し、成功するリプレイ攻撃を 99% 減らします。
- メッセージ整合性の検証:
securityフィールドを持つメッセージについて、メッセージ整合性の検証率 100% を達成します。 - パフォーマンスへの影響を最小限に抑制: 一般的な MCP メッセージに対するセキュリティ処理による追加レイテンシを 5% 以下に抑えます。
11.3 プロトコルレベルのセキュリティアノテーション(実装の詳細)
11.3.1 JSON Schema
次の JSON スキーマは、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 分類レベル
Phase 1 では、次の分類レベルを定義します。
- PUBLIC: 公開しても問題のない情報です。アクセスや配布に制限はありません。
- INTERNAL: 組織および認可されたパートナーに限定される情報です。
- SENSITIVE: プライバシー、法律、または規制上の要件により、追加の保護が必要な情報です。
- RESTRICTED: 厳格なアクセス制御が必要な、極めて機密性の高い情報です。開示されると、重大な損害を引き起こすおそれがあります。
11.3.4 完全性の検証
integrity フィールドでは、SHA-256 を使用する HMAC(ハッシュベースのメッセージ認証コード)を使用します。HMAC は、initialize ハンドシェイク中に導出された共有秘密を使用して計算されます(DSC のセキュアな初期化を参照)。
式:
integrity = "sha256:" + HMAC-SHA256(sharedSecret, message)
ここで、message は、security フィールド自体を除外した MCP メッセージの JSON 文字列表現です。異なる JSON 実装間で一貫した結果を得るには、sort_keys=True が不可欠です。
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 信頼スコアアルゴリズム
信頼スコアは、MCP クライアントまたはサーバーの信頼性を表す、0 から 100 までの数値です。
アルゴリズム:
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 - Validatorモジュール:
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 - Validatorモジュール:
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 - Classifierモジュール(簡略版):
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 - Classifierモジュール(簡略版):
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モジュールは定義された順序で実行されます。Phase 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 の準備 | - フェーズ 1 の実装詳細を反映した更新済み README.md。 |
| - フェーズ 1 のコンポーネントの使用に関する開発者向けドキュメント。 | ||
| - フェーズ 2 の計画(範囲、リソース、タイムライン)。 |
11.7 テスト戦略
フェーズ 1 のテストには、次の内容が含まれます。
- ユニットテスト: 個々のクラスと関数(DSC、ESM モジュール、ヘルパー関数)が正しく動作することを検証します。
- 統合テスト: ESM と MCP クライアント/サーバー間のやり取りをテストし、メッセージが正しく処理され、セキュリティ注釈が適切に扱われることを確認します。
- セキュリティテスト:
- リプレイ攻撃のシミュレーション: 同じシーケンス番号を持つ重複メッセージを送信し、リプレイ攻撃対策を検証します。
- 不正な署名の挿入: 内容を変更したメッセージや、不正な整合性ハッシュを含むメッセージを送信し、整合性チェックを検証します。
- セキュリティフィールドの欠落:
securityフィールドのないメッセージを送信し、拒否される(または設定に応じて適切に処理される)ことを検証します。 - 不正な分類: 不正な分類値を含むメッセージを送信します。
- 境界条件テスト: 仕様上の限界にあるパラメーター(非常に長い文字列、非常に大きな数値など)を使ってテストします。
- パフォーマンスベンチマーク: セキュリティ処理によって追加されるレイテンシを測定します。
11.8 リソース要件
- 人員:
- セキュリティ経験を持つシニアソフトウェアエンジニア 1~2 名。
- テスト担当の QA エンジニア 1 名。
- ツール:
- 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): DSC の信頼スコアに基づいてリソース制限を動的に調整する ARQ を実装します。既存の README の ARQ セクションへのリンク。
- 包括的な監査およびコンプライアンスレポート: コンプライアンス要件に対応できるよう、より詳細な監査ログとレポート機能で SECR を強化します。
- エンタープライズ統合コネクター: GUARDRAIL をエンタープライズセキュリティシステム(SIEM、IAM)と統合するためのコネクターを開発します。
- 高度な脅威検出機能: より高度な脅威検出手法(機械学習を用いた異常検知など)を検討します。
12.3 長期(フェーズ 4 - 12 か月以上)
-
プライバシー強化技術: 差分プライバシーや準同型暗号などのプライバシー強化技術を統合します。
-
マルチテナントのセキュリティ分離: 共有 MCP 環境内で異なるテナントを分離する仕組みを提供します。
-
連合型セキュリティモデル: 異なる組織間または MCP デプロイメント間で、安全な通信と信頼管理を可能にします。
-
自動セキュリティテストフレームワーク: GUARDRAIL および MCP アプリケーションのセキュリティを自動的にテストするツールを開発します。
-
AI による脅威検出: AML 内でリアルタイムの異常検知を行う機械学習を統合し、プロアクティブな脅威ハンティングを強化します。
-
動的なポリシー更新: SECR を介して外部の脅威インテリジェンスフィードからリアルタイムでポリシーを更新できるようにし、対応力を高めます。
-
分散型 ID: プライバシーを保護する認証のため、W3C 分散型識別子(DID)を CVL に組み込みます。
-
耐量子性の強化: 将来の量子脅威に備えて、暗号プリミティブ(格子ベースのアルゴリズムなど)をアップグレードします。
-
12.4. オープンソースのリファレンス実装: GUARDRAIL のオープンソースのリファレンス実装を開発・維持し、導入とコミュニティからの貢献を促進します。
13. コミュニティとガバナンス
13.1 指針
- セキュリティ第一: あらゆる意思決定において、セキュリティを最優先します。
- 実践的な導入: 段階的な導入と使いやすさを重視します。
- オープンなガバナンス: 透明性のある意思決定と、誰もが参加できる体制を実現します。
- 技術的卓越性: コード品質とセキュリティの高い基準を維持します。
- 持続可能性: プロジェクトの長期的な存続を確保します。
- 相互運用性: 進化する MCP 標準との互換性を維持します。
- コミュニティ主導: コミュニティのニーズと貢献に応えます。
13.2 ガバナンス体制
- 技術運営委員会(TSC): 技術的なリーダーシップを担い、アーキテクチャの方向性を定め、主要な設計上の決定を承認します。
- ワーキンググループ: 特定の分野(セキュリティレイヤー、統合、ドキュメントなど)に取り組みます。
- メンテナー: 特定のコンポーネントの日々の技術的なリーダーシップを担います。
- コントリビューター: プロジェクトに貢献するすべての人を指します。
13.3 貢献プロセス
- 提案する変更について話し合うため、GitHub のイシューを作成します。
- コード、テスト、ドキュメントを含むプルリクエストを送信します。
- メンテナーがプルリクエストをレビューして承認します。
- 変更がメインブランチにマージされます。
13.4 コミュニケーションチャネル
- GitHub: コード、イシュー、プルリクエストを管理する主要なリポジトリです。
- ディスカッションフォーラム: 議論や提案に使用します。
- チャットプラットフォーム:(Discord、Slack など)リアルタイムのコミュニケーションに使用します。
- メーリングリスト: お知らせに使用します。
13.5 行動規範
すべての参加者は、敬意と包摂性を重視するコミュニティを促進する行動規範を遵守することが求められます。
コメント
AI アプリケーションが外部のツールやデータソースに統一的な方法で接続するためのオープンなプロトコルです。クライアントはモデルとの対話を担い、MCP サーバーはツール・リソース・プロンプトテンプレートを提供します。標準メッセージでやり取りするため、1 つのサーバーを複数のクライアントで再利用できます。
