"""汇联易 (Huilianyi) open-API driver — outbound identity sync. Carries the tenant credentials (企业ID / 企业密钥) for the ``huilianyi`` identity-sync target kind, which pushes Helix's HRIS-of-truth employee + department master data into 汇联易, replacing the legacy 汇联易→Authing daily full-pull. Credential facts (verified against the live tenant + opendocs implement guide, 2026-06-10): * Credentials = 权限ID + 权限密钥, minted per-permission in 汇联易后台 → 系统对接 → 开放接口权限. The permission row ALSO gates every endpoint individually — enable 登录(/oauth/token) + 员工 + 部门 module endpoints on it, or calls fail with permission errors even with a valid token. * AccessToken: ``POST {base_url}/oauth/token`` with ``Authorization: Basic base64(权限ID:权限密钥)`` + form ``grant_type=client_credentials&scope=write``; SaaS gateway base is ``https://api.huilianyi.com/gateway``. ``token_path`` / ``token_auth_style`` stay configurable for private-deploy tenants. Wire-protocol work lives in ``app.domains.identity_sync.runners.huilianyi`` (employee + department runners). The driver is catalog + ``test_connection`` only. To create the sync target in the Identity Sync Center, pick this integration row + the ``huilianyi`` preset. """ from __future__ import annotations import logging from typing import Any from app.integrations.driver_base import IntegrationDriver from app.integrations.registry import register_driver from app.integrations.schemas import ( IntegrationFieldInfo, IntegrationTestResult, ) logger = logging.getLogger(__name__) _CONFIG_SCHEMA = [ IntegrationFieldInfo( key="base_url", label="API Base URL", is_secret=False, required=True, placeholder="https://api.huilianyi.com/gateway", hint=( "汇联易开放 API 网关地址。SaaS 标准租户为 " "https://api.huilianyi.com/gateway;私有化/专属集群按汇联易" "交付文档填写。" ), ), IntegrationFieldInfo( key="client_id", label="权限ID (Client ID)", is_secret=False, required=True, hint=( "汇联易后台 → 系统对接 → 开放接口权限 → 新建接口权限 生成的" "权限ID;记得在该权限上开通 登录 + 员工 + 部门 模块的接口。" ), ), IntegrationFieldInfo( key="client_secret", label="权限密钥 (Client Secret)", is_secret=True, required=True, hint="与权限ID 配对的权限密钥,仅在轮换时重新粘贴。", ), IntegrationFieldInfo( key="token_path", label="Token 端点路径", is_secret=False, required=False, placeholder="/oauth/token", hint=( "AccessToken 获取路径,默认 /oauth/token(已对照租户接口目录" "核实)。私有化集群不同时按交付文档调整。" ), ), IntegrationFieldInfo( key="token_auth_style", label="Token 凭证方式", is_secret=False, required=False, placeholder="basic", hint=( "basic = HTTP Basic 头(默认,标准网关用这个);" "form = 凭证放表单体。「测试连接」报 401/400 时切换尝试。" ), ), IntegrationFieldInfo( key="ca_bundle_path", label="Approved CA Bundle Path", is_secret=False, required=False, placeholder="/etc/helix/ca-certificates/huilianyi.pem", hint=( "私有化网关可填写挂载在 /etc/helix/ca-certificates 下的 " ".pem/.crt CA 文件；证书与主机名校验不会关闭。" ), ), ] @register_driver class HuilianyiDriver(IntegrationDriver): """汇联易 open API — outbound employee + department master data.""" kind = "identity_sync_huilianyi" provider = "huilianyi" display_name = "汇联易 (Huilianyi)" description = ( "汇联易开放 API (/api/open)。支持 huilianyi 同步目标:内部分两个阶段" "(部门 custDeptNumber 同步 → 员工同步,工号 employeeID 为唯一键)," "管理员视角是一个目标一个开关。离职 = 停用(离职员工V2),不删数据。" "凭证在汇联易后台「系统对接 → 开放接口权限」生成。" ) supports_multi_instance = False config_schema = _CONFIG_SCHEMA def _client(self, *, transport=None): from app.domains.identity_sync.runners.huilianyi import HuilianyiClient return HuilianyiClient( base_url=str(self.config.get("base_url") or ""), client_id=str(self.config.get("client_id") or ""), client_secret=str(self.config.get("client_secret") or ""), token_path=str(self.config.get("token_path") or "/oauth/token"), token_auth_style=str( self.config.get("token_auth_style") or "basic" ), transport=transport, ca_bundle_path=self.config.get("ca_bundle_path"), ) async def test_connection(self) -> IntegrationTestResult: """Two-stage READ-ONLY probe: 1. Token exchange — credentials + token endpoint shape valid. 2. ``GET /api/open/user`` (1-row page WITH the mandatory date window — omitting startDate/endDate draws errorCode 120003 「缺少必填项」, which would misread an authorized credential as a permission problem) — open-API 人员读权限 granted to this 企业ID (汇联易 scopes open-API permissions per-module in 开放接口权限). Stage 2 failure downgrades to a warning (the tenant may scope permissions write-only); stage 1 failure is a hard fail with the raw status + body excerpt so the admin can tell apart wrong-path (404) / wrong-style (400/401) / wrong-credential. """ from app.domains.identity_sync.runners.huilianyi import ( HuilianyiError, _error_code, _is_ok, redact_credentials, ) from app.domains.identity_sync.runners.tls import ( build_identity_sync_tls_verify, ) from app.integrations._url_safety import ( UnsafeURLError, safe_async_http_transport, ) try: tls_verify = build_identity_sync_tls_verify( base_url=str(self.config.get("base_url") or ""), verify_ssl=True, ca_bundle_path=self.config.get("ca_bundle_path"), ) transport = await safe_async_http_transport( str(self.config.get("base_url") or ""), verify=tls_verify, ) except (UnsafeURLError, ValueError) as e: # e.message = host + reason (full-url redaction contract). message = e.message if isinstance(e, UnsafeURLError) else str(e) return IntegrationTestResult( ok=False, message=f"base_url/TLS 配置无效: {message:.200}", detail={"stage": "config"}, ) try: client = self._client(transport=transport) except ValueError as e: return IntegrationTestResult( ok=False, message=f"配置不完整: {e}", detail={"stage": "config"}, ) probes: dict[str, dict[str, Any]] = {} try: token = await client.get_access_token() probes["token"] = {"ok": True} except HuilianyiError as e: await client.aclose() secret = str(self.config.get("client_secret") or "") return IntegrationTestResult( ok=False, message=redact_credentials( f"Token 获取失败: {e}。404 → 调整「Token 端点路径」;" "400/401 → 检查企业ID/企业密钥或切换「Token 凭证方式」。", secret, ), detail={ "stage": "token", "status_code": e.status_code, # Vendor error bodies can echo the inbound # Authorization header — redact OUR credential # material before the excerpt reaches the admin # API response (the generic scrub only covers # JWT/base64 shapes, not opaque tokens). "body_excerpt": redact_credentials( e.body_excerpt or "", secret, ), }, ) except Exception as e: # noqa: BLE001 await client.aclose() return IntegrationTestResult( ok=False, message=f"Token 获取失败: {type(e).__name__}: {e}", detail={"stage": "token"}, ) try: from app.domains.identity_sync.runners.huilianyi import ( _FULL_WINDOW_START, _full_window_end, ) status, body, raw = await client.request( "GET", "/api/open/user", params={ "page": "1", "size": "1", # Mandatory on 增量查询员工 — without the window the # API returns errorCode 120003 (missing params), # which is a validation error, not a permission # verdict, and would fail an authorized credential. "startDate": _FULL_WINDOW_START, "endDate": _full_window_end(), }, ) # 汇联易 wraps logical/permission errors as HTTP 200 + # non-0000 errorCode — a bare status check would report # "ok" for a credential that can't actually read 员工. if _is_ok(status, body): probes["user_read"] = {"ok": True} else: probes["user_read"] = { "ok": False, "error": redact_credentials( f"HTTP {status} errorCode={_error_code(body)}: " f"{raw[:200]}", str(self.config.get("client_secret") or ""), token, ), "hint": ( "开放接口权限里未开通「人员」模块读权限,或该" "企业ID 无集团权限。同步推送(写)不一定受影响," "但增量对账读不了。" ), } except Exception as e: # noqa: BLE001 probes["user_read"] = { "ok": False, "error": f"{type(e).__name__}: {e}", } finally: await client.aclose() user_read_ok = probes.get("user_read", {}).get("ok") is True msg = "OK · access token 获取成功" if not user_read_ok: msg += " · ⚠ 人员读接口探测失败,详见 detail(写权限可能不受影响)" return IntegrationTestResult( ok=True, message=msg, detail={"probes": probes}, )