ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Litestar 自定义认证中间件实战:AbstractAuthenticationMiddleware 扩展机制全解析

Litestar 自定义认证中间件实战:AbstractAuthenticationMiddleware 扩展机制全解析 Litestar 自定义认证中间件实战AbstractAuthenticationMiddleware 扩展机制全解析【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar本篇指南基于 Litestar 官方文档docs/usage/security/abstract-authentication-middleware.rst及其配套示例与源码讲解如何通过继承AbstractAuthenticationMiddleware实现自定义认证中间件包括authenticate_request抽象方法的契约、AuthenticationResult结果结构、user/auth值在 ASGI scope 中的流转路径以及exclude、exclude_from_auth、exclude_http_methods、scopes四类路由豁免机制的完整用法。读完后你可以为任意 HTTP / WebSocket 应用构建一套可排除特定路由、支持类型化request.user的认证层并理解其底层跳过逻辑与内置 JWT、SessionAuth 安全后端复用的同一套基类之间的关系。一、AbstractAuthenticationMiddleware认证中间件的抽象基类Litestar 从litestar.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: # 在此处实现你的认证逻辑 ...从源码 litestar/middleware/authentication.py 可以看到该抽象方法的契约它必须被子类覆写接收一个ASGIConnection实例即当前 HTTP 或 WebSocket 连接认证成功时返回一个AuthenticationResult实例认证失败时应当抛出NotAuthorizedException401或PermissionDeniedException403由框架统一的异常处理机制将其转换为 HTTP 响应。二、AuthenticationResult认证结果的载体与 scope 流转authenticate_request的返回值是一个标准 dataclassAuthenticationResult定义在 litestar/middleware/authentication.pydataclass class AuthenticationResult: __slots__ (auth, user) user: Any The user model, this can be any value corresponding to a user of the API. auth: Any The auth value, this can for example be a JWT token.两个属性的语义user非可选值代表用户。类型标注为Any因此可以接收任意值包括None——它可以是 dataclass 模型、ORM 实例也可以是任意自定义对象auth代表认证方案scheme的值例如 JWT token、API key 对象等默认为None。这两个值如何到达你的路由处理器看基类的 ASGI 入口__call__litestar/middleware/authentication.pyasync def __call__(self, scope: Scope, receive: Receive, send: Send) - None: if not should_bypass_middleware( exclude_http_methodsself.exclude_http_methods, exclude_opt_keyself.exclude_opt_key, exclude_path_patternself.exclude, scopescope, scopesself.scopes, ): auth_result await self.authenticate_request(ASGIConnection(scope)) scope[user] auth_result.user scope[auth] auth_result.auth await self.app(scope, receive, send)关键流程先调用should_bypass_middleware判断当前连接是否应跳过认证豁免机制见下文第四节若不需要跳过则调用子类的authenticate_request把结果的user和auth写入 ASGIscope字典的scope[user]与scope[auth]随后无论如何都会把请求继续传给下一个 ASGI 应用self.app。写入 scope 之后这些值通过连接基类的属性暴露出来。litestar/connection/base.py 中property def auth(self) - AuthT: if auth not in self.scope: raise ImproperlyConfiguredException(auth is not defined in scope, install an AuthMiddleware to set it) return cast(AuthT, self.scope[auth]) property def user(self) - UserT: if user not in self.scope: raise ImproperlyConfiguredException(user is not defined in scope, install an AuthMiddleware to set it) return cast(UserT, self.scope[user])因此在HTTP 路由处理器中通过request.user/request.auth访问在WebSocket 路由处理器中通过socket.user/socket.auth访问。注意一个重要的行为细节如果应用没有安装任何认证中间件而处理器却访问了user或auth会抛出ImproperlyConfiguredException测试 tests/unit/test_middleware/test_base_authentication_middleware.py 验证了这一行为对应 HTTP 500 响应。三、完整实战示例从用户模型到路由豁免官方示例完整实现在 docs/examples/security/using_abstract_authentication_middleware.py下面按步骤完整走一遍。3.1 定义 user 与 token 模型用户模型可以用 msgspec、Pydantic、ODM、ORM 等任意方式实现示例采用 dataclassdataclass class MyUser: name: str dataclass class MyToken: api_key: str3.2 实现认证中间件API_KEY_HEADER X-API-KEY TOKEN_USER_DATABASE {1: user_authorized} class CustomAuthenticationMiddleware(AbstractAuthenticationMiddleware): async def authenticate_request(self, connection: ASGIConnection) - AuthenticationResult: Given a request, parse the request api key stored in the header and retrieve the user correlating to the token from the DB # retrieve the auth header auth_header connection.headers.get(API_KEY_HEADER) if not auth_header: raise NotAuthorizedException() # this would be a database call token MyToken(api_keyauth_header) if not (name : TOKEN_USER_DATABASE.get(token.api_key)): raise NotAuthorizedException() user MyUser(namename) return AuthenticationResult(useruser, authtoken)这个实现的认证逻辑是从请求头X-API-KEY取出 API key在模拟的数据库中查找对应用户名找不到则抛出NotAuthorizedException对应 HTTP 401成功则返回携带MyUser和MyToken的AuthenticationResult。3.3 在 HTTP 与 WebSocket 处理器中访问 user / auth认证中间件注册后它对每个请求都会运行处理器中的user/auth会获得正确的静态类型通过Request/WebSocket的泛型参数声明get(/, sync_to_threadFalse) def my_http_handler(request: Request[MyUser, MyToken, State]) - None: user request.user # correctly typed as MyUser auth request.auth # correctly typed as MyToken assert isinstance(user, MyUser) assert isinstance(auth, MyToken) websocket(/) async def my_ws_handler(socket: WebSocket[MyUser, MyToken, State]) - None: user socket.user # correctly typed as MyUser auth socket.auth # correctly typed as MyToken assert isinstance(user, MyUser) assert isinstance(auth, MyToken)3.4 排除单条路由exclude_from_auth对于不需要认证的个别路由如站点首页直接在路由上标记exclude_from_authTrueget(path/, exclude_from_authTrue) async def site_index() - Response: Site index exists await anyio.Path(index.html).exists() if exists: async with await anyio.open_file(anyio.Path(index.html)) as file: content await file.read() return Response(contentcontent, status_code200, media_typeMediaType.HTML) raise NotFoundException(Site index was not found)3.5 在依赖dependency中使用同样的机制也适用于依赖函数——依赖中同样可以拿到类型化的request.user/request.authasync def my_dependency(request: Request[MyUser, MyToken, State]) - Any: user request.user # correctly typed as MyUser auth request.auth # correctly typed as MyToken assert isinstance(user, MyUser) assert isinstance(auth, MyToken)3.6 注册中间件与应用最后把中间件传入Litestar构造器。这里用DefineMiddleware声明式包装并顺带演示按路径前缀排除认证排除所有挂载在/schema*下的路由# you can optionally exclude certain paths from authentication. # the following excludes all routes mounted at or under /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)}, )四、构造函数参数详解四类豁免机制AbstractAuthenticationMiddleware的构造函数签名为litestar/middleware/authentication.pydef __init__( self, app: ASGIApp, exclude: str | list[str] | None None, exclude_from_auth_key: str exclude_from_auth, exclude_http_methods: Sequence[Method] | None None, scopes: Scopes | None None, ) - None:各参数说明参数默认值说明app必填中间件链中的下一个 ASGI 应用使用DefineMiddleware时由框架自动注入excludeNone一个或多个正则表达式模式匹配到的路径跳过认证exclude_from_auth_keyexclude_from_auth路由上用于关闭认证的 opt-out 键名可自定义例如改为my_exclude_key然后在路由上写my_exclude_keyTrueexclude_http_methods(HttpMethod.OPTIONS,)不需要认证的 HTTP 方法序列默认自动排除 OPTIONS 请求scopes{ScopeType.HTTP, ScopeType.WEBSOCKET}该中间件处理的 ASGI scope 类型集合默认为 HTTP WebSocket4.1 exclude 路径模式如何编译exclude的值由 litestar/middleware/_utils.py 中的build_exclude_path_pattern编译为单个正则传入字符串时直接re.compile传入列表时用|连接后编译若正则不合法抛出ImproperlyConfiguredExceptionUnable to compile exclude patterns for middleware...若该模式会匹配所有路径实现上通过尝试匹配/和一个 UUID 路径来检测贪婪匹配会发出warn_middleware_excluded_on_all_routes警告——这是一个实用的安全提示防止你无意间把所有路由都排除在认证之外。4.2 should_bypass_middleware四级跳过判定每次请求进入__call__时should_bypass_middlewarelitestar/middleware/_utils.py按以下顺序判定是否跳过认证scope 类型检查scope[type]不在scopes集合中例如你只配置了 HTTP 而来了一个 WebSocket 升级→ 跳过路由 opt-out 检查从scope[route_handler].opt中读取exclude_from_auth或你自定义的键为真 → 跳过。这就是get(/, exclude_from_authTrue)的底层原理HTTP 方法检查scope[method]在exclude_http_methods中默认含OPTIONS→ 跳过路径模式检查exclude正则匹配scope[path]对 mount 路由则匹配raw_path→ 跳过。任一条件命中即跳过认证直接放行到下一个 ASGI 应用全部未命中才执行authenticate_request。五、测试验证的行为边界单元测试 tests/unit/test_middleware/test_base_authentication_middleware.py 对上述机制做了系统性验证可作为行为参照test_authentication_middleware_http_routes/test_authentication_middleware_websocket_routes认证失败时PermissionDeniedException对应403返回AuthenticationResult后 HTTP 与 WebSocket 处理器都能断言request.user/socket.user的类型test_authentication_middleware_not_installed_raises_for_*未安装认证中间件时访问user/auth抛出ImproperlyConfiguredExceptionHTTP 500 / WebSocket 断开test_authentication_middleware_excludeDefineMiddleware(AuthMiddleware, exclude[north, south])后/north/1与/south返回 200/west返回 403test_authentication_middleware_exclude_from_auth/..._custom_key路由级exclude_from_authTrue以及自定义exclude_from_auth_keymy_exclude_key均能豁免认证test_authentication_exclude_http_methods/..._defaultexclude_http_methods[HttpMethod.GET]时 GET 放行而 OPTIONS 被拦截不配置时 OPTIONS 默认放行——印证了默认排除 OPTIONS的构造函数行为。六、同一基类的复用者JWT 与 SessionAuth 安全后端理解AbstractAuthenticationMiddleware还有一个附带收益Litestar 内置的多个安全后端正是基于它构建的。例如JWTlitestar/security/jwt/auth.py 中BaseJWTAuth声明的authentication_middleware_class必须继承JWTAuthenticationMiddleware其继承链最终到AbstractAuthenticationMiddleware。BaseJWTAuth的token_secret、retrieve_user_handler、algorithm、auth_header、accepted_audiences、require_claims等配置项最终都服务于该中间件的authenticate_request实现Session 认证litestar/security/session_auth/middleware.py 中SessionAuthMiddleware(AbstractAuthenticationMiddleware)的authenticate_request检查connection.session为空时清空会话并抛出NotAuthorizedException否则调用retrieve_user_handler解析用户返回AuthenticationResult(useruser, authconnection.session)。从源码结构看这套中间件写入 scope → 连接属性暴露 → 处理器/依赖消费的契约是所有认证方案共享的因此本文的自定义中间件知识与内置后端完全互通。七、小结与要点回顾实现自定义认证只需继承AbstractAuthenticationMiddleware并覆写authenticate_request(connection) - AuthenticationResult失败时抛NotAuthorizedException/PermissionDeniedExceptionAuthenticationResult.user与.auth会被写入scope[user]/scope[auth]进而通过request.user/request.auth/socket.user/socket.auth以声明的泛型类型暴露给处理器与依赖豁免认证有四条路径exclude正则支持列表非法正则报错、匹配全路径会告警、路由级exclude_from_authTrue键名可用exclude_from_auth_key自定义、exclude_http_methods默认排除 OPTIONS、scopes限定处理的 ASGI scope 类型忘记安装认证中间件却访问user/auth会得到ImproperlyConfiguredException这是定位为什么 request.user 报错的第一检查点。完整可运行示例见 docs/examples/security/using_abstract_authentication_middleware.py对应测试见 tests/unit/test_middleware/test_base_authentication_middleware.py。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表