Litestar 内置中间件详解：CORS、CSRF、Allowed Hosts、压缩、限流与会话的配置与实现原理

Litestar 内置中间件详解：CORS、CSRF、Allowed Hosts、压缩、限流与会话的配置与实现原理

【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar

本文以 Litestar 官方文档 docs/usage/middleware/builtin-middleware.rst 为主体，系统讲解框架自带的六类内置中间件：CORS 跨域、CSRF 防伪造、Allowed Hosts 主机白名单、响应压缩（gzip/brotli/zstd）、速率限制（Rate Limit）以及客户端/服务端 Session。结合 litestar/middleware/ 目录下的源码实现，每个中间件都会给出可直接复制运行的配置示例、完整参数说明（含默认值与取值范围），以及底层校验逻辑与调用链，帮助你在生产应用中正确启用并调优这些安全与性能能力。

内置中间件的两种启用方式

Litestar 的内置中间件分成两种接线模式，理解这一点可以看懂后文所有示例：

前四类通过Litestar构造器的专用关键字参数注入（对应实现事实来自各 config dataclass 的文档字符串）；速率限制与会则遵循统一的DefineMiddleware模式，即config.middleware属性返回一个可放入middleware列表的声明对象。下文按文档顺序逐一展开。

CORS：跨域资源共享

CORS（Cross-Origin Resource Sharing）是浏览器端常见的跨域安全机制，通常用中间件实现。在 Litestar 中启用它只需向Litestar构造器传入一个 CORSConfig 实例：

from litestar import Litestar from litestar.config.cors import CORSConfig cors_config = CORSConfig(allow_origins=["https://www.example.com"]) app = Litestar(route_handlers=[...], cors_config=cors_config)

CORSConfig 完整参数

结合 litestar/config/cors.py 中CORSConfigdataclass（L78-L192）的定义，各字段默认值与含义如下：

源码级行为解析

CORS 的响应头在配置实例化时就被预构建并缓存（cached_property），运行期只做轻量判断，具体规则见 litestar/config/cors.py：

• 预检（preflight）响应头（_build_preflight_headers）：始终携带Access-Control-Max-Age；仅当allow_origins含*时才写Access-Control-Allow-Origin: *，否则写Vary: Origin，以便缓存按来源区分。allow_headers不为*时，框架会把允许列表与内置常量DEFAULT_ALLOWED_CORS_HEADERS（定义于 litestar/constants.py）取并集后输出，常见默认头无需手工罗列。allow_methods为*时展开为DELETE, GET, HEAD, OPTIONS, PATCH, POST, PUT全集。
• 普通（simple）响应头（_build_simple_headers）：只包含Allow-Origin、Allow-Credentials与Expose-Headers。
• 来源匹配：allowed_origins_regex属性把allow_origins逐条转义后把*替换为.*，再与allow_origin_regex用|合并编译；is_origin_allowed()对 Origin 值做fullmatch全串匹配。也就是说*.example.com能匹配任意子域，但不会误匹配前缀相同的其他域。

内部处理流程的入口在 litestar/middleware/_internal/cors.py，测试覆盖位于 tests/e2e/test_cors/，可对照阅读。

CSRF：跨站请求伪造防护

CSRF（Cross-site Request Forgery）攻击利用用户处于活跃会话这一事实，诱导其浏览器向目标应用发出攻击者构造的请求，使应用将其误认为授权操作。官方文档给出的典型攻击报文如下：

POST /send-money HTTP/1.1 Host: target.web.app Content-Type: application/x-www-form-urlencoded amount=1000usd&to=attacker@evil.com

工作机制

Litestar 的 CSRF 中间件通过"双提交 Cookie + HMAC 签名"两步工作：

1. 首个"安全"请求（如 GET）时，服务端生成一个特殊 token 并写入 Cookie；
2. 后续每个"不安全"请求（如 POST），中间件校验请求中必须携带与该 Cookie 匹配的 token，来源可以是自定义请求头或固定表单字段。

启用方式——传入 CSRFConfig 实例：

from litestar import Litestar, get, post from litestar.config.csrf import CSRFConfig @get() async def get_resource() -> str: # GET 属于安全方法 return "some_resource" @post("{id:int}") async def create_resource(id: int) -> bool: # POST 属于不安全方法 return True csrf_config = CSRFConfig(secret="my-secret") app = Litestar([get_resource, create_resource], csrf_config=csrf_config)

如需自定义 Cookie 名与请求头名：

csrf_config = CSRFConfig( secret="my-secret", cookie_name="some-cookie-name", header_name="some-header-name", )

CSRFConfig 完整参数

依据 litestar/config/csrf.py（L13-L42）：

注意：表单字段名当前不可配置，只能使用固定键"_csrf_token"（见 litestar/middleware/csrf.py，form.get("_csrf_token", None)硬编码）。

客户端如何携带 token

任何支持 Cookie 持久化的 HTTP 客户端（如requests或httpx的 Session/Client）都可以访问受保护路由。官方文档给出的httpx.Client示例：

import httpx with httpx.Client() as client: get_response = client.get("http://localhost:8000/") # "csrftoken" 是默认 cookie 名 csrf = get_response.cookies["csrftoken"] # "x-csrftoken" 是默认 header 名 post_response_using_header = client.post("http://localhost:8000/1", headers={"x-csrftoken": csrf}) assert post_response_using_header.status_code == 201 # "_csrf_token" 是默认且不可配置的 form-data 键 post_response_using_form_data = client.post("http://localhost:8000/1", data={"_csrf_token": csrf}) assert post_response_using_form_data.status_code == 201 # 虽然有 header，但该请求会话中没有 cookie，因此会失败 # 注意这里用的是 httpx.post 而非 client.post post_response_with_no_persisted_cookie = httpx.post( "http://localhost:8000/1", headers={"x-csrftoken": csrf} ) assert post_response_with_no_persisted_cookie.status_code == 403 assert "CSRF token verification failed" in post_response_with_no_persisted_cookie.text

源码级 token 生成与校验

litestar/middleware/csrf.py 揭示了完整的令牌结构：

• generate_csrf_token()（L52-L63）：用secrets.token_hex(32)生成 64 位十六进制随机串，再拼接它在secret下的 HMAC-SHA256 摘要（generate_csrf_hash，L39-L49），最终 token 共 128 个十六进制字符；
• 安全方法分支（L122-L124）：若 Cookie 已存在则复用原 token，否则生成新 token，并通过包装 ASGIsend函数（create_send_wrapper，L135-L161）在http.response.start阶段注入Set-Cookie头，保证 token 只签发一次且随响应下发；
• 不安全方法分支（L125-L133）：要求请求 token 与 Cookie token 同时非空且通过_csrf_tokens_match校验——分别解出两段的随机部分并验证 HMAC，最后用secrets.compare_digest做恒定时间比较。任一失败即抛出PermissionDeniedException("CSRF token verification failed")，对应上面示例中的 403 响应。

按路由豁免

单个路由可通过 handler 的exclude_from_csrf=True选项豁免（对应exclude_from_csrf_key机制）：

@post("/post", exclude_from_csrf=True) def handler() -> None: ...

若要批量豁免多条路径，使用CSRFConfig.exclude关键字参数，它接受路径模式列表：

CSRFConfig(secret="my-secret", exclude=["/webhook/*"])

Allowed Hosts：主机名白名单

另一个常见安全机制是要求每个进入请求携带Host头，并限制其属于受信任域名集合——即"allowed hosts"。Litestar 提供 AllowedHostsMiddleware，通过传入 AllowedHostsConfig 实例（或域名列表）到Litestar启用：

from litestar import Litestar from litestar.config.allowed_hosts import AllowedHostsConfig app = Litestar( route_handlers=[...], allowed_hosts=AllowedHostsConfig( allowed_hosts=["*.example.com", "www.wikipedia.org"] ), )

AllowedHostsConfig 完整参数

依据 litestar/config/allowed_hosts.py（L15-L43）：

关于通配符，官方文档有明确约束：*.example.com可匹配www.example.com、x.y.z.example.com等任意深度子域；直接写*等于允许全部，与关闭中间件等价（这种情况下建议干脆不启用）；通配符只能出现在域名前缀，放在中间或结尾会在配置构造时抛出ImproperlyConfiguredException（见 litestar/config/allowed_hosts.py 的__post_init__校验）。

源码级校验流程

litestar/middleware/allowed_hosts.py 的处理逻辑（L24-L80）：

1. 若白名单含*，构造器直接返回不做任何检查（allowed_hosts_regex保持None），请求原样放行；
2. 否则把每个*.domain编译为.*\.domain$形式的正则，其余域名精确转义，合并成一个fullmatch正则；
3. 运行期读取请求Host头并剥掉端口（host.split(":")[0]），全串匹配命中即放行；
4. 未命中但命中www.剥离后的重定向正则时，用ASGIRedirectResponse将请求 307 式重定向到www.版本（url.with_replacements(netloc=f"www.{url.netloc}")）；
5. 其他情况返回 400，响应体为{"message":"invalid host header"}。

响应压缩：gzip、Brotli、Zstd

HTTP 响应可选压缩。Litestar 支持 gzip、brotli 与 zstd 三种后端：gzip 开箱即用，Brotli 需安装brotliextra（pip install 'litestar[brotli]'），Zstd 需zstdextra（pip install 'litestar[zstd]'，依赖backports.zstd包）。启用方式是向compression_config传入 CompressionConfig 实例，并设置backend。

GZIP

from litestar import Litestar from litestar.config.compression import CompressionConfig app = Litestar( route_handlers=[...], compression_config=CompressionConfig(backend="gzip", gzip_compress_level=9), )

gzip 专属参数：

• minimum_size：启用压缩的最小响应字节阈值，更小的响应不压缩，默认500（半 KB）；
• gzip_compress_level：取值 0-9，语义同 Python 标准库gzip，默认9（最大压缩级别）。

Brotli

app = Litestar( route_handlers=[...], compression_config=CompressionConfig(backend="brotli", brotli_gzip_fallback=True), )

Brotli 专属参数：

• minimum_size：同上，默认500；
• brotli_quality：范围 [0-11]，控制压缩速度与压缩率权衡，越高越慢，默认5；
• brotli_mode："generic"（混合内容）/"text"（UTF-8 文本）/"font"（WOFF 2.0），默认"text"；
• brotli_lgwin：窗口大小的以 2 为底的对数，范围 [10-24]，默认22；
• brotli_lgblock：最大输入块大小的以 2 为底的对数，范围 [16-24]；设为0时由 quality 决定，默认0；
• brotli_gzip_fallback：客户端不支持 brotli 时是否回退 gzip，默认True。

Zstd

app = Litestar( route_handlers=[...], compression_config=CompressionConfig(backend="zstd", zstd_gzip_fallback=True), )

Zstd 专属参数：

• minimum_size：同上，默认500；
• zstd_compress_level：>= 0的整数，值越大压缩比越高但越慢；0表示使用库默认级别（通常是 3），默认0；
• zstd_gzip_fallback：客户端不支持 zstd 时是否回退 gzip，默认True。

源码级配置校验

litestar/config/compression.py 的__post_init__（L71-L100）会在构造期做严格校验，非法配置直接抛出ImproperlyConfiguredException：

• minimum_size <= 0被拒绝；
• gzip后端要求gzip_compress_level在 0-9；
• brotli后端要求brotli_quality0-11、brotli_lgwin10-24，并在此分支中动态导入 BrotliCompression 作为压缩 facade，同时把brotli_gzip_fallback同步到通用的gzip_fallback标志；
• zstd后端从 ZstdCompression 读取upper_bound上限来校验zstd_compress_level，避免硬编码上限随依赖版本失效。

抽象层与门面实现位于 litestar/middleware/compression/：facade.py定义压缩门面协议，gzip_facade.py/brotli_facade.py/zstd_facade.py是三种具体实现，middleware.py则是负责Content-Encoding协商与实际压缩的 CompressionMiddleware。

速率限制：RateLimitMiddleware

Litestar 提供可选的 RateLimitMiddleware，遵循 IETF RateLimit 草案的响应头规范。官方示例文件 docs/examples/middleware/rate_limit.py 如下：

from litestar import Litestar, MediaType, get from litestar.middleware.rate_limit import RateLimitConfig rate_limit_config = RateLimitConfig(rate_limit=("minute", 1), exclude=["/schema"]) @get("/", media_type=MediaType.TEXT, sync_to_thread=False) def handler() -> str: """Handler which should not be accessed more than once per minute.""" return "ok" app = Litestar(route_handlers=[handler], middleware=[rate_limit_config.middleware])

唯一必填项是rate_limit：一个二元组(时间单位, 配额整数)，单位只能是"second"、"minute"、"hour"、"day"。源码中 litestar/middleware/rate_limit.py 定义了单位换算表DURATION_VALUES = {"second": 1, "minute": 60, "hour": 3600, "day": 86400}，用于滑动窗口重置与存储过期时间。

RateLimitConfig 完整参数

工作机制与限流响应头

从 litestar/middleware/rate_limit.py 的__call__（L80-L113）可以看到：中间件按"限流标识 + 路由"生成存储键（挂载路由会追加::mount后缀），在anyio.Lock()保护下从 Store 读取时间戳历史（CacheObject），窗口过期即重置；历史长度达到max_requests时抛出TooManyRequestsException（429），否则把当前时间戳压入历史并写回 Store（过期时间等于窗口长度，实现基于滑动窗口的计数）。

开启set_rate_limit_headers时，create_response_headers（L192-L213）会在每个响应上注入四个头（键名可配置）：

• RateLimit-Policy: {limit}; w={窗口秒数}
• RateLimit-Limit: {limit}
• RateLimit-Remaining: 剩余请求数
• RateLimit-Reset: 距窗口重置的秒数

在反向代理后面使用

默认模式用客户端 IP 作为唯一标识。应用若跑在代理后面，看到的地址是代理的地址而非终端用户。虽然代理会设置X-FORWARDED-FOR等头，但这些头不能隐式信任——任何客户端都可伪造，攻击者每请求换一个随机地址即可绕过限流（源码中get_remote_address（L48-L57）刻意不读取X-FORWARDED-FOR，无客户端信息时回退为127.0.0.1）。

推荐做法是叠加一个安全更新客户端地址的 ASGI 层，例如 uvicorn 的ProxyHeaderMiddleware或 hypercorn 的ProxyFixMiddleware，让 Litestar 拿到的request.client.host才是真实来源；也可通过identifier_for_request传入自定义函数，从认证令牌、可信头中派生标识。

Session 中间件：客户端与服务端会话

Litestar 提供 SessionMiddleware，同时支持客户端与服务端两类会话。服务端会话基于 Litestar 的 stores 体系，支持内存、文件、Redis、Valkey 四种存储后端。

基本设置

创建任意后端配置对象，把config.middleware加入应用中间件栈即可。完整示例（docs/examples/middleware/session/cookies_full_example.py）：

from os import urandom from litestar import Litestar, Request, delete, get, post from litestar.middleware.session.client_side import CookieBackendConfig # 用 16 字节（128 bit）密钥初始化；生产环境应从环境变量注入 session_config = CookieBackendConfig(secret=urandom(16)) @get("/session", sync_to_thread=False) def check_session_handler(request: Request) -> dict[str, bool]: """Handler function that accesses request.session.""" return {"has_session": request.session != {}} @post("/session", sync_to_thread=False) def create_session_handler(request: Request) -> None: """Handler to set the session.""" if not request.session: # value 可以是 dict 或 pydantic model request.set_session({"username": "moishezuchmir"}) @delete("/session", sync_to_thread=False) def delete_session_handler(request: Request) -> None: """Handler to clear the session.""" if request.session: request.clear_session() app = Litestar( route_handlers=[check_session_handler, create_session_handler, delete_session_handler], middleware=[session_config.middleware], )

由于客户端与服务端会话都依赖 Cookie（前者存会话数据，后者存会话 ID），二者共享大部分 Cookie 配置项，完整字段参考 BaseBackendConfig。

客户端会话（Client-side）

通过 ClientSideSessionBackend 提供，会话数据加密后整体放入 Cookie，支持 Cookie 拆分以应对浏览器 4 KB 限制。

重要：ClientSideSessionBackend依赖cryptography库，可作为 extra 一并安装：pip install 'litestar[cryptography]'。

最小示例（docs/examples/middleware/session/cookie_backend.py）：

from os import urandom from litestar import Litestar from litestar.middleware.session.client_side import CookieBackendConfig session_config = CookieBackendConfig(secret=urandom(16)) app = Litestar(middleware=[session_config.middleware])

更多 Cookie 配置项（secret、Cookie 的 path/secure/samesite 等）见 CookieBackendConfig。

服务端会话（Server-side）

服务端会话把数据存在服务端而非 Cookie 中：浏览器只持有一个随机生成的会话 ID Cookie，中间件用它到 Store 中加载对应数据。以文件存储为例（docs/examples/middleware/session/file_store.py）：

from pathlib import Path from litestar import Litestar from litestar.middleware.session.server_side import ServerSideSessionConfig from litestar.stores.file import FileStore app = Litestar( middleware=[ServerSideSessionConfig().middleware], stores={"sessions": FileStore(path=Path("session_data"))}, )

换成 Redis/Valkey/内存存储时，只需替换stores中注册的实现，详见 docs/usage/stores.rst 与 ServerSideSessionConfig。

Logging 中间件

内置日志中间件（LoggingMiddleware，实现见 litestar/middleware/logging.py）的文档内容已整体迁移，官方指引指向 docs/usage/logging.rst，本篇不再展开。

小结与延伸阅读

• 安全三件套：CORS（litestar/config/cors.py）、CSRF（litestar/middleware/csrf.py）、Allowed Hosts（litestar/middleware/allowed_hosts.py）都通过构造器专用参数启用，配置类自带取值校验，非法配置在应用启动前即暴露；
• 性能与防护：压缩（gzip/brotli/zstd，litestar/config/compression.py）与限流（litestar/middleware/rate_limit.py）互补——前者减小传输体积，后者基于 Store 的滑动窗口防止滥用，且限流标识的代理陷阱需在部署时显式处理；
• 会话：客户端会话（加密 Cookie，需cryptographyextra）与服务端会话（Session ID + Store，支持内存/文件/Redis/Valkey）按场景选型，二者共享BaseBackendConfig的 Cookie 配置面；
• 想要自定义中间件或理解中间件栈顺序，继续阅读 docs/usage/middleware/creating-middleware.rst 与 docs/usage/middleware/using-middleware.rst；端到端行为验证可参考 tests/e2e/test_cors/、tests/unit/test_middleware/ 下的测试用例。

【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar