diff --git a/scripts/hjc_test_push_out.py b/scripts/hjc_test_push_out.py new file mode 100644 index 0000000..3893566 --- /dev/null +++ b/scripts/hjc_test_push_out.py @@ -0,0 +1,657 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +汇吉采 → 一站式:**出向**推送联调探测脚本(POST {base-url}{create-purchase-details-path}) + +与同目录 hjc_test_push.py(入向:一站式 → 官方网 /api/hjc/push/project)互为反向。 + +用途:一条命令回答「出向对接到底通没通、卡在哪一层」。 +脚本会自动做三件事: + 1. 按文档约定的签名方案(sign = MD5(appKey + password + timestamp))发起真实推送; + 2. 额外遍历若干鉴权头变体(含 password 明文头等历史解读),用于区分「签名不对」与「根本没校验签名」; + 3. 用对照探测把失败定位到具体一层:登录过滤器 / 路由不存在 / 验签失败 / 网络不可达。 + +默认参数自动从仓库读取,避免与代码脱节: + - application.yml → hjc.one-stop.base-url、create-purchase-details-path + - HjcOneStopAuthUtil.java → APP_KEY、PASSWORD + +用法:: + + python hjc_test_push_out.py # 全量:约定方案 + 变体矩阵 + 对照探测 + python hjc_test_push_out.py --quick # 只打约定方案(日常回归用) + python hjc_test_push_out.py --repeat 3 # 约定方案连打 3 次(同一订单号,验幂等) + python hjc_test_push_out.py --base http://x:8810 --path /api/biz/createPurchaseDetails + python hjc_test_push_out.py --order-no HJC2026080101ABCD --json result.json + python hjc_test_push_out.py --dry-run # 只打印将要发出的请求,不真的发 + +订单号默认是固定值 ``TEST-HJC-ORDER-20260801``(不随时间变化):它同时是一站式的幂等键, +固定住才能重复验证幂等/状态流转,也不会每跑一次就在对方库里堆一条新订单;要换单号显式传 ``--order-no``。 + +退款报文(status=REFUNDED,必带 refundTime + refundReason,见 docs/一站式平台对接-接口文档.md §3.4):: + + python hjc_test_push_out.py --refund --dry-run # 看退款报文长什么样 + python hjc_test_push_out.py --refund --quick # 打退款报文 + python hjc_test_push_out.py --paid-then-refund # 同一订单号先 PAID 再 REFUNDED(推荐的联调姿势) + python hjc_test_push_out.py --refund --order-no HJC2026080101ABCD \\ + --refund-reason "项目终止,客户申请退款" --refund-time "2026-08-03 09:15:00" + +无第三方依赖,仅标准库。时间统一按北京时间(UTC+8)。 + +退出码:0=出向推送成功(收到 2xx);1=失败(并打印定位结论)。 +""" +from __future__ import annotations + +import argparse +import datetime as dt +import hashlib +import json +import os +import re +import ssl +import sys +import time +import urllib.error +import urllib.parse +import urllib.request + +try: # Windows 控制台默认 GBK,强制 UTF-8 避免中文乱码 + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + sys.stderr.reconfigure(encoding="utf-8", errors="replace") +except Exception: + pass + +# --------------------------------------------------------------------------- # +# 默认值 / 配置读取 +# --------------------------------------------------------------------------- # +HERE = os.path.dirname(os.path.abspath(__file__)) +MP_JAVA = os.path.dirname(HERE) +APPLICATION_YML = os.path.join(MP_JAVA, "src", "main", "resources", "application.yml") +AUTH_UTIL_JAVA = os.path.join( + MP_JAVA, "src", "main", "java", "com", "gxwebsoft", "hjc", "util", "HjcOneStopAuthUtil.java" +) + +FALLBACK_BASE = "http://180.141.88.21:8810" +FALLBACK_PATH = "/api/biz/createPurchaseDetails" +FALLBACK_APP_KEY = "HJC_Official_Website" +FALLBACK_PASSWORD = "vQ8$kR3#mW6@xP2!nF" + +TS_FMT = "%Y%m%d%H%M" +DT_FMT = "%Y-%m-%d %H:%M:%S" +# argparse 的 help 会对 % 做格式化,展示用的时间格式需转义 +DT_FMT_HELP = DT_FMT.replace("%", "%%") + +STATUS_PAID = "PAID" +STATUS_REFUNDED = "REFUNDED" +# 固定探测订单号:不随时间变化,重复运行始终命中一站式同一条记录。 +# 订单号同时是一站式幂等键(HJC_ORD_),固定值才能稳定验证幂等与状态流转, +# 也不会每跑一次就在对方库里堆一条新订单。(与 hjc_test_push.py 固定 projectNo 同一思路) +DEFAULT_ORDER_NO = "PROBE_202609111127" +# 与 HjcBizServiceImpl.DEFAULT_REFUND_REASON 对齐:退款原因留空时官方网会兜底这个值 +DEFAULT_REFUND_REASON = "联调探测:客户申请退款" +# 与 hjc_order.refund_reason 字段长度对齐(varchar(255)) +REFUND_REASON_MAX = 255 + + +def _read_text(path: str) -> str: + try: + with open(path, encoding="utf-8") as f: + return f.read() + except OSError: + return "" + + +def load_config() -> tuple[str, str]: + """从仓库读取 base-url / path,读不到则回退到内置值。""" + base, path = FALLBACK_BASE, FALLBACK_PATH + yml = _read_text(APPLICATION_YML) + if yml: + m = re.search(r'base-url:\s*"([^"]*)"', yml) + if m and m.group(1).strip(): + base = m.group(1).strip() + m = re.search(r"create-purchase-details-path:\s*(\S+)", yml) + if m and m.group(1).strip(): + path = m.group(1).strip().strip('"') + return base, path + + +def load_credentials() -> tuple[str, str]: + """从 HjcOneStopAuthUtil.java 读取 APP_KEY / PASSWORD,读不到则回退。""" + app_key, password = FALLBACK_APP_KEY, FALLBACK_PASSWORD + java = _read_text(AUTH_UTIL_JAVA) + if java: + m = re.search(r'APP_KEY\s*=\s*"([^"]*)"', java) + if m: + app_key = m.group(1) + m = re.search(r'PASSWORD\s*=\s*"([^"]*)"', java) + if m: + password = m.group(1) + return app_key, password + + +# --------------------------------------------------------------------------- # +# 签名 / 时间 +# --------------------------------------------------------------------------- # +def bj_now() -> dt.datetime: + return dt.datetime.now(dt.timezone.utc).astimezone(dt.timezone(dt.timedelta(hours=8))) + + +def timestamp_str(now: dt.datetime) -> str: + return now.strftime(TS_FMT) + + +def sign_of(app_key: str, password: str, ts: str) -> str: + return hashlib.md5((app_key + password + ts).encode("utf-8")).hexdigest() + + +def md5(text: str) -> str: + return hashlib.md5(text.encode("utf-8")).hexdigest() + + +# --------------------------------------------------------------------------- # +# HTTP +# --------------------------------------------------------------------------- # +def post(url: str, body: bytes, headers: dict, timeout: float, insecure: bool = False): + """返回 (http_code, response_text, elapsed_ms, error)。http_code=0 表示网络层失败。""" + req = urllib.request.Request(url, data=body, method="POST", headers=headers) + ctx = ssl._create_unverified_context() if insecure else None + t0 = time.monotonic() + try: + with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp: + text = resp.read().decode("utf-8", errors="replace") + return resp.status, text, (time.monotonic() - t0) * 1000, None + except urllib.error.HTTPError as e: + text = e.read().decode("utf-8", errors="replace") + return e.code, text, (time.monotonic() - t0) * 1000, None + except Exception as e: # noqa: BLE001 - 网络层任何异常都归为不可达 + return 0, "", (time.monotonic() - t0) * 1000, f"{type(e).__name__}: {e}" + + +# --------------------------------------------------------------------------- # +# 响应判定 +# --------------------------------------------------------------------------- # +LOGIN_HINTS = ("token失效", "请重新登录", "未登录", "not login", "unauthorized") +SIGN_HINTS = ("签名", "验签", "sign校验", "invalid signature", "signature") +TS_HINTS = ("时间戳", "timestamp失效", "timestamp expired") + +VERDICT_LABEL = { + "OK": "✅ 通过", + "LOGIN_FILTER": "🔒 登录过滤器拦截", + "SIGN_REJECT": "❌ 验签被拒", + "TS_REJECT": "❌ 时间戳被拒", + "NOT_FOUND": "🚫 路由不存在(404)", + "UNREACHABLE": "🌐 网络不可达", + "OTHER": "❓ 其他", +} + + +def classify(status: int, body: str) -> str: + if status == 0: + return "UNREACHABLE" + if 200 <= status < 300: + return "OK" + low = (body or "").lower() + if any(h in low for h in TS_HINTS): + return "TS_REJECT" + if any(h in low for h in SIGN_HINTS): + return "SIGN_REJECT" + if any(h in low for h in LOGIN_HINTS): + return "LOGIN_FILTER" + if status == 404: + return "NOT_FOUND" + return "OTHER" + + +def brief(body: str, limit: int = 160) -> str: + one = " ".join((body or "").split()) + return one if len(one) <= limit else one[:limit] + "…" + + +def fingerprint(status: int, body: str) -> str: + """响应的「错误语义指纹」:忽略每次都变的 timestamp/path 字段,只保留结论性字段。 + + 用于判断两次响应是否表达同一个错误(例如真实接口与不存在路径都被同一个过滤器拒绝)。 + """ + try: + obj = json.loads(body) + if isinstance(obj, dict): + keep = {k: obj[k] for k in ("status", "error", "message") if k in obj} + if keep: + return json.dumps(keep, ensure_ascii=False, sort_keys=True) + except (ValueError, TypeError): + pass + return " ".join((body or "").split())[:200] or f"HTTP {status}" + + +# --------------------------------------------------------------------------- # +# 报文 +# --------------------------------------------------------------------------- # +def default_payload( + order_no: str, + now: dt.datetime, + status: str = STATUS_PAID, + refund_reason: str | None = None, + refund_time: dt.datetime | None = None, +) -> dict: + """构造 createPurchaseDetails 请求体。 + + status=REFUNDED 时追加 refundTime / refundReason(官方网侧同样只在退款单推这两个字段, + 见 HjcBizServiceImpl#buildCreatePurchaseDetails)。 + """ + payload = { + "idempotencyKey": "HJC_ORD_" + order_no, + "orderNo": order_no, + "projectNo": "HJC-GXZCY2-2026002", + "projectName": "出向推送联调探测", + "tenderPrice": 0.01, + "quantity": 1, + "totalAmount": 0.01, + "buyer": { + "enterpriseName": "联调探测企业", + "creditCode": "91450000PROBE01", + "contactName": "联调", + "contactPhone": "13800000000", + "contactEmail": "probe@example.com", + }, + "paidAt": now.strftime(DT_FMT), + "payMethod": "WECHAT_NATIVE", + "status": status, + "invoiceStatus": "NONE", + } + if status == STATUS_REFUNDED: + reason = (refund_reason or "").strip() or DEFAULT_REFUND_REASON + payload["refundTime"] = (refund_time or now).strftime(DT_FMT) + payload["refundReason"] = reason[:REFUND_REASON_MAX] + return payload + + +def refund_payload_issues(payload: dict) -> list[str]: + """退款报文的本地自检:官方网/一站式都要求这两个字段非空,先在这里拦住低级错误。""" + issues: list[str] = [] + if payload.get("status") != STATUS_REFUNDED: + return issues + for field in ("refundTime", "refundReason"): + value = payload.get(field) + if not isinstance(value, str) or not value.strip(): + issues.append(f"{field} 缺失或为空(退款推送必带)") + rt = payload.get("refundTime") + if isinstance(rt, str) and rt.strip(): + try: + dt.datetime.strptime(rt.strip(), DT_FMT) + except ValueError: + issues.append(f"refundTime 格式应为 {DT_FMT}(当前: {rt})") + reason = payload.get("refundReason") + if isinstance(reason, str) and len(reason) > REFUND_REASON_MAX: + issues.append(f"refundReason 超过 {REFUND_REASON_MAX} 字符(当前: {len(reason)})") + return issues + + +def case_matrix(app_key: str, password: str, ts: str, sign: str) -> list[tuple[str, dict, str]]: + """返回 [(方案名, 请求头, 说明)],方案 1 为文档约定方案。""" + return [ + ( + "1 文档约定方案", + {"appKey": app_key, "timestamp": ts, "sign": sign}, + "appKey+timestamp+sign(sign=MD5(appKey+password+timestamp))", + ), + ( + "2 约定 + password 明文头", + {"appKey": app_key, "timestamp": ts, "sign": sign, "password": password}, + "在原需求「明文 password 头」与现规范之间的折中", + ), + ( + "3 仅 password 明文头", + {"password": password}, + "只带密钥,不带 appKey/sign", + ), + ( + "4 appKey + password 明文", + {"appKey": app_key, "password": password, "timestamp": ts}, + "按原文需求字面实现(无 sign)", + ), + ( + "5 双 MD5 解读", + {"appKey": md5(app_key + ts), "password": md5(password + ts), "timestamp": ts}, + "原文「appKey+年月日时分,password+年月日时分,MD5加密」的另一种读法", + ), + ( + "6 Authorization 头", + {"appKey": app_key, "timestamp": ts, "Authorization": sign}, + "把签名放 Authorization,适配标准网关", + ), + ( + "7 无任何鉴权头", + {}, + "基准:对比有无鉴权头的响应差异", + ), + ] + + +# --------------------------------------------------------------------------- # +# 主流程 +# --------------------------------------------------------------------------- # +def main() -> int: + ap = argparse.ArgumentParser( + description="汇吉采 → 一站式 出向推送联调探测", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + cfg_base, cfg_path = load_config() + cfg_key, cfg_pwd = load_credentials() + + ap.add_argument("--base", default=cfg_base, help=f"一站式 base-url(默认读配置: {cfg_base})") + ap.add_argument("--path", default=cfg_path, help=f"推送路径(默认读配置: {cfg_path})") + ap.add_argument("--appkey", default=cfg_key, help="appKey") + ap.add_argument("--password", default=cfg_pwd, help="签名密钥 password") + ap.add_argument("--order-no", default=DEFAULT_ORDER_NO, help=f"订单号(默认固定值 {DEFAULT_ORDER_NO})") + ap.add_argument("--payload", default=None, help="自定义请求体 JSON 文件") + ap.add_argument( + "--refund", + action="store_true", + help="推退款报文(status=REFUNDED,带 refundTime/refundReason)", + ) + ap.add_argument( + "--paid-then-refund", + action="store_true", + help="同一订单号先推 PAID 基线、再推 REFUNDED(模拟真实退款状态流转)", + ) + ap.add_argument( + "--refund-reason", + default=None, + help=f"退款原因(默认「{DEFAULT_REFUND_REASON}」,最长 {REFUND_REASON_MAX} 字符)", + ) + ap.add_argument( + "--refund-time", + default=None, + help=f"退款时间,格式 {DT_FMT_HELP}(默认当前北京时间)", + ) + ap.add_argument("--repeat", type=int, default=1, help="约定方案重复次数,用于验幂等(默认 1)") + ap.add_argument("--timeout", type=float, default=15.0, help="单请求超时秒数(默认 15)") + ap.add_argument("--quick", action="store_true", help="只跑约定方案,跳过变体矩阵与对照探测") + ap.add_argument("--no-control", action="store_true", help="跳过对照探测(不额外打扰对方服务)") + ap.add_argument("--dry-run", action="store_true", help="只打印请求,不发送") + ap.add_argument("--insecure", action="store_true", help="跳过 HTTPS 证书校验") + ap.add_argument("--json", dest="json_out", default=None, help="把完整结果写入 JSON 文件") + args = ap.parse_args() + + now = bj_now() + ts = timestamp_str(now) + sign = sign_of(args.appkey, args.password, ts) + order_no = args.order_no + url = args.base.rstrip("/") + args.path + + refund_time = None + if args.refund_time: + try: + refund_time = dt.datetime.strptime(args.refund_time.strip(), DT_FMT) + except ValueError: + print(f"❌ --refund-time 格式应为 {DT_FMT},收到:{args.refund_time}", file=sys.stderr) + return 2 + + want_refund = args.refund or args.paid_then_refund + status = STATUS_REFUNDED if want_refund else STATUS_PAID + if not want_refund and not args.payload and (args.refund_reason or args.refund_time): + print( + "⚠️ --refund-reason/--refund-time 仅在 --refund 或 --paid-then-refund 下生效" + "(当前为 PAID 报文)", + file=sys.stderr, + ) + + if args.payload: + if want_refund: + print("⚠️ --payload 已指定,忽略 --refund/--paid-then-refund/--refund-reason/--refund-time", file=sys.stderr) + with open(args.payload, encoding="utf-8") as f: + payload = json.load(f) + else: + payload = default_payload(order_no, now, status, args.refund_reason, refund_time) + + # 退款报文自检:本地先拦掉「必带字段缺失/时间格式错/超长」,避免白跑一轮联调 + issues = refund_payload_issues(payload) + if issues: + print("❌ 退款报文自检未通过:", file=sys.stderr) + for it in issues: + print(f" · {it}", file=sys.stderr) + return 2 + + # --paid-then-refund:同一订单号先落一条 PAID,再发 REFUNDED + paid_body = None + if args.paid_then_refund and not args.payload: + paid_body = json.dumps( + default_payload(order_no, now, STATUS_PAID), ensure_ascii=False + ).encode("utf-8") + + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + + print("=" * 78) + print("汇吉采 → 一站式 出向推送探测" + ("(退款报文)" if payload.get("status") == STATUS_REFUNDED else "")) + print("=" * 78) + print(f"目标 URL : {url}") + print(f"北京时间 : {now.strftime(DT_FMT)} → timestamp={ts}") + print(f"appKey : {args.appkey}") + print(f"sign : {sign}") + print(f"订单号 : {order_no}") + print(f"订单状态 : {payload.get('status')}") + if payload.get("status") == STATUS_REFUNDED: + print(f"退款时间 : {payload.get('refundTime')}") + print(f"退款原因 : {payload.get('refundReason')}") + print(f"请求体 : {body.decode('utf-8')}") + if paid_body is not None: + print() + print("(--paid-then-refund:先发 PAID 基线)") + print(f"请求体 : {paid_body.decode('utf-8')}") + print() + + if args.dry_run: + print("[dry-run] 未发送任何请求。等价 curl:") + if paid_body is not None: + print( + f" # 1) 同一订单号先落 PAID\n" + f" curl -i -X POST '{url}' \\\n" + f" -H 'Content-Type: application/json' \\\n" + f" -H 'appKey: {args.appkey}' -H 'timestamp: {ts}' -H 'sign: {sign}' \\\n" + f" -d '{paid_body.decode('utf-8')}'\n" + f" # 2) 再推 REFUNDED(带退款时间/原因)" + ) + print( + f" curl -i -X POST '{url}' \\\n" + f" -H 'Content-Type: application/json' \\\n" + f" -H 'appKey: {args.appkey}' -H 'timestamp: {ts}' -H 'sign: {sign}' \\\n" + f" -d '{body.decode('utf-8')}'" + ) + return 0 + + results: list[dict] = [] + + def run_case(name: str, headers: dict, note: str, target_url: str = url, use_body: bytes = body): + h = {"Content-Type": "application/json", **headers} + status_code, text, ms, err = post(target_url, use_body, h, args.timeout, args.insecure) + verdict = classify(status_code, text) + results.append( + { + "name": name, + "note": note, + "url": target_url, + "status": status_code, + "verdict": verdict, + "fingerprint": fingerprint(status_code, text), + "elapsed_ms": round(ms, 1), + "body": text, + "error": err, + } + ) + shown = f"HTTP {status_code}" if status_code else "无响应" + print(f" {name:<22} {shown:<9} {VERDICT_LABEL[verdict]:<18} {brief(err or text)}") + return results[-1] + + matrix = case_matrix(args.appkey, args.password, ts, sign) + + # ---------------- 阶段 0:(可选)PAID 基线 ---------------- + # 退款推送要针对「已在一站式存在的订单」,故先用同一订单号落一条 PAID 记录。 + if paid_body is not None: + print("── 阶段 0:PAID 基线(同一订单号,先建立已支付记录)──") + run_case("0 PAID 基线", matrix[0][1], "退款前先落一条已支付记录", use_body=paid_body) + print() + + # ---------------- 阶段 1:约定方案(含 repeat 幂等验证) ---------------- + print(f"── 阶段 1:文档约定方案{'(连打 %d 次验幂等)' % args.repeat if args.repeat > 1 else ''} ──") + primary = run_case(matrix[0][0], matrix[0][1], matrix[0][2]) + for i in range(2, args.repeat + 1): + run_case(f"1 文档约定方案 #{i}", matrix[0][1], "幂等复打") + + if not args.quick: + # ---------------- 阶段 2:鉴权头变体 ---------------- + print() + print("── 阶段 2:鉴权头变体矩阵(用于区分「签名不对」与「未校验签名」)──") + for name, headers, note in matrix[1:]: + run_case(name, headers, note) + + # query 参数形式单独处理(不带请求头) + q = urllib.parse.urlencode( + {"appKey": args.appkey, "password": args.password, "timestamp": ts, "sign": sign} + ) + run_case("8 query 参数", {}, "凭据放 URL query", target_url=f"{url}?{q}") + + # ---------------- 阶段 3:对照探测 ---------------- + # 仅在失败时才有诊断价值;已成功就不再给对方发多余请求。 + controls: list[dict] = [] + if not args.no_control and primary["verdict"] != "OK": + print() + print("── 阶段 3:对照探测(定位失败发生在哪一层)──") + parts = [p for p in args.path.strip("/").split("/") if p] + if len(parts) >= 2: + protected_nope = "/" + "/".join(parts[:-1] + ["__probe_nope__"]) + reference_nope = "/" + "/".join(parts[:1] + ["auth", "__probe_nope__"]) + ctl_headers = matrix[0][1] + for label, p, note in ( + ("C1 同前缀不存在路径", protected_nope, "若与目标接口响应一致 → 拦截发生在路由解析之前"), + ("C2 另一前缀不存在路径", reference_nope, "未受保护的参照组,应为干净的 404"), + ): + controls.append( + run_case(label, ctl_headers, note, target_url=args.base.rstrip("/") + p) + ) + else: + print(" (路径层级不足,跳过对照探测)") + + # ---------------- 结论 ---------------- + print() + print("=" * 78) + print("结论") + print("=" * 78) + + primary_verdict = primary["verdict"] + success = primary_verdict == "OK" + + # 与「同前缀不存在路径」对照:错误语义一致说明根本没走到路由 + same_as_control = bool( + controls + and controls[0]["status"] == primary["status"] + and controls[0]["fingerprint"] == primary["fingerprint"] + ) + + if success: + print("✅ 出向对接成功:目标接口返回 2xx,对方已接受并(应已)落盘。") + if args.repeat > 1: + ok_all = all(r["verdict"] == "OK" for r in results if r["name"].startswith("1 ")) + print( + f" 幂等复打 {args.repeat} 次:{'全部 2xx ✅' if ok_all else '存在非 2xx,请检查幂等处理 ❌'}" + ) + else: + print(f"❌ 出向对接失败:约定方案判定为「{VERDICT_LABEL[primary_verdict]}」。") + print() + if primary_verdict == "LOGIN_FILTER": + print("定位:请求被对方的**登录态过滤器**拒绝,与签名无关。依据:") + if same_as_control: + print(" · 同一套凭据打「同前缀下不存在的路径」,响应与真实接口**完全一致**") + print(" → 拒绝发生在路由解析之前,接口是否存在、签名是否正确都影响不到它。") + print(" · 响应体是登录错误而非「签名校验失败」,说明对方的验签逻辑没有被执行。") + print() + print("需对方处理:把该接口加入免登录白名单并实现约定验签,或提供获取调用 Token 的方式。") + elif primary_verdict == "SIGN_REJECT": + print("定位:对方已放行到验签逻辑,但签名不通过 → 属于**算法/密钥不一致**。") + print("建议:让对方用下面这组固定值离线自证,无需部署:") + print(f" appKey={args.appkey} timestamp={ts} sign={sign}") + print(" 若对方算不出同一 sign,核对 password 取值与拼接顺序(是否含分隔符、大小写)。") + elif primary_verdict == "TS_REJECT": + print("定位:时间戳被拒 → 核对对方服务器时间与北京时区,以及允许窗口(约定 ±10 分钟)。") + elif primary_verdict == "NOT_FOUND": + print("定位:路由不存在 → 路径或 base-url 不对,需对方确认接口实际地址与端口。") + elif primary_verdict == "UNREACHABLE": + print(f"定位:网络不可达 → {primary['error']}") + print("建议:确认对方服务已启动、端口开放、本机到该地址无防火墙/白名单限制。") + else: + print("未能自动归类,请把上面的响应原文提供给对方排查。") + + # 变体是否出现「与基准不同」的响应——若全一致,进一步佐证签名未被校验 + if not args.quick: + variants = [ + r + for r in results + if r["name"][0].isdigit() and not r["name"].startswith(("0 ", "1 ")) + ] + distinct = {r["fingerprint"] for r in variants} + if len(distinct) <= 1 and variants: + print() + print("补充:全部鉴权头变体(含不带任何头的基准)响应**完全相同**") + print(" → 对方当前没有对请求头做任何区分,签名校验尚未生效。") + + # 退款模式补充结论:PAID 基线 + REFUNDED 是否都通过(退款字段是否被对方接受) + if payload.get("status") == STATUS_REFUNDED: + print() + paid_base = next((r for r in results if r["name"].startswith("0 ")), None) + if paid_base is not None: + print( + f"PAID 基线 : {VERDICT_LABEL[paid_base['verdict']]}(同一订单号 {order_no})" + + ("" if paid_base["verdict"] == "OK" else " ← 基线未通过,退款结果可能不可信") + ) + print( + f"REFUNDED 报文: {VERDICT_LABEL[primary_verdict]}," + f"refundTime={payload.get('refundTime')} / refundReason={payload.get('refundReason')}" + ) + + print() + print(f"复现命令(可直接发给对方):") + if paid_body is not None: + print( + f" # 1) 同一订单号先落 PAID\n" + f" curl -i -X POST '{url}' -H 'Content-Type: application/json' \\\n" + f" -H 'appKey: {args.appkey}' -H 'timestamp: {ts}' -H 'sign: {sign}' \\\n" + f" -d '{paid_body.decode('utf-8')}'\n" + f" # 2) 再推 REFUNDED(带退款时间/原因)" + ) + print( + f" curl -i -X POST '{url}' \\\n" + f" -H 'Content-Type: application/json' \\\n" + f" -H 'appKey: {args.appkey}' -H 'timestamp: {ts}' -H 'sign: {sign}' \\\n" + f" -d '{body.decode('utf-8')}'" + ) + + if args.json_out: + with open(args.json_out, "w", encoding="utf-8") as f: + json.dump( + { + "base": args.base, + "path": args.path, + "url": url, + "timestamp": ts, + "sign": sign, + "appKey": args.appkey, + "orderNo": order_no, + "orderStatus": payload.get("status"), + "refundTime": payload.get("refundTime"), + "refundReason": payload.get("refundReason"), + "paidThenRefund": args.paid_then_refund, + "payload": payload, + "results": results, + "controls": controls, + "verdict": primary_verdict, + "success": success, + }, + f, + ensure_ascii=False, + indent=2, + ) + print(f"\n完整结果已写入:{args.json_out}") + + return 0 if success else 1 + + +if __name__ == "__main__": + sys.exit(main())