MCP Agent Attestation

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

攻击模拟结果

攻击模拟套件演示了对以下情况的保护：

检测率：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审查、测试和扩展。

______________________________________________________________________