weicw1996
fdff54a6cf
与已入库的 scripts/hjc_test_push.py（入向：一站式 → 官网 /api/hjc/push/project）互为反向， 一条命令回答「出向对接通没通、卡在哪一层」： - 按文档约定的签名方案（sign = MD5(appKey + password + timestamp)）发起真实推送； - 额外遍历若干鉴权头变体，用于区分「签名不对」与「根本没校验签名」； - 用对照探测把失败定位到具体一层：登录过滤器 / 路由不存在 / 验签失败 / 网络不可达。 默认参数直接从仓库的 application.yml 读取，避免脚本与配置脱节。
658 lines
29 KiB
Python
658 lines
29 KiB
Python
#!/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_<orderNo>），固定值才能稳定验证幂等与状态流转，
|
||
# 也不会每跑一次就在对方库里堆一条新订单。（与 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())
|