架构概述
基于微服务的身份认证架构采用OAuth 2.0/OpenID Connect协议，OpenIddict作为认证服务器，FastAPI作为资源服务器。系统包含三个核心组件：认证服务、API网关和业务微服务。OpenIddict负责颁发令牌，FastAPI通过JWT验证访问权限。
技术栈选型
- 认证服务器: OpenIddict 4.8（基于ASP.NET Core）
- 资源服务器: FastAPI 0.95+（Python 3.10+）
- 令牌格式: JWT with RSA256签名
- 数据库: PostgreSQL 14用于存储客户端和令牌数据
- 消息队列: Redis 7用于授权码流处理
认证服务设计
OpenIddict配置需启用授权码、客户端凭证和刷新令牌流程。关键配置项包括：
services.AddOpenIddict().AddCore(options => options.UseEntityFrameworkCore().UseDbContext<AuthDbContext>()).AddServer(options => {options.SetTokenEndpointUris("/connect/token");options.SetAuthorizationEndpointUris("/connect/authorize");options.AllowAuthorizationCodeFlow().AllowClientCredentialsFlow();options.RegisterScopes("api1", "offline_access");options.AddDevelopmentEncryptionCertificate().AddDevelopmentSigningCertificate();}).AddValidation(options => options.UseLocalServer());
FastAPI 资源服务器实现
安装必需依赖：
pip install fastapi uvicorn python-jose[cryptography] passlib bcrypt
JWT验证中间件示例：
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2AuthorizationCodeBearer
from jose import JWTError, jwtoauth2_scheme = OAuth2AuthorizationCodeBearer(authorizationUrl="http://auth-server/connect/authorize",tokenUrl="http://auth-server/connect/token"
)async def verify_token(token: str = Depends(oauth2_scheme)):try:payload = jwt.decode(token,key=public_key, # 从认证服务获取的公钥algorithms=["RS256"],audience="api1")return payloadexcept JWTError:raise HTTPException(status_code=401, detail="Invalid token")
微服务间通信安全
服务到服务通信采用客户端凭证流程，需在OpenIddict注册机器客户端：
{"client_id": "service_account","client_secret": "your-secret","grant_types": ["client_credentials"],"scopes": ["api1"]
}
Python客户端实现示例：
import httpxasync def get_service_token():async with httpx.AsyncClient() as client:response = await client.post("http://auth-server/connect/token",data={"grant_type": "client_credentials","client_id": "service_account","client_secret": "your-secret","scope": "api1"})return response.json()["access_token"]
权限控制设计
基于RBAC模型实现细粒度权限控制，OpenIddict scope与角色权限映射示例：
from fastapi import Securitydef require_permission(permission: str):async def checker(payload: dict = Security(verify_token)):if permission not in payload.get("scope", "").split():raise HTTPException(status_code=403, detail="Insufficient permissions")return payloadreturn checker# 使用示例
@app.get("/admin")
async def admin_route(user: dict = Depends(require_permission("admin"))):return {"message": "Admin access granted"}
性能优化策略
- 令牌缓存: Redis缓存已验证的JWT，减少重复验证开销
- 公钥轮换: JWKS端点实现自动密钥轮换
- 批量验证: 对微服务批量请求采用令牌哈希校验
Redis缓存实现示例：
from redis import asyncio as aioredisasync def cached_verify_token(token: str):redis = aioredis.from_url("redis://localhost")cache_key = f"token:{hashlib.sha256(token.encode()).hexdigest()}"if await redis.exists(cache_key):return json.loads(await redis.get(cache_key))payload = await verify_token(token)await redis.setex(cache_key, 3600, json.dumps(payload))return payload
安全增强措施
- 令牌绑定: 实现DPoP机制防止令牌重放
- 审计日志: 记录所有令牌颁发和验证事件
- 令牌撤销: 实时检查Redis黑名单
部署架构
建议采用容器化部署方案：
# Auth服务
FROM mcr.microsoft.com/dotnet/sdk:7.0 AS auth-server
EXPOSE 5000
COPY --from=builder /app .
ENTRYPOINT ["dotnet", "AuthServer.dll"]# FastAPI服务
FROM python:3.10-slim
EXPOSE 8000
COPY ./app /app
RUN pip install -r /app/requirements.txt
CMD ["uvicorn", "main:app", "--host", "0.0.0.0"]
监控指标
关键监控项包括：
- 认证服务QPS和响应时间
- JWT验证失败率
- 令牌缓存命中率
- 权限检查延迟
Prometheus监控示例配置：
scrape_configs:- job_name: 'auth'metrics_path: '/metrics'static_configs:- targets: ['auth-server:5000']- job_name: 'api'static_configs:- targets: ['api-service:8000']
