#!/usr/bin/env python3 """API_INBOUND 联调用例。签名按方案 §5.2：METHOD&path&key&ts&nonce&sha256&idem。 密钥只从 --secret 或环境变量 AIDOP_INBOUND_SECRET 读取，不写死生产 secret。 联调种子 TESTKPARTNER 的公开占位口令可用 --use-seed-secret（仅开发库）。 """ from __future__ import annotations import argparse import hashlib import hmac import json import os import sys import time import urllib.error import urllib.request import uuid from base64 import b64encode from datetime import datetime, timezone, timedelta # 公开联调种子，不是生产密钥 _SEED_SECRET = "TESTKPARTNER_SECRET_DO_NOT_USE_IN_PROD" def sha256_hex(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def sign(secret: str, message: str) -> str: digest = hmac.new(secret.encode("utf-8"), message.encode("utf-8"), hashlib.sha256).digest() return b64encode(digest).decode("ascii") def now_iso() -> str: return datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%dT%H:%M:%S+08:00") def sales_row(**extra) -> dict: row = { "bill_no": extra.pop("bill_no", "SO-INB-DEMO-001"), "entry_seq": extra.pop("entry_seq", 1), "seorder_id": extra.pop("seorder_id", 900001), "qty": extra.pop("qty", 10), "item_number": "ITEM-INB-001", "sourceUpdatedAt": extra.pop("sourceUpdatedAt", now_iso()), } row.update(extra) return row def envelope(rows: list, snapshot_id: str | None = None, seq: int | None = None) -> dict: data: dict = {"list": rows} if snapshot_id: data["snapshotId"] = snapshot_id if seq is not None: data["seq"] = seq return {"data": data} class InboundClient: def __init__(self, base_url: str, access_key: str, secret: str): self.base = base_url.rstrip("/") self.key = access_key self.secret = secret def request( self, method: str, path: str, body: bytes | None = None, *, idem: str | None = None, ts: int | None = None, nonce: str | None = None, skip_idem: bool = False, tamper_body: bool = False, extra_headers: dict | None = None, ) -> tuple[int, dict | str]: raw = body if body is not None else b"" digest = sha256_hex(raw) if tamper_body: raw = raw + b"x" ts_s = str(ts if ts is not None else int(time.time())) nonce = nonce or uuid.uuid4().hex idem = idem or f"IDEM-{uuid.uuid4().hex}" message = f"{method.upper()}&{path}&{self.key}&{ts_s}&{nonce}&{digest}&{idem}" headers = { "Content-Type": "application/json", "X-Access-Key": self.key, "X-Timestamp": ts_s, "X-Nonce": nonce, "X-Content-SHA256": digest, "X-Signature": sign(self.secret, message), } if not skip_idem: headers["Idempotency-Key"] = idem if extra_headers: headers.update(extra_headers) req = urllib.request.Request(self.base + path, data=raw if method != "GET" else None, method=method, headers=headers) try: with urllib.request.urlopen(req, timeout=30) as resp: text = resp.read().decode("utf-8") return resp.status, _parse(text) except urllib.error.HTTPError as ex: text = ex.read().decode("utf-8") return ex.code, _parse(text) except urllib.error.URLError as ex: return 0, {"message": str(ex.reason)} def _parse(text: str): try: return json.loads(text) if text else {} except json.JSONDecodeError: return text def report(name: str, ok: bool, detail) -> bool: mark = "PASS" if ok else "FAIL" print(f"[{mark}] {name}: {detail}") return ok def case_happy(c: InboundClient, entity: str) -> bool: path = f"/api/mdp/inbound/{entity}" body = json.dumps(envelope([sales_row()]), ensure_ascii=False).encode("utf-8") status, data = c.request("POST", path, body) return report("happy", status == 202 and isinstance(data, dict) and data.get("code") in (0, 2), (status, data)) def case_missing_idem(c: InboundClient, entity: str) -> bool: path = f"/api/mdp/inbound/{entity}" body = json.dumps(envelope([sales_row()]), ensure_ascii=False).encode("utf-8") status, data = c.request("POST", path, body, skip_idem=True) msg = data.get("message") if isinstance(data, dict) else data return report("missing Idempotency-Key", status == 400 and "Idempotency-Key" in str(msg), (status, data)) def case_tamper_body(c: InboundClient, entity: str) -> bool: path = f"/api/mdp/inbound/{entity}" body = json.dumps(envelope([sales_row()]), ensure_ascii=False).encode("utf-8") status, data = c.request("POST", path, body, tamper_body=True) msg = data.get("message") if isinstance(data, dict) else data return report("body+1 byte", status == 400 and "digest" in str(msg).lower(), (status, data)) def case_expired_ts(c: InboundClient, entity: str) -> bool: path = f"/api/mdp/inbound/{entity}" body = json.dumps(envelope([sales_row()]), ensure_ascii=False).encode("utf-8") status, data = c.request("POST", path, body, ts=int(time.time()) - 3600) return report("expired timestamp", status == 401, (status, data)) def case_nonce_replay(c: InboundClient, entity: str) -> bool: path = f"/api/mdp/inbound/{entity}" body = json.dumps(envelope([sales_row(bill_no="SO-NONCE-1")]), ensure_ascii=False).encode("utf-8") nonce = uuid.uuid4().hex status1, _ = c.request("POST", path, body, nonce=nonce) status2, data2 = c.request("POST", path, body, nonce=nonce, idem=f"IDEM-{uuid.uuid4().hex}") return report("nonce replay", status1 == 202 and status2 == 401, (status1, status2, data2)) def case_same_key_replay(c: InboundClient, entity: str) -> bool: path = f"/api/mdp/inbound/{entity}" body = json.dumps(envelope([sales_row(bill_no="SO-REPLAY-1")]), ensure_ascii=False).encode("utf-8") idem = f"IDEM-REPLAY-{uuid.uuid4().hex}" status1, data1 = c.request("POST", path, body, idem=idem) status2, data2 = c.request("POST", path, body, idem=idem) replay = isinstance(data2, dict) and (data2.get("data") or {}).get("idempotentReplay") is True return report("same-key replay", status1 == 202 and status2 == 202 and replay, (status1, status2, data2)) def case_same_key_diff_body(c: InboundClient, entity: str) -> bool: path = f"/api/mdp/inbound/{entity}" idem = f"IDEM-DIFF-{uuid.uuid4().hex}" b1 = json.dumps(envelope([sales_row(bill_no="SO-DIFF-1")]), ensure_ascii=False).encode("utf-8") b2 = json.dumps(envelope([sales_row(bill_no="SO-DIFF-2")]), ensure_ascii=False).encode("utf-8") status1, _ = c.request("POST", path, b1, idem=idem) status2, data2 = c.request("POST", path, b2, idem=idem) return report("same-key different body", status1 == 202 and status2 == 409, (status1, status2, data2)) def case_partial(c: InboundClient, entity: str) -> bool: path = f"/api/mdp/inbound/{entity}" rows = [sales_row(bill_no="SO-PART-OK"), {"qty": 1, "sourceUpdatedAt": now_iso()}] body = json.dumps(envelope(rows), ensure_ascii=False).encode("utf-8") status, data = c.request("POST", path, body) code = data.get("code") if isinstance(data, dict) else None return report("partial", status == 202 and code == 2, (status, data)) def case_stale(c: InboundClient, entity: str) -> bool: path = f"/api/mdp/inbound/{entity}" newer = "2026-09-11T12:00:00+08:00" older = "2026-09-01T12:00:00+08:00" b1 = json.dumps(envelope([sales_row(bill_no="SO-STALE-1", sourceUpdatedAt=newer)]), ensure_ascii=False).encode("utf-8") b2 = json.dumps(envelope([sales_row(bill_no="SO-STALE-1", sourceUpdatedAt=older)]), ensure_ascii=False).encode("utf-8") status1, _ = c.request("POST", path, b1) status2, data2 = c.request("POST", path, b2) rejected = ((data2.get("data") or {}).get("staleRejected") if isinstance(data2, dict) else 0) or 0 return report("stale", status1 == 202 and status2 == 202 and rejected >= 1, (status1, status2, data2)) def case_snapshot(c: InboundClient, entity: str) -> bool: open_path = f"/api/mdp/inbound/{entity}/snapshots" status, data = c.request("POST", open_path, b"") snap = (data.get("data") or {}).get("snapshotId") if isinstance(data, dict) else None if status not in (200, 201) or not snap: return report("snapshot open", False, (status, data)) push = f"/api/mdp/inbound/{entity}" b1 = json.dumps(envelope([sales_row(bill_no="SO-SNAP-1")], snap, 1), ensure_ascii=False).encode("utf-8") b2 = json.dumps(envelope([sales_row(bill_no="SO-SNAP-2")], snap, 2), ensure_ascii=False).encode("utf-8") s1, _ = c.request("POST", push, b1) s2, _ = c.request("POST", push, b2) commit_path = f"/api/mdp/inbound/{entity}/snapshots/{snap}/commit" sc, dc = c.request("POST", commit_path, b"") committed = isinstance(dc, dict) and ((dc.get("data") or {}).get("status") == "COMMITTED" or dc.get("code") == 0) return report("snapshot open→2 batches→commit", s1 == 202 and s2 == 202 and sc in (200, 202) and committed, (s1, s2, sc, dc)) CASES = { "happy": case_happy, "missing-idem": case_missing_idem, "tamper-body": case_tamper_body, "expired-ts": case_expired_ts, "nonce-replay": case_nonce_replay, "same-key-replay": case_same_key_replay, "same-key-diff-body": case_same_key_diff_body, "partial": case_partial, "stale": case_stale, "snapshot": case_snapshot, } def main() -> int: p = argparse.ArgumentParser(description="API_INBOUND mock push") p.add_argument("--base-url", default="http://127.0.0.1:5005") p.add_argument("--access-key", default="TESTKPARTNER") p.add_argument("--secret", default=os.environ.get("AIDOP_INBOUND_SECRET", "")) p.add_argument("--use-seed-secret", action="store_true", help="使用公开联调种子口令（非生产）") p.add_argument("--entity", default="S1_SALES_ORDER_ENTRY") p.add_argument("--case", default="all", choices=["all", *CASES]) args = p.parse_args() secret = args.secret if args.use_seed_secret: secret = _SEED_SECRET if not secret: print("missing --secret or AIDOP_INBOUND_SECRET（开发库可加 --use-seed-secret）", file=sys.stderr) return 2 client = InboundClient(args.base_url, args.access_key, secret) names = list(CASES) if args.case == "all" else [args.case] ok = True for name in names: ok = CASES[name](client, args.entity) and ok return 0 if ok else 1 if __name__ == "__main__": raise SystemExit(main())