FastAPI限流与IP白名单：完整指南与最佳实践

FastAPI限流与IP白名单完整指南与最佳实践【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapiFastAPI作为高性能的现代Python Web框架在生产环境中需要有效的安全防护措施。本文将详细介绍如何在FastAPI应用中实现限流Rate Limiting和IP白名单IP Whitelisting机制保护你的API免受恶意攻击和滥用。为什么需要限流和IP白名单在真实的Web应用部署中API接口面临着各种安全威胁恶意爬虫、DDoS攻击、暴力破解等。限流机制可以有效控制单个客户端在特定时间内的请求频率而IP白名单则允许你精确控制哪些IP地址可以访问你的API。核心概念解析限流Rate Limiting限制客户端在特定时间窗口内的请求次数IP白名单IP Whitelisting只允许特定IP地址访问API中间件MiddlewareFastAPI处理请求和响应的核心机制获取客户端IP地址在FastAPI中获取客户端IP地址非常简单。通过request.client.host属性你可以轻松获取到客户端的IP地址from fastapi import FastAPI, Request app FastAPI() app.get(/items/{item_id}) async def read_item(item_id: int, request: Request): client_ip request.client.host return {client_ip: client_ip, item_id: item_id}这是FastAPI中处理客户端IP的基础所有限流和IP过滤都建立在这个基础之上。实现IP白名单中间件IP白名单是保护API的第一道防线。下面是一个完整的IP白名单中间件实现from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse app FastAPI() # 允许访问的IP地址列表 ALLOWED_IPS [127.0.0.1, 192.168.1.100, 10.0.0.1] app.middleware(http) async def ip_whitelist_middleware(request: Request, call_next): # 获取客户端IP client_ip request.client.host # 检查IP是否在白名单中 if client_ip not in ALLOWED_IPS: return JSONResponse( status_code403, content{detail: Access denied. Your IP is not allowed.} ) # IP在白名单中继续处理请求 response await call_next(request) return response app.get(/protected) async def protected_endpoint(): return {message: This endpoint is protected by IP whitelist}高级IP白名单功能对于生产环境你可能需要更灵活的IP白名单配置import ipaddress from typing import List class IPWhitelistMiddleware: def __init__(self, app, allowed_ips: List[str], allow_cidr: bool True): self.app app self.allowed_ips set() self.cidr_ranges [] for ip in allowed_ips: if / in ip and allow_cidr: # 处理CIDR表示法如 192.168.1.0/24 self.cidr_ranges.append(ipaddress.ip_network(ip, strictFalse)) else: self.allowed_ips.add(ip) async def __call__(self, scope, receive, send): if scope[type] http: request Request(scope, receive) client_ip request.client.host # 检查精确IP匹配 if client_ip in self.allowed_ips: return await self.app(scope, receive, send) # 检查CIDR范围匹配 client_ip_obj ipaddress.ip_address(client_ip) for cidr in self.cidr_ranges: if client_ip_obj in cidr: return await self.app(scope, receive, send) # IP不在白名单中 response JSONResponse( status_code403, content{detail: Access denied} ) await response(scope, receive, send) return await self.app(scope, receive, send)实现限流中间件限流是保护API免受过载的关键技术。下面是一个简单的令牌桶算法实现import time from collections import defaultdict from fastapi import FastAPI, Request, HTTPException app FastAPI() # 限流配置 RATE_LIMIT 100 # 每分钟最大请求数 TIME_WINDOW 60 # 时间窗口秒 # 存储请求计数 request_counts defaultdict(list) app.middleware(http) async def rate_limit_middleware(request: Request, call_next): client_ip request.client.host current_time time.time() # 清理过期记录 request_counts[client_ip] [ req_time for req_time in request_counts[client_ip] if current_time - req_time TIME_WINDOW ] # 检查是否超过限制 if len(request_counts[client_ip]) RATE_LIMIT: raise HTTPException( status_code429, detailfRate limit exceeded. Try again in {TIME_WINDOW} seconds. ) # 记录当前请求 request_counts[client_ip].append(current_time) # 添加限流头部信息 response await call_next(request) response.headers[X-RateLimit-Limit] str(RATE_LIMIT) response.headers[X-RateLimit-Remaining] str( RATE_LIMIT - len(request_counts[client_ip]) ) response.headers[X-RateLimit-Reset] str( int(current_time TIME_WINDOW) ) return response使用Redis实现分布式限流在生产环境中你可能需要分布式限流解决方案import redis import time from fastapi import FastAPI, Request, HTTPException app FastAPI() # Redis连接 redis_client redis.Redis(hostlocalhost, port6379, db0) def check_rate_limit(ip: str, limit: int 100, window: int 60) - bool: 使用Redis实现滑动窗口限流 current_time time.time() key frate_limit:{ip} # 使用Redis有序集合存储请求时间戳 redis_client.zremrangebyscore(key, 0, current_time - window) # 获取当前窗口内的请求数量 request_count redis_client.zcard(key) if request_count limit: return False # 添加当前请求 redis_client.zadd(key, {str(current_time): current_time}) redis_client.expire(key, window) return True app.middleware(http) async def redis_rate_limit_middleware(request: Request, call_next): client_ip request.client.host if not check_rate_limit(client_ip): raise HTTPException( status_code429, detailToo many requests. Please try again later. ) return await call_next(request)结合IP白名单和限流在实际应用中通常需要同时使用IP白名单和限流from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse import time from collections import defaultdict app FastAPI() # 配置 ALLOWED_IPS [127.0.0.1, 192.168.1.0/24] RATE_LIMITS { default: {limit: 100, window: 60}, # 默认限流 premium: {limit: 1000, window: 60}, # 高级用户 } request_counts defaultdict(list) def is_ip_allowed(ip: str) - bool: 检查IP是否在白名单中 import ipaddress for allowed_ip in ALLOWED_IPS: if / in allowed_ip: # CIDR范围检查 if ipaddress.ip_address(ip) in ipaddress.ip_network(allowed_ip): return True elif ip allowed_ip: return True return False def get_rate_limit_config(ip: str): 根据IP获取限流配置 # 这里可以根据IP或用户身份返回不同的限流配置 return RATE_LIMITS[default] app.middleware(http) async def security_middleware(request: Request, call_next): client_ip request.client.host # 1. IP白名单检查 if not is_ip_allowed(client_ip): return JSONResponse( status_code403, content{detail: Access denied} ) # 2. 限流检查 config get_rate_limit_config(client_ip) current_time time.time() # 清理过期记录 request_counts[client_ip] [ req_time for req_time in request_counts[client_ip] if current_time - req_time config[window] ] # 检查限流 if len(request_counts[client_ip]) config[limit]: raise HTTPException( status_code429, detailfRate limit exceeded. Limit: {config[limit]} requests per {config[window]} seconds ) # 记录请求 request_counts[client_ip].append(current_time) # 处理请求 response await call_next(request) # 添加安全头部 response.headers[X-Content-Type-Options] nosniff response.headers[X-Frame-Options] DENY response.headers[X-XSS-Protection] 1; modeblock return response使用第三方库简化实现虽然可以自己实现限流和IP过滤但使用成熟的第三方库通常更可靠1. slowapi - 专业的限流库from fastapi import FastAPI from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded app FastAPI() # 初始化限流器 limiter Limiter(key_funcget_remote_address) app.state.limiter limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.get(/home) limiter.limit(5/minute) async def homepage(request): return {message: Hello World} app.get(/api/data) limiter.limit(100/day) async def get_data(request): return {data: sensitive information}2. fastapi-limiter - 专门为FastAPI设计的限流库from fastapi import FastAPI from fastapi_limiter import FastAPILimiter from fastapi_limiter.depends import RateLimiter import redis app FastAPI() app.on_event(startup) async def startup(): redis_client redis.Redis(hostlocalhost, port6379, db0) await FastAPILimiter.init(redis_client) app.get(/, dependencies[RateLimiter(times10, seconds60)]) async def index(): return {message: Hello World}最佳实践建议1. 分层限流策略# 不同端点的不同限流策略 RATE_LIMIT_CONFIG { /api/public/: {limit: 100, window: 60}, /api/auth/: {limit: 10, window: 60}, # 登录接口更严格 /api/admin/: {limit: 1000, window: 60}, # 管理员接口更高限制 }2. 动态IP白名单import sqlite3 from contextlib import contextmanager contextmanager def get_db_connection(): conn sqlite3.connect(ip_whitelist.db) try: yield conn finally: conn.close() def is_ip_allowed_dynamic(ip: str) - bool: 从数据库动态检查IP白名单 with get_db_connection() as conn: cursor conn.cursor() cursor.execute(SELECT 1 FROM allowed_ips WHERE ip ?, (ip,)) return cursor.fetchone() is not None3. 监控和日志记录import logging from datetime import datetime logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) app.middleware(http) async def logging_middleware(request: Request, call_next): start_time datetime.now() client_ip request.client.host try: response await call_next(request) process_time (datetime.now() - start_time).total_seconds() logger.info( f{client_ip} - {request.method} {request.url.path} - fStatus: {response.status_code} - Time: {process_time:.3f}s ) return response except Exception as e: logger.error( f{client_ip} - {request.method} {request.url.path} - fError: {str(e)} ) raise部署注意事项1. 反向代理后的真实IP当使用Nginx等反向代理时需要配置获取真实客户端IPdef get_real_ip(request: Request) - str: 获取真实客户端IP考虑反向代理 # 优先从X-Forwarded-For头部获取 forwarded_for request.headers.get(X-Forwarded-For) if forwarded_for: # X-Forwarded-For可能包含多个IP取第一个 return forwarded_for.split(,)[0].strip() # 回退到client.host return request.client.host2. Docker容器化配置# Dockerfile示例 FROM python:3.9 WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . # 设置环境变量 ENV ALLOWED_IPS127.0.0.1,192.168.1.0/24 ENV RATE_LIMIT100 ENV RATE_WINDOW60 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]测试你的安全配置编写测试确保限流和IP白名单正常工作from fastapi.testclient import TestClient from main import app client TestClient(app) def test_ip_whitelist(): # 测试允许的IP response client.get(/protected, headers{X-Forwarded-For: 127.0.0.1}) assert response.status_code 200 # 测试不允许的IP response client.get(/protected, headers{X-Forwarded-For: 8.8.8.8}) assert response.status_code 403 def test_rate_limiting(): # 发送大量请求测试限流 for i in range(101): # 超过100次限制 response client.get(/api/data) if i 100: assert response.status_code 429 else: assert response.status_code 200总结FastAPI提供了灵活的中间件系统使得实现IP白名单和限流变得简单而高效。通过本文介绍的方法你可以✅ 实现基于IP的访问控制✅ 配置灵活的限流策略✅ 结合使用IP白名单和限流✅ 使用成熟的第三方库简化实现✅ 在生产环境中部署和测试安全配置记住安全是一个持续的过程。定期审查你的IP白名单、调整限流策略并监控异常访问模式这样才能确保你的FastAPI应用始终安全可靠。️通过合理的限流和IP白名单配置你不仅可以保护API免受恶意攻击还能确保合法用户获得稳定可靠的服务体验。开始实施这些安全措施让你的FastAPI应用更加健壮吧【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考