MCP PKI 认证系统
使用ed25519公钥基础设施的模型上下文协议(MCP)互认证系统,实现MCP服务器与客户端之间的双向身份验证。
目录
概述
MCP PKI认证系统使用ed25519数字签名,为MCP连接提供安全的双向认证。它实现了一个包含4条消息的挑战-响应协议,在建立通信之前验证客户端和服务器双方的身份。
主要特点
- 双向认证客户端和服务器都相互验证对方的身份
- ed25519 密码学现代、快速的椭圆曲线签名
- 重放保护时间戳和随机数验证可防止重放攻击
- 白名单管理通过公钥指纹实现细粒度访问控制
- 高性能低于10毫秒的认证开销,无状态服务器设计
- 审计日志记录全面的身份验证事件追踪
- 多种运输方式支持HTTP/HTTPS和WebSocket
- Docker 准备就绪容器化测试与部署
快速入门
1. 生成密钥对
# Generate server key pair
mcp-keygen --output-dir ./keys --key-name server
# Generate client key pair
mcp-keygen --output-dir ./keys --key-name client2. 设置允许列表
# Add client's public key to server allowlist
mcp-allowlist add --config server_config.yml \
--key-file ./keys/client_public.pem \
--metadata '{"role": "trusted_client", "org": "example_org"}'
# Add server's public key to client allowlist
mcp-allowlist add --config client_config.yml \
--key-file ./keys/server_public.pem \
--metadata '{"role": "mcp_server", "endpoint": "api.example.com"}'3. 配置身份验证
创造 server_config.yml:
keys:
private_key_path: "./keys/server_private.pem"
public_key_path: "./keys/server_public.pem"
acl:
allowlist_path: "./keys/server_allowlist.json"
auth:
timestamp_tolerance: 300 # 5 minutes
nonce_cache_size: 10000
audit:
enabled: true
log_file: "./logs/server_audit.jsonl"
log_level: "INFO"
transport:
type: "http"
host: "0.0.0.0"
port: 8443
ssl:
cert_file: "./certs/server.crt"
key_file: "./certs/server.key"4. 测试认证
# Start server (in one terminal)
python -m mcp_pki_auth.examples.server --config server_config.yml
# Test from client (in another terminal)
python -m mcp_pki_auth.examples.client \
--config client_config.yml \
--server-url https://localhost:8443/auth建筑学
该系统由四个核心组件构成:
┌─────────────────┐ ┌─────────────────┐
│ MCP Client │ │ MCP Server │
│ │ │ │
│ ┌─────────────┐ │ │ ┌─────────────┐ │
│ │ Key Manager │ │ │ │ Key Manager │ │
│ └─────────────┘ │ │ └─────────────┘ │
│ ┌─────────────┐ │ │ ┌─────────────┐ │
│ │ ACL Manager │ │ │ │ ACL Manager │ │
│ └─────────────┘ │ │ └─────────────┘ │
│ ┌─────────────┐ │ │ ┌─────────────┐ │
│ │ Auth Engine│ │◄──►│ │ Auth Engine │ │
│ └─────────────┘ │ │ └─────────────┘ │
│ ┌─────────────┐ │ │ ┌─────────────┐ │
│ │ Transport │ │ │ │ Transport │ │
│ └─────────────┘ │ │ └─────────────┘ │
└─────────────────┘ └─────────────────┘组件
密钥管理器处理ed25519密钥对生成、PEM格式存储以及SHA-256指纹计算。
ACL管理器使用基于指纹的索引维护允许列表,以实现O(1)复杂度的元数据支持查找。
认证引擎执行包含签名验证、时间戳验证(±5分钟)和一次性随机数重放保护的4消息协议。
传输层提供HTTP/HTTPS和WebSocket连接,支持连接池、重试以及SSL/TLS。
协议流程
Client Server
│ │
│ 1. AuthRequest │
│ ─────────────────────► │
│ {public_key} │
│ │
│ 2. AuthChallenge │
│ ◄───────────────────── │
│ {challenge, sig} │
│ │
│ 3. AuthResponse │
│ ─────────────────────► │
│ {response, counter} │
│ │
│ 4. AuthComplete │
│ ◄───────────────────── │
│ {success, sig} │
│ │安装
要求
- Python 3.8及以上版本
- Docker(用于容器化测试)
安装方法
来自来源
git clone https://github.com/your-org/mcp-sec.git
cd mcp-sec
pip install -e .使用 Docker
git clone https://github.com/your-org/mcp-sec.git
cd mcp-sec
make setup-test-env
make build依赖项
核心依赖项会自动安装:
cryptography>=41.0.0- ed25519 密码运算pyyaml>=6.0- 配置文件解析click>=8.0.0- CLI(命令行界面)框架websockets>=11.0- 支持WebSocket传输tabulate>=0.9.0- CLI 表格格式化
配置
配置文件结构
# Key management
keys:
private_key_path: "./keys/private.pem"
public_key_path: "./keys/public.pem"
auto_generate: false # Generate keys if not found
# Access control
acl:
allowlist_path: "./keys/allowlist.json"
auto_create: true # Create empty allowlist if not found
metadata_required: false # Require metadata for all entries
# Authentication settings
auth:
timestamp_tolerance: 300 # seconds (±5 minutes)
nonce_cache_size: 10000 # Max cached nonces
nonce_cache_ttl: 3600 # Nonce TTL in seconds
# Audit logging
audit:
enabled: true
log_file: "./logs/audit.jsonl" # Use null for stdout
log_level: "INFO" # DEBUG, INFO, WARN, ERROR
max_file_size: "10MB"
backup_count: 5
include_performance: true
filter_events: ["auth_success", "auth_failure", "key_validation"]
# Transport configuration
transport:
type: "http" # "http", "https", "ws", "wss"
host: "localhost"
port: 8443
timeout: 30
max_connections: 100
retry_attempts: 3
retry_delay: 1.0
# SSL/TLS settings (for https/wss)
ssl:
cert_file: "./certs/server.crt"
key_file: "./certs/server.key"
ca_file: "./certs/ca.pem" # Optional CA bundle
verify_mode: "required" # "none", "optional", "required"环境变量
配置值可以通过环境变量进行覆盖:
# Key paths
export MCP_PRIVATE_KEY_PATH="./keys/private.pem"
export MCP_PUBLIC_KEY_PATH="./keys/public.pem"
# ACL settings
export MCP_ALLOWLIST_PATH="./keys/allowlist.json"
# Auth settings
export MCP_TIMESTAMP_TOLERANCE=300
export MCP_NONCE_CACHE_SIZE=10000
# Audit settings
export MCP_AUDIT_ENABLED=true
export MCP_AUDIT_LOG_FILE="./logs/audit.jsonl"
export MCP_AUDIT_LOG_LEVEL="INFO"
# Transport settings
export MCP_TRANSPORT_TYPE="https"
export MCP_TRANSPORT_HOST="0.0.0.0"
export MCP_TRANSPORT_PORT=8443CLI 使用方法
密钥管理
生成密钥对
# Basic key generation
mcp-keygen --output-dir ./keys --key-name mykey
# Generate with custom parameters
mcp-keygen \
--output-dir ./keys \
--key-name server \
--comment "Production server key - 2024" \
--format json # Output key info as JSON
# Show fingerprint of existing key
mcp-keygen --show-fingerprint ./keys/server_public.pem提取公钥
# Extract public key from private key
mcp-keygen --extract-public \
--private-key ./keys/server_private.pem \
--output-file ./keys/server_public_extracted.pem白名单管理
添加密钥
# Add key with metadata
mcp-allowlist add \
--config server_config.yml \
--key-file ./keys/client_public.pem \
--metadata '{"role": "api_client", "org": "example", "expires": "2024-12-31"}'
# Add key by fingerprint
mcp-allowlist add \
--config server_config.yml \
--fingerprint "a1b2c3d4..." \
--metadata '{"role": "backup_server"}'列出密钥
# List all keys
mcp-allowlist list --config server_config.yml
# List with specific format
mcp-allowlist list \
--config server_config.yml \
--format table \
--show-metadata
# Filter by metadata
mcp-allowlist list \
--config server_config.yml \
--filter-role "api_client"移除钥匙
# Remove by fingerprint
mcp-allowlist remove \
--config server_config.yml \
--fingerprint "a1b2c3d4..."
# Remove multiple keys
mcp-allowlist remove \
--config server_config.yml \
--fingerprint "a1b2c3d4..." "e5f6g7h8..."进口/出口
# Export allowlist
mcp-allowlist export \
--config server_config.yml \
--output ./backups/allowlist_backup.json
# Import allowlist (merges with existing)
mcp-allowlist import \
--config server_config.yml \
--input ./backups/allowlist_backup.json \
--merge
# Show statistics
mcp-allowlist stats --config server_config.yml测试
该系统包含可在Docker中运行的综合测试套件:
# Run all tests
make test
# Run specific test categories
make test-unit # Unit tests only
make test-integration # Integration tests only手动测试
您可以使用Python API手动测试身份验证流程:
# See the integration tests in tests/integration/test_auth_flow.py
# for complete examples of testing authentication flowsAPI 参考文档
核心类
密钥管理器
from mcp_pki_auth.key_manager import KeyManager
# Initialize
key_mgr = KeyManager()
# Generate new key pair
private_key, public_key = key_mgr.generate_key_pair()
# Save keys in PEM format
key_mgr.save_private_key(private_key, "./keys/private.pem")
key_mgr.save_public_key(public_key, "./keys/public.pem")
# Load existing keys
private_key = key_mgr.load_private_key("./keys/private.pem")
public_key = key_mgr.load_public_key("./keys/public.pem")
# Calculate fingerprint
fingerprint = key_mgr.get_fingerprint(public_key)
# Sign data
signature = key_mgr.sign_data(private_key, b"message")
# Verify signature
is_valid = key_mgr.verify_signature(public_key, b"message", signature)ACLManager(可翻译为“访问控制列表管理器”)
from mcp_pki_auth.acl_manager import ACLManager
# Initialize
acl_mgr = ACLManager("./keys/allowlist.json")
# Add key to allowlist
metadata = {"role": "client", "org": "example"}
acl_mgr.add_key(public_key, metadata)
# Check if key is allowed
is_allowed, metadata = acl_mgr.is_key_allowed(public_key)
# Remove key
acl_mgr.remove_key(public_key)
# List all keys
keys_info = acl_mgr.list_keys()认证引擎
from mcp_pki_auth.auth_engine import AuthenticationEngine
from mcp_pki_auth.config import Config
# Initialize
config = Config.load_from_file("config.yml")
auth_engine = AuthenticationEngine(config, key_manager, acl_manager)
# Client-side authentication
async def client_auth():
# Step 1: Create auth request
request = auth_engine.create_auth_request()
# Step 3: Process server challenge
response = auth_engine.process_challenge(challenge_msg)
# Verify final response
success = auth_engine.verify_auth_complete(complete_msg)
return success
# Server-side authentication
async def server_auth():
# Step 2: Process client request and create challenge
challenge = auth_engine.process_auth_request(request_msg)
# Step 4: Process response and complete auth
result = auth_engine.process_auth_response(response_msg)
return result传输层
HTTP传输
from mcp_pki_auth.transport import HTTPTransport
# Client
transport = HTTPTransport(
base_url="https://api.example.com:8443",
ssl_cert_file="./certs/client.crt",
ssl_key_file="./certs/client.key",
timeout=30,
retry_attempts=3
)
# Send authentication request
response = await transport.send_auth_message(auth_request)
# Server
server_transport = HTTPTransport(
host="0.0.0.0",
port=8443,
ssl_cert_file="./certs/server.crt",
ssl_key_file="./certs/server.key"
)
# Handle incoming authentication
async def handle_auth(request):
return await auth_engine.process_auth_request(request)
server_transport.set_auth_handler(handle_auth)
await server_transport.start_server()WebSocket 传输
from mcp_pki_auth.transport import WebSocketTransport
# Client WebSocket connection
ws_transport = WebSocketTransport(
url="wss://api.example.com:8443/ws",
ssl_cert_file="./certs/client.crt",
max_connections=10
)
# Persistent connection for multiple auths
async with ws_transport.connect() as connection:
response = await connection.authenticate(auth_request)
# Server WebSocket handler
ws_server = WebSocketTransport(
host="0.0.0.0",
port=8443,
ssl_cert_file="./certs/server.crt"
)
await ws_server.start_server(auth_handler)审计日志记录
from mcp_pki_auth.audit import AuditLogger
# Initialize audit logger
audit = AuditLogger(
log_file="./logs/audit.jsonl",
log_level="INFO",
include_performance=True
)
# Log authentication events
audit.log_auth_attempt(
client_fingerprint="a1b2c3d4...",
server_fingerprint="e5f6g7h8...",
success=True,
duration_ms=8.5,
metadata={"transport": "https", "endpoint": "/auth"}
)
# Log key validation
audit.log_key_validation(
fingerprint="a1b2c3d4...",
valid=True,
metadata={"source": "allowlist"}
)
# Get performance metrics
metrics = audit.get_performance_metrics()
print(f"Average auth time: {metrics['avg_auth_time_ms']:.2f}ms")安全
加密详细信息
- 算法ed25519(使用SHA-512的Curve25519)
- 密钥长度32字节的公钥,64字节的签名
- 哈希函数使用SHA-256进行指纹处理,使用SHA-512进行签名处理
安全特性
重放保护
- 时间戳所有消息均包含在±5分钟范围内验证的Unix时间戳
- 使节独特的随机数(nonce)防止重复挑战被重用
- 签名覆盖率签名包括所有消息字段以及分隔符
访问控制
- 仅允许名单(或白名单)没有隐式信任;所有密钥必须明确允许
- 基于指纹的使用SHA-256指纹进行密钥识别
- 元数据支持用于密钥管理和审计的丰富元数据
运输安全
- TLS 加密使用可配置的密码套件的HTTPS/WSS传输
- 证书验证完整的证书链验证
- 连接限制可配置的连接池和速率限制
最佳实践
密钥管理
# Generate keys with proper permissions
umask 077
mcp-keygen --output-dir ./keys --key-name production
# Store private keys securely
chmod 600 ./keys/*_private.pem
chmod 644 ./keys/*_public.pem
# Regular key rotation (recommended: annually)
mcp-keygen --output-dir ./keys --key-name production_2024配置安全
# Use strong timestamp tolerance (not too permissive)
auth:
timestamp_tolerance: 300 # 5 minutes max
# Enable comprehensive audit logging
audit:
enabled: true
log_level: "INFO"
include_performance: true
# Use TLS with proper certificates
transport:
type: "https"
ssl:
verify_mode: "required"
cert_file: "/path/to/cert.pem"
key_file: "/path/to/key.pem"操作安全
- 监控审计日志以检测失败的认证尝试
- 定期轮换密钥和证书
- 为不同的环境使用不同的密钥对
- 实施适当的备份和恢复程序
- 监控性能指标以检测异常
发展
项目结构
mcp-sec/
├── src/mcp_pki_auth/ # Main package
│ ├── __init__.py
│ ├── key_manager.py # Key generation and management
│ ├── acl_manager.py # Allowlist management
│ ├── auth_engine.py # Authentication protocol
│ ├── protocol.py # Message handling
│ ├── config.py # Configuration management
│ ├── audit.py # Audit logging
│ ├── transport.py # Network transport layer
│ ├── exceptions.py # Custom exceptions
│ └── cli/ # CLI tools
│ ├── __init__.py
│ ├── keygen.py # Key generation CLI
│ └── allowlist.py # Allowlist management CLI
├── tests/ # Test suite
│ ├── unit/ # Unit tests
│ └── integration/ # Integration tests
├── examples/ # Example implementations (planned)
├── Dockerfile.test # Docker test environment
├── docker-compose.test.yml # Docker Compose for testing
├── Makefile # Build automation
├── pyproject.toml # Project metadata
├── requirements.txt # Dependencies
├── requirements-dev.txt # Development dependencies
└── README.md # This file设置开发环境
使用 Docker(推荐)
# Clone and setup
git clone https://github.com/your-org/mcp-sec.git
cd mcp-sec
# Setup test environment
make setup-test-env
# Build development container
make build
# Run tests in container
make test
# Interactive development
make dev-shell本地开发
# Create virtual environment
python -m venv venv
source venv/bin/activate # Linux/Mac
# or: venv\Scripts\activate # Windows
# Install in development mode
pip install -e .
pip install -r requirements-dev.txt
# Run tests locally
pytest tests/
# Run linting
make lintGit 工作流
功能开发
# Create feature branch
git checkout -b feature/new-transport-layer
# Make changes and test
make test
make lint
# Commit and push
git add .
git commit -m "Add WebSocket transport layer with connection pooling"
git push origin feature/new-transport-layer实施分支
# For language-specific implementations
git checkout -b impl/rust-performance
git checkout -b impl/go-server
git checkout -b impl/nodejs-client
# Use worktrees for parallel development
git worktree add ../mcp-sec-rust impl/rust-performance
cd ../mcp-sec-rust
# Work on Rust implementation代码风格
- python遵循PEP 8规范,使用
black用于格式化,pylint用于代码检查(或语法检查) - 类型提示为所有公共API使用类型注解
- 文档所有公共类和函数的文档字符串
- 测试目标是达到核心认证路径95%以上的测试覆盖率
测试
测试类别
单元测试
# Run all unit tests
pytest tests/unit/
# Test specific component
pytest tests/unit/test_key_manager.py
pytest tests/unit/test_auth_engine.py
# With coverage
pytest tests/unit/ --cov=mcp_pki_auth --cov-report=html集成测试
# Full authentication flow tests
pytest tests/integration/
# Test with Docker environment
make test-integration
# Test specific scenarios
pytest tests/integration/test_full_auth_flow.py -v安全测试
# Security-focused tests
pytest tests/security/
# Replay attack prevention
pytest tests/security/test_replay_protection.py
# Timing attack resistance
pytest tests/security/test_timing_attacks.py性能测试
# Performance benchmarks
pytest tests/performance/ --benchmark-only
# Specific performance metrics
pytest tests/performance/test_auth_performance.py -s
# Memory usage testing
pytest tests/performance/test_memory_usage.py基于Docker的测试
# All tests in clean environment
make test
# Specific test suites
make test-unit
make test-integration
make test-security
make test-performance
# Cross-language integration (when available)
make test-cross-language持续集成
该项目包含GitHub Actions工作流:
- 单元测试每次提交时运行
- 集成测试在主分支的拉取请求(PRs)上运行
- 安全测试每晚运行
- 性能测试在发布版本上运行
做出贡献
入门指南/开始使用
- 为仓库创建分支(或“克隆仓库”)
- 创建一个特性分支
- 进行你的更改
- 为新功能添加测试
- 确保所有测试通过
- 提交拉取请求
发展指南
代码质量
- 所有代码都必须通过(测试/验证)
make lint支票 - 新功能需要进行覆盖率超过90%的测试
- 公共API需要文档说明
- 遵循现有的代码模式和规范
拉取请求流程
- 描述清晰阐述变化及动机
- 测试包含对新功能的测试
- 文档更新相关文件
- 演出考虑变更对性能的影响
- 安全强调任何安全影响
问题报告
在报告问题时,请包含:
- Python版本和操作系统
- 完整的错误信息和堆栈跟踪
- 最小化复现步骤
- 配置文件(已清理)
架构决策
在实施之前,应通过 GitHub 问题跟踪系统讨论重大的架构变更。请考虑:
- 性能影响
- 安全影响
- 向后兼容性
- 跨语言实现的挑战
未来路线图
立即(下一发布版)
- 额外的命令行工具认证测试
mcp-auth-test) 和配置验证 (mcp-config-validate) - 示例应用服务器/客户端示例实现
- 配置管理完整的YAML配置加载和验证
中期
- 增强的运输(能力/系统)带有SSL/TLS的完整HTTP/WebSocket服务器实现
- 性能优化针对高吞吐量场景的基准测试与优化
- 安全增强额外的安全测试和防范时序攻击
长期
- 多语言支持Go、Rust、Node.js 的实现
- 密钥轮换自动化密钥轮换协议
- 联邦跨域身份验证支持
- 硬件安全HSM(硬件安全模块)和硬件密钥支持
- 监测Prometheus指标集成
当前实施状态
✅ 已完成的功能
- 核心认证完成4条消息的ed25519双向认证协议
- 密钥管理密钥生成、加载、保存和指纹识别
- 访问控制支持元数据的白名单管理
- 审计日志记录带有性能指标的全面结构化日志记录
- 命令行界面(CLI)工具密钥生成(
mcp-keygen) 和白名单管理 (mcp-allowlist) - 协议处理消息的序列化/反序列化及验证
- Docker 支持容器化的测试与开发环境
- 测试套件单元测试和集成测试,覆盖率70%以上
🚧(施工中/道路封闭/请绕行) 在开发中
- 传输层HTTP/WebSocket 实现(基本结构已存在)
- 配置管理YAML 配置系统(部分实现)
📋 表格/清单 计划中的功能
- 额外的命令行工具身份验证测试和配置验证
- 示例应用完整的服务器/客户端实现
- 增强的文档记录API文档和部署指南
______________________________________________________________________
许可证
这个项目遵循MIT许可证授权。参见 许可证 详情见文件。
支持
- 文档: 文档/
- 问题:
- 讨论:
- 安全请将安全问题报告至 security@example.com
______________________________________________________________________
版本1.0.0 最后更新时间2025年10月 维护者M.Hjorleifsson
