FastAPI限流白名单：API密钥的终极指南
·
FastAPI限流白名单：API密钥的终极指南
项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi FastAPI作为现代高性能Python Web框架，提供了强大的API密钥认证机制和灵活的限流策略。本文将详细介绍如何结合API密钥实现智能的限流白名单功能，保护你的API免受恶意攻击和滥用。🚀
什么是API密钥认证？
API密钥是FastAPI中最简单的认证方式之一，它允许客户端通过查询参数、请求头或Cookie传递一个特定的密钥来访问受保护的API端点。FastAPI内置了三种API密钥认证方式：
- APIKeyQuery - 通过查询参数传递API密钥
- APIKeyHeader - 通过HTTP头部传递API密钥
- APIKeyCookie - 通过Cookie传递API密钥
FastAPI自动生成的Swagger UI文档界面
为什么需要限流白名单？
限流（Rate Limiting）是保护API的重要机制，它可以防止：
- 恶意攻击：如DDoS攻击、暴力破解
- 资源滥用：防止单个用户消耗过多服务器资源
- 服务稳定性：确保所有用户都能公平访问服务
白名单机制允许特定API密钥享受更高的请求限制或完全免除限制，这对于以下场景特别有用：
✅ 合作伙伴API：商业合作伙伴需要更高的调用频率 ✅ 内部服务：内部系统间的通信需要更高的优先级 ✅ 付费用户：付费用户应获得更好的服务质量 ✅ 开发测试：开发环境需要宽松的限制
FastAPI API密钥基础实现
首先，让我们看看如何在FastAPI中使用API密钥：
from fastapi import Depends, FastAPI
from fastapi.security import APIKeyHeader
app = FastAPI()
api_key_header = APIKeyHeader(name="X-API-Key")
@app.get("/protected/")
async def protected_route(api_key: str = Depends(api_key_header)):
return {"message": "Access granted", "api_key": api_key}
在这个示例中，我们创建了一个需要X-API-Key请求头的受保护路由。客户端必须在请求头中包含有效的API密钥才能访问该端点。
实现智能限流白名单系统
现在让我们构建一个完整的限流白名单系统：
步骤1：定义API密钥验证器
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import APIKeyHeader
from datetime import datetime, timedelta
from collections import defaultdict
app = FastAPI()
# 存储API密钥和对应的限流配置
API_KEYS = {
"partner-key-123": {
"name": "Partner Company",
"rate_limit": 100, # 每分钟100次
"whitelisted": True # 在白名单中
},
"internal-key-456": {
"name": "Internal Service",
"rate_limit": 1000, # 每分钟1000次
"whitelisted": True
},
"default-key-789": {
"name": "Regular User",
"rate_limit": 10, # 每分钟10次
"whitelisted": False
}
}
# 请求跟踪字典
request_tracker = defaultdict(list)
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
def validate_api_key(api_key: str = Depends(api_key_header)):
if not api_key:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="API key is required"
)
if api_key not in API_KEYS:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API key"
)
return api_key
步骤2：实现限流中间件
from fastapi import Request
from fastapi.responses import JSONResponse
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
# 检查是否有API密钥
api_key = request.headers.get("X-API-Key")
if not api_key:
# 如果没有API密钥，使用默认限制
return await apply_rate_limit(request, call_next, "anonymous")
if api_key in API_KEYS:
key_info = API_KEYS[api_key]
# 检查是否在白名单中
if key_info["whitelisted"]:
# 白名单用户，应用特殊限制
return await apply_rate_limit(
request, call_next, api_key,
key_info["rate_limit"]
)
else:
# 普通用户，应用默认限制
return await apply_rate_limit(
request, call_next, api_key,
key_info["rate_limit"]
)
# 无效的API密钥
return JSONResponse(
status_code=401,
content={"detail": "Invalid API key"}
)
async def apply_rate_limit(
request: Request,
call_next,
identifier: str,
limit: int = 10 # 默认每分钟10次
):
current_time = datetime.now()
one_minute_ago = current_time - timedelta(minutes=1)
# 清理过期的请求记录
request_tracker[identifier] = [
timestamp for timestamp in request_tracker[identifier]
if timestamp > one_minute_ago
]
# 检查是否超过限制
if len(request_tracker[identifier]) >= limit:
return JSONResponse(
status_code=429,
content={
"detail": f"Rate limit exceeded. Limit: {limit} requests per minute",
"retry_after": 60 # 60秒后重试
},
headers={"Retry-After": "60"}
)
# 记录本次请求
request_tracker[identifier].append(current_time)
# 继续处理请求
response = await call_next(request)
return response
步骤3：创建受保护的路由
@app.get("/api/data")
async def get_data(api_key: str = Depends(validate_api_key)):
key_info = API_KEYS[api_key]
return {
"message": "Data retrieved successfully",
"api_key_info": {
"name": key_info["name"],
"rate_limit": key_info["rate_limit"],
"whitelisted": key_info["whitelisted"]
},
"remaining_requests": calculate_remaining_requests(api_key)
}
def calculate_remaining_requests(api_key: str):
current_time = datetime.now()
one_minute_ago = current_time - timedelta(minutes=1)
recent_requests = [
timestamp for timestamp in request_tracker[api_key]
if timestamp > one_minute_ago
]
limit = API_KEYS[api_key]["rate_limit"]
return max(0, limit - len(recent_requests))
高级白名单功能实现
基于角色的访问控制
from enum import Enum
class UserRole(str, Enum):
ADMIN = "admin"
PARTNER = "partner"
PREMIUM = "premium"
STANDARD = "standard"
# 扩展API密钥配置
ENHANCED_API_KEYS = {
"admin-key-001": {
"name": "System Admin",
"role": UserRole.ADMIN,
"rate_limit": 1000,
"whitelisted": True,
"permissions": ["read", "write", "delete", "manage_users"]
},
"partner-key-002": {
"name": "Gold Partner",
"role": UserRole.PARTNER,
"rate_limit": 500,
"whitelisted": True,
"permissions": ["read", "write"]
}
}
def check_permission(api_key: str, required_permission: str):
key_info = ENHANCED_API_KEYS.get(api_key)
if not key_info:
return False
return required_permission in key_info["permissions"]
@app.get("/api/admin/users")
async def get_users(
api_key: str = Depends(validate_api_key),
permission_check: bool = Depends(lambda: check_permission(api_key, "manage_users"))
):
if not permission_check:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Insufficient permissions"
)
return {"users": ["user1", "user2", "user3"]}
动态白名单管理
from typing import Dict
import redis
import json
# 使用Redis存储动态白名单配置
class DynamicWhitelist:
def __init__(self):
self.redis_client = redis.Redis(
host='localhost',
port=6379,
db=0
)
def add_to_whitelist(self, api_key: str, config: Dict):
"""添加API密钥到白名单"""
self.redis_client.set(
f"whitelist:{api_key}",
json.dumps(config),
ex=86400 # 24小时过期
)
def is_whitelisted(self, api_key: str) -> bool:
"""检查API密钥是否在白名单中"""
return self.redis_client.exists(f"whitelist:{api_key}")
def get_config(self, api_key: str) -> Dict:
"""获取白名单配置"""
config_json = self.redis_client.get(f"whitelist:{api_key}")
if config_json:
return json.loads(config_json)
return None
# 初始化动态白名单管理器
whitelist_manager = DynamicWhitelist()
@app.post("/admin/whitelist/add")
async def add_to_whitelist(
api_key: str,
rate_limit: int = 100,
expires_in: int = 86400
):
config = {
"rate_limit": rate_limit,
"whitelisted": True,
"created_at": datetime.now().isoformat(),
"expires_in": expires_in
}
whitelist_manager.add_to_whitelist(api_key, config)
return {"message": "API key added to whitelist", "config": config}
最佳实践和注意事项
1. 安全存储API密钥
- 使用环境变量存储主密钥
- 定期轮换API密钥
- 记录所有API密钥的使用情况
2. 监控和日志
- 记录所有API调用，包括成功和失败的尝试
- 监控异常访问模式
- 设置警报机制
3. 性能优化
- 使用Redis或Memcached存储限流计数器
- 考虑分布式限流策略
- 使用缓存减少数据库查询
4. 错误处理
- 提供清晰的错误消息
- 包含重试时间信息
- 记录详细的调试信息
实际应用场景
场景1：合作伙伴API集成
# 为合作伙伴提供更高的请求限制
PARTNER_WHITELIST = {
"partner-company-a": {
"rate_limit": 1000, # 每分钟1000次
"features": ["data_export", "batch_processing"]
},
"partner-company-b": {
"rate_limit": 500, # 每分钟500次
"features": ["data_export"]
}
}
场景2：付费用户分级
# 根据用户等级设置不同的限制
USER_TIERS = {
"free": {"rate_limit": 10, "whitelisted": False},
"basic": {"rate_limit": 100, "whitelisted": True},
"premium": {"rate_limit": 1000, "whitelisted": True},
"enterprise": {"rate_limit": 10000, "whitelisted": True}
}
总结
FastAPI的API密钥认证结合智能限流白名单机制，为你的API提供了强大的保护层。通过本文介绍的方法，你可以：
🎯 轻松实现API密钥认证：支持查询参数、请求头和Cookie三种方式 🛡️ 有效防止API滥用：基于IP、用户和API密钥的多维度限流 ⚡ 提供差异化服务：为白名单用户提供更高的请求限制 📊 实时监控和调整：动态管理白名单和限流策略
FastAPI的ReDoc文档界面，支持API密钥认证的文档展示
记住，良好的API安全实践不仅仅是技术实现，更是持续监控、调整和优化的过程。通过合理的限流白名单策略，你可以在保护API安全的同时，为不同用户群体提供优质的服务体验。
下一步行动
- 测试你的实现：使用不同的API密钥测试限流效果
- 监控API使用情况：设置监控仪表板跟踪API使用模式
- 定期审查白名单：定期清理不再需要的白名单条目
- 考虑使用第三方库：如
slowapi或fastapi-limiter获得更多功能
通过本文的指导，你已经掌握了在FastAPI中实现API密钥限流白名单的核心技术。现在就开始为你的API添加这一重要保护层吧！💪
更多推荐
30
0- 0
已为社区贡献8条内容
所有评论(0)