diff --git a/.env.example b/.env.example index 6615eb2..265112d 100644 --- a/.env.example +++ b/.env.example @@ -26,11 +26,10 @@ PMS_WEB_PORT=38100 # ---- QMT WebSocket 直连通道的密钥 (协议 QMT_WS_PROTOCOL.md §10.1.1) ---- # 这两项是**密钥**, 只在这里维护: 不入库、不进 ParamStore、不上管理页面、不进代码。 -# 生成一对新密钥并打印公钥 (公钥带外交给 QMT 侧, 私钥不出机器): -# docker compose run --rm pms-web python -c "\ -# import secrets; from app.core import ws_codec as w; \ -# s=secrets.token_hex(32); print('seed(私钥, 填下面):', s); \ -# print('pubkey(公钥, 给对方):', w.public_key_b64(s))" +# 生成一对新密钥 (公钥带外交给 QMT 侧, 私钥不出机器): +# docker compose run --rm pms-web python scripts/gen_keys.py +# 填完两项后自检 (含协议 §2.1.1 测试向量): +# docker compose run --rm pms-web python scripts/gen_keys.py --check PMS_QMT_SIGN_SEED_HEX= # QMT 侧的 Ed25519 公钥 (base64), 用来验上行签名。上下行是两对不同的密钥, 互不复用。 PMS_QMT_PEER_PUBKEY_B64= diff --git a/README.md b/README.md index 5b72f38..bbfe801 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,7 @@ scripts/ test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 32 例 init_db.py 建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练) check_db.py 实机连通性与表结构自检 (需真实 .env) + gen_keys.py ws 通道密钥: 生成 / 只取公钥(--pubkey) / 自检(--check) ``` ## 三条铁律 @@ -195,13 +196,11 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶ ### 切 ws 通道的操作顺序 ```bash -# 1. 生成本端密钥对, seed 填进 .env, 公钥带外交给 QMT 侧 -docker compose run --rm pms-web python -c "\ - import secrets; from app.core import ws_codec as w; \ - s=secrets.token_hex(32); print('seed:', s); print('pubkey:', w.public_key_b64(s))" +# 1. 生成本端密钥对: seed 填进 .env 的 PMS_QMT_SIGN_SEED_HEX, 公钥带外交给 QMT 侧 +docker compose run --rm pms-web python scripts/gen_keys.py # 2. 把 QMT 侧公钥填进 .env 的 PMS_QMT_PEER_PUBKEY_B64; 把本机内网 IP 报给对方加白 -# 3. 双方各自用协议 §2.1.1 的测试向量互验 —— 单测已覆盖本端: -docker compose run --rm pms-web python scripts/test_batch6_units.py +# 3. 密钥自检 (含协议 §2.1.1 测试向量, 双方各跑一次对上再联调) +docker compose run --rm pms-web python scripts/gen_keys.py --check # 4. 建新表 (幂等) 并自检 docker compose run --rm pms-web python scripts/init_db.py --yes docker compose run --rm pms-web python scripts/check_db.py # [6] 段看通道状态 @@ -211,6 +210,8 @@ 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`。 + 任一步不放心都可以退回去：把 `PMS_DISPATCH_MODE` 改回 `shadow` 即恢复人工执行，`pms-ws` 停掉也只是让指令拒发（保持原状），不会产生半截状态。 **待外部协商（剩余）**：`QMT_INTERFACE_REQUIREMENTS.md` 的 A/C/D 各项。已闭环的有：A2 状态枚举与成交均价、A3 资金快照、A4 逐笔成交、Q1~Q11 与 R1/R2（详见协议 §10）。已知存疑项：`trading_log.extra_data` 里的 `total_filled` 口径与示例数据矛盾，PMS 事后核对绕开该列，只读 `traded_volume / traded_price / traded_amount`（协议 §10.2）。 diff --git a/scripts/gen_keys.py b/scripts/gen_keys.py new file mode 100644 index 0000000..e0f6ba5 --- /dev/null +++ b/scripts/gen_keys.py @@ -0,0 +1,173 @@ +# -*- 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 + + # ③ 自检: 跑协议 §2.1.1 的测试向量 + 校验当前 .env 里两把钥匙是否可用 + docker compose run --rm pms-web python scripts/gen_keys.py --check + +为什么单独写个脚本而不是一行 `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("【公钥】带外交给 QMT 侧 (对方用它验我们下发的指令签名):\n") + print(f" {wsc.public_key_b64(seed)}\n") + print("接下来:") + print(" 1. 把上面的 seed 写进 .env; 确认 .env 在 .gitignore 里 (本仓库已配)") + print(" 2. 把公钥用 U 盘/内网 IM 等带外方式发给 QMT 侧") + print(" 3. 拿到对方的公钥, 填进 .env 的 PMS_QMT_PEER_PUBKEY_B64") + print(" 4. 双方各跑一次 --check, 用协议 §2.1.1 的测试向量互验") + print(" 5. 把本机内网 IP 报给对方加入入站白名单 (§10.1.1)") + print("\n注意: 这条命令把私钥打到了终端上。用完清一下 shell 历史与终端回滚缓冲,") + print(" 如果终端有日志转存, 也一并处理掉。") + + +def show_pubkey(): + wsc = _codec() + from config.settings import settings + seed = (settings.PMS_QMT_SIGN_SEED_HEX or "").strip() + if not seed: + print("PMS_QMT_SIGN_SEED_HEX 还没配 —— 先跑一次不带参数的本脚本生成密钥对") + return 1 + print(f"本端公钥 (交给 QMT 侧): {wsc.public_key_b64(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"可用; 本端公钥 = {pub}") + print(" ↑ 这一串交给 QMT 侧, 对方用它验我们的签名") + 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: + import base64 + try: + raw = base64.b64decode(peer, validate=True) + if len(raw) != 32: + bad(f"不是 32 字节 Ed25519 公钥 (实际 {len(raw)} 字节)") + else: + ok(f"格式正确 ({peer[:16]}...)") + print(" 注: 格式对不代表就是对方那把 —— 要等对端发第一条消息验签通过") + except Exception as e: + bad(f"不是合法 base64: {e}") + + 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", "p"): + sys.exit(show_pubkey()) + if arg in ("help", "h"): + print(__doc__) + sys.exit(0) + if arg: + print(f"未知参数 {sys.argv[1]!r}; 可用: (无) / --pubkey / --check / --help") + sys.exit(2) + generate() + + +if __name__ == "__main__": + main()