|
|
@@ -0,0 +1,176 @@
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+"""乐收赢测试环境 **主扫下单** 探针：向上游发一次真实的 /v3/qrpay，拿 code_url。
|
|
|
+
|
|
|
+为什么要打这一发（前面已经验过签名了）：
|
|
|
+ - 之前那条只读 /v3/queryorder 证明了「签名规则 + 信封」对；
|
|
|
+ - 但**下单**才走完整的业务校验（mch_no 是否有效、shop_no 要不要、pay_trace_no 唯一性、
|
|
|
+ total_amount 形态、time_expire 取值）——这些都是「签名对」证明不了的。
|
|
|
+ - 2026-09-28 用户已与上游确认：文档里的示例参数可用于测试。
|
|
|
+
|
|
|
+★ 与生产代码的关系（重要，别把本探针当生产量具）：
|
|
|
+ 报文**逐字段抄自** `LsyOrderService.createQrOrder`（主扫那条腿）在 2026-09-28 的读法：
|
|
|
+ mch_no / [shop_no] / pay_trace_no / pay_time / total_amount / time_expire
|
|
|
+ / [order_body] / [attach] / [notify_url] + inst_no + sign（由 client 自动补）
|
|
|
+ 本脚本会把**实际发出去的 body** 原样打出来，好让人把它与上面那份字段清单逐格对
|
|
|
+ —— 转写件必须能被核对，否则它只是一段漂亮的猜测。
|
|
|
+ 但本探针仍是**另一支量具**：它证明的是「这个字段集被上游接受」，不是「Java 客户端发出的字节被接受」。
|
|
|
+
|
|
|
+★ 不传 notify_url：上游文档说「不传取机构默认配置回调」。传了就等于让上游的测试环境
|
|
|
+ 去回调我们的地址，那是一个**有副作用**的动作，本次不需要。
|
|
|
+
|
|
|
+★ 凭据（GC3/GC5）：机构密钥只从 application.yml 就地读，只打印长度与来源，绝不落文件、不上命令行。
|
|
|
+★ 出站 body 里的 sign 打码（它是这一笔的鉴权件）——字段形状仍然完整可见。
|
|
|
+"""
|
|
|
+import json
|
|
|
+import os
|
|
|
+import re
|
|
|
+import sys
|
|
|
+import time
|
|
|
+import urllib.request
|
|
|
+import urllib.error
|
|
|
+from hashlib import md5
|
|
|
+
|
|
|
+YML = "java/src/main/resources/application.yml"
|
|
|
+VAR = re.compile(r"^\$\{([A-Za-z0-9_]+):(.*)\}$")
|
|
|
+OUT = "logs/_lsy-qrpay-probe.txt"
|
|
|
+RESP_BIN = "logs/_lsy-qrpay-resp.bin"
|
|
|
+
|
|
|
+buf = []
|
|
|
+
|
|
|
+
|
|
|
+def say(s=""):
|
|
|
+ buf.append(s)
|
|
|
+ print(s.encode("ascii", "replace").decode())
|
|
|
+
|
|
|
+
|
|
|
+def read_block(path, top):
|
|
|
+ lines = open(path, encoding="utf-8", errors="replace").read().split("\n")
|
|
|
+ start = None
|
|
|
+ for i, ln in enumerate(lines):
|
|
|
+ if re.match(r"^%s:\s*$" % re.escape(top), ln):
|
|
|
+ start = i
|
|
|
+ break
|
|
|
+ if start is None:
|
|
|
+ raise SystemExit("找不到顶层段 %s:" % top)
|
|
|
+ out = {}
|
|
|
+ for j in range(start + 1, len(lines)):
|
|
|
+ ln = lines[j]
|
|
|
+ if ln.strip() == "" or ln.lstrip().startswith("#"):
|
|
|
+ continue
|
|
|
+ if not ln.startswith((" ", "\t")):
|
|
|
+ break
|
|
|
+ m = re.match(r"^\s+([A-Za-z0-9_.\-]+):\s*(.*)$", ln)
|
|
|
+ if m:
|
|
|
+ v = re.split(r"\s+#", m.group(2), maxsplit=1)[0].strip().strip("'\"")
|
|
|
+ out[m.group(1)] = v
|
|
|
+ return out
|
|
|
+
|
|
|
+
|
|
|
+def resolve(raw):
|
|
|
+ m = VAR.match(raw)
|
|
|
+ if not m:
|
|
|
+ return raw, "application.yml 字面量"
|
|
|
+ var, default = m.group(1), m.group(2)
|
|
|
+ if os.environ.get(var):
|
|
|
+ return os.environ[var], "环境变量 %s" % var
|
|
|
+ return default, "application.yml 默认值（键 %s）" % var
|
|
|
+
|
|
|
+
|
|
|
+cfg = read_block(YML, "leshouying")
|
|
|
+base_url, base_src = resolve(cfg["base-url"])
|
|
|
+inst_no, inst_src = resolve(cfg["inst-no"])
|
|
|
+key, key_src = resolve(cfg["key"])
|
|
|
+expire_min = resolve(cfg.get("default-qr-expire-minutes", "15"))[0]
|
|
|
+
|
|
|
+say("── 取值来源（不含密钥值）──")
|
|
|
+say(" base-url = %s [%s]" % (base_url, base_src))
|
|
|
+say(" inst-no = %s [%s]" % (inst_no, inst_src))
|
|
|
+say(" key = <len=%d> [%s]" % (len(key), key_src))
|
|
|
+say(" default-qr-expire-minutes= %s [取自 application.yml]" % expire_min)
|
|
|
+say()
|
|
|
+
|
|
|
+# ── 组报文：字段集逐格抄自 LsyOrderService.createQrOrder ─────────────────
|
|
|
+now = time.strftime("%Y%m%d%H%M%S")
|
|
|
+trace = "probe" + now + "qr"
|
|
|
+payload = {
|
|
|
+ "inst_no": inst_no, # client 的 signedPayload 自动补
|
|
|
+ "mch_no": "Y52211090004", # 文档测试商户号
|
|
|
+ "pay_trace_no": trace, # 上游要求商户系统内唯一；本探针每跑一次都不同
|
|
|
+ "pay_time": now, # yyyyMMddHHmmss
|
|
|
+ "total_amount": "1", # 1 分
|
|
|
+ "time_expire": str(expire_min),
|
|
|
+ "order_body": "联调探针-主扫",
|
|
|
+ # shop_no / attach / notify_url 生产侧都是「非 null 才发」，此处留空不传
|
|
|
+}
|
|
|
+
|
|
|
+A = "&".join("%s=%s" % (k, payload[k]) for k in sorted(payload))
|
|
|
+sign = md5((A + "&key=" + key).encode("utf-8")).hexdigest().lower()
|
|
|
+payload["sign"] = sign
|
|
|
+
|
|
|
+say("── 将发送（不含密钥；sign 打码）──")
|
|
|
+say(" A = %s" % A)
|
|
|
+say()
|
|
|
+body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
|
+shown = json.loads(body.decode("utf-8"))
|
|
|
+shown["sign"] = "<len=%d>" % len(sign)
|
|
|
+say(" body = %s" % json.dumps(shown, ensure_ascii=False, indent=2))
|
|
|
+say()
|
|
|
+say(" 字段清单（= 生产 createQrOrder 发的集合）：%s" % ", ".join(sorted(payload)))
|
|
|
+say()
|
|
|
+
|
|
|
+url = base_url.rstrip("/") + "/v3/qrpay"
|
|
|
+say("── 出站 ──")
|
|
|
+say(" POST %s" % url)
|
|
|
+say(" Content-Type: application/json;charset=utf-8")
|
|
|
+say(" bytes = %d" % len(body))
|
|
|
+say()
|
|
|
+
|
|
|
+req = urllib.request.Request(
|
|
|
+ url, data=body,
|
|
|
+ headers={"Content-Type": "application/json;charset=utf-8"},
|
|
|
+ method="POST",
|
|
|
+)
|
|
|
+try:
|
|
|
+ with urllib.request.urlopen(req, timeout=30) as r:
|
|
|
+ code, raw = r.status, r.read()
|
|
|
+except urllib.error.HTTPError as e:
|
|
|
+ code, raw = e.code, e.read()
|
|
|
+except Exception as e:
|
|
|
+ say("!! 请求未完成: %r" % (e,))
|
|
|
+ open(OUT, "w", encoding="utf-8").write("\n".join(buf))
|
|
|
+ sys.exit(1)
|
|
|
+
|
|
|
+open(RESP_BIN, "wb").write(raw)
|
|
|
+say("── 上游应答 ──")
|
|
|
+say(" http_code = %s" % code)
|
|
|
+say(" bytes = %d" % len(raw))
|
|
|
+say(" 原始字节已落 %s" % RESP_BIN)
|
|
|
+say()
|
|
|
+
|
|
|
+try:
|
|
|
+ resp = json.loads(raw.decode("utf-8"))
|
|
|
+ say(" 解码 = utf-8 成功")
|
|
|
+except Exception as e:
|
|
|
+ say(" utf-8 解码失败: %r" % (e,))
|
|
|
+ open(OUT, "w", encoding="utf-8").write("\n".join(buf))
|
|
|
+ sys.exit(1)
|
|
|
+
|
|
|
+say(" return_code = %s" % resp.get("return_code"))
|
|
|
+say(" return_msg = %s" % resp.get("return_msg"))
|
|
|
+say(" result_code = %s" % resp.get("result_code"))
|
|
|
+say(" code_url = %s" % resp.get("code_url"))
|
|
|
+say()
|
|
|
+
|
|
|
+# ── 入站验签：用上一条已经实证过的「null 不参与」读法 ─────────────────
|
|
|
+rkey = resp.get("sign")
|
|
|
+if rkey:
|
|
|
+ nonnull = {k: v for k, v in resp.items() if k != "sign" and v is not None}
|
|
|
+ A2 = "&".join("%s=%s" % (k, nonnull[k]) for k in sorted(nonnull))
|
|
|
+ mine = md5((A2 + "&key=" + key).encode("utf-8")).hexdigest().lower()
|
|
|
+ say("── 入站验签（null 不参与）──")
|
|
|
+ say(" 参与签名的字段数 = %d / 应答字段数 %d" % (len(nonnull), len(resp) - 1))
|
|
|
+ say(" A = %s" % A2)
|
|
|
+ say(" 重算 == 上游 sign : %s" % (mine == rkey))
|
|
|
+ say()
|
|
|
+
|
|
|
+open(OUT, "w", encoding="utf-8").write("\n".join(buf))
|
alphaH