一,什么是中间件
中间件是一种软件组件,它在请求到达应用程序处理程序之前和/或响应发送回客户端之前执行操作。
请求从客户端发出。
请求首先经过Middleware 1。
然后经过Middleware 2。
请求到达FastAPI路由处理器。
响应从路由处理器返回。
响应经过Middleware 2。
最后经过Middleware 1。
响应返回给客户端。
🌰:
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponseapp = FastAPI()@app.middleware("http")
async def simple_middleware(request: Request, call_next):# 在请求处理之前执行的代码print(f"Received request: {request.method} {request.url}")# 调用下一个中间件或路由处理器response = await call_next(request)# 在响应返回之前执行的代码:如果响应是JSONResponse,我们在响应头中添加一个自定义字段。if isinstance(response, JSONResponse):response.headers["X-Processed-By"] = "SimpleMiddleware"print(f"Processed response: {response.status_code}")return response@app.get("/")
async def root():return {"message": "Hello World"}if '__main__' == __name__:uvicorn.run(app, host='127.0.0.1', port=8088)
在FastAPI中,中间件在很多场景下都非常有用,比如:
- 请求日志记录
- 认证和授权
- 响应修改
- 性能监控
- 跨域资源共享(CORS)处理
二,常用的中间件
(一)CORSMiddleware
1,同源策略与跨域资源共享
同源策略是一个重要的安全概念,由网页浏览器强制执行。它限制了一个源(origin,如果两个 URL 的协议、端口和主机都相同的话,则这两个 URL 是同源的)中加载的文档或脚本如何与来自另一个源的资源进行交互。
- 定义:如果两个URL的协议、域名和端口都相同,则它们被认为是同源的。
- 目的:防止恶意网站读取另一个网站的敏感数据。
- 限制:脚本只能访问来自同一源的数据,不能直接访问不同源的资源。
例如:
- https://example.com/page1 可以访问 https://example.com/page2
- https://example.com 不能直接访问 https://api.example.com 或 http://example.com(不同协议)
CORS是一种机制,它使用额外的 HTTP 头来告诉浏览器让运行在一个源(domain)上的 Web 应用被准许访问来自不同源服务器上的指定的资源。
- 目的:允许服务器声明哪些源可以访问它们的资源,从而放宽同源策略的限制。
- 工作原理:服务器在响应中包含特定的 HTTP 头,告诉浏览器允许跨域请求。
关键的 CORS 头部:
- Access-Control-Allow-Origin: 指定允许访问资源的源。
- Access-Control-Allow-Methods: 指定允许的 HTTP 方法。
- Access-Control-Allow-Headers: 指定允许的请求头。
同源策略和CORS看似矛盾,但实际上它们共同构成了web安全和功能性之间的平衡。为什么在有同源策略的情况下还需要CORS?
-
Web的演变:
- 早期Web:最初的Web主要由静态页面组成,不同源之间的交互很少。- 同源策略的引入:随着Web变得更加动态,同源策略被引入以防止潜在的跨站点脚本攻击(XSS)和数据窃取。- Web 2.0时代:随着AJAX的兴起,Web应用变得更加动态和交互式。- 现代Web:现在的Web充满了单页应用(SPAs)、微服务架构和复杂的API驱动的应用。
-
同源策略的局限性:
- 虽然同源策略提供了重要的安全保护,但它也限制了合法的跨域请求。- 在现代Web应用中,前端和后端经常部署在不同的域上,或者一个应用需要访问多个不同域的API。
-
CORS的必要性:
- 业务需求:公司可能需要在多个子域或完全不同的域之间共享资源。- API经济:许多公司提供API服务,这些API需要被不同域的客户端访问。- 微服务架构:不同的服务可能部署在不同的域上,但需要相互通信。- 开发和测试:开发环境和生产环境可能使用不同的域。
-
CORS如何平衡安全和功能:
- 控制访问:CORS允许服务器明确指定哪些域可以访问其资源。- 细粒度控制:可以控制允许的HTTP方法、头部等。- 预检请求:对于非简单请求,CORS使用预检请求机制,增加了一层安全检查。- 保持同源策略:CORS并没有废除同源策略,而是提供了一种受控的方式来放宽限制。
-
CORS的优势:
- 安全性:虽然允许跨域请求,但CORS通过明确的服务器配置来维护安全性。- 灵活性:开发者可以构建更复杂、分布式的应用架构。- 标准化:CORS提供了一个标准化的方法来处理跨域请求,取代了之前的一些不安全或复杂的变通方法(如JSONP)。
-
实际应用场景:
- 前后端分离:前端可能托管在CDN上,而后端API在不同的域。- 第三方服务集成:如嵌入地图、支付服务或社交媒体小工具。- 多环境部署:开发、测试和生产环境可能使用不同的域。
2,使用CORSMiddleware
以前后端分离项目为例。
在前端,本身不需要特别的 CORS 配置,因为 CORS 主要是由服务器端控制的。但是,你需要确保你的 API 请求使用了正确的 URL,例如:
import axios from 'axios';const api = axios.create({baseURL: 'http://localhost:8000', // 你的 FastAPI 服务器地址
});// 使用方式
api.get('/some-endpoint').then(response => {console.log(response.data);
});
在 fastapi 中,需要配置 CORS 中间件来允许跨源请求:
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()# 配置 CORS
app.add_middleware(CORSMiddleware,allow_origins=["http://localhost:8080"], # 允许的源,这里假设你的前端应用运行在 8080 端口allow_credentials=True,allow_methods=["*"], # 允许所有方法allow_headers=["*"], # 允许所有头
)# 你的路由和其他代码...@app.get("/")
async def root():return {"message": "Hello World"}if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)
(二)GZipMiddleware
1,HTTP 响应压缩
自动压缩 HTTP 响应是一个优化技术,旨在减少传输的数据量,从而加快网页加载速度并减少带宽使用。这种方式尤其适用于文本内容,如 HTML、CSS、JavaScript 和 JSON。
1,确保 Web 服务器(如 Nginx 或 Apache)配置了 Gzip 或 Brotli 压缩。服务器会在发送响应前自动压缩响应数据。
-
在 Nginx 中启用 Gzip 压缩:
http {gzip on;gzip_types text/plain text/css application/json application/javascript text/xml application/xml application/xml+rss text/javascript;gzip_proxied any;gzip_min_length 1000; }
-
在 Apache 中启用 Gzip 压缩:
<IfModule mod_deflate.c>AddOutputFilterByType DEFLATE text/html text/plain text/xml text/css application/json application/javascript text/javascript </IfModule>
2,客户端支持:现代浏览器默认支持 Gzip 和 Brotli 压缩。在请求头中,浏览器会发送 Accept-Encoding 字段,指明支持的压缩算法:
Accept-Encoding: gzip, deflate, br
3,服务器响应:服务器检查 Accept-Encoding
字段,并使用适当的压缩算法对响应内容进行压缩,同时在响应头中添加 Content-Encoding
字段,指明使用的压缩算法:
HTTP/1.1 200 OK
Content-Type: text/html
Content-Encoding: gzip
Content-Length: 512<compressed content>
4,客户端对数据进行解压以恢复原始内容。
2,使用 GZipMiddleware
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponseapp = FastAPI()# 添加 GZip 中间件,在响应大小超过 100000 字节时才进行压缩
app.add_middleware(GZipMiddleware, minimum_size=100000)@app.get("/item/")
async def test():file_path = "02 部署前准备--配置Settings.py.mp4"# # 获取文件大小# import os# file_size = os.path.getsize(file_path)return FileResponse(file_path,media_type="video/mp4",filename=file_path,# headers={"Content-Length": str(file_size)})if '__main__' == __name__:uvicorn.run(app, host='127.0.0.1', port=8088)
(三)TrustedHostMiddleware
1,HTTP Host Header攻击
HTTP Host Header攻击是一种利用 HTTP 请求中 Host 头部的安全漏洞。这种攻击可能导致各种安全问题,包括但不限于网站重定向、缓存污染和密码重置漏洞。
HTTP Host header attacks
如何识别和利用HTTP Host头的漏洞
2,使用 TrustedHostMiddleware
Trusted Host Middleware 是一种安全机制,用于限制应用程序只接受来自特定主机或域名的请求。这是一个重要的安全特性。
工作原理:
- 检查请求头:中间件检查每个incoming请求的 Host 头。
- 比对允许列表:将 Host 头与预先配置的允许主机列表进行比对。
- 处理结果:如果 Host 头匹配允许列表中的一项,请求被允许通过;否则,请求被拒绝。
from fastapi import FastAPI
from starlette.middleware.trustedhost import TrustedHostMiddlewareapp = FastAPI()app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com", "www.example.com"]
)
(四)更多中间件
Starlette 官档 - 中间件 ASGI
Awesome 列表
(六)一些有趣的自定义中间件
- 限流中间件:限制每个IP在特定时间窗口内的请求次数。
- 响应时间模拟中间件:、为每个请求添加随机延迟。用于测试前端应用对不同响应时间的处理能力。可以模拟真实世界的网络延迟,帮助发现潜在的超时问题。
- 请求ID中间件:为每个请求分配一个唯一的ID。方便跟踪和调试请求,特别是在分布式系统中。
- 响应内容修改中间件:修改JSON响应中的特定内容。可以用于统一处理某些响应,如敏感信息脱敏。
- 日志中间件:记录每个请求和响应的详细信息,对于调试和监控非常有用。
- 错误处理中间件:全局捕获异常并自定义错误响应。
- 安全头中间件:添加安全相关的 HTTP 头,提升应用安全性。
- 统计中间件:记录请求的统计信息,如请求数量、响应时间等。
- HTTP 缓存中间件:添加 HTTP 缓存头,提高页面加载速度。
- 请求重试中间件:在特定情况下对请求进行重试。
import asyncio
import logging
import random
import timeimport uvicorn
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddlewareapp = FastAPI()# 1. 限流中间件
class RateLimitMiddleware(BaseHTTPMiddleware):def __init__(self, app, max_requests: int, time_window: int):super().__init__(app)self.max_requests = max_requestsself.time_window = time_windowself.request_counts = {}async def dispatch(self, request: Request, call_next):client_ip = request.client.hostcurrent_time = time.time()if client_ip in self.request_counts:if current_time - self.request_counts[client_ip]["timestamp"] > self.time_window:self.request_counts[client_ip] = {"count": 1, "timestamp": current_time}else:self.request_counts[client_ip]["count"] += 1if self.request_counts[client_ip]["count"] > self.max_requests:return JSONResponse(status_code=429, content={"error": "Too many requests"})else:self.request_counts[client_ip] = {"count": 1, "timestamp": current_time}return await call_next(request)# 2. 响应时间模拟中间件
class ResponseTimeSimulatorMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):# 模拟0.1秒到1秒的随机延迟delay = random.uniform(0.1, 1)await asyncio.sleep(delay)response = await call_next(request)response.headers["X-Simulated-Delay"] = f"{delay:.2f}s"return response# 3. 请求ID中间件
class RequestIDMiddleware(BaseHTTPMiddleware):def __init__(self, app):super().__init__(app)self.request_id_counter = 0async def dispatch(self, request: Request, call_next):self.request_id_counter += 1request_id = f"REQ-{self.request_id_counter:05d}"request.state.request_id = request_idresponse = await call_next(request)response.headers["X-Request-ID"] = request_idreturn response# 4. 响应内容修改中间件
class ResponseModifierMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):response = await call_next(request)if isinstance(response, JSONResponse):content = response.body.decode()modified_content = content.replace("example", "EXAMPLE")return Response(content=modified_content, status_code=response.status_code,headers=dict(response.headers), media_type=response.media_type)return responselogging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)# 5. 日志中间件
class LoggingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):start_time = time.time()response = await call_next(request)process_time = time.time() - start_timelogger.info(f"Request: {request.method} {request.url} completed in {process_time:.4f}s")return response# 6. 错误处理中间件
class ErrorHandlingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):try:response = await call_next(request)return responseexcept Exception as exc:return JSONResponse(content={"error": str(exc)}, status_code=500)# 7. 安全头中间件
class SecurityHeadersMiddleware(BaseHTTPMiddleware):async def dispatch(self, request, call_next):response = await call_next(request)response.headers["Content-Security-Policy"] = "default-src 'self'"response.headers["X-Content-Type-Options"] = "nosniff"response.headers["X-Frame-Options"] = "DENY"response.headers["X-XSS-Protection"] = "1; mode=block"return response# 8. 统计中间件
class StatsMiddleware(BaseHTTPMiddleware):def __init__(self, app):super().__init__(app)self.total_requests = 0self.total_time = 0.0async def dispatch(self, request: Request, call_next):start_time = time.time()response = await call_next(request)process_time = time.time() - start_timeself.total_requests += 1self.total_time += process_timeresponse.headers["X-Total-Requests"] = str(self.total_requests)response.headers["X-Total-Time"] = f"{self.total_time:.2f}s"response.headers["X-Process-Time"] = f"{process_time:.2f}s"return response# 9. HTTP 缓存中间件
class HTTPCacheMiddleware(BaseHTTPMiddleware):async def dispatch(self, request, call_next):response = await call_next(request)response.headers["Cache-Control"] = "public, max-age=3600"return response# 10. 请求重试中间件
class RetryMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):retries = 3for attempt in range(retries):response = await call_next(request)if response.status_code == 200:return responsetime.sleep(1)return response# 添加中间件到应用
app.add_middleware(RateLimitMiddleware, max_requests=5, time_window=60)
app.add_middleware(ResponseTimeSimulatorMiddleware)
app.add_middleware(RequestIDMiddleware)
app.add_middleware(ResponseModifierMiddleware)
app.add_middleware(LoggingMiddleware)
app.add_middleware(ErrorHandlingMiddleware)
app.add_middleware(SecurityHeadersMiddleware)@app.get("/")
async def root():return {"message": "This is an example response"}if '__main__' == __name__:uvicorn.run(app, host='127.0.0.1', port=8088)
三,使用中间件的注意事项
(一)顺序
中间件是按照添加的顺序依次执行的。顺序会影响请求和响应的处理流程,所以要注意中间件的添加顺序。
假设我们有两个中间件,一个是日志记录中间件,另一个是 GZip 压缩中间件。我们希望日志记录在请求处理之前和之后都能记录信息,而压缩应该在响应返回之前进行。
from fastapi import FastAPI, Request
from fastapi.middleware.gzip import GZipMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
import time
import logginglogging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)class LoggingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):start_time = time.time()logger.info(f"Request: {request.method} {request.url}")response = await call_next(request)process_time = time.time() - start_timelogger.info(f"Completed in {process_time:.4f}s")return responseapp = FastAPI()# 日志记录中间件在 GZip 中间件之前
app.add_middleware(LoggingMiddleware)
app.add_middleware(GZipMiddleware, minimum_size=1000)@app.get("/")
async def read_root():return {"message": "Hello, World!"}
资源管理:
- 中间件可能需要管理资源(如数据库连接)。
- 确保正确地打开和关闭资源,考虑使用上下文管理器。
异步操作:
- FastAPI支持异步操作,中间件也应该尽可能是异步的。
- 使用 async/await 语法,避免阻塞操作。
(二)性能
中间件会增加请求处理的开销,要避免添加过多不必要的中间件,尤其是在高并发场景下。
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time
import logging
import oslogging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)class LoggingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):start_time = time.time()logger.info(f"Request: {request.method} {request.url}")response = await call_next(request)process_time = time.time() - start_timelogger.info(f"Completed in {process_time:.4f}s")return responseapp = FastAPI()if os.getenv("ENV") == "development":app.add_middleware(LoggingMiddleware)@app.get("/")
async def read_root():return {"message": "Hello, World!"}
中间件会对每个请求都执行,如果包含耗时操作,会显著影响应用性能。
尽量保持中间件轻量,避免在中间件中执行耗时操作。如果必须,考虑异步操作或缓存策略。
(三)异常处理
中间件中处理的异常不会自动传递给 FastAPI 的全局异常处理器,需要在中间件中捕获并处理异常。
from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponseclass ErrorHandlingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):try:response = await call_next(request)return responseexcept HTTPException as exc:return JSONResponse(content={"error": exc.detail}, status_code=exc.status_code)except Exception as exc:return JSONResponse(content={"error": "Internal Server Error"}, status_code=500)app = FastAPI()
app.add_middleware(ErrorHandlingMiddleware)@app.get("/")
async def read_root():raise Exception("An unexpected error occurred")
中间件可以捕获和处理异常,但也可能掩盖重要的错误信息。
在捕获异常时,确保记录足够的信息用于调试。考虑只捕获特定类型的异常。
(四)状态共享
使用 request.state 可以在中间件和路由处理器之间共享状态信息。
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddlewareclass RequestIDMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):request.state.request_id = "unique-request-id"response = await call_next(request)response.headers["X-Request-ID"] = request.state.request_idreturn responseapp = FastAPI()
app.add_middleware(RequestIDMiddleware)@app.get("/")
async def read_root(request: Request):return {"request_id": request.state.request_id}
读取请求体:
- 读取请求体后,它就被消耗了,后续的处理(包括路由处理函数)将无法再次读取。
- 如果必须在中间件中读取请求体,考虑缓存它或者使用 request.stream()
来允许多次读取。
(五)响应修改
中间件可以修改响应对象,但需要确保不会破坏响应的结构和内容。
class ResponseModifierMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):response = await call_next(request)if isinstance(response, JSONResponse):response.content = {"modified": response.content}return response
修改响应可能会影响应用的预期行为,特别是当多个中间件都修改响应时。
谨慎修改响应,确保修改不会破坏响应的结构或语义。考虑使用装饰器而不是中间件来修改特定路由的响应。