MCP代理认证
模型上下文协议代理的加密身份验证
概述
此项目为实现加密证明扩展 模型上下文协议（MCP） 这使得服务器能够验证连接的AI代理的身份和来源。
问题：MCP处理用户授权（OAuth 2.1），但缺少代理身份验证。服务器无法验证正在连接的模型、部署它的人，或者它的来源链是否完整。
解决方案：基于JWT的认证令牌，带有Ed25519签名，使用MCP与代理一起旅行 experimental 非中断协议扩展的能力字段。
特性
- 🔐 Ed25519签名：快速、安全的加密验证
- 🎫 基于JWT的代币：具有SPIFFE兼容标识符的标准格式
- 🔄 重播保护：基于JTI的缓存阻止令牌重用
- 🏢 企业就绪：SPIFFE ID格式，JWKS密钥分发
- 🛡️ 策略强制：必需/首选/可选认证模式
- 🧪 攻击模拟：演示套件证明了针对常见攻击的安全性
快速开始
安装
# Clone the repository
git clone https://github.com/joelv/mcp-agent-attestation.git
cd mcp-agent-attestation
# Create virtual environment with uv (recommended)
uv venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
# Install with all dependencies
uv pip install -e ".[all]"
# Or with pip
pip install -e ".[all]"运行演示
# Core attestation demo
python -m attestation.core
# Attack simulation suite
python -m attestation.attacks
# Protocol extension demo
python -m attestation.protocol项目结构
mcp-agent-attestation/
├── SPEC.md # Technical specification
├── README.md # This file
├── pyproject.toml # Project configuration
├── src/
│ └── attestation/
│ ├── __init__.py # Package exports
│ ├── core.py # Core attestation primitives
│ ├── protocol.py # MCP protocol extension
│ ├── attacks.py # Attack simulations
│ ├── jwks.py # JWKS HTTP fetcher
│ ├── cache.py # Redis/in-memory replay cache
│ ├── mcp_client.py # MCP SDK client integration
│ └── mcp_server.py # MCP SDK server integration
└── tests/
└── ...用法
创建认证令牌（代理/客户端）
from attestation import (
AttestationProvider,
AgentIdentity,
KeyPair,
)
# In production, Anthropic would run this
keypair = KeyPair.generate("anthropic-2025-01")
provider = AttestationProvider(
issuer="https://api.anthropic.com",
keypair=keypair,
)
# Create identity for the agent
identity = AgentIdentity(
model_family="claude-4",
model_version="claude-sonnet-4-20250514",
provider="anthropic",
)
# Generate token for a specific server
token = provider.create_token(
identity=identity,
audience="https://mcp-server.example.com",
)验证证明（服务器端）
from attestation import (
AttestationVerifier,
InMemoryKeyResolver,
VerificationPolicy,
)
# Setup key resolver (would fetch from JWKS in production)
key_resolver = InMemoryKeyResolver()
key_resolver.add_keypair("https://api.anthropic.com", keypair)
# Create verifier
verifier = AttestationVerifier(
trusted_issuers=["https://api.anthropic.com"],
key_resolver=key_resolver,
policy=VerificationPolicy.REQUIRED,
)
# Verify incoming token
result = await verifier.verify(token)
if result.verified:
print(f"Verified agent: {result.subject}")
print(f"Trust level: {result.trust_level}")
else:
print(f"Verification failed: {result.error}")MCP SDK集成
该库提供与MCP Python SDK的直接集成：
客户端：验证ClientSession
from mcp.client.stdio import stdio_client
from attestation import AttestationProvider, AgentIdentity, KeyPair
from attestation.mcp_client import AttestingClientSession
# Setup attestation
keypair = KeyPair.generate("my-key")
provider = AttestationProvider(issuer="https://api.anthropic.com", keypair=keypair)
identity = AgentIdentity(model_family="claude-4", model_version="claude-sonnet-4", provider="anthropic")
# Connect with attestation
async with stdio_client(server_params) as (read, write):
session = AttestingClientSession(
read_stream=read,
write_stream=write,
attestation_provider=provider,
agent_identity=identity,
target_audience="https://my-mcp-server.com",
)
result = await session.initialize() # Token injected automatically
# Check attestation was verified
if session.attestation_verified:
print(f"Verified with trust level: {session.attestation_status['trust_level']}")服务器端：认证服务器
from mcp.server.lowlevel.server import Server
from attestation import AttestationVerifier, InMemoryKeyResolver, VerificationPolicy
from attestation.mcp_server import AttestingServer, create_attesting_server
# Setup verifier
key_resolver = InMemoryKeyResolver()
key_resolver.add_keypair("https://api.anthropic.com", trusted_keypair)
verifier = AttestationVerifier(
trusted_issuers=["https://api.anthropic.com"],
key_resolver=key_resolver,
policy=VerificationPolicy.REQUIRED,
)
# Create attesting server
attesting_server = create_attesting_server("my-server", verifier, version="1.0.0")
# Register handlers (same as normal MCP server)
@attesting_server.list_tools()
async def list_tools():
return [...]
# Protect specific tools
@attesting_server.call_tool()
@attesting_server.require_attestation(trust_level=TrustLevel.PROVIDER)
async def call_tool(name, arguments):
return [...]低级协议集成
from attestation import (
AttestingAgent,
AttestationMiddleware,
ServerAttestationCapability,
)
# Client side: Attach attestation to initialize request
agent = AttestingAgent(provider=provider, identity=identity)
capabilities = agent.inject_into_capabilities(
{"sampling": {}, "roots": {"listChanged": True}},
audience="https://mcp-server.example.com"
)
# Server side: Verify in middleware
middleware = AttestationMiddleware(
verifier=verifier,
capability=ServerAttestationCapability(
policy="required",
trusted_issuers=["https://api.anthropic.com"]
)
)
result = await middleware.process_initialize(request_params)
if result.should_proceed:
session.attestation = result.context
else:
return result.error_response攻击模拟结果
攻击模拟套件演示了对以下情况的保护：
| 攻击 | 状态 | 防御 |
|---|---|---|
| 模型欺骗 | ✅ 已阻止 | 签名验证 |
| 产地伪造 | ✅ 已阻止 | 受信任的发行者列表 |
| 令牌重播 | ✅ 已阻止 | JTI缓存 |
| 代币篡改 | ✅ 已阻止 | 签名验证 |
| 发卡行打字 | ✅ 已阻止 | 严格匹配发卡行 |
| 降级攻击 | ✅ 已阻止 | 策略执行 |
| 观众不匹配 | ✅ 已阻止 | 受众验证 |
| 安全降级 | ✅ 已阻止 | 已签署的索赔 |
检测率：100%
规格
看 规格.md 完整的技术规范，包括：
- JWT令牌结构和声明
- MCP协议扩展格式
- 密钥管理（JWKS）
- 安全注意事项
- 企业扩展（SPIFFE/IdP）
CLI工具
用于令牌管理的命令行实用程序：
# Generate a key pair
python -m attestation keygen --kid my-key-2025
# Generate an attestation token
python -m attestation generate \
--issuer https://api.anthropic.com \
--audience https://my-server.com \
--model-version claude-sonnet-4 \
--output full
# Inspect a token (without verification)
python -m attestation inspect
# Run attack simulation suite
python -m attestation attack可观测性
内置指标和跟踪支持：
from attestation import get_metrics, trace_verification, AttestationEventHandler
# Access metrics
metrics = get_metrics()
print(metrics.verification_total)
print(metrics.to_prometheus()) # Prometheus format
# Trace operations (OpenTelemetry-compatible)
with trace_verification("https://api.anthropic.com") as span:
result = await verifier.verify(token)
span.set_attribute("verified", result.verified)
# Custom event handlers
class MyHandler(AttestationEventHandler):
def on_replay_detected(self, issuer, jti):
alert_security_team(issuer, jti)
register_event_handler(MyHandler())路线图
- \[x\] 核心证明原语
- \[x\] MCP协议扩展
- \[x\] 攻击模拟套件
- \[x\] MCP SDK集成（认证客户端会话、认证服务器）
- \[x\] 带缓存的JWKS HTTP提取器
- \[x\] Redis支持的回放缓存
- \[x\] CLI工具
- \[x\] 可观察性（度量、跟踪、事件处理程序）
- \[x\] TypeScript实现
- \[\]行为指纹识别（未来研究）
贡献
欢迎投稿和反馈！
许可证
MIT许可证-请参阅 许可证 了解详情。
作者
乔尔·维拉里诺\ 莱斯大学，计算机科学与统计学\ joelavillarino@gmail.com
致谢
初步实施是在Claude（Anthropic）的协助下开发的。 所有代码均由Joel Villarino审查、测试和扩展。
______________________________________________________________________