"""Oracle NetSuite REST record API driver — outbound identity sync. Carries the tenant credentials (Account ID + TBA Consumer Key/Secret + Token ID/Secret) for the ``netsuite`` identity-sync target kind, which pushes Helix's HRIS-of-truth employee master data into NetSuite via the REST record API. Credential facts (NetSuite TBA, verified against the live PingCAP account 2026-06-23): * Token-Based Authentication (OAuth 1.0a, HMAC-SHA256). The four static credentials are minted in NetSuite: Setup → Integration → Manage Integrations (Consumer Key/Secret) + Setup → Users/Roles → Access Tokens (Token ID/Secret). A dedicated role with "REST Web Services" + "Log in using Access Tokens" + the record permission is required — the Administrator role is NOT usable with TBA. * The REST host is derived from the Account ID, so only the id is configured here: ``{id_lower_dashed}.suitetalk.api.netsuite.com``. Wire-protocol work lives in ``app.domains.identity_sync.runners.netsuite`` (employee runner). The driver is catalog + ``test_connection`` only. To create the sync target in the Identity Sync Center, pick this integration row + the ``netsuite`` preset. """ from __future__ import annotations import logging 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="account_id", label="Account ID", is_secret=False, required=True, placeholder="7429932", hint=( "NetSuite 账户 ID(Setup → Company → Company Information → " "ACCOUNT ID)。REST 主机由它推导:7429932 → " "7429932.suitetalk.api.netsuite.com;沙箱 1234567_SB1 → " "1234567-sb1.suitetalk.api.netsuite.com。" ), ), IntegrationFieldInfo( key="consumer_key", label="Consumer Key", is_secret=True, required=True, hint=( "Setup → Integration → Manage Integrations 新建集成应用(勾 " "Token-Based Authentication)生成的 Consumer Key,仅在轮换时" "重新粘贴。" ), ), IntegrationFieldInfo( key="consumer_secret", label="Consumer Secret", is_secret=True, required=True, hint="与 Consumer Key 配对的 Consumer Secret(只显示一次)。", ), IntegrationFieldInfo( key="token_id", label="Token ID", is_secret=True, required=True, hint=( "Setup → Users/Roles → Access Tokens 新建访问令牌(集成 + 用户 " "+ 专用角色)生成的 Token ID。" ), ), IntegrationFieldInfo( key="token_secret", label="Token Secret", is_secret=True, required=True, hint="与 Token ID 配对的 Token Secret(只显示一次)。", ), IntegrationFieldInfo( key="ca_bundle_path", label="Approved CA Bundle Path", is_secret=False, required=False, placeholder="/etc/helix/ca-certificates/netsuite.pem", hint=( "仅在受控 TLS inspection/private PKI 场景填写挂载目录内的 " ".pem/.crt CA 文件；证书与主机名校验不会关闭。" ), ), ] @register_driver class NetSuiteDriver(IntegrationDriver): """Oracle NetSuite REST record API — outbound employee master data.""" kind = "identity_sync_netsuite" provider = "netsuite" display_name = "Oracle NetSuite" description = ( "NetSuite REST record API (/services/rest/record/v1)。支持 netsuite " "同步目标:把 HRIS 员工主数据推到 NetSuite 员工记录,外部 ID " "(externalId)= HRIS 工号为唯一键,PUT eid: 幂等 upsert(无需先查)。" "离职 = 停用(isInactive=true),不硬删。认证 = Token-Based " "Authentication(OAuth 1.0a / HMAC-SHA256),四个静态凭据逐请求签名," "无 token 交换。OneWorld 账户员工创建需 subsidiary —— 用 company_code " "value_map 映射子公司内部 ID,或在目标 runner_options 配默认子公司。" ) supports_multi_instance = False config_schema = _CONFIG_SCHEMA def _client(self): from app.domains.identity_sync.runners.netsuite import NetSuiteClient return NetSuiteClient( account_id=str(self.config.get("account_id") or ""), consumer_key=str(self.config.get("consumer_key") or ""), consumer_secret=str(self.config.get("consumer_secret") or ""), token_id=str(self.config.get("token_id") or ""), token_secret=str(self.config.get("token_secret") or ""), ca_bundle_path=self.config.get("ca_bundle_path"), ) def build_suiteql_client(self): """Public accessor for a signed NetSuite SuiteQL/record client. Used by the finance NetSuite coordinator (GL actuals inbound) to run SuiteQL via the SAME four TBA credentials this driver carries — GL read reuses the exact transport the identity sync already proved against the live account, with no credential-field duplication. Raises ``ValueError`` (from the client constructor) when the credentials are incomplete.""" return self._client() # Sentinel external id used by the read-only connection probe. The # ``eid:`` form mirrors how the runner actually addresses every record # (``employee/eid:{工号}``), so the probe validates the SAME capability # the sync uses. No real 工号 will ever collide with this literal. _PROBE_EID = "eid:__helix_connection_probe__" async def test_connection(self) -> IntegrationTestResult: """READ-ONLY probe: ``GET …/employee/eid:__helix_connection_probe__``. Why a targeted ``eid:`` lookup and NOT a collection ``GET /employee?limit=1``: this connector is upsert-only — it addresses every person by external id (``PUT employee/eid:{工号}``) and never enumerates employees. A collection/list GET requires the role to be able to *search* all employees, which a minimal upsert integration role legitimately lacks — NetSuite then returns ``400`` with ``o:errorCode = USER_ERROR`` / "您的当前角色没有执行此操作的权限", a false negative that has nothing to do with whether the sync works. A targeted ``eid:`` GET needs only the same record access the sync needs, so it is the faithful health check. Interpreting the result: * ``404`` (record-not-found for the sentinel id) → **healthy**: the four TBA credentials signed correctly, the role can reach the employee record via REST, the id simply doesn't exist (expected). * ``2xx`` → also healthy (only possible if someone really created that sentinel id; still proves access). * ``400 / 401 / 403`` → hard fail (bad credential, or the role genuinely lacks REST Web Services / Log in using Access Tokens / Employees record access). Never writes. """ from app.domains.identity_sync.runners.netsuite import ( NetSuiteError, _error_code, _vendor_message, redact_credentials, ) try: client = self._client() except ValueError as e: return IntegrationTestResult( ok=False, message=f"配置不完整: {e}", detail={"stage": "config"}, ) secret = str(self.config.get("token_secret") or "") try: status, body, raw, _loc = await client.request( "GET", f"/services/rest/record/v1/employee/{self._PROBE_EID}", ) except NetSuiteError as e: await client.aclose() return IntegrationTestResult( ok=False, message=redact_credentials( f"连接失败: {e}。检查 Account ID / 网络。", secret, ), detail={"stage": "request", "status_code": e.status_code}, ) finally: # aclose is idempotent; the except branch already closed. try: await client.aclose() except Exception: # noqa: BLE001 pass # 404 = the sentinel id doesn't exist (expected) but the role COULD # look it up → auth + REST + employee-record access all good. if (200 <= status < 300) or status == 404: return IntegrationTestResult( ok=True, message=f"OK · host={client.host} · employee 记录访问正常", detail={"host": client.host, "status_code": status}, ) hint = "" if status in (401, 403): hint = ( " · 401/403 = 凭据错误或角色缺少 REST Web Services / " "Log in using Access Tokens / 员工记录 权限" ) elif status == 400 and (_error_code(body) or "").upper() == "USER_ERROR": hint = ( " · 400 USER_ERROR = 角色可认证但无该操作权限;" "确认绑定角色的「列表→员工」权限,以及账户已启用 REST Web 服务" ) return IntegrationTestResult( ok=False, message=redact_credentials( f"HTTP {status}: {_vendor_message(body, raw)}{hint}", secret, ), detail={"stage": "request", "status_code": status}, )