Litestar 安全实现指南认证中间件、内置安全后端、Guards 与密钥数据处理【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestarLitestar 对安全方案本身保持“无主见”agnostic立场你可以自由采用任何标准或非标准的认证机制但框架同时提供了一组内置组件让你可以低摩擦地落地认证Authentication与授权Authorization。本文基于仓库中docs/usage/security/下的安全文档章节及其配套示例与源码实现完整讲解 Litestar 安全体系的四个支柱自定义认证中间件AbstractAuthenticationMiddleware、内置安全后端Session / JWT、基于 Guards 的角色授权、路由级认证排除/包含规则以及SecretString/SecretBytes密钥数据结构。读完后你可以为一个 Litestar 应用搭建完整的“认证 授权 密钥安全”链路并理解其底层 ASGI 作用域scope注入机制。安全体系总览认证与授权两条主线从仓库文档组织看安全章节索引Litestar 的安全能力分为两大层次认证Authentication确认“你是谁”。核心抽象是 AbstractAuthenticationMiddleware位于litestar/middleware/authentication.py内置后端Session Auth、JWT Auth 等则统一继承自 AbstractSecurityConfig位于 litestar/security/base.py。授权Authorization确认“你能做什么”。核心抽象是Guards——接收连接与路由处理器副本的可调用对象校验失败时抛出PermissionDeniedExceptionHTTP 403见 Guards 文档。此外密钥数据处理SecretString/SecretBytes为在请求参数与请求体中安全地传递敏感值提供了容器类型。自定义认证AbstractAuthenticationMiddleware核心 APIAuthenticationResult 与 authenticate_requestlitestar.middleware导出 AbstractAuthenticationMiddleware这是一个实现了MiddlewareProtocol的抽象基类ABC。你只需继承它并实现抽象方法authenticate_requestfrom litestar.middleware import ( AbstractAuthenticationMiddleware, AuthenticationResult, ) from litestar.connection import ASGIConnection class MyAuthenticationMiddleware(AbstractAuthenticationMiddleware): async def authenticate_request( self, connection: ASGIConnection ) - AuthenticationResult: # 在这里完成认证逻辑 ...authenticate_request是一个异步函数接收连接实例返回 AuthenticationResult 实例。该 dataclass 只有两个属性__slots__ (auth, user)user非可空代表用户类型为Any可接收包括None在内的任意值auth可选代表认证凭证如 JWT token默认None。从源码的__call__实现litestar/middleware/authentication.py#L71-L92可以看到其工作方式中间件先通过should_bypass_middleware判断是否应跳过依据exclude路径正则、exclude_http_methods、路由上的exclude_opt_key选项、scopes作用域若不跳过则调用authenticate_request并把结果写入 ASGI scopeif not should_bypass_middleware(...): auth_result await self.authenticate_request(ASGIConnection(scope)) scope[user] auth_result.user scope[auth] auth_result.auth await self.app(scope, receive, send)这两个 scope 值随后分别作为Request.user/Request.auth在 HTTP 路由处理器中可用作为WebSocket.user/WebSocket.auth在 WebSocket 路由处理器中可用——即同一个认证结果对 HTTP 与 WS 两种作用域统一生效。构造参数方面__init__L48-L69支持参数说明默认值app中间件栈中下一个 ASGI handler由框架注入必填exclude一个或一组跳过认证的路径正则模式Noneexclude_from_auth_key路由上用于关闭认证的选项键exclude_from_authexclude_http_methods不需要认证的方法序列(HttpMethod.OPTIONS,)scopes处理的 ASGI scope 集合{ScopeType.HTTP, ScopeType.WEBSOCKET}完整实战示例基于 API Key 的认证中间件仓库示例 using_abstract_authentication_middleware.py 演示了从用户模型到应用装配的完整链路。第一步定义用户与 token 模型可用 msgspec、Pydantic、ODM/ORM 实现这里用 dataclassdataclass class MyUser: name: str dataclass class MyToken: api_key: str第二步实现认证中间件——从请求头读取 API Key查询“数据库”换取用户API_KEY_HEADER X-API-KEY TOKEN_USER_DATABASE {1: user_authorized} class CustomAuthenticationMiddleware(AbstractAuthenticationMiddleware): async def authenticate_request(self, connection: ASGIConnection) - AuthenticationResult: 从请求头解析 API Key并检索对应 token 的用户 # 读取认证头 auth_header connection.headers.get(API_KEY_HEADER) if not auth_header: raise NotAuthorizedException() # 此处实际为数据库调用 token MyToken(api_keyauth_header) if not (name : TOKEN_USER_DATABASE.get(token.api_key)): raise NotAuthorizedException() user MyUser(namename) return AuthenticationResult(useruser, authtoken)第三步把中间件传入Litestar构造函数用DefineMiddleware包装以便附加参数# 可选地排除某些路径这里排除所有挂载在 /schema* 下或之上的路由 auth_mw DefineMiddleware(CustomAuthenticationMiddleware, excludeschema) app Litestar( route_handlers[site_index, my_http_handler, my_ws_handler], middleware[auth_mw], dependencies{some_dependency: Provide(my_dependency)}, )至此CustomAuthenticationMiddleware会对每个请求运行。在处理器中访问认证结果时Request/WebSocket支持泛型标注以获得静态类型提示get(/, sync_to_threadFalse) def my_http_handler(request: Request[MyUser, MyToken, State]) - None: user request.user # 静态类型正确推断为 MyUser auth request.auth # 静态类型正确推断为 MyTokenWebSocket 侧同理websocket(/) async def my_ws_handler(socket: WebSocket[MyUser, MyToken, State]) - None: user socket.user # 类型为 MyUser auth socket.auth # 类型为 MyToken此外还有两个控制粒度的机制按路由排除在处理器上声明exclude_from_authTrue与中间件的exclude_from_auth_key对应get(path/, exclude_from_authTrue) async def site_index() - Response: ...依赖中复用依赖注入函数同样能拿到带类型的request.user/request.authasync def my_dependency(request: Request[MyUser, MyToken, State]) - Any: user request.user # 类型为 MyUser auth request.auth # 类型为 MyToken失败时的异常约定authenticate_request的文档约定L94-L110认证失败时应抛出NotAuthorizedException或PermissionDeniedException。注意NotAuthorizedException对应 HTTP 401而PermissionDeniedException对应 403——前者语义上表示“未认证/认证失败”与 Guards 的授权失败场景相互区分。内置安全后端AbstractSecurityConfig 及其子类基类 AbstractSecurityConfigAbstractSecurityConfig 是 Litestar 提供的全部安全后端的基类也是自定义后端的起点。它是一个泛型类Generic[UserType, AuthType]关键字段如下源码字段说明retrieve_user_handler接收认证值auth与连接、返回user值的可调用对象可同步可异步同步会被自动包装exclude跳过认证的路径正则字符串或列表exclude_opt_key路由上关闭认证/授权检查的选项键默认exclude_from_authexclude_http_methods不需要认证的方法默认[OPTIONS, HEAD]scopes处理的 ASGI scopeNone时同时处理http与websocketguards授权用的 Guards 迭代器route_handlers/dependencies后端可自带的处理器与依赖type_encoders类型到序列化编码器的映射每个后端还提供on_app_init(app_config)方法L75 附近在应用初始化时把中间件、Guards 等注入应用——这就是下例中on_app_init[session_auth.on_app_init]的作用。Session Auth 后端Litestar 内置开箱即用的 Session Auth 后端可与会话中间件支持的所有会话后端内存、服务端、Redis 等配合使用。完整示例见 using_session_auth.py核心结构# User 模型可以是任意类型Pydantic、SQLAlchemy 模型等 class User(BaseModel): id: UUID name: str email: EmailStr # 会话字典 → 用户的回调可同步可异步 async def retrieve_user_handler( session: dict[str, Any], connection: ASGIConnection[Any, Any, Any, Any] ) - User | None: return MOCK_DB.get(user_id) if (user_id : session.get(user_id)) else None # 登录成功后写入会话 post(/login) async def login(data: UserLoginPayload, request: Request[Any, Any, Any]) - User: ... request.set_session({user_id: user_id}) # 支持 dict 或 Pydantic 模型 return MOCK_DB[user_id] # 受保护路由request.user 由 retrieve_user_handler 的结果注入 get(/user, sync_to_threadFalse) def get_user(request: Request[User, dict[Literal[user_id], str], Any]) - Any: return request.user # 安全后端配置 session_auth SessionAuthUser, ServerSideSessionBackend, # 排除无需认证的 URL文档、注册与登录 exclude[/login, /signup, /schema], ) app Litestar( route_handlers[login, signup, get_user], on_app_init[session_auth.on_app_init], # 注入中间件与 OpenAPI 安全方案 openapi_configopenapi_config, )会话字典的内容由你自己决定——上面用{user_id: ...}即可登录处理中还会用到SecretStr承接密码字段避免明文在对象上流转。JWT 后端JWT 安全后端属于可选组件需要先安装pyjwt与cryptography依赖或者直接用 extra 安装见 JWT 文档pip install litestar[jwt]Litestar 提供三个 JWT 后端JWTAuth基础后端token 通过请求头传输服务端也从同一头键读取 token默认头为AuthorizationJWTCookieAuth继承JWTAuth区别是 token 放在 Cookie 中而非请求头OAuth2PasswordBearerAuth继承JWTCookieAuth用于 OAuth 2.0 Bearer 密码流程。以JWTAuth为例完整示例见 using_jwt_auth.py# JWTAuth 需要一个 retrieve handler接收 JWT token 模型与连接返回对应 User async def retrieve_user_handler(token: Token, connection: ASGIConnection[Any, Any, Any, Any]) - User | None: # 在这里实现用户检索逻辑 return MOCK_DB.get(token.sub) jwt_auth JWTAuthUser), # 指定哪些端点排除在认证之外登录端点与 OpenAPI 文档 exclude[/login, /schema], ) # 由 JWTAuth 实例生成登录处理器 post(/login) async def login_handler(data: User) - Response[User]: MOCK_DB[str(data.id)] data # 可对 Response 实例做任意更新如 response.set_cookie(...) return jwt_auth.login(identifierstr(data.id), token_extras{email: data.email}, response_bodydata) get(/some-path, sync_to_threadFalse) def some_route_handler(request: Request[User, Token, Any]) - Any: assert isinstance(request.user, User) # 中间件注入的用户 assert isinstance(request.auth, Token) # 从认证头解析出的 Token从 JWTAuth 源码字段定义 可以看到其关键参数与默认值参数默认值说明token_secret必填生成 token 哈希的密钥建议从环境变量注入algorithmHS256JWT 签名算法auth_headerAuthorization读取 token 的请求头键可改为X-Api-Key等default_token_expiration1 天timedelta(days1)token 默认过期时间revoked_token_handlerNone判断 token 是否被撤销的回调返回True表示已撤销openapi_security_scheme_nameBearerToken注入 OpenAPI 的安全方案名自定义 Token 类Token 类可以携带任意自定义字段继承 Token 子类化后在 backend 上指定即可示例见 custom_token_cls.py。token 会在 JSON 与目标类型之间做基础类型转换。需要注意的是涉及第三方库Pydantic、attrs或自定义type_decoders的复杂转换在 token 上不可用要支持复杂转换必须在子类中覆写Token.encode/Token.decode。校验 issuer 与 audience要校验 JWT 的ississuer与audaudience声明可在认证后端上设置accepted_issuers/accepted_audiences列表示例见 verify_issuer_audience.pyjwt_auth JWTAuthUser, retrieve_user_handlerretrieve_user_handler, accepted_audiences[https://api.testserver.local], accepted_issuers[https://auth.testserver.local], )解码 JWT 时token 上的 issuer/audience 会与白名单逐一比对任何一项不匹配都会抛出NotAuthorizedException返回401 Unauthorized响应。自定义 token 校验与撤销自定义解码覆写Token.decode_payload示例见 custom_decode_payload.py。它由Token.decode以编码后的 token 字符串调用必须返回表示 payload 的字典decode再据此构造 token 类实例。token 撤销维护一份已撤销 token 列表在认证时检查。JWTAuth提供revoked_token_handler回调L285-L287接收 auth 值返回True表示该 token 已撤销后续携带该 token 的请求将被拒绝示例见 using_token_revocation.py。端点的排除与包含规则本节对应 excluding-and-including-endpoints.rst。默认规则配置在所用的Auth对象AbstractSecurityConfig子类上下文以SessionAuth为例JWTAuth与JWTCookieAuth的用法完全一致。1. 排除路由excludeexclude接受字符串或字符串列表被解释为正则模式并针对完整路径匹配。由于模式不做隐式锚定/schema会匹配任意包含/schema的路径而不只是以它开头的路径。要只匹配前缀请用^锚定session_auth SessionAuthUser, ServerSideSessionBackend, # 对除 /login、/signup、/schema 开头之外的所有端点启用认证 exclude[r^/login, r^/signup, r^/schema], )用^/schema锚定后/schema/swagger也被覆盖无需单独排除。文档特别给出一个警告传入/会禁用全部路由的认证因为它作为正则匹配任何路径。排除模式在底层由 build_exclude_path_pattern 编译并在should_bypass_middleware中统一用于中间件跳过判断authentication.py#L82-L88。2. 反向包含只保护特定路径由于exclude是正则可以写一条“取反”规则使只有指定路径受认证保护。下例中仅/secured下的端点需要认证其余路由全部放行session_auth SessionAuthUser, ServerSideSessionBackend, # 排除“非 /secured 结尾”的所有路径 —— 等效于只保护 /secured exclude[r^(?!.*\/secured$).*$], )3. 路由级排除exclude_from_auth如果想让某条路由整体跳过认证可在处理器上直接声明exclude_from_authTrueget(/secured) def secured_route() - Any: ... get(/unsecured, exclude_from_authTrue) def unsecured_route() - Any: ...选项键本身也可以改名——在安全配置中设置exclude_opt_key默认值是exclude_from_authget(/unsecured, no_authTrue) def unsecured_route() - Any: ... session_auth SessionAuthUser, ServerSideSessionBackend, exclude[r^/login, r^/signup, r^/schema], exclude_opt_keyno_auth, # 默认值是 exclude_from_auth )Guards基于角色的授权Guards 是接收两个参数的可调用对象connectionRequest或WebSocket实例二者都是ASGIConnection子类与route_handlerBaseRouteHandler的副本。它们的职责是授权——验证该连接是否允许到达目标处理器校验失败时应抛出HTTPException通常是带403状态码的PermissionDeniedException。以仓库示例 guards.py 为例实现一个基于角色的授权系统。先定义角色枚举与用户模型class UserRole(StrEnum): CONSUMER consumer ADMIN admin class User(BaseModel): id: UUID4 role: UserRole property def is_admin(self) - bool: 判断用户是否为管理员 return self.role UserRole.ADMIN然后写一个只放行管理员的 guard并挂到具体路由上def admin_user_guard(connection: ASGIConnection, _: BaseRouteHandler) - None: if not connection.user.is_admin: raise PermissionDeniedException() post(path/user, guards[admin_user_guard], sync_to_threadFalse) def create_user(data: User) - User: raise NotImplementedError这里的connection.user正是前面认证环节中间件或 JWT 后端的retrieve_user_handler注入到连接上的对象——Guards 天然依赖认证层的结果工作。只有 admin 用户才能调用create_user。Guard 的作用域与分层Guards 是 Litestar 分层架构的一部分可以声明在任意层Litestar 应用实例、Router、Controller 或单个路由处理器def my_guard(connection: ASGIConnection, handler: BaseRouteHandler) - None: ... # controller 层 class UserController(Controller): path /user guards [my_guard] # router 层 admin_router Router(pathadmin, route_handlers[UserController], guards[my_guard]) # app 层 app Litestar(route_handlers[admin_router], guards[my_guard])放置位置取决于需要的访问控制粒度针对单个处理器某个 controller 的全部操作某个 router 管辖的所有路由还是整个应用两个关键特性可叠加guards是一个 list每层可以挂多个 guard。与依赖注入不同Guards 在不同层之间不相互覆盖而是累积——各层声明的 guard 会合并执行注意 OPTIONS 请求若 guard 放在 controller 或 app 层它们同样会在所有OPTIONS请求上执行该问题见上游 issue #2314文档建议参考其 workaround。借助 opt 键实现“令牌守卫”有时需要把值直接挂在路由处理器上权限标记、私有 token 等这可以通过路由处理器的opt参数键实现。示例一个受“秘密 token”保护的端点def secret_token_guard(connection: ASGIConnection, route_handler: BaseRouteHandler) - None: if ( route_handler.opt.get(secret) and not connection.headers.get(Secret-Header, ) route_handler.opt[secret] ): raise PermissionDeniedException() get(path/secret, guards[secret_token_guard], opt{secret: environ.get(SECRET)}) def secret_endpoint() - None: ...guard 在运行时从route_handler.opt[secret]读取配置值与请求头Secret-Header比对不一致即拒绝——演示了“声明期配置 运行期校验”的组合模式。密钥数据处理SecretString 与 SecretBytesLitestar 提供两个容器类型协助在 Web 服务中处理敏感数据源码见 litestar/datastructures/secret_values.pySecretStringSecretValue[str]子类SecretBytesSecretValue[bytes]子类两者的共同行为由基类SecretValue定义私有存储_secret_value通过get_secret()取出真实值__str__与__repr__均返回遮蔽表示——SecretString返回******SecretBytes返回b******。这意味着即使对象被打印进日志或调试输出真实值也不会泄露。密钥作为请求参数示例 secret_header.py 演示把SecretString作为 GET 请求的 Header 参数接收并用secrets.compare_digest做常量时间比较以抵御时序攻击SECRET_VALUE super-secret # 示例值生产环境应安全存储 get(sync_to_threadFalse) def get_handler(secret: Annotated[SecretString, HeaderParameter(namex-secret)]) - Sensitive: if not compare_digest(secret.get_secret(), SECRET_VALUE): raise NotAuthorizedException return Sensitive(valuesensitive data)文档同时提醒ASGIConnection.headers保存的头部是按 ASGI 消息解析出的原样值务必注意不要让这些头部被日志记录或以其他方式暴露。密钥作为请求体字段示例 secret_body.py 演示把SecretString作为数据结构的字段从 HTTP body 接收敏感值dataclass class Sensitive: value: SecretString post(sync_to_threadFalse) def post_handler(data: Sensitive) - Sensitive: return data安全实践要点用环境变量、密钥管理服务或加密数据库安全存储密钥比较密钥值时始终使用secrets.compare_digest之类的常量时间函数缓解时序攻击实施访问控制与日志审计限制谁能访问敏感信息。小结Litestar 的安全体系可以概括为四层协作认证由AbstractAuthenticationMiddleware自研方案或SessionAuth/JWTAuth/JWTCookieAuth/OAuth2PasswordBearerAuth内置后端统一派生自AbstractSecurityConfig完成认证结果通过scope[user]/scope[auth]注入连接对象授权由 Guards 在应用、Router、Controller、处理器各层累积执行失败抛PermissionDeniedException403路由粒度通过exclude正则、exclude_from_auth/自定义exclude_opt_key选项精确控制密钥传递则交由SecretString/SecretBytes容器与常量时间比较保护。所有机制对 HTTP 与 WebSocket 作用域一致生效并可自动注入 OpenAPI 安全方案文档。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Litestar 安全实现指南：认证中间件、内置安全后端、Guards 与密钥数据处理
Litestar 安全实现指南认证中间件、内置安全后端、Guards 与密钥数据处理【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestarLitestar 对安全方案本身保持“无主见”agnostic立场你可以自由采用任何标准或非标准的认证机制但框架同时提供了一组内置组件让你可以低摩擦地落地认证Authentication与授权Authorization。本文基于仓库中docs/usage/security/下的安全文档章节及其配套示例与源码实现完整讲解 Litestar 安全体系的四个支柱自定义认证中间件AbstractAuthenticationMiddleware、内置安全后端Session / JWT、基于 Guards 的角色授权、路由级认证排除/包含规则以及SecretString/SecretBytes密钥数据结构。读完后你可以为一个 Litestar 应用搭建完整的“认证 授权 密钥安全”链路并理解其底层 ASGI 作用域scope注入机制。安全体系总览认证与授权两条主线从仓库文档组织看安全章节索引Litestar 的安全能力分为两大层次认证Authentication确认“你是谁”。核心抽象是 AbstractAuthenticationMiddleware位于litestar/middleware/authentication.py内置后端Session Auth、JWT Auth 等则统一继承自 AbstractSecurityConfig位于 litestar/security/base.py。授权Authorization确认“你能做什么”。核心抽象是Guards——接收连接与路由处理器副本的可调用对象校验失败时抛出PermissionDeniedExceptionHTTP 403见 Guards 文档。此外密钥数据处理SecretString/SecretBytes为在请求参数与请求体中安全地传递敏感值提供了容器类型。自定义认证AbstractAuthenticationMiddleware核心 APIAuthenticationResult 与 authenticate_requestlitestar.middleware导出 AbstractAuthenticationMiddleware这是一个实现了MiddlewareProtocol的抽象基类ABC。你只需继承它并实现抽象方法authenticate_requestfrom litestar.middleware import ( AbstractAuthenticationMiddleware, AuthenticationResult, ) from litestar.connection import ASGIConnection class MyAuthenticationMiddleware(AbstractAuthenticationMiddleware): async def authenticate_request( self, connection: ASGIConnection ) - AuthenticationResult: # 在这里完成认证逻辑 ...authenticate_request是一个异步函数接收连接实例返回 AuthenticationResult 实例。该 dataclass 只有两个属性__slots__ (auth, user)user非可空代表用户类型为Any可接收包括None在内的任意值auth可选代表认证凭证如 JWT token默认None。从源码的__call__实现litestar/middleware/authentication.py#L71-L92可以看到其工作方式中间件先通过should_bypass_middleware判断是否应跳过依据exclude路径正则、exclude_http_methods、路由上的exclude_opt_key选项、scopes作用域若不跳过则调用authenticate_request并把结果写入 ASGI scopeif not should_bypass_middleware(...): auth_result await self.authenticate_request(ASGIConnection(scope)) scope[user] auth_result.user scope[auth] auth_result.auth await self.app(scope, receive, send)这两个 scope 值随后分别作为Request.user/Request.auth在 HTTP 路由处理器中可用作为WebSocket.user/WebSocket.auth在 WebSocket 路由处理器中可用——即同一个认证结果对 HTTP 与 WS 两种作用域统一生效。构造参数方面__init__L48-L69支持参数说明默认值app中间件栈中下一个 ASGI handler由框架注入必填exclude一个或一组跳过认证的路径正则模式Noneexclude_from_auth_key路由上用于关闭认证的选项键exclude_from_authexclude_http_methods不需要认证的方法序列(HttpMethod.OPTIONS,)scopes处理的 ASGI scope 集合{ScopeType.HTTP, ScopeType.WEBSOCKET}完整实战示例基于 API Key 的认证中间件仓库示例 using_abstract_authentication_middleware.py 演示了从用户模型到应用装配的完整链路。第一步定义用户与 token 模型可用 msgspec、Pydantic、ODM/ORM 实现这里用 dataclassdataclass class MyUser: name: str dataclass class MyToken: api_key: str第二步实现认证中间件——从请求头读取 API Key查询“数据库”换取用户API_KEY_HEADER X-API-KEY TOKEN_USER_DATABASE {1: user_authorized} class CustomAuthenticationMiddleware(AbstractAuthenticationMiddleware): async def authenticate_request(self, connection: ASGIConnection) - AuthenticationResult: 从请求头解析 API Key并检索对应 token 的用户 # 读取认证头 auth_header connection.headers.get(API_KEY_HEADER) if not auth_header: raise NotAuthorizedException() # 此处实际为数据库调用 token MyToken(api_keyauth_header) if not (name : TOKEN_USER_DATABASE.get(token.api_key)): raise NotAuthorizedException() user MyUser(namename) return AuthenticationResult(useruser, authtoken)第三步把中间件传入Litestar构造函数用DefineMiddleware包装以便附加参数# 可选地排除某些路径这里排除所有挂载在 /schema* 下或之上的路由 auth_mw DefineMiddleware(CustomAuthenticationMiddleware, excludeschema) app Litestar( route_handlers[site_index, my_http_handler, my_ws_handler], middleware[auth_mw], dependencies{some_dependency: Provide(my_dependency)}, )至此CustomAuthenticationMiddleware会对每个请求运行。在处理器中访问认证结果时Request/WebSocket支持泛型标注以获得静态类型提示get(/, sync_to_threadFalse) def my_http_handler(request: Request[MyUser, MyToken, State]) - None: user request.user # 静态类型正确推断为 MyUser auth request.auth # 静态类型正确推断为 MyTokenWebSocket 侧同理websocket(/) async def my_ws_handler(socket: WebSocket[MyUser, MyToken, State]) - None: user socket.user # 类型为 MyUser auth socket.auth # 类型为 MyToken此外还有两个控制粒度的机制按路由排除在处理器上声明exclude_from_authTrue与中间件的exclude_from_auth_key对应get(path/, exclude_from_authTrue) async def site_index() - Response: ...依赖中复用依赖注入函数同样能拿到带类型的request.user/request.authasync def my_dependency(request: Request[MyUser, MyToken, State]) - Any: user request.user # 类型为 MyUser auth request.auth # 类型为 MyToken失败时的异常约定authenticate_request的文档约定L94-L110认证失败时应抛出NotAuthorizedException或PermissionDeniedException。注意NotAuthorizedException对应 HTTP 401而PermissionDeniedException对应 403——前者语义上表示“未认证/认证失败”与 Guards 的授权失败场景相互区分。内置安全后端AbstractSecurityConfig 及其子类基类 AbstractSecurityConfigAbstractSecurityConfig 是 Litestar 提供的全部安全后端的基类也是自定义后端的起点。它是一个泛型类Generic[UserType, AuthType]关键字段如下源码字段说明retrieve_user_handler接收认证值auth与连接、返回user值的可调用对象可同步可异步同步会被自动包装exclude跳过认证的路径正则字符串或列表exclude_opt_key路由上关闭认证/授权检查的选项键默认exclude_from_authexclude_http_methods不需要认证的方法默认[OPTIONS, HEAD]scopes处理的 ASGI scopeNone时同时处理http与websocketguards授权用的 Guards 迭代器route_handlers/dependencies后端可自带的处理器与依赖type_encoders类型到序列化编码器的映射每个后端还提供on_app_init(app_config)方法L75 附近在应用初始化时把中间件、Guards 等注入应用——这就是下例中on_app_init[session_auth.on_app_init]的作用。Session Auth 后端Litestar 内置开箱即用的 Session Auth 后端可与会话中间件支持的所有会话后端内存、服务端、Redis 等配合使用。完整示例见 using_session_auth.py核心结构# User 模型可以是任意类型Pydantic、SQLAlchemy 模型等 class User(BaseModel): id: UUID name: str email: EmailStr # 会话字典 → 用户的回调可同步可异步 async def retrieve_user_handler( session: dict[str, Any], connection: ASGIConnection[Any, Any, Any, Any] ) - User | None: return MOCK_DB.get(user_id) if (user_id : session.get(user_id)) else None # 登录成功后写入会话 post(/login) async def login(data: UserLoginPayload, request: Request[Any, Any, Any]) - User: ... request.set_session({user_id: user_id}) # 支持 dict 或 Pydantic 模型 return MOCK_DB[user_id] # 受保护路由request.user 由 retrieve_user_handler 的结果注入 get(/user, sync_to_threadFalse) def get_user(request: Request[User, dict[Literal[user_id], str], Any]) - Any: return request.user # 安全后端配置 session_auth SessionAuthUser, ServerSideSessionBackend, # 排除无需认证的 URL文档、注册与登录 exclude[/login, /signup, /schema], ) app Litestar( route_handlers[login, signup, get_user], on_app_init[session_auth.on_app_init], # 注入中间件与 OpenAPI 安全方案 openapi_configopenapi_config, )会话字典的内容由你自己决定——上面用{user_id: ...}即可登录处理中还会用到SecretStr承接密码字段避免明文在对象上流转。JWT 后端JWT 安全后端属于可选组件需要先安装pyjwt与cryptography依赖或者直接用 extra 安装见 JWT 文档pip install litestar[jwt]Litestar 提供三个 JWT 后端JWTAuth基础后端token 通过请求头传输服务端也从同一头键读取 token默认头为AuthorizationJWTCookieAuth继承JWTAuth区别是 token 放在 Cookie 中而非请求头OAuth2PasswordBearerAuth继承JWTCookieAuth用于 OAuth 2.0 Bearer 密码流程。以JWTAuth为例完整示例见 using_jwt_auth.py# JWTAuth 需要一个 retrieve handler接收 JWT token 模型与连接返回对应 User async def retrieve_user_handler(token: Token, connection: ASGIConnection[Any, Any, Any, Any]) - User | None: # 在这里实现用户检索逻辑 return MOCK_DB.get(token.sub) jwt_auth JWTAuthUser), # 指定哪些端点排除在认证之外登录端点与 OpenAPI 文档 exclude[/login, /schema], ) # 由 JWTAuth 实例生成登录处理器 post(/login) async def login_handler(data: User) - Response[User]: MOCK_DB[str(data.id)] data # 可对 Response 实例做任意更新如 response.set_cookie(...) return jwt_auth.login(identifierstr(data.id), token_extras{email: data.email}, response_bodydata) get(/some-path, sync_to_threadFalse) def some_route_handler(request: Request[User, Token, Any]) - Any: assert isinstance(request.user, User) # 中间件注入的用户 assert isinstance(request.auth, Token) # 从认证头解析出的 Token从 JWTAuth 源码字段定义 可以看到其关键参数与默认值参数默认值说明token_secret必填生成 token 哈希的密钥建议从环境变量注入algorithmHS256JWT 签名算法auth_headerAuthorization读取 token 的请求头键可改为X-Api-Key等default_token_expiration1 天timedelta(days1)token 默认过期时间revoked_token_handlerNone判断 token 是否被撤销的回调返回True表示已撤销openapi_security_scheme_nameBearerToken注入 OpenAPI 的安全方案名自定义 Token 类Token 类可以携带任意自定义字段继承 Token 子类化后在 backend 上指定即可示例见 custom_token_cls.py。token 会在 JSON 与目标类型之间做基础类型转换。需要注意的是涉及第三方库Pydantic、attrs或自定义type_decoders的复杂转换在 token 上不可用要支持复杂转换必须在子类中覆写Token.encode/Token.decode。校验 issuer 与 audience要校验 JWT 的ississuer与audaudience声明可在认证后端上设置accepted_issuers/accepted_audiences列表示例见 verify_issuer_audience.pyjwt_auth JWTAuthUser, retrieve_user_handlerretrieve_user_handler, accepted_audiences[https://api.testserver.local], accepted_issuers[https://auth.testserver.local], )解码 JWT 时token 上的 issuer/audience 会与白名单逐一比对任何一项不匹配都会抛出NotAuthorizedException返回401 Unauthorized响应。自定义 token 校验与撤销自定义解码覆写Token.decode_payload示例见 custom_decode_payload.py。它由Token.decode以编码后的 token 字符串调用必须返回表示 payload 的字典decode再据此构造 token 类实例。token 撤销维护一份已撤销 token 列表在认证时检查。JWTAuth提供revoked_token_handler回调L285-L287接收 auth 值返回True表示该 token 已撤销后续携带该 token 的请求将被拒绝示例见 using_token_revocation.py。端点的排除与包含规则本节对应 excluding-and-including-endpoints.rst。默认规则配置在所用的Auth对象AbstractSecurityConfig子类上下文以SessionAuth为例JWTAuth与JWTCookieAuth的用法完全一致。1. 排除路由excludeexclude接受字符串或字符串列表被解释为正则模式并针对完整路径匹配。由于模式不做隐式锚定/schema会匹配任意包含/schema的路径而不只是以它开头的路径。要只匹配前缀请用^锚定session_auth SessionAuthUser, ServerSideSessionBackend, # 对除 /login、/signup、/schema 开头之外的所有端点启用认证 exclude[r^/login, r^/signup, r^/schema], )用^/schema锚定后/schema/swagger也被覆盖无需单独排除。文档特别给出一个警告传入/会禁用全部路由的认证因为它作为正则匹配任何路径。排除模式在底层由 build_exclude_path_pattern 编译并在should_bypass_middleware中统一用于中间件跳过判断authentication.py#L82-L88。2. 反向包含只保护特定路径由于exclude是正则可以写一条“取反”规则使只有指定路径受认证保护。下例中仅/secured下的端点需要认证其余路由全部放行session_auth SessionAuthUser, ServerSideSessionBackend, # 排除“非 /secured 结尾”的所有路径 —— 等效于只保护 /secured exclude[r^(?!.*\/secured$).*$], )3. 路由级排除exclude_from_auth如果想让某条路由整体跳过认证可在处理器上直接声明exclude_from_authTrueget(/secured) def secured_route() - Any: ... get(/unsecured, exclude_from_authTrue) def unsecured_route() - Any: ...选项键本身也可以改名——在安全配置中设置exclude_opt_key默认值是exclude_from_authget(/unsecured, no_authTrue) def unsecured_route() - Any: ... session_auth SessionAuthUser, ServerSideSessionBackend, exclude[r^/login, r^/signup, r^/schema], exclude_opt_keyno_auth, # 默认值是 exclude_from_auth )Guards基于角色的授权Guards 是接收两个参数的可调用对象connectionRequest或WebSocket实例二者都是ASGIConnection子类与route_handlerBaseRouteHandler的副本。它们的职责是授权——验证该连接是否允许到达目标处理器校验失败时应抛出HTTPException通常是带403状态码的PermissionDeniedException。以仓库示例 guards.py 为例实现一个基于角色的授权系统。先定义角色枚举与用户模型class UserRole(StrEnum): CONSUMER consumer ADMIN admin class User(BaseModel): id: UUID4 role: UserRole property def is_admin(self) - bool: 判断用户是否为管理员 return self.role UserRole.ADMIN然后写一个只放行管理员的 guard并挂到具体路由上def admin_user_guard(connection: ASGIConnection, _: BaseRouteHandler) - None: if not connection.user.is_admin: raise PermissionDeniedException() post(path/user, guards[admin_user_guard], sync_to_threadFalse) def create_user(data: User) - User: raise NotImplementedError这里的connection.user正是前面认证环节中间件或 JWT 后端的retrieve_user_handler注入到连接上的对象——Guards 天然依赖认证层的结果工作。只有 admin 用户才能调用create_user。Guard 的作用域与分层Guards 是 Litestar 分层架构的一部分可以声明在任意层Litestar 应用实例、Router、Controller 或单个路由处理器def my_guard(connection: ASGIConnection, handler: BaseRouteHandler) - None: ... # controller 层 class UserController(Controller): path /user guards [my_guard] # router 层 admin_router Router(pathadmin, route_handlers[UserController], guards[my_guard]) # app 层 app Litestar(route_handlers[admin_router], guards[my_guard])放置位置取决于需要的访问控制粒度针对单个处理器某个 controller 的全部操作某个 router 管辖的所有路由还是整个应用两个关键特性可叠加guards是一个 list每层可以挂多个 guard。与依赖注入不同Guards 在不同层之间不相互覆盖而是累积——各层声明的 guard 会合并执行注意 OPTIONS 请求若 guard 放在 controller 或 app 层它们同样会在所有OPTIONS请求上执行该问题见上游 issue #2314文档建议参考其 workaround。借助 opt 键实现“令牌守卫”有时需要把值直接挂在路由处理器上权限标记、私有 token 等这可以通过路由处理器的opt参数键实现。示例一个受“秘密 token”保护的端点def secret_token_guard(connection: ASGIConnection, route_handler: BaseRouteHandler) - None: if ( route_handler.opt.get(secret) and not connection.headers.get(Secret-Header, ) route_handler.opt[secret] ): raise PermissionDeniedException() get(path/secret, guards[secret_token_guard], opt{secret: environ.get(SECRET)}) def secret_endpoint() - None: ...guard 在运行时从route_handler.opt[secret]读取配置值与请求头Secret-Header比对不一致即拒绝——演示了“声明期配置 运行期校验”的组合模式。密钥数据处理SecretString 与 SecretBytesLitestar 提供两个容器类型协助在 Web 服务中处理敏感数据源码见 litestar/datastructures/secret_values.pySecretStringSecretValue[str]子类SecretBytesSecretValue[bytes]子类两者的共同行为由基类SecretValue定义私有存储_secret_value通过get_secret()取出真实值__str__与__repr__均返回遮蔽表示——SecretString返回******SecretBytes返回b******。这意味着即使对象被打印进日志或调试输出真实值也不会泄露。密钥作为请求参数示例 secret_header.py 演示把SecretString作为 GET 请求的 Header 参数接收并用secrets.compare_digest做常量时间比较以抵御时序攻击SECRET_VALUE super-secret # 示例值生产环境应安全存储 get(sync_to_threadFalse) def get_handler(secret: Annotated[SecretString, HeaderParameter(namex-secret)]) - Sensitive: if not compare_digest(secret.get_secret(), SECRET_VALUE): raise NotAuthorizedException return Sensitive(valuesensitive data)文档同时提醒ASGIConnection.headers保存的头部是按 ASGI 消息解析出的原样值务必注意不要让这些头部被日志记录或以其他方式暴露。密钥作为请求体字段示例 secret_body.py 演示把SecretString作为数据结构的字段从 HTTP body 接收敏感值dataclass class Sensitive: value: SecretString post(sync_to_threadFalse) def post_handler(data: Sensitive) - Sensitive: return data安全实践要点用环境变量、密钥管理服务或加密数据库安全存储密钥比较密钥值时始终使用secrets.compare_digest之类的常量时间函数缓解时序攻击实施访问控制与日志审计限制谁能访问敏感信息。小结Litestar 的安全体系可以概括为四层协作认证由AbstractAuthenticationMiddleware自研方案或SessionAuth/JWTAuth/JWTCookieAuth/OAuth2PasswordBearerAuth内置后端统一派生自AbstractSecurityConfig完成认证结果通过scope[user]/scope[auth]注入连接对象授权由 Guards 在应用、Router、Controller、处理器各层累积执行失败抛PermissionDeniedException403路由粒度通过exclude正则、exclude_from_auth/自定义exclude_opt_key选项精确控制密钥传递则交由SecretString/SecretBytes容器与常量时间比较保护。所有机制对 HTTP 与 WebSocket 作用域一致生效并可自动注入 OpenAPI 安全方案文档。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考