# -*- coding: utf-8 -*- """ ws 通道密钥工具 (协议 QMT_WS_PROTOCOL.md §2.1 / §10.1.1) ========================================================= 用法: # ① 生成一对新密钥。seed 填进 .env 的 PMS_QMT_SIGN_SEED_HEX, 公钥带外交给 QMT 侧 docker compose run --rm pms-web python scripts/gen_keys.py # ② 已经有 seed 了, 只想再把公钥要出来发给对方 (私钥不会打印) docker compose run --rm pms-web python scripts/gen_keys.py --pubkey # ③ 只输出 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。 密钥这种一次性操作不值得跟引号较劲。 """ import os import secrets import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # 协议 §2.1.1 的测试向量 (仅供联调互验, 不得用于生产) VEC_SEED = "00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff" VEC_PUB = "PM0kHP/Js2GARLl9A22GFFk9iwF8NA8d7odzOFUXZUs=" VEC_PAYLOAD = {"instruction_id": "INS-20260728-a3f19c04", "ts_code": "600000.SH", "side": "sell", "qty": 2000, "limit_price": 12.35, "valid_until": 1769512500000, "intent": "TRIM", "note": "保垫减仓·垫厚先收"} VEC_SIG = ("NzCILYw7ampmMtg41EBMubtlnDwr/jjGO+1cMTIxVdqaSaT779Kop5DAdADHL2aSbqrw2sp8" "hCQb+apCACIYCw==") BAR = "=" * 66 def _codec(): from app.core import ws_codec as wsc return wsc def generate(): wsc = _codec() seed = secrets.token_hex(32) print(BAR) print("新的 Ed25519 密钥对已生成") print(BAR) 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("【公钥·裸 base64】协议 §2.1.1 的形态, 自己留档备查:\n") print(f" {wsc.public_key_b64(seed)}\n") 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. 把 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 _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 还没配 —— 先跑一次不带参数的本脚本生成密钥对", file=sys.stderr) return seed def show_pubkey(): wsc = _codec() seed = _own_seed() if not seed: return 1 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 def check(): wsc = _codec() from config.settings import settings failed = [] def ok(msg): print(f" OK {msg}") def bad(msg): failed.append(msg) print(f" FAIL {msg}") print(BAR) print("ws 通道密钥自检") print(BAR) print("\n[1] 协议 §2.1.1 测试向量 (双方实现必须先对上这一组)") canon = wsc.canonical(v=1, type_="place_order", msg_id="m-20260728-000123", ts=1769500000123, nonce="9f2c8a1b7d3e4056", payload_hash=wsc.payload_sha256(VEC_PAYLOAD)) if wsc.public_key_b64(VEC_SEED) == VEC_PUB: ok("由测试 seed 推出的公钥与协议一致") else: bad("测试 seed 推出的公钥对不上 —— cryptography 装错了?") if wsc.sign(canon, VEC_SEED) == VEC_SIG: ok("签名值与协议向量逐字节一致") else: bad("签名值对不上协议向量 —— 规范化串拼接被改坏了, 先看 ws_codec.canonical") print("\n[2] 本端私钥 (PMS_QMT_SIGN_SEED_HEX, 签下发指令用)") seed = (settings.PMS_QMT_SIGN_SEED_HEX or "").strip() if not seed: bad("未配置 —— ws 通道无法签名, 切 ws 模式后指令一律发不出去") else: try: pub = wsc.public_key_b64(seed) sig = wsc.sign(canon, seed) if wsc.verify(canon, sig, pub): 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: bad(f"私钥非法: {type(e).__name__}: {e}") if seed == VEC_SEED: bad("用的是协议里的**测试**密钥 —— 联调可以, 上实盘前必须换掉") print("\n[3] 对端公钥 (PMS_QMT_PEER_PUBKEY_B64, 验上行回报用)") peer = (settings.PMS_QMT_PEER_PUBKEY_B64 or "").strip() if not peer: bad("未配置 —— 所有上行消息都会验签失败被丢弃, 成交一条也收不到") else: form = wsc.pubkey_form(peer) try: raw_b64 = wsc.normalize_pubkey(peer) ok(f"可用; 输入形态 = {form}") if form != "裸 base64 (协议形态)": print(f" 已自动归一成协议形态: {raw_b64}") print(" 注: 格式对不代表就是对方那把 —— 要等对端发第一条消息验签通过") except Exception as 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 leaked = [k for k in param_store.SECRET_KEYS if param_store.get(k)] if leaked: bad(f"密钥可经 ParamStore 读出: {leaked}") else: ok("ParamStore 读不到密钥, 页面参数列表里也不会出现") print("\n" + "-" * 66) if failed: print(f"FAILED: {len(failed)} 项") for m in failed: print(" - " + m) return 1 print("ALL OK —— 密钥就绪, 可以起 pms-ws 了:") print(" docker compose --profile ws up -d pms-ws && docker compose logs -f pms-ws") return 0 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"): 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 / --pem / --check / --help") sys.exit(2) generate() if __name__ == "__main__": main()