Pydantic 边界强类型校验与错误防御

跨进程通信（IPC）最大的风险在于：永远不要相信从网络/TCP Socket 另一端传过来的任何数据。如果传入的数据缺斤少两（缺少字段）、类型不对（把数字传成了字符串）或者格式错乱，没有防护的系统很容易在代码深处直接报KeyError或TypeError崩溃。

bus/envelope.py和bus/commands.py模块通过Pydantic v2构建了一套严密的两层强类型防御墙。

为什么要进行数据校验？

1. 保证系统安全性（防止安全漏洞攻击）

未经校验的输入是绝大多数网络安全漏洞的根源。攻击者专门利用程序对输入的“无条件信任”来注入恶性数据。

• 防止 SQL 注入（SQL Injection）：如果用户输入的用户名是admin' OR '1'='1且未经过校验与过滤，数据库可能会直接泄漏所有数据。

• 防止跨站脚本攻击（XSS）：在前端输入框中输入包含<script>的代码，若未经转义和校验直接存储并展示，会导致其他用户的 Cookie 被窃取。

• 防止内存溢出与拒绝服务攻击（DoS）：如果系统不校验上传文件或请求体的大小（例如传输了一个 10GB 的超级大行），很容易把服务器的内存（OOM）或 CPU 直接撑爆。

2. 提高系统健壮性，防止崩溃（ Crash ）

动态语言（如 Python、JavaScript）在处理数据时非常自由，但这份自由也带来了极高的隐患。

• 脏数据引发的运行时异常：如果代码预期拿到的是一个数字age: 18，但外部系统传过来的是字符串age: "hello"或空值null，后续的数学运算或数据库写入就会直接触发TypeError、KeyError或NullPointerException，导致服务进程崩溃。

• “快速失败”原则（Fail-Fast）：在数据刚进入系统边界（如 API 接口、跨进程通信入口）时就进行校验。如果不合法，立刻拒绝；这比让脏数据一路渗透到业务逻辑深处乃至数据库内部才报错要安全得多。

3. 保证业务逻辑与数据的准确性（Data Integrity）

许多业务规则在数据库层是难以完全约束的，必须依赖数据校验来确保数据符合现实世界中的“业务语义”。

• 符合业务范式：例如，用户的年龄不能为负数（age > 0），电子邮件必须包含@符号，手机号必须是 11 位数字，订单金额不能低于 0 元等。

• 多字段关联约束：例如，结束时间必须晚于开始时间（end_time > start_time），或者选择“已支付”状态时必须同时附带“支付流水号”。

4. 简化业务层代码，提升开发效率

如果在数据入口处不做校验，这些校验逻辑就会散落到每一个业务函数中。

• 不进行统一校验的代码（满屏防守）：

  Python

  def process_user(user_dict): if "name" not in user_dict or not user_dict["name"]: raise ValueError("Name is required") if "age" not in user_dict or not isinstance(user_dict["age"], int): raise ValueError("Invalid age") # ...代码中充斥着大量的 if/else 判空和类型判断

• 进行统一数据校验后（如使用 Pydantic / Schema）：

  Python

  def process_user(user: UserModel): # 进入这里的 user 一定是类型正确、字段齐备的！ # 开发者可以 100% 专注于核心业务逻辑 do_something(user.name, user.age)

5. 提供友好的用户体验与明确的错误反馈

当数据输入有误时，系统需要准确告知调用方（无论是前端用户还是下游系统）到底错在哪里，而不是返回一个笼统的“服务器内部错误（500 Internal Server Error）”。

通过数据校验，系统可以返回结构化的错误提示：

❌"Invalid Request: Field 'email' is missing, and 'age' must be greater than 0."

这极大地降低了前端与后端、或者服务与服务之间的沟通和排错成本。

💡 总结

数据校验的本质就是：将所有外部输入（HTTP 请求、IPC 通信、文件读取、用户输入等）一律视为“不可信数据”，通过在系统边界设立严密的规矩，确保进入内部的数据绝对干净、类型明确且符合业务逻辑。

一、 第一层防御：传输外壳校验（JsonRpcRequest/JsonRpcSuccess/JsonRpcError）

当客户端通过 TCP 发送一段 NDJSON 过来时，Daemon（服务端）接收到的本质上只是一串完全不可信的原始字节流/字符串。

1. 协议外壳的模型定义 (bus/envelope.py)

Python

from typing import Any, Literal from pydantic import BaseModel, Field # 1.1 请求外壳 class JsonRpcRequest(BaseModel): jsonrpc: Literal["2.0"] = "2.0" # 强制限制必须是字符串 "2.0" id: str # 请求 ID，用于匹配响应 method: str # 路由方法名（如 "core.ping"） params: dict[str, Any] = Field(default_factory=dict) # 业务参数字典 # 1.2 成功响应外壳 class JsonRpcSuccess(BaseModel): jsonrpc: Literal["2.0"] = "2.0" id: str result: Any # 具体的业务返回结果 # 1.3 错误响应外壳 class JsonRpcError(BaseModel): jsonrpc: Literal["2.0"] = "2.0" id: str | None = None error: JsonRpcErrorObject # 结构化的错误详情

2. 第一层防御拦截了什么？

这一层只关心符合不符合 JSON-RPC 2.0 规范，不关心具体业务逻辑。

Python

# 当服务端收到一段原始 JSON 字符串 line 时： try: raw = json.loads(line) except json.JSONDecodeError as e: # ❌ 连 JSON 都不是（比如发了一串乱码） # 防御生效：返回 -32700 Parse Error await self._send(writer, make_error(None, PARSE_ERROR, f"Parse error: {e}")) return try: req = JsonRpcRequest.model_validate(raw) except ValidationError as e: # ❌ 是 JSON，但缺少了 id 或 method 字段，或者 jsonrpc 版本不对 # 防御生效：返回 -32600 Invalid Request await self._send(writer, make_error(None, INVALID_REQUEST, "Invalid Request", str(e))) return

二、 第二层防御：业务 Payload 强校验（Discriminator 与 Command 校验）

当外壳校验通过后，我们拿到了req.method和req.params。但params里的字段对不对，需要由业务 Command 模型来做第二层把关。

1. 业务 Command 的模型定义 (bus/commands.py)

Python

from typing import Annotated, Literal from pydantic import BaseModel, Discriminator, Tag class PingCommand(BaseModel): type: Literal["core.ping"] = "core.ping" # 业务类型鉴别标识 client: str # 必填字段：客户端名称 class PongResult(BaseModel): server_version: str uptime_ms: int received_at: str # 利用 Discriminator（判别联合）为未来扩展多种命令做准备 Command = Annotated[ PingCommand, # 今后这里可以继续扩展 | RunAgentCommand | SubscribeEventCommand Discriminator("type") ]

2. 第二层防御拦截了什么？

在具体的方法处理器_ping_handler中：

Python

async def _ping_handler(self, params: dict[str, Any]) -> PongResult: try: # Pydantic 自动检查 params 里是否有 client 字段、类型是否为 str cmd = PingCommand.model_validate(params) except ValidationError as e: # ❌ 缺少必填参数（如没传 client），或者类型传错（如 client 传了 123） # 防御生效：抛出 ValidationError，上层 capture 捕获后返回 -32602 Invalid Params raise logger.debug(f"ping from {cmd.client}") return PongResult(...)

三、 结构化错误响应机制（完整的 Error Code 防线）

Pydantic 拦截到非法数据后，最关键的是如何优雅、结构化地告诉客户端炸在哪里，而不是直接 Crash 退出。

整个模块定义了一套与 JSON-RPC 2.0 规范严格对齐的错误代码体系：

Python

# 标准 JSON-RPC 错误码映射 PARSE_ERROR = -32700 # 1. 语法解析失败（非法 JSON） INVALID_REQUEST = -32600 # 2. 请求结构不合规（外壳校验失败） METHOD_NOT_FOUND = -32601 # 3. 找不到对应的 handler（未注册的 method） INVALID_PARAMS = -32602 # 4. 参数校验失败（Pydantic 业务 Payload 校验失败） INTERNAL_ERROR = -32603 # 5. Handler 内部代码执行崩溃（未知 Runtime 异常）

在_handle_line的核心分发逻辑中，实现了全方位的try...except兜底链条：

接收到 TCP 数据流 │ ▼ [ json.loads ] ────────── Exception ─────────► 返回 -32700 (Parse Error) │ ▼ [ JsonRpcRequest.model_validate ] ── Exception ─► 返回 -32600 (Invalid Request) │ ▼ [ 查找 req.method ] ────── Not Found ────────► 返回 -32601 (Method Not Found) │ ▼ [ handler(req.params) ] │ ├─ Pydantic ValidationError ───────────► 返回 -32602 (Invalid Params) │ ├─ 其他业务逻辑 Exception ───────────────► 返回 -32603 (Internal Error) │ ▼ [ 打包 JsonRpcSuccess 返回 ]

四、 为什么说这种设计的工程价值极大？

1. 客户端与服务端“双向防御”：

   • 不仅Daemon 侧收到请求时用 Pydantic 校验；

   • 当CLI 侧收到数据后，同样会用JsonRpcSuccess.model_validate(raw)和PongResult.model_validate(resp.result)重新校验一遍。两端都把跨进程界面的数据当成不可信输入，彻底杜绝隐藏 Bug。

2. 零防御性代码脏入侵： 业务 Handler（如_ping_handler）完全不需要写类似if "client" not in params:、if not isinstance(params["client"], str):这种繁琐易错的校验逻辑。拿到cmd = PingCommand.model_validate(params)的瞬间，cmd.client就已经是强类型且安全可用的了。

3. “代码即协议事实来源”（Single Source of Truth）： 协议文档无需人工手写。因为有了这些 Pydantic 模型，我们可以通过脚本遍历模型自动导出 JSON Schema，并动态生成WIRE_PROTOCOL.md协议文档。只要改动 Pydantic 模型，文档就能做到 100% 同步更新。