Skip to Content

FastAPI (Python)

FastAPI 专为高性能异步 Web 应用而设计。为了保持高吞吐,我们使用自定义 BaseHTTPMiddleware 与异步 HTTP 客户端 (httpx) 集成 ProxyTracer,确保绝不阻塞底层事件循环。

中间件实现

此实现安全提取真实的客户端 IP,利用异步连接池高效查询 ProxyTracer,并在 proxy 标志返回 true 时返回严格的 403 Forbidden 响应。

import os import httpx from fastapi import FastAPI, Request, status from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware app = FastAPI() class ProxyTracerMiddleware(BaseHTTPMiddleware): def __init__(self, app): super().__init__(app) self.api_key = os.getenv("PROXYTRACER_API_KEY") # 初始化持久化异步客户端以实现连接池复用与极致性能 self.client = httpx.AsyncClient(timeout=0.5) # 严格的 500ms 超时 async def dispatch(self, request: Request, call_next): # 1. 安全提取负载均衡器/反向代理背后的真实 IP forwarded_for = request.headers.get("x-forwarded-for") if forwarded_for: ip = forwarded_for.split(",")[0].strip() else: ip = request.client.host if request.client else None # 跳过本地开发环境 IP if ip and ip not in ["127.0.0.1", "::1", "testclient"]: try: # 2. 异步查询 ProxyTracer API response = await self.client.get( f"https://api.proxytracer.com/v1/check/{ip}", headers={"Authorization": f"Bearer {self.api_key}"} ) if response.status_code == 200: data = response.json() # 3. 若检测到代理,立即丢弃连接 if data.get("proxy") is True: return JSONResponse( status_code=status.HTTP_403_FORBIDDEN, content={"error": "Access Denied: VPN or Proxy detected."} ) except httpx.RequestError as e: # 故障开放(Fail-open):记录网络错误并允许请求继续进行 print(f"ProxyTracer API timeout or failure: {e}") # 4. 流量正常,继续进入路由处理器 return await call_next(request) # 全局应用中间件 app.add_middleware(ProxyTracerMiddleware)

依赖提示:此方案需要 httpx 库以支持异步 HTTP 请求。可通过 pip install httpx 进行安装。

最后更新于