"""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, sign_path: str | None = None, skip_access_key: bool = False, access_key_override: str | None = None, ) -> tuple[int, dict | str, dict]: """返回 (http_status, parsed_body, sent_headers)。sent_headers 中签名已遮蔽。 sign_path：签名串路径与请求 URL 路径不同时使用（查询串不得进入签名， 对齐 Ai-DOP InboundSignatureHandler 用 Request.Path.Value 验签）。 skip_access_key：不发送 X-Access-Key（N3 缺 AccessKey）。 access_key_override：用指定值替代 AccessKey（N3 错误 AccessKey，签名自洽但未知）。""" 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}" sent_key = access_key_override or self.key message = build_message(method, sign_path or path, sent_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-Timestamp": ts_s, "X-Nonce": nonce, "X-Content-SHA256": digest, "X-Signature": signature, } if not skip_access_key: headers["X-Access-Key"] = sent_key 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): sign_path = f"/api/mdp/inbound/{entity}/snapshots/{urllib.parse.quote(snapshot_id)}/commit" path = sign_path + ("?force=true" if force else "") return self.request("POST", path, b"", sign_path=sign_path) def op_digest(self, entity: str, biz_date: str): sign_path = f"/api/mdp/inbound/{entity}/digest" path = sign_path + f"?date={urllib.parse.quote(biz_date)}" return self.request("GET", path, b"", sign_path=sign_path) # -- 必测用例（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) # -- 第二阶段 N3 新增用例 ------------------------------------------------ def case_missing_access_key(self, entity: str, envelope: dict): """缺 AccessKey：不发送 X-Access-Key 头。""" return self.op_push(entity, envelope, skip_access_key=True) def case_wrong_access_key(self, entity: str, envelope: dict): """错误 AccessKey：签名自洽但 AccessKey 未授权。""" return self.op_push(entity, envelope, access_key_override="SIM-UNKNOWN-ACCESS-KEY") def case_missing_required_field(self, entity: str, envelope: dict, field: str): """缺必填字段：逐行删除指定契约字段。""" env = json.loads(json.dumps(envelope, ensure_ascii=False)) for row in env["data"]["list"]: row.pop(field, None) return self.op_push(entity, env) def case_wrong_field_type(self, entity: str, envelope: dict, field: str, bad_value): """字段类型错误：把指定字段替换为错误类型值（如数值列塞字符串）。""" env = json.loads(json.dumps(envelope, ensure_ascii=False)) for row in env["data"]["list"]: row[field] = bad_value return self.op_push(entity, env) def case_stale_rows(self, entity: str, envelope: dict): """stale 数据：逐行把时间类字段退到 2000 年（快照 commit 时被 Ai-DOP 拒绝）。""" env = json.loads(json.dumps(envelope, ensure_ascii=False)) for row in env["data"]["list"]: for key in tuple(row): if str(key).lower() in ("sourceupdatedat", "updatedat", "bizdate", "occurredat"): row[key] = "2000-01-01T00:00:00" return self.op_push(entity, env) def case_unknown_entity(self, envelope: dict): """未授权/未注册 entityCode：推送到一个不存在于目录的实体码。""" return self.op_push("SIM_UNKNOWN_ENTITY_XX", envelope) def case_bulk_partial_failure(self, entity: str, ndjson_lines: list[dict], field: str, bad_idx: int = -1): """bulk 部分失败：指定行删除必填字段，其余行完好。""" lines = [json.loads(json.dumps(line, ensure_ascii=False)) for line in ndjson_lines] lines[bad_idx].pop(field, None) return self.op_bulk(entity, lines) def case_snapshot_commit_failure(self, entity: str): """snapshot commit 失败：提交一个不存在的 snapshotId。""" return self.op_snapshot_commit(entity, "SNAP-NOT-EXIST-0001")