"""Integration CRUD + driver resolution + health state. All database I/O for ``integrations`` / ``integration_approval_flows`` / ``platform_settings`` lives here. Domain code never touches the ORM rows directly — it asks this service for a *driver instance* and works through the driver's typed interface. Resolution helpers: * :func:`build_primary_driver(kind)` — use when the caller wants "THE HRIS" / "THE chat-approval" and there's exactly one primary. * :func:`build_driver_by_id(id_or_slug)` — explicit addressing for multi-instance kinds (cloud accounts, software-provisioning bindings). Both raise a clear :class:`AppException` when no active integration matches — the router turns that into a 503 with a message pointing the admin at ``/admin/integrations``. """ from __future__ import annotations import inspect import json import logging import re from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from uuid import UUID from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.credential_redaction import is_credential_key, redact_credentials from app.core.database import async_session_factory from app.core.exceptions import AppException, ConflictError, NotFoundError from app.core.txn import best_effort from app.domains.cmdb.credential_crypto import ( CredentialDecryptionError, decrypt_fields, encrypt_fields, ) from app.identity.person_models import IdentityPerson from app.integrations.driver_base import IntegrationDriver from app.integrations.models import ( Integration, IntegrationApprovalFlow, PlatformSetting, ) from app.integrations.registry import get_driver_class from app.integrations.schemas import ( ApprovalFlowResponse, IntegrationResponse, IntegrationTestResult, PlatformSettingResponse, ) logger = logging.getLogger(__name__) # Table names that may be checked for durable references before deleting an # integration. Interpolated into raw SQL (identifiers can't be parameterised), # so a compile-time allowlist prevents accidental injection. _INTEGRATION_DELETE_GUARD_TABLES: frozenset[str] = frozenset({ # ``cmdb_ai_resources`` dropped in Phase 2 (migration 151); # AI billing history now lives in cmdb_cost_facts which is # already gated by its own RESTRICT FK to integrations. "bailian_workspaces", "bailian_api_keys", "bailian_usage_daily", # DAX resource projections are runtime authority and audit/cutover # history; disabling an integration must not erase them. "data_access_resources", }) def _scrub_message(s: str) -> str: """Strip credential-shaped substrings from a free-form error message before it lands on a stored ``last_test_message`` or audit detail. The intent is defense-in-depth, not a hard guarantee — boto / Azure / Aliyun SDK error messages occasionally embed AK/SK fragments in request signatures, and a configuration mistake might also slip a raw credential into the message text. We scrub the obvious shapes: AWS access key IDs, PEM private-key blocks, the most common ``key=value`` credential aliases, and long base64-looking blobs that almost always carry a token, signature, or encrypted payload. """ return redact_credentials(s, max_len=1000, redact_long_base64=True) # Detail keys that carry credential material (even a prefix). Legacy or # third-party drivers may put ``token[:8] + "..."`` under these to help # admins eyeball WHICH token is configured — but even 6–8 chars of a bearer # token / app secret is credential material and must not reach the # admin-facing test payload or audit log. We drop the key entirely rather # than redact-in-place so there's zero fragment left. _CREDENTIAL_DETAIL_KEYS = frozenset({ "token_prefix", "token", "access_token", "app_secret", "secret", "app_access_token", "tenant_access_token", "user_access_token", }) def _scrub_detail(detail: Any) -> Any: """Recursively scrub a ``test_connection`` detail payload. Applies :func:`_scrub_message` to every string value and drops any key in :data:`_CREDENTIAL_DETAIL_KEYS`. Mirrors what we already do to ``result.message`` so a driver can't route a token fragment or a raw vendor exception repr around the message scrub via ``detail``. """ if isinstance(detail, dict): out: dict[Any, Any] = {} for k, v in detail.items(): if ( isinstance(k, str) and ( k.lower() in _CREDENTIAL_DETAIL_KEYS or is_credential_key(k) ) ): continue out[k] = _scrub_detail(v) return out if isinstance(detail, (list, tuple)): return type(detail)(_scrub_detail(v) for v in detail) if isinstance(detail, str): return _scrub_message(detail) return detail _SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9\-_]{0,62}[a-z0-9]$") @dataclass(slots=True) class ResolvedIntegration: """Integration row + decrypted config, ready to feed a driver.""" row: Integration config: dict[str, Any] # ─── Reads: resolve + build driver ──────────────────────────────── async def _load_by_id_or_slug( session: AsyncSession, key: str | UUID, *, for_update: bool = False, ) -> Integration | None: """Load an Integration row by UUID, UUID-string, or slug. ``for_update=True`` issues ``SELECT … FOR UPDATE`` so the caller can serialise mutations on the row (used by webhook-secret rotate to prevent two concurrent admins from overwriting each other's ``_previous`` slot bookkeeping). """ stmt: Any if isinstance(key, UUID): stmt = select(Integration).where(Integration.id == key) else: # Allow UUID strings too — fall back to slug lookup on parse error. try: uuid_key = UUID(str(key)) stmt = select(Integration).where(Integration.id == uuid_key) except (ValueError, TypeError): stmt = select(Integration).where(Integration.slug == str(key)) if for_update: stmt = stmt.with_for_update() return (await session.execute(stmt)).scalar_one_or_none() async def _resolve( session: AsyncSession, row: Integration, ) -> ResolvedIntegration: merged: dict[str, Any] = {} if row.config_public: merged.update(row.config_public) if row.secrets_encrypted and row.secrets_nonce: try: merged.update(decrypt_fields( row.secrets_encrypted, row.secrets_nonce, integration_id=str(row.id), )) except CredentialDecryptionError as e: # ``decrypt_fields`` converts every cryptography-level failure # (InvalidTag, bad base64, wrong nonce length, KEY rotation) # into ``CredentialDecryptionError`` with an actionable # message — surface that directly as 503 so the admin UI # shows what to do instead of a generic 500. raise AppException( code="ENCRYPTION_KEY_MISMATCH", message=( f"Integration {row.slug or row.id!r} secrets cannot be " f"decrypted: {e}. Re-enter the credentials at " f"/admin/integrations or restore the original key." ), status_code=503, ) from e except RuntimeError as e: # Non-decryption RuntimeError (e.g. missing KEK env). Distinct # error code so ops can tell apart misconfig vs corrupted row. raise AppException( code="ENCRYPTION_KEY_MISSING", message=str(e), status_code=503, ) from e return ResolvedIntegration(row=row, config=merged) async def get_primary( kind: str, *, session: AsyncSession | None = None, ) -> Integration: """Return the primary active integration for a kind, or raise 503.""" async def _body(s: AsyncSession) -> Integration: row = ( await s.execute( select(Integration) .where( Integration.kind == kind, Integration.enabled.is_(True), Integration.is_primary.is_(True), ) .order_by(Integration.created_at) ) ).scalars().first() if row is None: # Fall back to ANY single enabled instance of this kind enabled_rows = ( await s.execute( select(Integration).where( Integration.kind == kind, Integration.enabled.is_(True), ) ) ).scalars().all() if len(enabled_rows) == 1: return enabled_rows[0] raise AppException( code="INTEGRATION_NOT_CONFIGURED", message=( f"No primary {kind!r} integration is configured. " f"Add one under /admin/integrations and mark it primary." ), status_code=503, ) return row if session is not None: return await _body(session) async with async_session_factory() as owned: return await _body(owned) async def build_primary_driver( kind: str, *, session: AsyncSession | None = None, ) -> IntegrationDriver: """Resolve the primary integration for a kind and construct its driver.""" if session is not None: row = await get_primary(kind, session=session) resolved = await _resolve(session, row) else: async with async_session_factory() as owned: row = await get_primary(kind, session=owned) resolved = await _resolve(owned, row) return _build_driver_from_resolved(resolved) async def build_driver_by_id( integration_id: str | UUID, *, session: AsyncSession | None = None, ) -> IntegrationDriver: """Resolve by explicit id/slug. Raises 404 if unknown, 503 if disabled.""" async def _body(s: AsyncSession) -> IntegrationDriver: row = await _load_by_id_or_slug(s, integration_id) if row is None: raise NotFoundError("Integration", str(integration_id)) if not row.enabled: raise AppException( code="INTEGRATION_DISABLED", message=( f"Integration {row.slug!r} is disabled. Re-enable it " f"under /admin/integrations." ), status_code=503, ) resolved = await _resolve(s, row) return _build_driver_from_resolved(resolved) if session is not None: return await _body(session) async with async_session_factory() as owned: return await _body(owned) def _build_driver_from_resolved( resolved: ResolvedIntegration, ) -> IntegrationDriver: row = resolved.row cls = get_driver_class(row.kind, row.provider) if cls is None: raise AppException( code="INTEGRATION_DRIVER_UNKNOWN", message=( f"No driver registered for ({row.kind!r}, {row.provider!r})." " The driver module may have been removed from this build." ), status_code=500, ) # ``row.capabilities`` is the admin-customised per-instance set — # the driver-class ``capabilities`` attribute is the factory default # used only when the column is NULL/empty (a fresh integration the # admin hasn't yet pruned). Forwarding the per-instance set lets # diagnostic paths (``test_connection``) probe exactly what the # admin opted into. # Defense in depth: probe the concrete driver's ``__init__`` for # the ``enabled_capabilities`` kwarg before passing it. Most # drivers inherit ``IntegrationDriver.__init__`` (which accepts # it), but a concrete subclass with its own ``__init__`` override # may not — calling unconditionally with the kwarg would raise # ``TypeError`` and brick every code path that builds that # driver (provisioning, offboarding). Forwarding only when the # signature accepts the kwarg keeps the contract opt-in. init_params = inspect.signature(cls.__init__).parameters if "enabled_capabilities" in init_params or any( p.kind == inspect.Parameter.VAR_KEYWORD for p in init_params.values() ): return cls( integration_id=str(row.id), config=resolved.config, enabled_capabilities=list(row.capabilities or []) or None, ) return cls(integration_id=str(row.id), config=resolved.config) # ─── Approval-flow lookup (for chat_approval integrations) ──────── async def get_approval_flows_map( *, session: AsyncSession | None = None, integration_id: UUID | None = None, ) -> dict[str, str]: """All active approval flows on the primary chat_approval integration, keyed by ``flow_key`` → ``external_code``. Returns ``{}`` when nothing is configured (vs. raising); callers iterating over a fixed set of templates skip the missing ones. """ async def _body(s: AsyncSession) -> dict[str, str]: if integration_id is None: try: integ = await get_primary("chat_approval", session=s) except AppException: return {} target_id = integ.id else: target_id = integration_id rows = ( await s.execute( select(IntegrationApprovalFlow).where( IntegrationApprovalFlow.integration_id == target_id, IntegrationApprovalFlow.active.is_(True), ) ) ).scalars().all() return {r.flow_key: r.external_code for r in rows} if session is not None: return await _body(session) async with async_session_factory() as owned: return await _body(owned) async def get_approval_flow( flow_key: str, *, session: AsyncSession | None = None, integration_id: UUID | None = None, ) -> IntegrationApprovalFlow: """Resolve an approval flow definition by flow_key. By default we look up on the primary chat_approval integration. Pass ``integration_id`` to scope explicitly when you have multiple chat-approval systems configured. """ async def _body(s: AsyncSession) -> IntegrationApprovalFlow: if integration_id is None: integration = await get_primary("chat_approval", session=s) target_id = integration.id else: target_id = integration_id row = ( await s.execute( select(IntegrationApprovalFlow).where( IntegrationApprovalFlow.integration_id == target_id, IntegrationApprovalFlow.flow_key == flow_key, IntegrationApprovalFlow.active.is_(True), ) ) ).scalar_one_or_none() if row is None: raise AppException( code="APPROVAL_FLOW_NOT_CONFIGURED", message=( f"No active approval flow mapping for flow_key={flow_key!r}." " Add one under /admin/integrations//flows." ), status_code=503, ) return row if session is not None: return await _body(session) async with async_session_factory() as owned: return await _body(owned) # ─── Admin CRUD ──────────────────────────────────────────────────── class IntegrationService: """Admin-facing CRUD + test-connection orchestration.""" def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) async def list( self, *, kind: str | None = None, enabled: bool | None = None, ) -> list[IntegrationResponse]: stmt = select(Integration).order_by(Integration.kind, Integration.name) if kind is not None: stmt = stmt.where(Integration.kind == kind) if enabled is not None: stmt = stmt.where(Integration.enabled.is_(enabled)) rows = (await self.db.execute(stmt)).scalars().all() return await self._serialize_many(list(rows)) async def get(self, key: str | UUID) -> IntegrationResponse: row = await _load_by_id_or_slug(self.db, key) if row is None: raise NotFoundError("Integration", str(key)) return (await self._serialize_many([row]))[0] async def create( self, *, kind: str, provider: str, name: str, slug: str, description: str | None, is_primary: bool, enabled: bool, capabilities: list[str] | None, config: dict[str, Any], trace_id: str, actor_user_id: UUID, ) -> IntegrationResponse: driver_cls = get_driver_class(kind, provider) if driver_cls is None: raise AppException( code="INTEGRATION_DRIVER_UNKNOWN", message=f"No driver registered for ({kind!r}, {provider!r})", status_code=400, ) _validate_slug(slug) # Uniqueness existing = ( await self.db.execute( select(Integration).where(Integration.slug == slug) ) ).scalar_one_or_none() if existing is not None: raise ConflictError(f"An integration with slug {slug!r} already exists") # Multi-instance check if not driver_cls.supports_multi_instance: other = ( await self.db.execute( select(Integration).where( Integration.kind == kind, Integration.provider == provider, ) ) ).scalars().first() if other is not None: raise ConflictError( f"Driver ({kind!r}, {provider!r}) is single-instance; " f"one already exists ({other.slug})", ) public, secrets = self._split_fields(driver_cls, config) self._require_fields(driver_cls, public, secrets, is_new=True) # Pre-launch hardening 2026-05-06: AAD = integration row id. # We need the row's id BEFORE encrypting so the AEAD tag binds # to the row identity. Insert with a generated UUID (default # is uuid.uuid4 on the column) THEN encrypt with that id — # i.e. encode placeholders first, flush to materialize the id, # encrypt with id-as-AAD, write back. row = Integration( kind=kind, provider=provider, name=name, slug=slug, description=description, enabled=enabled, is_primary=False, config_public=public or None, secrets_encrypted=None, secrets_nonce=None, capabilities=capabilities, created_by=actor_user_id, updated_by=actor_user_id, ) self.db.add(row) await self.db.flush() # materialize row.id if secrets: cipher, nonce = encrypt_fields( secrets, integration_id=str(row.id), ) row.secrets_encrypted = cipher row.secrets_nonce = nonce await self.db.flush() if is_primary: # Set ours first, then unset siblings — both changes land in # the same flush so there is never a zero-primary window. row.is_primary = True await self._unset_other_primaries(row) await self.db.flush() # Cloud integrations: auto-provision per-capability scheduled_jobs # so /admin/scheduled-jobs immediately shows them and APScheduler # starts syncing on the next boot. Non-cloud kinds are ignored. if kind == "cloud": from app.domains.cmdb.scheduled_jobs_provisioning import ( sync_jobs_for_integration, ) await sync_jobs_for_integration(self.db, row) await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="integration.create", status="success", resource_type="integration", resource_id=str(row.id), detail={ "kind": kind, "provider": provider, "slug": slug, "public_keys": sorted(public.keys()), "secret_keys_set": sorted(secrets.keys()), }, ) return (await self._serialize_many([row]))[0] async def update( self, key: str | UUID, *, name: str | None, description: str | None, enabled: bool | None, is_primary: bool | None, capabilities: list[str] | None, config: dict[str, Any] | None, trace_id: str, actor_user_id: UUID, ) -> IntegrationResponse: row = await _load_by_id_or_slug(self.db, key) if row is None: raise NotFoundError("Integration", str(key)) driver_cls = get_driver_class(row.kind, row.provider) if driver_cls is None: raise AppException( code="INTEGRATION_DRIVER_UNKNOWN", message=f"No driver registered for ({row.kind!r}, {row.provider!r})", status_code=500, ) if name is not None: row.name = name if description is not None: row.description = description if enabled is not None: row.enabled = enabled if capabilities is not None: row.capabilities = capabilities if config is not None: existing_secrets: dict[str, Any] = {} if row.secrets_encrypted and row.secrets_nonce: # Round-1 review follow-up (Review-E P2-5): mirror the # GET path's CredentialDecryptionError → 503 mapping so # a PUT/PATCH against an integration with a corrupted # blob (KEK rotation, ciphertext swap, base64 drift) # returns a clean 503 with actionable text instead of # an opaque 500. Without this wrap, ``decrypt_fields`` # raises ``CredentialDecryptionError(RuntimeError)`` # straight through the global handler. try: existing_secrets = decrypt_fields( row.secrets_encrypted, row.secrets_nonce, integration_id=str(row.id), ) except CredentialDecryptionError as e: raise AppException( code="ENCRYPTION_KEY_MISMATCH", message=( f"Integration {row.slug or row.id!r} existing " f"secrets cannot be decrypted: {e}. Re-enter " "the credentials with the full config dict " "or restore the original " "CREDENTIAL_ENCRYPTION_KEY." ), status_code=503, ) from e except RuntimeError as e: raise AppException( code="ENCRYPTION_KEY_MISSING", message=str(e), status_code=503, ) from e new_public, new_secrets = self._split_fields(driver_cls, config) # Public: merge with existing so partial updates don't drop # fields that weren't in this payload. An explicit empty # value would have been dropped by _split_fields already, so # "clear" is not expressible via PATCH — use PUT-style # replacement by sending the full config dict. merged_public: dict[str, Any] = dict(row.config_public or {}) merged_public.update(new_public) # Secrets: merge — absent keys keep existing, empty string clears. merged_secrets = dict(existing_secrets) for k, v in new_secrets.items(): if v in (None, ""): merged_secrets.pop(k, None) else: merged_secrets[k] = v # P1-F (2026-06-02 audit): refuse to write # ``webhook_hmac_secret`` directly through PATCH — that # bypasses the rotation bookkeeping (no ``_previous`` slot # set, no 24h overlap, no audit row tagged # ``integration.webhook_secret_rotate``). Force admins # through ``POST /integrations/{key}/webhook-secret/rotate`` # where the dual-secret invariants are enforced. if ( "webhook_hmac_secret" in new_secrets and (new_secrets["webhook_hmac_secret"] or "").strip() ): raise AppException( code="WEBHOOK_SECRET_MUST_USE_ROTATE", message=( "webhook_hmac_secret cannot be edited directly. " "Use POST /integrations/{key}/webhook-secret/rotate " "to generate a new secret with the 24h overlap " "window. Saving the field directly would bypass " "the rotation bookkeeping and lock SF / CPI out " "the moment the new secret lands." ), status_code=400, ) self._require_fields( driver_cls, merged_public, merged_secrets, is_new=False, ) row.config_public = merged_public or None if merged_secrets: cipher, nonce = encrypt_fields( merged_secrets, integration_id=str(row.id), ) row.secrets_encrypted = cipher row.secrets_nonce = nonce else: row.secrets_encrypted = None row.secrets_nonce = None if is_primary is True and not row.is_primary: # Set ours first, then unset siblings — both changes land in # the same flush so there is never a zero-primary window. row.is_primary = True await self._unset_other_primaries(row) elif is_primary is False: row.is_primary = False row.updated_by = actor_user_id await self.db.flush() # Cloud integrations: diff scheduled_jobs. If the admin toggled # capabilities we need to add/remove the matching job rows. # Display-name refresh also propagates renames. if row.kind == "cloud": from app.domains.cmdb.scheduled_jobs_provisioning import ( sync_jobs_for_integration, ) await sync_jobs_for_integration(self.db, row) await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="integration.update", status="success", resource_type="integration", resource_id=str(row.id), detail={"slug": row.slug}, ) return (await self._serialize_many([row]))[0] async def delete( self, key: str | UUID, *, trace_id: str, actor_user_id: UUID, ) -> None: row = await _load_by_id_or_slug(self.db, key) if row is None: raise NotFoundError("Integration", str(key)) # Cloud integrations: clear scheduled_jobs rows first so the FK # in their config JSON doesn't outlive the integration. removed_jobs = 0 if row.kind == "cloud": from app.domains.cmdb.scheduled_jobs_provisioning import ( remove_jobs_for_integration, ) removed_jobs = await remove_jobs_for_integration(self.db, str(row.id)) # Pre-launch hardening 2026-05-06 (round 2): migration 124 # converted every AI/Bailian parent FK from CASCADE to RESTRICT. # DAX resource projections follow the same durable-reference rule: # deleting an integration with history/runtime rows should surface a # clear 409 instead of either leaking a raw FK error or erasing rows. history_ref_count = 0 for table in _INTEGRATION_DELETE_GUARD_TABLES: from sqlalchemy import text as _text async with best_effort( self.db, label="service best-effort step", log=logger, ) as be_755_step: cnt = await self.db.scalar( _text( f"SELECT COUNT(1) FROM {table} " f"WHERE integration_id = :iid" ), {"iid": str(row.id)}, ) history_ref_count += int(cnt or 0) if not be_755_step.ok: # Table may not exist on all deployments (e.g. tests # without migration 079). Skip silently — RESTRICT FK # is the actual enforcement; this is just nicer UX. continue if history_ref_count > 0: raise AppException( code="INTEGRATION_HAS_HISTORY", message=( f"Integration {row.slug or row.id!r} has " f"{history_ref_count} durable history/resource " f"row(s) referring to it. Hard delete is refused " f"to preserve audit/runtime history. Disable via " f"`Integration.enabled=False` instead — that " f"halts new traffic without losing the ledger." ), status_code=409, ) await self.db.delete(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="integration.delete", status="success", resource_type="integration", resource_id=str(row.id), detail={"slug": row.slug, "removed_scheduled_jobs": removed_jobs}, ) async def test_connection( self, key: str | UUID, *, trace_id: str, actor_user_id: UUID | None, actor_type: str = "user", ) -> IntegrationTestResult: """Probe one integration and persist the verdict. ``actor_user_id`` is optional and ``actor_type`` defaults to ``"user"`` so every existing caller is unchanged. The daily ``integrations.health_check`` cron passes ``None`` / ``"system"``: before it existed, ``last_test_status`` was written **only** when an admin clicked the button, so an expired credential was never noticed here — it was noticed as failures piling up in five downstream domains. ``updated_by`` is left alone on a system probe: a health check is not an edit, and overwriting the last human editor would lose real provenance. """ row = await _load_by_id_or_slug(self.db, key) if row is None: raise NotFoundError("Integration", str(key)) resolved = await _resolve(self.db, row) driver = _build_driver_from_resolved(resolved) result: IntegrationTestResult try: result = await driver.test_connection() except AppException as e: result = IntegrationTestResult( ok=False, message=_scrub_message(e.message or ""), detail={"code": e.code}, ) except Exception as e: # noqa: BLE001 result = IntegrationTestResult( ok=False, message=_scrub_message(str(e)), detail={"type": type(e).__name__}, ) scrubbed = _scrub_message(result.message or "") if result.message else None row.last_tested_at = datetime.now(UTC) row.last_test_status = "ok" if result.ok else "failed" row.last_test_message = scrubbed[:1000] if scrubbed else None if actor_user_id is not None: row.updated_by = actor_user_id await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type=actor_type, actor_user_id=actor_user_id, action="integration.test_connection", status="success" if result.ok else "failure", resource_type="integration", resource_id=str(row.id), detail={"slug": row.slug, "message": (scrubbed or "")[:200]}, ) # Surface a fully-scrubbed result back to the caller so the admin # UI doesn't re-expose what we stripped before storing — this # covers BOTH the free-form message AND the structured detail # (which drivers use for token_prefix / raw vendor exception # reprs that would otherwise bypass the message scrub). return IntegrationTestResult( ok=result.ok, message=scrubbed if scrubbed is not None else result.message, detail=_scrub_detail(result.detail), ) async def rotate_webhook_hmac_secret( self, key: str | UUID, *, trace_id: str, actor_user_id: UUID, overlap_seconds: int = 24 * 60 * 60, ) -> dict[str, Any]: # Serialise mutations on this Integration row — two admins # clicking "Rotate" concurrently would otherwise both read the # same primary, generate independent new secrets, and the # second write would orphan the first new secret as the # ``_previous`` slot (admin distributed it to HRIS but Helix's # next rotate already aged it out). FOR UPDATE blocks the # second transaction until the first commits, then it sees # the post-rotate state and produces a clean overlap. # Also closes the TOCTOU window where the row could be # DELETE'd between load and write. """Generate a fresh ``webhook_hmac_secret`` for an integration and keep the previous one valid for ``overlap_seconds`` (default 24h) as ``webhook_hmac_secret_previous`` + ``…_previous_expires_at``. The plaintext of the NEW secret is returned exactly once — the caller is responsible for handing it to the upstream integrator (e.g. the HRIS team / CPI iflow author) over a secure channel. It is encrypted before persistence; no admin endpoint reveals an already-rotated secret again. Webhook auth tries the primary first, then the previous secret if its expiry hasn't passed — so SF / CPI / Slack-style rotations with a brief deployment window don't drop events. """ import secrets as _secrets from datetime import UTC, datetime, timedelta row = await _load_by_id_or_slug(self.db, key, for_update=True) if row is None: raise NotFoundError("Integration", str(key)) driver_cls = get_driver_class(row.kind, row.provider) if driver_cls is None: raise AppException( code="INTEGRATION_DRIVER_UNKNOWN", message=f"No driver registered for ({row.kind!r}, {row.provider!r})", status_code=500, ) # Guard: only drivers that declared a ``webhook_hmac_secret`` # field in their config schema can be rotated. Prevents the # rotate button from showing up on integrations that don't # receive callbacks (e.g. Atlassian outbound-only). field_keys = {f.key for f in (driver_cls.config_schema or [])} if "webhook_hmac_secret" not in field_keys: raise AppException( code="WEBHOOK_SECRET_NOT_APPLICABLE", message=( f"Integration provider {row.provider!r} does not " "declare a ``webhook_hmac_secret`` field; nothing " "to rotate." ), status_code=400, ) existing: dict[str, Any] = {} if row.secrets_encrypted and row.secrets_nonce: try: existing = decrypt_fields( row.secrets_encrypted, row.secrets_nonce, integration_id=str(row.id), ) except CredentialDecryptionError as e: raise AppException( code="ENCRYPTION_KEY_MISMATCH", message=( f"Integration {row.slug or row.id!r} existing " f"secrets cannot be decrypted: {e}. Restore the " "original CREDENTIAL_ENCRYPTION_KEY before rotating." ), status_code=503, ) from e except RuntimeError as e: raise AppException( code="ENCRYPTION_KEY_MISSING", message=str(e), status_code=503, ) from e previous_value = existing.get("webhook_hmac_secret") # 256 bits as 64-char hex — fits Antd Input.Password preview # cleanly and matches the entropy of the prior token-only auth. new_secret = _secrets.token_hex(32) existing["webhook_hmac_secret"] = new_secret if previous_value: existing["webhook_hmac_secret_previous"] = previous_value existing["webhook_hmac_secret_previous_expires_at"] = ( datetime.now(UTC) + timedelta(seconds=overlap_seconds) ).isoformat() else: # First-ever generation — no overlap window applies. existing.pop("webhook_hmac_secret_previous", None) existing.pop("webhook_hmac_secret_previous_expires_at", None) cipher, nonce = encrypt_fields(existing, integration_id=str(row.id)) row.secrets_encrypted = cipher row.secrets_nonce = nonce row.updated_by = actor_user_id await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="integration.webhook_secret_rotate", status="success", resource_type="integration", resource_id=str(row.id), detail={ "slug": row.slug, "rotated_first_time": previous_value is None, "previous_expires_at": existing.get( "webhook_hmac_secret_previous_expires_at" ), }, ) # NOTE: never log or audit the plaintext ``new_secret``. return { "secret": new_secret, "previous_expires_at": existing.get( "webhook_hmac_secret_previous_expires_at" ), "rotated_first_time": previous_value is None, } async def webhook_secret_status( self, key: str | UUID, ) -> dict[str, Any]: """Read-only status of the webhook secret pair without ever revealing the plaintext — drives the admin UI badge ("primary set + previous expires in 23h") so an admin can plan the swap with HRIS without re-generating a new secret just to peek.""" from datetime import UTC, datetime row = await _load_by_id_or_slug(self.db, key) if row is None: raise NotFoundError("Integration", str(key)) existing: dict[str, Any] = {} if row.secrets_encrypted and row.secrets_nonce: try: existing = decrypt_fields( row.secrets_encrypted, row.secrets_nonce, integration_id=str(row.id), ) except (CredentialDecryptionError, RuntimeError): # Mirror the GET path: surface as "no secret" rather # than 503-ing the status endpoint and blocking the UI. return { "has_primary": False, "has_previous": False, "previous_expires_at": None, } has_primary = bool(existing.get("webhook_hmac_secret")) prev_exp_raw = existing.get("webhook_hmac_secret_previous_expires_at") has_previous = False if existing.get("webhook_hmac_secret_previous") and prev_exp_raw: try: exp_dt = datetime.fromisoformat(str(prev_exp_raw)) if exp_dt.tzinfo is None: exp_dt = exp_dt.replace(tzinfo=UTC) has_previous = exp_dt > datetime.now(UTC) except ValueError: has_previous = False return { "has_primary": has_primary, "has_previous": has_previous, "previous_expires_at": prev_exp_raw if has_previous else None, } def _require_webhook_receiver(self, row: Integration) -> None: """Guard: the integration's driver must declare a ``webhook_hmac_secret`` field — i.e. it actually receives inbound callbacks. Reused by every webhook-auth admin action so the Basic-credential / IP-allowlist controls never appear on an outbound-only integration (e.g. Atlassian).""" driver_cls = get_driver_class(row.kind, row.provider) if driver_cls is None: raise AppException( code="INTEGRATION_DRIVER_UNKNOWN", message=f"No driver registered for ({row.kind!r}, {row.provider!r})", status_code=500, ) field_keys = {f.key for f in (driver_cls.config_schema or [])} if "webhook_hmac_secret" not in field_keys: raise AppException( code="WEBHOOK_AUTH_NOT_APPLICABLE", message=( f"Integration provider {row.provider!r} does not " "receive webhooks; nothing to configure." ), status_code=400, ) def _decrypt_secrets_or_raise(self, row: Integration) -> dict[str, Any]: """Decrypt ``secrets_encrypted`` for a mutation path, mapping crypto failures to the same 503 codes the rotate flow uses.""" if not (row.secrets_encrypted and row.secrets_nonce): return {} try: return decrypt_fields( row.secrets_encrypted, row.secrets_nonce, integration_id=str(row.id), ) except CredentialDecryptionError as e: raise AppException( code="ENCRYPTION_KEY_MISMATCH", message=( f"Integration {row.slug or row.id!r} existing secrets " f"cannot be decrypted: {e}. Restore the original " "CREDENTIAL_ENCRYPTION_KEY first." ), status_code=503, ) from e except RuntimeError as e: raise AppException( code="ENCRYPTION_KEY_MISSING", message=str(e), status_code=503, ) from e async def rotate_webhook_basic_credential( self, key: str | UUID, *, trace_id: str, actor_user_id: UUID, ) -> dict[str, Any]: """Generate an HTTP Basic-Auth credential (username + fresh random password) for an inbound webhook receiver. This is the *simple* alternative to the HMAC scheme for upstream tools that cannot compute a per-request body signature (e.g. SAP SuccessFactors Integration Center, where the integrator just picks "基本验证" from a dropdown and pastes user + password). The webhook endpoint accepts EITHER a valid Basic credential OR an HMAC-signed request, so enabling Basic does not disable the HMAC path. Security trade-off vs HMAC: a leaked Basic credential lets an attacker forge requests (no body binding / replay window). Mitigated by TLS, the always-on mass-leaver brake, and the optional webhook IP allowlist (see :meth:`set_webhook_ip_allowlist`). Plaintext password is returned exactly once; it is encrypted before persistence and no endpoint reveals it again. """ import secrets as _secrets row = await _load_by_id_or_slug(self.db, key, for_update=True) if row is None: raise NotFoundError("Integration", str(key)) self._require_webhook_receiver(row) existing = self._decrypt_secrets_or_raise(row) cfg_public = dict(row.config_public or {}) # Keep an already-chosen username; default to a stable readable one. username = ( (cfg_public.get("webhook_basic_username") or "").strip() or "sf-webhook" ) # 24 bytes urlsafe ≈ 32 chars — strong, copy-pasteable, no ':' that # would break the Basic ``user:pass`` split. password = _secrets.token_urlsafe(24) existing["webhook_basic_password"] = password cfg_public["webhook_basic_username"] = username cipher, nonce = encrypt_fields(existing, integration_id=str(row.id)) row.secrets_encrypted = cipher row.secrets_nonce = nonce row.config_public = cfg_public # reassign so JSON column is dirtied row.updated_by = actor_user_id await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="integration.webhook_basic_credential_rotate", status="success", resource_type="integration", resource_id=str(row.id), detail={"slug": row.slug, "username": username}, ) # NOTE: never log or audit the plaintext password. return {"username": username, "password": password} async def set_webhook_ip_allowlist( self, key: str | UUID, *, cidrs: list[str], trace_id: str, actor_user_id: UUID, ) -> dict[str, Any]: """Set the inbound-webhook IP allowlist for an integration. Each entry is a bare IP or CIDR (IPv4/IPv6). An **empty** list clears the allowlist, which means *allow all source IPs* (the webhook then relies on Basic/HMAC auth alone). When non-empty, the webhook endpoint rejects (403) any caller whose resolved client IP is not inside one of the ranges — recovering most of the security lost by allowing the simpler Basic scheme. """ import ipaddress row = await _load_by_id_or_slug(self.db, key, for_update=True) if row is None: raise NotFoundError("Integration", str(key)) self._require_webhook_receiver(row) normalized: list[str] = [] seen: set[str] = set() for raw in cidrs or []: entry = (raw or "").strip() if not entry: continue try: net = ipaddress.ip_network(entry, strict=False) except ValueError as exc: raise AppException( code="WEBHOOK_IP_ALLOWLIST_INVALID", message=f"{entry!r} 不是合法的 IP 或 CIDR。", status_code=400, ) from exc canon = str(net) if canon not in seen: seen.add(canon) normalized.append(canon) cfg_public = dict(row.config_public or {}) if normalized: cfg_public["webhook_ip_allowlist"] = normalized else: cfg_public.pop("webhook_ip_allowlist", None) row.config_public = cfg_public row.updated_by = actor_user_id await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="integration.webhook_ip_allowlist_set", status="success", resource_type="integration", resource_id=str(row.id), detail={"slug": row.slug, "count": len(normalized)}, ) return {"ip_allowlist": normalized} async def webhook_auth_status( self, key: str | UUID, ) -> dict[str, Any]: """Non-secret status for the webhook-auth admin UI: which schemes are configured + the Basic username + the IP allowlist. Never returns the HMAC secret or the Basic password.""" row = await _load_by_id_or_slug(self.db, key) if row is None: raise NotFoundError("Integration", str(key)) existing: dict[str, Any] = {} if row.secrets_encrypted and row.secrets_nonce: try: existing = decrypt_fields( row.secrets_encrypted, row.secrets_nonce, integration_id=str(row.id), ) except (CredentialDecryptionError, RuntimeError): existing = {} cfg_public = row.config_public or {} return { "has_hmac_secret": bool(existing.get("webhook_hmac_secret")), "has_basic_credential": bool(existing.get("webhook_basic_password")), "basic_username": cfg_public.get("webhook_basic_username"), "ip_allowlist": list(cfg_public.get("webhook_ip_allowlist") or []), } # ─── Approval flows sub-CRUD ─ async def list_approval_flows( self, integration_key: str | UUID, ) -> list[ApprovalFlowResponse]: integration = await _load_by_id_or_slug(self.db, integration_key) if integration is None: raise NotFoundError("Integration", str(integration_key)) rows = ( await self.db.execute( select(IntegrationApprovalFlow) .where(IntegrationApprovalFlow.integration_id == integration.id) .order_by(IntegrationApprovalFlow.flow_key) ) ).scalars().all() return [ ApprovalFlowResponse( id=r.id, integration_id=r.integration_id, flow_key=r.flow_key, label=r.label, external_code=r.external_code, payload_schema=r.payload_schema, active=r.active, created_at=r.created_at, updated_at=r.updated_at, ) for r in rows ] async def upsert_approval_flow( self, integration_key: str | UUID, *, flow_key: str, label: str, external_code: str, payload_schema: dict[str, Any] | None, active: bool, trace_id: str, actor_user_id: UUID, ) -> ApprovalFlowResponse: integration = await _load_by_id_or_slug(self.db, integration_key) if integration is None: raise NotFoundError("Integration", str(integration_key)) if integration.kind != "chat_approval": raise AppException( code="APPROVAL_FLOW_WRONG_KIND", message=( f"Cannot define approval flow on a {integration.kind!r} integration; " f"kind must be 'chat_approval'." ), status_code=400, ) existing = ( await self.db.execute( select(IntegrationApprovalFlow).where( IntegrationApprovalFlow.integration_id == integration.id, IntegrationApprovalFlow.flow_key == flow_key, ) ) ).scalar_one_or_none() if existing is None: row = IntegrationApprovalFlow( integration_id=integration.id, flow_key=flow_key, label=label, external_code=external_code, payload_schema=payload_schema, active=active, ) self.db.add(row) else: row = existing row.label = label row.external_code = external_code row.payload_schema = payload_schema row.active = active await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="integration.approval_flow.upsert", status="success", resource_type="integration_approval_flow", resource_id=str(row.id), detail={"flow_key": flow_key, "integration_slug": integration.slug}, ) return ApprovalFlowResponse( id=row.id, integration_id=row.integration_id, flow_key=row.flow_key, label=row.label, external_code=row.external_code, payload_schema=row.payload_schema, active=row.active, created_at=row.created_at, updated_at=row.updated_at, ) async def delete_approval_flow( self, integration_key: str | UUID, flow_id: UUID, *, trace_id: str, actor_user_id: UUID, ) -> None: integration = await _load_by_id_or_slug(self.db, integration_key) if integration is None: raise NotFoundError("Integration", str(integration_key)) row = await self.db.get(IntegrationApprovalFlow, flow_id) if row is None or row.integration_id != integration.id: raise NotFoundError("IntegrationApprovalFlow", str(flow_id)) await self.db.delete(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="integration.approval_flow.delete", status="success", resource_type="integration_approval_flow", resource_id=str(flow_id), detail={"flow_key": row.flow_key, "integration_slug": integration.slug}, ) # ─── internals ─ def _split_fields( self, driver_cls: type[IntegrationDriver], payload: dict[str, Any], ) -> tuple[dict[str, Any], dict[str, Any]]: """Return (public_dict, secret_dict) based on the driver schema.""" public: dict[str, Any] = {} secrets: dict[str, Any] = {} for f in driver_cls.config_schema: if not self._visible(f, payload): continue if f.key not in payload: continue v = payload[f.key] if isinstance(v, str): v = v.strip() if f.key == "ca_bundle_path" and v not in (None, ""): if not isinstance(v, str): raise AppException( code="INTEGRATION_CA_BUNDLE_INVALID", message="ca_bundle_path must be a string", status_code=400, ) from app.domains.identity_sync.runners.tls import ( validate_identity_sync_ca_bundle_path, ) try: validate_identity_sync_ca_bundle_path(v) except ValueError as exc: raise AppException( code="INTEGRATION_CA_BUNDLE_INVALID", message=str(exc), status_code=400, ) from exc if f.is_secret: secrets[f.key] = v else: if v in (None, ""): continue public[f.key] = v return public, secrets def _require_fields( self, driver_cls: type[IntegrationDriver], public: dict[str, Any], secrets: dict[str, Any], *, is_new: bool, ) -> None: merged: dict[str, Any] = dict(public) merged.update(secrets) for field in driver_cls.config_schema: if not field.required: continue if not self._visible(field, merged): continue if field.is_secret: if field.key not in secrets or secrets.get(field.key) in (None, ""): # on update, absent-means-keep if is_new or field.key in secrets: raise AppException( code="INTEGRATION_CONFIG_REQUIRED", message=f"Missing required field: {field.label}", status_code=400, ) else: if public.get(field.key) in (None, ""): raise AppException( code="INTEGRATION_CONFIG_REQUIRED", message=f"Missing required field: {field.label}", status_code=400, ) @staticmethod def _visible(field: Any, values: dict[str, Any]) -> bool: """Return True when ``field.visible_when`` is satisfied.""" if not getattr(field, "visible_when", None): return True for k, expected in field.visible_when.items(): if str(values.get(k, "")) != str(expected): return False return True async def _unset_other_primaries(self, row: Integration) -> None: """Ensure at most one ``is_primary=True`` row per kind. Uses ``SELECT ... FOR UPDATE`` on the existing-primary rows so concurrent admins flipping primary on different rows of the same kind serialize at the row lock instead of racing past each other and leaving two primaries committed. Callers must set ``row.is_primary = True`` BEFORE calling this method so that the new primary is already in the session when the others are unset. Both changes are then flushed / committed together by the caller — no intermediate flush here avoids the zero-primary window between unset and set. """ others = ( await self.db.execute( select(Integration) .where( Integration.kind == row.kind, Integration.id != row.id, Integration.is_primary.is_(True), ) .with_for_update() ) ).scalars().all() for other in others: other.is_primary = False async def _serialize_many( self, rows: list[Integration], ) -> list[IntegrationResponse]: if not rows: return [] # Lazy-import to avoid a cmdb→integrations import cycle. from app.domains.cmdb.scheduled_jobs_provisioning import ( resolve_sync_capabilities, ) updater_ids = {r.updated_by for r in rows if r.updated_by} updaters: dict[UUID, IdentityPerson] = {} if updater_ids: people = ( await self.db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(updater_ids)) ) ).scalars().all() updaters = {p.id: p for p in people} responses: list[IntegrationResponse] = [] for row in rows: driver_cls = get_driver_class(row.kind, row.provider) redacted: dict[str, Any] = dict(row.config_public or {}) if driver_cls is not None: for field in driver_cls.config_schema: if field.is_secret: redacted[f"{field.key}_set"] = bool(row.secrets_encrypted) updater = updaters.get(row.updated_by) if row.updated_by else None # Resolve effective capabilities the same way the cron # scheduler does — driver default when DB row is blank, # admin override otherwise. ``resolve_sync_capabilities`` # internally calls ``get_provider_class(provider)`` which # raises ``ValueError`` for unknown providers (returns [] # from the helper itself in that case — see source). # Narrowing the catch to ``ValueError`` here so any other # exception (DB driver error, lazy-import bug) doesn't get # silently swallowed; falling back to raw row.capabilities # only on the known "no provider class" path. effective: list[str] = [] if row.kind == "cloud": try: effective = resolve_sync_capabilities(row) except ValueError: effective = list(row.capabilities or []) else: effective = list(row.capabilities or []) responses.append(IntegrationResponse( id=row.id, kind=row.kind, provider=row.provider, name=row.name, slug=row.slug, description=row.description, enabled=row.enabled, is_primary=row.is_primary, config=redacted, capabilities=row.capabilities, effective_capabilities=effective, last_tested_at=row.last_tested_at, last_test_status=row.last_test_status, last_test_message=row.last_test_message, created_at=row.created_at, updated_at=row.updated_at, created_by=row.created_by, updated_by=row.updated_by, updated_by_name=updater.display_name if updater else None, )) return responses # ─── Platform settings (non-integration, key-value) ─────────────── class PlatformSettingService: _SENSITIVE_MARKERS = ("secret", "token", "password", "key", "credential", "webhook") def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) async def list(self) -> list[PlatformSettingResponse]: rows = ( await self.db.execute( select(PlatformSetting).order_by(PlatformSetting.key) ) ).scalars().all() return await self._serialize(list(rows)) async def get(self, key: str) -> PlatformSettingResponse: row = await self.db.get(PlatformSetting, key) if row is None: raise NotFoundError("PlatformSetting", key) return (await self._serialize([row]))[0] async def upsert( self, key: str, value: Any, description: str | None, *, trace_id: str, actor_user_id: UUID, ) -> PlatformSettingResponse: row = await self.db.get(PlatformSetting, key) created = row is None if row is None: row = PlatformSetting(key=key, value=value, description=description) self.db.add(row) else: row.value = value if description is not None: row.description = description row.updated_by = actor_user_id await self.db.flush() # Audit detail: log the key's *name*, value *type*, and a size # hint. We deliberately do NOT log the value itself: the # namespace-grouped settings UI makes it easy for admins to # stash secret-shaped values (webhook signing secrets, # transitional API tokens, etc.) under arbitrary keys, and # audit_logs is read by Security Auditor role — broader than # platform_setting write scope. redacted = any(m in key.lower() for m in self._SENSITIVE_MARKERS) try: size_hint = len(json.dumps(value)) except (TypeError, ValueError): size_hint = -1 await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="platform_setting.create" if created else "platform_setting.update", status="success", resource_type="platform_setting", resource_id=key, detail={ "key": key, "value_type": type(value).__name__, "value_size": size_hint, "redacted_by_key_name": redacted, }, ) return (await self._serialize([row]))[0] async def delete( self, key: str, *, trace_id: str, actor_user_id: UUID, ) -> None: row = await self.db.get(PlatformSetting, key) if row is None: raise NotFoundError("PlatformSetting", key) await self.db.delete(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="platform_setting.delete", status="success", resource_type="platform_setting", resource_id=key, ) async def _serialize( self, rows: list[PlatformSetting], ) -> list[PlatformSettingResponse]: updater_ids = {r.updated_by for r in rows if r.updated_by} updaters: dict[UUID, IdentityPerson] = {} if updater_ids: people = ( await self.db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(updater_ids)) ) ).scalars().all() updaters = {p.id: p for p in people} return [ PlatformSettingResponse( key=r.key, value=( "" if any(m in r.key.lower() for m in self._SENSITIVE_MARKERS) else r.value ), description=r.description, updated_by=r.updated_by, updated_by_name=( updaters[r.updated_by].display_name if r.updated_by and r.updated_by in updaters else None ), updated_at=r.updated_at, created_at=r.created_at, ) for r in rows ] # ─── util ──────────────────────────────────────────────────────── def _validate_slug(slug: str) -> None: if not _SLUG_RE.fullmatch(slug): raise AppException( code="INVALID_SLUG", message=( "Slug must be lowercase letters/digits/hyphens/underscores," " 2-64 chars, starting and ending with a letter or digit." ), status_code=400, ) async def get_platform_setting( key: str, default: Any = None, *, session: AsyncSession | None = None, ) -> Any: """Convenience reader for domain code.""" async def _body(s: AsyncSession) -> Any: row = await s.get(PlatformSetting, key) return row.value if row is not None else default if session is not None: return await _body(session) async with async_session_factory() as owned: return await _body(owned) __all__ = [ "IntegrationService", "PlatformSettingService", "build_driver_by_id", "build_primary_driver", "get_approval_flow", "get_approval_flows_map", "get_platform_setting", "get_primary", ]