mcp-auth-py——ASGI应用程序的可插拔身份验证（FastAPI友好）

![CI](https://github.com/cbritt0n/mcp-auth-py/actions/workflows/ci.yml)

太长，读不下去了 FastAPI/ASGI的可插拔身份验证——交换提供者（本地/google/aws/aazure），支持异步，带JWKS缓存（可选Redis）。

🔒 mcp-auth-py是一个小型、框架友好的库，它添加了可插入的身份验证提供程序适用于FastAPI（或其他ASGI应用程序）。它提供了一个轻量级的中间件和一个小型提供者注册表，以便您可以交换本地（JWT）、谷歌、AWS（Cognito）的身份验证后端，以及Azure（AAD），而无需更改应用程序代码。

为什么要使用mcp-auth-py？

即插即用提供商：swap local, google, aws （Cognito）， azure, github，或 discord 而无需更改应用程序代码。
云感知：支持Google ID令牌、AWS Cognito OIDC、Azure AD OIDC、GitHub OAuth2和Discord OAuth2开箱即用。
企业就绪：多租户架构、合规性监控（GDPR、HIPAA、SOX）和性能优化。
异步友好：中间件和提供者可以异步；阻塞SDK被卸载到线程池中。
JWKS缓存：每进程TTL缓存加上可选的Redis支持的JWKS适配器，用于多进程共享。
最小化核心负担：让你的应用程序保持轻量；提供者SDK是您根据需要安装的可选附加组件。
可测试和CI就绪：包括单元测试和CI预提交挂钩，以保持高质量。

这个图书馆做什么

为ASGI应用程序添加一个小型中间件和提供者注册表（提供了FastAPI示例）。
提供规范 Principal 模型和a AuthResult 合同，以便提供商返回统一的形状。
使添加新提供者变得容易：实现 Provider 接口并注册。
生产安全：全面的JWT验证、速率限制、暴力保护、安全标头
企业RBAC：基于角色的访问控制，具有分层权限和特定于资源的授权
实时通知：基于WebSocket的实时更新，用于安全事件和权限更改
高性能缓存：基于Redis的分布式缓存，具有智能失效模式
审计与合规：通过安全分析和合规报告完成审计跟踪

特性

🔐 多提供商身份验证：本地JWT、谷歌OAuth2、AWS Cognito、Azure AD、GitHub OAuth2和Discord OAuth2
🏢 企业多租户：数据库/模式隔离、分层组织、条件访问策略
🛡️ 生产安全：企业级JWT验证、速率限制、暴力保护、安全标头
📋 合规性监测：自动化GDPR、HIPAA、SOX合规性，实时评估和报告
⚡ 高性能：基于Redis的分布式缓存，具有断路器和性能监控功能
🚀 RBAC系统：完整的基于角色的访问控制，具有分层权限和特定于资源的授权
📡 实时更新：基于WebSocket的安全事件和权限更改实时通知
📊 审计与合规：具有安全分析和合规性报告的全面审计跟踪
🔧 开发者友好：支持异步的流、FastAPI装饰器、全面的示例和文档
☁️ 云原生：Kubernetes部署、Docker支持、生产就绪配置

安装

🚀 快速设置

# Clone and install locally
git clone https://github.com/cbritt0n/mcp-auth-py.git
cd mcp-auth-py
pip install -e .

# Run example application
uvicorn examples.server:app --reload

# Visit http://localhost:8000/docs to see the API

📦 安装选项

# Clone the repository
git clone https://github.com/cbritt0n/mcp-auth-py.git
cd mcp-auth-py

# Basic installation (local JWT only)
pip install -e .

# With specific cloud providers
pip install -e .[google]        # Google OAuth2
pip install -e .[aws]           # AWS Cognito
pip install -e .[azure]         # Azure AD
pip install -e .[github]        # GitHub OAuth2 (built-in)
pip install -e .[discord]       # Discord OAuth2 (built-in)
pip install -e .[redis_jwks]    # Redis caching
pip install -e .[rbac]          # RBAC Extension
pip install -e .[realtime]      # WebSocket real-time features
pip install -e .[audit]         # Audit trail and analytics
pip install -e .[enterprise]    # Multi-tenant enterprise features

All providers + RBAC + Real-time + Audit + Security

pip install -e .[full]

Production security with comprehensive hardening

pip install -e .[security] # JWT validation, rate limiting, security headers

Development with all testing tools

pip install -e .[dev] # Testing, linting, pre-commit hooks

Quick start

Run the example app in examples/server.py:

uvicorn examples.server:app --reload

访问http://localhost:8000/hello--安装了中间件，它将阻止没有有效令牌的请求。

具有生产安全性的单文件FastAPI示例

这是一个具有全面安全性的最小复制粘贴FastAPI应用程序：

from fastapi import FastAPI, Request, Depends
from mcp_auth.settings import Settings
from mcp_auth.setup import setup_auth
from mcp_auth.security import get_validated_principal, require_admin_principal
from mcp_auth.middleware_security import setup_production_security

# Configure settings for production
settings = Settings(
    auth_provider="local",
    jwt_secret="prod-your-super-secure-256-bit-jwt-secret-key-here-minimum-32-chars",
    enable_rate_limiting=True,
    enable_security_headers=True,
    require_https=False,  # Set to True in production
    max_login_attempts=5,
    rate_limit_requests_per_minute=100
)

# Create FastAPI app with authentication and security
app = FastAPI(title="Secure API")
app = setup_auth(app, settings)
setup_production_security(app, settings)

@app.get("/hello")
def hello(principal=Depends(get_validated_principal)):
    """Public endpoint with authentication required"""
    return {
        "message": f"Hello {principal.name or principal.id}!",
        "user_id": principal.id,
        "provider": principal.provider,
        "roles": principal.roles
    }

@app.get("/admin/stats")
def admin_stats(principal=Depends(require_admin_principal)):
    """Admin-only endpoint with enhanced security"""
    return {
        "message": f"Admin access granted to {principal.name}",
        "system_status": "healthy",
        "security_level": "high"
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行测试

# Install with dev dependencies
pip install -e .[dev]

# Run the test suite
pytest -v

提供商SDK（可选）

mcp-auth-py使用云提供商的可选依赖关系来保持核心包的轻量级：

# Install specific providers as needed
pip install -e .[google]        # Google OAuth2 (google-auth)
pip install -e .[aws]           # AWS Cognito (boto3)
pip install -e .[azure]         # Azure AD (OIDC only, no extra deps)
pip install -e .[github]        # GitHub OAuth2 (built-in, no extra deps)
pip install -e .[discord]       # Discord OAuth2 (built-in, no extra deps)
pip install -e .[redis_jwks]    # Redis JWKS caching
pip install -e .[enterprise]    # Multi-tenant enterprise features

# Or install everything
pip install -e .[full]          # All providers + Redis + uvicorn + enterprise

异步支持和生产准备

异步支持

中间件和提供者支持异步。提供者可以实现为协程或同步

函数——中间件将 await 协程结果自动生成。

用于阻止SDK（例如。 boto3)我们使用线程池卸载来避免阻塞事件循环。
为了在生产环境中完全无阻塞，请启用支持异步的库，如 httpx （JWKS）和 aioredis.

生产检查表

使用 redis_jwks 用于跨进程共享JWKS缓存的适配器。集 redis_url 在 Settings.
仅安装生产中所需的提供程序SDK（避免运送开发垫片）。
跑 pre-commit run --all-files 在推动保持CI绿色之前。
在生产ASGI服务器（Uvicorn/Gunicorn，带workers）下运行，并在边缘终止TLS。
配置JWKS获取和缓存命中率的超时、连接池和监控。

适配器（非ASGI/MCP服务器）

该套餐包括 mcp_auth.adapters 与非ASGI呼叫提供商的助手或同步MCP服务器：

authenticate_request(provider, request) --等待提供者结果的异步助手。
authenticate_request_sync(provider, request) --旧服务器的同步包装器。
token_to_principal(provider, token) / token_to_principal_sync(...) --最小令牌助手。

示例（同步服务器）：

from mcp_auth.providers.registry import get_provider
from mcp_auth.adapters import token_to_principal_sync

provider = get_provider("aws")
principal = token_to_principal_sync(provider, token)
if principal is None:
    # unauthorized
    ...

提供程序配置

集 auth_provider 和 provider_config 通过 mcp_auth.settings.Settings （Pydantic设置（如果可用））。

本地（默认）

from mcp_auth.settings import Settings
settings = Settings(auth_provider="local")

谷歌

settings = Settings(
    auth_provider="google",
    provider_config={"audience": "GOOGLE_CLIENT_ID"},
)

AWS Cognito

settings = Settings(
    auth_provider="aws",
    provider_config={
        "cognito_region": "us-west-2",
        "cognito_user_pool_id": "us-west-2_XXXXXXXXX",
        "audience": "YOUR_COGNITO_APP_CLIENT_ID",
        "use_cognito_get_user": False,
    },
)

Azure AD

settings = Settings(
    auth_provider="azure",
    provider_config={"tenant": "your-tenant-id", "audience": "APP_CLIENT_ID"},
)

GitHub OAuth2

settings = Settings(
    auth_provider="github",
    provider_config={
        "client_id": "your_github_client_id",
        "client_secret": "your_github_client_secret",  # Optional
        "scopes": ["user:email", "read:org"],
        "allowed_organizations": ["your-org", "partner-org"],  # Optional
    },
)

Discord OAuth2

settings = Settings(
    auth_provider="discord",
    provider_config={
        "client_id": "your_discord_client_id",
        "client_secret": "your_discord_client_secret",  # Optional
        "bot_token": "your_discord_bot_token",  # Optional, for role verification
        "scopes": ["identify", "email", "guilds"],
        "allowed_guilds": ["123456789012345678"],  # Optional server restrictions
    },
)

🏢 企业多租户

MCP Auth为企业级多租户提供多种隔离策略：

设置企业功能

# Install enterprise features
pip install -e .[enterprise]

# Configure multi-tenancy
python tests/setup_wizard.py  # Choose option 7: Enterprise

多租户配置

from mcp_auth.enterprise import MultiTenantAuth, TenantStrategy
from mcp_auth.enterprise.compliance import ComplianceMonitor

# Configure tenant isolation strategy
settings = Settings(
    auth_provider="google",  # Any provider works
    tenant_strategy=TenantStrategy.ROW_LEVEL_SECURITY,  # or DATABASE_PER_TENANT, SCHEMA_PER_TENANT
    tenant_resolver="header",  # header, subdomain, path, jwt
    redis_url="redis://localhost:6379/0"
)

# Setup multi-tenant authentication
app = FastAPI()
tenant_auth = MultiTenantAuth(settings)
app = tenant_auth.setup_app(app)

# Compliance monitoring
compliance = ComplianceMonitor(settings)
app.include_router(compliance.get_router(), prefix="/compliance")

@app.get("/api/data")
async def get_tenant_data(
    principal=Depends(get_validated_principal),
    tenant=Depends(get_current_tenant)
):
    # Automatic tenant isolation based on strategy
    return {
        "tenant_id": tenant.id,
        "user": principal.name,
        "isolation": tenant.strategy.value
    }

多租户功能

灵活隔离：每个租户的数据库、每个租户的架构或行级安全性
层级组织：具有继承权限的父子租户关系
条件接收政策：IP限制、基于时间的访问、设备要求
租户管理：用于租户管理和配置的REST API
性能优化：租户感知缓存和连接池
合规性集成：自动租户级合规性监控

看 docs/enterprise_guide.md 完成企业设置和 examples/enterprise_demo.py 作为工作示例。

🔧 快速设置向导

使用交互式安装向导配置任何提供程序：

# Run the setup wizard
python tests/setup_wizard.py

# Choose your provider:
# 1. Local (JWT with secret key)
# 2. Google (OAuth2)
# 3. AWS (Cognito)
# 4. Azure (Active Directory)
# 5. GitHub (OAuth2)
# 6. Discord (OAuth2)
# 7. Enterprise (Multi-tenant)

向导将：

生成安全 .env 配置文件
引导您完成特定于提供商的设置（OAuth应用程序、客户端ID等）
配置可选功能（Redis缓存、企业功能）
提供后续步骤和测试命令

使用Redis支持的JWKS缓存

安装可选 redis_jwks 额外的，并通过以下方式为每个提供商启用 redis_jwks=True 和 redis_url:

pip install .[redis_jwks]

from mcp_auth.providers.aws import AWSProvider

provider = AWSProvider({
    "cognito_region": "us-west-2",
    "cognito_user_pool_id": "us-west-2_XXXX",
    "redis_jwks": True,
    "redis_url": "redis://redis.example.local:6379/0",
})

适配器是可选的；取消设置后，提供程序将回退到进程内缓存。

重要说明

JWKS按提供程序实例缓存；使用Redis实现多进程共享。
该测试套件包括供应商垫片，用于测试而不需要所有云SDK。

贡献

看 CONTRIBUTING.md 以获取贡献者指南。

许可证

Apache-2.0--参见 LICENSE.

通过以下方式支持环境变量 pydantic-settings （参见 Settings.Config.env_file).

🚀 生产部署

Docker（推荐）

# Quick start with Docker
docker build -t mcp-auth .
docker run -p 8000:8000 --env-file .env mcp-auth

# Multi-provider setup with docker-compose
docker-compose up -d  # Runs local, AWS, and Google providers

Kubernetes

kubectl apply -f k8s/deployment.yaml

看 文档/生产部署.md 获取全面的生产设置指南，包括：

完整的安全配置和强化
JWT代币安全和轮换政策
速率限制和暴力保护
AWS Cognito、Google OAuth2、Azure AD设置
Redis集群和高可用性
Kubernetes部署和负载平衡
监控、警报和事件响应
安全最佳实践和合规性

🛡️ 生产安全功能

MCP Auth包括用于生产部署的企业级安全功能：

身份验证安全

JWT令牌验证：具有过期、受众和发行者验证的行业标准JWT令牌
多提供商支持：在AWS Cognito、Azure AD、Google OAuth和本地身份验证之间无缝切换
令牌安全：可配置的过期、安全的秘密管理、令牌轮换支持
管理员授权：具有增强安全验证功能的专用管理端点

安全强化

速率限制：具有可配置限制和自适应限制的按IP请求限制
暴力保护：登录尝试跟踪，自动锁定和延迟升级
安全标头：全面的HTTP安全标头（HSTS、CSP、X-Frame-Options等）
请求验证：输入净化和恶意负载检测
HTTPS强制：可配置的HTTPS要求，具有适当的重定向处理

监控与合规

安全事件日志记录：完整的审计跟踪，包括安全事件分类和风险评分
实时监控：带管理仪表板的基于WebSocket的安全事件通知
异常检测：请求模式分析和自动威胁检测
合规报告：SOX、GDPR、HIPAA符合自动报告

安全配置示例

from mcp_auth.settings import Settings
from mcp_auth.security import TokenValidator, RateLimiter, AdminAuthorizer

# Production security settings
settings = Settings(
    # JWT Security
    jwt_secret="prod-your-super-secure-256-bit-jwt-secret-key-here-minimum-32-chars",
    jwt_access_token_expire_minutes=60,  # 1 hour
    jwt_audience="api.yourcompany.com",
    jwt_issuer="auth.yourcompany.com",

    # Rate Limiting & Protection
    enable_rate_limiting=True,
    rate_limit_requests_per_minute=100,
    max_login_attempts=5,
    lockout_duration_minutes=15,

    # Security Headers & HTTPS
    enable_security_headers=True,
    require_https=True,
    hsts_max_age=31536000,  # 1 year

    # Redis for distributed security
    redis_url="redis://redis-server:6379/0",
    redis_password="your-secure-redis-password"
)

# Initialize security components
token_validator = TokenValidator(settings)
rate_limiter = RateLimiter(settings)
admin_authorizer = AdminAuthorizer()

# Validate JWT tokens with comprehensive checks
principal = await token_validator.validate_token(
    token,
    check_expiration=True,
    check_audience=True,
    check_issuer=True
)

# Apply rate limiting with automatic IP tracking
await rate_limiter.check_rate_limit(request.client.host)

# Require admin privileges for sensitive operations
await admin_authorizer.require_admin_access(principal, "system.admin")

看 示例/生产示例.py 用于启用所有安全功能的完整生产设置。

💡 示例和用例

complete_app.py --带有用户端点的完整FastAPI应用程序
multi-provider.py --一个应用程序中的不同身份验证提供者
**** --生产集装箱化部署
rbac_demo.py --具有基于角色的访问控制的完整RBAC系统

🔐 RBAC扩展

RBAC（基于角色的访问控制）扩展增加了全面的授权功能：

from mcp_auth.rbac import RBACEngine, Role, Permission, require_permissions

# Setup RBAC engine
engine = RBACEngine()

# Create roles with hierarchical permissions
admin_role = Role("admin", "Administrator", [
    Permission.from_string("*:*:*")  # Full access
])
editor_role = Role("editor", "Content Editor", [
    Permission.from_string("posts:create"),
    Permission.from_string("posts:*:edit"),
    Permission.from_string("posts:*:delete")
])

engine.add_role(admin_role)
engine.add_role(editor_role)
engine.assign_role("user123", "editor")

# Protect endpoints with decorators
@app.post("/posts")
@require_permissions("posts:create")
async def create_post():
    return {"message": "Post created"}

@app.put("/posts/{post_id}")
@require_permissions("posts:edit")  # Auto-resolves to posts:{post_id}:edit
async def update_post(post_id: str):
    return {"message": f"Post {post_id} updated"}

RBAC功能

分层角色：角色继承父角色的权限
资源特定权限：支持通配符的细粒度控制
FastAPI装饰器: @require_permissions, @require_roles, @require_access
管理界面：用于管理角色和权限的REST API
灵活的体系结构：适用于任何身份验证提供程序

看 docs/rbacextension.md 获取完整的RBAC文档和 examples/rbac_demo.py 作为一个工作示例。

🌐 实时功能

为即时RBAC事件通知添加实时WebSocket支持：

from mcp_auth.realtime import setup_realtime_system, notify_rbac_event

# Enable WebSocket real-time features
realtime_router = setup_realtime_system(app)

# Client WebSocket connection at /ws
# Automatic broadcasting of permission changes, role assignments, security events
await notify_rbac_event(RBACEvent(
    event_type=EventType.PERMISSION_GRANTED,
    user_id="user123",
    resource="documents",
    action="read"
))

实时功能：

WebSocket管理：自动连接生命周期和身份验证
事件广播：权限更改和安全事件的实时通知
Redis分发：事件分布在多个服务器实例上
故障弱化：使用或不使用Redis
客户端过滤：用户只接收相关事件

看 docs/realtime_guide.md 获取完整的WebSocket集成指南。

⚡ 高性能缓存

基于Redis的分布式缓存显著提高了授权性能：

from mcp_auth.caching import setup_caching_system, enable_rbac_caching

# Setup Redis caching
await setup_caching_system(redis_url="redis://localhost:6379/0")
enable_rbac_caching(app)

# Permission checks are automatically cached
# 25x performance improvement for repeated operations
# Intelligent cache invalidation on role/permission changes

缓存功能：

分布式Redis缓存：跨多个服务器实例共享缓存
智能失效：权限更改时自动清理
性能监控：内置命中率和时间指标
批量操作：高效的批量获取/设置操作
基于模式的清理：智能缓存密钥管理

看 docs/caching.guide.md 用于缓存配置和优化。

📊 审计跟踪和安全分析

全面的审计日志记录和安全分析，用于合规性和监控：

from mcp_auth.audit import setup_audit_system, get_audit_logger

# Enable audit system with analytics dashboard
audit_router = setup_audit_system(app, enable_analytics=True)

# All RBAC operations automatically logged with context
# Custom security events
audit = get_audit_logger()
await audit.log_security_event(
    AuditEventType.SECURITY_VIOLATION,
    "Multiple failed login attempts detected",
    risk_score=85
)

# Built-in analytics dashboard at /audit/dashboard
# Security metrics, user access patterns, compliance reports

审核功能：

综合录井：所有具有完整上下文的RBAC操作
安全分析：异常检测和风险评分
合规报告：SOX、GDPR、HIPAA报告
访问模式分析：用户行为监控
实时警报：与安全监控系统集成
性能跟踪：授权性能和缓存指标

看 docs/audit_guide.md 用于完整的审计和分析文档。

MCP Auth Py

快速接入

详细介绍