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_request：

from 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实例；
认证失败时应当抛出NotAuthorizedException（401）或PermissionDeniedException（403），由框架统一的异常处理机制将其转换为 HTTP 响应。

二、AuthenticationResult：认证结果的载体与 scope 流转

authenticate_request的返回值是一个标准 dataclassAuthenticationResult，定义在 litestar/middleware/authentication.py：

@dataclass 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.py）：

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: if not should_bypass_middleware( exclude_http_methods=self.exclude_http_methods, exclude_opt_key=self.exclude_opt_key, exclude_path_pattern=self.exclude, scope=scope, scopes=self.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 等任意方式实现，示例采用 dataclass：

@dataclass class MyUser: name: str @dataclass class MyToken: api_key: str

3.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_key=auth_header) if not (name := TOKEN_USER_DATABASE.get(token.api_key)): raise NotAuthorizedException() user = MyUser(name=name) return AuthenticationResult(user=user, auth=token)

这个实现的认证逻辑是：从请求头X-API-KEY取出 API key，在（模拟的）数据库中查找对应用户名，找不到则抛出NotAuthorizedException（对应 HTTP 401），成功则返回携带MyUser和MyToken的AuthenticationResult。

3.3 在 HTTP 与 WebSocket 处理器中访问 user / auth

认证中间件注册后，它对每个请求都会运行，处理器中的user/auth会获得正确的静态类型（通过Request/WebSocket的泛型参数声明）：

@get("/", sync_to_thread=False) 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_auth=True：

@get(path="/", exclude_from_auth=True) 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(content=content, status_code=200, media_type=MediaType.HTML) raise NotFoundException("Site index was not found")

3.5 在依赖（dependency）中使用

同样的机制也适用于依赖函数——依赖中同样可以拿到类型化的request.user/request.auth：

async 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, exclude="schema") 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.py）：

def __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`时由框架自动注入）
`exclude`	`None`	一个或多个正则表达式模式，匹配到的路径跳过认证
`exclude_from_auth_key`	`"exclude_from_auth"`	路由上用于关闭认证的 opt-out 键名，可自定义（例如改为`my_exclude_key`，然后在路由上写`my_exclude_key=True`）
`exclude_http_methods`	`(HttpMethod.OPTIONS,)`	不需要认证的 HTTP 方法序列；默认自动排除 OPTIONS 请求
`scopes`	`{ScopeType.HTTP, ScopeType.WEBSOCKET}`	该中间件处理的 ASGI scope 类型集合，默认为 HTTP + WebSocket

4.1 exclude 路径模式如何编译

exclude的值由 litestar/middleware/_utils.py 中的build_exclude_path_pattern编译为单个正则：

传入字符串时直接re.compile；传入列表时用|连接后编译；
若正则不合法，抛出ImproperlyConfiguredException（"Unable to compile exclude patterns for middleware..."）；
若该模式会匹配所有路径（实现上通过尝试匹配/和一个 UUID 路径来检测"贪婪"匹配），会发出warn_middleware_excluded_on_all_routes警告——这是一个实用的安全提示，防止你无意间把所有路由都排除在认证之外。

4.2 should_bypass_middleware：四级跳过判定

每次请求进入__call__时，should_bypass_middleware（litestar/middleware/_utils.py）按以下顺序判定是否跳过认证：

scope 类型检查：scope["type"]不在scopes集合中（例如你只配置了 HTTP 而来了一个 WebSocket 升级）→ 跳过；
路由 opt-out 检查：从scope["route_handler"].opt中读取exclude_from_auth（或你自定义的键），为真 → 跳过。这就是@get("/", exclude_from_auth=True)的底层原理；
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抛出ImproperlyConfiguredException（HTTP 500 / WebSocket 断开）；
test_authentication_middleware_exclude：DefineMiddleware(AuthMiddleware, exclude=["north", "south"])后，/north/1与/south返回 200，/west返回 403；
test_authentication_middleware_exclude_from_auth/..._custom_key：路由级exclude_from_auth=True（以及自定义exclude_from_auth_key="my_exclude_key"）均能豁免认证；
test_authentication_exclude_http_methods/..._default：exclude_http_methods=[HttpMethod.GET]时 GET 放行而 OPTIONS 被拦截；不配置时 OPTIONS 默认放行——印证了"默认排除 OPTIONS"的构造函数行为。

六、同一基类的复用者：JWT 与 SessionAuth 安全后端

理解AbstractAuthenticationMiddleware还有一个附带收益：Litestar 内置的多个安全后端正是基于它构建的。例如：

JWT：litestar/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(user=user, auth=connection.session)。

从源码结构看，这套"中间件写入 scope → 连接属性暴露 → 处理器/依赖消费"的契约是所有认证方案共享的，因此本文的自定义中间件知识与内置后端完全互通。

七、小结与要点回顾

实现自定义认证只需继承AbstractAuthenticationMiddleware并覆写authenticate_request(connection) -> AuthenticationResult，失败时抛NotAuthorizedException/PermissionDeniedException；
AuthenticationResult.user与.auth会被写入scope["user"]/scope["auth"]，进而通过request.user/request.auth/socket.user/socket.auth以声明的泛型类型暴露给处理器与依赖；
豁免认证有四条路径：exclude正则（支持列表，非法正则报错、匹配全路径会告警）、路由级exclude_from_auth=True（键名可用exclude_from_auth_key自定义）、exclude_http_methods（默认排除 OPTIONS）、scopes（限定处理的 ASGI scope 类型）；
忘记安装认证中间件却访问user/auth会得到ImproperlyConfiguredException，这是定位"为什么 request.user 报错"的第一检查点。