"""API_INBOUND 签名客户端:与 MdpInboundController / InboundSignatureHandler 契约一致。 签名串(方案 §5.2): message = METHOD&path&accessKey×tamp&nonce&bodySha256&idempotencyKey signature = Base64(HMAC-SHA256(secret, message)) Secret 只驻留本进程内存,不出现在返回值、日志与页面(脱敏由 redaction 模块负责)。 七类操作:schema / push / receipt / bulk / snapshot open / snapshot commit / digest。 测试用例语义与 tools/mock/inbound/push_demo.py 对齐(复用协议、不迁移文件)。 """ from __future__ import annotations import hashlib import hmac import json import time import urllib.error import urllib.parse import urllib.request import uuid from base64 import b64encode CONTRACT_VERSION_HEADER = "X-Mdp-Contract-Version" 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 build_message(method: str, path: str, access_key: str, timestamp: str, nonce: str, body_sha256: str, idempotency_key: str) -> str: return f"{method.upper()}&{path}&{access_key}&{timestamp}&{nonce}&{body_sha256}&{idempotency_key}" def _parse(text: str): try: return json.loads(text) if text else {} except json.JSONDecodeError: return text class InboundClient: def __init__(self, base_url: str, access_key: str, secret: str, timeout: int = 30): self.base = base_url.rstrip("/") self.key = access_key self._secret = secret self.timeout = timeout # -- 底层签名请求 --------------------------------------------------------- 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, bad_signature: bool = False, contract_version: str | None = None, extra_headers: dict | None = None, ) -> tuple[int, dict | str, dict]: """返回 (http_status, parsed_body, sent_headers)。sent_headers 中签名已遮蔽。""" 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 = build_message(method, path, self.key, ts_s, nonce, digest, idem) signature = sign(self._secret, message) if bad_signature: signature = sign(self._secret + "-wrong", message) headers = { "Content-Type": "application/json", "X-Access-Key": self.key, "X-Timestamp": ts_s, "X-Nonce": nonce, "X-Content-SHA256": digest, "X-Signature": signature, } if not skip_idem: headers["Idempotency-Key"] = idem if contract_version: headers[CONTRACT_VERSION_HEADER] = contract_version if extra_headers: headers.update(extra_headers) data = raw if method.upper() != "GET" else None req = urllib.request.Request(self.base + path, data=data, method=method.upper(), headers=headers) redacted = dict(headers) redacted["X-Signature"] = "***REDACTED***" try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: return resp.status, _parse(resp.read().decode("utf-8")), redacted except urllib.error.HTTPError as ex: return ex.code, _parse(ex.read().decode("utf-8")), redacted except urllib.error.URLError as ex: return 0, {"message": f"SOURCE_UNREACHABLE: {ex.reason}"}, redacted except TimeoutError: return 0, {"message": "SOURCE_UNREACHABLE: timeout"}, redacted # -- 七类操作 ------------------------------------------------------------- def op_schema(self, entity: str, contract_version: str | None = None): path = f"/api/mdp/inbound/{entity}/schema" return self.request("GET", path, b"", contract_version=contract_version) def op_push(self, entity: str, envelope: dict, **kwargs): path = f"/api/mdp/inbound/{entity}" body = json.dumps(envelope, ensure_ascii=False).encode("utf-8") return self.request("POST", path, body, **kwargs) def op_receipt(self, sync_batch_id: str): path = f"/api/mdp/inbound/receipts/{urllib.parse.quote(sync_batch_id)}" return self.request("GET", path, b"") def op_bulk(self, entity: str, ndjson_lines: list[dict], **kwargs): path = f"/api/mdp/inbound/{entity}/bulk" body = ("\n".join(json.dumps(line, ensure_ascii=False) for line in ndjson_lines)).encode("utf-8") headers = {"Content-Type": "application/x-ndjson"} extra = dict(kwargs.pop("extra_headers", None) or {}) extra.update(headers) return self.request("POST", path, body, extra_headers=extra, **kwargs) def op_snapshot_open(self, entity: str): path = f"/api/mdp/inbound/{entity}/snapshots" return self.request("POST", path, b"") def op_snapshot_commit(self, entity: str, snapshot_id: str, force: bool = False): path = f"/api/mdp/inbound/{entity}/snapshots/{urllib.parse.quote(snapshot_id)}/commit" if force: path += "?force=true" return self.request("POST", path, b"") def op_digest(self, entity: str, biz_date: str): path = f"/api/mdp/inbound/{entity}/digest?date={urllib.parse.quote(biz_date)}" return self.request("GET", path, b"") # -- 必测用例(P1-E)------------------------------------------------------- def case_expired_ts(self, entity: str, envelope: dict): return self.op_push(entity, envelope, ts=int(time.time()) - 3600) def case_nonce_replay(self, entity: str, envelope: dict): body = json.dumps(envelope, ensure_ascii=False).encode("utf-8") path = f"/api/mdp/inbound/{entity}" nonce = uuid.uuid4().hex s1, d1, _ = self.request("POST", path, body, nonce=nonce) s2, d2, _ = self.request("POST", path, body, nonce=nonce, idem=f"IDEM-{uuid.uuid4().hex}") return (s1, d1), (s2, d2) def case_same_key_replay(self, entity: str, envelope: dict): body = json.dumps(envelope, ensure_ascii=False).encode("utf-8") path = f"/api/mdp/inbound/{entity}" idem = f"IDEM-REPLAY-{uuid.uuid4().hex}" s1, d1, _ = self.request("POST", path, body, idem=idem) s2, d2, _ = self.request("POST", path, body, idem=idem) return (s1, d1), (s2, d2) def case_same_key_diff_body(self, entity: str, env1: dict, env2: dict): path = f"/api/mdp/inbound/{entity}" idem = f"IDEM-DIFF-{uuid.uuid4().hex}" b1 = json.dumps(env1, ensure_ascii=False).encode("utf-8") b2 = json.dumps(env2, ensure_ascii=False).encode("utf-8") s1, d1, _ = self.request("POST", path, b1, idem=idem) s2, d2, _ = self.request("POST", path, b2, idem=idem) return (s1, d1), (s2, d2) def case_missing_idem(self, entity: str, envelope: dict): return self.op_push(entity, envelope, skip_idem=True) def case_tamper_body(self, entity: str, envelope: dict): return self.op_push(entity, envelope, tamper_body=True) def case_bad_signature(self, entity: str, envelope: dict): return self.op_push(entity, envelope, bad_signature=True)