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

"""Ai-DOP 只读预检（任务书 P1-G）。

只做只读探测，不写任何 Ai-DOP 表：
- 可达性：GET {base}/api/mdp/inbound/__probe_not_exists（404/401 均视为服务在线）；
- 实体预检：用真实签名调 GET schema，按 HTTP 状态与消息分层判定
  READY / CONFIG_NOT_REGISTERED / CONFIG_NOT_ENABLED / GRANT_MISSING / SOURCE_UNREACHABLE；
- 管理 API 预检（可选）：仅当 AIDOP_SIM_ADMIN_TOKEN 提供时查询来源/实体配置，全部只读。

目标 host 必须命中白名单，否则硬阻断（默认拒绝非白名单目标）。
"""
from __future__ import annotations

import json
import urllib.error
import urllib.request
from urllib.parse import urlparse

from clients.inbound_client import InboundClient

GRANT_HINTS = ("grant", "授权", "accesskey", "access key", "开放身份", "identity")


class HostNotAllowedError(RuntimeError):
    pass


def assert_host_allowed(base_url: str, allowed_hosts: list[str]) -> None:
    host = (urlparse(base_url).hostname or "").lower()
    allowed = {h.lower() for h in allowed_hosts}
    if host not in allowed:
        raise HostNotAllowedError(
            f"target host '{host}' not in allowedTargetHosts {sorted(allowed)}；生产目标默认阻断")


def check_reachable(base_url: str, allowed_hosts: list[str], timeout: int = 5) -> dict:
    assert_host_allowed(base_url, allowed_hosts)
    url = base_url.rstrip("/") + "/api/mdp/inbound/__probe_not_exists"
    req = urllib.request.Request(url, method="GET")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return {"reachable": True, "httpStatus": resp.status}
    except urllib.error.HTTPError as ex:
        # 401/404 说明服务在线（缺签名头或路径不存在）
        return {"reachable": True, "httpStatus": ex.code}
    except Exception as ex:  # URLError / timeout
        return {"reachable": False, "error": f"{type(ex).__name__}: {ex}"}


def classify_inbound_precheck(status: int, body) -> str:
    """schema 探测结果 → 配置层状态码。"""
    if status == 0:
        return "SOURCE_UNREACHABLE"
    if status == 200:
        return "READY"
    message = ""
    if isinstance(body, dict):
        message = str(body.get("message") or "")
    if status == 404:
        return "CONFIG_NOT_REGISTERED"
    if status in (401, 403):
        low = message.lower()
        if any(h in low for h in GRANT_HINTS):
            return "GRANT_MISSING"
        return "CONFIG_NOT_ENABLED"
    return "RUN_FAILED"


def precheck_inbound_entity(client: InboundClient, entity_code: str,
                            allowed_hosts: list[str], contract_version: str | None = None) -> dict:
    assert_host_allowed(client.base, allowed_hosts)
    status, body, headers = client.op_schema(entity_code, contract_version=contract_version)
    state = classify_inbound_precheck(status, body)
    result = {
        "entityCode": entity_code,
        "configState": state,
        "httpStatus": status,
        "requestHeaders": headers,
    }
    if state == "READY" and isinstance(body, dict):
        data = body.get("data") or {}
        fields = data.get("fields") or []
        result["requiredFields"] = [f.get("name") for f in fields if f.get("required")]
        result["bizKeyExpr"] = data.get("bizKeyExpr")
    elif isinstance(body, dict):
        result["message"] = body.get("message")
    return result


def precheck_all_inbound(client: InboundClient, entity_codes: list[str],
                         allowed_hosts: list[str]) -> list[dict]:
    return [precheck_inbound_entity(client, code, allowed_hosts) for code in entity_codes]


def query_admin_api(base_url: str, admin_token: str, path: str,
                    allowed_hosts: list[str], timeout: int = 10):
    """可选只读管理 API 查询；未提供 admin token 时由调用方跳过。"""
    assert_host_allowed(base_url, allowed_hosts)
    req = urllib.request.Request(
        base_url.rstrip("/") + path,
        headers={"Authorization": f"Bearer {admin_token}"},
        method="GET")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return resp.status, json.loads(resp.read().decode("utf-8") or "{}")
    except urllib.error.HTTPError as ex:
        try:
            return ex.code, json.loads(ex.read().decode("utf-8") or "{}")
        except json.JSONDecodeError:
            return ex.code, {}
    except Exception as ex:
        return 0, {"message": f"{type(ex).__name__}: {ex}"}