174 lines
6.9 KiB
Python
174 lines
6.9 KiB
Python
# -*- 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()
|