配置导入接口返回成功，回读结果却与提交内容不一致。先检查客户端发出的原始 JSON：同一个对象里可能出现了两次 quota，而解析器只留下了后一个值。等数据变成字典，再严格的字段校验也找不回被覆盖的那一项。
本文以“批量任务额度配置”为例，把排查推进到解析入口，并提供可以独立运行的实现与回归测试。示例使用 Python 标准库，不依赖 Web 框架。
一、适用场景与数据合同
适用于接收 JSON 配置文件、管理接口请求体或消息内容的 Python 服务。示例约定一个完整替换配置的入口：
- 输入为不超过 64 KiB 的 UTF-8 原始字节，不接受 BOM。
- 顶层必须是对象，且恰好包含
enabled与quota两个字段。 enabled必须是布尔值；quota必须是 1 到 1000 的整数，不接受布尔值冒充整数。- 任意层级的对象都不允许重复键；非有限数值也必须拒绝。
这是一份明确的业务合同。真实接口若支持部分更新，应单独设计缺省值与字段删除语义，不要直接照搬“恰好两个字段”的限制。
二、先复现：strict=True 仍然覆盖重复键
把下面代码保存为 reproduce.py，执行 python reproduce.py：
import json
raw = b'{"enabled":true,"quota":10,"quota":900}'
parsed = json.loads(raw, strict=True)
assert parsed == {"enabled": True, "quota": 900}
print(parsed)
这里没有解析异常，quota 最终是 900。strict=True 主要控制字符串内未转义控制字符的处理，并不提供重复键校验；Python 默认解码行为保留同名键的最后一个值。相关行为见 Python 3.14 json 文档。
因此，“已经调用 json.loads”“已经使用 strict=True”“已经用模型校验 quota 范围”都不能证明请求不存在重复键。900 本来就在允许范围内，后置的业务模型完全可能放行。
三、定位时追踪解析发生的位置
先在开发环境用合成请求重现，再沿入口检查以下顺序：
- 请求字节在哪一层第一次被转换成字典：网关、框架中间件、请求模型，还是业务函数？
- 校验函数接收的是字节还是已经解析完成的对象？若是后者，重复信息已经丢失。
- 客户端是否通过字符串拼接合并配置，而非构造对象后统一序列化？
- 不同语言服务是否采用不同的重复键处理策略？不要假设上下游都选最后一个值。
不要为排查开启原始请求体日志。生产环境只统计入口、错误类型、请求标识和请求字节数；用不含真实账户数据的固定样本重放。若涉及请求验签，继续针对原始字节验签，验签成功后再执行本文解析，不能先重序列化再验签。
四、实现：在键值对变成字典之前拒绝重复
object_pairs_hook 接收每个 JSON 对象的键值对列表，可以在构造字典前检查同名键。object_hook 接收的已经是字典，时间太晚。该钩子也会处理嵌套对象和数组中的对象，参数含义见 官方解码接口说明。
将以下内容保存为 strict_config.py：
"""在配置入口拒绝含歧义的 JSON，并校验业务字段。"""
import json
import math
from dataclasses import dataclass
from typing import NoReturn
MAX_BYTES = 64 * 1024
class ConfigInputError(ValueError):
"""表示可向调用方返回的配置输入错误。"""
@dataclass(frozen=True)
class JobConfig:
"""表示已完成边界校验的任务配置。"""
enabled: bool
quota: int
def reject_duplicate_keys(pairs: list[tuple[str, object]]) -> dict[str, object]:
"""在重复键尚未被字典覆盖时中断解析。"""
result: dict[str, object] = {}
for key, value in pairs:
if key in result:
# 键名同样来自外部输入，不将其拼入可读错误。
raise ConfigInputError("JSON 对象包含重复键")
result[key] = value
return result
def reject_constant(value: str) -> NoReturn:
"""拒绝解码器扩展支持的非标准数值常量。"""
raise ConfigInputError("JSON 不允许非有限数值")
def parse_finite_float(value: str) -> float:
"""同时拦截标准指数写法转换成无穷大的情况。"""
number = float(value)
if not math.isfinite(number):
raise ConfigInputError("JSON 浮点数超出有限范围")
return number
def load_job_config(raw: bytes) -> JobConfig:
"""从有大小限制的原始字节构造可供业务使用的配置。"""
if not isinstance(raw, bytes):
raise ConfigInputError("配置入口必须接收原始字节")
if len(raw) > MAX_BYTES:
raise ConfigInputError("配置超过 64 KiB 限制")
try:
text = raw.decode("utf-8")
except UnicodeDecodeError as exc:
raise ConfigInputError("配置必须使用 UTF-8 编码") from exc
if text.startswith("\ufeff"):
raise ConfigInputError("配置不得包含 UTF-8 BOM")
try:
data = json.loads(
text,
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_constant,
parse_float=parse_finite_float,
)
except ConfigInputError:
raise
except (ValueError, RecursionError) as exc:
raise ConfigInputError("JSON 语法或解析限制校验失败") from exc
if not isinstance(data, dict):
raise ConfigInputError("配置顶层必须是对象")
if set(data) != {"enabled", "quota"}:
raise ConfigInputError("配置必须且只能包含 enabled 与 quota")
enabled = data["enabled"]
quota = data["quota"]
if type(enabled) is not bool:
raise ConfigInputError("enabled 必须是布尔值")
# bool 是 int 的子类，需要避免 True 被当成额度 1。
if type(quota) is not int or not 1 <= quota <= 1000:
raise ConfigInputError("quota 必须是 1 到 1000 的整数")
return JobConfig(enabled=enabled, quota=quota)
几个实现细节需要看清：
- 不用正则扫描键名。转义后的
"qu\u006fta"与"quota"会解码成同一个字符串，必须在解码器提供的键值对中比较。 parse_constant拒绝NaN、Infinity、-Infinity；1e400不走这个回调，所以额外检查浮点转换结果。此处拒绝无穷大是入口合同，不代表 JSON 语法本身禁止这种指数写法。- 捕获
ValueError同时覆盖 JSON 语法异常及整数转换限制；显式保留自定义校验错误，避免把重复键原因改写成笼统异常。 - 错误链用于保留内部诊断上下文，但底层异常可能持有文档或输入片段。返回错误时只取安全的外层说明，日志中也不要直接记录异常链或开启
exc_info=True。
五、回归测试：不能只检查最终字典
将下面代码保存为同目录下的 test_strict_config.py。重复键样本必须保留原始字符串，不能先创建字典再 json.dumps，否则测试还没开始，重复键就已经消失。
"""验证正常配置、重复键与输入边界。"""
import unittest
from strict_config import MAX_BYTES, ConfigInputError, JobConfig, load_job_config
class JobConfigTests(unittest.TestCase):
"""覆盖解析入口与业务合同。"""
def test_valid_boundaries(self) -> None:
for quota in (1, 1000):
with self.subTest(quota=quota):
raw = ('{"enabled":true,"quota":%d}' % quota).encode("utf-8")
self.assertEqual(load_job_config(raw), JobConfig(True, quota))
def test_duplicate_top_level(self) -> None:
with self.assertRaisesRegex(ConfigInputError, "重复键"):
load_job_config(b'{"enabled":true,"quota":10,"quota":900}')
def test_duplicate_nested_objects(self) -> None:
for raw in (
b'{"extra":{"x":1,"x":2}}',
b'{"extra":[{"x":1,"x":2}]}',
):
with self.subTest(raw=raw):
with self.assertRaisesRegex(ConfigInputError, "重复键"):
load_job_config(raw)
def test_escaped_duplicate(self) -> None:
raw = rb'{"enabled":true,"quota":10,"qu\u006fta":900}'
with self.assertRaisesRegex(ConfigInputError, "重复键"):
load_job_config(raw)
def test_nonfinite_numbers(self) -> None:
for value in (b"NaN", b"Infinity", b"-Infinity", b"1e400"):
with self.subTest(value=value):
with self.assertRaisesRegex(ConfigInputError, "有限"):
load_job_config(b'{"enabled":true,"quota":' + value + b"}")
def test_invalid_business_fields(self) -> None:
for value in (b"true", b"false", b"0", b"1001", b"1.0", b'"10"', b"null"):
with self.subTest(value=value):
with self.assertRaisesRegex(ConfigInputError, "quota"):
load_job_config(b'{"enabled":true,"quota":' + value + b"}")
with self.assertRaisesRegex(ConfigInputError, "enabled"):
load_job_config(b'{"enabled":1,"quota":10}')
def test_invalid_shape(self) -> None:
for raw in (b"[]", b"null", b"{}", b'{"enabled":true,"quota":1,"extra":0}'):
with self.subTest(raw=raw):
with self.assertRaises(ConfigInputError):
load_job_config(raw)
def test_invalid_encoding(self) -> None:
for raw in (b"\xff", b"\xef\xbb\xbf{}"):
with self.subTest(raw=raw):
with self.assertRaisesRegex(ConfigInputError, "UTF-8"):
load_job_config(raw)
def test_invalid_syntax(self) -> None:
for raw in (b"", b'{"enabled":true,}', b"{} {}"):
with self.subTest(raw=raw):
with self.assertRaisesRegex(ConfigInputError, "解析限制"):
load_job_config(raw)
def test_size_limit(self) -> None:
raw = b'{"enabled":true,"quota":1}'
padded = raw + b" " * (MAX_BYTES - len(raw))
self.assertEqual(load_job_config(padded), JobConfig(True, 1))
with self.assertRaisesRegex(ConfigInputError, "64 KiB"):
load_job_config(padded + b" ")
if __name__ == "__main__":
unittest.main()
执行 python -m unittest -v test_strict_config，应明确看到执行了 10 项测试，不能把“Ran 0 tests”视为成功。第二、第三、第四项还会确认错误确实来自重复键检测，避免未知字段检查偶然让错误实现通过测试。
六、接入、上线与回退
将读取字节、解析、鉴权、写配置分别放在明确边界：完成身份认证与权限校验后，调用 load_job_config，只有返回 JobConfig 后才能进入持久化或任务调度逻辑。不要把框架已经解析的 dict 再编码后传给它，那样无法恢复重复信息。
读取文件时使用二进制模式，最多读取 MAX_BYTES + 1 字节，再交给该函数判断；读取 HTTP 请求时，应在网关和服务接收层限制实际读取量，并对解压后的体积施加限制。不能只信任 Content-Length，也不能等完整请求体进入内存后才认为已经控制了接收成本。
上线前先用合成报文覆盖完整链路，确认框架没有抢先解析；再回放经批准的脱敏样本并修复客户端序列化。错误指标可以用稳定原因码，如 duplicate_key、invalid_encoding、invalid_schema，与 request_id、body_bytes 一起记录，固定说明使用中文。解析失败映射为合同约定的客户端错误，不落库、不触发任务。
如果切换后失败量上升，应暂停扩大流量、定位并回退产生错误报文的客户端，保留服务端拒绝歧义输入的保护；已有配置继续按最近一次校验通过的版本服务。不要对解析失败的同一请求自动重试，也不要自动挑选第一个或最后一个重复值后继续写入。
这里的 64 KiB 限制与 RecursionError 转换只是基础防线，没有提供固定的嵌套深度预算，也没有证明高并发资源消耗可接受。面对公网或高度不可信的大规模输入，还需具备明确深度/节点限制的解析方案以及服务资源隔离。示例不负责调用者权限、配置版本冲突和并发更新控制。
七、验证范围与结论
本文发布前在 Python 3.14.6 中执行复现代码及正文的 10 项 unittest，覆盖顶层/嵌套/转义重复键、非有限数值、字段类型与范围、编码、语法和大小边界。未执行真实 Web 框架、中间件、网关或生产流量集成测试，也未验证其他 Python 版本；本地环境未提供 Ruff，未安装额外依赖。
这类问题的关键在于检查时机：重复键检测必须发生在键值对被字典覆盖之前，业务字段校验则负责随后确认数据是否满足合同。将两步都放在可测试的入口，才能同时防止静默覆盖和错误类型进入业务逻辑。
Discussion
评论