AiDOPWarehouse/tools/integration-simulator/clients/inbound_client.py

"""API_INBOUND 签名客户端：与 MdpInboundController / InboundSignatureHandler 契约一致。

签名串（方案 §5.2）：
    message = METHOD&path&accessKey&timestamp&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,
    ) -> tuple[int, dict | str, dict]:
        """返回 (http_status, parsed_body, sent_headers)。sent_headers 中签名已遮蔽。

        sign_path：签名串路径与请求 URL 路径不同时使用（查询串不得进入签名，
        对齐 Ai-DOP InboundSignatureHandler 用 Request.Path.Value 验签）。"""
        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, sign_path or 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):
        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)