zlynv/drogue
GitHub: zlynv/drogue
drogue 是一个为 Python Web 框架提供生产级限流和 DDoS 防护的库，解决了传统限流方案侵入性强、缺乏 WebSocket 支持和信任区分等问题。
Stars: 1 | Forks: 0
# drogue
为 Python Web 应用提供限流和 DDoS 防护。提供简洁的 API、WebSocket 支持，并内置多层防御机制。
**[阅读文档](https://zlynv.github.io/drogue/)**
## drogue 解决什么问题？
Web 应用需要限流来防止滥用，但现有的解决方案存在以下不足：
1. **签名污染** —— 大多数限流器强制将 `request: Request` 塞进每个函数签名中，将你的业务逻辑与限流器耦合在一起。
2. **缺乏 WebSocket 防护** —— 使用 WebSocket 的实时应用没有内置的限流功能。
3. **缺乏 DDoS 检测** —— 简单的计数器可以捕捉到过度使用，但无法检测到每个客户端都保持在限制之下的分布式攻击。
4. **缺乏信任区分** —— 每个请求都经过相同的评估路径，即使是对已验证的用户也是如此。
drogue 解决了这四个问题。它通过身份（IP、用户、header）进行限流，而无需修改你的函数签名；它可以检测异常的流量模式，并为受信任的客户端提供快速通道。
## drogue 适用于谁？
- **API 开发者** —— 需要限流但不想被框架绑定
- **运行 FastAPI、Django 或 Flask 的团队** —— 希望在三种框架上使用统一的解决方案
- **面临 DDoS 或滥用的平台** —— 需要比简单请求计数更强大的防护
- **具有 WebSocket 连接的应用** —— 需要实时保护
## 工作原理
```
from fastapi import FastAPI
from drogue.adapters.fastapi import DrogueLimiter
app = FastAPI()
limiter = DrogueLimiter(app, default_limits=["100/minute"])
@app.get("/api/data")
@limiter.limit("10/minute")
async def get_data():
return {"data": "value"}
```
不需要 `request: Request` 参数。限流 header 会自动注入。同样的模式也适用于 Flask 和 Django。
**请求流程：**
1. 客户端发送请求
2. drogue 提取客户端 key（IP、用户 ID 或自定义 header）
3. 根据路由匹配规则
4. 算法评估请求（令牌桶、滑动窗口或固定窗口）
5. 运行保护层（DDoS 检查、信任状态、探测器检测）
6. 返回带有限流 header 的响应
## 功能
| 类别 | 你将获得的功能 |
|----------|-------------|
| **限流** | 令牌桶、滑动窗口、固定窗口、感知成本的限流、阻塞模式、突发流量控制 |
| **身份** | 基于 IP、基于用户、基于 header、复合提取器、防伪造 X-Forwarded-For |
| **框架** | FastAPI (ASGI)、Flask (装饰器)、Django (middleware + 装饰器)、Django REST Framework (throttle) |
| **DDoS 检测** | Z-score 异常检测、流式 Sentinel 模型、探测模式检测 |
| **自动封禁** | 封禁时长倍增的渐进式封禁（5分钟至160分钟），阈值可配置 |
| **信任系统** | 状态机（未知/正常/受信任/不受信任/已封禁），已验证用户吞吐量提升 9 倍 |
| **熔断器** | Closed/Open/HalfOpen 状态，自动恢复 |
| **自适应限制** | 基于 CPU 和内存的动态调整，在系统负载下降低限制 |
| **防御随机化** | 会话级差异、蜜罐路径、防指纹追踪 |
| **CIDR 过滤** | 支持从配置或文件加载的允许/阻止列表，兼容 IPv4 和 IPv6 |
| **概率存储** | Count-Min Sketch（1M 个 key 仅需 10MB）、布隆过滤器、布谷鸟过滤器、HyperLogLog |
| **可观测性** | Prometheus 指标、OpenTelemetry 链路追踪、结构化 JSON 日志 |
| **影子模式** | 在不实际生效的情况下测试规则，上线前收集指标 |
## 性能
| 指标 | drogue | 备注 |
|--------|--------|-------|
| 令牌桶 | ~1.4us | 每次获取的中位延迟 |
| 滑动窗口 | ~1.6us | 每次获取的中位延迟 |
| 固定窗口 | ~1.1us | 每次获取的中位延迟 |
| 吞吐量 | 700K+ req/s | 单个 worker，内存存储 |
| 每个 key 的内存占用 | ~150 bytes | 进程内存储 |
*测试环境：Intel Core Ultra 5 225F，Python 3.13，asyncio 单个 worker，内存存储，10万次迭代。*
### 基准测试
运行完整的基准测试套件：
```
pip install locust pytest-benchmark
pip install -e ".[fastapi]"
# 函数级 benchmarks（无需 server）
pytest benchmarks/test_algorithm_latency.py -v --benchmark-only
pytest benchmarks/test_algorithm_throughput.py -v --benchmark-only
pytest benchmarks/test_memory.py -v --benchmark-only
# HTTP load test
python -m uvicorn benchmarks.apps.fastapi_app:app --port 8000
locust -f benchmarks/locustfile.py --headless -u 100 -r 10 --run-time 60s -H http://localhost:8000
# DDoS protection test
python -m uvicorn benchmarks.apps.ddos_app:app --port 8000
locust -f benchmarks/ddos_locustfile.py --headless -u 100 -r 10 --run-time 60s -H http://localhost:8000
```
**测试结果 (Windows, Python 3.13, MemoryStorage)：**
| 测试项 | 结果 |
|------|--------|
| 函数级 ops/sec | 80-88K 次 acquire 调用/秒 |
| HTTP 吞吐量 | 2,400+ req/sec |
| DDoS 防护 | 97.8% 攻击者被封禁，p50：5ms |
## 安装
```
pip install drogue
# 使用 framework extras
pip install drogue[fastapi] # FastAPI + Starlette
pip install drogue[django] # Django
pip install drogue[flask] # Flask
pip install drogue[drf] # Django REST Framework
pip install drogue[redis] # Redis backend
pip install drogue[all] # Everything
```
## 快速入门
**FastAPI:**
```
from fastapi import FastAPI
from drogue.adapters.fastapi import DrogueLimiter
app = FastAPI()
limiter = DrogueLimiter(app, default_limits=["100/minute"])
@app.get("/api/data")
@limiter.limit("10/minute")
async def get_data():
return {"data": "value"}
```
**Flask:**
```
from flask import Flask
from drogue.adapters.flask import DrogueLimiter
app = Flask(__name__)
limiter = DrogueLimiter(app, default_limits=["100/minute"])
@app.route("/api/data")
@limiter.limit("10/minute")
def get_data():
return {"data": "value"}
```
**Django:**
```
# settings.py
MIDDLEWARE = [
"drogue.adapters.django.DrogueMiddleware",
]
# views.py
from drogue.adapters.django import DrogueRateLimiter
from django.http import JsonResponse
limiter = DrogueRateLimiter()
@limiter.limit("10/minute")
def get_data(request):
return JsonResponse({"data": "value"})
```
## 从 slowapi 迁移
```
# Before（slowapi）
from slowapi import Limiter
from slowapi.util import get_remote_address
from starlette.requests import Request
limiter = Limiter(key_func=get_remote_address)
@app.get("/api/data")
@limiter.limit("10/minute")
async def get_data(request: Request):
return {"data": "value"}
# After（drogue）
from drogue.adapters.fastapi import DrogueLimiter
limiter = DrogueLimiter(app)
@app.get("/api/data")
@limiter.limit("10/minute")
async def get_data():
return {"data": "value"}
```
## 已知限制
- 默认情况下，封禁状态仅存储在内存中。计划在 v0.3 版本支持 Redis 持久化。
- 信任缓存是基于单个进程的。多 worker 部署环境需要每个 worker 维护独立的信任状态。
- 对于返回字典的 Flask 视图，其限流 header 不会自动注入。
## 安全
请通过 GitHub 的私有漏洞报告功能报告安全漏洞。不要为安全漏洞开启公开的 issue。响应时间：48小时。
## 路线图
- **v0.1** —— 核心限流、DDoS 检测、支持三大框架（当前版本）
- **v0.2** —— 基于 Redis 的封禁状态、支持 Django 和 Flask 的 WebSocket
- **v0.3** —— 信任缓存跨进程同步、高级 Sentinel 功能
- **v1.0** —— 生产就绪、完整的文档站点
## 开发
```
pip install drogue[dev]
pytest
ruff check src/drogue/
```
## 许可证
MIT 许可证。详情请参阅 [LICENSE](LICENSE)。
由 [Zlynv](https://github.com/Zlynv) 创建
标签：API保护, DDoS防护, Python, WebSocket, Web中间件, 依赖分析, 搜索引擎查询, 无后门, 熔断器, 用户代理, 自定义请求头, 逆向工具, 限流