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 的内置中间件分成两种接线模式理解这一点可以看懂后文所有示例中间件启用入口对应配置类CORSLitestar(cors_config...)CORSConfigCSRFLitestar(csrf_config...)CSRFConfigAllowed HostsLitestar(allowed_hosts...)AllowedHostsConfig压缩Litestar(compression_config...)CompressionConfig速率限制Litestar(middleware[config.middleware])RateLimitConfigSessionLitestar(middleware[config.middleware])CookieBackendConfig / ServerSideSessionConfig前四类通过Litestar构造器的专用关键字参数注入对应实现事实来自各 config dataclass 的文档字符串速率限制与会则遵循统一的DefineMiddleware模式即config.middleware属性返回一个可放入middleware列表的声明对象。下文按文档顺序逐一展开。CORS跨域资源共享CORSCross-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_configcors_config)CORSConfig 完整参数结合 litestar/config/cors.py 中CORSConfigdataclassL78-L192的定义各字段默认值与含义如下参数默认值说明allow_origins[*]允许的来源列表路径任意组件中可用*通配如domain.*映射到Access-Control-Allow-Originallow_methods[*]允许的 HTTP 方法映射到Access-Control-Allow-Methodsallow_headers[*]允许的请求头构造时会被统一转为小写映射到Access-Control-Allow-Headersallow_credentialsFalse是否设置Access-Control-Allow-Credentialsallow_origin_regexNone额外的来源正则与allow_origins合并匹配expose_headers[]通过Access-Control-Expose-Headers暴露的响应头max_age600预检响应缓存 TTL秒映射到Access-Control-Max-Age源码级行为解析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跨站请求伪造防护CSRFCross-site Request Forgery攻击利用用户处于活跃会话这一事实诱导其浏览器向目标应用发出攻击者构造的请求使应用将其误认为授权操作。官方文档给出的典型攻击报文如下POST /send-money HTTP/1.1 Host: target.web.app Content-Type: application/x-www-form-urlencoded amount1000usdtoattackerevil.com工作机制Litestar 的 CSRF 中间件通过双提交 Cookie HMAC 签名两步工作首个安全请求如 GET时服务端生成一个特殊 token 并写入 Cookie后续每个不安全请求如 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(secretmy-secret) app Litestar([get_resource, create_resource], csrf_configcsrf_config)如需自定义 Cookie 名与请求头名csrf_config CSRFConfig( secretmy-secret, cookie_namesome-cookie-name, header_namesome-header-name, )CSRFConfig 完整参数依据 litestar/config/csrf.pyL13-L42参数默认值说明secret必填用于对 token 做 HMAC 签名的密钥字符串cookie_namecsrftokenCSRF Cookie 名称cookie_path/Cookie 路径header_namex-csrftoken校验时读取的请求头名称cookie_secureFalse是否设置 Cookie 的Secure属性cookie_httponlyFalse是否设置HttpOnly属性cookie_samesitelaxlax/strict/nonecookie_domainNone限定哪些主机可收到该 Cookiesafe_methods{GET, HEAD, OPTIONS}允许下发 token Cookie 的安全方法集合excludeNone跳过 CSRF 校验的路径模式单个或列表exclude_from_csrf_keyexclude_from_csrf路由上用于单独禁用 CSRF 的标识键注意表单字段名当前不可配置只能使用固定键_csrf_token见 litestar/middleware/csrf.pyform.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_hashL39-L49最终 token 共 128 个十六进制字符安全方法分支L122-L124若 Cookie 已存在则复用原 token否则生成新 token并通过包装 ASGIsend函数create_send_wrapperL135-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_csrfTrue选项豁免对应exclude_from_csrf_key机制post(/post, exclude_from_csrfTrue) def handler() - None: ...若要批量豁免多条路径使用CSRFConfig.exclude关键字参数它接受路径模式列表CSRFConfig(secretmy-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_hostsAllowedHostsConfig( allowed_hosts[*.example.com, www.wikipedia.org] ), )AllowedHostsConfig 完整参数依据 litestar/config/allowed_hosts.pyL15-L43参数默认值说明allowed_hosts[*]受信任主机列表*.example.com允许所有子域*允许全部excludeNone跳过校验的路径模式单个或列表exclude_opt_keyNone路由上禁用主机检查的标识键scopesNone处理的 ASGI scope为None时http与websocket都处理www_redirectTrue是否把www.前缀且其余部分匹配受信任主机的请求重定向到www.版本关于通配符官方文档有明确约束*.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若白名单含*构造器直接返回不做任何检查allowed_hosts_regex保持None请求原样放行否则把每个*.domain编译为.*\.domain$形式的正则其余域名精确转义合并成一个fullmatch正则运行期读取请求Host头并剥掉端口host.split(:)[0]全串匹配命中即放行未命中但命中www.剥离后的重定向正则时用ASGIRedirectResponse将请求 307 式重定向到www.版本url.with_replacements(netlocfwww.{url.netloc})其他情况返回 400响应体为{message:invalid host header}。响应压缩gzip、Brotli、ZstdHTTP 响应可选压缩。Litestar 支持 gzip、brotli 与 zstd 三种后端gzip 开箱即用Brotli 需安装brotliextrapip install litestar[brotli]Zstd 需zstdextrapip install litestar[zstd]依赖backports.zstd包。启用方式是向compression_config传入 CompressionConfig 实例并设置backend。GZIPfrom litestar import Litestar from litestar.config.compression import CompressionConfig app Litestar( route_handlers[...], compression_configCompressionConfig(backendgzip, gzip_compress_level9), )gzip 专属参数minimum_size启用压缩的最小响应字节阈值更小的响应不压缩默认500半 KBgzip_compress_level取值 0-9语义同 Python 标准库gzip默认9最大压缩级别。Brotliapp Litestar( route_handlers[...], compression_configCompressionConfig(backendbrotli, brotli_gzip_fallbackTrue), )Brotli 专属参数minimum_size同上默认500brotli_quality范围 [0-11]控制压缩速度与压缩率权衡越高越慢默认5brotli_modegeneric混合内容/textUTF-8 文本/fontWOFF 2.0默认textbrotli_lgwin窗口大小的以 2 为底的对数范围 [10-24]默认22brotli_lgblock最大输入块大小的以 2 为底的对数范围 [16-24]设为0时由 quality 决定默认0brotli_gzip_fallback客户端不支持 brotli 时是否回退 gzip默认True。Zstdapp Litestar( route_handlers[...], compression_configCompressionConfig(backendzstd, zstd_gzip_fallbackTrue), )Zstd 专属参数minimum_size同上默认500zstd_compress_level 0的整数值越大压缩比越高但越慢0表示使用库默认级别通常是 3默认0zstd_gzip_fallback客户端不支持 zstd 时是否回退 gzip默认True。源码级配置校验litestar/config/compression.py 的__post_init__L71-L100会在构造期做严格校验非法配置直接抛出ImproperlyConfiguredExceptionminimum_size 0被拒绝gzip后端要求gzip_compress_level在 0-9brotli后端要求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。速率限制RateLimitMiddlewareLitestar 提供可选的 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_typeMediaType.TEXT, sync_to_threadFalse) 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 完整参数参数默认值说明rate_limit必填(unit, count)元组如(minute, 10)excludeNone跳过限流的路径模式exclude_opt_keyNone路由级禁用的标识键identifier_for_requestget_remote_address从请求提取限流标识的可调用对象check_throttle_handlerNone返回 bool 决定是否对某请求执行限流检查middleware_classRateLimitMiddleware使用的中间件类set_rate_limit_headersTrue是否在响应上写限流头rate_limit_policy_header_keyRateLimit-Policy策略头键名rate_limit_remaining_header_keyRateLimit-Remaining剩余配额头键名rate_limit_reset_header_keyRateLimit-Reset窗口重置倒计时头键名rate_limit_limit_header_keyRateLimit-Limit总配额头键名storerate_limit使用的 Store 名称经app.stores.get(name)解析工作机制与限流响应头从 litestar/middleware/rate_limit.py 的__call__L80-L113可以看到中间件按限流标识 路由生成存储键挂载路由会追加::mount后缀在anyio.Lock()保护下从 Store 读取时间戳历史CacheObject窗口过期即重置历史长度达到max_requests时抛出TooManyRequestsException429否则把当前时间戳压入历史并写回 Store过期时间等于窗口长度实现基于滑动窗口的计数。开启set_rate_limit_headers时create_response_headersL192-L213会在每个响应上注入四个头键名可配置RateLimit-Policy: {limit}; w{窗口秒数}RateLimit-Limit: {limit}RateLimit-Remaining: 剩余请求数RateLimit-Reset: 距窗口重置的秒数在反向代理后面使用默认模式用客户端 IP 作为唯一标识。应用若跑在代理后面看到的地址是代理的地址而非终端用户。虽然代理会设置X-FORWARDED-FOR等头但这些头不能隐式信任——任何客户端都可伪造攻击者每请求换一个随机地址即可绕过限流源码中get_remote_addressL48-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.pyfrom 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(secreturandom(16)) get(/session, sync_to_threadFalse) def check_session_handler(request: Request) - dict[str, bool]: Handler function that accesses request.session. return {has_session: request.session ! {}} post(/session, sync_to_threadFalse) 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_threadFalse) 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.pyfrom os import urandom from litestar import Litestar from litestar.middleware.session.client_side import CookieBackendConfig session_config CookieBackendConfig(secreturandom(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.pyfrom 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(pathPath(session_data))}, )换成 Redis/Valkey/内存存储时只需替换stores中注册的实现详见 docs/usage/stores.rst 与 ServerSideSessionConfig。Logging 中间件内置日志中间件LoggingMiddleware实现见 litestar/middleware/logging.py的文档内容已整体迁移官方指引指向 docs/usage/logging.rst本篇不再展开。小结与延伸阅读安全三件套CORSlitestar/config/cors.py、CSRFlitestar/middleware/csrf.py、Allowed Hostslitestar/middleware/allowed_hosts.py都通过构造器专用参数启用配置类自带取值校验非法配置在应用启动前即暴露性能与防护压缩gzip/brotli/zstdlitestar/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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考