From 3a1191660b6f84cf572e3cbd35896d85090effda Mon Sep 17 00:00:00 2001 From: zlt Date: Tue, 28 Jul 2026 17:12:49 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=A0=E5=AF=86=E7=9B=B8=E5=85=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 8 +++- README.md | 15 +++++- app/core/ws_codec.py | 91 ++++++++++++++++++++++++++++++++++-- scripts/gen_keys.py | 85 ++++++++++++++++++++++++--------- scripts/run_tests.py | 2 +- scripts/test_batch6_units.py | 45 ++++++++++++++++++ 6 files changed, 218 insertions(+), 28 deletions(-) diff --git a/.env.example b/.env.example index 265112d..35b1ad4 100644 --- a/.env.example +++ b/.env.example @@ -31,5 +31,11 @@ PMS_WEB_PORT=38100 # 填完两项后自检 (含协议 §2.1.1 测试向量): # docker compose run --rm pms-web python scripts/gen_keys.py --check PMS_QMT_SIGN_SEED_HEX= -# QMT 侧的 Ed25519 公钥 (base64), 用来验上行签名。上下行是两对不同的密钥, 互不复用。 +# QMT 侧的 Ed25519 公钥, 用来验上行签名。上下行是两对不同的密钥, 互不复用。 +# +# 对方多半发的是 PEM 文件。整段 PEM 有换行放不进 .env —— **取中间那行 base64 粘过来**: +# -----BEGIN PUBLIC KEY----- +# MCowBQYDK2VwAyEAOh3Cd30lcdKkU4+VDIvU9FOFeDfJvK7Cx6A0WE+RdWU= ← 粘这一行 +# -----END PUBLIC KEY----- +# 裸 32 字节 base64 / 64 位 hex 也都认, 代码会自动归一 (ws_codec.normalize_pubkey)。 PMS_QMT_PEER_PUBKEY_B64= diff --git a/README.md b/README.md index bbfe801..6310fec 100644 --- a/README.md +++ b/README.md @@ -196,8 +196,10 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶ ### 切 ws 通道的操作顺序 ```bash -# 1. 生成本端密钥对: seed 填进 .env 的 PMS_QMT_SIGN_SEED_HEX, 公钥带外交给 QMT 侧 +# 1. 生成本端密钥对: seed 填进 .env 的 PMS_QMT_SIGN_SEED_HEX docker compose run --rm pms-web python scripts/gen_keys.py +# 公钥交给 QMT 侧要 PEM 形态 (他们用 load_pem_public_key 读文件): +docker compose run --rm pms-web python scripts/gen_keys.py --pem > pms_public.pem # 2. 把 QMT 侧公钥填进 .env 的 PMS_QMT_PEER_PUBKEY_B64; 把本机内网 IP 报给对方加白 # 3. 密钥自检 (含协议 §2.1.1 测试向量, 双方各跑一次对上再联调) docker compose run --rm pms-web python scripts/gen_keys.py --check @@ -210,7 +212,16 @@ docker compose --profile ws up -d pms-ws && docker compose logs -f pms-ws # 7. S2/S3 联调通过后, 页面把 PMS_DISPATCH_MODE 改成 ws ``` -> `scripts/gen_keys.py` 还有 `--pubkey`（只打印当前私钥对应的公钥，不打印私钥——公钥发丢了重发用）。**别用 `python -c` 加 shell 续行 `\` 拼这段**：续行会把下一行的行首缩进一起带进 Python 源码字符串，直接 `IndentationError: unexpected indent`。 +> `scripts/gen_keys.py` 还有 `--pubkey`（打印当前私钥对应的公钥两种形态，不打印私钥）。**别用 `python -c` 加 shell 续行 `\` 拼这段**：续行会把下一行的行首缩进一起带进 Python 源码字符串，直接 `IndentationError: unexpected indent`。 + +**公钥的两种形态**（2026-07-28 交换密钥时踩到，两边要的不是同一种）： + +| 谁 | 要什么 | 长什么样 | +|---|---|---| +| PMS 的 `.env`（`PMS_QMT_PEER_PUBKEY_B64`） | 一行字符串 | 裸 32 字节 base64 `Oh3Cd30l...`；也接受 PEM 正文那一行、PEM 全文、64 位 hex | +| QMT 侧的 `crypto.py`（`load_pem_public_key`） | 一个 `.pem` 文件 | `-----BEGIN PUBLIC KEY-----` 三行 | + +两者装的是同一把钥匙：PEM 正文解出来是 44 字节 = 12 字节 SPKI 固定头 + 32 字节裸公钥。协议 §2.1.1 正文用的是裸 base64，所以把 PEM 正文原样粘进 `.env` 会报「44 字节，长度不对」。`ws_codec.normalize_pubkey()` 现在四种形态全收并自动归一——对方发来什么直接粘上去即可，`gen_keys.py --check` 会告诉你它识别成了哪种。 任一步不放心都可以退回去：把 `PMS_DISPATCH_MODE` 改回 `shadow` 即恢复人工执行，`pms-ws` 停掉也只是让指令拒发（保持原状），不会产生半截状态。 diff --git a/app/core/ws_codec.py b/app/core/ws_codec.py index 0315beb..bb13ed9 100644 --- a/app/core/ws_codec.py +++ b/app/core/ws_codec.py @@ -163,20 +163,105 @@ def sign(canon: str, seed_hex: str) -> str: def public_key_b64(seed_hex: str) -> str: - """由私钥 seed 推出公钥 —— 部署时把它带外送给 QMT 侧 (§10.1.1 第 2 条)。""" + """由私钥 seed 推出公钥, **裸 32 字节 base64** —— 协议 §2.1.1 正文用的形态。""" if not CRYPTO_AVAILABLE: raise CodecError("INTERNAL", "缺少 cryptography 依赖") sk = Ed25519PrivateKey.from_private_bytes(_seed_bytes(seed_hex)) return base64.b64encode(sk.public_key().public_bytes_raw()).decode() +def public_key_pem(seed_hex: str) -> str: + """由私钥 seed 推出公钥的 **PEM** 形态。 + + QMT 侧的 `crypto.py` 用 `serialization.load_pem_public_key(path)` 加载我方公钥, + 也就是说他们要的是一个 `.pem` 文件而不是一串裸 base64。带外交换密钥时把这个给过去, + 对方存成文件即可直接用 —— 省掉一轮"你发的这串我读不出来"。 + """ + if not CRYPTO_AVAILABLE: + raise CodecError("INTERNAL", "缺少 cryptography 依赖") + from cryptography.hazmat.primitives import serialization + sk = Ed25519PrivateKey.from_private_bytes(_seed_bytes(seed_hex)) + return sk.public_key().public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo).decode() + + +# Ed25519 的 SPKI(DER) 固定前缀: 12 字节头 + 32 字节裸公钥 = 44 字节。 +# PEM 正文那一行 base64 解出来就是这 44 字节。 +_SPKI_ED25519_PREFIX = bytes.fromhex("302a300506032b6570032100") + + +def _pubkey_bytes(value: str) -> bytes: + """把对端公钥的各种形态收敛成 32 字节裸公钥。 + + 协议 §2.1.1 写的公钥是**裸 32 字节 base64**(`PM0kHP/...`), 而 cryptography 默认 + 导出的是 PEM —— 正文是 44 字节的 SPKI DER(12 字节固定头 + 32 字节裸公钥)。对方按 + PEM 交换密钥是很自然的做法 (他们的 crypto.py 就是 `load_pem_public_key`), 直接把 + PEM 正文粘进 .env 会得到"44 字节, 长度不对"这种让人一头雾水的报错。 + + 所以这里四种形态全收: + 裸 base64 44 字符 → 32 字节 协议正文形态 + SPKI base64 60 字符 → 44 字节 PEM 正文单行, 自动剥掉 12 字节头 + PEM 全文 含 -----BEGIN----- 头尾 真换行或 .env 里的字面 \\n 都认 + hex 64 字符 → 32 字节 顺手也认 + + 多认几种格式换来的是"对方发来什么直接粘上去就能用"。这类互通摩擦不值得让两边改代码, + 更不值得在联调现场耗掉一轮。 + """ + s = str(value or "").strip() + if not s: + raise CodecError("SIG_INVALID", "公钥为空") + if "BEGIN" in s.upper(): + s = s.replace("\\n", "\n") # .env 里常见的字面 \n + s = "".join(ln for ln in s.splitlines() if "-----" not in ln) + s = "".join(s.split()) # 去掉所有空白与换行 + if len(s) == 64 and all(c in "0123456789abcdefABCDEF" for c in s): + return bytes.fromhex(s) + try: + raw = base64.b64decode(s, validate=True) + except Exception as e: + raise CodecError("SIG_INVALID", f"公钥既不是合法 base64 也不是 64 位 hex: {e}") from e + if len(raw) == 32: + return raw + if len(raw) == 44 and raw.startswith(_SPKI_ED25519_PREFIX): + return raw[len(_SPKI_ED25519_PREFIX):] + raise CodecError("SIG_INVALID", + f"不是 Ed25519 公钥: 解出 {len(raw)} 字节 " + f"(裸公钥应为 32 字节, PEM/SPKI 正文应为 44 字节且带固定头)") + + +def normalize_pubkey(value: str) -> str: + """任意形态 → 协议用的裸 32 字节 base64。非法输入抛 CodecError。""" + return base64.b64encode(_pubkey_bytes(value)).decode() + + +def pubkey_form(value: str) -> str: + """识别输入是哪种形态, 给自检脚本输出人话用。""" + s = str(value or "").strip() + if not s: + return "空" + if "BEGIN" in s.upper(): + return "PEM 全文" + s2 = "".join(s.split()) + if len(s2) == 64 and all(c in "0123456789abcdefABCDEF" for c in s2): + return "hex" + try: + n = len(base64.b64decode(s2, validate=True)) + except Exception: + return "无法识别" + return {32: "裸 base64 (协议形态)", 44: "SPKI base64 (PEM 正文)"}.get(n, f"{n} 字节, 非法") + + def verify(canon: str, sig_b64: str, pubkey_b64: str) -> bool: """验签。任何异常 (公钥非法/签名非法/长度不对/base64 烂) 一律 False —— 验不过就是验不过, - 不区分「为什么验不过」, 免得给探测方留下侧信道。""" + 不区分「为什么验不过」, 免得给探测方留下侧信道。 + + 公钥四种形态都接受, 见 _pubkey_bytes。 + """ if not CRYPTO_AVAILABLE: raise CodecError("INTERNAL", "缺少 cryptography 依赖, 无法验签") try: - pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pubkey_b64)) + pk = Ed25519PublicKey.from_public_bytes(_pubkey_bytes(pubkey_b64)) pk.verify(base64.b64decode(sig_b64), canon.encode("utf-8")) return True except Exception: # noqa: BLE001 —— 见上, 故意不细分 diff --git a/scripts/gen_keys.py b/scripts/gen_keys.py index e0f6ba5..0f86b29 100644 --- a/scripts/gen_keys.py +++ b/scripts/gen_keys.py @@ -10,9 +10,19 @@ ws 通道密钥工具 (协议 QMT_WS_PROTOCOL.md §2.1 / §10.1.1) # ② 已经有 seed 了, 只想再把公钥要出来发给对方 (私钥不会打印) docker compose run --rm pms-web python scripts/gen_keys.py --pubkey - # ③ 自检: 跑协议 §2.1.1 的测试向量 + 校验当前 .env 里两把钥匙是否可用 + # ③ 只输出 PEM (无任何装饰, 可直接重定向存文件交给 QMT 侧) + docker compose run --rm pms-web python scripts/gen_keys.py --pem > pms_public.pem + + # ④ 自检: 跑协议 §2.1.1 的测试向量 + 校验当前 .env 里两把钥匙是否可用 docker compose run --rm pms-web python scripts/gen_keys.py --check +公钥的两种形态 (2026-07-28 联调踩到, 记一下): + 裸 32 字节 base64 `PM0kHP/...` —— 协议 §2.1.1 正文用的, 填进本端 .env 的那种 + PEM / SPKI `-----BEGIN PUBLIC KEY-----` —— QMT 侧 crypto.py 用 + `load_pem_public_key()` 加载, 他们要的是这种 +两者装的是同一把钥匙: PEM 正文解出来是 44 字节 = 12 字节固定头 + 32 字节裸公钥。 +本端 ws_codec 四种形态 (裸/SPKI/PEM/hex) 都认, 对方发来什么直接粘进 .env 即可。 + 为什么单独写个脚本而不是一行 `python -c`: 那条命令里既有中文、又有单双引号、还得换行, 用 shell 续行 `\\` 拼的话行首缩进会一起进到 python 的源码字符串里, 直接 IndentationError。 密钥这种一次性操作不值得跟引号较劲。 @@ -49,26 +59,54 @@ def generate(): print("\n【私钥 seed】填进 .env 的 PMS_QMT_SIGN_SEED_HEX, 到此为止 —— 不入库、不进") print(" ParamStore、不进代码、不发给任何人 (协议 §10.1.1):\n") print(f" PMS_QMT_SIGN_SEED_HEX={seed}\n") - print("【公钥】带外交给 QMT 侧 (对方用它验我们下发的指令签名):\n") + print("【公钥·裸 base64】协议 §2.1.1 的形态, 自己留档备查:\n") print(f" {wsc.public_key_b64(seed)}\n") - print("接下来:") + print("【公钥·PEM】**交给 QMT 侧的是这个** —— 他们的 crypto.py 用") + print(" load_pem_public_key() 从 .pem 文件加载:\n") + for ln in wsc.public_key_pem(seed).strip().splitlines(): + print(f" {ln}") + print("\n 直接存成文件给对方:") + print(" docker compose run --rm pms-web python scripts/gen_keys.py --pem " + "> pms_public.pem") + print("\n接下来:") print(" 1. 把上面的 seed 写进 .env; 确认 .env 在 .gitignore 里 (本仓库已配)") - print(" 2. 把公钥用 U 盘/内网 IM 等带外方式发给 QMT 侧") - print(" 3. 拿到对方的公钥, 填进 .env 的 PMS_QMT_PEER_PUBKEY_B64") + print(" 2. 把 PEM 用 U 盘/内网 IM 等带外方式发给 QMT 侧") + print(" 3. 拿到对方的公钥, 原样粘进 .env 的 PMS_QMT_PEER_PUBKEY_B64 ——") + print(" 裸 base64 / PEM 正文那一行 / hex 都认。PEM 全文有换行不好放 .env,") + print(" 取中间那行 base64 粘上去就行") print(" 4. 双方各跑一次 --check, 用协议 §2.1.1 的测试向量互验") print(" 5. 把本机内网 IP 报给对方加入入站白名单 (§10.1.1)") print("\n注意: 这条命令把私钥打到了终端上。用完清一下 shell 历史与终端回滚缓冲,") print(" 如果终端有日志转存, 也一并处理掉。") -def show_pubkey(): - wsc = _codec() +def _own_seed(): from config.settings import settings seed = (settings.PMS_QMT_SIGN_SEED_HEX or "").strip() if not seed: - print("PMS_QMT_SIGN_SEED_HEX 还没配 —— 先跑一次不带参数的本脚本生成密钥对") + print("PMS_QMT_SIGN_SEED_HEX 还没配 —— 先跑一次不带参数的本脚本生成密钥对", + file=sys.stderr) + return seed + + +def show_pubkey(): + wsc = _codec() + seed = _own_seed() + if not seed: return 1 - print(f"本端公钥 (交给 QMT 侧): {wsc.public_key_b64(seed)}") + print(f"裸 base64 (协议形态): {wsc.public_key_b64(seed)}") + print("PEM (交给 QMT 侧, 他们用 load_pem_public_key 加载):") + print(wsc.public_key_pem(seed).strip()) + return 0 + + +def show_pem(): + """只吐 PEM, 不带任何装饰 —— 好让 `> pms_public.pem` 直接得到一个可用文件。""" + wsc = _codec() + seed = _own_seed() + if not seed: + return 1 + sys.stdout.write(wsc.public_key_pem(seed)) return 0 @@ -110,8 +148,10 @@ def check(): pub = wsc.public_key_b64(seed) sig = wsc.sign(canon, seed) if wsc.verify(canon, sig, pub): - ok(f"可用; 本端公钥 = {pub}") - print(" ↑ 这一串交给 QMT 侧, 对方用它验我们的签名") + ok(f"可用; 本端公钥 (裸 base64) = {pub}") + print(" 交给 QMT 侧的是 PEM 形态 (他们用 load_pem_public_key 加载):") + for ln in wsc.public_key_pem(seed).strip().splitlines(): + print(f" {ln}") else: bad("自签自验没通过 —— 密钥有问题") except Exception as e: @@ -124,16 +164,17 @@ def check(): if not peer: bad("未配置 —— 所有上行消息都会验签失败被丢弃, 成交一条也收不到") else: - import base64 + form = wsc.pubkey_form(peer) try: - raw = base64.b64decode(peer, validate=True) - if len(raw) != 32: - bad(f"不是 32 字节 Ed25519 公钥 (实际 {len(raw)} 字节)") - else: - ok(f"格式正确 ({peer[:16]}...)") - print(" 注: 格式对不代表就是对方那把 —— 要等对端发第一条消息验签通过") + raw_b64 = wsc.normalize_pubkey(peer) + ok(f"可用; 输入形态 = {form}") + if form != "裸 base64 (协议形态)": + print(f" 已自动归一成协议形态: {raw_b64}") + print(" 注: 格式对不代表就是对方那把 —— 要等对端发第一条消息验签通过") except Exception as e: - bad(f"不是合法 base64: {e}") + bad(f"无法解析 (识别为「{form}」): {e}") + print(" 提示: 对方给的若是 PEM, 取 -----BEGIN----- 与 -----END----- 之间") + print(" 那一行 base64 粘进 .env 即可 (整段 PEM 有换行, 放不进 .env)") print("\n[4] 密钥不落库自证 (协议 §10.1.1)") from app.services import param_store @@ -158,13 +199,15 @@ def main(): arg = (sys.argv[1] if len(sys.argv) > 1 else "").strip().lstrip("-").lower() if arg in ("check", "c"): sys.exit(check()) - if arg in ("pubkey", "pub", "p"): + if arg in ("pubkey", "pub"): sys.exit(show_pubkey()) + if arg in ("pem", "p"): + sys.exit(show_pem()) if arg in ("help", "h"): print(__doc__) sys.exit(0) if arg: - print(f"未知参数 {sys.argv[1]!r}; 可用: (无) / --pubkey / --check / --help") + print(f"未知参数 {sys.argv[1]!r}; 可用: (无) / --pubkey / --pem / --check / --help") sys.exit(2) generate() diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 7fac3c5..bcb5673 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -10,7 +10,7 @@ test_batch3_units.py 规则闸 / 择时执行器实现B 纯逻辑 (19 例) test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 (11 例) test_batch5_units.py 决策系统信号流解析与消化口径 (8 例) - test_batch6_units.py ws 通道: 协议测试向量/签名/水位/单表守卫 (44 例) + test_batch6_units.py ws 通道: 协议测试向量/签名/公钥形态/水位/单表守卫 (49 例) test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (32 例) 任一子集失败即整体失败 (退出码 1)。 """ diff --git a/scripts/test_batch6_units.py b/scripts/test_batch6_units.py index f16e6eb..b34deb0 100644 --- a/scripts/test_batch6_units.py +++ b/scripts/test_batch6_units.py @@ -16,6 +16,7 @@ G. 单表访问守卫: upsert (ON DUPLICATE KEY UPDATE) 不得被误判成多表。 H. dispatcher 纯助手: valid_until 归一、子单→父指令反推。 """ +import base64 import os import sys import traceback @@ -126,6 +127,50 @@ def run(): except wsc.CodecError as e: eq(e.code, "SIG_INVALID") + # --- 公钥形态。2026-07-28 与 QMT 侧交换密钥时踩到: 对方按 PEM 交换 (其 crypto.py + # 用 load_pem_public_key), 而协议 §2.1.1 正文写的是裸 32 字节 base64。PEM 正文 + # 解出来是 44 字节 SPKI DER = 12 字节固定头 + 32 字节裸公钥, 直接粘进 .env 会得到 + # 「44 字节, 长度不对」这种看不出所以然的报错。现在四种形态全收。 + VEC_SPKI = "MCowBQYDK2VwAyEA" + VEC_PUB + VEC_PEM = f"-----BEGIN PUBLIC KEY-----\n{VEC_SPKI}\n-----END PUBLIC KEY-----\n" + + @case("公钥四种形态都归一到裸 32 字节 base64") + def _(): + for form in (VEC_PUB, VEC_SPKI, VEC_PEM, + VEC_PEM.replace("\n", "\\n"), # .env 里的字面 \n + base64.b64decode(VEC_PUB).hex()): # 64 位 hex + eq(wsc.normalize_pubkey(form), VEC_PUB, f"归一失败: {form[:24]}...") + + @case("公钥四种形态都能直接验签") + def _(): + for form in (VEC_PUB, VEC_SPKI, VEC_PEM, VEC_PEM.replace("\n", "\\n")): + assert wsc.verify(VEC_CANON, VEC_SIG, form), f"验签失败: {form[:24]}..." + + @case("公钥形态识别 (自检脚本给人话提示用)") + def _(): + eq(wsc.pubkey_form(VEC_PUB), "裸 base64 (协议形态)") + eq(wsc.pubkey_form(VEC_SPKI), "SPKI base64 (PEM 正文)") + eq(wsc.pubkey_form(VEC_PEM), "PEM 全文") + eq(wsc.pubkey_form(""), "空") + + @case("公钥非法输入仍然拒绝, 不能因为放宽格式就什么都收") + def _(): + for bad_ in ("", "zzz!!!", "aGVsbG8=", "MCowBQYDK2VwAyEA"): # 空/非法/太短/只有头 + try: + wsc.normalize_pubkey(bad_) + raise AssertionError(f"应当拒绝: {bad_!r}") + except wsc.CodecError as e: + eq(e.code, "SIG_INVALID", f"输入 {bad_!r}") + + @case("本端公钥的 PEM 形态可被对方 load_pem_public_key 读取") + def _(): + pem = wsc.public_key_pem(VEC_SEED) + assert pem.startswith("-----BEGIN PUBLIC KEY-----"), pem[:40] + eq(wsc.normalize_pubkey(pem), VEC_PUB, "PEM 与裸 base64 必须是同一把钥匙") + from cryptography.hazmat.primitives import serialization + k = serialization.load_pem_public_key(pem.encode()) # 对方那条路径 + eq(base64.b64encode(k.public_bytes_raw()).decode(), VEC_PUB) + print("\n[C] 信封组包与解析") @case("build → parse 往返一致")