Litestar JWT 安全后端实战JWTAuth、Cookie Auth、OAuth2 密码流与令牌吊销【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar本文以 Litestar 官方文档中的 JWT 安全后端章节为主体系统讲解JWTAuth、JWTCookieAuth、OAuth2PasswordBearerAuth三种认证后端的配置与用法并深入 Token 令牌模型、认证中间件 与 配置基类 的源码实现覆盖自定义令牌字段、issuer/audience 校验、载荷解码定制与令牌吊销等全部进阶场景。读完后你可以独立完成一个带 JWT 登录、鉴权、OpenAPI 安全方案注入的完整应用。安装使用 jwt 依赖扩展Litestar 的 JWT 安全后端是可选组件依赖pyjwt和cryptography两个第三方包。最简单的安装方式是启用 Litestar 的jwtextrapip install litestar[jwt]从当前仓库的 pyproject.toml 可以确认该 extra 的具体内容jwt [cryptography, pyjwt2.9.0]即jwt扩展要求pyjwt 2.9.0源码中直接import jwt即 PyJWT 库来编解码令牌。如果不用 extra也可以单独安装这两个包。JWTAuth 后端基于请求头的基础认证JWTAuth是最基础的 JWT 认证后端它从请求头中读取 JWT 令牌认证通过后由你提供的回调把令牌映射为业务用户对象。完整示例见 using_jwt_auth.pyimport secrets from os import environ from typing import Any from uuid import UUID from pydantic import BaseModel, EmailStr from litestar import Litestar, Request, Response, get, post from litestar.connection import ASGIConnection from litestar.openapi.config import OpenAPIConfig from litestar.security.jwt import JWTAuth, Token # 假设我们有一个 User 模型这里用 pydantic也可以是 SQLAlchemy 模型等任意值 class User(BaseModel): id: UUID name: str email: EmailStr MOCK_DB: dict[str, User] {} # JWTAuth 要求一个 retrieve handler 回调接收 Token 与 ASGI 连接返回对应的 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 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: # request.user 由中间件设置为 retrieve_user_handler 返回的用户实例 assert isinstance(request.user, User) # request.auth 是由认证头数据构建的 Token 实例 assert isinstance(request.auth, Token) openapi_config OpenAPIConfig(titleMy API, version1.0.0) # 把 jwt_auth.on_app_init 作为初始化钩子传入它会注入 JWT 中间件与 OpenAPI 安全配置 app Litestar( route_handlers[login_handler, some_route_handler], on_app_init[jwt_auth.on_app_init], openapi_configopenapi_config, )核心配置参数源码级说明对照 auth.py 中JWTAuth的 dataclass 定义各参数及其默认值如下参数默认值说明token_secret必填签名密钥。官方建议以环境变量注入不要硬编码retrieve_user_handler必填接收(Token, ASGIConnection)返回用户对象可为任意值同步/异步均可algorithmHS256JWT 签名算法auth_headerAuthorization读取令牌的请求头键也可改为X-Api-Key等default_token_expirationtimedelta(days1)默认令牌过期时间excludeNone跳过认证的 URL 模式单个或列表exclude_opt_keyexclude_from_auth用于在单个路由上禁用认证的检查键exclude_http_methods[OPTIONS, HEAD]无需认证的 HTTP 方法scopesNone中间件处理的 ASGI scopeNone时同时处理 http 与 websocketguardsNone认证之后的授权守卫列表route_handlers/dependenciesNone随配置一起注册的处理器与依赖openapi_security_scheme_nameBearerTokenOpenAPI 安全方案的命名token_clsTokenJWT 载荷转换的目标类型可自定义见下文令牌如何从请求头中被提取从 middleware.py 的JWTAuthenticationMiddleware.authenticate_request可以看到提取逻辑auth_header connection.headers.get(self.auth_header) if not auth_header: raise NotAuthorizedException(No JWT token found in request header) encoded_token auth_header.partition( )[-1]即中间件按空格分割请求头取最后一段作为令牌——这正是Bearer token格式中取token的方式。JWTAuth的login()响应头也是按同一规则生成的format_auth_header会依据 OpenAPI 方案拼出Bearer encoded_token见 auth.py。认证失败与 401在authenticate_tokenmiddleware.py中令牌解码、用户查找、吊销检查三步任一失败都会抛出NotAuthorizedExceptiontoken self.token_cls.decode( encoded_tokenencoded_token, secretself.token_secret, algorithmself.algorithm, audienceself.token_audience, issuerself.token_issuer, require_claimsself.require_claims, verify_expself.verify_expiry, verify_nbfself.verify_not_before, strict_audienceself.strict_audience, ) user await self.retrieve_user_handler(token, connection) token_revoked False if self.revoked_token_handler: token_revoked await self.revoked_token_handler(token, connection) if not user or token_revoked: raise NotAuthorizedException() return AuthenticationResult(useruser, authtoken)成功时返回AuthenticationResult(user..., authtoken)框架将其写入scope[user]和scope[auth]因此处理器内可通过request.user与request.auth访问。JWTCookieAuth 后端HttpOnly Cookie 传递令牌JWTCookieAuth继承自同一套基础配置与JWTAuth的唯一本质区别是令牌放在HttpOnlyCookie 中而不是请求头。示例见 using_jwt_cookie_auth.pyfrom litestar.security.jwt import JWTCookieAuth, Token jwt_cookie_auth JWTCookieAuthUser), exclude[/login, /schema], # 可选的 Cookie 参数例如开启 secure # secureTrue, ) post(/login) async def login_handler(data: User) - Response[User]: MOCK_DB[str(data.id)] data return jwt_cookie_auth.login(identifierstr(data.id), response_bodydata)Cookie 相关的附加参数auth.py参数默认值说明keytokenCookie 名称path/Cookie 有效的路径范围domainNoneCookie 有效域secureNone是否强制 HTTPS 传输samesitelax跨站请求是否携带 Cookie可选lax/strict/none源码中JWTCookieAuth.login()auth.py会构造一个httponlyTrue的Cookie其max_age取自token_expiration或default_token_expiration同时保留响应头中的令牌保证同一登录响应既种 Cookie 又返回认证头。对应的JWTCookieAuthenticationMiddlewaremiddleware.py在提取令牌时优先请求头、回退 Cookieencoded_token ( connection.headers.get(self.auth_header, ).partition( )[-1] or connection.cookies.get(self.auth_cookie_key, ).split( )[-1] ) if not encoded_token: raise NotAuthorizedException(No JWT token found in request header or cookies)OAuth2PasswordBearerAuthOAuth 2.0 密码流OAuth2PasswordBearerAuth面向 OAuth 2.0 Bearer 密码流场景。它在功能上基于 Cookie 认证机制额外要求一个token_url参数指向登录路由并向 OpenAPI 文档注入标准的oauth2密码流安全方案。示例见 using_oauth2_password_bearer.pyfrom litestar.security.jwt import OAuth2Login, OAuth2PasswordBearerAuth, Token oauth2_auth OAuth2PasswordBearerAuthUser), token_url/login, # 获取新令牌的 URL即登录路由 exclude[/login, /schema], ) post(/login) async def login_handler(request: Request[Any, Any, Any], data: User) - Response[OAuth2Login]: MOCK_DB[str(data.id)] data # 不传 response_body 时返回标准 OAuth2 令牌响应 return oauth2_auth.login(identifierstr(data.id)) post(/login_custom) async def login_custom_response_handler(data: User) - Response[User]: MOCK_DB[str(data.id)] data # 也可以自定义响应体 return oauth2_auth.login(identifierstr(data.id), response_bodydata)源码上的差异点auth.pytoken_url: str为必填用于生成 OpenAPIOAuthFlow(token_url..., scopes...)安全方案类型为typeoauth2、flowsOAuthFlows(password...)可选oauth_scopes声明令牌可用的 scopelogin()的send_token_as_response_body默认即为True与JWTAuth不同默认响应体是OAuth2LoginDTOdataclass class OAuth2Login: access_token: str token_type: str refresh_token: str | None None expires_in: int | None None即标准 OAuth2 密码流的{access_token, token_type: bearer, expires_in}响应格式。使用自定义令牌类Token本身是一个 dataclass你可以通过子类化添加任意字段并把子类传给后端的token_cls。最小示例见 custom_token_cls.pyimport dataclasses import secrets from typing import Any from litestar import Litestar, Request, get from litestar.connection import ASGIConnection from litestar.security.jwt import JWTAuth, Token dataclasses.dataclass class CustomToken(Token): token_flag: bool False dataclasses.dataclass class User: id: str async def retrieve_user_handler(token: CustomToken, connection: ASGIConnection) - User: return User(idtoken.sub) TOKEN_SECRET secrets.token_hex() jwt_auth JWTAuthUser get(/, sync_to_threadFalse) def handler(request: Request[User, CustomToken, Any]) - dict[str, Any]: return {id: request.user.id, token_flag: request.auth.token_flag} app Litestar(middleware[jwt_auth.middleware])内置Token的完整字段定义见 token.pyexp过期时间datetime、sub主题必填非空字符串、iat签发时间默认当前 UTC 时间、iss/aud/jti均可选以及extras令牌中出现的其他自定义字段都会自动归入这个 dict。__post_init__会做严格校验sub必须非空、exp必须是未来时间、iat必须是当前或过去时间否则抛出ImproperlyConfiguredException。需要注意的边界官方文档的 important 提示令牌从 JSON 转换到 token 类时只做基础类型转换decode中通过msgspec.convert(payload, cls, strictFalse)完成见 token.py。涉及 Pydantic、attrs 等第三方库的复杂类型转换或自定义type_decoders不可用如需要必须在子类中重写Token.encode与Token.decode方法。验证 issuer 与 audience要校验 JWT 的ississuer与audaudience声明在认证后端上设置accepted_issuers/accepted_audiences列表即可。令牌解码时其 issuer/audience 值会与列表比对任一不匹配即抛出NotAuthorizedException返回401 Unauthorized。示例见 verify_issuer_audience.pyjwt_auth JWTAuthUser, retrieve_user_handlerretrieve_user_handler, accepted_audiences[https://api.testserver.local], accepted_issuers[https://auth.testserver.local], )底层实现链路JWTAuth把accepted_issuers/accepted_audiences透传给中间件的token_issuer/token_audienceauth.py 的middleware属性再由Token.decode组装 PyJWT 的optionsoptions: JWTDecodeOptions { verify_aud: bool(audience), verify_iss: bool(issuer), } if require_claims: options[require] list(require_claims)也就是说只有传了 issuer/audience 才开启对应校验。此外Token.decode还支持require_claims要求令牌必须包含指定声明缺失则 401verify_exp/verify_nbf是否校验exp在未来、nbf在过去后端侧对应verify_expiry/verify_not_before默认均为Truestrict_audience要求aud为单值且精确匹配此模式下accepted_audiences必须是长度为 1 的序列token.py 中有显式检查并抛出ValueError。定制令牌解码重写 decode_payloadToken.decode内部通过类方法decode_payload完成实际的 PyJWT 解码token.py。因此只要子类重写decode_payload就能在拿到原始 payload 字典后做任意预处理再交给父类流程构建 token 实例。示例见 custom_decode_payload.pyimport dataclasses from collections.abc import Sequence from typing import Any from litestar.security.jwt.token import JWTDecodeOptions, Token dataclasses.dataclass class CustomToken(Token): classmethod def decode_payload( cls, encoded_token: str, secret: str | bytes, algorithms: list[str], issuer: str | Sequence[str] | None None, audience: str | Sequence[str] | None None, options: JWTDecodeOptions | None None, ) - Any: payload super().decode_payload( encoded_tokenencoded_token, secretsecret, algorithmsalgorithms, issuerissuer, audienceaudience, optionsoptions, ) # 自定义逻辑把 userexample.com 形式的 sub 截取为域名部分 payload[sub] payload[sub].split(, maxsplit1)[1] return payload调用约定decode_payload由Token.decode以编码后的令牌字符串调用必须返回一个代表解码载荷的 dict随后decode用它构造 token 类实例包括exp/iat的时间戳→datetime 转换、sub必填检查、多余字段归入extras等步骤。令牌吊销Token RevocationJWT 本身无状态无法真正撤销但可以维护一个吊销列表并在认证时校验。框架为此提供了revoked_token_handler参数接收(Token, ASGIConnection)返回True表示该令牌已吊销。完整示例见 using_token_revocation.pyMOCK_DB: dict[str, User] {} BLOCKLIST: dict[str, str] {} async def retrieve_user_handler(token: Token, connection: ASGIConnection[Any, Any, Any, Any]) - User | None: return MOCK_DB.get(token.sub) # 使用吊销功能必须提供 revoked_token_handler async def revoked_token_handler(token: Token, connection: ASGIConnection[Any, Any, Any, Any]) - bool: jti token.jti # 令牌唯一标识JWT ID if jti: revoked BLOCKLIST.get(jti) if revoked: return True return False jwt_auth JWTAuthUser, exclude[/login, /schema], ) post(/login) async def login_handler(data: User) - Response[User]: MOCK_DB[str(data.id)] data # 登录时为每个令牌分配唯一 jti便于精确吊销 return jwt_auth.login( identifierstr(data.id), token_unique_jwt_iduuid4().hex, token_extras{email: data.email}, response_bodydata, ) post(/logout) async def logout_handler(request: Request[User, Token, Any]) - dict[str, str]: jti request.auth.jti if jti: BLOCKLIST[jti] revoked return {message: Token has been revoked.} return {message: No valid token found.}要点登录时用token_unique_jwt_id写入jti——create_token会把该值写入令牌的jti声明auth.py这是按令牌精确吊销的前提登出时把request.auth.jti加入黑名单中间件在用户查找之后执行revoked_token_handler见上文 middleware.py 的源码已吊销的令牌同样返回 401。生产环境黑名单应放在 Redis 等持久化存储中并为其设置与令牌剩余寿命一致的 TTL。挂载方式小结on_app_init 与手动 middleware从 base.py 的AbstractSecurityConfig.on_app_init可以看到把jwt_auth.on_app_init传给Litestar(on_app_init...)时框架自动完成三件事app_config.middleware.insert(0, self.middleware)—— 把 JWT 认证中间件插到中间件栈最内层位置 0若配置了OpenAPIConfig把openapi_components安全方案与security_requirement安全需求注入其中合并guards、dependencies、route_handlers等配置项。如果只需要认证中间件而不需要 OpenAPI 注入也可以像 custom_token_cls.py 那样直接Litestar(middleware[jwt_auth.middleware])jwt_auth.middleware属性返回一个封装了JWTAuthenticationMiddleware的DefineMiddleware所有校验参数issuer、audience、claims、过期等都会透传auth.py。小结Litestar 的 JWT 安全后端把令牌编解码与业务用户映射解耦Token负责声明结构与校验retrieve_user_handler/revoked_token_handler承载业务逻辑JWTAuth/JWTCookieAuth/OAuth2PasswordBearerAuth三种后端分别对应请求头、HttpOnly Cookie、OAuth2 密码流三种传输形态并通过on_app_init一并完成中间件与 OpenAPI 安全方案的注册。相关源码集中在 litestar/security/jwt/auth.py、middleware.py、token.py可运行示例集中在 docs/examples/security/jwt/API 参考见 docs/reference/security/ 下的 JWT 条目。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Litestar JWT 安全后端实战：JWTAuth、Cookie Auth、OAuth2 密码流与令牌吊销
Litestar JWT 安全后端实战JWTAuth、Cookie Auth、OAuth2 密码流与令牌吊销【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar本文以 Litestar 官方文档中的 JWT 安全后端章节为主体系统讲解JWTAuth、JWTCookieAuth、OAuth2PasswordBearerAuth三种认证后端的配置与用法并深入 Token 令牌模型、认证中间件 与 配置基类 的源码实现覆盖自定义令牌字段、issuer/audience 校验、载荷解码定制与令牌吊销等全部进阶场景。读完后你可以独立完成一个带 JWT 登录、鉴权、OpenAPI 安全方案注入的完整应用。安装使用 jwt 依赖扩展Litestar 的 JWT 安全后端是可选组件依赖pyjwt和cryptography两个第三方包。最简单的安装方式是启用 Litestar 的jwtextrapip install litestar[jwt]从当前仓库的 pyproject.toml 可以确认该 extra 的具体内容jwt [cryptography, pyjwt2.9.0]即jwt扩展要求pyjwt 2.9.0源码中直接import jwt即 PyJWT 库来编解码令牌。如果不用 extra也可以单独安装这两个包。JWTAuth 后端基于请求头的基础认证JWTAuth是最基础的 JWT 认证后端它从请求头中读取 JWT 令牌认证通过后由你提供的回调把令牌映射为业务用户对象。完整示例见 using_jwt_auth.pyimport secrets from os import environ from typing import Any from uuid import UUID from pydantic import BaseModel, EmailStr from litestar import Litestar, Request, Response, get, post from litestar.connection import ASGIConnection from litestar.openapi.config import OpenAPIConfig from litestar.security.jwt import JWTAuth, Token # 假设我们有一个 User 模型这里用 pydantic也可以是 SQLAlchemy 模型等任意值 class User(BaseModel): id: UUID name: str email: EmailStr MOCK_DB: dict[str, User] {} # JWTAuth 要求一个 retrieve handler 回调接收 Token 与 ASGI 连接返回对应的 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 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: # request.user 由中间件设置为 retrieve_user_handler 返回的用户实例 assert isinstance(request.user, User) # request.auth 是由认证头数据构建的 Token 实例 assert isinstance(request.auth, Token) openapi_config OpenAPIConfig(titleMy API, version1.0.0) # 把 jwt_auth.on_app_init 作为初始化钩子传入它会注入 JWT 中间件与 OpenAPI 安全配置 app Litestar( route_handlers[login_handler, some_route_handler], on_app_init[jwt_auth.on_app_init], openapi_configopenapi_config, )核心配置参数源码级说明对照 auth.py 中JWTAuth的 dataclass 定义各参数及其默认值如下参数默认值说明token_secret必填签名密钥。官方建议以环境变量注入不要硬编码retrieve_user_handler必填接收(Token, ASGIConnection)返回用户对象可为任意值同步/异步均可algorithmHS256JWT 签名算法auth_headerAuthorization读取令牌的请求头键也可改为X-Api-Key等default_token_expirationtimedelta(days1)默认令牌过期时间excludeNone跳过认证的 URL 模式单个或列表exclude_opt_keyexclude_from_auth用于在单个路由上禁用认证的检查键exclude_http_methods[OPTIONS, HEAD]无需认证的 HTTP 方法scopesNone中间件处理的 ASGI scopeNone时同时处理 http 与 websocketguardsNone认证之后的授权守卫列表route_handlers/dependenciesNone随配置一起注册的处理器与依赖openapi_security_scheme_nameBearerTokenOpenAPI 安全方案的命名token_clsTokenJWT 载荷转换的目标类型可自定义见下文令牌如何从请求头中被提取从 middleware.py 的JWTAuthenticationMiddleware.authenticate_request可以看到提取逻辑auth_header connection.headers.get(self.auth_header) if not auth_header: raise NotAuthorizedException(No JWT token found in request header) encoded_token auth_header.partition( )[-1]即中间件按空格分割请求头取最后一段作为令牌——这正是Bearer token格式中取token的方式。JWTAuth的login()响应头也是按同一规则生成的format_auth_header会依据 OpenAPI 方案拼出Bearer encoded_token见 auth.py。认证失败与 401在authenticate_tokenmiddleware.py中令牌解码、用户查找、吊销检查三步任一失败都会抛出NotAuthorizedExceptiontoken self.token_cls.decode( encoded_tokenencoded_token, secretself.token_secret, algorithmself.algorithm, audienceself.token_audience, issuerself.token_issuer, require_claimsself.require_claims, verify_expself.verify_expiry, verify_nbfself.verify_not_before, strict_audienceself.strict_audience, ) user await self.retrieve_user_handler(token, connection) token_revoked False if self.revoked_token_handler: token_revoked await self.revoked_token_handler(token, connection) if not user or token_revoked: raise NotAuthorizedException() return AuthenticationResult(useruser, authtoken)成功时返回AuthenticationResult(user..., authtoken)框架将其写入scope[user]和scope[auth]因此处理器内可通过request.user与request.auth访问。JWTCookieAuth 后端HttpOnly Cookie 传递令牌JWTCookieAuth继承自同一套基础配置与JWTAuth的唯一本质区别是令牌放在HttpOnlyCookie 中而不是请求头。示例见 using_jwt_cookie_auth.pyfrom litestar.security.jwt import JWTCookieAuth, Token jwt_cookie_auth JWTCookieAuthUser), exclude[/login, /schema], # 可选的 Cookie 参数例如开启 secure # secureTrue, ) post(/login) async def login_handler(data: User) - Response[User]: MOCK_DB[str(data.id)] data return jwt_cookie_auth.login(identifierstr(data.id), response_bodydata)Cookie 相关的附加参数auth.py参数默认值说明keytokenCookie 名称path/Cookie 有效的路径范围domainNoneCookie 有效域secureNone是否强制 HTTPS 传输samesitelax跨站请求是否携带 Cookie可选lax/strict/none源码中JWTCookieAuth.login()auth.py会构造一个httponlyTrue的Cookie其max_age取自token_expiration或default_token_expiration同时保留响应头中的令牌保证同一登录响应既种 Cookie 又返回认证头。对应的JWTCookieAuthenticationMiddlewaremiddleware.py在提取令牌时优先请求头、回退 Cookieencoded_token ( connection.headers.get(self.auth_header, ).partition( )[-1] or connection.cookies.get(self.auth_cookie_key, ).split( )[-1] ) if not encoded_token: raise NotAuthorizedException(No JWT token found in request header or cookies)OAuth2PasswordBearerAuthOAuth 2.0 密码流OAuth2PasswordBearerAuth面向 OAuth 2.0 Bearer 密码流场景。它在功能上基于 Cookie 认证机制额外要求一个token_url参数指向登录路由并向 OpenAPI 文档注入标准的oauth2密码流安全方案。示例见 using_oauth2_password_bearer.pyfrom litestar.security.jwt import OAuth2Login, OAuth2PasswordBearerAuth, Token oauth2_auth OAuth2PasswordBearerAuthUser), token_url/login, # 获取新令牌的 URL即登录路由 exclude[/login, /schema], ) post(/login) async def login_handler(request: Request[Any, Any, Any], data: User) - Response[OAuth2Login]: MOCK_DB[str(data.id)] data # 不传 response_body 时返回标准 OAuth2 令牌响应 return oauth2_auth.login(identifierstr(data.id)) post(/login_custom) async def login_custom_response_handler(data: User) - Response[User]: MOCK_DB[str(data.id)] data # 也可以自定义响应体 return oauth2_auth.login(identifierstr(data.id), response_bodydata)源码上的差异点auth.pytoken_url: str为必填用于生成 OpenAPIOAuthFlow(token_url..., scopes...)安全方案类型为typeoauth2、flowsOAuthFlows(password...)可选oauth_scopes声明令牌可用的 scopelogin()的send_token_as_response_body默认即为True与JWTAuth不同默认响应体是OAuth2LoginDTOdataclass class OAuth2Login: access_token: str token_type: str refresh_token: str | None None expires_in: int | None None即标准 OAuth2 密码流的{access_token, token_type: bearer, expires_in}响应格式。使用自定义令牌类Token本身是一个 dataclass你可以通过子类化添加任意字段并把子类传给后端的token_cls。最小示例见 custom_token_cls.pyimport dataclasses import secrets from typing import Any from litestar import Litestar, Request, get from litestar.connection import ASGIConnection from litestar.security.jwt import JWTAuth, Token dataclasses.dataclass class CustomToken(Token): token_flag: bool False dataclasses.dataclass class User: id: str async def retrieve_user_handler(token: CustomToken, connection: ASGIConnection) - User: return User(idtoken.sub) TOKEN_SECRET secrets.token_hex() jwt_auth JWTAuthUser get(/, sync_to_threadFalse) def handler(request: Request[User, CustomToken, Any]) - dict[str, Any]: return {id: request.user.id, token_flag: request.auth.token_flag} app Litestar(middleware[jwt_auth.middleware])内置Token的完整字段定义见 token.pyexp过期时间datetime、sub主题必填非空字符串、iat签发时间默认当前 UTC 时间、iss/aud/jti均可选以及extras令牌中出现的其他自定义字段都会自动归入这个 dict。__post_init__会做严格校验sub必须非空、exp必须是未来时间、iat必须是当前或过去时间否则抛出ImproperlyConfiguredException。需要注意的边界官方文档的 important 提示令牌从 JSON 转换到 token 类时只做基础类型转换decode中通过msgspec.convert(payload, cls, strictFalse)完成见 token.py。涉及 Pydantic、attrs 等第三方库的复杂类型转换或自定义type_decoders不可用如需要必须在子类中重写Token.encode与Token.decode方法。验证 issuer 与 audience要校验 JWT 的ississuer与audaudience声明在认证后端上设置accepted_issuers/accepted_audiences列表即可。令牌解码时其 issuer/audience 值会与列表比对任一不匹配即抛出NotAuthorizedException返回401 Unauthorized。示例见 verify_issuer_audience.pyjwt_auth JWTAuthUser, retrieve_user_handlerretrieve_user_handler, accepted_audiences[https://api.testserver.local], accepted_issuers[https://auth.testserver.local], )底层实现链路JWTAuth把accepted_issuers/accepted_audiences透传给中间件的token_issuer/token_audienceauth.py 的middleware属性再由Token.decode组装 PyJWT 的optionsoptions: JWTDecodeOptions { verify_aud: bool(audience), verify_iss: bool(issuer), } if require_claims: options[require] list(require_claims)也就是说只有传了 issuer/audience 才开启对应校验。此外Token.decode还支持require_claims要求令牌必须包含指定声明缺失则 401verify_exp/verify_nbf是否校验exp在未来、nbf在过去后端侧对应verify_expiry/verify_not_before默认均为Truestrict_audience要求aud为单值且精确匹配此模式下accepted_audiences必须是长度为 1 的序列token.py 中有显式检查并抛出ValueError。定制令牌解码重写 decode_payloadToken.decode内部通过类方法decode_payload完成实际的 PyJWT 解码token.py。因此只要子类重写decode_payload就能在拿到原始 payload 字典后做任意预处理再交给父类流程构建 token 实例。示例见 custom_decode_payload.pyimport dataclasses from collections.abc import Sequence from typing import Any from litestar.security.jwt.token import JWTDecodeOptions, Token dataclasses.dataclass class CustomToken(Token): classmethod def decode_payload( cls, encoded_token: str, secret: str | bytes, algorithms: list[str], issuer: str | Sequence[str] | None None, audience: str | Sequence[str] | None None, options: JWTDecodeOptions | None None, ) - Any: payload super().decode_payload( encoded_tokenencoded_token, secretsecret, algorithmsalgorithms, issuerissuer, audienceaudience, optionsoptions, ) # 自定义逻辑把 userexample.com 形式的 sub 截取为域名部分 payload[sub] payload[sub].split(, maxsplit1)[1] return payload调用约定decode_payload由Token.decode以编码后的令牌字符串调用必须返回一个代表解码载荷的 dict随后decode用它构造 token 类实例包括exp/iat的时间戳→datetime 转换、sub必填检查、多余字段归入extras等步骤。令牌吊销Token RevocationJWT 本身无状态无法真正撤销但可以维护一个吊销列表并在认证时校验。框架为此提供了revoked_token_handler参数接收(Token, ASGIConnection)返回True表示该令牌已吊销。完整示例见 using_token_revocation.pyMOCK_DB: dict[str, User] {} BLOCKLIST: dict[str, str] {} async def retrieve_user_handler(token: Token, connection: ASGIConnection[Any, Any, Any, Any]) - User | None: return MOCK_DB.get(token.sub) # 使用吊销功能必须提供 revoked_token_handler async def revoked_token_handler(token: Token, connection: ASGIConnection[Any, Any, Any, Any]) - bool: jti token.jti # 令牌唯一标识JWT ID if jti: revoked BLOCKLIST.get(jti) if revoked: return True return False jwt_auth JWTAuthUser, exclude[/login, /schema], ) post(/login) async def login_handler(data: User) - Response[User]: MOCK_DB[str(data.id)] data # 登录时为每个令牌分配唯一 jti便于精确吊销 return jwt_auth.login( identifierstr(data.id), token_unique_jwt_iduuid4().hex, token_extras{email: data.email}, response_bodydata, ) post(/logout) async def logout_handler(request: Request[User, Token, Any]) - dict[str, str]: jti request.auth.jti if jti: BLOCKLIST[jti] revoked return {message: Token has been revoked.} return {message: No valid token found.}要点登录时用token_unique_jwt_id写入jti——create_token会把该值写入令牌的jti声明auth.py这是按令牌精确吊销的前提登出时把request.auth.jti加入黑名单中间件在用户查找之后执行revoked_token_handler见上文 middleware.py 的源码已吊销的令牌同样返回 401。生产环境黑名单应放在 Redis 等持久化存储中并为其设置与令牌剩余寿命一致的 TTL。挂载方式小结on_app_init 与手动 middleware从 base.py 的AbstractSecurityConfig.on_app_init可以看到把jwt_auth.on_app_init传给Litestar(on_app_init...)时框架自动完成三件事app_config.middleware.insert(0, self.middleware)—— 把 JWT 认证中间件插到中间件栈最内层位置 0若配置了OpenAPIConfig把openapi_components安全方案与security_requirement安全需求注入其中合并guards、dependencies、route_handlers等配置项。如果只需要认证中间件而不需要 OpenAPI 注入也可以像 custom_token_cls.py 那样直接Litestar(middleware[jwt_auth.middleware])jwt_auth.middleware属性返回一个封装了JWTAuthenticationMiddleware的DefineMiddleware所有校验参数issuer、audience、claims、过期等都会透传auth.py。小结Litestar 的 JWT 安全后端把令牌编解码与业务用户映射解耦Token负责声明结构与校验retrieve_user_handler/revoked_token_handler承载业务逻辑JWTAuth/JWTCookieAuth/OAuth2PasswordBearerAuth三种后端分别对应请求头、HttpOnly Cookie、OAuth2 密码流三种传输形态并通过on_app_init一并完成中间件与 OpenAPI 安全方案的注册。相关源码集中在 litestar/security/jwt/auth.py、middleware.py、token.py可运行示例集中在 docs/examples/security/jwt/API 参考见 docs/reference/security/ 下的 JWT 条目。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考